github
/
fail2ban
mirror of https://github.com/fail2ban/fail2ban
1
0
Fork 0
Code Issues Releases Wiki Activity

closes #2615: systemd backend would seek to last known position (or `now - findtime`) in journal at start.

pull/2630/head
sebres 2020-01-28 21:45:30 +01:00
parent 569dea2b19
commit 12b3ac684a
6 changed files with 135 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
fail2ban/server/database.py
View File

@ -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):

36
fail2ban/server/filtersystemd.py
View File

@ -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

4
fail2ban/server/jail.py
View File

@ -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.

9
fail2ban/tests/databasetestcase.py
View File

@ -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"])

11
fail2ban/tests/dummyjail.py
View File

@ -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;

49
fail2ban/tests/filtertestcase.py
View File

@ -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()