From 9e6d07d928d1016be8195613aac2c0a2b53b9342 Mon Sep 17 00:00:00 2001 From: sebres Date: Wed, 22 Jan 2020 17:19:35 +0100 Subject: [PATCH 1/3] testSampleRegexsFactory: `time` is not mandatory anymore (check time only if set in json), allows usage of same line(s) matching different `logtype` option: `# filterOptions: [{"logtype": "file"}, {"logtype": "short"}, {"logtype": "journal"}]` --- fail2ban/tests/samplestestcase.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/fail2ban/tests/samplestestcase.py b/fail2ban/tests/samplestestcase.py index 1039b65e..0bbd05f5 100644 --- a/fail2ban/tests/samplestestcase.py +++ b/fail2ban/tests/samplestestcase.py @@ -223,11 +223,9 @@ def testSampleRegexsFactory(name, basedir): try: fail = {} # for logtype "journal" we don't need parse timestamp (simulate real systemd-backend handling): - checktime = True if opts.get('logtype') != 'journal': ret = flt.processLine(line) else: # simulate journal processing, time is known from journal (formatJournalEntry): - checktime = False if opts.get('test.prefix-line'): # journal backends creates common prefix-line: line = opts.get('test.prefix-line') + line ret = flt.processLine(('', TEST_NOW_STR, line.rstrip('\r\n')), TEST_NOW) @@ -271,7 +269,7 @@ def testSampleRegexsFactory(name, basedir): self.assertEqual(fv, v) t = faildata.get("time", None) - if checktime or t is not None: + if t is not None: try: jsonTimeLocal = datetime.datetime.strptime(t, "%Y-%m-%dT%H:%M:%S") except ValueError: From 569dea2b197b5a6b4c9f7b6450d6a87da58586ae Mon Sep 17 00:00:00 2001 From: sebres Date: Wed, 22 Jan 2020 17:24:40 +0100 Subject: [PATCH 2/3] filter.d/mysqld-auth.conf: capture user name in filter (can be more strict if user switched, used in action or fail2ban-regex output); also add coverage for mariadb 10.4 log format (gh-2611) --- config/filter.d/mysqld-auth.conf | 2 +- fail2ban/tests/files/logs/mysqld-auth | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/config/filter.d/mysqld-auth.conf b/config/filter.d/mysqld-auth.conf index 97b37920..930c9b5a 100644 --- a/config/filter.d/mysqld-auth.conf +++ b/config/filter.d/mysqld-auth.conf @@ -17,7 +17,7 @@ before = common.conf _daemon = mysqld -failregex = ^%(__prefix_line)s(?:(?:\d{6}|\d{4}-\d{2}-\d{2})[ T]\s?\d{1,2}:\d{2}:\d{2} )?(?:\d+ )?\[\w+\] (?:\[[^\]]+\] )*Access denied for user '[^']+'@'' (to database '[^']*'|\(using password: (YES|NO)\))*\s*$ +failregex = ^%(__prefix_line)s(?:(?:\d{6}|\d{4}-\d{2}-\d{2})[ T]\s?\d{1,2}:\d{2}:\d{2} )?(?:\d+ )?\[\w+\] (?:\[[^\]]+\] )*Access denied for user '[^']+'@'' (to database '[^']*'|\(using password: (YES|NO)\))*\s*$ ignoreregex = diff --git a/fail2ban/tests/files/logs/mysqld-auth b/fail2ban/tests/files/logs/mysqld-auth index 0b0827f9..29faeb71 100644 --- a/fail2ban/tests/files/logs/mysqld-auth +++ b/fail2ban/tests/files/logs/mysqld-auth @@ -33,3 +33,7 @@ Sep 16 21:30:32 catinthehat mysqld: 130916 21:30:32 [Warning] Access denied for 2019-09-06T01:45:18 srv mysqld: 2019-09-06 1:45:18 140581192722176 [Warning] Access denied for user 'global'@'192.0.2.2' (using password: YES) # failJSON: { "time": "2019-09-24T13:16:50", "match": true , "host": "192.0.2.3", "desc": "ISO timestamp within log message" } 2019-09-24T13:16:50 srv mysqld[1234]: 2019-09-24 13:16:50 8756 [Warning] Access denied for user 'root'@'192.0.2.3' (using password: YES) + +# filterOptions: [{"logtype": "file"}, {"logtype": "short"}, {"logtype": "journal"}] +# failJSON: { "match": true , "host": "192.0.2.1", "user":"root", "desc": "mariadb 10.4 log format, gh-2611" } +2020-01-16 21:34:14 4644 [Warning] Access denied for user 'root'@'192.0.2.1' (using password: YES) From 12b3ac684a637e2adfef05997d4aa3ad36b68970 Mon Sep 17 00:00:00 2001 From: sebres Date: Tue, 28 Jan 2020 21:45:30 +0100 Subject: [PATCH 3/3] closes #2615: systemd backend would seek to last known position (or `now - findtime`) in journal at start. --- fail2ban/server/database.py | 64 +++++++++++++++++++++++------- fail2ban/server/filtersystemd.py | 36 +++++++++++------ fail2ban/server/jail.py | 4 ++ fail2ban/tests/databasetestcase.py | 9 +++++ fail2ban/tests/dummyjail.py | 11 +---- fail2ban/tests/filtertestcase.py | 49 ++++++++++++++++++++++- 6 files changed, 135 insertions(+), 38 deletions(-) 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()