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

amend to 4ff8d051f49808ac769709c5aff8591fcd79040a: fixed fail2ban-regex with journalmatch using systemd-journal

Browse Source
pull/1698/head
sebres 8 years ago
parent 22afdbd536
commit 8bcaeb9022
2 changed files with 39 additions and 34 deletions
Show Stats Download Patch File Download Diff File

49
fail2ban/client/fail2banregex.py
Unescape View File

@ -235,7 +235,7 @@ class Fail2banRegex(object):
else: else:
self._maxlines = 20 self._maxlines = 20
if opts.journalmatch is not None: if opts.journalmatch is not None:
self.setJournalMatch(opts.journalmatch.split()) self.setJournalMatch(shlex.split(opts.journalmatch))
if opts.datepattern: if opts.datepattern:
self.setDatePattern(opts.datepattern) self.setDatePattern(opts.datepattern)
if opts.usedns: if opts.usedns:
@ -243,6 +243,7 @@ class Fail2banRegex(object):
self._filter.returnRawHost = opts.raw self._filter.returnRawHost = opts.raw
self._filter.checkFindTime = False self._filter.checkFindTime = False
self._filter.checkAllRegex = True self._filter.checkAllRegex = True
self._opts = opts
def decode_line(self, line): def decode_line(self, line):
return FileContainer.decode_line('<LOG>', self._encoding, line) return FileContainer.decode_line('<LOG>', self._encoding, line)
@ -265,8 +266,7 @@ class Fail2banRegex(object):
output( "Use maxlines : %d" % self._filter.getMaxLines() ) output( "Use maxlines : %d" % self._filter.getMaxLines() )
def setJournalMatch(self, v): def setJournalMatch(self, v):
if self._journalmatch is None: self._journalmatch = v
self._journalmatch = v
def readRegex(self, value, regextype): def readRegex(self, value, regextype):
assert(regextype in ('fail', 'ignore')) assert(regextype in ('fail', 'ignore'))
@ -297,33 +297,38 @@ class Fail2banRegex(object):
if opt[0] == 'multi-set': if opt[0] == 'multi-set':
optval = opt[3] optval = opt[3]
elif opt[0] == 'set': elif opt[0] == 'set':
optval = [opt[3]] optval = opt[3:]
else: else:
continue continue
for optval in optval: try:
try: if opt[2] == "prefregex":
if opt[2] == "prefregex": for optval in optval:
self._filter.prefRegex = optval self._filter.prefRegex = optval
elif opt[2] == "addfailregex": elif opt[2] == "addfailregex":
stor = regex_values.get('fail') stor = regex_values.get('fail')
if not stor: stor = regex_values['fail'] = list() if not stor: stor = regex_values['fail'] = list()
for optval in optval:
stor.append(RegexStat(optval)) stor.append(RegexStat(optval))
#self._filter.addFailRegex(optval) #self._filter.addFailRegex(optval)
elif opt[2] == "addignoreregex": elif opt[2] == "addignoreregex":
stor = regex_values.get('ignore') stor = regex_values.get('ignore')
if not stor: stor = regex_values['ignore'] = list() if not stor: stor = regex_values['ignore'] = list()
for optval in optval:
stor.append(RegexStat(optval)) stor.append(RegexStat(optval))
#self._filter.addIgnoreRegex(optval) #self._filter.addIgnoreRegex(optval)
elif opt[2] == "maxlines": elif opt[2] == "maxlines":
for optval in optval:
self.setMaxLines(optval) self.setMaxLines(optval)
elif opt[2] == "datepattern": elif opt[2] == "datepattern":
for optval in optval:
self.setDatePattern(optval) self.setDatePattern(optval)
elif opt[2] == "addjournalmatch": elif opt[2] == "addjournalmatch":
if self._opts.journalmatch is None:
self.setJournalMatch(optval) self.setJournalMatch(optval)
except ValueError as e: # pragma: no cover except ValueError as e: # pragma: no cover
output( "ERROR: Invalid value for %s (%r) " \ output( "ERROR: Invalid value for %s (%r) " \
"read from %s: %s" % (opt[2], optval, value, e) ) "read from %s: %s" % (opt[2], optval, value, e) )
return False return False
else: else:
output( "Use %11s line : %s" % (regex, shortstr(value)) ) output( "Use %11s line : %s" % (regex, shortstr(value)) )
@ -510,7 +515,7 @@ class Fail2banRegex(object):
for line in hdlr: for line in hdlr:
yield self.decode_line(line) yield self.decode_line(line)
def start(self, opts, args): def start(self, args):
cmd_log, cmd_regex = args[:2] cmd_log, cmd_regex = args[:2]
@ -603,5 +608,5 @@ def exec_command_line(*args):
logSys.addHandler(stdout) logSys.addHandler(stdout)
fail2banRegex = Fail2banRegex(opts) fail2banRegex = Fail2banRegex(opts)
if not fail2banRegex.start(opts, args): if not fail2banRegex.start(args):
sys.exit(-1) sys.exit(-1)

24
fail2ban/tests/fail2banregextestcase.py
Unescape View File

@ -96,7 +96,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
(opts, args, fail2banRegex) = _Fail2banRegex( (opts, args, fail2banRegex) = _Fail2banRegex(
"test", r".** from <HOST>$" "test", r".** from <HOST>$"
) )
self.assertFalse(fail2banRegex.start(opts, args)) self.assertFalse(fail2banRegex.start(args))
self.assertLogged("Unable to compile regular expression") self.assertLogged("Unable to compile regular expression")
def testWrongIngnoreRE(self): def testWrongIngnoreRE(self):
@ -104,7 +104,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
"--datepattern", "{^LN-BEG}EPOCH", "--datepattern", "{^LN-BEG}EPOCH",
"test", r".*? from <HOST>$", r".**" "test", r".*? from <HOST>$", r".**"
) )
self.assertFalse(fail2banRegex.start(opts, args)) self.assertFalse(fail2banRegex.start(args))
self.assertLogged("Unable to compile regular expression") self.assertLogged("Unable to compile regular expression")
def testDirectFound(self): def testDirectFound(self):
@ -114,7 +114,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
"Dec 31 11:59:59 [sshd] error: PAM: Authentication failure for kevin from 192.0.2.0", "Dec 31 11:59:59 [sshd] error: PAM: Authentication failure for kevin from 192.0.2.0",
r"Authentication failure for .*? from <HOST>$" r"Authentication failure for .*? from <HOST>$"
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 1 lines, 0 ignored, 1 matched, 0 missed') self.assertLogged('Lines: 1 lines, 0 ignored, 1 matched, 0 missed')
def testDirectNotFound(self): def testDirectNotFound(self):
@ -123,7 +123,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
"Dec 31 11:59:59 [sshd] error: PAM: Authentication failure for kevin from 192.0.2.0", "Dec 31 11:59:59 [sshd] error: PAM: Authentication failure for kevin from 192.0.2.0",
r"XYZ from <HOST>$" r"XYZ from <HOST>$"
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 1 lines, 0 ignored, 0 matched, 1 missed') self.assertLogged('Lines: 1 lines, 0 ignored, 0 matched, 1 missed')
def testDirectIgnored(self): def testDirectIgnored(self):
@ -133,7 +133,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
r"Authentication failure for .*? from <HOST>$", r"Authentication failure for .*? from <HOST>$",
r"kevin from 192.0.2.0$" r"kevin from 192.0.2.0$"
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 1 lines, 1 ignored, 0 matched, 0 missed') self.assertLogged('Lines: 1 lines, 1 ignored, 0 matched, 0 missed')
def testDirectRE_1(self): def testDirectRE_1(self):
@ -143,7 +143,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
Fail2banRegexTest.FILENAME_01, Fail2banRegexTest.FILENAME_01,
Fail2banRegexTest.RE_00 Fail2banRegexTest.RE_00
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 19 lines, 0 ignored, 13 matched, 6 missed') self.assertLogged('Lines: 19 lines, 0 ignored, 13 matched, 6 missed')
self.assertLogged('Error decoding line'); self.assertLogged('Error decoding line');
@ -159,7 +159,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
Fail2banRegexTest.FILENAME_01, Fail2banRegexTest.FILENAME_01,
Fail2banRegexTest.RE_00 Fail2banRegexTest.RE_00
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 19 lines, 0 ignored, 16 matched, 3 missed') self.assertLogged('Lines: 19 lines, 0 ignored, 16 matched, 3 missed')
def testDirectRE_1raw_noDns(self): def testDirectRE_1raw_noDns(self):
@ -169,7 +169,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
Fail2banRegexTest.FILENAME_01, Fail2banRegexTest.FILENAME_01,
Fail2banRegexTest.RE_00 Fail2banRegexTest.RE_00
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 19 lines, 0 ignored, 13 matched, 6 missed') self.assertLogged('Lines: 19 lines, 0 ignored, 13 matched, 6 missed')
def testDirectRE_2(self): def testDirectRE_2(self):
@ -179,7 +179,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
Fail2banRegexTest.FILENAME_02, Fail2banRegexTest.FILENAME_02,
Fail2banRegexTest.RE_00 Fail2banRegexTest.RE_00
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 13 lines, 0 ignored, 5 matched, 8 missed') self.assertLogged('Lines: 13 lines, 0 ignored, 5 matched, 8 missed')
def testVerbose(self): def testVerbose(self):
@ -189,7 +189,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
Fail2banRegexTest.FILENAME_02, Fail2banRegexTest.FILENAME_02,
Fail2banRegexTest.RE_00 Fail2banRegexTest.RE_00
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 13 lines, 0 ignored, 5 matched, 8 missed') self.assertLogged('Lines: 13 lines, 0 ignored, 5 matched, 8 missed')
self.assertLogged('141.3.81.106 Sun Aug 14 11:53:59 2005') self.assertLogged('141.3.81.106 Sun Aug 14 11:53:59 2005')
@ -200,7 +200,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
"--datepattern", "^(?:%a )?%b %d %H:%M:%S(?:\.%f)?(?: %ExY)?", "--datepattern", "^(?:%a )?%b %d %H:%M:%S(?:\.%f)?(?: %ExY)?",
Fail2banRegexTest.FILENAME_WRONGCHAR, Fail2banRegexTest.FILTER_SSHD Fail2banRegexTest.FILENAME_WRONGCHAR, Fail2banRegexTest.FILTER_SSHD
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 4 lines, 0 ignored, 2 matched, 2 missed') self.assertLogged('Lines: 4 lines, 0 ignored, 2 matched, 2 missed')
self.assertLogged('Error decoding line') self.assertLogged('Error decoding line')
@ -215,7 +215,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
"--debuggex", "--print-all-matched", "--debuggex", "--print-all-matched",
Fail2banRegexTest.FILENAME_WRONGCHAR, Fail2banRegexTest.FILTER_SSHD Fail2banRegexTest.FILENAME_WRONGCHAR, Fail2banRegexTest.FILTER_SSHD
) )
self.assertTrue(fail2banRegex.start(opts, args)) self.assertTrue(fail2banRegex.start(args))
self.assertLogged('Lines: 4 lines, 0 ignored, 2 matched, 2 missed') self.assertLogged('Lines: 4 lines, 0 ignored, 2 matched, 2 missed')
self.assertLogged('https://') self.assertLogged('https://')

Write Preview
Loading…
Cancel
Save