prevents to stop processing after first unhandled exception inside processLineAndAdd (filter going idle after 100th error);

fail2ban/server/filter.py

@@ -85,6 +85,8 @@ class Filter(JailThread):
 		self.__lastDate = None
 		## External command
 		self.__ignoreCommand = False
+		## Error counter
+		self.__errors = 0
 
 		self.dateDetector = DateDetector()
 		self.dateDetector.addDefaultTemplate()
@@ -414,26 +416,39 @@ class Filter(JailThread):
 	def processLineAndAdd(self, line, date=None):
 		"""Processes the line for failures and populates failManager
 		"""
-		for element in self.processLine(line, date)[1]:
-			ip = element[1]
-			unixTime = element[2]
-			lines = element[3]
-			fail = {}
-			if len(element) > 4:
-				fail = element[4]
-			logSys.debug("Processing line with time:%s and ip:%s", 
-					unixTime, ip)
-			if unixTime < MyTime.time() - self.getFindTime():
-				logSys.debug("Ignore line since time %s < %s - %s", 
-					unixTime, MyTime.time(), self.getFindTime())
-				break
-			if self.inIgnoreIPList(ip, log_ignore=True):
-				continue
-			logSys.info(
-				"[%s] Found %s - %s", self.jail.name, ip, datetime.datetime.fromtimestamp(unixTime).strftime("%Y-%m-%d %H:%M:%S")
-			)
-			tick = FailTicket(ip, unixTime, lines, data=fail)
-			self.failManager.addFailure(tick)
+		try:
+			for element in self.processLine(line, date)[1]:
+				ip = element[1]
+				unixTime = element[2]
+				lines = element[3]
+				fail = {}
+				if len(element) > 4:
+					fail = element[4]
+				logSys.debug("Processing line with time:%s and ip:%s", 
+						unixTime, ip)
+				if unixTime < MyTime.time() - self.getFindTime():
+					logSys.debug("Ignore line since time %s < %s - %s", 
+						unixTime, MyTime.time(), self.getFindTime())
+					break
+				if self.inIgnoreIPList(ip, log_ignore=True):
+					continue
+				logSys.info(
+					"[%s] Found %s - %s", self.jail.name, ip, datetime.datetime.fromtimestamp(unixTime).strftime("%Y-%m-%d %H:%M:%S")
+				)
+				tick = FailTicket(ip, unixTime, lines, data=fail)
+				self.failManager.addFailure(tick)
+			# reset (halve) error counter (successfully processed line):
+			if self.__errors:
+				self.__errors //= 2
+		except Exception as e:
+			logSys.error("Failed to process line: %r, caught exception: %r", line, e,
+				exc_info=logSys.getEffectiveLevel()<=logging.DEBUG)
+			# incr error counter, stop processing this :
+			self.__errors += 1
+			if self.__errors >= 100:
+				logSys.error("Too many errors at once (%s), going idle", self.__errors)
+				self.__errors //= 2
+				self.idle = True
 
 	##
 	# Returns true if the line should be ignored.

fail2ban/tests/filtertestcase.py

@@ -604,6 +604,27 @@ class LogFileMonitor(LogCaptureTestCase):
 		self.filter.getFailures(self.name)
 		self.assertLogged('Unable to open %s' % self.name)
 
+	def testErrorProcessLine(self):
+		## produce error with not callable processLine:
+		_org_processLine = self.filter.processLine
+		self.filter.processLine = None
+		for i in range(100):
+			self.file.write("line%d\n" % 1)
+		self.file.flush()
+		for i in range(100):
+			self.filter.getFailures(self.name)
+		self.assertLogged('Failed to process line:')
+		self.assertLogged('Too many errors at once')
+		self.pruneLog()
+		self.assertTrue(self.filter.idle)
+		self.filter.idle = False
+		self.filter.getFailures(self.name)
+		self.filter.processLine = _org_processLine
+		self.file.write("line%d\n" % 1)
+		self.file.flush()
+		self.filter.getFailures(self.name)
+		self.assertNotLogged('Failed to process line:')
+
 	def testRemovingFailRegex(self):
 		self.filter.delFailRegex(0)
 		self.assertNotLogged('Cannot remove regular expression. Index 0 is not valid')