diff --git a/fail2ban/server/filter.py b/fail2ban/server/filter.py index d2cd0ce9..f8417d2d 100644 --- a/fail2ban/server/filter.py +++ b/fail2ban/server/filter.py @@ -104,6 +104,10 @@ class Filter(JailThread): ## Error counter (protected, so can be used in filter implementations) ## if it reached 100 (at once), run-cycle will go idle self._errors = 0 + ## Next time to update log or journal position in database: + self._nextUpdateTM = 0 + ## Pending updates (must be executed at next update time or during stop): + self._pendDBUpdates = {} ## return raw host (host is not dns): self.returnRawHost = False ## check each regex (used for test purposes): @@ -1023,9 +1027,6 @@ class FileFilter(Filter): log = self.__logs.pop(path) except KeyError: return - db = self.jail.database - if db is not None: - db.updateLog(self.jail, log) logSys.info("Removed logfile: %r", path) self._delLogPath(path) return @@ -1145,9 +1146,15 @@ class FileFilter(Filter): self.processLineAndAdd(line.rstrip('\r\n')) finally: log.close() - db = self.jail.database - if db is not None: - db.updateLog(self.jail, log) + if self.jail.database is not None: + self._pendDBUpdates[log] = 1 + if ( + self.ticks % 100 == 0 + or MyTime.time() >= self._nextUpdateTM + or not self.active + ): + self._updateDBPending() + self._nextUpdateTM = MyTime.time() + Utils.DEFAULT_SLEEP_TIME * 5 return True ## @@ -1247,12 +1254,33 @@ class FileFilter(Filter): ret.append(("File list", path)) return ret - def stop(self): - """Stop monitoring of log-file(s) + def _updateDBPending(self): + """Apply pending updates (log position) to database. """ + db = self.jail.database + while True: + try: + log, args = self._pendDBUpdates.popitem() + except KeyError: + break + db.updateLog(self.jail, log) + + def onStop(self): + """Stop monitoring of log-file(s). Invoked after run method. + """ + # ensure positions of pending logs are up-to-date: + if self._pendDBUpdates and self.jail.database: + self._updateDBPending() # stop files monitoring: for path in self.__logs.keys(): self.delLogPath(path) + + def stop(self): + """Stop filter + """ + # normally onStop will be called automatically in thread after its run ends, + # but for backwards compatibilities we'll invoke it in caller of stop method. + self.onStop() # stop thread: super(Filter, self).stop() @@ -1304,6 +1332,15 @@ class FileContainer: ## shows that log is in operation mode (expecting new messages only from here): self.inOperation = tail + def __hash__(self): + return hash(self.__filename) + def __eq__(self, other): + return (id(self) == id(other) or + self.__filename == (other.__filename if isinstance(other, FileContainer) else other) + ) + def __repr__(self): + return 'file-log:'+self.__filename + def getFileName(self): return self.__filename diff --git a/fail2ban/server/filtersystemd.py b/fail2ban/server/filtersystemd.py index a305b6cf..e2220105 100644 --- a/fail2ban/server/filtersystemd.py +++ b/fail2ban/server/filtersystemd.py @@ -61,7 +61,6 @@ 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") @@ -321,9 +320,10 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover #self.__journal.wait(self.sleeptime) != journal.NOP ## ## wait for entries without sleep in intervals, because "sleeping" in journal.wait: - Utils.wait_for(lambda: not self.active or \ - self.__journal.wait(Utils.DEFAULT_SLEEP_INTERVAL) != journal.NOP, - self.sleeptime, 0.00001) + if not logentry: + Utils.wait_for(lambda: not self.active or \ + self.__journal.wait(Utils.DEFAULT_SLEEP_INTERVAL) != journal.NOP, + self.sleeptime, 0.00001) if self.idle: # because journal.wait will returns immediatelly if we have records in journal, # just wait a little bit here for not idle, to prevent hi-load: @@ -368,15 +368,17 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover if self.ticks % 10 == 0: self.performSvc() # update position in log (time and iso string): - if (line and self.jail.database and ( - self.ticks % 10 == 0 - or MyTime.time() >= self.__nextUpdateTM + if self.jail.database: + if line: + self._pendDBUpdates['systemd-journal'] = (tm, line[1]) + line = None + if self._pendDBUpdates and ( + self.ticks % 100 == 0 + or MyTime.time() >= self._nextUpdateTM or not self.active - ) - ): - self.jail.database.updateJournal(self.jail, 'systemd-journal', tm, line[1]) - self.__nextUpdateTM = MyTime.time() + Utils.DEFAULT_SLEEP_TIME * 5 - line = None + ): + self._updateDBPending() + self._nextUpdateTM = MyTime.time() + Utils.DEFAULT_SLEEP_TIME * 5 except Exception as e: # pragma: no cover if not self.active: # if not active - error by stop... break @@ -403,3 +405,22 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover ret.append(("Journal matches", [" + ".join(" ".join(match) for match in self.__matches)])) return ret + + def _updateDBPending(self): + """Apply pending updates (jornal position) to database. + """ + db = self.jail.database + while True: + try: + log, args = self._pendDBUpdates.popitem() + except KeyError: + break + db.updateJournal(self.jail, log, *args) + + def onStop(self): + """Stop monitoring of journal. Invoked after run method. + """ + # ensure positions of pending logs are up-to-date: + if self._pendDBUpdates and self.jail.database: + self._updateDBPending() + diff --git a/fail2ban/server/jailthread.py b/fail2ban/server/jailthread.py index 94f34542..67955a06 100644 --- a/fail2ban/server/jailthread.py +++ b/fail2ban/server/jailthread.py @@ -67,6 +67,8 @@ class JailThread(Thread): def run_with_except_hook(*args, **kwargs): try: run(*args, **kwargs) + # call on stop callback to do some finalizations: + self.onStop() except Exception as e: # avoid very sporadic error "'NoneType' object has no attribute 'exc_info'" (https://bugs.python.org/issue7336) # only extremely fast systems are affected ATM (2.7 / 3.x), if thread ends nothing is available here. @@ -97,6 +99,12 @@ class JailThread(Thread): self.active = True super(JailThread, self).start() + @abstractmethod + def onStop(self): # pragma: no cover - absract + """Abstract - Called when thread ends (after run). + """ + pass + def stop(self): """Sets `active` property to False, to flag run method to return. """