NF: Add systemd journal backend

bin/fail2ban-regex

@@ -25,6 +25,12 @@ __license__ = "GPL"
 import getopt, sys, time, logging, os, locale
 from ConfigParser import NoOptionError, NoSectionError, MissingSectionHeaderError
 
+try:
+	from fail2ban.server.filtersystemd import FilterSystemd
+	from systemd import journal
+except:
+	journal = None
+
 from fail2ban.version import version
 from fail2ban.client.configparserinc import SafeConfigParserWithIncludes
 from fail2ban.server.filter import Filter
@@ -69,6 +75,7 @@ class Fail2banRegex:
 		self.__filter = Filter(None)
 		self.__ignoreregex = list()
 		self.__failregex = list()
+		self.__journalmatch = ""
 		self.__verbose = False
 		self.__maxlines_set = False		  # so we allow to override maxlines in cmdline
 		self.encoding = locale.getpreferredencoding()
@@ -111,10 +118,14 @@ class Fail2banRegex:
 		print "    -V, --version           print the version"
 		print "    -v, --verbose           verbose output"
 		print "    -l INT, --maxlines=INT  set maxlines for multi-line regex default: 1"
+		print "    -m MATCHES, --matches=MATCHES"
+		print "                            journalctl style matches, overriding filter file."
+		print "                            Special value \"ALL\" searches entire journal"
 		print
 		print "Log:"
 		print "    string                  a string representing a log line"
 		print "    filename                path to a log file (/var/log/auth.log)"
+		print "    \"systemd-journal\"       search systemd journal (systemd python required)"
 		print
 		print "Regex:"
 		print "    string                  a string representing a 'failregex'"
@@ -223,6 +234,10 @@ class Fail2banRegex:
 					print "ERROR: Invalid value for maxlines (%(maxlines)r) " \
 					      "read from %(value)s" % locals()
 					return False
+			try:
+				self.__journalmatch = reader.get("Init", "journalmatch")
+			except (NoSectionError, NoOptionError):
+				pass
 		else:
 			if len(value) > 53:
 				stripReg = value[0:50] + "..."
@@ -342,13 +357,16 @@ class Fail2banRegex:
 			print "information."
 			return True
 
+	def getJournalMatch(self):
+		return self.__journalmatch
 
 if __name__ == "__main__":
 	fail2banRegex = Fail2banRegex()
 	# Reads the command line options.
 	try:
-		cmdOpts = 'hVcvl:e:'
-		cmdLongOpts = ['help', 'version', 'verbose', 'maxlines=', 'encoding=']
+		cmdOpts = 'hVcvl:e:m:'
+		cmdLongOpts = ['help', 'version', 'verbose', 'maxlines=', 'encoding=',
+			'matches=']
 		optList, args = getopt.getopt(sys.argv[1:], cmdOpts, cmdLongOpts)
 	except getopt.GetoptError:
 		fail2banRegex.dispUsage()
@@ -391,6 +409,42 @@ if __name__ == "__main__":
 				print e
 				print
 				sys.exit(-1)
+		elif cmd_log == "systemd-journal":
+			if journal is None:
+				print "Error: systemd library not found. Exiting..."
+				sys.exit(-1)
+			myjournal = journal.Reader()
+			journalmatch = ""
+			# Parse journal matches from command line
+			for opt in optList:
+				if opt[0] in ["-m", "--matches"]:
+					journalmatch = opt[1]
+			# If no command line option, take journal match from filter
+			if not journalmatch:
+				journalmatch = fail2banRegex.getJournalMatch()
+			try:
+				if journalmatch != "ALL":
+					for element in journalmatch.split():
+						if element == "+":
+							myjournal.add_disjunction()
+						else:
+							myjournal.add_match(element)
+			except ValueError:
+				print "Error: Invalid journal match: %s" % journalmatch
+				print "Exiting..."
+				sys.exit(-1)
+			print "Use systemd journal match: %s" % (journalmatch or "ALL")
+			while True:
+				try:
+					entry = myjournal.get_next()
+				except OSError:
+					continue
+				else:
+					if not entry:
+						break
+				line = FilterSystemd.formatJournalEntry(entry)
+				fail2banRegex.testIgnoreRegex(line)
+				fail2banRegex.testRegex(line)
 		else:
 			if len(sys.argv[1]) > 53:
 				stripLog = cmd_log[0:50] + "..."

config/filter.d/dovecot.conf

@@ -21,3 +21,11 @@ failregex = .*(?:pop3-login|imap-login):.*(?:Authentication failure|Aborted logi
 # Values:  TEXT
 #
 ignoreregex = 
+
+[Init]
+
+# Option:  journalmatch
+# Notes.:  systemd journalctl style match filter for journal based backends
+# Values:  TEXT
+#
+journalmatch = _SYSTEMD_UNIT=dovecot.service

config/filter.d/postfix.conf

@@ -21,3 +21,11 @@ failregex = reject: RCPT from (.*)\[<HOST>\]: 554
 # Values:  TEXT
 #
 ignoreregex = 
+
+[Init]
+
+# Option:  journalmatch
+# Notes.:  systemd journalctl style match filter for journal based backends
+# Values:  TEXT
+#
+journalmatch = _SYSTEMD_UNIT=postfix.service

config/filter.d/recidive.conf

@@ -36,3 +36,11 @@ failregex = fail2ban.actions:\s+WARNING\s+\[(?:.*)\]\s+Ban\s+<HOST>
 #
 # Ignore our own bans, to keep our counts exact.
 ignoreregex = fail2ban.actions:\s+WARNING\s+\[%(_jailname)s\]\s+Ban\s+<HOST>
+
+[Init]
+
+# Option:  journalmatch
+# Notes.:  systemd journalctl style match filter for journal based backends
+# Values:  TEXT
+#
+journalmatch = _SYSTEMD_UNIT=fail2ban.service

config/filter.d/sshd-ddos.conf

@@ -34,3 +34,11 @@ failregex = ^%(__prefix_line)sDid not receive identification string from <HOST>\
 # Values:  TEXT
 #
 ignoreregex = 
+
+[Init]
+
+# Option:  journalmatch
+# Notes.:  systemd journalctl style match filter for journal based backend
+# Values:  TEXT
+#
+journalmatch = _SYSTEMD_UNIT=sshd.service + _COMM=sshd

config/filter.d/sshd.conf

@@ -39,3 +39,11 @@ failregex = ^%(__prefix_line)s(?:error: PAM: )?[aA]uthentication (?:failure|erro
 # Values:  TEXT
 #
 ignoreregex = 
+
+[Init]
+
+# Option:  journalmatch
+# Notes.:  systemd journalctl style match filter for journal based backend
+# Values:  TEXT
+#
+journalmatch = _SYSTEMD_UNIT=sshd.service + _COMM=sshd

config/jail.conf

@@ -42,7 +42,7 @@ findtime  = 600
 maxretry = 5
 
 # "backend" specifies the backend used to get files modification.
-# Available options are "pyinotify", "gamin", "polling" and "auto".
+# Available options are "pyinotify", "gamin", "polling", "systemd" and "auto".
 # This option can be overridden in each jail as well.
 #
 # pyinotify: requires pyinotify (a file alteration monitor) to be installed.
@@ -50,6 +50,9 @@ maxretry = 5
 # gamin:     requires Gamin (a file alteration monitor) to be installed.
 #              If Gamin is not installed, Fail2ban will use auto.
 # polling:   uses a polling algorithm which does not require external libraries.
+# systemd:   uses systemd python library to access the systemd journal.
+#              Specifying "logpath" is not valid for this backend.
+#              See "journalmatch" in the jails associated filter config
 # auto:      will try to use the following backends, in order:
 #              pyinotify, gamin, polling.
 backend = auto

fail2ban/client/beautifier.py

@@ -113,6 +113,12 @@ class Beautifier:
 			elif inC[2] == "logencoding":
 				msg = "Current log encoding is set to:\n"
 				msg = msg + response
+			elif inC[2] in ("journalmatch", "addjournalmatch", "deljournalmatch"):
+				if len(response) == 0:
+					msg = "No journal match filter set"
+				else:
+					msg = "Current match filter:\n"
+					msg += ' + '.join(response)
 			elif inC[2] in ("ignoreip", "addignoreip", "delignoreip"):
 				if len(response) == 0:
 					msg = "No IP address/network is ignored"

fail2ban/client/filterreader.py

@@ -56,5 +56,9 @@ class FilterReader(DefinitionInitConfigReader):
 		if self._initOpts:
 			if 'maxlines' in self._initOpts:
 				stream.append(["set", self._jailName, "maxlines", self._initOpts["maxlines"]])
+			# Do not send a command if the match is empty.
+			if self._initOpts.get("journalmatch", '') != '':
+				for match in self._initOpts["journalmatch"].split("\n"):
+					stream.append(["set", self._jailName, "addjournalmatch", match])
 		return stream
 		

fail2ban/protocol.py

@@ -55,6 +55,8 @@ protocol = [
 ["set <JAIL> addlogpath <FILE>", "adds <FILE> to the monitoring list of <JAIL>"], 
 ["set <JAIL> dellogpath <FILE>", "removes <FILE> from the monitoring list of <JAIL>"],
 ["set <JAIL> logencoding <ENCODING>", "sets the <ENCODING> of the log files for <JAIL>"],
+["set <JAIL> addjournalmatch <MATCH>", "adds <MATCH> to the journal filter of <JAIL>"],
+["set <JAIL> deljournalmatch <MATCH>", "removes <MATCH> from the journal filter of <JAIL>"],
 ["set <JAIL> addfailregex <REGEX>", "adds the regular expression <REGEX> which must match failures for <JAIL>"], 
 ["set <JAIL> delfailregex <INDEX>", "removes the regular expression at <INDEX> for failregex"], 
 ["set <JAIL> addignoreregex <REGEX>", "adds the regular expression <REGEX> which should match pattern to exclude for <JAIL>"],
@@ -79,6 +81,7 @@ protocol = [
 ['', "JAIL INFORMATION", ""],
 ["get <JAIL> logpath", "gets the list of the monitored files for <JAIL>"],
 ["get <JAIL> logencoding <ENCODING>", "gets the <ENCODING> of the log files for <JAIL>"],
+["get <JAIL> journalmatch", "gets the journal filter match for <JAIL>"],
 ["get <JAIL> ignoreip", "gets the list of ignored IP addresses for <JAIL>"],
 ["get <JAIL> failregex", "gets the list of regular expressions which matches the failures for <JAIL>"],
 ["get <JAIL> ignoreregex", "gets the list of regular expressions which matches patterns to ignore for <JAIL>"],

fail2ban/server/filter.py

@@ -643,6 +643,21 @@ class FileContainer:
 			self.__handler = None
 
 
+##
+# JournalFilter class.
+#
+# Base interface class for systemd journal filters
+
+class JournalFilter(Filter):
+
+	def addJournalMatch(self, match):
+		pass
+
+	def delJournalMatch(self, match):
+		pass
+
+	def getJournalMatch(self, match):
+		return []
 
 ##
 # Utils class for DNS and IP handling.

fail2ban/server/filtersystemd.py

@@ -0,0 +1,237 @@
+# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: t -*-
+# vi: set ft=python sts=4 ts=4 sw=4 noet :
+
+# This file is part of Fail2Ban.
+#
+# Fail2Ban is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# Fail2Ban is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Fail2Ban; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+
+# Original author: Cyril Jaquier
+
+__author__ = "Cyril Jaquier, Lee Clemens, Yaroslav Halchenko, Steven Hiscocks"
+__copyright__ = "Copyright (c) 2004 Cyril Jaquier, 2011-2012 Lee Clemens, 2012 Yaroslav Halchenko, 2013 Steven Hiscocks"
+__license__ = "GPL"
+
+import logging, datetime
+from distutils.version import LooseVersion
+
+from systemd import journal
+if LooseVersion(getattr(journal, '__version__', "0")) < '204':
+	raise ImportError("Fail2Ban requires systemd >= 204")
+
+from failmanager import FailManagerEmpty
+from filter import JournalFilter
+from mytime import MyTime
+
+
+# Gets the instance of the logger.
+logSys = logging.getLogger("fail2ban.filter")
+
+##
+# Journal reader class.
+#
+# This class reads from systemd journal and detects login failures or anything
+# else that matches a given regular expression. This class is instantiated by
+# a Jail object.
+
+class FilterSystemd(JournalFilter):
+	##
+	# Constructor.
+	#
+	# Initialize the filter object with default values.
+	# @param jail the jail object
+
+	def __init__(self, jail, **kwargs):
+		JournalFilter.__init__(self, jail, **kwargs)
+		self.__modified = False
+		# Initialise systemd-journal connection
+		self.__journal = journal.Reader(converters={'__CURSOR': lambda x: x})
+		self.__matches = []
+		logSys.debug("Created FilterSystemd")
+
+	##
+	# Add a journal match filter
+	#
+	# @param match journalctl syntax matches
+
+	def addJournalMatch(self, match):
+		if self.__matches:
+			self.__journal.add_disjunction() # Add OR
+		try:
+			for match_element in match.split():
+				if match_element == "+":
+					self.__journal.add_disjunction()
+				else:
+					self.__journal.add_match(match_element)
+		except:
+			logSys.error("Error adding journal match for: %s", match)
+			self.resetJournalMatches()
+		else:
+			for match_element in match.split('+'):
+				self.__matches.append(match_element.strip())
+			logSys.debug("Adding journal match for: %s", match)
+	##
+	# Reset a journal match filter called on removal or failure
+	#
+	# @return None 
+
+	def resetJournalMatches(self):
+		self.__journal.flush_matches()
+		logSys.debug("Flushed all journal matches")
+		match_copy = self.__matches[:]
+		self.__matches = []
+		for match in match_copy:
+			self.addJournalMatch(match)
+
+	##
+	# Delete a journal match filter
+	#
+	# @param match journalctl syntax matches
+
+	def delJournalMatch(self, match):
+		if match in self.__matches:
+			del self.__matches[self.__matches.index(match)]
+			self.resetJournalMatches()
+
+	##
+	# Get current journal match filter
+	#
+	# @return journalctl syntax matches
+
+	def getJournalMatch(self):
+		return self.__matches
+
+    ##
+    # Join group of log elements which may be a mix of bytes and strings
+    #
+    # @param elements list of strings and bytes
+    # @return elements joined as string
+
+	@staticmethod
+	def _joinStrAndBytes(elements):
+		strElements = []
+		for element in elements:
+			if isinstance(element, str):
+				strElements.append(element)
+			else:
+				strElements.append(str(element, errors='ignore'))
+		return " ".join(strElements)
+
+	##
+	# Format journal log entry into syslog style
+	#
+	# @param entry systemd journal entry dict
+	# @return format log line
+
+	@staticmethod
+	def formatJournalEntry(logentry):
+		logelements = [logentry.get('_SOURCE_REALTIME_TIMESTAMP',
+			logentry.get('__REALTIME_TIMESTAMP')).strftime("%b %d %H:%M:%S %Y")]
+		if logentry.get('_HOSTNAME'):
+			logelements.append(logentry['_HOSTNAME'])
+		if logentry.get('SYSLOG_IDENTIFIER'):
+			logelements.append(logentry['SYSLOG_IDENTIFIER'])
+			if logentry.get('_PID'):
+				logelements[-1] += ("[%i]" % logentry['_PID'])
+			logelements[-1] += ":"
+		elif logentry.get('_COMM'):
+			logelements.append(logentry['_COMM'])
+			if logentry.get('_PID'):
+				logelements[-1] += ("[%i]" % logentry['_PID'])
+			logelements[-1] += ":"
+		if logelements[-1] == "kernel:":
+			if '_SOURCE_MONOTONIC_TIMESTAMP' in logentry:
+				monotonic = logentry.get('_SOURCE_MONOTONIC_TIMESTAMP')
+			else:
+				monotonic = logentry.get('__MONOTONIC_TIMESTAMP')[0]
+			logelements.append("[%12.6f]" % monotonic.total_seconds())
+		if isinstance(logentry.get('MESSAGE',''), list):
+			logelements.append(" ".join(logentry['MESSAGE']))
+		else:
+			logelements.append(logentry.get('MESSAGE', ''))
+
+		try:
+			logline = u" ".join(logelements) + u"\n"
+		except UnicodeDecodeError:
+			# Python 2, so treat as string
+			logline = " ".join([str(logline) for logline in logelements]) + "\n"
+		except TypeError:
+			# Python 3, one or more elements bytes
+			logSys.warning("Error decoding log elements from journal: %s" %
+				repr(logelements))
+			logline =  self._joinStrAndBytes(logelements) + "\n"
+
+		logSys.debug("Read systemd journal entry: %s" % repr(logline))
+		return logline
+
+	##
+	# Main loop.
+	#
+	# Peridocily check for new journal entries matching the filter and
+	# handover to FailManager
+
+	def run(self):
+		self.setActive(True)
+
+		# Seek to now - findtime in journal
+		start_time = datetime.datetime.now() - \
+				datetime.timedelta(seconds=int(self.getFindTime()))
+		self.__journal.seek_realtime(start_time)
+		# Move back one entry to ensure do not end up in dead space
+		# if start time beyond end of journal
+		try:
+			self.__journal.get_previous()
+		except OSError:
+			pass # Reading failure, so safe to ignore
+
+		while self._isActive():
+			if not self.getIdle():
+				while self._isActive():
+					try:
+						logentry = self.__journal.get_next()
+					except OSError:
+						logSys.warning(
+							"Error reading line from systemd journal")
+						continue
+					if logentry:
+						self.processLineAndAdd(
+							self.formatJournalEntry(logentry))
+						self.__modified = True
+					else:
+						break
+				if self.__modified:
+					try:
+						while True:
+							ticket = self.failManager.toBan()
+							self.jail.putFailTicket(ticket)
+					except FailManagerEmpty:
+						self.failManager.cleanup(MyTime.time())
+					self.dateDetector.sortTemplate()
+					self.__modified = False
+			self.__journal.wait(self.getSleepTime())
+		logSys.debug((self.jail is not None and self.jail.getName()
+                      or "jailless") +" filter terminated")
+		return True
+
+    ##
+    # Get the status of the filter.
+    #           
+    # Get some informations about the filter state such as the total
+    # number of failures.
+    # @return a list with tuple
+
+	def status(self):
+		ret = JournalFilter.status(self)
+		ret.append(("Journal matches", [" + ".join(self.__matches)]))
+		return ret

fail2ban/server/jail.py

@@ -35,7 +35,7 @@ class Jail:
 	#Known backends. Each backend should have corresponding __initBackend method
 	# yoh: stored in a list instead of a tuple since only
 	#      list had .index until 2.6
-	_BACKENDS = ['pyinotify', 'gamin', 'polling']
+	_BACKENDS = ['pyinotify', 'gamin', 'polling', 'systemd']
 
 	def __init__(self, name, backend = "auto"):
 		self.__name = name
@@ -101,6 +101,13 @@ class Jail:
 		from filterpyinotify import FilterPyinotify
 		self.__filter = FilterPyinotify(self)
 	
+	def _initSystemd(self):
+		# Try to import systemd
+		import systemd
+		logSys.info("Jail '%s' uses systemd" % self.__name)
+		from filtersystemd import FilterSystemd
+		self.__filter = FilterSystemd(self)
+	
 	def setName(self, name):
 		self.__name = name
 	

fail2ban/server/server.py

@@ -26,6 +26,7 @@ __license__ = "GPL"
 
 from threading import Lock, RLock
 from jails import Jails
+from filter import FileFilter, JournalFilter
 from transmitter import Transmitter
 from asyncserver import AsyncServer
 from asyncserver import AsyncServerException
@@ -169,14 +170,41 @@ class Server:
 		return self.__jails.getFilter(name).getIgnoreIP()
 	
 	def addLogPath(self, name, fileName):
-		self.__jails.getFilter(name).addLogPath(fileName)
+		filter_ = self.__jails.getFilter(name)
+		if isinstance(filter_, FileFilter):
+			filter_.addLogPath(fileName)
 	
 	def delLogPath(self, name, fileName):
-		self.__jails.getFilter(name).delLogPath(fileName)
+		filter_ = self.__jails.getFilter(name)
+		if isinstance(filter_, FileFilter):
+			self.__jails.getFilter(name).delLogPath(fileName)
 	
 	def getLogPath(self, name):
-		return [m.getFileName()
-				for m in self.__jails.getFilter(name).getLogPath()]
+		filter_ = self.__jails.getFilter(name)
+		if isinstance(filter_, FileFilter):
+			return [m.getFileName()
+					for m in filter_.getLogPath()]
+		else:
+			logSys.info("Jail %s is not a FileFilter instance" % name)
+			return []
+	
+	def addJournalMatch(self, name, match):
+		filter_ = self.__jails.getFilter(name)
+		if isinstance(filter_, JournalFilter):
+			filter_.addJournalMatch(match)
+	
+	def delJournalMatch(self, name, match):
+		filter_ = self.__jails.getFilter(name)
+		if isinstance(filter_, JournalFilter):
+			filter_.delJournalMatch(match)
+	
+	def getJournalMatch(self, name):
+		filter_ = self.__jails.getFilter(name)
+		if isinstance(filter_, JournalFilter):
+			return filter_.getJournalMatch()
+		else:
+			logSys.info("Jail %s is not a JournalFilter instance" % name)
+			return []
 	
 	def setLogEncoding(self, name, encoding):
 		return self.__jails.getFilter(name).setLogEncoding(encoding)

fail2ban/server/transmitter.py

@@ -144,6 +144,14 @@ class Transmitter:
 			value = command[2]
 			self.__server.setLogEncoding(name, value)
 			return self.__server.getLogEncoding(name)
+		elif command[1] == "addjournalmatch":
+			value = ' '.join(command[2:])
+			self.__server.addJournalMatch(name, value)
+			return self.__server.getJournalMatch(name)
+		elif command[1] == "deljournalmatch":
+			value = ' '.join(command[2:])
+			self.__server.delJournalMatch(name, value)
+			return self.__server.getJournalMatch(name)
 		elif command[1] == "addfailregex":
 			value = command[2]
 			self.__server.addFailRegex(name, value)
@@ -250,6 +258,8 @@ class Transmitter:
 			return self.__server.getLogPath(name)
 		elif command[1] == "logencoding":
 			return self.__server.getLogEncoding(name)
+		elif command[1] == "journalmatch":
+			return self.__server.getJournalMatch(name)
 		elif command[1] == "ignoreip":
 			return self.__server.getIgnoreIP(name)
 		elif command[1] == "failregex":

fail2ban/tests/files/testcase-journal.log

@@ -0,0 +1,19 @@
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from failed.dns.ch
+error: PAM: Authentication failure for kevin from failed.dns.ch
+error: PAM: Authentication failure for kevin from failed.dns.ch
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 193.168.0.128
+error: PAM: Authentication failure for kevin from 87.142.124.10
+error: PAM: Authentication failure for kevin from 87.142.124.10
+error: PAM: Authentication failure for kevin from 87.142.124.10
+error: PAM: Authentication failure for kevin from 87.142.124.10

fail2ban/tests/filtertestcase.py

@@ -29,6 +29,11 @@ import sys
 import time
 import tempfile
 
+try:
+	from systemd import journal
+except ImportError:
+	journal = None
+
 from fail2ban.server.jail import Jail
 from fail2ban.server.filterpoll import FilterPoll
 from fail2ban.server.filter import FileFilter, DNSUtils
@@ -160,6 +165,34 @@ def _copy_lines_between_files(in_, fout, n=None, skip=0, mode='a', terminal_line
 	time.sleep(0.1)
 	return fout
 
+def _copy_lines_to_journal(in_, fields={},n=None, skip=0, terminal_line=""):
+	"""Copy lines from one file to systemd journal
+
+	Returns None
+	"""
+	if isinstance(in_, str): # pragma: no branch - only used with str in test cases
+		fin = open(in_, 'r')
+	else:
+		fin = in_
+	# Required for filtering
+	fields.update({"SYSLOG_IDENTIFIER": "fail2ban-testcases",
+					"PRIORITY": "7",
+					})
+	# Skip
+	for i in xrange(skip):
+		_ = fin.readline()
+	# Read/Write
+	i = 0
+	while n is None or i < n:
+		l = fin.readline()
+		if terminal_line is not None and l == terminal_line:
+			break
+		journal.send(MESSAGE=l.strip(), **fields)
+		i += 1
+	if isinstance(in_, str): # pragma: no branch - only used with str in test cases
+		# Opened earlier, therefore must close it
+		fin.close()
+
 #
 #  Actual tests
 #
@@ -574,6 +607,129 @@ def get_monitor_failures_testcase(Filter_):
 			  % (Filter_.__name__, testclass_name) # 'tempfile')
 	return MonitorFailures
 
+def get_monitor_failures_journal_testcase(Filter_):
+	"""Generator of TestCase's for journal based filters/backends
+	"""
+
+	class MonitorJournalFailures(unittest.TestCase):
+		def setUp(self):
+			"""Call before every test case."""
+			self.test_file = os.path.join(TEST_FILES_DIR, "testcase-journal.log")
+			self.jail = DummyJail()
+			self.filter = Filter_(self.jail)
+			# UUID used to ensure that only meeages generated
+			# as part of this test are picked up by the filter
+			import uuid
+			self.test_uuid = str(uuid.uuid4())
+			self.name = "monitorjournalfailures-%s" % self.test_uuid
+			self.filter.addJournalMatch(
+				"SYSLOG_IDENTIFIER=fail2ban-testcases "
+				"TEST_FIELD=1 "
+				"TEST_UUID=%s" % str(self.test_uuid))
+			self.filter.addJournalMatch(
+				"SYSLOG_IDENTIFIER=fail2ban-testcases "
+				"TEST_FIELD=2 "
+				"TEST_UUID=%s" % self.test_uuid)
+			self.journal_fields = {
+				'TEST_FIELD': "1", 'TEST_UUID': self.test_uuid}
+			self.filter.setActive(True)
+			self.filter.addFailRegex("(?:(?:Authentication failure|Failed [-/\w+]+) for(?: [iI](?:llegal|nvalid) user)?|[Ii](?:llegal|nvalid) user|ROOT LOGIN REFUSED) .*(?: from|FROM) <HOST>")
+			self.filter.start()
+
+		def tearDown(self):
+			self.filter.stop()
+			self.filter.join()		  # wait for the thread to terminate
+			pass
+
+		def __str__(self):
+			return "MonitorJournalFailures%s(%s)" \
+			  % (Filter_, hasattr(self, 'name') and self.name or 'tempfile')
+
+		def isFilled(self, delay=2.):
+			"""Wait up to `delay` sec to assure that it was modified or not
+			"""
+			time0 = time.time()
+			while time.time() < time0 + delay:
+				if len(self.jail):
+					return True
+				time.sleep(0.1)
+			return False
+
+		def isEmpty(self, delay=0.4):
+			# shorter wait time for not modified status
+			return not self.isFilled(delay)
+
+		def assert_correct_ban(self, test_ip, test_attempts):
+			self.assertTrue(self.isFilled(10)) # give Filter a chance to react
+			ticket = self.jail.getFailTicket()
+
+			attempts = ticket.getAttempt()
+			ip = ticket.getIP()
+			matches = ticket.getMatches()
+
+			self.assertEqual(ip, test_ip)
+			self.assertEqual(attempts, test_attempts)
+
+		def test_grow_file(self):
+			self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
+
+			# Now let's feed it with entries from the file
+			_copy_lines_to_journal(
+				self.test_file, self.journal_fields, n=2)
+			self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
+			# and our dummy jail is empty as well
+			self.assertFalse(len(self.jail))
+			# since it should have not been enough
+
+			_copy_lines_to_journal(
+				self.test_file, self.journal_fields, skip=2, n=3)
+			self.assertTrue(self.isFilled(6))
+			# so we sleep for up to 6 sec for it not to become empty,
+			# and meanwhile pass to other thread(s) and filter should
+			# have gathered new failures and passed them into the
+			# DummyJail
+			self.assertEqual(len(self.jail), 1)
+			# and there should be no "stuck" ticket in failManager
+			self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
+			self.assert_correct_ban("193.168.0.128", 3)
+			self.assertEqual(len(self.jail), 0)
+
+			# Lets read some more to check it bans again
+			_copy_lines_to_journal(
+				self.test_file, self.journal_fields, skip=5, n=4)
+			self.assert_correct_ban("193.168.0.128", 3)
+
+		def test_delJournalMatch(self):
+			# Smoke test for removing of match
+
+			# basic full test
+			_copy_lines_to_journal(
+				self.test_file, self.journal_fields, n=5)
+			self.assert_correct_ban("193.168.0.128", 3)
+
+			# and now remove the JournalMatch
+			self.filter.delJournalMatch(
+				"SYSLOG_IDENTIFIER=fail2ban-testcases "
+				"TEST_FIELD=1 "
+				"TEST_UUID=%s" % str(self.test_uuid))
+
+			_copy_lines_to_journal(
+				self.test_file, self.journal_fields, n=5, skip=5)
+			# so we should get no more failures detected
+			self.assertTrue(self.isEmpty(2))
+
+			# but then if we add it back again
+			self.filter.addJournalMatch(
+				"SYSLOG_IDENTIFIER=fail2ban-testcases "
+				"TEST_FIELD=1 "
+				"TEST_UUID=%s" % str(self.test_uuid))
+			self.assert_correct_ban("193.168.0.128", 4)
+			_copy_lines_to_journal(
+				self.test_file, self.journal_fields, n=6, skip=10)
+			# we should detect the failures
+			self.assertTrue(self.isFilled(6))
+
+	return MonitorJournalFailures
 
 class GetFailures(unittest.TestCase):
 

fail2ban/tests/utils.py

@@ -201,6 +201,12 @@ def gatherTests(regexps=None, no_network=False):
 	for Filter_ in filters:
 		tests.addTest(unittest.makeSuite(
 			filtertestcase.get_monitor_failures_testcase(Filter_)))
+	try:
+		from fail2ban.server.filtersystemd import FilterSystemd
+		tests.addTest(unittest.makeSuite(filtertestcase.get_monitor_failures_journal_testcase(FilterSystemd)))
+	except Exception, e: # pragma: no cover
+		logSys.warning("I: Skipping systemd backend testing. Got exception '%s'" % e)
+
 
 	# Server test for logging elements which break logging used to support
 	# testcases analysis