fail2ban.conf: added new fail2ban configuration option "allowipv6" (default auto), can be used to allow or disallow IPv6 interface in fail2ban immediately by start (e. g. if fail2ban starts before network interfaces).

closes gh-2804

config/fail2ban.conf

@@ -55,6 +55,12 @@ socket = /var/run/fail2ban/fail2ban.sock
 #
 pidfile = /var/run/fail2ban/fail2ban.pid
 
+# Option: allowipv6
+# Notes.: Allows IPv6 interface:
+#         Default: auto
+# Values: [ auto yes (on, true, 1) no (off, false, 0) ] Default: auto
+#allowipv6 = auto
+
 # Options: dbfile
 # Notes.: Set the file for the fail2ban persistent data to be stored.
 #         A value of ":memory:" means database is only stored in memory 

fail2ban/client/fail2banreader.py

@@ -53,6 +53,7 @@ class Fail2banReader(ConfigReader):
 		opts = [["string", "loglevel", "INFO" ],
 				["string", "logtarget", "STDERR"],
 				["string", "syslogsocket", "auto"],
+				["string", "allowipv6", "auto"],
 				["string", "dbfile", "/var/lib/fail2ban/fail2ban.sqlite3"],
 				["int",    "dbmaxmatches", None],
 				["string", "dbpurgeage", "1d"]]
@@ -74,6 +75,7 @@ class Fail2banReader(ConfigReader):
 		# Also dbfile should be set before all other database options.
 		# So adding order indices into items, to be stripped after sorting, upon return
 		order = {"thread":0, "syslogsocket":11, "loglevel":12, "logtarget":13,
+			"allowipv6": 14,
 			"dbfile":50, "dbmaxmatches":51, "dbpurgeage":51}
 		stream = list()
 		for opt in self.__opts:

fail2ban/server/ipdns.py

@@ -169,27 +169,31 @@ class DNSUtils:
 		DNSUtils.CACHE_ipToName.set(key, name)
 		return name
 
+	# key find cached own hostnames (this tuple-key cannot be used elsewhere):
+	_getSelfNames_key = ('self','dns')
+
 	@staticmethod
 	def getSelfNames():
 		"""Get own host names of self"""
-		# try find cached own hostnames (this tuple-key cannot be used elsewhere):
-		key = ('self','dns')
-		names = DNSUtils.CACHE_ipToName.get(key)
+		# try find cached own hostnames:
+		names = DNSUtils.CACHE_ipToName.get(DNSUtils._getSelfNames_key)
 		# get it using different ways (a set with names of localhost, hostname, fully qualified):
 		if names is None:
 			names = set([
 				'localhost', DNSUtils.getHostname(False), DNSUtils.getHostname(True)
 			]) - set(['']) # getHostname can return ''
 		# cache and return :
-		DNSUtils.CACHE_ipToName.set(key, names)
+		DNSUtils.CACHE_ipToName.set(DNSUtils._getSelfNames_key, names)
 		return names
 
+	# key to find cached own IPs (this tuple-key cannot be used elsewhere):
+	_getSelfIPs_key = ('self','ips')
+
 	@staticmethod
 	def getSelfIPs():
 		"""Get own IP addresses of self"""
-		# try find cached own IPs (this tuple-key cannot be used elsewhere):
-		key = ('self','ips')
-		ips = DNSUtils.CACHE_nameToIp.get(key)
+		# to find cached own IPs:
+		ips = DNSUtils.CACHE_nameToIp.get(DNSUtils._getSelfIPs_key)
 		# get it using different ways (a set with IPs of localhost, hostname, fully qualified):
 		if ips is None:
 			ips = set()
@@ -199,13 +203,30 @@ class DNSUtils:
 				except Exception as e: # pragma: no cover
 					logSys.warning("Retrieving own IPs of %s failed: %s", hostname, e)
 		# cache and return :
-		DNSUtils.CACHE_nameToIp.set(key, ips)
+		DNSUtils.CACHE_nameToIp.set(DNSUtils._getSelfIPs_key, ips)
 		return ips
 
+	_IPv6IsAllowed = None
+
+	@staticmethod
+	def setIPv6IsAllowed(value):
+		DNSUtils._IPv6IsAllowed = value
+		logSys.debug("IPv6 is %s", ('on' if value else 'off') if value is not None else 'auto')
+		return value
+
+	# key to find cached value of IPv6 allowance (this tuple-key cannot be used elsewhere):
+	_IPv6IsAllowed_key = ('self','ipv6-allowed')
+
 	@staticmethod
 	def IPv6IsAllowed():
-		# return os.path.exists("/proc/net/if_inet6") || any((':' in ip) for ip in DNSUtils.getSelfIPs())
-		return any((':' in ip.ntoa) for ip in DNSUtils.getSelfIPs())
+		if DNSUtils._IPv6IsAllowed is not None:
+			return DNSUtils._IPv6IsAllowed
+		v = DNSUtils.CACHE_nameToIp.get(DNSUtils._IPv6IsAllowed_key)
+		if v is not None:
+			return v
+		v = any((':' in ip.ntoa) for ip in DNSUtils.getSelfIPs())
+		DNSUtils.CACHE_nameToIp.set(DNSUtils._IPv6IsAllowed_key, v)
+		return v
 
 
 ##

fail2ban/server/server.py

@@ -770,6 +770,11 @@ class Server:
 				logSys.info("flush performed on %s" % self.__logTarget)
 			return "flushed"
 			
+	@staticmethod
+	def setIPv6IsAllowed(value):
+		value = _as_bool(value) if value != 'auto' else None
+		return DNSUtils.setIPv6IsAllowed(value)
+
 	def setThreadOptions(self, value):
 		for o, v in value.iteritems():
 			if o == 'stacksize':

fail2ban/server/transmitter.py

@@ -173,6 +173,11 @@ class Transmitter:
 				return self.__server.getSyslogSocket()
 			else:
 				raise Exception("Failed to change syslog socket")
+		elif name == "allowipv6":
+			value = command[1]
+			self.__server.setIPv6IsAllowed(value)
+			if self.__quiet: return
+			return value
 		#Thread
 		elif name == "thread":
 			value = command[1]

fail2ban/tests/clientreadertestcase.py

@@ -975,6 +975,7 @@ class JailsReaderTest(LogCaptureTestCase):
 		  ['set', 'syslogsocket', 'auto'],
 		  ['set', 'loglevel', "INFO"],
 		  ['set', 'logtarget', '/var/log/fail2ban.log'],
+		  ['set', 'allowipv6', 'auto'],
 		  ['set', 'dbfile', '/var/lib/fail2ban/fail2ban.sqlite3'],
 		  ['set', 'dbmaxmatches', 10],
 		  ['set', 'dbpurgeage', '1d'],

fail2ban/tests/servertestcase.py

@@ -35,7 +35,7 @@ import platform
 from ..server.failregex import Regex, FailRegex, RegexException
 from ..server import actions as _actions
 from ..server.server import Server
-from ..server.ipdns import IPAddr
+from ..server.ipdns import DNSUtils, IPAddr
 from ..server.jail import Jail
 from ..server.jailthread import JailThread
 from ..server.ticket import BanTicket
@@ -175,6 +175,19 @@ class Transmitter(TransmitterBase):
 	def testVersion(self):
 		self.assertEqual(self.transm.proceed(["version"]), (0, version.version))
 
+	def testSetIPv6(self):
+		try:
+			self.assertEqual(self.transm.proceed(["set", "allowipv6", 'yes']), (0, 'yes'))
+			self.assertTrue(DNSUtils.IPv6IsAllowed())
+			self.assertLogged("IPv6 is on"); self.pruneLog()
+			self.assertEqual(self.transm.proceed(["set", "allowipv6", 'no']), (0, 'no'))
+			self.assertFalse(DNSUtils.IPv6IsAllowed())
+			self.assertLogged("IPv6 is off"); self.pruneLog()
+		finally:
+			# restore back to auto:
+			self.assertEqual(self.transm.proceed(["set", "allowipv6", "auto"]), (0, "auto"))
+			self.assertLogged("IPv6 is auto"); self.pruneLog()
+
 	def testSleep(self):
 		if not unittest.F2B.fast:
 			t0 = time.time()

man/jail.conf.5

@@ -151,6 +151,11 @@ PID filename.  Default: /var/run/fail2ban/fail2ban.pid
 .br
 This is used to store the process ID of the fail2ban server.
 .TP
+.B allowipv6
+option to allow IPv6 interface - auto, yes (on, true, 1) or no (off, false, 0).  Default: auto
+.br
+This value can be used to declare fail2ban whether IPv6 is allowed or not.
+.TP
 .B dbfile
 Database filename. Default: /var/lib/fail2ban/fail2ban.sqlite3
 .br