diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 4ea0e7eb..231cca1e 100644 --- a/.github/workflows/main.yml +++ b/.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: diff --git a/fail2ban/server/filter.py b/fail2ban/server/filter.py index f514b337..e9000447 100644 --- a/fail2ban/server/filter.py +++ b/fail2ban/server/filter.py @@ -722,7 +722,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): diff --git a/fail2ban/server/filterpoll.py b/fail2ban/server/filterpoll.py index 7ee00540..196955e5 100644 --- a/fail2ban/server/filterpoll.py +++ b/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 diff --git a/fail2ban/server/filterpyinotify.py b/fail2ban/server/filterpyinotify.py index d62348a2..16b6cfd5 100644 --- a/fail2ban/server/filterpyinotify.py +++ b/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 diff --git a/fail2ban/server/filtersystemd.py b/fail2ban/server/filtersystemd.py index 925109d1..88f8c292 100644 --- a/fail2ban/server/filtersystemd.py +++ b/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) diff --git a/fail2ban/tests/filtertestcase.py b/fail2ban/tests/filtertestcase.py index 3cc17fb1..5166bc43 100644 --- a/fail2ban/tests/filtertestcase.py +++ b/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",