Merge branch '0.10' into 0.11

.github/workflows/main.yml

@@ -22,7 +22,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        python-version: [2.7, 3.5, 3.6, 3.7, 3.8, 3.9, '3.10.0-beta.1', pypy2, pypy3]
+        python-version: [2.7, 3.5, 3.6, 3.7, 3.8, 3.9, '3.10.0-rc.2', pypy2, pypy3]
       fail-fast: false
     # Steps represent a sequence of tasks that will be executed as part of the job
     steps:

fail2ban/server/filter.py

@@ -726,7 +726,7 @@ class Filter(JailThread):
 			# incr common error counter:
 			self.commonError()
 
-	def commonError(self):
+	def commonError(self, reason="common", exc=None):
 		# incr error counter, stop processing (going idle) after 100th error :
 		self._errors += 1
 		# sleep a little bit (to get around time-related errors):

fail2ban/server/filterpoll.py

@@ -122,7 +122,7 @@ class FilterPoll(FileFilter):
 				logSys.error("Caught unhandled exception in main cycle: %r", e,
 					exc_info=logSys.getEffectiveLevel()<=logging.DEBUG)
 				# incr common error counter:
-				self.commonError()
+				self.commonError("unhandled", e)
 		logSys.debug("[%s] filter terminated", self.jailName)
 		return True
 

fail2ban/server/filterpyinotify.py

@@ -165,7 +165,7 @@ class FilterPyinotify(FileFilter):
 			return
 		found = {}
 		minTime = 60
-		for path, (retardTM, isDir) in self.__pending.iteritems():
+		for path, (retardTM, isDir) in list(self.__pending.items()):
 			if ntm - self.__pendingChkTime < retardTM:
 				if minTime > retardTM: minTime = retardTM
 				continue
@@ -188,7 +188,7 @@ class FilterPyinotify(FileFilter):
 				self._refreshWatcher(path, isDir=isDir)
 			if isDir:
 				# check all files belong to this dir:
-				for logpath in self.__watchFiles:
+				for logpath in list(self.__watchFiles):
 					if logpath.startswith(path + pathsep):
 						# if still no file - add to pending, otherwise refresh and process:
 						if not os.path.isfile(logpath):
@@ -268,15 +268,13 @@ class FilterPyinotify(FileFilter):
 
 	def _addLogPath(self, path):
 		self._addFileWatcher(path)
-		# initial scan:
+		# notify (wake up if in waiting):
 		if self.active:
-			# we can execute it right now:
-			self._process_file(path)
-		else:
-			# retard until filter gets started, isDir=None signals special case: process file only (don't need to refresh monitor):
-			self._addPending(path, ('INITIAL', path), isDir=None)
+			self.__pendingMinTime = 0
+		# retard until filter gets started, isDir=None signals special case: process file only (don't need to refresh monitor):
+		self._addPending(path, ('INITIAL', path), isDir=None)
 
-    ##
+	##
 	# Delete a log path
 	#
 	# @param path the log file to delete
@@ -287,7 +285,7 @@ class FilterPyinotify(FileFilter):
 			logSys.error("Failed to remove watch on path: %s", path)
 
 		path_dir = dirname(path)
-		for k in self.__watchFiles:
+		for k in list(self.__watchFiles):
 			if k.startswith(path_dir + pathsep):
 				path_dir = None
 				break
@@ -341,12 +339,17 @@ class FilterPyinotify(FileFilter):
 				self.__notifier.process_events()
 
 				# wait for events / timeout:
-				notify_maxtout = self.__notify_maxtout
 				def __check_events():
-					return not self.active or self.__notifier.check_events(timeout=notify_maxtout)
-				if Utils.wait_for(__check_events, min(self.sleeptime, self.__pendingMinTime)):
+					return (
+						not self.active
+						or bool(self.__notifier.check_events(timeout=self.__notify_maxtout))
+						or (self.__pendingMinTime and self.__pending)
+					)
+				wres = Utils.wait_for(__check_events, min(self.sleeptime, self.__pendingMinTime))
+				if wres:
 					if not self.active: break
-					self.__notifier.read_events()
+					if not isinstance(wres, dict):
+						self.__notifier.read_events()
 
 				self.ticks += 1
 
@@ -363,7 +366,7 @@ class FilterPyinotify(FileFilter):
 				logSys.error("Caught unhandled exception in main cycle: %r", e,
 					exc_info=logSys.getEffectiveLevel()<=logging.DEBUG)
 				# incr common error counter:
-				self.commonError()
+				self.commonError("unhandled", e)
 			
 		logSys.debug("[%s] filter exited (pyinotifier)", self.jailName)
 		self.__notifier = None

fail2ban/server/filtersystemd.py

@@ -61,6 +61,7 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover
 		# Initialise systemd-journal connection
 		self.__journal = journal.Reader(**jrnlargs)
 		self.__matches = []
+		self.__nextUpdateTM = 0
 		self.setDatePattern(None)
 		logSys.debug("Created FilterSystemd")
 
@@ -285,6 +286,7 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover
 		except OSError:
 			pass # Reading failure, so safe to ignore
 
+		line = None
 		while self.active:
 			# wait for records (or for timeout in sleeptime seconds):
 			try:
@@ -326,15 +328,22 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover
 				if self.ticks % 10 == 0:
 					self.performSvc()
 				# update position in log (time and iso string):
-				if self.jail.database is not None:
+				if (line and self.jail.database and (
+				    self.ticks % 10 == 0
+				    or MyTime.time() >= self.__nextUpdateTM
+				    or not self.active
+				  )
+				):
 					self.jail.database.updateJournal(self.jail, 'systemd-journal', line[1], line[0][1])
+					self.__nextUpdateTM = MyTime.time() + Utils.DEFAULT_SLEEP_TIME * 5
+					line = None
 			except Exception as e: # pragma: no cover
 				if not self.active: # if not active - error by stop...
 					break
 				logSys.error("Caught unhandled exception in main cycle: %r", e,
 					exc_info=logSys.getEffectiveLevel()<=logging.DEBUG)
 				# incr common error counter:
-				self.commonError()
+				self.commonError("unhandled", e)
 
 		logSys.debug("[%s] filter terminated", self.jailName)
 

fail2ban/tests/filtertestcase.py

@@ -979,6 +979,10 @@ class CommonMonitorTestCase(unittest.TestCase):
 		super(CommonMonitorTestCase, self).setUp()
 		self._failTotal = 0
 
+	def tearDown(self):
+		super(CommonMonitorTestCase, self).tearDown()
+		self.assertFalse(hasattr(self, "_unexpectedError"))
+
 	def waitFailTotal(self, count, delay=1):
 		"""Wait up to `delay` sec to assure that expected failure `count` reached
 		"""
@@ -1004,6 +1008,16 @@ class CommonMonitorTestCase(unittest.TestCase):
 		last_ticks = self.filter.ticks
 		return Utils.wait_for(lambda: self.filter.ticks >= last_ticks + ticks, _maxWaitTime(delay))
 
+	def commonFltError(self, reason="common", exc=None):
+		""" Mock-up for default common error handler to find catched unhandled exceptions
+		could occur in filters
+		"""
+		self._commonFltError(reason, exc)
+		if reason == "unhandled":
+			DefLogSys.critical("Caught unhandled exception in main cycle of %r : %r", self.filter, exc, exc_info=True)
+			self._unexpectedError = True
+		# self.assertNotEqual(reason, "unhandled")
+
 
 def get_monitor_failures_testcase(Filter_):
 	"""Generator of TestCase's for different filters/backends
@@ -1026,6 +1040,8 @@ def get_monitor_failures_testcase(Filter_):
 			self.file = open(self.name, 'a')
 			self.jail = DummyJail()
 			self.filter = Filter_(self.jail)
+			# mock-up common error to find catched unhandled exceptions:
+			self._commonFltError, self.filter.commonError = self.filter.commonError, self.commonFltError
 			self.filter.addLogPath(self.name, autoSeek=False)
 			# speedup search using exact date pattern:
 			self.filter.setDatePattern(r'^(?:%a )?%b %d %H:%M:%S(?:\.%f)?(?: %ExY)?')
@@ -1324,6 +1340,8 @@ def get_monitor_failures_journal_testcase(Filter_): # pragma: systemd no cover
 		def _initFilter(self, **kwargs):
 			self._getRuntimeJournal() # check journal available
 			self.filter = Filter_(self.jail, **kwargs)
+			# mock-up common error to find catched unhandled exceptions:
+			self._commonFltError, self.filter.commonError = self.filter.commonError, self.commonFltError
 			self.filter.addJournalMatch([
 				"SYSLOG_IDENTIFIER=fail2ban-testcases",
 				"TEST_FIELD=1",