fail2ban-regex: extended with same logic as fail2ban-server (sets `logtype` to `journal` if systemd backend is used (`systemd-journal` specified), to apply short prefix-line in filter)

fail2ban/client/fail2banregex.py

@@ -261,6 +261,7 @@ class Fail2banRegex(object):
 		self._filter.checkFindTime = False
 		self._filter.checkAllRegex = True
 		self._opts = opts
+		self._backend = 'auto'
 
 	def decode_line(self, line):
 		return FileContainer.decode_line('<LOG>', self._encoding, line)
@@ -327,6 +328,8 @@ class Fail2banRegex(object):
 				basedir = None
 				if not os.path.isabs(fltName): # avoid join with "filter.d" inside FilterReader
 					fltName = os.path.abspath(fltName)
+			if not fltOpt.get('logtype'):
+				fltOpt['logtype'] = ['file','journal'][int(self._backend.startswith("systemd"))]
 			if fltOpt:
 				output( "Use   filter options : %r" % fltOpt )
 			reader = FilterReader(fltName, 'fail2ban-regex-jail', fltOpt, share_config=self.share_config, basedir=basedir)
@@ -597,6 +600,9 @@ class Fail2banRegex(object):
 
 		cmd_log, cmd_regex = args[:2]
 
+		if cmd_log.startswith("systemd-journal"): # pragma: no cover
+			self._backend = 'systemd'
+
 		try:
 			if not self.readRegex(cmd_regex, 'fail'): # pragma: no cover
 				return False

fail2ban/tests/fail2banregextestcase.py

@@ -25,6 +25,7 @@ __license__ = "GPL"
 
 import os
 import sys
+import unittest
 
 from ..client import fail2banregex
 from ..client.fail2banregex import Fail2banRegex, get_opt_parser, exec_command_line, output, str2LogLevel
@@ -315,6 +316,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
 		_decode_line_warn.clear()
 
 	def testWronChar(self):
+		unittest.F2B.SkipIfCfgMissing(stock=True)
 		self._reset()
 		(opts, args, fail2banRegex) = _Fail2banRegex(
 			"-l", "notice", # put down log-level, because of too many debug-messages
@@ -331,6 +333,7 @@ class Fail2banRegexTest(LogCaptureTestCase):
 		self.assertLogged('Nov  8 00:16:12 main sshd[32547]: pam_succeed_if(sshd:auth): error retrieving information about user llinco')
 
 	def testWronCharDebuggex(self):
+		unittest.F2B.SkipIfCfgMissing(stock=True)
 		self._reset()
 		(opts, args, fail2banRegex) = _Fail2banRegex(
 			"-l", "notice", # put down log-level, because of too many debug-messages
@@ -381,3 +384,27 @@ class Fail2banRegexTest(LogCaptureTestCase):
 			'-v', '-d', '%:%.%-', 'LOG', 'RE'
 		), 0)
 		self.assertLogged('Failed to set datepattern')
+
+	def testLogtypeSystemdJournal(self): # pragma: no cover
+		if not fail2banregex.FilterSystemd:
+			raise unittest.SkipTest('Skip test because no systemd backand available')
+		(opts, args, fail2banRegex) = _Fail2banRegex(
+			"systemd-journal", Fail2banRegexTest.FILTER_ZZZ_GEN
+			  +'[journalmatch="SYSLOG_IDENTIFIER=\x01\x02dummy\x02\x01",'
+				+' failregex="^\x00\x01\x02dummy regex, never match <F-ID>xxx</F-ID>"]'
+		)
+		self.assertTrue(fail2banRegex.start(args))
+		self.assertLogged("'logtype': 'journal'")
+		self.assertNotLogged("'logtype': 'file'")
+		self.assertLogged('Lines: 0 lines, 0 ignored, 0 matched, 0 missed')
+		self.pruneLog()
+		# logtype specified explicitly (should win in filter):
+		(opts, args, fail2banRegex) = _Fail2banRegex(
+			"systemd-journal", Fail2banRegexTest.FILTER_ZZZ_GEN
+			  +'[logtype=file,'
+			  +' journalmatch="SYSLOG_IDENTIFIER=\x01\x02dummy\x02\x01",'
+				+' failregex="^\x00\x01\x02dummy regex, never match <F-ID>xxx</F-ID>"]'
+		)
+		self.assertTrue(fail2banRegex.start(args))
+		self.assertLogged("'logtype': 'file'")
+		self.assertNotLogged("'logtype': 'journal'")