Merge branch '0.10' into 0.11, rewrite updateDb because it can be executed after repair, and some tables can be missing.

# Conflicts:
#	fail2ban/server/database.py
#	fail2ban/tests/fail2banclienttestcase.py
#	fail2ban/tests/sockettestcase.py

ChangeLog

@@ -76,6 +76,9 @@ ver. 0.11.0-dev-0 (2017/??/??) - development nightly edition
   - `datetime` - add date-time to the message (default on, ignored if `format` specified);
   - `format` - specify own format how it will be logged, for example for short-log into STDOUT:
       `fail2ban-server -f --logtarget 'stdout[format="%(relativeCreated)5d | %(message)s"]' start`;
+* Automatically recover or recreate corrupt persistent database (e. g. if failed to open with 
+  'database disk image is malformed'). Fail2ban will create a backup, try to repair the database,
+  if repair fails - recreate new database (gh-1465, gh-2004).
 
 
 ver. 0.10.1 (2017/10/12) - succeeded-before-friday-the-13th

config/action.d/firewallcmd-ipset.conf

@@ -21,8 +21,10 @@ before = firewallcmd-common.conf
 actionstart = ipset create <ipmset> hash:ip timeout <default-timeout><familyopt>
               firewall-cmd --direct --add-rule <family> filter <chain> 0 -p <protocol> -m multiport --dports <port> -m set --match-set <ipmset> src -j <blocktype>
 
+actionflush = ipset flush <ipmset>
+
 actionstop = firewall-cmd --direct --remove-rule <family> filter <chain> 0 -p <protocol> -m multiport --dports <port> -m set --match-set <ipmset> src -j <blocktype>
-             ipset flush <ipmset>
+             <actionflush>
              ipset destroy <ipmset>
 
 actionban = ipset add <ipmset> <ip> timeout <bantime> -exist

config/action.d/firewallcmd-new.conf

@@ -16,7 +16,7 @@ actionstop = firewall-cmd --direct --remove-rule <family> filter <chain> 0 -m st
              firewall-cmd --direct --remove-rules <family> filter f2b-<name>
              firewall-cmd --direct --remove-chain <family> filter f2b-<name>
 
-actioncheck = firewall-cmd --direct --get-chains <family> filter | grep -q 'f2b-<name>$'
+actioncheck = firewall-cmd --direct --get-chains <family> filter | sed -e 's, ,\n,g' | grep -q 'f2b-<name>$'
 
 actionban = firewall-cmd --direct --add-rule <family> filter f2b-<name> 0 -s <ip> -j <blocktype>
 

fail2ban/server/actions.py

@@ -277,8 +277,10 @@ class Actions(JailThread, Mapping):
 					exc_info=logSys.getEffectiveLevel()<=logging.DEBUG)
 		while self.active:
 			if self.idle:
+				logSys.debug("Actions: enter idle mode")
 				Utils.wait_for(lambda: not self.active or not self.idle,
-					self.sleeptime * 10, self.sleeptime)
+					lambda: False, self.sleeptime)
+				logSys.debug("Actions: leave idle mode")
 				continue
 			if not Utils.wait_for(lambda: not self.active or self.__checkBan(), self.sleeptime):
 				self.__checkUnBan()

fail2ban/server/asyncserver.py

@@ -122,7 +122,9 @@ def loop(active, timeout=None, use_poll=False):
 	if timeout is None:
 		timeout = Utils.DEFAULT_SLEEP_TIME
 	poll = asyncore.poll
-	if use_poll and asyncore.poll2 and hasattr(asyncore.select, 'poll'): # pragma: no cover
+	if callable(use_poll):
+		poll = use_poll
+	elif use_poll and asyncore.poll2 and hasattr(asyncore.select, 'poll'): # pragma: no cover
 		logSys.debug('Server listener (select) uses poll')
 		# poll2 expected a timeout in milliseconds (but poll and loop in seconds):
 		timeout = float(timeout) / 1000

fail2ban/server/database.py

@@ -22,6 +22,7 @@ __copyright__ = "Copyright (c) 2013 Steven Hiscocks"
 __license__ = "GPL"
 
 import json
+import os
 import shutil
 import sqlite3
 import sys
@@ -31,6 +32,7 @@ from threading import RLock
 
 from .mytime import MyTime
 from .ticket import FailTicket
+from .utils import Utils
 from ..helpers import getLogger, PREFER_ENC
 
 # Gets the instance of the logger.
@@ -127,14 +129,15 @@ class Fail2BanDb(object):
 	purgeage
 	"""
 	__version__ = 4
-	# Note all _TABLE_* strings must end in ';' for py26 compatibility
+	# Note all SCRIPTS strings must end in ';' for py26 compatibility
-	_TABLE_fail2banDb = "CREATE TABLE fail2banDb(version INTEGER);"
+	_CREATE_SCRIPTS = (
-	_TABLE_jails = "CREATE TABLE jails(" \
+		 ('fail2banDb', "CREATE TABLE IF NOT EXISTS fail2banDb(version INTEGER);")
+		,('jails', "CREATE TABLE IF NOT EXISTS jails(" \
 			"name TEXT NOT NULL UNIQUE, " \
 			"enabled INTEGER NOT NULL DEFAULT 1" \
 			");" \
-			"CREATE INDEX jails_name ON jails(name);"
+			"CREATE INDEX IF NOT EXISTS jails_name ON jails(name);")
-	_TABLE_logs = "CREATE TABLE logs(" \
+		,('logs', "CREATE TABLE IF NOT EXISTS logs(" \
 			"jail TEXT NOT NULL, " \
 			"path TEXT, " \
 			"firstlinemd5 TEXT, " \
@@ -143,13 +146,13 @@ class Fail2BanDb(object):
 			"UNIQUE(jail, path)," \
 			"UNIQUE(jail, path, firstlinemd5)" \
 			");" \
-			"CREATE INDEX logs_path ON logs(path);" \
+			"CREATE INDEX IF NOT EXISTS logs_path ON logs(path);" \
-			"CREATE INDEX logs_jail_path ON logs(jail, path);"
+			"CREATE INDEX IF NOT EXISTS logs_jail_path ON logs(jail, path);")
 			#TODO: systemd journal features \
 			#"journalmatch TEXT, " \
 			#"journlcursor TEXT, " \
-			#"lastfiletime INTEGER DEFAULT 0, " # is this easily available \
+			#"lastfiletime INTEGER DEFAULT 0, " # is this easily available
-	_TABLE_bans = "CREATE TABLE bans(" \
+		,('bans', "CREATE TABLE IF NOT EXISTS bans(" \
 			"jail TEXT NOT NULL, " \
 			"ip TEXT, " \
 			"timeofban INTEGER NOT NULL, " \
@@ -158,11 +161,10 @@ class Fail2BanDb(object):
 			"data JSON, " \
 			"FOREIGN KEY(jail) REFERENCES jails(name) " \
 			");" \
-			"CREATE INDEX bans_jail_timeofban_ip ON bans(jail, timeofban);" \
+			"CREATE INDEX IF NOT EXISTS bans_jail_timeofban_ip ON bans(jail, timeofban);" \
-			"CREATE INDEX bans_jail_ip ON bans(jail, ip);" \
+			"CREATE INDEX IF NOT EXISTS bans_jail_ip ON bans(jail, ip);" \
-			"CREATE INDEX bans_ip ON bans(ip);" \
+			"CREATE INDEX IF NOT EXISTS bans_ip ON bans(ip);")
-
+		,('bips', "CREATE TABLE IF NOT EXISTS bips(" \
-	_TABLE_bips = "CREATE TABLE bips(" \
 			"ip TEXT NOT NULL, " \
 			"jail TEXT NOT NULL, " \
 			"timeofban INTEGER NOT NULL, " \
@@ -172,20 +174,26 @@ class Fail2BanDb(object):
 			"PRIMARY KEY(ip, jail), " \
 			"FOREIGN KEY(jail) REFERENCES jails(name) " \
 			");" \
-			"CREATE INDEX bips_timeofban ON bips(timeofban);" \
+			"CREATE INDEX IF NOT EXISTS bips_timeofban ON bips(timeofban);" \
-			"CREATE INDEX bips_ip ON bips(ip);" \
+			"CREATE INDEX IF NOT EXISTS bips_ip ON bips(ip);")
+	)
+	_CREATE_TABS = dict(_CREATE_SCRIPTS)
 
 
 	def __init__(self, filename, purgeAge=24*60*60, outDatedFactor=3):
 		self.maxEntries = 50
-		try:
 		self._lock = RLock()
-			self._db = sqlite3.connect(
-				filename, check_same_thread=False,
-				detect_types=sqlite3.PARSE_DECLTYPES)
 		self._dbFilename = filename
 		self._purgeAge = purgeAge
 		self._outDatedFactor = outDatedFactor;
+		self._connectDB()
+
+	def _connectDB(self, checkIntegrity=False):
+		filename = self._dbFilename
+		try:
+			self._db = sqlite3.connect(
+				filename, check_same_thread=False,
+				detect_types=sqlite3.PARSE_DECLTYPES)
 
 			self._bansMergedCache = {}
 
@@ -206,6 +214,7 @@ class Fail2BanDb(object):
 			pypy = False
 
 		cur = self._db.cursor()
+		try:
 			cur.execute("PRAGMA foreign_keys = ON")
 			# speedup: write data through OS without syncing (no wait):
 			cur.execute("PRAGMA synchronous = OFF")
@@ -215,24 +224,44 @@ class Fail2BanDb(object):
 			# speedup: temporary tables and indices are kept in memory:
 			cur.execute("PRAGMA temp_store = MEMORY")
 
-		try:
 			cur.execute("SELECT version FROM fail2banDb LIMIT 1")
 		except sqlite3.OperationalError:
-			logSys.warning("New database created. Version '%i'",
+			logSys.warning("New database created. Version '%r'",
 				self.createDb())
+		except sqlite3.Error as e:
+			logSys.error(
+				"Error opening fail2ban persistent database '%s': %s",
+				filename, e.args[0])
+			# if not a file - raise an error:
+			if not os.path.isfile(filename):
+				raise
+			# try to repair it:
+			cur.close()
+			cur = None
+			self.repairDB()
 		else:
 			version = cur.fetchone()[0]
 			if version < Fail2BanDb.__version__:
 				newversion = self.updateDb(version)
 				if newversion == Fail2BanDb.__version__:
-					logSys.warning( "Database updated from '%i' to '%i'",
+					logSys.warning( "Database updated from '%r' to '%r'",
 						version, newversion)
 				else: # pragma: no cover
-					logSys.error( "Database update failed to achieve version '%i'"
+					logSys.error( "Database update failed to achieve version '%r'"
-						": updated from '%i' to '%i'",
+						": updated from '%r' to '%r'",
 						Fail2BanDb.__version__, version, newversion)
 					raise RuntimeError('Failed to fully update')
 		finally:
+			if checkIntegrity:
+				logSys.debug("  Create missing tables/indices ...")
+				self._createDb(cur, incremental=True)
+				logSys.debug("  -> ok")
+				logSys.debug("  Check integrity ...")
+				cur.execute("PRAGMA integrity_check")
+				for s in cur.fetchall():
+					logSys.debug("  -> %s", ' '.join(s))
+				self._db.commit()
+			if cur:
 				# pypy: set journal mode after possible upgrade db:
 				if pypy:
 					cur.execute("PRAGMA journal_mode = MEMORY")
@@ -243,6 +272,47 @@ class Fail2BanDb(object):
 		self._db.close()
 		logSys.info("Connection to database closed.")
 
+	@property
+	def _dbBackupFilename(self):
+		try:
+			return self.__dbBackupFilename
+		except AttributeError:
+			self.__dbBackupFilename = self._dbFilename + '.' + time.strftime('%Y%m%d-%H%M%S', MyTime.gmtime())
+			return self.__dbBackupFilename
+	
+	def repairDB(self):
+		class RepairException(Exception):
+			pass
+		# avoid endless recursion if reconnect failed again for some reasons:
+		_repairDB = self.repairDB
+		self.repairDB = None
+		try:
+			# backup
+			logSys.info("Trying to repair database %s", self._dbFilename)
+			shutil.move(self._dbFilename, self._dbBackupFilename)
+			logSys.info("  Database backup created: %s", self._dbBackupFilename)
+
+			# first try to repair using dump/restore in order 
+			Utils.executeCmd((r"""f2b_db=$0; f2b_dbbk=$1; sqlite3 "$f2b_dbbk" ".dump" | sqlite3 "$f2b_db" """,
+				self._dbFilename, self._dbBackupFilename))
+			dbFileSize = os.stat(self._dbFilename).st_size
+			if dbFileSize:
+				logSys.info("  Repair seems to be successful, restored %d byte(s).", dbFileSize)
+				# succeeded - try to reconnect:
+				self._connectDB(checkIntegrity=True)
+			else:
+				logSys.info("  Repair seems to be failed, restored %d byte(s).", dbFileSize)
+				raise RepairException('Recreate ...')
+		except Exception as e:
+			# if still failed, just recreate database as fallback:
+			logSys.error("  Error repairing of fail2ban database '%s': %s",
+				self._dbFilename, e.args[0], 
+				exc_info=(not isinstance(e, RepairException) and logSys.getEffectiveLevel() <= 10))
+			os.remove(self._dbFilename)
+			self._connectDB(checkIntegrity=True)
+		finally:
+			self.repairDB = _repairDB
+
 	@property
 	def filename(self):
 		"""File name of SQLite3 database file.
@@ -259,26 +329,29 @@ class Fail2BanDb(object):
 	def purgeage(self, value):
 		self._purgeAge = MyTime.str2seconds(value)
 
-	@commitandrollback
+	def _createDb(self, cur, incremental=False):
-	def createDb(self, cur):
 		"""Creates a new database, called during initialisation.
 		"""
-		# Version info
+		# create all (if not exists):
-		cur.executescript(Fail2BanDb._TABLE_fail2banDb)
+		for (n, s) in Fail2BanDb._CREATE_SCRIPTS:
-		cur.execute("INSERT INTO fail2banDb(version) VALUES(?)",
+			cur.executescript(s)
+		# save current database version (if not already set):			
+		cur.execute("INSERT INTO fail2banDb(version)"
+			" SELECT ? WHERE NOT EXISTS (SELECT 1 FROM fail2banDb LIMIT 1)",
 			(Fail2BanDb.__version__, ))
-		# Jails
-		cur.executescript(Fail2BanDb._TABLE_jails)
-		# Logs
-		cur.executescript(Fail2BanDb._TABLE_logs)
-		# Bans
-		cur.executescript(Fail2BanDb._TABLE_bans)
-		# BIPs (bad ips)
-		cur.executescript(Fail2BanDb._TABLE_bips)
-
 		cur.execute("SELECT version FROM fail2banDb LIMIT 1")
 		return cur.fetchone()[0]
 
+	@commitandrollback
+	def createDb(self, cur, incremental=False):
+		return self._createDb(cur, incremental);
+
+	def _tableExists(self, cur, table):
+		cur.execute("select 1 where exists ("
+			"select 1 from sqlite_master WHERE type='table' AND name=?)", (table,))
+		res = cur.fetchone()
+		return res is not None and res[0]
+
 	@commitandrollback
 	def updateDb(self, cur, version):
 		"""Update an existing database, called during initialisation.
@@ -288,12 +361,13 @@ class Fail2BanDb(object):
 		if version > Fail2BanDb.__version__:
 			raise NotImplementedError(
 						"Attempt to travel to future version of database ...how did you get here??")
-
+		try:
-		self._dbBackupFilename = self.filename + '.' + time.strftime('%Y%m%d-%H%M%S', MyTime.gmtime())
+			logSys.info("Upgrade database: %s from version '%r'", self._dbBackupFilename, version)
+			if not os.path.isfile(self._dbBackupFilename):
 				shutil.copyfile(self.filename, self._dbBackupFilename)
-		logSys.info("Database backup created: %s", self._dbBackupFilename)
+				logSys.info("  Database backup created: %s", self._dbBackupFilename)
 
-		if version < 2:
+			if version < 2 and self._tableExists(cur, "logs"):
 				cur.executescript("BEGIN TRANSACTION;"
 							"CREATE TEMPORARY TABLE logs_temp AS SELECT * FROM logs;"
 							"DROP TABLE logs;"
@@ -301,24 +375,30 @@ class Fail2BanDb(object):
 							"INSERT INTO logs SELECT * from logs_temp;"
 							"DROP TABLE logs_temp;"
 							"UPDATE fail2banDb SET version = 2;"
-						"COMMIT;" % Fail2BanDb._TABLE_logs)
+							"COMMIT;" % Fail2BanDb._CREATE_TABS['logs'])
 
-		if version < 3:
+			if version < 3 and self._tableExists(cur, "bans"):
 				cur.executescript("BEGIN TRANSACTION;"
 							"CREATE TEMPORARY TABLE bans_temp AS SELECT jail, ip, timeofban, 600 as bantime, 1 as bancount, data FROM bans;"
 							"DROP TABLE bans;"
 							"%s;"
 							"INSERT INTO bans SELECT * from bans_temp;"
 							"DROP TABLE bans_temp;"
-						"COMMIT;" % Fail2BanDb._TABLE_bans)
+							"COMMIT;" % Fail2BanDb._CREATE_TABS['bans'])
 			if version < 4:
 				cur.executescript("BEGIN TRANSACTION;"
 							"%s;"
 							"UPDATE fail2banDb SET version = 4;"
-						"COMMIT;" % Fail2BanDb._TABLE_bips)
+							"COMMIT;" % Fail2BanDb._CREATE_TABS['bips'])
 
 			cur.execute("SELECT version FROM fail2banDb LIMIT 1")
 			return cur.fetchone()[0]
+		except Exception as e:
+			# if still failed, just recreate database as fallback:
+			logSys.error("Failed to upgrade database '%s': %s",
+				self._dbFilename, e.args[0], 
+				exc_info=logSys.getEffectiveLevel() <= 10)
+			raise
 
 	@commitandrollback
 	def addJail(self, cur, jail):

fail2ban/tests/actiontestcase.py

@@ -30,9 +30,10 @@ import time
 import unittest
 
 from ..server.action import CommandAction, CallingMap, substituteRecursiveTags
-from ..server.actions import OrderedDict
+from ..server.actions import OrderedDict, Actions
 from ..server.utils import Utils
 
+from .dummyjail import DummyJail
 from .utils import LogCaptureTestCase
 from .utils import pid_exists
 
@@ -568,3 +569,19 @@ class CommandActionTest(LogCaptureTestCase):
 		self.assertIn("'b': 11", s)
 		self.assertIn("'c': ", s) # presents as callable
 		self.assertNotIn("'c': ''", s) # but not empty
+
+	def testActionsIdleMode(self):
+		a = Actions(DummyJail())
+		a.sleeptime = 0.0001;   # don't need to wait long
+		# enter idle mode right now (start idle):
+		a.idle = True;
+		# start:
+		a.start()
+		# wait for enter/leave of idle mode:
+		self.assertLogged("Actions: enter idle mode", wait=10)
+		# leave idle mode:
+		a.idle = False
+		self.assertLogged("Actions: leave idle mode", wait=10)
+		# stop it:
+		a.active = False
+		a.join()

fail2ban/tests/databasetestcase.py

@@ -62,7 +62,18 @@ class DatabaseTest(LogCaptureTestCase):
 		self.dbFilename = None
 		if not unittest.F2B.memory_db:
 			_, self.dbFilename = tempfile.mkstemp(".db", "fail2ban_")
-		self.db = getFail2BanDb(self.dbFilename)
+		self._db = ':auto-create-in-memory:'
+
+	@property
+	def db(self):
+		if isinstance(self._db, basestring) and self._db == ':auto-create-in-memory:':
+			self._db = getFail2BanDb(self.dbFilename)
+		return self._db
+	@db.setter
+	def db(self, value):
+		if isinstance(self._db, Fail2BanDb): # pragma: no cover
+			self._db.close()
+		self._db = value
 
 	def tearDown(self):
 		"""Call after every test case."""
@@ -106,10 +117,46 @@ class DatabaseTest(LogCaptureTestCase):
 			self.jail.name in self.db.getJailNames(),
 			"Jail not retained in Db after disconnect reconnect.")
 
+	def testRepairDb(self):
+		if Fail2BanDb is None: # pragma: no cover
+			return
+		self.db = None
+		if self.dbFilename is None: # pragma: no cover
+			_, self.dbFilename = tempfile.mkstemp(".db", "fail2ban_")
+		# test truncated database with different sizes:
+		#   - 14000 bytes - seems to be reparable,
+		#   - 4000  bytes - is totally broken.
+		for truncSize in (14000, 4000):
+			self.pruneLog("[test-repair], next phase - file-size: %d" % truncSize)
+			shutil.copyfile(
+				os.path.join(TEST_FILES_DIR, 'database_v1.db'), self.dbFilename)
+			# produce currupt database:
+			f = os.open(self.dbFilename, os.O_RDWR)
+			os.ftruncate(f, truncSize)
+			os.close(f)
+			# test repair:
+			try:
+				self.db = Fail2BanDb(self.dbFilename)
+				if truncSize == 14000: # restored:
+					self.assertLogged("Repair seems to be successful",
+						"Check integrity", "Database updated", all=True)
+					self.assertEqual(self.db.getLogPaths(), set(['/tmp/Fail2BanDb_pUlZJh.log']))
+					self.assertEqual(len(self.db.getJailNames()), 1)
+				else: # recreated:
+					self.assertLogged("Repair seems to be failed",
+						"Check integrity", "New database created.", all=True)
+					self.assertEqual(len(self.db.getLogPaths()), 0)
+					self.assertEqual(len(self.db.getJailNames()), 0)
+			finally:
+				if self.db and self.db._dbFilename != ":memory:":
+					os.remove(self.db._dbBackupFilename)
+					self.db = None
+
 	def testUpdateDb(self):
 		if Fail2BanDb is None: # pragma: no cover
 			return
 		self.db = None
+		try:
 			if self.dbFilename is None: # pragma: no cover
 				_, self.dbFilename = tempfile.mkstemp(".db", "fail2ban_")
 			shutil.copyfile(
@@ -122,6 +169,8 @@ class DatabaseTest(LogCaptureTestCase):
 
 			self.assertEqual(self.db.updateDb(Fail2BanDb.__version__), Fail2BanDb.__version__)
 			self.assertRaises(NotImplementedError, self.db.updateDb, Fail2BanDb.__version__ + 1)
+		finally:
+			if self.db and self.db._dbFilename != ":memory:":
 				os.remove(self.db._dbBackupFilename)
 
 	def testUpdateDb2(self):

fail2ban/tests/fail2banclienttestcase.py

@@ -126,6 +126,9 @@ class FailExitException(fail2bancmdline.ExitException):
 	pass
 
 
+SUCCESS = ExitException
+FAILED = FailExitException
+
 INTERACT = []
 
 
@@ -329,7 +332,20 @@ def with_foreground_server_thread(startextra={}):
 				)
 				th.daemon = True
 				th.start()
-				try:
+				# to wait for end of server, default accept any exit code, because multi-threaded, 
+				# thus server can exit in-between...
+				def _stopAndWaitForServerEnd(code=(SUCCESS, FAILED)):
+					# if seems to be down - try to catch end phase (wait a bit for end:True to recognize down state):
+					if not phase.get('end', None) and not os.path.exists(pjoin(tmp, "f2b.pid")):
+						Utils.wait_for(lambda: phase.get('end', None) is not None, MID_WAITTIME)
+					# stop (if still running):
+					if not phase.get('end', None):
+						self.execCmd(code, startparams, "stop")
+						# wait for end sign:
+						Utils.wait_for(lambda: phase.get('end', None) is not None, MAX_WAITTIME)
+						self.assertTrue(phase.get('end', None))
+						self.assertLogged("Shutdown successful", "Exiting Fail2ban", all=True)
+				self.stopAndWaitForServerEnd = _stopAndWaitForServerEnd
 				# wait for start thread:
 				Utils.wait_for(lambda: phase.get('start', None) is not None, MAX_WAITTIME)
 				self.assertTrue(phase.get('start', None))
@@ -340,24 +356,16 @@ def with_foreground_server_thread(startextra={}):
 				# several commands to server in body of decorated function:
 				return f(self, tmp, startparams, *args, **kwargs)
 			finally:
+				if th:
+					# wait for server end (if not yet already exited):
 					DefLogSys.info('=== within server: end.  ===')
 					self.pruneLog()
-					# if seems to be down - try to catch end phase (wait a bit for end:True to recognize down state):
+					self.stopAndWaitForServerEnd()
-					if not phase.get('end', None) and not os.path.exists(pjoin(tmp, "f2b.pid")):
-						Utils.wait_for(lambda: phase.get('end', None) is not None, MID_WAITTIME)
-					# stop (if still running):
-					if not phase.get('end', None):
-						self.execSuccess(startparams, "stop")
-						# wait for end:
-						Utils.wait_for(lambda: phase.get('end', None) is not None, MAX_WAITTIME)
-						self.assertTrue(phase.get('end', None))
-						self.assertLogged("Shutdown successful", "Exiting Fail2ban", all=True)
-			finally:
-				if th:
 					# we start client/server directly in current process (new thread),
 					# so don't kill (same process) - if success, just wait for end of worker:
 					if phase.get('end', None):
 						th.join()
+				self.stopAndWaitForServerEnd = None
 				tearDownMyTime()
 		return wrapper
 	return _deco_wrapper
@@ -420,11 +428,9 @@ class Fail2banClientServerBase(LogCaptureTestCase):
 				logSys.debug("No log file %s to examine details of error", log)
 			raise
 
-	def execSuccess(self, startparams, *args):
+	def execCmd(self, exitType, startparams, *args):
-		raise NotImplementedError("To be defined in subclass")
+		self.assertRaises(exitType, self.exec_command_line[0],
-
+			(self.exec_command_line[1:] + startparams + args))
-	def execFailed(self, startparams, *args):
-		raise NotImplementedError("To be defined in subclass")
 
 	#
 	# Common tests
@@ -433,7 +439,7 @@ class Fail2banClientServerBase(LogCaptureTestCase):
 		# start and wait to end (foreground):
 		logSys.debug("start of test worker")
 		phase['start'] = True
-		self.execSuccess(("-f",) + startparams, "start")
+		self.execCmd(SUCCESS, ("-f",) + startparams, "start")
 		# end :
 		phase['end'] = True
 		logSys.debug("end of test worker")
@@ -441,46 +447,40 @@ class Fail2banClientServerBase(LogCaptureTestCase):
 	@with_foreground_server_thread()
 	def testStartForeground(self, tmp, startparams):
 		# several commands to server:
-		self.execSuccess(startparams, "ping")
+		self.execCmd(SUCCESS, startparams, "ping")
-		self.execFailed(startparams, "~~unknown~cmd~failed~~")
+		self.execCmd(FAILED, startparams, "~~unknown~cmd~failed~~")
-		self.execSuccess(startparams, "echo", "TEST-ECHO")
+		self.execCmd(SUCCESS, startparams, "echo", "TEST-ECHO")
 
 
 class Fail2banClientTest(Fail2banClientServerBase):
 
-	def execSuccess(self, startparams, *args):
+	exec_command_line = (_exec_client, CLIENT,)
-		self.assertRaises(ExitException, _exec_client,
-			((CLIENT,) + startparams + args))
-
-	def execFailed(self, startparams, *args):
-		self.assertRaises(FailExitException, _exec_client,
-			((CLIENT,) + startparams + args))
 
 	def testConsistency(self):
 		self.assertTrue(isfile(pjoin(BIN, CLIENT)))
 		self.assertTrue(isfile(pjoin(BIN, SERVER)))
 
 	def testClientUsage(self):
-		self.execSuccess((), "-h")
+		self.execCmd(SUCCESS, (), "-h")
 		self.assertLogged("Usage: " + CLIENT)
 		self.assertLogged("Report bugs to ")
 		self.pruneLog()
-		self.execSuccess((), "-vq", "-V")
+		self.execCmd(SUCCESS, (), "-vq", "-V")
 		self.assertLogged("Fail2Ban v" + fail2bancmdline.version)
 		self.pruneLog()
-		self.execSuccess((), "--str2sec", "1d12h30m")
+		self.execCmd(SUCCESS, (), "--str2sec", "1d12h30m")
 		self.assertLogged("131400")
 
 	@with_tmpdir
 	def testClientDump(self, tmp):
 		# use here the stock configuration (if possible)
 		startparams = _start_params(tmp, True)
-		self.execSuccess(startparams, "-vvd")
+		self.execCmd(SUCCESS, startparams, "-vvd")
 		self.assertLogged("Loading files")
 		self.assertLogged("['set', 'logtarget',")
 		self.pruneLog()
 		# pretty dump:
-		self.execSuccess(startparams, "--dp")
+		self.execCmd(SUCCESS, startparams, "--dp")
 		self.assertLogged("['set', 'logtarget',")
 		
 	@with_tmpdir
@@ -489,28 +489,28 @@ class Fail2banClientTest(Fail2banClientServerBase):
 		# use once the stock configuration (to test starting also)
 		startparams = _start_params(tmp, True)
 		# start:
-		self.execSuccess(("-b",) + startparams, "start")
+		self.execCmd(SUCCESS, ("-b",) + startparams, "start")
 		# wait for server (socket and ready):
 		self._wait_for_srv(tmp, True, startparams=startparams)
 		self.assertLogged("Server ready")
 		self.assertLogged("Exit with code 0")
 		try:
-			self.execSuccess(startparams, "echo", "TEST-ECHO")
+			self.execCmd(SUCCESS, startparams, "echo", "TEST-ECHO")
-			self.execFailed(startparams, "~~unknown~cmd~failed~~")
+			self.execCmd(FAILED, startparams, "~~unknown~cmd~failed~~")
 			self.pruneLog()
 			# start again (should fail):
-			self.execFailed(("-b",) + startparams, "start")
+			self.execCmd(FAILED, ("-b",) + startparams, "start")
 			self.assertLogged("Server already running")
 		finally:
 			self.pruneLog()
 			# stop:
-			self.execSuccess(startparams, "stop")
+			self.execCmd(SUCCESS, startparams, "stop")
 			self.assertLogged("Shutdown successful")
 			self.assertLogged("Exit with code 0")
 
 		self.pruneLog()
 		# stop again (should fail):
-		self.execFailed(startparams, "stop")
+		self.execCmd(FAILED, startparams, "stop")
 		self.assertLogged("Failed to access socket path")
 		self.assertLogged("Is fail2ban running?")
 
@@ -521,7 +521,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
 		startparams = _start_params(tmp, logtarget=pjoin(tmp, "f2b.log"))
 		# if fast, start server process from client started direct here:
 		if unittest.F2B.fast: # pragma: no cover
-			self.execSuccess(startparams + ("start",))
+			self.execCmd(SUCCESS, startparams + ("start",))
 		else:
 			# start (in new process, using the same python version):
 			cmd = (sys.executable, pjoin(BIN, CLIENT))
@@ -535,12 +535,12 @@ class Fail2banClientTest(Fail2banClientServerBase):
 		self.pruneLog()
 		try:
 			# echo from client (inside):
-			self.execSuccess(startparams, "echo", "TEST-ECHO")
+			self.execCmd(SUCCESS, startparams, "echo", "TEST-ECHO")
 			self.assertLogged("TEST-ECHO")
 			self.assertLogged("Exit with code 0")
 			self.pruneLog()
 			# test ping timeout:
-			self.execSuccess(startparams, "ping", "0.1")
+			self.execCmd(SUCCESS, startparams, "ping", "0.1")
 			self.assertLogged("Server replied: pong")
 			self.pruneLog()
 			# python 3 seems to bypass such short timeouts also, 
@@ -551,7 +551,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
 				os.kill(pid, signal.SIGSTOP); # or SIGTSTP?
 				time.sleep(Utils.DEFAULT_SHORT_INTERVAL)
 				# test ping with short timeout:
-				self.execFailed(startparams, "ping", "1e-10")
+				self.execCmd(FAILED, startparams, "ping", "1e-10")
 			finally:
 				# resume:
 				os.kill(pid, signal.SIGCONT)
@@ -563,7 +563,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
 				"status",
 				"exit"
 			]
-			self.execSuccess(startparams, "-i")
+			self.execCmd(SUCCESS, startparams, "-i")
 			self.assertLogged("INTERACT-ECHO")
 			self.assertLogged("Status", "Number of jail:")
 			self.assertLogged("Exit with code 0")
@@ -574,7 +574,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
 				"restart",
 				"exit"
 			]
-			self.execSuccess(startparams, "-i")
+			self.execCmd(SUCCESS, startparams, "-i")
 			self.assertLogged("Reading config files:")
 			self.assertLogged("Shutdown successful")
 			self.assertLogged("Server ready")
@@ -585,18 +585,18 @@ class Fail2banClientTest(Fail2banClientServerBase):
 				"reload ~~unknown~jail~fail~~",
 				"exit"
 			]
-			self.execSuccess(startparams, "-i")
+			self.execCmd(SUCCESS, startparams, "-i")
 			self.assertLogged("Failed during configuration: No section: '~~unknown~jail~fail~~'")
 			self.pruneLog()
 			# test reload missing jail (direct):
-			self.execFailed(startparams, "reload", "~~unknown~jail~fail~~")
+			self.execCmd(FAILED, startparams, "reload", "~~unknown~jail~fail~~")
 			self.assertLogged("Failed during configuration: No section: '~~unknown~jail~fail~~'")
 			self.assertLogged("Exit with code -1")
 			self.pruneLog()
 		finally:
 			self.pruneLog()
 			# stop:
-			self.execSuccess(startparams, "stop")
+			self.execCmd(SUCCESS, startparams, "stop")
 			self.assertLogged("Shutdown successful")
 			self.assertLogged("Exit with code 0")
 
@@ -607,33 +607,33 @@ class Fail2banClientTest(Fail2banClientServerBase):
 		startparams = _start_params(tmp, logtarget="INHERITED")
 
 		## wrong config directory
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"--async", "-c", pjoin(tmp, "miss"), "start")
 		self.assertLogged("Base configuration directory " + pjoin(tmp, "miss") + " does not exist")
 		self.pruneLog()
 
 		## wrong socket
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"--async", "-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "miss/f2b.sock"), "start")
 		self.assertLogged("There is no directory " + pjoin(tmp, "miss") + " to contain the socket file")
 		self.pruneLog()
 
 		## not running
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"), "reload")
 		self.assertLogged("Could not find server")
 		self.pruneLog()
 
 		## already exists:
 		open(pjoin(tmp, "f2b.sock"), 'a').close()
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"--async", "-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"), "start")
 		self.assertLogged("Fail2ban seems to be in unexpected state (not running but the socket exists)")
 		self.pruneLog()
 		os.remove(pjoin(tmp, "f2b.sock"))
 
 		## wrong option:
-		self.execFailed((), "-s")
+		self.execCmd(FAILED, (), "-s")
 		self.assertLogged("Usage: ")
 		self.pruneLog()
 
@@ -643,13 +643,13 @@ class Fail2banClientTest(Fail2banClientServerBase):
 		startparams = _start_params(tmp, logtarget="INHERITED")
 
 		# not started:
-		self.execFailed(startparams,
+		self.execCmd(FAILED, startparams,
 			"reload", "jail")
 		self.assertLogged("Could not find server")
 		self.pruneLog()
 
 		# unexpected arg:
-		self.execFailed(startparams,
+		self.execCmd(FAILED, startparams,
 			"--async", "reload", "--xxx", "jail")
 		self.assertLogged("Unexpected argument(s) for reload:")
 		self.pruneLog()
@@ -669,16 +669,10 @@ class Fail2banClientTest(Fail2banClientServerBase):
 
 class Fail2banServerTest(Fail2banClientServerBase):
 
-	def execSuccess(self, startparams, *args):
+	exec_command_line = (_exec_server, SERVER,)
-		self.assertRaises(ExitException, _exec_server,
-			((SERVER,) + startparams + args))
-
-	def execFailed(self, startparams, *args):
-		self.assertRaises(FailExitException, _exec_server,
-			((SERVER,) + startparams + args))
 
 	def testServerUsage(self):
-		self.execSuccess((), "-h")
+		self.execCmd(SUCCESS, (), "-h")
 		self.assertLogged("Usage: " + SERVER)
 		self.assertLogged("Report bugs to ")
 
@@ -698,12 +692,12 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		self.assertLogged("Server ready")
 		self.pruneLog()
 		try:
-			self.execSuccess(startparams, "echo", "TEST-ECHO")
+			self.execCmd(SUCCESS, startparams, "echo", "TEST-ECHO")
-			self.execFailed(startparams, "~~unknown~cmd~failed~~")
+			self.execCmd(FAILED, startparams, "~~unknown~cmd~failed~~")
 		finally:
 			self.pruneLog()
 			# stop:
-			self.execSuccess(startparams, "stop")
+			self.execCmd(SUCCESS, startparams, "stop")
 			self.assertLogged("Shutdown successful")
 			self.assertLogged("Exit with code 0")
 
@@ -714,20 +708,20 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		startparams = _start_params(tmp, logtarget="INHERITED")
 
 		## wrong config directory
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"-c", pjoin(tmp, "miss"))
 		self.assertLogged("Base configuration directory " + pjoin(tmp, "miss") + " does not exist")
 		self.pruneLog()
 
 		## wrong socket
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"-c", pjoin(tmp, "config"), "-x", "-s", pjoin(tmp, "miss/f2b.sock"))
 		self.assertLogged("There is no directory " + pjoin(tmp, "miss") + " to contain the socket file")
 		self.pruneLog()
 
 		## already exists:
 		open(pjoin(tmp, "f2b.sock"), 'a').close()
-		self.execFailed((),
+		self.execCmd(FAILED, (),
 			"-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"))
 		self.assertLogged("Fail2ban seems to be in unexpected state (not running but the socket exists)")
 		self.pruneLog()
@@ -742,7 +736,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# test configuration is correct:
 		self.pruneLog("[test-phase 0]")
-		self.execSuccess(startparams, "--test")
+		self.execCmd(SUCCESS, startparams, "--test")
 		self.assertLogged("OK: configuration test is successful")
 
 		# append one wrong configured jail:
@@ -751,14 +745,14 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# first try test config:
 		self.pruneLog("[test-phase 0a]")
-		self.execFailed(startparams, "--test")
+		self.execCmd(FAILED, startparams, "--test")
 		self.assertLogged("Unable to read the filter 'broken-jail-filter'",
 			"Errors in jail 'broken-jail'.",
 			"ERROR: test configuration failed", all=True)
 
 		# failed to start with test config:
 		self.pruneLog("[test-phase 0b]")
-		self.execFailed(startparams, "-t", "start")
+		self.execCmd(FAILED, startparams, "-t", "start")
 		self.assertLogged("Unable to read the filter 'broken-jail-filter'",
 			"Errors in jail 'broken-jail'.",
 			"ERROR: test configuration failed", all=True)
@@ -889,7 +883,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		self.pruneLog("[test-phase 1a]")
 		if unittest.F2B.log_level < logging.DEBUG: # pragma: no cover
 			_out_file(test1log)
-		self.execSuccess(startparams, "reload")
+		self.execCmd(SUCCESS, startparams, "reload")
 		self.assertLogged(
 			"Reload finished.",
 			"1 ticket(s) in 'test-jail1", all=True, wait=MID_WAITTIME)
@@ -918,7 +912,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		_write_file(test1log, "w+")
 		if unittest.F2B.log_level < logging.DEBUG: # pragma: no cover
 			_out_file(test1log)
-		self.execSuccess(startparams, "reload")
+		self.execCmd(SUCCESS, startparams, "reload")
 		self.assertLogged("Reload finished.", all=True, wait=MID_WAITTIME)
 		# test not unbanned / banned again:
 		self.assertNotLogged(
@@ -950,7 +944,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 			start= "               echo '[<name>] %s: started.'" % "test-action1",
 			reload="               echo '[<name>] %s: reloaded.'" % "test-action1", 
 			stop=  "               echo '[<name>] %s: stopped.'" % "test-action1")
-		self.execSuccess(startparams, "reload")
+		self.execCmd(SUCCESS, startparams, "reload")
 		self.assertLogged("Reload finished.", all=True, wait=MID_WAITTIME)
 		# test not unbanned / banned again:
 		self.assertNotLogged(
@@ -1013,7 +1007,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# restart jail without unban all:
 		self.pruneLog("[test-phase 2c]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"restart", "test-jail2")
 		self.assertLogged(
 			"Reload finished.",
@@ -1041,9 +1035,9 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# ban manually to test later flush by unban all:
 		self.pruneLog("[test-phase 2d]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"set", "test-jail2", "banip", "192.0.2.21")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"set", "test-jail2", "banip", "192.0.2.22")
 		self.assertLogged(
 			"stdout: '[test-jail2] test-action3: ++ ban 192.0.2.22",
@@ -1051,7 +1045,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# restart jail with unban all:
 		self.pruneLog("[test-phase 2e]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"restart", "--unban", "test-jail2")
 		self.assertLogged(
 			"Reload finished.",
@@ -1084,7 +1078,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# reload jail1 without restart (without ban/unban):
 		self.pruneLog("[test-phase 3]")
-		self.execSuccess(startparams, "reload", "test-jail1")
+		self.execCmd(SUCCESS, startparams, "reload", "test-jail1")
 		self.assertLogged(
 			"Reload finished.", all=True, wait=MID_WAITTIME)
 		self.assertLogged(
@@ -1099,7 +1093,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		# whole reload, but this time with jail1 only (jail2 should be stopped via configuration):
 		self.pruneLog("[test-phase 4]")
 		_write_jail_cfg(enabled=[1])
-		self.execSuccess(startparams, "reload")
+		self.execCmd(SUCCESS, startparams, "reload")
 		self.assertLogged("Reload finished.", all=True, wait=MID_WAITTIME)
 		# test both jails should be reloaded:
 		self.assertLogged(
@@ -1136,7 +1130,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# unban single ips:
 		self.pruneLog("[test-phase 6]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"--async", "unban", "192.0.2.5", "192.0.2.6")
 		self.assertLogged(
 			"192.0.2.5 is not banned",
@@ -1145,7 +1139,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# reload all (one jail) with unban all:
 		self.pruneLog("[test-phase 7]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"reload", "--unban")
 		self.assertLogged("Reload finished.", all=True, wait=MID_WAITTIME)
 		# reloads unbanned all:
@@ -1168,7 +1162,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# unban all (just to test command, already empty - nothing to unban):
 		self.pruneLog("[test-phase 7b]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"--async", "unban", "--all")
 		self.assertLogged(
 			"Flush ban list",
@@ -1177,7 +1171,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		# backend-switch (restart instead of reload):
 		self.pruneLog("[test-phase 8a]")
 		_write_jail_cfg(enabled=[1], backend="xxx-unknown-backend-zzz")
-		self.execFailed(startparams, "reload")
+		self.execCmd(FAILED, startparams, "reload")
 		self.assertLogged("Reload finished.", all=True, wait=MID_WAITTIME)
 		self.assertLogged(
 			"Restart jail 'test-jail1' (reason: 'polling' != ", 
@@ -1185,18 +1179,18 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		self.pruneLog("[test-phase 8b]")
 		_write_jail_cfg(enabled=[1])
-		self.execSuccess(startparams, "reload")
+		self.execCmd(SUCCESS, startparams, "reload")
 		self.assertLogged("Reload finished.", all=True, wait=MID_WAITTIME)	
 
 		# several small cases (cover several parts):
 		self.pruneLog("[test-phase end-1]")
 		# wrong jail (not-started):
-		self.execFailed(startparams,
+		self.execCmd(FAILED, startparams,
 			"--async", "reload", "test-jail2")
 		self.assertLogged("the jail 'test-jail2' does not exist")
 		self.pruneLog()
 		# unavailable jail (but exit 0), using --if-exists option:
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"--async", "reload", "--if-exists", "test-jail2")
 		self.assertNotLogged(
 			"Creating new jail 'test-jail2'",
@@ -1204,7 +1198,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 
 		# restart all jails (without restart server):
 		self.pruneLog("[test-phase end-2]")
-		self.execSuccess(startparams,
+		self.execCmd(SUCCESS, startparams,
 			"--async", "reload", "--restart", "--all")
 		self.assertLogged(
 			"Jail 'test-jail1' stopped", 
@@ -1249,7 +1243,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 			"[nginx-blck-lst] Ban 125-000-003",
 			"[nginx-blck-lst] Ban 125-000-004",
 			"[nginx-blck-lst] Ban 125-000-005",
-			"Banned 5 / 5, 5 ticket(s)",
+			"5 ticket(s)",
 			all=True, wait=MID_WAITTIME
 		)
 		_out_file(mpfn)
@@ -1261,7 +1255,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		self.assertIn('\\125-000-005 1;\n', mp)
 
 		# unban 1, 2 and 5:
-		self.execSuccess(startparams, 'unban', '125-000-001', '125-000-002', '125-000-005')
+		self.execCmd(SUCCESS, startparams, 'unban', '125-000-001', '125-000-002', '125-000-005')
 		_out_file(mpfn)
 		# check really unbanned but other sessions are still present (blacklisted in map-file):
 		mp = _read_file(mpfn)
@@ -1272,8 +1266,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		self.assertIn('\\125-000-004 1;\n', mp)
 
 		# stop server and wait for end:
-		self.execSuccess(startparams, 'stop')
+		self.stopAndWaitForServerEnd(SUCCESS)
-		self.assertLogged("Shutdown successful", "Exiting Fail2ban", all=True, wait=MID_WAITTIME)
 
 		# check flushed (all sessions were deleted from map-file):
 		self.assertLogged("[nginx-blck-lst] Flush ticket(s) with nginx-block-map")
@@ -1331,7 +1324,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
 		_write_file(test1log, "w")
 		# initial start:
 		self.pruneLog("[test-phase 0) time-0]")
-		self.execSuccess(startparams, "reload")
+		self.execCmd(SUCCESS, startparams, "reload")
 		# generate bad ip:
 		_write_file(test1log, "w+", *(
 		  (str(int(MyTime.time())) + " failure 401 from 192.0.2.11: I'm bad \"hacker\" `` $(echo test)",) * 3

fail2ban/tests/servertestcase.py

@@ -1677,6 +1677,10 @@ class ServerConfigReaderTests(LogCaptureTestCase):
 						"`ipset create f2b-j-w-fwcmd-ipset6 hash:ip timeout 0 family inet6`",
 						"`firewall-cmd --direct --add-rule ipv6 filter INPUT_direct 0 -p tcp -m multiport --dports http -m set --match-set f2b-j-w-fwcmd-ipset6 src -j REJECT --reject-with icmp6-port-unreachable`",
 					),
+					'flush': (
+						"`ipset flush f2b-j-w-fwcmd-ipset`",
+						"`ipset flush f2b-j-w-fwcmd-ipset6`",
+					),
 					'stop': (
 						"`firewall-cmd --direct --remove-rule ipv4 filter INPUT_direct 0 -p tcp -m multiport --dports http -m set --match-set f2b-j-w-fwcmd-ipset src -j REJECT --reject-with icmp-port-unreachable`",
 						"`ipset flush f2b-j-w-fwcmd-ipset`",

fail2ban/tests/sockettestcase.py

@@ -31,8 +31,10 @@ import threading
 import time
 import unittest
 
+from .utils import LogCaptureTestCase
+
 from .. import protocol
-from ..server.asyncserver import asyncore, RequestHandler, AsyncServer, AsyncServerException
+from ..server.asyncserver import asyncore, RequestHandler, loop, AsyncServer, AsyncServerException
 from ..server.utils import Utils
 from ..client.csocket import CSocket
 
@@ -219,7 +221,20 @@ class Socket(LogCaptureTestCase):
 		self.assertFalse(os.path.exists(self.sock_name))
 
 
-class ClientMisc(unittest.TestCase):
+class ClientMisc(LogCaptureTestCase):
+
+	def testErrorsInLoop(self):
+		phase = {'cntr': 0}
+		def _active():
+			return phase['cntr'] < 40
+		def _poll(*args):
+			phase['cntr'] += 1
+			raise Exception('test *%d*' % phase['cntr'])
+		# test errors "catched" and logged:
+		loop(_active, use_poll=_poll)
+		self.assertLogged("test *1*", "test *10*", "test *20*", all=True)
+		self.assertLogged("Too many errors - stop logging connection errors")
+		self.assertNotLogged("test *21*", "test *22*", "test *23*", all=True)
 
 	def testPrintFormattedAndWiki(self):
 		# redirect stdout to devnull