**not ready** testActionsConsistencyCheck fixed, but several **broken** tests (todo: fix public interface like action.start()/stop()).

fail2ban/server/action.py

@@ -382,7 +382,7 @@ class CommandAction(ActionBase):
 			addrepl=(lambda tag:family if tag == 'family' else None),
 			cache=self.__substCache)
 
-	def _executeOperation(self, tag, operation, family=[]):
+	def _executeOperation(self, tag, operation, family=[], forceExec=False, afterExec=None):
 		"""Executes the operation commands (like "actionstart", "actionstop", etc).
 
 		Replace the tags in the action command with actions properties
@@ -395,12 +395,17 @@ class CommandAction(ActionBase):
 			cmd = self._getOperation(tag, 'inet4')
 			if not family or 'inet4' in family:
 				if cmd:
-					res &= self.executeCmd(cmd, self.timeout)
+					ret = self.executeCmd(cmd, self.timeout)
+					res &= ret
+					if afterExec: afterExec('inet4' if 'inet4' in family else '', ret)
 			# execute ipv6 operation if available (and not the same as ipv4):
 			if allowed_ipv6 and (not family or 'inet6' in family):
 				cmd6 = self._getOperation(tag, 'inet6')
-				if cmd6 and cmd6 != cmd: # - avoid double execution of same command
+				forceExec |= (family and 'inet6' in family and 'inet4' not in family)
-					res &= self.executeCmd(cmd6, self.timeout)
+				if cmd6 and (forceExec or cmd6 != cmd): # - avoid double execution of same command
+					ret = self.executeCmd(cmd6, self.timeout)
+					res &= ret
+					if afterExec: afterExec('inet6' if 'inet6' in family else '', ret)
 			if not res:
 				raise RuntimeError("Error %s action %s/%s" % (operation, self._jail, self._name,))
 		except ValueError as e:
@@ -440,10 +445,11 @@ class CommandAction(ActionBase):
 		if self._startOnDemand:
 			if not forceStart:
 				return True
-		elif self.__started.get(family): # pragma: no cover - normally unreachable
+		elif not forceStart and self.__started.get(family): # pragma: no cover - normally unreachable
 			return True
-		ret = self._executeOperation('<actionstart>', 'starting', family=family)
+		ret = self._executeOperation('<actionstart>', 'starting', family=[family], forceExec=forceStart)
-		self.__started[family] = ret
+		if ret:
+			self.__started[family] = 1
 		return ret
 
 	def ban(self, aInfo):
@@ -459,13 +465,14 @@ class CommandAction(ActionBase):
 			the ban.
 		"""
 		# if we should start the action on demand (conditional by family):
+		family = aInfo.get('family', '')
 		if self._startOnDemand:
-			family = aInfo.get('family')
 			if not self.__started.get(family):
 				self.start(family, forceStart=True)
 		# ban:
 		if not self._processCmd('<actionban>', aInfo):
 			raise RuntimeError("Error banning %(ip)s" % aInfo)
+		self.__started[family] |= 2; # contains items
 
 	def unban(self, aInfo):
 		"""Executes the "actionunban" command.
@@ -479,8 +486,10 @@ class CommandAction(ActionBase):
 			Dictionary which includes information in relation to
 			the ban.
 		"""
-		if not self._processCmd('<actionunban>', aInfo):
+		family = aInfo.get('family', '')
-			raise RuntimeError("Error unbanning %(ip)s" % aInfo)
+		if self.__started.get(family) & 2: # contains items
+			if not self._processCmd('<actionunban>', aInfo):
+				raise RuntimeError("Error unbanning %(ip)s" % aInfo)
 
 	def flush(self):
 		"""Executes the "actionflush" command.
@@ -491,27 +500,34 @@ class CommandAction(ActionBase):
 		Replaces the tags in the action command with actions properties
 		and executes the resulting command.
 		"""
-		family = []
+		# collect started families, may be started on demand (conditional):
-		# collect started families, if started on demand (conditional):
+		family = [f for (f,v) in self.__started.iteritems() if v & 3 == 3]; # started and contains items
-		if self._startOnDemand:
+		# if nothing contains items:
-			family = [f for (f,v) in self.__started.iteritems() if v]
+		if not family: return True
-			# if no started (on demand) actions:
+		# flush:
-			if not family: return True
+		def _afterFlush(family, ret):
-		return self._executeOperation('<actionflush>', 'flushing', family=family)
+			if ret and self.__started.get(family):
+				self.__started[family] &= ~2; # no items anymore
+		return self._executeOperation('<actionflush>', 'flushing', family=family, afterExec=_afterFlush)
 
-	def stop(self):
+	def stop(self, family=None):
 		"""Executes the "actionstop" command.
 
 		Replaces the tags in the action command with actions properties
 		and executes the resulting command.
 		"""
-		family = []
 		# collect started families, if started on demand (conditional):
-		if self._startOnDemand:
+		if family is None:
 			family = [f for (f,v) in self.__started.iteritems() if v]
 			# if no started (on demand) actions:
 			if not family: return True
-		self.__started = {}
+			self.__started = {}
+		else:
+			try:
+				self.__started[family] &= 0
+				family = [family]
+			except KeyError: # pragma: no cover
+				return True
 		return self._executeOperation('<actionstop>', 'stopping', family=family)
 
 	def reload(self, **kwargs):
@@ -533,8 +549,9 @@ class CommandAction(ActionBase):
 		ret = True
 		# for each started family:
 		if self.actioncheck:
-			for (family, started) in self.__started.iteritems():
+			for (family, started) in self.__started.items():
 				if started and not self._invariantCheck(family, beforeRepair):
+					self.__started[family] = 0
 					ret &= False
 		return ret
 
@@ -733,6 +750,9 @@ class CommandAction(ActionBase):
 	def _invariantCheck(self, family='', beforeRepair=None):
 		"""Executes a substituted `actioncheck` command.
 		"""
+		# for started action/family only (avoid check not started inet4 if inet6 gets broken):
+		if family != '' and family not in self.__started and not self.__started.get(''):
+			return True
 		checkCmd = self._getOperation('<actioncheck>', family)
 		if not checkCmd or self.executeCmd(checkCmd, self.timeout):
 			return True
@@ -754,10 +774,10 @@ class CommandAction(ActionBase):
 			#    but the tickets are still in BanManager, so in case of new failures
 			#    it will not be banned, because "already banned" will happen.
 			try:
-				self.stop()
+				self.stop(family)
 			except RuntimeError: # bypass error in stop (if start/check succeeded hereafter).
 				pass
-			self.start()
+			self.start(family)
 		if not self.executeCmd(checkCmd, self.timeout):
 			self._logSys.critical("Unable to restore environment")
 			return False

fail2ban/server/actions.py

@@ -516,23 +516,24 @@ class Actions(JailThread, Mapping):
 			try:
 				if hasattr(action, 'flush') and (not isinstance(action, CommandAction) or action.actionflush):
 					logSys.notice("[%s] Flush ticket(s) with %s", self._jail.name, name)
-					action.flush()
+					if action.flush():
-				else:
+						continue
-					unbactions[name] = action
 			except Exception as e:
 				logSys.error("Failed to flush bans in jail '%s' action '%s': %s",
 					self._jail.name, name, e,
 					exc_info=logSys.getEffectiveLevel()<=logging.DEBUG)
 				logSys.info("No flush occured, do consistency check")
-				def _beforeRepair():
+				if hasattr(action, 'consistencyCheck'):
-					if stop:
+					def _beforeRepair():
-						self._logSys.error("Invariant check failed. Flush is impossible.")
+						if stop and not getattr(action, 'actionrepair_on_unban', None): # don't need repair on stop
-						return False
+							self._logSys.error("Invariant check failed. Flush is impossible.")
-					return True
+							return False
-				if not hasattr(action, 'consistencyCheck') or action.consistencyCheck(_beforeRepair):
+						return True
-					# fallback to single unbans:
+					action.consistencyCheck(_beforeRepair)
-					logSys.info("unban tickets each individualy")
+					continue
-					unbactions[name] = action
+			# fallback to single unbans:
+			logSys.debug("  Unban tickets each individualy")
+			unbactions[name] = action
 		actions = unbactions
 		# flush the database also:
 		if db and self._jail.database is not None:

fail2ban/tests/actionstestcase.py

@@ -48,15 +48,15 @@ class ExecuteActions(LogCaptureTestCase):
 	def tearDown(self):
 		super(ExecuteActions, self).tearDown()
 
-	def defaultAction(self):
+	def defaultAction(self, o={}):
 		self.__actions.add('ip')
 		act = self.__actions['ip']
-		act.actionstart = 'echo ip start'
+		act.actionstart = 'echo ip start'+o.get('start', '')
 		act.actionban = 'echo ip ban <ip>'
 		act.actionunban = 'echo ip unban <ip>'
-		act.actioncheck = 'echo ip check'
+		act.actioncheck = 'echo ip check'+o.get('check', '')
-		act.actionflush = 'echo ip flush <family>'
+		act.actionflush = 'echo ip flush'+o.get('flush', '')
-		act.actionstop = 'echo ip stop'
+		act.actionstop = 'echo ip stop'+o.get('stop', '')
 		return act
 
 	def testActionsAddDuplicateName(self):
@@ -211,11 +211,10 @@ class ExecuteActions(LogCaptureTestCase):
 		self.assertLogged('Unbanned 30, 0 ticket(s)')
 		self.assertNotLogged('Unbanned 50, 0 ticket(s)')
 
-	@with_alt_time
 	def testActionsConsistencyCheck(self):
+		act = self.defaultAction({'check':' <family>', 'flush':' <family>'})
 		# flush for inet6 is intentionally "broken" here - test no unhandled except and invariant check:
-		act = self.defaultAction()
+		act['actionflush?family=inet6'] = act.actionflush + '; exit 1'
-		act['actionflush?family=inet6'] = 'echo ip flush <family>; exit 1'
 		act.actionstart_on_demand = True
 		self.__actions.start()
 		self.assertNotLogged("stdout: %r" % 'ip start')
@@ -230,26 +229,137 @@ class ExecuteActions(LogCaptureTestCase):
 
 		# check should fail (so cause stop/start):
 		self.pruneLog('[test-phase 1] simulate inconsistent env')
-		act['actioncheck?family=inet6'] = 'echo ip check <family>; exit 1'
+		act['actioncheck?family=inet6'] = act.actioncheck + '; exit 1'
 		self.__actions._Actions__flushBan()
-		self.assertLogged('Failed to flush bans',
+		self.assertLogged(
+			"stdout: %r" % 'ip flush inet4',
+			"stdout: %r" % 'ip flush inet6',
+			'Failed to flush bans',
 			'No flush occured, do consistency check',
 			'Invariant check failed. Trying to restore a sane environment',
-			"stdout: %r" % 'ip stop',
+			"stdout: %r" % 'ip stop',  # same for both families
-			"stdout: %r" % 'ip start',
+			'Unable to restore environment',
 			all=True, wait=True)
 
 		# check succeeds:
 		self.pruneLog('[test-phase 2] consistent env')
 		act['actioncheck?family=inet6'] = act.actioncheck
+		self.assertEqual(self.__actions.addBannedIP('2001:db8::1'), 1)
+		self.assertLogged('Ban 2001:db8::1',
+			"stdout: %r" % 'ip start',   # same for both families
+			"stdout: %r" % 'ip ban 2001:db8::1',
+			all=True, wait=True)
+		self.assertNotLogged("stdout: %r" % 'ip check inet4',
+			all=True)
+
+		self.pruneLog('[test-phase 3] failed flush in consistent env')
 		self.__actions._Actions__flushBan()
 		self.assertLogged('Failed to flush bans',
 			'No flush occured, do consistency check',
-			"stdout: %r" % 'ip ban 192.0.2.1',
+			"stdout: %r" % 'ip flush inet6',
+			"stdout: %r" % 'ip check inet6',
 			all=True, wait=True)
+		self.assertNotLogged(
+			"stdout: %r" % 'ip flush inet4',
+			"stdout: %r" % 'ip stop',
+			"stdout: %r" % 'ip start',
+			'Unable to restore environment',
+			all=True)
 
+		# stop, flush succeeds:
+		self.pruneLog('[test-phase end] flush successful')
 		act['actionflush?family=inet6'] = act.actionflush
-
 		self.__actions.stop()
 		self.__actions.join()
+		self.assertLogged(
+			"stdout: %r" % 'ip flush inet6',
+			"stdout: %r" % 'ip stop',    # same for both families
+			'action ip terminated',
+			all=True, wait=True)
+		# no flush for inet4 (already successfully flushed):
+		self.assertNotLogged("ERROR",
+			"stdout: %r" % 'ip flush inet4',
+			'Unban tickets each individualy',
+			all=True)
 
+	def testActionsConsistencyCheckDiffFam(self):
+		# same as testActionsConsistencyCheck, but different start/stop commands for both families
+		act = self.defaultAction({'start':' <family>', 'check':' <family>', 'flush':' <family>', 'stop':' <family>'})
+		# flush for inet6 is intentionally "broken" here - test no unhandled except and invariant check:
+		act['actionflush?family=inet6'] = act.actionflush + '; exit 1'
+		act.actionstart_on_demand = True
+		self.__actions.start()
+		self.assertNotLogged("stdout: %r" % 'ip start')
+
+		self.assertEqual(self.__actions.addBannedIP('192.0.2.1'), 1)
+		self.assertEqual(self.__actions.addBannedIP('2001:db8::1'), 1)
+		self.assertLogged('Ban 192.0.2.1', 'Ban 2001:db8::1',
+			"stdout: %r" % 'ip start inet4',
+			"stdout: %r" % 'ip ban 192.0.2.1',
+			"stdout: %r" % 'ip start inet6',
+			"stdout: %r" % 'ip ban 2001:db8::1',
+			all=True, wait=True)
+
+		# check should fail (so cause stop/start):
+		self.pruneLog('[test-phase 1] simulate inconsistent env')
+		act['actioncheck?family=inet6'] = act.actioncheck + '; exit 1'
+		self.__actions._Actions__flushBan()
+		self.assertLogged(
+			"stdout: %r" % 'ip flush inet4',
+			"stdout: %r" % 'ip flush inet6',
+			'Failed to flush bans',
+			'No flush occured, do consistency check',
+			'Invariant check failed. Trying to restore a sane environment',
+			"stdout: %r" % 'ip stop inet6',
+			'Unable to restore environment',
+			all=True, wait=True)
+		# start/stop should be called for inet6 only:
+		self.assertNotLogged(
+			"stdout: %r" % 'ip stop inet4',
+			all=True)
+
+		# check succeeds:
+		self.pruneLog('[test-phase 2] consistent env')
+		act['actioncheck?family=inet6'] = act.actioncheck
+		self.assertEqual(self.__actions.addBannedIP('2001:db8::1'), 1)
+		self.assertLogged('Ban 2001:db8::1',
+			"stdout: %r" % 'ip start inet6',
+			"stdout: %r" % 'ip ban 2001:db8::1',
+			all=True, wait=True)
+		self.assertNotLogged(
+			"stdout: %r" % 'ip check inet4',
+			"stdout: %r" % 'ip start inet4',
+			all=True)
+
+		self.pruneLog('[test-phase 3] failed flush in consistent env')
+		act['actioncheck?family=inet6'] = act.actioncheck
+		self.__actions._Actions__flushBan()
+		self.assertLogged('Failed to flush bans',
+			'No flush occured, do consistency check',
+			"stdout: %r" % 'ip flush inet6',
+			"stdout: %r" % 'ip check inet6',
+			all=True, wait=True)
+		self.assertNotLogged(
+			"stdout: %r" % 'ip flush inet4',
+			"stdout: %r" % 'ip stop inet4',
+			"stdout: %r" % 'ip start inet4',
+			"stdout: %r" % 'ip stop inet6',
+			"stdout: %r" % 'ip start inet6',
+			all=True)
+
+		# stop, flush succeeds:
+		self.pruneLog('[test-phase end] flush successful')
+		act['actionflush?family=inet6'] = act.actionflush
+		self.__actions.stop()
+		self.__actions.join()
+		self.assertLogged(
+			"stdout: %r" % 'ip flush inet6',
+			"stdout: %r" % 'ip stop inet4',
+			"stdout: %r" % 'ip stop inet6',
+			'action ip terminated',
+			all=True, wait=True)
+		# no flush for inet4 (already successfully flushed):
+		self.assertNotLogged("ERROR",
+			"stdout: %r" % 'ip flush inet4',
+			'Unban tickets each individualy',
+			all=True)