fail2ban-regex: implemented new feature `-i` or `--invert` - inverting the sense of matching, to output non-matching lines.

pull/4001/head
sebres 2025-05-06 18:15:05 +02:00
parent f7aaaf50b8
commit cca2de984f
2 changed files with 32 additions and 4 deletions

View File

@ -172,6 +172,8 @@ def get_opt_parser():
help="Disable check for all regex's"),
Option("-o", "--out", action="store", dest="out", default=None,
help="Set token to print failure information only (row, id, ip, msg, host, ip4, ip6, dns, matches, ...)"),
Option("-i", "--invert", action="store_true", dest="invert",
help="Invert the sense of matching, to output non-matching lines."),
Option("--print-no-missed", action='store_true',
help="Do not print any missed lines"),
Option("--print-no-ignored", action='store_true',
@ -529,7 +531,7 @@ class Fail2banRegex(object):
except RegexException as e: # pragma: no cover
output( 'ERROR: %s' % e )
return None, 0, None
if self._filter.getMaxLines() > 1:
if self._filter.getMaxLines() > 1 and not self._opts.out:
for bufLine in orgLineBuffer[int(fullBuffer):]:
if bufLine not in self._filter._Filter__lineBuffer:
try:
@ -619,8 +621,10 @@ class Fail2banRegex(object):
def process(self, test_lines):
t0 = time.time()
out = None
if self._opts.out: # get out function
out = self._prepaireOutput()
outinv = self._opts.invert
for line in test_lines:
if isinstance(line, tuple):
line_datetimestripped, ret, is_ignored = self.testRegex(line[0], line[1])
@ -632,8 +636,13 @@ class Fail2banRegex(object):
continue
line_datetimestripped, ret, is_ignored = self.testRegex(line)
if self._opts.out: # (formatted) output:
if len(ret) > 0 and not is_ignored: out(ret)
if out: # (formatted) output:
if len(ret) > 0 and not is_ignored:
if not outinv: out(ret)
elif outinv: # inverted output (currently only time and message as matches):
if not len(ret): # [failRegexIndex, fid, date, fail]
ret = [[-1, "", self._filter._Filter__lastDate, {"fid":"", "matches":[line]}]]
out(ret)
continue
if is_ignored:

View File

@ -432,8 +432,16 @@ class Fail2banRegexTest(LogCaptureTestCase):
self.assertLogged('output: %s' % "['192.0.2.0'", "'ip4': '192.0.2.0'", "'user': 'kevin'", all=True)
self.pruneLog()
# log msg :
self.assertTrue(_test_exec('-o', 'msg', STR_00, RE_00_USER))
nmline = "Dec 31 12:00:00 [sshd] error: PAM: No failure for user from 192.0.2.123"
lines = STR_00+"\n"+nmline
self.assertTrue(_test_exec('-o', 'msg', lines, RE_00_USER))
self.assertLogged('output: %s' % STR_00)
self.assertNotLogged('output: %s' % nmline)
self.pruneLog()
# log msg (inverted) :
self.assertTrue(_test_exec('-o', 'msg', '-i', lines, RE_00_USER))
self.assertLogged('output: %s' % nmline)
self.assertNotLogged('output: %s' % STR_00)
self.pruneLog()
# item of match (user):
self.assertTrue(_test_exec('-o', 'user', STR_00, RE_00_USER))
@ -443,6 +451,17 @@ class Fail2banRegexTest(LogCaptureTestCase):
self.assertTrue(_test_exec('-o', '<ip>, <F-USER>, <family>', STR_00, RE_00_USER))
self.assertLogged('output: %s' % '192.0.2.0, kevin, inet4')
self.pruneLog()
# log msg :
lines = nmline+"\n"+STR_00; # just reverse lines (to cover possible order dependencies)
self.assertTrue(_test_exec('-o', '<time> : <msg>', lines, RE_00_USER))
self.assertLogged('output: %s : %s' % (1104490799.0, STR_00))
self.assertNotLogged('output: %s' % nmline)
self.pruneLog()
# log msg (inverted) :
self.assertTrue(_test_exec('-o', '<time> : <msg>', '-i', lines, RE_00_USER))
self.assertLogged('output: %s : %s' % (1104490800.0, nmline))
self.assertNotLogged('output: %s' % STR_00)
self.pruneLog()
def testStalledIPByNoFailFrmtOutput(self):
opts = (