diff --git a/fail2ban/server/database.py b/fail2ban/server/database.py index f584d74f..ab0de565 100644 --- a/fail2ban/server/database.py +++ b/fail2ban/server/database.py @@ -455,22 +455,24 @@ class Fail2BanDb(object): If log was already present in database, value of last position in the log file; else `None` """ + return self._addLog(cur, jail, container.getFileName(), container.getPos(), container.getHash()) + + def _addLog(self, cur, jail, name, pos=0, md5=None): lastLinePos = None cur.execute( "SELECT firstlinemd5, lastfilepos FROM logs " "WHERE jail=? AND path=?", - (jail.name, container.getFileName())) + (jail.name, name)) try: firstLineMD5, lastLinePos = cur.fetchone() except TypeError: - firstLineMD5 = False + firstLineMD5 = None - cur.execute( - "INSERT OR REPLACE INTO logs(jail, path, firstlinemd5, lastfilepos) " - "VALUES(?, ?, ?, ?)", - (jail.name, container.getFileName(), - container.getHash(), container.getPos())) - if container.getHash() != firstLineMD5: + if not firstLineMD5 and (pos or md5): + cur.execute( + "INSERT OR REPLACE INTO logs(jail, path, firstlinemd5, lastfilepos) " + "VALUES(?, ?, ?, ?)", (jail.name, name, md5, pos)) + if md5 is not None and md5 != firstLineMD5: lastLinePos = None return lastLinePos @@ -499,7 +501,7 @@ class Fail2BanDb(object): return set(row[0] for row in cur.fetchmany()) @commitandrollback - def updateLog(self, cur, *args, **kwargs): + def updateLog(self, cur, jail, container): """Updates hash and last position in log file. Parameters @@ -509,14 +511,48 @@ class Fail2BanDb(object): container : FileContainer File container of the log file being updated. """ - self._updateLog(cur, *args, **kwargs) + self._updateLog(cur, jail, container.getFileName(), container.getPos(), container.getHash()) - def _updateLog(self, cur, jail, container): + def _updateLog(self, cur, jail, name, pos, md5): cur.execute( "UPDATE logs SET firstlinemd5=?, lastfilepos=? " - "WHERE jail=? AND path=?", - (container.getHash(), container.getPos(), - jail.name, container.getFileName())) + "WHERE jail=? AND path=?", (md5, pos, jail.name, name)) + # be sure it is set (if not available): + if not cur.rowcount: + cur.execute( + "INSERT OR REPLACE INTO logs(jail, path, firstlinemd5, lastfilepos) " + "VALUES(?, ?, ?, ?)", (jail.name, name, md5, pos)) + + @commitandrollback + def getJournalPos(self, cur, jail, name, time=0, iso=None): + """Get journal position from database. + + Parameters + ---------- + jail : Jail + Jail of which the journal belongs to. + name, time, iso : + Journal name (typically systemd-journal) and last known time. + + Returns + ------- + int (or float) + Last position (as time) if it was already present in database; else `None` + """ + return self._addLog(cur, jail, name, time, iso); # no hash, just time as iso + + @commitandrollback + def updateJournal(self, cur, jail, name, time, iso): + """Updates last position (as time) of journal. + + Parameters + ---------- + jail : Jail + Jail of which the journal belongs to. + name, time, iso : + Journal name (typically systemd-journal) and last known time. + """ + self._updateLog(cur, jail, name, time, iso); # no hash, just time as iso @commitandrollback def addBan(self, cur, jail, ticket): diff --git a/fail2ban/server/filtersystemd.py b/fail2ban/server/filtersystemd.py index c2a72598..870b3058 100644 --- a/fail2ban/server/filtersystemd.py +++ b/fail2ban/server/filtersystemd.py @@ -190,6 +190,13 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover def getJournalReader(self): return self.__journal + def getJrnEntTime(self, logentry): + """ Returns time of entry as tuple (ISO-str, Posix).""" + date = logentry.get('_SOURCE_REALTIME_TIMESTAMP') + if date is None: + date = logentry.get('__REALTIME_TIMESTAMP') + return (date.isoformat(), time.mktime(date.timetuple()) + date.microsecond/1.0E6) + ## # Format journal log entry into syslog style # @@ -222,9 +229,8 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover logelements[-1] += v logelements[-1] += ":" if logelements[-1] == "kernel:": - if '_SOURCE_MONOTONIC_TIMESTAMP' in logentry: - monotonic = logentry.get('_SOURCE_MONOTONIC_TIMESTAMP') - else: + monotonic = logentry.get('_SOURCE_MONOTONIC_TIMESTAMP') + if monotonic is None: monotonic = logentry.get('__MONOTONIC_TIMESTAMP')[0] logelements.append("[%12.6f]" % monotonic.total_seconds()) msg = logentry.get('MESSAGE','') @@ -235,13 +241,11 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover logline = " ".join(logelements) - date = logentry.get('_SOURCE_REALTIME_TIMESTAMP', - logentry.get('__REALTIME_TIMESTAMP')) + date = self.getJrnEntTime(logentry) logSys.log(5, "[%s] Read systemd journal entry: %s %s", self.jailName, - date.isoformat(), logline) + date[0], logline) ## use the same type for 1st argument: - return ((logline[:0], date.isoformat(), logline.replace('\n', '\\n')), - time.mktime(date.timetuple()) + date.microsecond/1.0E6) + return ((logline[:0], date[0], logline.replace('\n', '\\n')), date[1]) def seekToTime(self, date): if not isinstance(date, datetime.datetime): @@ -262,9 +266,12 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover "Jail regexs will be checked against all journal entries, " "which is not advised for performance reasons.") - # Seek to now - findtime in journal - start_time = datetime.datetime.now() - \ - datetime.timedelta(seconds=int(self.getFindTime())) + # Try to obtain the last known time (position of journal) + start_time = 0 + if self.jail.database is not None: + start_time = self.jail.database.getJournalPos(self.jail, 'systemd-journal') or 0 + # Seek to max(last_known_time, now - findtime) in journal + start_time = max( start_time, MyTime.time() - int(self.getFindTime()) ) self.seekToTime(start_time) # Move back one entry to ensure do not end up in dead space # if start time beyond end of journal @@ -303,8 +310,8 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover e, exc_info=logSys.getEffectiveLevel() <= logging.DEBUG) self.ticks += 1 if logentry: - self.processLineAndAdd( - *self.formatJournalEntry(logentry)) + line = self.formatJournalEntry(logentry) + self.processLineAndAdd(*line) self.__modified += 1 if self.__modified >= 100: # todo: should be configurable break @@ -313,6 +320,9 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover if self.__modified: self.performBan() self.__modified = 0 + # update position in log (time and iso string): + if self.jail.database is not None: + self.jail.database.updateJournal(self.jail, 'systemd-journal', line[1], line[0][1]) except Exception as e: # pragma: no cover if not self.active: # if not active - error by stop... break diff --git a/fail2ban/server/jail.py b/fail2ban/server/jail.py index 1deea49b..89912556 100644 --- a/fail2ban/server/jail.py +++ b/fail2ban/server/jail.py @@ -156,6 +156,10 @@ class Jail(object): """ return self.__db + @database.setter + def database(self, value): + self.__db = value; + @property def filter(self): """The filter which the jail is using to monitor log files. diff --git a/fail2ban/tests/databasetestcase.py b/fail2ban/tests/databasetestcase.py index af9298a2..06f92d8c 100644 --- a/fail2ban/tests/databasetestcase.py +++ b/fail2ban/tests/databasetestcase.py @@ -225,6 +225,15 @@ class DatabaseTest(LogCaptureTestCase): self.db.addLog(self.jail, self.fileContainer), None) os.remove(filename) + def testUpdateJournal(self): + self.testAddJail() # Jail required + # not yet updated: + self.assertEqual(self.db.getJournalPos(self.jail, 'systemd-journal'), None) + # update 3 times (insert and 2 updates) and check it was set (and overwritten): + for t in (1500000000, 1500000001, 1500000002): + self.db.updateJournal(self.jail, 'systemd-journal', t, 'TEST'+str(t)) + self.assertEqual(self.db.getJournalPos(self.jail, 'systemd-journal'), t) + def testAddBan(self): self.testAddJail() ticket = FailTicket("127.0.0.1", 0, ["abc\n"]) diff --git a/fail2ban/tests/dummyjail.py b/fail2ban/tests/dummyjail.py index 19f97f4e..9e9aaeed 100644 --- a/fail2ban/tests/dummyjail.py +++ b/fail2ban/tests/dummyjail.py @@ -35,7 +35,6 @@ class DummyJail(Jail): self.lock = Lock() self.queue = [] super(DummyJail, self).__init__(name='DummyJail', backend=backend) - self.__db = None self.__actions = Actions(self) def __len__(self): @@ -63,7 +62,7 @@ class DummyJail(Jail): @property def name(self): - return "DummyJail #%s with %d tickets" % (id(self), len(self)) + return "DummyJail" + ("" if self.database else " #%s with %d tickets" % (id(self), len(self))) @property def idle(self): @@ -73,14 +72,6 @@ class DummyJail(Jail): def idle(self, value): pass - @property - def database(self): - return self.__db; - - @database.setter - def database(self, value): - self.__db = value; - @property def actions(self): return self.__actions; diff --git a/fail2ban/tests/filtertestcase.py b/fail2ban/tests/filtertestcase.py index 6ca8162b..91cb7eb6 100644 --- a/fail2ban/tests/filtertestcase.py +++ b/fail2ban/tests/filtertestcase.py @@ -43,7 +43,8 @@ from ..server.failmanager import FailManagerEmpty from ..server.ipdns import asip, getfqdn, DNSUtils, IPAddr from ..server.mytime import MyTime from ..server.utils import Utils, uni_decode -from .utils import setUpMyTime, tearDownMyTime, mtimesleep, with_tmpdir, LogCaptureTestCase, \ +from .databasetestcase import getFail2BanDb +from .utils import setUpMyTime, tearDownMyTime, mtimesleep, with_alt_time, with_tmpdir, LogCaptureTestCase, \ logSys as DefLogSys, CONFIG_DIR as STOCK_CONF_DIR from .dummyjail import DummyJail @@ -1397,6 +1398,52 @@ def get_monitor_failures_journal_testcase(Filter_): # pragma: systemd no cover self.test_file, self.journal_fields, skip=5, n=4) self.assert_correct_ban("193.168.0.128", 3) + @with_alt_time + def test_grow_file_with_db(self): + + def _gen_falure(ip): + # insert new failures ans check it is monitored: + fields = self.journal_fields + fields.update(TEST_JOURNAL_FIELDS) + journal.send(MESSAGE="error: PAM: Authentication failure for test from "+ip, **fields) + self.waitForTicks(1) + self.assert_correct_ban(ip, 1) + + # coverage for update log: + self.jail.database = getFail2BanDb(':memory:') + self.jail.database.addJail(self.jail) + MyTime.setTime(time.time()) + self._test_grow_file() + # stop: + self.filter.stop() + self.filter.join() + MyTime.setTime(time.time() + 2) + # update log manually (should cause a seek to end of log without wait for next second): + self.jail.database.updateJournal(self.jail, 'systemd-journal', MyTime.time(), 'TEST') + # check seek to last (simulated) position succeeds (without bans of previous copied tickets): + self._failTotal = 0 + self._initFilter() + self.filter.setMaxRetry(1) + self.filter.start() + self.waitForTicks(1) + # check new IP but no old IPs found: + _gen_falure("192.0.2.5") + self.assertFalse(self.jail.getFailTicket()) + + # now the same with increased time (check now - findtime case): + self.filter.stop() + self.filter.join() + MyTime.setTime(time.time() + 10000) + self._failTotal = 0 + self._initFilter() + self.filter.setMaxRetry(1) + self.filter.start() + self.waitForTicks(1) + MyTime.setTime(time.time() + 3) + # check new IP but no old IPs found: + _gen_falure("192.0.2.6") + self.assertFalse(self.jail.getFailTicket()) + def test_delJournalMatch(self): self._initFilter() self.filter.start()