*WiP* try to solve RC on jails with too many failures without ban, gh-2945 ...

fail2ban/server/failmanager.py

@@ -124,9 +124,10 @@ class FailManager:
 		return len(self.__failList)
 	
 	def cleanup(self, time):
+		time -= self.__maxTime
 		with self.__lock:
 			todelete = [fid for fid,item in self.__failList.iteritems() \
-				if item.getTime() + self.__maxTime <= time]
+				if item.getTime() <= time]
 			if len(todelete) == len(self.__failList):
 				# remove all:
 				self.__failList = dict()
@@ -140,7 +141,7 @@ class FailManager:
 			else:
 				# create new dictionary without items to be deleted:
 				self.__failList = dict((fid,item) for fid,item in self.__failList.iteritems() \
-					if item.getTime() + self.__maxTime > time)
+					if item.getTime() > time)
 		self.__bgSvc.service()
 	
 	def delFailure(self, fid):

fail2ban/server/filter.py

@@ -93,6 +93,8 @@ class Filter(JailThread):
 		## Store last time stamp, applicable for multi-line
 		self.__lastTimeText = ""
 		self.__lastDate = None
+		## Next service (cleanup) time
+		self.__nextSvcTime = -(1<<63)
 		## if set, treat log lines without explicit time zone to be in this time zone
 		self.__logtimezone = None
 		## Default or preferred encoding (to decode bytes from file or journal):
@@ -114,10 +116,10 @@ class Filter(JailThread):
 		self.checkFindTime = True
 		## shows that filter is in operation mode (processing new messages):
 		self.inOperation = True
-		## if true prevents against retarded banning in case of RC by too many failures (disabled only for test purposes):
-		self.banASAP = True
 		## Ticks counter
 		self.ticks = 0
+		## Processed lines counter
+		self.procLines = 0
 		## Thread name:
 		self.name="f2b/f."+self.jailName
 
@@ -441,12 +443,23 @@ class Filter(JailThread):
 
 	def performBan(self, ip=None):
 		"""Performs a ban for IPs (or given ip) that are reached maxretry of the jail."""
-		try: # pragma: no branch - exception is the only way out
+		while True:
-			while True:
+			try:
 				ticket = self.failManager.toBan(ip)
-				self.jail.putFailTicket(ticket)
+			except FailManagerEmpty:
-		except FailManagerEmpty:
+				break
-			self.failManager.cleanup(MyTime.time())
+			self.jail.putFailTicket(ticket)
+			if ip: break
+		self.performSvc()
+
+	def performSvc(self, force=False):
+		"""Performs a service tasks (clean failure list)."""
+		tm = MyTime.time()
+		# avoid too early clean up:
+		if force or tm >= self.__nextSvcTime:
+			self.__nextSvcTime = tm + 5
+			# clean up failure list:
+			self.failManager.cleanup(tm)
 
 	def addAttempt(self, ip, *matches):
 		"""Generate a failed attempt for ip"""
@@ -694,8 +707,12 @@ class Filter(JailThread):
 				attempts = self.failManager.addFailure(tick)
 				# avoid RC on busy filter (too many failures) - if attempts for IP/ID reached maxretry,
 				# we can speedup ban, so do it as soon as possible:
-				if self.banASAP and attempts >= self.failManager.getMaxRetry():
+				if attempts >= self.failManager.getMaxRetry():
 					self.performBan(ip)
+				self.procLines += 1
+			# every 100 lines check need to perform service tasks:
+			if self.procLines % 100 == 0:
+				self.performSvc()
 			# reset (halve) error counter (successfully processed line):
 			if self._errors:
 				self._errors //= 2
@@ -1064,6 +1081,7 @@ class FileFilter(Filter):
 	# is created and is added to the FailManager.
 
 	def getFailures(self, filename, inOperation=None):
+		if self.idle: return False
 		log = self.getLog(filename)
 		if log is None:
 			logSys.error("Unable to get failures in %s", filename)

fail2ban/server/filtergamin.py

@@ -55,7 +55,6 @@ class FilterGamin(FileFilter):
 
 	def __init__(self, jail):
 		FileFilter.__init__(self, jail)
-		self.__modified = False
 		# Gamin monitor
 		self.monitor = gamin.WatchMonitor()
 		fd = self.monitor.get_fd()
@@ -67,21 +66,9 @@ class FilterGamin(FileFilter):
 		logSys.log(4, "Got event: " + repr(event) + " for " + path)
 		if event in (gamin.GAMCreated, gamin.GAMChanged, gamin.GAMExists):
 			logSys.debug("File changed: " + path)
-			self.__modified = True
 
 		self.ticks += 1
-		self._process_file(path)
-
-	def _process_file(self, path):
-		"""Process a given file
-
-		TODO -- RF:
-		this is a common logic and must be shared/provided by FileFilter
-		"""
 		self.getFailures(path)
-		if not self.banASAP: # pragma: no cover
-			self.performBan()
-		self.__modified = False
 
 	##
 	# Add a log file path
@@ -128,6 +115,9 @@ class FilterGamin(FileFilter):
 			Utils.wait_for(lambda: not self.active or self._handleEvents(),
 				self.sleeptime)
 			self.ticks += 1
+			if self.ticks % 10 == 0:
+				self.performSvc()
+
 		logSys.debug("[%s] filter terminated", self.jailName)
 		return True
 

fail2ban/server/filterpoll.py

@@ -27,9 +27,7 @@ __license__ = "GPL"
 import os
 import time
 
-from .failmanager import FailManagerEmpty
 from .filter import FileFilter
-from .mytime import MyTime
 from .utils import Utils
 from ..helpers import getLogger, logging
 
@@ -55,7 +53,6 @@ class FilterPoll(FileFilter):
 
 	def __init__(self, jail):
 		FileFilter.__init__(self, jail)
-		self.__modified = False
 		## The time of the last modification of the file.
 		self.__prevStats = dict()
 		self.__file404Cnt = dict()
@@ -115,13 +112,10 @@ class FilterPoll(FileFilter):
 					break
 				for filename in modlst:
 					self.getFailures(filename)
-					self.__modified = True
 
 				self.ticks += 1
-				if self.__modified:
+				if self.ticks % 10 == 0:
-					if not self.banASAP: # pragma: no cover
+					self.performSvc()
-						self.performBan()
-					self.__modified = False
 			except Exception as e: # pragma: no cover
 				if not self.active: # if not active - error by stop...
 					break

fail2ban/server/filterpyinotify.py

@@ -75,7 +75,6 @@ class FilterPyinotify(FileFilter):
 
 	def __init__(self, jail):
 		FileFilter.__init__(self, jail)
-		self.__modified = False
 		# Pyinotify watch manager
 		self.__monitor = pyinotify.WatchManager()
 		self.__notifier = None
@@ -140,9 +139,6 @@ class FilterPyinotify(FileFilter):
 		"""
 		if not self.idle:
 			self.getFailures(path)
-			if not self.banASAP: # pragma: no cover
-				self.performBan()
-			self.__modified = False
 
 	def _addPending(self, path, reason, isDir=False):
 		if path not in self.__pending:
@@ -352,9 +348,14 @@ class FilterPyinotify(FileFilter):
 					if not self.active: break
 					self.__notifier.read_events()
 
+				self.ticks += 1
+
 				# check pending files/dirs (logrotate ready):
-				if not self.idle:
+				if self.idle:
-					self._checkPending()
+					continue
+				self._checkPending()
+				if self.ticks % 10 == 0:
+					self.performSvc()
 
 			except Exception as e: # pragma: no cover
 				if not self.active: # if not active - error by stop...
@@ -364,8 +365,6 @@ class FilterPyinotify(FileFilter):
 				# incr common error counter:
 				self.commonError()
 			
-			self.ticks += 1
-
 		logSys.debug("[%s] filter exited (pyinotifier)", self.jailName)
 		self.__notifier = None
 

fail2ban/server/filtersystemd.py

@@ -322,13 +322,12 @@ class FilterSystemd(JournalFilter): # pragma: systemd no cover
 							break
 					else:
 						break
-				if self.__modified:
+				self.__modified = 0
-					if not self.banASAP: # pragma: no cover
+				if self.ticks % 10 == 0:
-						self.performBan()
+					self.performSvc()
-					self.__modified = 0
+				# update position in log (time and iso string):
-					# update position in log (time and iso string):
+				if self.jail.database is not None:
-					if self.jail.database is not None:
+					self.jail.database.updateJournal(self.jail, 'systemd-journal', line[1], line[0][1])
-						self.jail.database.updateJournal(self.jail, 'systemd-journal', line[1], line[0][1])
 			except Exception as e: # pragma: no cover
 				if not self.active: # if not active - error by stop...
 					break

fail2ban/tests/filtertestcase.py

@@ -800,7 +800,6 @@ class LogFileMonitor(LogCaptureTestCase):
 		_, self.name = tempfile.mkstemp('fail2ban', 'monitorfailures')
 		self.file = open(self.name, 'a')
 		self.filter = FilterPoll(DummyJail())
-		self.filter.banASAP = False # avoid immediate ban in this tests
 		self.filter.addLogPath(self.name, autoSeek=False)
 		self.filter.active = True
 		self.filter.addFailRegex(r"(?:(?:Authentication failure|Failed [-/\w+]+) for(?: [iI](?:llegal|nvalid) user)?|[Ii](?:llegal|nvalid) user|ROOT LOGIN REFUSED) .*(?: from|FROM) <HOST>")
@@ -952,15 +951,18 @@ class LogFileMonitor(LogCaptureTestCase):
 		self.file.close()
 		self.file = _copy_lines_between_files(GetFailures.FILENAME_01, self.name,
 											  n=14, mode='w')
+		print('=========='*10)
 		self.filter.getFailures(self.name)
+		print('=========='*10)
 		self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
 		self.assertEqual(self.filter.failManager.getFailTotal(), 2)
 
 		# move aside, but leaving the handle still open...
+		print('=========='*10)
 		os.rename(self.name, self.name + '.bak')
 		_copy_lines_between_files(GetFailures.FILENAME_01, self.name, skip=14, n=1).close()
 		self.filter.getFailures(self.name)
-		_assert_correct_last_attempt(self, self.filter, GetFailures.FAILURES_01)
+		#_assert_correct_last_attempt(self, self.filter, GetFailures.FAILURES_01)
 		self.assertEqual(self.filter.failManager.getFailTotal(), 3)
 
 
@@ -1018,7 +1020,6 @@ def get_monitor_failures_testcase(Filter_):
 			self.file = open(self.name, 'a')
 			self.jail = DummyJail()
 			self.filter = Filter_(self.jail)
-			self.filter.banASAP = False # avoid immediate ban in this tests
 			self.filter.addLogPath(self.name, autoSeek=False)
 			# speedup search using exact date pattern:
 			self.filter.setDatePattern(r'^(?:%a )?%b %d %H:%M:%S(?:\.%f)?(?: %ExY)?')
@@ -1277,14 +1278,14 @@ def get_monitor_failures_testcase(Filter_):
 			# tail written before, so let's not copy anything yet
 			#_copy_lines_between_files(GetFailures.FILENAME_01, self.name, n=100)
 			# we should detect the failures
-			self.assert_correct_last_attempt(GetFailures.FAILURES_01, count=6) # was needed if we write twice above
+			self.assert_correct_last_attempt(GetFailures.FAILURES_01, count=3) # was needed if we write twice above
 
 			# now copy and get even more
 			_copy_lines_between_files(GetFailures.FILENAME_01, self.file, skip=12, n=3)
 			# check for 3 failures (not 9), because 6 already get above...
 			self.assert_correct_last_attempt(GetFailures.FAILURES_01)
 			# total count in this test:
-			self.assertEqual(self.filter.failManager.getFailTotal(), 12)
+			self.assertEqual(self.filter.failManager.getFailTotal(), 9)
 
 	cls = MonitorFailures
 	cls.__qualname__ = cls.__name__ = "MonitorFailures<%s>(%s)" \
@@ -1316,7 +1317,6 @@ def get_monitor_failures_journal_testcase(Filter_): # pragma: systemd no cover
 		def _initFilter(self, **kwargs):
 			self._getRuntimeJournal() # check journal available
 			self.filter = Filter_(self.jail, **kwargs)
-			self.filter.banASAP = False # avoid immediate ban in this tests
 			self.filter.addJournalMatch([
 				"SYSLOG_IDENTIFIER=fail2ban-testcases",
 				"TEST_FIELD=1",
@@ -1570,7 +1570,6 @@ class GetFailures(LogCaptureTestCase):
 		setUpMyTime()
 		self.jail = DummyJail()
 		self.filter = FileFilter(self.jail)
-		self.filter.banASAP = False # avoid immediate ban in this tests
 		self.filter.active = True
 		# speedup search using exact date pattern:
 		self.filter.setDatePattern(r'^(?:%a )?%b %d %H:%M:%S(?:\.%f)?(?: %ExY)?')
@@ -1771,7 +1770,6 @@ class GetFailures(LogCaptureTestCase):
 			self.pruneLog("[test-phase useDns=%s]" % useDns)
 			jail = DummyJail()
 			filter_ = FileFilter(jail, useDns=useDns)
-			filter_.banASAP = False # avoid immediate ban in this tests
 			filter_.active = True
 			filter_.failManager.setMaxRetry(1)	# we might have just few failures