[Web] Add identity provider outage translation, remove unused needs_restart

data/Dockerfiles/netfilter/docker-entrypoint.sh

@@ -17,8 +17,8 @@ if [ $nftables_found -gt $iptables_found ]; then
 fi
 
 if [ $nftables_found -eq 0 ] && [ $nftables_found -eq $iptables_found ]; then
-  nftables_lines=$(nft list ruleset | wc -l)
-  iptables_lines=$(iptables-save | wc -l)
+  nftables_lines=$(nft list ruleset 2>/dev/null | wc -l)
+  iptables_lines=$(iptables-save 2>/dev/null | wc -l)
   if [ $nftables_lines -gt $iptables_lines ]; then
     backend=nftables
   else

data/Dockerfiles/netfilter/main.py

@@ -1,548 +1,57 @@
-#!/usr/bin/env python3
-
-DEBUG = False
-
-import re
-import os
-import sys
-import time
-import atexit
-import signal
-import ipaddress
-from collections import Counter
-from random import randint
-from threading import Thread
-from threading import Lock
-import redis
-import json
-import dns.resolver
-import dns.exception
-import uuid
-from modules.Logger import Logger
-from modules.IPTables import IPTables
-from modules.NFTables import NFTables
-
-def logdebug(msg):
-  if DEBUG:
-    logger.logInfo("DEBUG: %s" % msg)
-
-# Globals
-WHITELIST = []
-BLACKLIST = []
-bans = {}
-quit_now = False
-exit_code = 0
-lock = Lock()
-chain_name = "MAILCOW"
-r = None
-pubsub = None
-clear_before_quit = False
-
-def refreshF2boptions():
-  global f2boptions
-  global quit_now
-  global exit_code
-  f2boptions = {}
-
-  if not r.get('F2B_OPTIONS'):
-    f2boptions['ban_time'] = r.get('F2B_BAN_TIME')
-    f2boptions['max_ban_time'] = r.get('F2B_MAX_BAN_TIME')
-    f2boptions['ban_time_increment'] = r.get('F2B_BAN_TIME_INCREMENT')
-    f2boptions['max_attempts'] = r.get('F2B_MAX_ATTEMPTS')
-    f2boptions['retry_window'] = r.get('F2B_RETRY_WINDOW')
-    f2boptions['netban_ipv4'] = r.get('F2B_NETBAN_IPV4')
-    f2boptions['netban_ipv6'] = r.get('F2B_NETBAN_IPV6')
-  else:
-    try:
-      f2boptions = json.loads(r.get('F2B_OPTIONS'))
-    except ValueError as e:
-      logger.logCrit(
-        'Error loading F2B options: F2B_OPTIONS is not json. Exception: %s' % e)
-      quit_now = True
-      exit_code = 2
-
-  verifyF2boptions(f2boptions)
-  r.set('F2B_OPTIONS', json.dumps(f2boptions, ensure_ascii=False))
-
-def verifyF2boptions(f2boptions):
-  verifyF2boption(f2boptions, 'ban_time', 1800)
-  verifyF2boption(f2boptions, 'max_ban_time', 10000)
-  verifyF2boption(f2boptions, 'ban_time_increment', True)
-  verifyF2boption(f2boptions, 'max_attempts', 10)
-  verifyF2boption(f2boptions, 'retry_window', 600)
-  verifyF2boption(f2boptions, 'netban_ipv4', 32)
-  verifyF2boption(f2boptions, 'netban_ipv6', 128)
-  verifyF2boption(f2boptions, 'banlist_id', str(uuid.uuid4()))
-  verifyF2boption(f2boptions, 'manage_external', 0)
-
-def verifyF2boption(f2boptions, f2boption, f2bdefault):
-  f2boptions[f2boption] = f2boptions[f2boption] if f2boption in f2boptions and f2boptions[f2boption] is not None else f2bdefault
-
-def refreshF2bregex():
-  global f2bregex
-  global quit_now
-  global exit_code
-  if not r.get('F2B_REGEX'):
-    f2bregex = {}
-    f2bregex[1] = r'mailcow UI: Invalid password for .+ by ([0-9a-f\.:]+)'
-    f2bregex[2] = r'Rspamd UI: Invalid password by ([0-9a-f\.:]+)'
-    f2bregex[3] = r'warning: .*\[([0-9a-f\.:]+)\]: SASL .+ authentication failed: (?!.*Connection lost to authentication server).+'
-    f2bregex[4] = r'warning: non-SMTP command from .*\[([0-9a-f\.:]+)]:.+'
-    f2bregex[5] = r'NOQUEUE: reject: RCPT from \[([0-9a-f\.:]+)].+Protocol error.+'
-    f2bregex[6] = r'\w+\([^,]+,([0-9a-f\.:]+),<[^>]+>\): Password mismatch \(SHA1 of given password: [a-f0-9]+\)'
-    f2bregex[7] = r'\w+\([^,]+,([0-9a-f\.:]+),<[^>]+>\): unknown user \(SHA1 of given password: [a-f0-9]+\)'
-    f2bregex[8] = r'SOGo.+ Login from \'([0-9a-f\.:]+)\' for user .+ might not have worked'
-    f2bregex[9] = r'([0-9a-f\.:]+) \"GET \/SOGo\/.* HTTP.+\" 403 .+'
-    r.set('F2B_REGEX', json.dumps(f2bregex, ensure_ascii=False))
-  else:
-    try:
-      f2bregex = {}
-      f2bregex = json.loads(r.get('F2B_REGEX'))
-    except ValueError:
-      logger.logCrit('Error loading F2B options: F2B_REGEX is not json')
-      quit_now = True
-      exit_code = 2
-
-def get_ip(address):
-  ip = ipaddress.ip_address(address)
-  if type(ip) is ipaddress.IPv6Address and ip.ipv4_mapped:
-    ip = ip.ipv4_mapped
-  if ip.is_private or ip.is_loopback:
-    return False
-
-  return ip
-
-def ban(address):
-  global f2boptions
-  global lock
-  logdebug("ban() called with address=%s" % address)
-  refreshF2boptions()
-  MAX_ATTEMPTS = int(f2boptions['max_attempts'])
-  RETRY_WINDOW = int(f2boptions['retry_window'])
-  NETBAN_IPV4 = '/' + str(f2boptions['netban_ipv4'])
-  NETBAN_IPV6 = '/' + str(f2boptions['netban_ipv6'])
-
-  ip = get_ip(address)
-  if not ip:
-    logdebug("No valid IP -- skipping ban()")
-    return
-  address = str(ip)
-  self_network = ipaddress.ip_network(address)
-
-  with lock:
-    temp_whitelist = set(WHITELIST)
-    logdebug("Checking if %s overlaps with any WHITELIST entries" % self_network)
-    if temp_whitelist:
-      for wl_key in temp_whitelist:
-        wl_net = ipaddress.ip_network(wl_key, False)
-        logdebug("Checking overlap between %s and %s" % (self_network, wl_net))
-        if wl_net.overlaps(self_network):
-          logger.logInfo(
-            'Address %s is allowlisted by rule %s' % (self_network, wl_net))
-          return
-
-  net = ipaddress.ip_network(
-    (address + (NETBAN_IPV4 if type(ip) is ipaddress.IPv4Address else NETBAN_IPV6)), strict=False)
-  net = str(net)
-  logdebug("Ban net: %s" % net)
-
-  if not net in bans:
-    bans[net] = {'attempts': 0, 'last_attempt': 0, 'ban_counter': 0}
-    logdebug("Initing new ban counter for %s" % net)
-
-  current_attempt = time.time()
-  logdebug("Current attempt ts=%s, previous: %s, retry_window: %s" %
-           (current_attempt, bans[net]['last_attempt'], RETRY_WINDOW))
-  if current_attempt - bans[net]['last_attempt'] > RETRY_WINDOW:
-    bans[net]['attempts'] = 0
-    logdebug("Ban counter for %s reset as window expired" % net)
-
-  bans[net]['attempts'] += 1
-  bans[net]['last_attempt'] = current_attempt
-  logdebug("%s attempts now %d" % (net, bans[net]['attempts']))
-
-  if bans[net]['attempts'] >= MAX_ATTEMPTS:
-    cur_time = int(round(time.time()))
-    NET_BAN_TIME = calcNetBanTime(bans[net]['ban_counter'])
-    logger.logCrit('Banning %s for %d minutes' % (net, NET_BAN_TIME / 60 ))
-    if type(ip) is ipaddress.IPv4Address and int(f2boptions['manage_external']) != 1:
-      with lock:
-        logdebug("Calling tables.banIPv4(%s)" % net)
-        tables.banIPv4(net)
-    elif int(f2boptions['manage_external']) != 1:
-      with lock:
-        logdebug("Calling tables.banIPv6(%s)" % net)
-        tables.banIPv6(net)
-
-    logdebug("Updating F2B_ACTIVE_BANS[%s]=%d" %
-              (net, cur_time + NET_BAN_TIME))
-    r.hset('F2B_ACTIVE_BANS', '%s' % net, cur_time + NET_BAN_TIME)
-  else:
-    logger.logWarn('%d more attempts in the next %d seconds until %s is banned' % (
-      MAX_ATTEMPTS - bans[net]['attempts'], RETRY_WINDOW, net))
-
-def unban(net):
-  global lock
-  logdebug("Calling unban() with net=%s" % net)
-  if not net in bans:
-    logger.logInfo(
-      '%s is not banned, skipping unban and deleting from queue (if any)' % net)
-    r.hdel('F2B_QUEUE_UNBAN', '%s' % net)
-    return
-  logger.logInfo('Unbanning %s' % net)
-  if type(ipaddress.ip_network(net)) is ipaddress.IPv4Network:
-    with lock:
-      logdebug("Calling tables.unbanIPv4(%s)" % net)
-      tables.unbanIPv4(net)
-  else:
-    with lock:
-      logdebug("Calling tables.unbanIPv6(%s)" % net)
-      tables.unbanIPv6(net)
-  r.hdel('F2B_ACTIVE_BANS', '%s' % net)
-  r.hdel('F2B_QUEUE_UNBAN', '%s' % net)
-  if net in bans:
-    logdebug("Unban for %s, setting attempts=0, ban_counter+=1" % net)
-    bans[net]['attempts'] = 0
-    bans[net]['ban_counter'] += 1
-
-def permBan(net, unban=False):
-  global f2boptions
-  global lock
-
-  is_unbanned = False
-  is_banned = False
-  if type(ipaddress.ip_network(net, strict=False)) is ipaddress.IPv4Network:
-    with lock:
-      if unban:
-        is_unbanned = tables.unbanIPv4(net)
-      elif int(f2boptions['manage_external']) != 1:
-        is_banned = tables.banIPv4(net)
-  else:
-    with lock:
-      if unban:
-        is_unbanned = tables.unbanIPv6(net)
-      elif int(f2boptions['manage_external']) != 1:
-        is_banned = tables.banIPv6(net)
-
-
-  if is_unbanned:
-    r.hdel('F2B_PERM_BANS', '%s' % net)
-    logger.logCrit('Removed host/network %s from denylist' % net)
-  elif is_banned:
-    r.hset('F2B_PERM_BANS', '%s' % net, int(round(time.time())))
-    logger.logCrit('Added host/network %s to denylist' % net)
-
-def clear():
-  global lock
-  logger.logInfo('Clearing all bans')
-  for net in bans.copy():
-    logdebug("Unbanning net: %s" % net)
-    unban(net)
-  with lock:
-    logdebug("Clearing IPv4/IPv6 table")
-    tables.clearIPv4Table()
-    tables.clearIPv6Table()
-    try:
-      if r is not None:
-        r.delete('F2B_ACTIVE_BANS')
-        r.delete('F2B_PERM_BANS')
-    except Exception as ex:
-      logger.logWarn('Error clearing redis keys F2B_ACTIVE_BANS and F2B_PERM_BANS: %s' % ex)
-
-def watch():
-  global pubsub
-  global quit_now
-  global exit_code
-
-  logger.logInfo('Watching Redis channel F2B_CHANNEL')
-  pubsub.subscribe('F2B_CHANNEL')
-
-  while not quit_now:
-    try:
-      for item in pubsub.listen():
-        refreshF2bregex()
-        for rule_id, rule_regex in f2bregex.items():
-          if item['data'] and item['type'] == 'message':
-            try:
-              result = re.search(rule_regex, item['data'])
-            except re.error:
-              result = False
-            if result:
-              addr = result.group(1)
-              ip = ipaddress.ip_address(addr)
-              if ip.is_private or ip.is_loopback:
-                continue
-              logger.logWarn('%s matched rule id %s (%s)' % (addr, rule_id, item['data']))
-              ban(addr)
-    except Exception as ex:
-      logger.logWarn('Error reading log line from pubsub: %s' % ex)
-      pubsub = None
-      quit_now = True
-      exit_code = 2
-
-def snat4(snat_target):
-  global lock
-  global quit_now
-
-  while not quit_now:
-    time.sleep(10)
-    with lock:
-      tables.snat4(snat_target, os.getenv('IPV4_NETWORK', '172.22.1') + '.0/24')
-
-def snat6(snat_target):
-  global lock
-  global quit_now
-
-  while not quit_now:
-    time.sleep(10)
-    with lock:
-      tables.snat6(snat_target, os.getenv('IPV6_NETWORK', 'fd4d:6169:6c63:6f77::/64'))
-
-def autopurge():
-  global f2boptions
-  logdebug("autopurge thread started")
-  while not quit_now:
-    logdebug("autopurge tick")
-    time.sleep(10)
-    refreshF2boptions()
-    MAX_ATTEMPTS = int(f2boptions['max_attempts'])
-    QUEUE_UNBAN = r.hgetall('F2B_QUEUE_UNBAN')
-    logdebug("QUEUE_UNBAN: %s" % QUEUE_UNBAN)
-    if QUEUE_UNBAN:
-      for net in QUEUE_UNBAN:
-        logdebug("Autopurge: unbanning queued net: %s" % net)
-        unban(str(net))
-    # Only check expiry for actively banned IPs:
-    active_bans = r.hgetall('F2B_ACTIVE_BANS')
-    now = time.time()
-    for net_str, expire_str in active_bans.items():
-      logdebug("Checking ban expiry for (actively banned): %s" % net_str)
-      # Defensive: always process if timer missing or expired
-      try:
-        expire = float(expire_str)
-      except Exception:
-        logdebug("Invalid expire time for %s; unbanning" % net_str)
-        unban(net_str)
-        continue
-      time_left = expire - now
-      logdebug("Time left for %s: %.1f seconds" % (net_str, time_left))
-      if time_left <= 0:
-        logdebug("Ban expired for %s" % net_str)
-        unban(net_str)
-
-def mailcowChainOrder():
-  global lock
-  global quit_now
-  global exit_code
-  while not quit_now:
-    time.sleep(10)
-    with lock:
-      quit_now, exit_code = tables.checkIPv4ChainOrder()
-      if quit_now: return
-      quit_now, exit_code = tables.checkIPv6ChainOrder()
-
-def calcNetBanTime(ban_counter):
-  global f2boptions
-
-  BAN_TIME = int(f2boptions['ban_time'])
-  MAX_BAN_TIME = int(f2boptions['max_ban_time'])
-  BAN_TIME_INCREMENT = bool(f2boptions['ban_time_increment'])
-  NET_BAN_TIME = BAN_TIME if not BAN_TIME_INCREMENT else BAN_TIME * 2 ** ban_counter
-  NET_BAN_TIME = max([BAN_TIME, min([NET_BAN_TIME, MAX_BAN_TIME])])
-  return NET_BAN_TIME
-
-def isIpNetwork(address):
-  try:
-    ipaddress.ip_network(address, False)
-  except ValueError:
-    return False
-  return True
-
-def genNetworkList(list):
-  resolver = dns.resolver.Resolver()
-  hostnames = []
-  networks = []
-  for key in list:
-    if isIpNetwork(key):
-      networks.append(key)
-    else:
-      hostnames.append(key)
-  for hostname in hostnames:
-    hostname_ips = []
-    for rdtype in ['A', 'AAAA']:
-      try:
-        answer = resolver.resolve(qname=hostname, rdtype=rdtype, lifetime=3)
-      except dns.exception.Timeout:
-        logger.logInfo('Hostname %s timedout on resolve' % hostname)
-        break
-      except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
-        continue
-      except dns.exception.DNSException as dnsexception:
-        logger.logInfo('%s' % dnsexception)
-        continue
-      for rdata in answer:
-        hostname_ips.append(rdata.to_text())
-    networks.extend(hostname_ips)
-  return set(networks)
-
-def whitelistUpdate():
-  global lock
-  global quit_now
-  global WHITELIST
-  while not quit_now:
-    start_time = time.time()
-    list = r.hgetall('F2B_WHITELIST')
-    new_whitelist = []
-    if list:
-      new_whitelist = genNetworkList(list)
-    with lock:
-      if Counter(new_whitelist) != Counter(WHITELIST):
-        WHITELIST = new_whitelist
-        logger.logInfo('Allowlist was changed, it has %s entries' % len(WHITELIST))
-    time.sleep(60.0 - ((time.time() - start_time) % 60.0))
-
-def blacklistUpdate():
-  global quit_now
-  global BLACKLIST
-  while not quit_now:
-    start_time = time.time()
-    list = r.hgetall('F2B_BLACKLIST')
-    new_blacklist = []
-    if list:
-      new_blacklist = genNetworkList(list)
-    if Counter(new_blacklist) != Counter(BLACKLIST):
-      addban = set(new_blacklist).difference(BLACKLIST)
-      delban = set(BLACKLIST).difference(new_blacklist)
-      BLACKLIST = new_blacklist
-      logger.logInfo('Denylist was changed, it has %s entries' % len(BLACKLIST))
-      if addban:
-        for net in addban:
-          permBan(net=net)
-      if delban:
-        for net in delban:
-          permBan(net=net, unban=True)
-    time.sleep(60.0 - ((time.time() - start_time) % 60.0))
-
-def sigterm_quit(signum, frame):
-  global clear_before_quit
-  logdebug("SIGTERM received, setting clear_before_quit to True and exiting")
-  clear_before_quit = True
-  sys.exit(exit_code)
-
-def before_quit():
-  logdebug("before_quit called, clear_before_quit=%s" % clear_before_quit)
-  if clear_before_quit:
-    clear()
-  if pubsub is not None:
-    pubsub.unsubscribe()
-
-if __name__ == '__main__':
-  logger = Logger()
-  logdebug("Sys.argv: %s" % sys.argv)
-  atexit.register(before_quit)
-  signal.signal(signal.SIGTERM, sigterm_quit)
-
-  backend = sys.argv[1]
-  logdebug("Backend: %s" % backend)
-  if backend == "nftables":
-    logger.logInfo('Using NFTables backend')
-    tables = NFTables(chain_name, logger)
-  else:
-    logger.logInfo('Using IPTables backend')
-    logger.logWarn(
-        "DEPRECATION: iptables-legacy is deprecated and will be removed in future releases. "
-        "Please switch to nftables on your host to ensure complete compatibility."
-    )
-    time.sleep(5)
-    tables = IPTables(chain_name, logger)
-
-  clear()
-  logger.logInfo("Initializing mailcow netfilter chain")
-  tables.initChainIPv4()
-  tables.initChainIPv6()
-
-  if os.getenv("DISABLE_NETFILTER_ISOLATION_RULE", "").lower() in ("y", "yes"):
-    logger.logInfo(f"Skipping {chain_name} isolation")
-  else:
-    logger.logInfo(f"Setting {chain_name} isolation")
-    tables.create_mailcow_isolation_rule("br-mailcow", [3306, 6379, 8983, 12345], os.getenv("MAILCOW_REPLICA_IP"))
-
-  # connect to redis
-  while True:
-    try:
-      redis_slaveof_ip = os.getenv('REDIS_SLAVEOF_IP', '')
-      redis_slaveof_port = os.getenv('REDIS_SLAVEOF_PORT', '')
-      logdebug(
-        "Connecting redis (SLAVEOF_IP:%s, PORT:%s)" % (redis_slaveof_ip, redis_slaveof_port))
-      if "".__eq__(redis_slaveof_ip):
-        r = redis.StrictRedis(
-          host=os.getenv('IPV4_NETWORK', '172.22.1') + '.249', decode_responses=True, port=6379, db=0, password=os.environ['REDISPASS'])
-      else:
-        r = redis.StrictRedis(
-          host=redis_slaveof_ip, decode_responses=True, port=redis_slaveof_port, db=0, password=os.environ['REDISPASS'])
-      r.ping()
-      pubsub = r.pubsub()
-    except Exception as ex:
-      logdebug(
-        'Redis connection failed: %s - trying again in 3 seconds' % (ex))
-      time.sleep(3)
-    else:
-      break
-  logger.set_redis(r)
-  logdebug("Redis connection established, setting up F2B keys")
-
-  if r.exists('F2B_LOG'):
-    logdebug("Renaming F2B_LOG to NETFILTER_LOG")
-    r.rename('F2B_LOG', 'NETFILTER_LOG')
-  r.delete('F2B_ACTIVE_BANS')
-  r.delete('F2B_PERM_BANS')
-
-  refreshF2boptions()
-
-  watch_thread = Thread(target=watch)
-  watch_thread.daemon = True
-  watch_thread.start()
-
-  if os.getenv('SNAT_TO_SOURCE') and os.getenv('SNAT_TO_SOURCE') != 'n':
-    try:
-      snat_ip = os.getenv('SNAT_TO_SOURCE')
-      snat_ipo = ipaddress.ip_address(snat_ip)
-      if type(snat_ipo) is ipaddress.IPv4Address:
-        snat4_thread = Thread(target=snat4, args=(snat_ip,))
-        snat4_thread.daemon = True
-        snat4_thread.start()
-    except ValueError:
-      print(os.getenv('SNAT_TO_SOURCE') + ' is not a valid IPv4 address')
-
-  if os.getenv('SNAT6_TO_SOURCE') and os.getenv('SNAT6_TO_SOURCE') != 'n':
-    try:
-      snat_ip = os.getenv('SNAT6_TO_SOURCE')
-      snat_ipo = ipaddress.ip_address(snat_ip)
-      if type(snat_ipo) is ipaddress.IPv6Address:
-        snat6_thread = Thread(target=snat6,args=(snat_ip,))
-        snat6_thread.daemon = True
-        snat6_thread.start()
-    except ValueError:
-      print(os.getenv('SNAT6_TO_SOURCE') + ' is not a valid IPv6 address')
-
-  autopurge_thread = Thread(target=autopurge)
-  autopurge_thread.daemon = True
-  autopurge_thread.start()
-
-  mailcowchainwatch_thread = Thread(target=mailcowChainOrder)
-  mailcowchainwatch_thread.daemon = True
-  mailcowchainwatch_thread.start()
-
-  blacklistupdate_thread = Thread(target=blacklistUpdate)
-  blacklistupdate_thread.daemon = True
-  blacklistupdate_thread.start()
-
-  whitelistupdate_thread = Thread(target=whitelistUpdate)
-  whitelistupdate_thread.daemon = True
-  whitelistupdate_thread.start()
-
-  while not quit_now:
-    time.sleep(0.5)
-
-  logdebug("Exiting with code %s" % exit_code)
-  sys.exit(exit_code)
+#!/usr/bin/env python3
+
+import os
+import sys
+import time
+import signal
+import redis
+from modules.Logger import Logger
+from modules.Netfilter import Netfilter
+
+chain_name = "MAILCOW"
+
+def connect_redis(logger):
+  if os.getenv('REDIS_SLAVEOF_IP', ''):
+    host, port = os.getenv('REDIS_SLAVEOF_IP'), os.getenv('REDIS_SLAVEOF_PORT', '')
+  else:
+    host, port = os.getenv('IPV4_NETWORK', '172.22.1') + '.249', 6379
+  while True:
+    try:
+      # log lines are not always valid UTF-8
+      r = redis.StrictRedis(host=host, port=port, db=0, password=os.environ['REDISPASS'], decode_responses=True,
+                            encoding_errors='replace', socket_connect_timeout=5, socket_timeout=30)
+      r.ping()
+      return r
+    except Exception as ex:
+      logger.logWarn('Redis connection failed: %s - trying again in 3 seconds' % ex)
+      time.sleep(3)
+
+if __name__ == '__main__':
+  logger = Logger()
+
+  backend = sys.argv[1]
+  if backend == "nftables":
+    from modules.NFTables import NFTables
+    logger.logInfo('Using NFTables backend')
+    tables = NFTables(chain_name, logger)
+  else:
+    from modules.IPTables import IPTables
+    logger.logInfo('Using IPTables backend')
+    logger.logWarn(
+        "DEPRECATION: iptables-legacy is deprecated and will be removed in future releases. "
+        "Please switch to nftables on your host to ensure complete compatibility."
+    )
+    time.sleep(5)
+    tables = IPTables(chain_name, logger)
+
+  r = connect_redis(logger)
+  logger.set_redis(r)
+
+  netfilter = Netfilter(r, tables, logger, os.environ)
+  signal.signal(signal.SIGTERM, lambda signum, frame: netfilter.stop.set())
+  signal.signal(signal.SIGINT, lambda signum, frame: netfilter.stop.set())
+
+  netfilter.setup()
+  netfilter.run()
+  netfilter.shutdown()
+  sys.exit(0)

data/Dockerfiles/netfilter/modules/BanManager.py

@@ -0,0 +1,236 @@
+import ipaddress
+import time
+
+
+class BanManager:
+  """Failed attempts and automatic bans (Redis F2B_ACTIVE_BANS: network => expiry), the denylist (F2B_PERM_BANS)
+  and the allowlist. Every firewall change runs under lock, which also guards the other firewall work of netfilter
+  (chains, SNAT), because the backends must not run in parallel."""
+
+  # Seconds failed logins of an address are not counted after the auth backend was unavailable for it
+  EXCUSE_TIME = 60
+
+  def __init__(self, r, tables, options, logger, lock, clock=time.time):
+    self.r = r
+    self.tables = tables
+    self.options = options
+    self.logger = logger
+    self.lock = lock
+    self.clock = clock
+    self.counters = {}
+    self.banned = set()
+    self.denied = {}
+    self.allowlist = []
+    self.excused = {}
+
+  def excuse(self, ip):
+    """The auth backend could not check a login of ip: its failed logins are not counted for EXCUSE_TIME"""
+    with self.lock:
+      now = self.clock()
+      if self.excused.get(ip, 0) <= now:
+        self.logger.logWarn('Auth backend unavailable for %s, its failed logins of the next %d seconds are not counted' % (
+          ip, self.EXCUSE_TIME))
+      self.excused[ip] = now + self.EXCUSE_TIME
+
+  def attempt(self, ip):
+    """Counts a failed attempt of ip and bans its network after max_attempts within retry_window"""
+    options = self.options.values
+    prefix = options['netban_ipv4'] if ip.version == 4 else options['netban_ipv6']
+    net = str(ipaddress.ip_network('%s/%d' % (ip, prefix), strict=False))
+    with self.lock:
+      now = self.clock()
+      allowed = self.allowed_by(ip)
+      if allowed is not None:
+        self.logger.logInfo('Address %s is allowlisted by rule %s' % (ip, allowed))
+        return
+      if self.excused.get(ip, 0) > now:
+        self.logger.logInfo('Not counting the failed login of %s, the auth backend was unavailable' % ip)
+        return
+      if net in self.banned:
+        # log lines of connections from before the ban
+        return
+      counter = self.counter(net)
+      if now - counter['last_attempt'] > options['retry_window']:
+        counter['attempts'] = 0
+      counter['attempts'] += 1
+      counter['last_attempt'] = now
+      if counter['attempts'] < options['max_attempts']:
+        self.logger.logWarn('%d more attempts in the next %d seconds until %s is banned' % (
+          options['max_attempts'] - counter['attempts'], options['retry_window'], net))
+        return
+
+      ban_time = self.options.ban_time(counter['ban_counter'])
+      self.logger.logCrit('Banning %s for %d minutes' % (net, ban_time / 60))
+      # Redis first: a firewall rule without expiry would never be lifted
+      self.r.hset('F2B_ACTIVE_BANS', net, int(round(now + ban_time)))
+      self.banned.add(net)
+      if options['manage_external'] != 1:
+        self.apply([net])
+
+  def unban(self, net):
+    """Lifts the automatic ban of a network; the rule stays while the denylist bans the same network"""
+    with self.lock:
+      normalized = self.normalize(net)
+      if normalized is None:
+        self.logger.logWarn('Removing invalid network %s from the active bans' % net)
+      else:
+        self.logger.logInfo('Unbanning %s' % net)
+        if normalized not in self.denied:
+          self.remove([normalized])
+      self.r.hdel('F2B_ACTIVE_BANS', net)
+      self.r.hdel('F2B_QUEUE_UNBAN', net)
+      if net in self.banned:
+        self.banned.discard(net)
+        counter = self.counter(net)
+        counter['attempts'] = 0
+        counter['ban_counter'] += 1
+
+  def unban_queued(self, entry):
+    """Unban request of the UI: a banned network, or a network that contains banned ones (allowlisting)"""
+    network = self.network(entry)
+    if network is None:
+      self.logger.logWarn('Removing invalid network %s from the unban queue' % entry)
+      self.r.hdel('F2B_QUEUE_UNBAN', entry)
+      return
+    with self.lock:
+      banned = set(self.r.hkeys('F2B_ACTIVE_BANS')) | self.banned
+      targets = [net for net in banned if self.overlaps(net, network)]
+      for net in targets:
+        self.unban(net)
+      if not targets and str(network) not in self.denied:
+        # a rule without known ban, e.g. left by an older netfilter
+        if self.remove([str(network)]):
+          self.logger.logInfo('Unbanning %s' % network)
+        else:
+          self.logger.logInfo('%s is not banned, removing it from the unban queue' % network)
+      self.r.hdel('F2B_QUEUE_UNBAN', entry)
+
+  def purge(self):
+    """Processes the unban queue of the UI and lifts expired bans"""
+    for entry in self.r.hkeys('F2B_QUEUE_UNBAN'):
+      self.unban_queued(entry)
+    now = self.clock()
+    for net, expires in self.r.hgetall('F2B_ACTIVE_BANS').items():
+      try:
+        expired = self.network(net) is None or float(expires) <= now
+      except (TypeError, ValueError):
+        expired = True
+      if expired:
+        self.unban(net)
+    with self.lock:
+      # bans whose Redis entry is gone, e.g. deleted by hand
+      for net in self.banned - set(self.r.hkeys('F2B_ACTIVE_BANS')):
+        self.unban(net)
+    self.forget_idle(now)
+
+  def restore(self):
+    """Applies the unexpired automatic bans of Redis again after a start of netfilter and drops the expired ones"""
+    now = self.clock()
+    restored = []
+    with self.lock:
+      for net, expires in self.r.hgetall('F2B_ACTIVE_BANS').items():
+        try:
+          valid = self.network(net) is not None and float(expires) > now
+        except (TypeError, ValueError):
+          valid = False
+        if not valid:
+          self.r.hdel('F2B_ACTIVE_BANS', net)
+          continue
+        self.banned.add(net)
+        self.counter(net)
+        restored.append(net)
+      if restored and self.options.values['manage_external'] != 1:
+        self.apply(restored)
+    if restored:
+      self.logger.logInfo('Restored %d active bans' % len(restored))
+
+  def reapply(self):
+    """Applies all bans again, e.g. after the MAILCOW chain had to be recreated"""
+    with self.lock:
+      if self.options.values['manage_external'] != 1:
+        self.apply(list(self.banned) + list(self.denied))
+
+  def set_denylist(self, entries):
+    """Bans the networks of the denylist permanently and lifts the bans of removed entries"""
+    networks = {}
+    for entry in entries:
+      net = self.normalize(entry)
+      if net is None:
+        self.logger.logWarn('Skipping invalid denylist entry %s' % entry)
+        continue
+      networks[net] = entry
+    with self.lock:
+      if networks.keys() == self.denied.keys():
+        return
+      self.logger.logInfo('Denylist was changed, it has %d entries' % len(networks))
+      added = [net for net in networks if net not in self.denied]
+      removed = [net for net in self.denied if net not in networks]
+      if added and self.options.values['manage_external'] != 1:
+        self.apply(added)
+      for net in added:
+        self.r.hset('F2B_PERM_BANS', networks[net], int(round(self.clock())))
+        self.logger.logCrit('Added host/network %s to denylist' % networks[net])
+      # an automatic ban of the same network keeps its rule
+      auto_banned = {self.normalize(net) for net in self.banned}
+      self.remove([net for net in removed if net not in auto_banned])
+      for net in removed:
+        self.r.hdel('F2B_PERM_BANS', self.denied[net])
+        self.logger.logCrit('Removed host/network %s from denylist' % self.denied[net])
+      self.denied = networks
+
+  def set_allowlist(self, entries):
+    networks = []
+    for entry in entries:
+      network = self.network(entry)
+      if network is None:
+        self.logger.logWarn('Skipping invalid allowlist entry %s' % entry)
+        continue
+      networks.append(network)
+    with self.lock:
+      if set(networks) != set(self.allowlist):
+        self.allowlist = networks
+        self.logger.logInfo('Allowlist was changed, it has %d entries' % len(networks))
+
+  def allowed_by(self, ip):
+    address = ipaddress.ip_network(ip)
+    for network in self.allowlist:
+      if network.version == address.version and network.overlaps(address):
+        return network
+    return None
+
+  def counter(self, net):
+    return self.counters.setdefault(net, {'attempts': 0, 'last_attempt': 0.0, 'ban_counter': 0})
+
+  def forget_idle(self, now):
+    """Drops counters of networks that were never banned and had no attempt within the retry window"""
+    window = self.options.values['retry_window']
+    with self.lock:
+      for net in [net for net, counter in self.counters.items()
+                  if net not in self.banned and counter['ban_counter'] == 0 and now - counter['last_attempt'] > window]:
+        del self.counters[net]
+      for ip in [ip for ip, until in self.excused.items() if until <= now]:
+        del self.excused[ip]
+
+  def apply(self, nets):
+    for net in self.tables.ban(nets):
+      self.logger.logWarn('Could not add the firewall rule for %s' % net)
+
+  def remove(self, nets):
+    return self.tables.unban(nets) if nets else []
+
+  @staticmethod
+  def network(entry):
+    try:
+      return ipaddress.ip_network(entry, strict=False)
+    except (TypeError, ValueError):
+      return None
+
+  @classmethod
+  def normalize(cls, entry):
+    network = cls.network(entry)
+    return str(network) if network is not None else None
+
+  @classmethod
+  def overlaps(cls, net, network):
+    other = cls.network(net)
+    return other is not None and other.version == network.version and other.overlaps(network)

data/Dockerfiles/netfilter/modules/IPTables.py

@@ -1,6 +1,7 @@
 import iptc
 import time
-import os
+import ipaddress
+import contextlib
 
 class IPTables:
   def __init__(self, chain_name, logger):
@@ -8,62 +9,10 @@ class IPTables:
     self.logger = logger
 
   def initChainIPv4(self):
-    if not iptc.Chain(iptc.Table(iptc.Table.FILTER), self.chain_name) in iptc.Table(iptc.Table.FILTER).chains:
-      iptc.Table(iptc.Table.FILTER).create_chain(self.chain_name)
-    for c in ['FORWARD', 'INPUT']:
-      chain = iptc.Chain(iptc.Table(iptc.Table.FILTER), c)
-      rule = iptc.Rule()
-      rule.src = '0.0.0.0/0'
-      rule.dst = '0.0.0.0/0'
-      target = iptc.Target(rule, self.chain_name)
-      rule.target = target
-      if rule not in chain.rules:
-        chain.insert_rule(rule)
+    self.init_chain(iptc.Table(iptc.Table.FILTER))
 
   def initChainIPv6(self):
-    if not iptc.Chain(iptc.Table6(iptc.Table6.FILTER), self.chain_name) in iptc.Table6(iptc.Table6.FILTER).chains:
-      iptc.Table6(iptc.Table6.FILTER).create_chain(self.chain_name)
-    for c in ['FORWARD', 'INPUT']:
-      chain = iptc.Chain(iptc.Table6(iptc.Table6.FILTER), c)
-      rule = iptc.Rule6()
-      rule.src = '::/0'
-      rule.dst = '::/0'
-      target = iptc.Target(rule, self.chain_name)
-      rule.target = target
-      if rule not in chain.rules:
-        chain.insert_rule(rule)
-
-  def checkIPv4ChainOrder(self):
-    filter_table = iptc.Table(iptc.Table.FILTER)
-    filter_table.refresh()
-    return self.checkChainOrder(filter_table)
-
-  def checkIPv6ChainOrder(self):
-    filter_table = iptc.Table6(iptc.Table6.FILTER)
-    filter_table.refresh()
-    return self.checkChainOrder(filter_table)
-
-  def checkChainOrder(self, filter_table):
-    err = False
-    exit_code = None
-
-    forward_chain = iptc.Chain(filter_table, 'FORWARD')
-    input_chain = iptc.Chain(filter_table, 'INPUT')
-    for chain in [forward_chain, input_chain]:
-      target_found = False
-      for position, item in enumerate(chain.rules):
-        if item.target.name == self.chain_name:
-          target_found = True
-          if position > 2:
-            self.logger.logCrit('Error in %s chain: %s target not found, restarting container' % (chain.name, self.chain_name))
-            err = True
-            exit_code = 2
-      if not target_found:
-        self.logger.logCrit('Error in %s chain: %s target not found, restarting container' % (chain.name, self.chain_name))
-        err = True
-        exit_code = 2
-
-    return err, exit_code
+    self.init_chain(iptc.Table6(iptc.Table6.FILTER))
 
   def clearIPv4Table(self):
     self.clearTable(iptc.Table(iptc.Table.FILTER))
@@ -71,68 +20,133 @@ class IPTables:
   def clearIPv6Table(self):
     self.clearTable(iptc.Table6(iptc.Table6.FILTER))
 
-  def clearTable(self, filter_table):
-    filter_table.autocommit = False
-    forward_chain = iptc.Chain(filter_table, "FORWARD")
-    input_chain = iptc.Chain(filter_table, "INPUT")
-    mailcow_chain = iptc.Chain(filter_table, self.chain_name)
-    if mailcow_chain in filter_table.chains:
-      for rule in mailcow_chain.rules:
-        mailcow_chain.delete_rule(rule)
-      for rule in forward_chain.rules:
-        if rule.target.name == self.chain_name:
-          forward_chain.delete_rule(rule)
-      for rule in input_chain.rules:
-        if rule.target.name == self.chain_name:
-          input_chain.delete_rule(rule)
-      filter_table.delete_chain(self.chain_name)
-    filter_table.commit()
+  @contextlib.contextmanager
+  def transaction(self, table):
+    # One commit for all changes, autocommit rewrites the whole table for every change
+    table.autocommit = False
+    table.refresh()
+    try:
+      yield
+      table.commit()
+    finally:
+      table.refresh()
+      table.autocommit = True
+
+  def new_rule(self, table):
+    return iptc.Rule6() if isinstance(table, iptc.Table6) else iptc.Rule()
+
+  def family(self, table):
+    return 'IPv6' if isinstance(table, iptc.Table6) else 'IPv4'
+
+  def init_chain(self, filter_table):
+    # Empty MAILCOW chain with one jump at the top of the input and forward chains
     filter_table.refresh()
-    filter_table.autocommit = True
+    if filter_table.is_chain(self.chain_name):
+      filter_table.flush_entries(self.chain_name)
+    else:
+      filter_table.create_chain(self.chain_name)
+    self.ensure_jumps(filter_table)
 
-  def banIPv4(self, source):
-    chain = iptc.Chain(iptc.Table(iptc.Table.FILTER), self.chain_name)
-    rule = iptc.Rule()
-    rule.src = source
-    target = iptc.Target(rule, "REJECT")
-    rule.target = target
-    if rule in chain.rules:
-      return False
-    chain.insert_rule(rule)
-    return True
+  def ensure_chains(self):
+    # Recreates missing MAILCOW chains and moves the jumps back to the top, True if a chain was recreated
+    recreated = False
+    for filter_table in [iptc.Table(iptc.Table.FILTER), iptc.Table6(iptc.Table6.FILTER)]:
+      filter_table.refresh()
+      if not filter_table.is_chain(self.chain_name):
+        filter_table.create_chain(self.chain_name)
+        self.logger.logWarn(f'MAILCOW {self.family(filter_table)} chain was missing and has been recreated')
+        recreated = True
+      self.ensure_jumps(filter_table)
+    return recreated
 
-  def banIPv6(self, source):
-    chain = iptc.Chain(iptc.Table6(iptc.Table6.FILTER), self.chain_name)
-    rule = iptc.Rule6()
-    rule.src = source
-    target = iptc.Target(rule, "REJECT")
-    rule.target = target
-    if rule in chain.rules:
-      return False
-    chain.insert_rule(rule)
-    return True
+  def ensure_jumps(self, filter_table):
+    # Exactly one jump to MAILCOW as first rule: docker inserts its jump (DOCKER-USER) at the top
+    for name in ['INPUT', 'FORWARD']:
+      chain = iptc.Chain(filter_table, name)
+      positions = [position for position, rule in enumerate(chain.rules) if rule.target.name == self.chain_name]
+      if positions == [0]: continue
 
-  def unbanIPv4(self, source):
-    chain = iptc.Chain(iptc.Table(iptc.Table.FILTER), self.chain_name)
-    rule = iptc.Rule()
-    rule.src = source
-    target = iptc.Target(rule, "REJECT")
-    rule.target = target
-    if rule not in chain.rules: 
-      return False
-    chain.delete_rule(rule)
-    return True
+      with self.transaction(filter_table):
+        for rule in [rule for rule in chain.rules if rule.target.name == self.chain_name]:
+          chain.delete_rule(rule)
+        rule = self.new_rule(filter_table)
+        rule.target = iptc.Target(rule, self.chain_name)
+        chain.insert_rule(rule)
+      if positions:
+        self.logger.logInfo(f'Moved MAILCOW jump to the top of the {self.family(filter_table)} {name} chain, it was at position {positions[0] + 1}')
+      else:
+        self.logger.logInfo(f'Added MAILCOW jump to the {self.family(filter_table)} {name} chain')
 
-  def unbanIPv6(self, source):
-    chain = iptc.Chain(iptc.Table6(iptc.Table6.FILTER), self.chain_name)
-    rule = iptc.Rule6()
-    rule.src = source
-    target = iptc.Target(rule, "REJECT")
-    rule.target = target
-    if rule not in chain.rules:
-      return False
-    chain.delete_rule(rule)
-    return True
+  def clearTable(self, filter_table):
+    try:
+      with self.transaction(filter_table):
+        if not filter_table.is_chain(self.chain_name):
+          return
+        for chain in filter_table.chains:
+          for rule in chain.rules:
+            if rule.target.name == self.chain_name:
+              chain.delete_rule(rule)
+        filter_table.flush_entries(self.chain_name)
+        filter_table.delete_chain(self.chain_name)
+      self.logger.logInfo(f'Clear completed: {self.family(filter_table)}')
+    except Exception as ex:
+      self.logger.logWarn(f'Error clearing the {self.family(filter_table)} MAILCOW chain: {ex}')
+
+  def ban_rule_network(self, rule):
+    # Source network of a ban rule, None for every other rule
+    if rule.target.name not in ('REJECT', 'DROP') or rule.matches or rule.in_interface or rule.out_interface:
+      return None
+    try:
+      return ipaddress.ip_network(rule.src, strict=False)
+    except ValueError:
+      return None
+
+  def split_tables(self, sources):
+    networks = {4: [], 6: []}
+    for source in sources:
+      network = ipaddress.ip_network(source, strict=False)
+      networks[network.version].append(network)
+    tables = [(iptc.Table(iptc.Table.FILTER), networks[4]), (iptc.Table6(iptc.Table6.FILTER), networks[6])]
+    return [(table, table_networks) for table, table_networks in tables if table_networks]
+
+  def ban(self, sources):
+    # One REJECT rule per network, existing rules are kept. Returns the networks that could not be banned
+    failed = []
+    for filter_table, networks in self.split_tables(sources):
+      try:
+        with self.transaction(filter_table):
+          chain = iptc.Chain(filter_table, self.chain_name)
+          existing = {self.ban_rule_network(rule) for rule in chain.rules}
+          for network in dict.fromkeys(networks):
+            if network in existing: continue
+            rule = self.new_rule(filter_table)
+            rule.src = str(network)
+            rule.target = iptc.Target(rule, "REJECT")
+            chain.insert_rule(rule)
+      except Exception as ex:
+        self.logger.logWarn(f'iptables: {ex}')
+        failed += networks
+    return [str(network) for network in failed]
+
+  def unban(self, sources):
+    # Removes every rule of the networks, also duplicates of older versions. Returns the unbanned networks
+    removed = []
+    for filter_table, networks in self.split_tables(sources):
+      targets = set(networks)
+      try:
+        with self.transaction(filter_table):
+          chain = iptc.Chain(filter_table, self.chain_name)
+          found = set()
+          for rule in chain.rules:
+            network = self.ban_rule_network(rule)
+            if network in targets:
+              chain.delete_rule(rule)
+              found.add(network)
+      except Exception as ex:
+        self.logger.logWarn(f'iptables: {ex}')
+        continue
+      removed += found
+    return [str(network) for network in removed]
 
   def snat4(self, snat_target, source):
     try:
@@ -188,7 +202,7 @@ class IPTables:
           if item == new_rule:
             if position != 0:
               chain.delete_rule(new_rule)
-    
+
       table.commit()
       table.autocommit = True
     except:
@@ -215,7 +229,9 @@ class IPTables:
 
   def create_mailcow_isolation_rule(self, _interface:str, _dports:list, _allow:str = ""):
     try:
-      chain = iptc.Chain(iptc.Table(iptc.Table.FILTER), self.chain_name)
+      table = iptc.Table(iptc.Table.FILTER)
+      table.refresh()
+      chain = iptc.Chain(table, self.chain_name)
 
       # insert mailcow isolation rule
       rule = iptc.Rule()
@@ -231,7 +247,7 @@ class IPTables:
       chain.insert_rule(rule, position=0)
 
       # insert mailcow isolation exception rule
-      if _allow != "":
+      if _allow:
         rule = iptc.Rule()
         rule.src = _allow
         rule.in_interface = f'!{_interface}'
@@ -249,4 +265,4 @@ class IPTables:
       return True
     except Exception as e:
       self.logger.logCrit(f"Error adding {self.chain_name} isolation: {e}")
-      return False
+      return False

data/Dockerfiles/netfilter/modules/NFTables.py

@@ -1,8 +1,9 @@
 import nftables
 import ipaddress
-import os
 
 class NFTables:
+  ISOLATION_COMMENTS = ('mailcow isolation', 'mailcow isolation allow')
+
   def __init__(self, chain_name, logger):
     self.chain_name = chain_name
     self.logger = logger
@@ -12,41 +13,15 @@ class NFTables:
     self.nft.set_handle_output(True)
     self.nft_chain_names = {'ip': {'filter': {'input': '', 'forward': ''}, 'nat': {'postrouting': ''} },
                             'ip6': {'filter': {'input': '', 'forward': ''}, 'nat': {'postrouting': ''} } }
+    self.mailcow_chains = set()
 
     self.search_current_chains()
 
   def initChainIPv4(self):
-    self.insert_mailcow_chains("ip")
+    self.init_chain("ip")
 
   def initChainIPv6(self):
-    self.insert_mailcow_chains("ip6")
-
-  def checkIPv4ChainOrder(self):
-    return self.checkChainOrder("ip")
-
-  def checkIPv6ChainOrder(self):
-    return self.checkChainOrder("ip6")
-
-  def checkChainOrder(self, filter_table):
-    err = False
-    exit_code = None
-
-    for chain in ['input', 'forward']:
-      chain_position = self.check_mailcow_chains(filter_table, chain)
-      if chain_position is None: continue
-
-      if chain_position is False:
-        self.logger.logCrit(f'MAILCOW target not found in {filter_table} {chain} table, restarting container to fix it...')
-        err = True
-        exit_code = 2
-
-      if chain_position > 0:
-        chain_position += 1
-        self.logger.logCrit(f'MAILCOW target is in position {chain_position} in the {filter_table} {chain} table, restarting container to fix it...')
-        err = True
-        exit_code = 2
-
-    return err, exit_code
+    self.init_chain("ip6")
 
   def clearIPv4Table(self):
     self.clearTable("ip")
@@ -54,84 +29,19 @@ class NFTables:
   def clearIPv6Table(self):
     self.clearTable("ip6")
 
-  def clearTable(self, _family):
-    is_empty_dict = True
-    json_command = self.get_base_dict()
-    chain_handle = self.get_chain_handle(_family, "filter", self.chain_name)
-    # if no handle, the chain doesn't exists
-    if chain_handle is not None:
-      is_empty_dict = False
-      # flush chain
-      mailcow_chain = {'family': _family, 'table': 'filter', 'name': self.chain_name}
-      flush_chain = {'flush': {'chain': mailcow_chain}}
-      json_command["nftables"].append(flush_chain)
-
-    # remove rule in forward chain
-    # remove rule in input chain
-    chains_family = [self.nft_chain_names[_family]['filter']['input'],
-                    self.nft_chain_names[_family]['filter']['forward'] ]
-
-    for chain_base in chains_family:
-      if not chain_base: continue
-
-      rules_handle = self.get_rules_handle(_family, "filter", chain_base)
-      if rules_handle is not None:
-        for r_handle in rules_handle:
-          is_empty_dict = False
-          mailcow_rule = {'family':_family,
-                          'table': 'filter',
-                          'chain': chain_base,
-                          'handle': r_handle }
-          delete_rules = {'delete': {'rule': mailcow_rule} }
-          json_command["nftables"].append(delete_rules)
-
-    # remove chain
-    # after delete all rules referencing this chain
-    if chain_handle is not None:
-      mc_chain_handle = {'family':_family,
-                        'table': 'filter',
-                        'name': self.chain_name,
-                        'handle': chain_handle }
-      delete_chain = {'delete': {'chain': mc_chain_handle} }
-      json_command["nftables"].append(delete_chain)
-
-    if is_empty_dict == False:
-      if self.nft_exec_dict(json_command):
-        self.logger.logInfo(f"Clear completed: {_family}")
-
-  def banIPv4(self, source):
-    ban_dict = self.get_ban_ip_dict(source, "ip")
-    return self.nft_exec_dict(ban_dict)
-
-  def banIPv6(self, source):
-    ban_dict = self.get_ban_ip_dict(source, "ip6")
-    return self.nft_exec_dict(ban_dict)
-
-  def unbanIPv4(self, source):
-    unban_dict = self.get_unban_ip_dict(source, "ip")
-    if not unban_dict:
-      return False
-    return self.nft_exec_dict(unban_dict)
-
-  def unbanIPv6(self, source):
-    unban_dict = self.get_unban_ip_dict(source, "ip6")
-    if not unban_dict:
-      return False
-    return self.nft_exec_dict(unban_dict)
-
   def snat4(self, snat_target, source):
     self.snat_rule("ip", snat_target, source)
 
   def snat6(self, snat_target, source):
     self.snat_rule("ip6", snat_target, source)
 
-
-  def nft_exec_dict(self, query: dict):
+  def nft_exec_dict(self, query: dict, quiet: bool = False):
     if not query: return False
 
     rc, output, error = self.nft.json_cmd(query)
     if rc != 0:
-      #self.logger.logCrit(f"Nftables Error: {error}")
+      if not quiet:
+        self.logger.logWarn(f"nftables: {str(error).strip() or f'error {rc}'}")
       return False
 
     # Prevent returning False or empty string on commands that do not produce output
@@ -144,118 +54,165 @@ class NFTables:
     return {'nftables': [{ 'metainfo': { 'json_schema_version': 1} } ] }
 
   def search_current_chains(self):
+    nft_chain_names = {'ip': {'filter': {'input': '', 'forward': ''}, 'nat': {'postrouting': ''} },
+                       'ip6': {'filter': {'input': '', 'forward': ''}, 'nat': {'postrouting': ''} } }
     nft_chain_priority = {'ip': {'filter': {'input': None, 'forward': None}, 'nat': {'postrouting': None} },
-                      'ip6': {'filter': {'input': None, 'forward': None}, 'nat': {'postrouting': None} } }
+                          'ip6': {'filter': {'input': None, 'forward': None}, 'nat': {'postrouting': None} } }
+    mailcow_chains = set()
 
     # Command: 'nft list chains'
     _list = {'list' : {'chains': 'null'} }
     command = self.get_base_dict()
     command['nftables'].append(_list)
     kernel_ruleset = self.nft_exec_dict(command)
-    if kernel_ruleset:
-      for _object in kernel_ruleset['nftables']:
-        chain = _object.get("chain")
-        if not chain: continue
+    if not kernel_ruleset:
+      return
 
-        _family = chain['family']
-        _table = chain['table']
-        _hook = chain.get("hook")
-        _priority = chain.get("prio")
-        _name = chain['name']
-
-        if _family not in self.nft_chain_names: continue
-        if _table not in self.nft_chain_names[_family]: continue
-        if _hook not in self.nft_chain_names[_family][_table]: continue
-        if _priority is None: continue
-
-        _saved_priority = nft_chain_priority[_family][_table][_hook]
-        if _saved_priority is None or _priority < _saved_priority:
-          # at this point, we know the chain has:
-          # hook and priority set
-          # and it has the lowest priority
-          nft_chain_priority[_family][_table][_hook] = _priority
-          self.nft_chain_names[_family][_table][_hook] = _name
-
-  def search_for_chain(self, kernel_ruleset: dict, chain_name: str):
-    found = False
-    for _object in kernel_ruleset["nftables"]:
+    for _object in kernel_ruleset['nftables']:
       chain = _object.get("chain")
-      if not chain:
-        continue
-      ch_name = chain.get("name")
-      if ch_name == chain_name:
-        found = True
-        break
-    return found
+      if not chain: continue
+
+      _family = chain['family']
+      _table = chain['table']
+      _hook = chain.get("hook")
+      _priority = chain.get("prio")
+      _name = chain['name']
+
+      if _family not in nft_chain_names: continue
+      if _table not in nft_chain_names[_family]: continue
+      if _table == 'filter' and _name == self.chain_name:
+        mailcow_chains.add(_family)
+      if _hook not in nft_chain_names[_family][_table]: continue
+      if _priority is None: continue
+
+      _saved_priority = nft_chain_priority[_family][_table][_hook]
+      if _saved_priority is None or _priority < _saved_priority:
+        # at this point, we know the chain has:
+        # hook and priority set
+        # and it has the lowest priority
+        nft_chain_priority[_family][_table][_hook] = _priority
+        nft_chain_names[_family][_table][_hook] = _name
+
+    self.nft_chain_names = nft_chain_names
+    self.mailcow_chains = mailcow_chains
+
+  def has_filter_chains(self, _family: str):
+    # no base chains: the filter table does not exist, e.g. ip6 if docker has ipv6 disabled
+    return any(self.nft_chain_names[_family]['filter'].values())
+
+  def list_rules(self, _family: str, _table: str, _chain: str):
+    # Command: 'nft list chain {family} {table} {chain}', None if the chain does not exist
+    _list = {'list': {'chain': {'family': _family, 'table': _table, 'name': _chain}}}
+    command = self.get_base_dict()
+    command['nftables'].append(_list)
+    kernel_ruleset = self.nft_exec_dict(command, quiet=True)
+    if not kernel_ruleset or not isinstance(kernel_ruleset, dict):
+      return None
+    return [_object["rule"] for _object in kernel_ruleset["nftables"] if _object.get("rule")]
 
   def get_chain_dict(self, _family: str, _name: str):
-    # nft (add | create) chain [<family>] <table> <name> 
+    # nft (add | create) chain [<family>] <table> <name>
     _chain_opts = {'family': _family, 'table': 'filter', 'name': _name  }
     _add = {'add': {'chain': _chain_opts} }
     final_chain = self.get_base_dict()
     final_chain["nftables"].append(_add)
     return final_chain
 
-  def get_mailcow_jump_rule_dict(self, _family: str, _chain: str):
-    _jump_rule = self.get_base_dict()
-    _expr_opt=[]
-    _expr_counter = {'family': _family, 'table': 'filter', 'packets': 0, 'bytes': 0}
-    _counter_dict = {'counter': _expr_counter}
-    _expr_opt.append(_counter_dict)
-
-    _jump_opts = {'jump': {'target': self.chain_name} }
-
-    _expr_opt.append(_jump_opts)
-
+  def get_mailcow_jump_rule(self, _family: str, _chain: str):
+    _expr_opt = [{'counter': {'family': _family, 'table': 'filter', 'packets': 0, 'bytes': 0}},
+                 {'jump': {'target': self.chain_name}}]
     _rule_params = {'family': _family,
                     'table': 'filter',
                     'chain': _chain,
                     'expr': _expr_opt,
                     'comment': "mailcow" }
+    return {'insert': {'rule': _rule_params} }
 
-    _add_rule = {'insert': {'rule': _rule_params} }
+  def is_mailcow_jump(self, rule: dict):
+    # by target, jumps restored by iptables-restore carry no nft comment
+    return any(isinstance(expr, dict) and isinstance(expr.get('jump'), dict) and expr['jump'].get('target') == self.chain_name
+               for expr in rule.get('expr', []))
 
-    _jump_rule["nftables"].append(_add_rule)
+  def init_chain(self, _family: str):
+    # Empty MAILCOW chain with one jump at the top of the input and forward chains
+    if not self.has_filter_chains(_family): return
 
-    return _jump_rule
+    if self.list_rules(_family, 'filter', self.chain_name) is None:
+      if self.nft_exec_dict(self.get_chain_dict(_family, self.chain_name)):
+        self.logger.logInfo(f"MAILCOW {_family} chain created successfully.")
+    else:
+      flush_chain = {'flush': {'chain': {'family': _family, 'table': 'filter', 'name': self.chain_name}}}
+      command = self.get_base_dict()
+      command['nftables'].append(flush_chain)
+      self.nft_exec_dict(command)
 
-  def insert_mailcow_chains(self, _family: str):
-    nft_input_chain = self.nft_chain_names[_family]['filter']['input']
-    nft_forward_chain = self.nft_chain_names[_family]['filter']['forward']
-    # Command: 'nft list table <family> filter'
-    _table_opts = {'family': _family, 'name': 'filter'}
-    _list = {'list': {'table': _table_opts} }
+    self.ensure_jumps(_family)
+
+  def ensure_chains(self):
+    # Recreates missing MAILCOW chains and moves the jumps back to the top, True if a chain was recreated
+    self.search_current_chains()
+    recreated = False
+    for _family in ['ip', 'ip6']:
+      if not self.has_filter_chains(_family): continue
+
+      if _family not in self.mailcow_chains:
+        if self.nft_exec_dict(self.get_chain_dict(_family, self.chain_name)):
+          self.logger.logWarn(f"MAILCOW {_family} chain was missing and has been recreated")
+          recreated = True
+      self.ensure_jumps(_family)
+    return recreated
+
+  def ensure_jumps(self, _family: str):
+    # Exactly one jump to MAILCOW as first rule: docker inserts its jumps (DOCKER-USER, DOCKER-FORWARD) at the top
+    for hook in ['input', 'forward']:
+      base_chain = self.nft_chain_names[_family]['filter'][hook]
+      if not base_chain: continue
+
+      rules = self.list_rules(_family, 'filter', base_chain)
+      if rules is None: continue
+
+      positions = [position for position, rule in enumerate(rules) if self.is_mailcow_jump(rule)]
+      if positions == [0]: continue
+
+      # delete and insert in one transaction, the chain is never without jump
+      command = self.get_base_dict()
+      for position in positions:
+        _rule_opts = {'family': _family, 'table': 'filter', 'chain': base_chain, 'handle': rules[position]['handle']}
+        command['nftables'].append({'delete': {'rule': _rule_opts}})
+      command['nftables'].append(self.get_mailcow_jump_rule(_family, base_chain))
+      if self.nft_exec_dict(command):
+        if positions:
+          self.logger.logInfo(f"Moved MAILCOW jump to the top of the {_family} {base_chain} chain, it was at position {positions[0] + 1}")
+        else:
+          self.logger.logInfo(f"Added MAILCOW jump to the {_family} {base_chain} chain")
+
+  def clearTable(self, _family: str):
+    # Jumps first, the chain can only be deleted without references. Separate steps, so one failing step
+    # does not roll back the others
+    _list = {'list': {'table': {'family': _family, 'name': 'filter'}}}
     command = self.get_base_dict()
     command['nftables'].append(_list)
-    kernel_ruleset = self.nft_exec_dict(command)
-    if kernel_ruleset:
-      # chain
-      if not self.search_for_chain(kernel_ruleset, self.chain_name):
-        cadena = self.get_chain_dict(_family, self.chain_name)
-        if self.nft_exec_dict(cadena):
-          self.logger.logInfo(f"MAILCOW {_family} chain created successfully.")
+    kernel_ruleset = self.nft_exec_dict(command, quiet=True)
+    if not kernel_ruleset or not isinstance(kernel_ruleset, dict):
+      return
 
-      input_jump_found, forward_jump_found = False, False
+    delete_jumps = self.get_base_dict()
+    for _object in kernel_ruleset["nftables"]:
+      rule = _object.get("rule")
+      if rule and self.is_mailcow_jump(rule):
+        _rule_opts = {'family': _family, 'table': 'filter', 'chain': rule['chain'], 'handle': rule['handle']}
+        delete_jumps['nftables'].append({'delete': {'rule': _rule_opts}})
+    if len(delete_jumps['nftables']) > 1:
+      self.nft_exec_dict(delete_jumps)
 
-      for _object in kernel_ruleset["nftables"]:
-        if not _object.get("rule"):
-          continue
-
-        rule = _object["rule"]
-        if nft_input_chain and rule["chain"] == nft_input_chain:
-          if rule.get("comment") and rule["comment"] == "mailcow":
-            input_jump_found = True
-        if nft_forward_chain and rule["chain"] == nft_forward_chain:
-          if rule.get("comment") and rule["comment"] == "mailcow":
-            forward_jump_found = True
-
-      if not input_jump_found:
-        command = self.get_mailcow_jump_rule_dict(_family, nft_input_chain)
-        self.nft_exec_dict(command)
-
-      if not forward_jump_found:
-        command = self.get_mailcow_jump_rule_dict(_family, nft_forward_chain)
-        self.nft_exec_dict(command)
+    if self.list_rules(_family, 'filter', self.chain_name) is None:
+      return
+    mailcow_chain = {'family': _family, 'table': 'filter', 'name': self.chain_name}
+    command = self.get_base_dict()
+    command['nftables'].append({'flush': {'chain': mailcow_chain}})
+    command['nftables'].append({'delete': {'chain': mailcow_chain}})
+    if self.nft_exec_dict(command):
+      self.logger.logInfo(f"Clear completed: {_family}")
 
   def delete_nat_rule(self, _family:str, _chain: str, _handle:str):
     delete_command = self.get_base_dict()
@@ -285,23 +242,14 @@ class NFTables:
     # no postrouting chain, may occur if docker has ipv6 disabled.
     if not chain_name: return
 
-    # Command: nft list chain <family> nat <chain_name>
-    _chain_opts = {'family': _family, 'table': 'nat', 'name': chain_name}
-    _list = {'list':{'chain': _chain_opts} }
-    command = self.get_base_dict()
-    command['nftables'].append(_list)
-    kernel_ruleset = self.nft_exec_dict(command)
-    if not kernel_ruleset:
+    rules = self.list_rules(_family, 'nat', chain_name)
+    if rules is None:
       return
 
     rule_position = 0
     rule_handle = None
     rule_found = False
-    for _object in kernel_ruleset["nftables"]:
-      if not _object.get("rule"):
-        continue
-
-      rule = _object["rule"]
+    for rule in rules:
       if not rule.get("comment") or not rule["comment"] == "mailcow":
         rule_position +=1
         continue
@@ -314,23 +262,28 @@ class NFTables:
     target_net = ipaddress.ip_network(snat_target, strict=False)
 
     if rule_found:
-      saddr_ip = rule["expr"][0]["match"]["right"]["prefix"]["addr"]
-      saddr_len = int(rule["expr"][0]["match"]["right"]["prefix"]["len"])
+      try:
+        saddr_ip = rule["expr"][0]["match"]["right"]["prefix"]["addr"]
+        saddr_len = int(rule["expr"][0]["match"]["right"]["prefix"]["len"])
 
-      daddr_ip = rule["expr"][1]["match"]["right"]["prefix"]["addr"]
-      daddr_len = int(rule["expr"][1]["match"]["right"]["prefix"]["len"])
+        daddr_ip = rule["expr"][1]["match"]["right"]["prefix"]["addr"]
+        daddr_len = int(rule["expr"][1]["match"]["right"]["prefix"]["len"])
 
-      target_ip = rule["expr"][3]["snat"]["addr"]
+        target_ip = rule["expr"][3]["snat"]["addr"]
 
-      saddr_net = ipaddress.ip_network(saddr_ip + '/' + str(saddr_len), strict=False)
-      daddr_net = ipaddress.ip_network(daddr_ip + '/' + str(daddr_len), strict=False)
-      current_target_net = ipaddress.ip_network(target_ip, strict=False)
+        saddr_net = ipaddress.ip_network(saddr_ip + '/' + str(saddr_len), strict=False)
+        daddr_net = ipaddress.ip_network(daddr_ip + '/' + str(daddr_len), strict=False)
+        current_target_net = ipaddress.ip_network(target_ip, strict=False)
 
-      match = all((
-                dest_net == saddr_net,
-                dest_net == daddr_net,
-                target_net == current_target_net
-              ))
+        match = all((
+                  dest_net == saddr_net,
+                  dest_net == daddr_net,
+                  target_net == current_target_net
+                ))
+      except (KeyError, IndexError, TypeError, ValueError):
+        # rule of another format, replace it
+        saddr_net = dest_net
+        match = False
       try:
         if rule_position == 0:
           if not match:
@@ -377,49 +330,13 @@ class NFTables:
       except:
         self.logger.logCrit(f"Error running SNAT on {_family}, retrying...")
 
-  def get_chain_handle(self, _family: str, _table: str, chain_name: str):
-    chain_handle = None
-    # Command: 'nft list chains {family}'
-    _list = {'list': {'chains': {'family': _family} } }
-    command = self.get_base_dict()
-    command['nftables'].append(_list)
-    kernel_ruleset = self.nft_exec_dict(command)
-    if kernel_ruleset:
-      for _object in kernel_ruleset["nftables"]:
-        if not _object.get("chain"):
-          continue
-        chain = _object["chain"]
-        if chain["family"] == _family and chain["table"] == _table and chain["name"] == chain_name:
-          chain_handle = chain["handle"]
-          break
-    return chain_handle
-
   def get_rules_handle(self, _family: str, _table: str, chain_name: str, _comment_filter = "mailcow"):
-    rule_handle = []
-    # Command: 'nft list chain {family} {table} {chain_name}'
-    _chain_opts = {'family': _family, 'table': _table, 'name': chain_name}
-    _list = {'list': {'chain': _chain_opts} }
-    command = self.get_base_dict()
-    command['nftables'].append(_list)
-
-    kernel_ruleset = self.nft_exec_dict(command)
-    if kernel_ruleset:
-      for _object in kernel_ruleset["nftables"]:
-        if not _object.get("rule"):
-          continue
-
-        rule = _object["rule"]
-        if rule["family"] == _family and rule["table"] == _table and rule["chain"] == chain_name:
-          if rule.get("comment") and rule["comment"] == _comment_filter:
-            rule_handle.append(rule["handle"])
-    return rule_handle
-
-  def get_ban_ip_dict(self, ipaddr: str, _family: str):
-    json_command = self.get_base_dict()
+    rules = self.list_rules(_family, _table, chain_name) or []
+    return [rule["handle"] for rule in rules if rule.get("comment") == _comment_filter]
 
+  def get_ban_rule(self, network, _family: str):
     expr_opt = []
-    ipaddr_net = ipaddress.ip_network(ipaddr, strict=False)
-    right_dict = {'prefix': {'addr': str(ipaddr_net.network_address), 'len': int(ipaddr_net.prefixlen) } }
+    right_dict = {'prefix': {'addr': str(network.network_address), 'len': int(network.prefixlen) } }
 
     left_dict = {'payload': {'protocol': _family, 'field': 'saddr'} }
     match_dict = {'op': '==', 'left': left_dict, 'right': right_dict }
@@ -432,228 +349,115 @@ class NFTables:
 
     rule_dict = {'family': _family, 'table': "filter", 'chain': self.chain_name, 'expr': expr_opt}
 
-    base_dict = {'insert': {'rule': rule_dict} }
-    json_command["nftables"].append(base_dict)
+    return {'insert': {'rule': rule_dict} }
 
-    return json_command
+  def ban_rule_network(self, rule: dict, _family: str):
+    # Source network of a ban rule (saddr match and drop), None for every other rule
+    if rule.get("comment") in self.ISOLATION_COMMENTS:
+      return None
+    exprs = [expr for expr in rule.get("expr", []) if isinstance(expr, dict)]
+    if not any('drop' in expr or 'reject' in expr for expr in exprs):
+      return None
+    for expr in exprs:
+      match = expr.get("match")
+      if not isinstance(match, dict) or match.get("op") != "==":
+        continue
+      left = match.get("left")
+      payload = left.get("payload") if isinstance(left, dict) else None
+      if not isinstance(payload, dict) or payload.get("protocol") != _family or payload.get("field") != "saddr":
+        continue
+      right = match.get("right")
+      try:
+        if isinstance(right, dict) and isinstance(right.get("prefix"), dict):
+          return ipaddress.ip_network(f'{right["prefix"]["addr"]}/{right["prefix"]["len"]}', strict=False)
+        if isinstance(right, str):
+          return ipaddress.ip_network(right, strict=False)
+      except (KeyError, ValueError):
+        return None
+    return None
 
-  def get_unban_ip_dict(self, ipaddr:str, _family: str):
-    json_command = self.get_base_dict()
-    # Command: 'nft list chain {s_family} filter  MAILCOW'
-    _chain_opts = {'family': _family, 'table': 'filter', 'name': self.chain_name}
-    _list = {'list': {'chain': _chain_opts} }
-    command = self.get_base_dict()
-    command['nftables'].append(_list)
-    kernel_ruleset = self.nft_exec_dict(command)
-    rule_handle = None
-    if kernel_ruleset:
-      for _object in kernel_ruleset["nftables"]:
-        if not _object.get("rule"):
-          continue
+  def split_families(self, sources):
+    families = {}
+    for source in sources:
+      network = ipaddress.ip_network(source, strict=False)
+      families.setdefault('ip' if network.version == 4 else 'ip6', []).append(network)
+    return families
 
-        rule = _object["rule"]["expr"][0]["match"]
-        if not "payload" in rule["left"]:
-          continue
-        left_opt = rule["left"]["payload"]
-        if not left_opt["protocol"] == _family:
-          continue
-        if not left_opt["field"] =="saddr":
-          continue
+  def ban(self, sources):
+    # One drop rule per network, existing rules are kept. Returns the networks that could not be banned
+    failed = []
+    for _family, networks in self.split_families(sources).items():
+      rules = self.list_rules(_family, 'filter', self.chain_name)
+      if rules is None:
+        failed += networks
+        continue
+      existing = {self.ban_rule_network(rule, _family) for rule in rules}
+      missing = [network for network in dict.fromkeys(networks) if network not in existing]
+      if not missing: continue
 
-        # ip currently banned
-        rule_right = rule["right"]
-        if isinstance(rule_right, dict):
-          current_rule_ip = rule_right["prefix"]["addr"] + '/' + str(rule_right["prefix"]["len"])
-        else:
-          current_rule_ip = rule_right
-        current_rule_net = ipaddress.ip_network(current_rule_ip)
+      command = self.get_base_dict()
+      for network in missing:
+        command['nftables'].append(self.get_ban_rule(network, _family))
+      if not self.nft_exec_dict(command):
+        failed += missing
+    return [str(network) for network in failed]
 
-        # ip to ban
-        candidate_net = ipaddress.ip_network(ipaddr, strict=False)
+  def unban(self, sources):
+    # Removes every rule of the networks, also duplicates of older versions. Returns the unbanned networks
+    removed = []
+    for _family, networks in self.split_families(sources).items():
+      rules = self.list_rules(_family, 'filter', self.chain_name)
+      if not rules: continue
 
-        if current_rule_net == candidate_net:
-          rule_handle = _object["rule"]["handle"]
-          break
-
-      if rule_handle is not None:
-        mailcow_rule = {'family': _family, 'table': 'filter', 'chain': self.chain_name, 'handle': rule_handle}
-        delete_rule = {'delete': {'rule': mailcow_rule} }
-        json_command["nftables"].append(delete_rule)
-      else:
-        return False
-
-    return json_command
-
-  def check_mailcow_chains(self, family: str, chain: str):
-    position = 0
-    rule_found = False
-    chain_name = self.nft_chain_names[family]['filter'][chain]
-
-    if not chain_name: return None
-
-    _chain_opts = {'family': family, 'table': 'filter', 'name': chain_name}
-    _list = {'list': {'chain': _chain_opts}}
-    command = self.get_base_dict()
-    command['nftables'].append(_list)
-    kernel_ruleset = self.nft_exec_dict(command)
-    if kernel_ruleset:
-      for _object in kernel_ruleset["nftables"]:
-        if not _object.get("rule"):
-          continue
-        rule = _object["rule"]
-        if rule.get("comment") and rule["comment"] == "mailcow":
-          rule_found = True
-          break
-
-        position+=1
-
-    return position if rule_found else False
+      targets = set(networks)
+      command = self.get_base_dict()
+      found = set()
+      for rule in rules:
+        network = self.ban_rule_network(rule, _family)
+        if network in targets:
+          _rule_opts = {'family': _family, 'table': 'filter', 'chain': self.chain_name, 'handle': rule['handle']}
+          command['nftables'].append({'delete': {'rule': _rule_opts}})
+          found.add(network)
+      if found and self.nft_exec_dict(command):
+        removed += found
+    return [str(network) for network in removed]
 
   def create_mailcow_isolation_rule(self, _interface:str, _dports:list, _allow:str = ""):
-    family = "ip"
-    table = "filter"
-    comment_filter_drop = "mailcow isolation"
-    comment_filter_allow = "mailcow isolation allow"
-    json_command = self.get_base_dict()
+    # One rule per port: iptables-nft cannot show an anonymous set, a host that saves and restores the ruleset
+    # with iptables-save (e.g. netfilter-persistent) would get a drop for all ports
+    for comment in self.ISOLATION_COMMENTS:
+      for handle in self.get_rules_handle("ip", "filter", self.chain_name, comment):
+        self.delete_filter_rule("ip", self.chain_name, handle)
 
-    # Delete old mailcow isolation rules
-    handles = self.get_rules_handle(family, table, self.chain_name, comment_filter_drop)
-    for handle in handles:
-      self.delete_filter_rule(family, self.chain_name, handle)
-    handles = self.get_rules_handle(family, table, self.chain_name, comment_filter_allow)
-    for handle in handles:
-      self.delete_filter_rule(family, self.chain_name, handle)
-
-    # insert mailcow isolation rule
-    _match_dict_drop = [
-      {
-        "match": {
-          "op": "!=",
-          "left": {
-            "meta": {
-              "key": "iifname"
-            }
-          },
-          "right": _interface
-        }
-      },
-      {
-        "match": {
-          "op": "==",
-          "left": {
-            "meta": {
-              "key": "oifname"
-            }
-          },
-          "right": _interface
-        }
-      },
-      {
-        "match": {
-          "op": "==",
-          "left": {
-            "payload": {
-              "protocol": "tcp",
-              "field": "dport"
-            }
-          },
-          "right": {
-            "set": _dports
-          }
-        }
-      },
-      {
-        "counter": {
-          "packets": 0,
-          "bytes": 0
-        }
-      },
-      {
-        "drop": None
-      }
-    ]
-    rule_drop = { "insert": { "rule": {
-      "family": family,
-      "table": table,
-      "chain": self.chain_name,
-      "comment": comment_filter_drop,
-      "expr": _match_dict_drop
-    }}}
-    json_command["nftables"].append(rule_drop)
-
-    # insert mailcow isolation allow rule
-    if _allow != "":
-      _match_dict_allow = [
-        {
-          "match": {
-            "op": "==",
-            "left": {
-              "payload": {
-                "protocol": "ip",
-                "field": "saddr"
-              }
-            },
-            "right": _allow
-          }
-        },
-        {
-          "match": {
-            "op": "!=",
-            "left": {
-              "meta": {
-                "key": "iifname"
-              }
-            },
-            "right": _interface
-          }
-        },
-        {
-          "match": {
-            "op": "==",
-            "left": {
-              "meta": {
-                "key": "oifname"
-              }
-            },
-            "right": _interface
-          }
-        },
-        {
-          "match": {
-            "op": "==",
-            "left": {
-              "payload": {
-                "protocol": "tcp",
-                "field": "dport"
-              }
-            },
-            "right": {
-              "set": _dports
-            }
-          }
-        },
-        {
-          "counter": {
-            "packets": 0,
-            "bytes": 0
-          }
-        },
-        {
-          "accept": None
-        }
-      ]
-      rule_allow = { "insert": { "rule": {
-        "family": family,
-        "table": table,
-        "chain": self.chain_name,
-        "comment": comment_filter_allow,
-        "expr": _match_dict_allow
-      }}}
-      json_command["nftables"].append(rule_allow)
-
-    success = self.nft_exec_dict(json_command)
-    if success == False:
+    if not self.nft_exec_dict(self.get_isolation_dict(_interface, _dports, self.ISOLATION_COMMENTS[0], {"drop": None})):
       self.logger.logCrit(f"Error adding {self.chain_name} isolation")
       return False
 
-    return True
+    # separate, an invalid address must not prevent the drop rules. Inserted last, so they are above them
+    if _allow and not self.nft_exec_dict(self.get_isolation_dict(_interface, _dports, self.ISOLATION_COMMENTS[1], {"accept": None}, _allow)):
+      self.logger.logCrit(f"Error adding {self.chain_name} isolation exception for {_allow}")
+      return False
+
+    return True
+
+  def get_isolation_dict(self, _interface:str, _dports:list, _comment:str, _verdict:dict, _saddr:str = ""):
+    json_command = self.get_base_dict()
+    for port in _dports:
+      expr = []
+      if _saddr:
+        expr.append({"match": {"op": "==", "left": {"payload": {"protocol": "ip", "field": "saddr"}}, "right": _saddr}})
+      expr += [
+        {"match": {"op": "!=", "left": {"meta": {"key": "iifname"}}, "right": _interface}},
+        {"match": {"op": "==", "left": {"meta": {"key": "oifname"}}, "right": _interface}},
+        {"match": {"op": "==", "left": {"payload": {"protocol": "tcp", "field": "dport"}}, "right": int(port)}},
+        {"counter": {"packets": 0, "bytes": 0}},
+        _verdict
+      ]
+      json_command["nftables"].append({"insert": {"rule": {
+        "family": "ip",
+        "table": "filter",
+        "chain": self.chain_name,
+        "comment": _comment,
+        "expr": expr
+      }}})
+    return json_command

data/Dockerfiles/netfilter/modules/Netfilter.py

@@ -0,0 +1,200 @@
+import ipaddress
+import threading
+import dns.exception
+import dns.resolver
+from modules.BanManager import BanManager
+from modules.Options import Options
+from modules.Regexes import Regexes
+
+
+class Netfilter:
+  """Runs the workers of netfilter. Every worker is a loop that logs errors and goes on, so neither a broken log
+  line nor a Redis restart can stop the expiry of bans."""
+
+  ISOLATION_INTERFACE = 'br-mailcow'
+  ISOLATION_PORTS = [3306, 6379, 8983, 12345]
+
+  def __init__(self, r, tables, logger, env):
+    self.r = r
+    self.tables = tables
+    self.logger = logger
+    self.env = env
+    # reentrant, bans.reapply() runs within the chain check
+    self.lock = threading.RLock()
+    self.stop = threading.Event()
+    self.options = Options(logger)
+    self.regexes = Regexes(logger)
+    self.bans = BanManager(r, tables, self.options, logger, self.lock)
+    self.snat4_target = self.snat_target('SNAT_TO_SOURCE', 4)
+    self.snat6_target = self.snat_target('SNAT6_TO_SOURCE', 6)
+    self.resolved = {}
+    self.workers = {
+      'watch': (self.watch, 3),
+      'purge': (self.purge, 10),
+      'chains': (self.check_chains, 10),
+      'lists': (self.update_lists, 60),
+    }
+    if self.snat4_target or self.snat6_target:
+      self.workers['snat'] = (self.snat, 10)
+    self.threads = {}
+
+  def setup(self):
+    if self.r.exists('F2B_LOG'):
+      self.r.rename('F2B_LOG', 'NETFILTER_LOG')
+    # filled again by the lists worker
+    self.r.delete('F2B_PERM_BANS')
+    self.options.refresh(self.r)
+    self.regexes.refresh(self.r)
+    with self.lock:
+      self.tables.clearIPv4Table()
+      self.tables.clearIPv6Table()
+      self.logger.logInfo('Initializing mailcow netfilter chain')
+      self.tables.initChainIPv4()
+      self.tables.initChainIPv6()
+      self.isolate()
+    self.bans.restore()
+
+  def run(self):
+    for name in self.workers:
+      self.start(name)
+    while not self.stop.wait(5):
+      for name, thread in list(self.threads.items()):
+        if not thread.is_alive():
+          self.logger.logCrit('Worker %s stopped unexpectedly, restarting it' % name)
+          self.start(name)
+
+  def shutdown(self):
+    """Removes the firewall rules, the active bans stay in Redis and are restored on the next start"""
+    self.stop.set()
+    for thread in self.threads.values():
+      thread.join(timeout=5)
+    with self.lock:
+      self.logger.logInfo('Clearing all bans')
+      self.tables.clearIPv4Table()
+      self.tables.clearIPv6Table()
+    try:
+      self.r.delete('F2B_PERM_BANS')
+    except Exception as ex:
+      self.logger.logWarn('Error clearing redis key F2B_PERM_BANS: %s' % ex)
+
+  def start(self, name):
+    task, interval = self.workers[name]
+    thread = threading.Thread(target=self.loop, args=(name, task, interval), name=name, daemon=True)
+    self.threads[name] = thread
+    thread.start()
+
+  def loop(self, name, task, interval):
+    while not self.stop.is_set():
+      try:
+        task()
+      except Exception as ex:
+        self.logger.logWarn('Error in worker %s: %s' % (name, ex))
+      self.stop.wait(interval)
+
+  def watch(self):
+    pubsub = self.r.pubsub(ignore_subscribe_messages=True)
+    try:
+      pubsub.subscribe('F2B_CHANNEL')
+      self.logger.logInfo('Watching Redis channel F2B_CHANNEL')
+      while not self.stop.is_set():
+        message = pubsub.get_message(timeout=1.0)
+        if message and message['type'] == 'message' and message['data']:
+          self.handle_line(message['data'])
+    finally:
+      try:
+        pubsub.close()
+      except Exception:
+        pass
+
+  def handle_line(self, line):
+    # a broken line must not end the watcher
+    try:
+      excused = Regexes.temporary_failure(line)
+      if excused is not None:
+        ip = Regexes.parse_ip(excused)
+        if self.external(ip):
+          self.bans.excuse(ip)
+        return
+      result = self.regexes.match(line)
+      if result is None:
+        return
+      rule_id, address = result
+      ip = Regexes.parse_ip(address)
+      if ip is None:
+        self.logger.logWarn('Rule id %s matched an invalid address %s (%s)' % (rule_id, address, line))
+        return
+      if not self.external(ip):
+        return
+      self.logger.logWarn('%s matched rule id %s (%s)' % (ip, rule_id, line))
+      self.bans.attempt(ip)
+    except Exception as ex:
+      self.logger.logWarn('Error processing log line %r: %s' % (line, ex))
+
+  @staticmethod
+  def external(ip):
+    # private and loopback addresses are never banned
+    return ip is not None and not (ip.is_private or ip.is_loopback)
+
+  def purge(self):
+    self.options.refresh(self.r)
+    self.regexes.refresh(self.r)
+    self.bans.purge()
+
+  def check_chains(self):
+    with self.lock:
+      if self.tables.ensure_chains():
+        self.isolate()
+        self.bans.reapply()
+
+  def snat(self):
+    with self.lock:
+      if self.snat4_target:
+        self.tables.snat4(self.snat4_target, self.env.get('IPV4_NETWORK', '172.22.1') + '.0/24')
+      if self.snat6_target:
+        self.tables.snat6(self.snat6_target, self.env.get('IPV6_NETWORK', 'fd4d:6169:6c63:6f77::/64'))
+
+  def update_lists(self):
+    self.bans.set_allowlist(self.resolve(self.r.hkeys('F2B_WHITELIST')))
+    self.bans.set_denylist(self.resolve(self.r.hkeys('F2B_BLACKLIST')))
+
+  def isolate(self):
+    if self.env.get('DISABLE_NETFILTER_ISOLATION_RULE', '').lower() in ('y', 'yes'):
+      self.logger.logInfo('Skipping MAILCOW isolation')
+      return
+    self.logger.logInfo('Setting MAILCOW isolation')
+    self.tables.create_mailcow_isolation_rule(self.ISOLATION_INTERFACE, self.ISOLATION_PORTS, self.env.get('MAILCOW_REPLICA_IP', ''))
+
+  def resolve(self, entries):
+    """Networks of the entries; hostnames are resolved, on DNS errors to their last known addresses"""
+    resolver = dns.resolver.Resolver()
+    networks = set()
+    for entry in entries:
+      if BanManager.network(entry) is not None:
+        networks.add(entry)
+        continue
+      addresses = set()
+      for rdtype in ['A', 'AAAA']:
+        try:
+          answer = resolver.resolve(qname=entry, rdtype=rdtype, lifetime=3)
+        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
+          continue
+        except dns.exception.DNSException as ex:
+          self.logger.logInfo('Could not resolve %s: %s' % (entry, ex))
+          addresses = self.resolved.get(entry, set())
+          break
+        addresses.update(rdata.to_text() for rdata in answer)
+      self.resolved[entry] = addresses
+      networks |= addresses
+    return networks
+
+  def snat_target(self, name, version):
+    value = self.env.get(name, '')
+    if not value or value == 'n':
+      return None
+    try:
+      if ipaddress.ip_address(value).version == version:
+        return value
+    except ValueError:
+      pass
+    self.logger.logWarn('%s is not a valid IPv%d address' % (value, version))
+    return None

data/Dockerfiles/netfilter/modules/Options.py

@@ -0,0 +1,120 @@
+import json
+import uuid
+
+
+class Options:
+  """Fail2ban options of the mailcow UI (Redis F2B_OPTIONS). The values are replaced as a whole and never changed
+  in place, so a thread that took them (options.values) keeps a consistent set."""
+
+  DEFAULTS = {
+    'ban_time': 1800,
+    'max_ban_time': 10000,
+    'ban_time_increment': True,
+    'max_attempts': 10,
+    'retry_window': 600,
+    'netban_ipv4': 32,
+    'netban_ipv6': 128,
+    'manage_external': 0,
+  }
+  # Single keys of old installations, read while F2B_OPTIONS does not exist
+  LEGACY_KEYS = {
+    'ban_time': 'F2B_BAN_TIME',
+    'max_ban_time': 'F2B_MAX_BAN_TIME',
+    'ban_time_increment': 'F2B_BAN_TIME_INCREMENT',
+    'max_attempts': 'F2B_MAX_ATTEMPTS',
+    'retry_window': 'F2B_RETRY_WINDOW',
+    'netban_ipv4': 'F2B_NETBAN_IPV4',
+    'netban_ipv6': 'F2B_NETBAN_IPV6',
+  }
+
+  def __init__(self, logger):
+    self.logger = logger
+    self.values = self.normalize({})
+
+  def refresh(self, r):
+    """Reads the options of the UI; options it lacks are added to Redis once, existing ones are never overwritten"""
+    raw = r.get('F2B_OPTIONS')
+    if raw:
+      try:
+        stored = json.loads(raw)
+      except ValueError as ex:
+        self.logger.logWarn('F2B_OPTIONS is no valid JSON, keeping the current options: %s' % ex)
+        return
+      if not isinstance(stored, dict):
+        self.logger.logWarn('F2B_OPTIONS is no JSON object, keeping the current options')
+        return
+    else:
+      stored = {key: r.get(name) for key, name in self.LEGACY_KEYS.items()}
+      stored = {key: value for key, value in stored.items() if value is not None}
+
+    values = self.normalize(stored)
+    self.values = values
+    # First start or old installation: store the complete set, else only what the UI has not saved yet
+    missing = {key: value for key, value in values.items() if not raw or stored.get(key) is None}
+    if missing:
+      self.add_missing(r, missing)
+
+  def ban_time(self, ban_counter):
+    """Seconds of the next ban of a network that was banned ban_counter times before"""
+    values = self.values
+    ban_time = values['ban_time']
+    if values['ban_time_increment']:
+      # the exponent is capped, max_ban_time limits the result anyway
+      ban_time = ban_time * 2 ** min(ban_counter, 32)
+    return max(values['ban_time'], min(ban_time, values['max_ban_time']))
+
+  @classmethod
+  def normalize(cls, stored):
+    values = dict(stored)
+    for key, default in cls.DEFAULTS.items():
+      if isinstance(default, bool):
+        values[key] = cls.to_bool(stored.get(key), default)
+      else:
+        values[key] = cls.to_int(stored.get(key), default)
+    values['ban_time'] = max(1, values['ban_time'])
+    values['max_ban_time'] = max(1, values['max_ban_time'])
+    values['max_attempts'] = max(1, values['max_attempts'])
+    values['retry_window'] = max(1, values['retry_window'])
+    # A short prefix would ban whole regions or everything (/0)
+    values['netban_ipv4'] = min(32, max(8, values['netban_ipv4']))
+    values['netban_ipv6'] = min(128, max(8, values['netban_ipv6']))
+    values['manage_external'] = 1 if values['manage_external'] == 1 else 0
+    if not values.get('banlist_id'):
+      values['banlist_id'] = str(uuid.uuid4())
+    return values
+
+  @staticmethod
+  def to_int(value, default):
+    try:
+      return int(value)
+    except (TypeError, ValueError):
+      return default
+
+  @staticmethod
+  def to_bool(value, default):
+    if isinstance(value, bool):
+      return value
+    if isinstance(value, (int, float)):
+      return value != 0
+    if isinstance(value, str):
+      return value.strip().lower() in ('1', 'true', 'yes', 'on')
+    return default
+
+  def add_missing(self, r, missing):
+    # WATCH keeps a save of the UI between reading and writing
+    def fill(pipe):
+      raw = pipe.get('F2B_OPTIONS')
+      current = json.loads(raw) if raw else {}
+      if not isinstance(current, dict):
+        return
+      added = {key: value for key, value in missing.items() if current.get(key) is None}
+      if not added:
+        return
+      current.update(added)
+      pipe.multi()
+      pipe.set('F2B_OPTIONS', json.dumps(current, ensure_ascii=False))
+
+    try:
+      r.transaction(fill, 'F2B_OPTIONS')
+    except Exception as ex:
+      self.logger.logWarn('Could not add default options to F2B_OPTIONS: %s' % ex)

data/Dockerfiles/netfilter/modules/Regexes.py

@@ -0,0 +1,88 @@
+import ipaddress
+import json
+import re
+
+
+class Regexes:
+  """Rules for log lines (Redis F2B_REGEX): rule id => regex whose first group captures the client address"""
+
+  DEFAULTS = {
+    1: r'mailcow UI: Invalid password for .+ by ([0-9a-f\.:]+)',
+    2: r'Rspamd UI: Invalid password by ([0-9a-f\.:]+)',
+    3: r'warning: .*\[([0-9a-f\.:]+)\]: SASL .+ authentication failed: (?!.*Connection lost to authentication server).+',
+    4: r'warning: non-SMTP command from .*\[([0-9a-f\.:]+)]:.+',
+    5: r'NOQUEUE: reject: RCPT from \[([0-9a-f\.:]+)].+Protocol error.+',
+    6: r'\w+\([^,]+,([0-9a-f\.:]+),<[^>]+>\): Password mismatch \(SHA1 of given password: [a-f0-9]+\)',
+    7: r'\w+\([^,]+,([0-9a-f\.:]+),<[^>]+>\): unknown user \(SHA1 of given password: [a-f0-9]+\)',
+    8: r'SOGo.+ Login from \'([0-9a-f\.:]+)\' for user .+ might not have worked',
+    9: r'([0-9a-f\.:]+) \"GET \/SOGo\/.* HTTP.+\" 403 .+',
+  }
+  # Dovecot could not ask the auth backend (identity provider down, mailcowauth.php unreachable). Postfix logs
+  # such a temporary failure like a wrong password, so failed logins of the address are not counted for a while.
+  TEMPORARY_FAILURE = re.compile(r'lua\([^,]+,([0-9a-f\.:]+)(?:,<[^>]+>)?\): passdb-lua: Upstream (?:unavailable|unreachable)')
+
+  def __init__(self, logger):
+    self.logger = logger
+    self.rules = []
+    self.source = None
+
+  def refresh(self, r):
+    """Compiles the rules of the UI again when they changed; invalid rules are skipped"""
+    raw = r.get('F2B_REGEX')
+    if not raw:
+      raw = json.dumps(self.DEFAULTS, ensure_ascii=False)
+      r.set('F2B_REGEX', raw, nx=True)
+    if raw == self.source:
+      return
+    try:
+      stored = json.loads(raw)
+    except ValueError as ex:
+      self.logger.logWarn('F2B_REGEX is no valid JSON, keeping the current rules: %s' % ex)
+      return
+    if not isinstance(stored, dict):
+      self.logger.logWarn('F2B_REGEX is no JSON object, keeping the current rules')
+      return
+
+    rules = []
+    for rule_id, regex in sorted(stored.items(), key=lambda item: self.sort_key(item[0])):
+      try:
+        rules.append((str(rule_id), re.compile(regex)))
+      except (re.error, TypeError) as ex:
+        self.logger.logWarn('Skipping regex of rule id %s: %s' % (rule_id, ex))
+    self.rules = rules
+    self.source = raw
+
+  def match(self, line):
+    """(rule id, captured address) of the first rule that matches the line, None if no rule matches"""
+    for rule_id, regex in self.rules:
+      result = regex.search(line)
+      if result and result.lastindex and result.group(1):
+        return rule_id, result.group(1)
+    return None
+
+  @classmethod
+  def temporary_failure(cls, line):
+    """Address of a login that failed because the auth backend was unavailable, None for other lines"""
+    result = cls.TEMPORARY_FAILURE.search(line)
+    return result.group(1) if result else None
+
+  @staticmethod
+  def parse_ip(text):
+    """IP address of a captured address, IPv4-mapped IPv6 as IPv4; a trailing colon (e.g. "1.2.3.4:") is
+    tolerated. None if the text is no address."""
+    for candidate in (text, text.rstrip(':')):
+      try:
+        ip = ipaddress.ip_address(candidate)
+      except ValueError:
+        continue
+      if ip.version == 6 and ip.ipv4_mapped:
+        return ip.ipv4_mapped
+      return ip
+    return None
+
+  @staticmethod
+  def sort_key(rule_id):
+    try:
+      return (0, int(rule_id))
+    except (TypeError, ValueError):
+      return (1, str(rule_id))

data/conf/dovecot/auth/mailcowauth.php

@@ -101,7 +101,16 @@ if ($result === false){
   // Init Identity Provider
   $iam_provider = identity_provider('init');
   $iam_settings = identity_provider('get');
-  $result = user_login($post['username'], $post['password'], array('is_internal' => true, 'service' => $post['service']));
+  try {
+    $result = user_login($post['username'], $post['password'], array('is_internal' => true, 'service' => $post['service']));
+  } catch (\Mailcow\Auth\IdpUnavailable $e) {
+    // temporary failure for Dovecot, no failed login
+    error_log("MAILCOWAUTH: Identity provider unavailable for user " . $post['username'] . " with service " . $post['service'] . " from IP " . $post['real_rip'] . ": " . $e->getMessage());
+    http_response_code(503); // Service Unavailable
+    echo json_encode($return);
+    session_destroy();
+    exit;
+  }
   if ($result) {
     error_log('MAILCOWAUTH: User auth for user ' . $post['username'] . " with service " . $post['service'] . " from IP " . $post['real_rip']);
     set_sasl_log($post['username'], $post['real_rip'], $post['service']);

data/conf/dovecot/auth/passwd-verify.lua

@@ -41,6 +41,12 @@ function auth_password_verify(request, password)
     return dovecot.auth.PASSDB_RESULT_INTERNAL_FAILURE, "Upstream unreachable"
   end
 
+  -- 5xx: the backend or the identity provider cannot answer, a temporary failure and no failed login
+  if c >= 500 then
+    dovecot.i_info("HTTP request to auth backend failed with " .. tostring(c) .. " for user " .. request.user)
+    return dovecot.auth.PASSDB_RESULT_INTERNAL_FAILURE, "Upstream unavailable"
+  end
+
   if c ~= 200 and c ~= 401 then
     dovecot.i_info("HTTP request failed with " .. tostring(c) .. " for user " .. request.user)
     return dovecot.auth.PASSDB_RESULT_PASSWORD_MISMATCH, "Upstream error"

data/conf/dovecot/dovecot.conf

@@ -67,9 +67,11 @@ passdb {
 }
 # check for regular password - if empty (e.g. force-passwd-reset), previous pass=yes passdbs also fail
 # a return of the following passdb is mandatory
+# cache_key with password: the lua default (%u) would let a temporary failure fall back to the negative entry
+# of a passdb lookup (master logins) and log "unknown user", which netfilter counts as failed login
 passdb {
   driver = lua
-  args = file=/etc/dovecot/auth/passwd-verify.lua blocking=yes
+  args = file=/etc/dovecot/auth/passwd-verify.lua blocking=yes cache_key=%s:%u:%w
 }
 # Set doveadm_password=your-secret-password in data/conf/dovecot/extra.conf (create if missing)
 service doveadm {

data/web/inc/functions.auth.inc.php

@@ -36,7 +36,18 @@ function check_login($user, $pass, $extra = null) {
 
   // Try validate user
   if (!isset($role) || $role == "user") {
-    $result = user_login($user, $pass, $extra);
+    try {
+      $result = user_login($user, $pass, $extra);
+    } catch (\Mailcow\Auth\IdpUnavailable $e) {
+      // no failed login, netfilter must not count it
+      error_log("mailcow UI: Identity provider unavailable for " . $user . " by " . $_SERVER['REMOTE_ADDR'] . ": " . $e->getMessage());
+      $_SESSION['return'][] =  array(
+        'type' => 'danger',
+        'log' => array(__FUNCTION__, $user, '*', $e->getMessage()),
+        'msg' => 'iam_unavailable'
+      );
+      return false;
+    }
     if ($result !== false) {
       $real_rip = ($_SERVER['HTTP_X_REAL_IP'] ?? $_SERVER['REMOTE_ADDR']);
       set_sasl_log($user, $real_rip, $extra['service']);
@@ -516,11 +527,14 @@ function keycloak_mbox_login_rest($user, $pass, $extra = null){
     return false;
   }
   if (!$iam_provider) {
-    return false;
+    throw new \Mailcow\Auth\IdpUnavailable('Keycloak is not configured completely');
   }
 
   // get access_token for service account of mailcow client
   $admin_token = identity_provider("get-keycloak-admin-token");
+  if (!$admin_token) {
+    throw new \Mailcow\Auth\IdpUnavailable('Could not get an admin token from Keycloak');
+  }
 
   // get the mailcow_password attribute from keycloak user
   $url = "{$iam_settings['server_url']}/admin/realms/{$iam_settings['realm']}/users";
@@ -541,13 +555,9 @@ function keycloak_mbox_login_rest($user, $pass, $extra = null){
   $users = json_decode(curl_exec($curl), true);
   $code = curl_getinfo($curl, CURLINFO_HTTP_CODE);
   curl_close($curl);
+  // the password is checked below, a failed request is never a wrong password
   if ($code != 200) {
-    $_SESSION['return'][] =  array(
-      'type' => 'danger',
-      'log' => array(__FUNCTION__, $user, '*', 'Identity Provider returned HTTP ' . $code),
-      'msg' => 'generic_server_error'
-    );
-    return false;
+    throw new \Mailcow\Auth\IdpUnavailable('Keycloak returned HTTP ' . $code);
   }
   // exact=true can still return several users with that email
   if (!is_array($users) || count($users) != 1) {
@@ -687,7 +697,8 @@ function ldap_mbox_login($user, $pass, $extra = null){
     return false;
   }
   if (!$iam_provider) {
-    return false;
+    // identity_provider('init') could not connect
+    throw new \Mailcow\Auth\IdpUnavailable('No connection to the LDAP server');
   }
 
   try {
@@ -700,33 +711,17 @@ function ldap_mbox_login($user, $pass, $extra = null){
         \Mailcow\Identity\Directory\LdapDirectory::idAttributes($iam_settings['id_attribute'] ?? '')))));
 
     $user_res = $ldap_query->firstOrFail();
-  } catch (Exception $e) {
-    // clear $_SESSION['return'] to not leak data
-    $_SESSION['return'] = array();
-    $_SESSION['return'][] =  array(
-      'type' => 'danger',
-      'log' => array(__FUNCTION__, $user, '*', $e->getMessage()),
-      'msg' => 'generic_server_error'
-    );
+  } catch (\LdapRecord\Query\ObjectNotFoundException $e) {
     return false;
-  }
-  try {
-    if (!$iam_provider->auth()->attempt($user_res['dn'], $pass)) {
-      return false;
-    }
   } catch (Exception $e) {
-    // clear $_SESSION['return'] to not leak data
-    $_SESSION['return'] = array();
-    $_SESSION['return'][] =  array(
-      'type' => 'danger',
-      'log' => array(__FUNCTION__, $user, '*', $e->getMessage()),
-      'msg' => 'generic_server_error'
-    );
+    throw new \Mailcow\Auth\IdpUnavailable('LDAP search failed: ' . $e->getMessage(), 0, $e);
+  }
+  $idp_login = \Mailcow\Auth\AuthModule::fromGlobals()->idp;
+  if (!$idp_login->verifyLdapPassword($iam_provider, (string)$user_res['dn'], (string)$pass)) {
     return false;
   }
 
   // get mapped template
-  $idp_login = \Mailcow\Auth\AuthModule::fromGlobals()->idp;
   $mapped_template = $idp_login->mappedTemplate($user_res[$iam_settings['attribute_field']] ?? array());
   $external_id = \Mailcow\Identity\Directory\LdapDirectory::entryId($user_res, $iam_settings['id_attribute'] ?? '');
 

data/web/inc/functions.fail2ban.inc.php

@@ -60,13 +60,14 @@ function fail2ban($_action, $_data = null, $_extra = null) {
         $queue_unban = $redis->hGetAll('F2B_QUEUE_UNBAN');
         if (is_array($active_bans)) {
           foreach ($active_bans as $network => $banned_until) {
-            $queued_for_unban = (isset($queue_unban[$network]) && $queue_unban[$network] == 1) ? 1 : 0;
-            $difference = $banned_until - time();
+            // Expired bans are lifted by netfilter within seconds
+            $difference = max(0, intval($banned_until) - time());
+            $queued_for_unban = ($difference == 0 || (isset($queue_unban[$network]) && $queue_unban[$network] == 1)) ? 1 : 0;
             $f2b_options['active_bans'][] = array(
               'queued_for_unban' => $queued_for_unban,
               'network' => $network,
               'ip' => strtok($network,'/'),
-              'banned_until' => sprintf('%02dh %02dm %02ds', ($difference/3600), ($difference/60%60), $difference%60)
+              'banned_until' => sprintf('%02dh %02dm %02ds', intdiv($difference, 3600), intdiv($difference, 60) % 60, $difference % 60)
             );
           }
         }
@@ -108,16 +109,15 @@ function fail2ban($_action, $_data = null, $_extra = null) {
             );
             return false;
           }
-          // Rules will also be recreated on log events, but rules may seem empty for a second in the UI
-          docker('post', 'netfilter-mailcow', 'restart');
+          // netfilter recreates the default rules within 10 seconds
           $fail_count = 0;
           $regex_result = json_decode($redis->Get('F2B_REGEX'), true);
-          while (empty($regex_result) && $fail_count < 10) {
+          while (empty($regex_result) && $fail_count < 15) {
             $regex_result = json_decode($redis->Get('F2B_REGEX'), true);
             $fail_count++;
             sleep(1);
           }
-          if ($fail_count >= 10) {
+          if (empty($regex_result)) {
             $_SESSION['return'][] = array(
               'type' => 'danger',
               'log' => array(__FUNCTION__, $_action, $_data_log),
@@ -141,7 +141,7 @@ function fail2ban($_action, $_data = null, $_extra = null) {
           $_SESSION['return'][] = array(
             'type' => 'success',
             'log' => array(__FUNCTION__, $_action, $_data_log),
-            'msg' => array('object_modified', htmlspecialchars($network))
+            'msg' => 'f2b_modified'
           );
           return true;
         }
@@ -172,7 +172,7 @@ function fail2ban($_action, $_data = null, $_extra = null) {
               if (valid_network($network)) {
                 try {
                   $redis->hSet('F2B_WHITELIST', $network, 1);
-                  $redis->hDel('F2B_BLACKLIST', $network, 1);
+                  $redis->hDel('F2B_BLACKLIST', $network);
                   $redis->hSet('F2B_QUEUE_UNBAN', $network, 1);
                 }
                 catch (RedisException $e) {
@@ -199,14 +199,13 @@ function fail2ban($_action, $_data = null, $_extra = null) {
               if (valid_network($network) && !in_array($network, array(
                 '0.0.0.0',
                 '0.0.0.0/0',
-                getenv('IPV4_NETWORK') . '0/24',
-                getenv('IPV4_NETWORK') . '0',
+                getenv('IPV4_NETWORK') . '.0/24',
+                getenv('IPV4_NETWORK') . '.0',
                 getenv('IPV6_NETWORK')
               ))) {
                 try {
                   $redis->hSet('F2B_BLACKLIST', $network, 1);
-                  $redis->hDel('F2B_WHITELIST', $network, 1);
-                  //$response = docker('post', 'netfilter-mailcow', 'restart');
+                  $redis->hDel('F2B_WHITELIST', $network);
                 }
                 catch (RedisException $e) {
                   $_SESSION['return'][] = array(
@@ -261,62 +260,60 @@ function fail2ban($_action, $_data = null, $_extra = null) {
       $f2b_options['ban_time'] = ($ban_time < 60) ? 60 : $ban_time;
       $f2b_options['ban_time_increment'] = ($ban_time_increment == 1) ? true : false;
       $f2b_options['max_ban_time'] = ($max_ban_time < 60) ? 60 : $max_ban_time;
-      $f2b_options['netban_ipv4'] = ($netban_ipv4 < 8) ? 8 : $netban_ipv4;
-      $f2b_options['netban_ipv6'] = ($netban_ipv6 < 8) ? 8 : $netban_ipv6;
-      $f2b_options['netban_ipv4'] = ($netban_ipv4 > 32) ? 32 : $netban_ipv4;
-      $f2b_options['netban_ipv6'] = ($netban_ipv6 > 128) ? 128 : $netban_ipv6;
+      $f2b_options['netban_ipv4'] = min(32, max(8, $netban_ipv4));
+      $f2b_options['netban_ipv6'] = min(128, max(8, $netban_ipv6));
       $f2b_options['max_attempts'] = ($max_attempts < 1) ? 1 : $max_attempts;
       $f2b_options['retry_window'] = ($retry_window < 1) ? 1 : $retry_window;
-      $f2b_options['banlist_id'] = $is_now['banlist_id'];
+      $f2b_options['banlist_id'] = (!empty($is_now['banlist_id'])) ? $is_now['banlist_id'] : uuid4();
       $f2b_options['manage_external'] = ($manage_external > 0) ? 1 : 0;
+      $wl_items = array();
+      if (!empty($wl)) {
+        foreach (array_filter(array_map('trim', preg_split("/( |,|;|\n)/", $wl))) as $wl_item) {
+          if (valid_network($wl_item) || valid_hostname($wl_item)) {
+            $wl_items[$wl_item] = 1;
+          }
+          else {
+            $_SESSION['return'][] = array(
+              'type' => 'danger',
+              'log' => array(__FUNCTION__, $_action, $_data_log),
+              'msg' => array('network_host_invalid', $wl_item)
+            );
+          }
+        }
+      }
+      $bl_items = array();
+      if (!empty($bl)) {
+        foreach (array_filter(array_map('trim', preg_split("/( |,|;|\n)/", $bl))) as $bl_item) {
+          if (valid_network($bl_item) && !in_array($bl_item, array(
+            '0.0.0.0',
+            '0.0.0.0/0',
+            getenv('IPV4_NETWORK') . '.0/24',
+            getenv('IPV4_NETWORK') . '.0',
+            getenv('IPV6_NETWORK')
+          ))) {
+            $bl_items[$bl_item] = 1;
+          }
+          else {
+            $_SESSION['return'][] = array(
+              'type' => 'danger',
+              'log' => array(__FUNCTION__, $_action, $_data_log),
+              'msg' => array('network_host_invalid', $bl_item)
+            );
+          }
+        }
+      }
       try {
+        // One transaction, netfilter must never see empty lists in between
+        $redis->multi();
         $redis->Set('F2B_OPTIONS', json_encode($f2b_options));
-        $redis->Del('F2B_WHITELIST');
-        $redis->Del('F2B_BLACKLIST');
-        if(!empty($wl)) {
-          $wl_array = array_map('trim', preg_split( "/( |,|;|\n)/", $wl));
-          $wl_array = array_filter($wl_array);
-          if (is_array($wl_array)) {
-            foreach ($wl_array as $wl_item) {
-              if (valid_network($wl_item) || valid_hostname($wl_item)) {
-                $redis->hSet('F2B_WHITELIST', $wl_item, 1);
-              }
-              else {
-                $_SESSION['return'][] = array(
-                  'type' => 'danger',
-                  'log' => array(__FUNCTION__, $_action, $_data_log),
-                  'msg' => array('network_host_invalid', $wl_item)
-                );
-                continue;
-              }
-            }
-          }
+        $redis->Del('F2B_WHITELIST', 'F2B_BLACKLIST');
+        if (!empty($wl_items)) {
+          $redis->hMSet('F2B_WHITELIST', $wl_items);
         }
-        if(!empty($bl)) {
-          $bl_array = array_map('trim', preg_split( "/( |,|;|\n)/", $bl));
-          $bl_array = array_filter($bl_array);
-          if (is_array($bl_array)) {
-            foreach ($bl_array as $bl_item) {
-              if (valid_network($bl_item) && !in_array($bl_item, array(
-                '0.0.0.0',
-                '0.0.0.0/0',
-                getenv('IPV4_NETWORK') . '0/24',
-                getenv('IPV4_NETWORK') . '0',
-                getenv('IPV6_NETWORK')
-              ))) {
-                $redis->hSet('F2B_BLACKLIST', $bl_item, 1);
-              }
-              else {
-                $_SESSION['return'][] = array(
-                  'type' => 'danger',
-                  'log' => array(__FUNCTION__, $_action, $_data_log),
-                  'msg' => array('network_host_invalid', $bl_item)
-                );
-                continue;
-              }
-            }
-          }
+        if (!empty($bl_items)) {
+          $redis->hMSet('F2B_BLACKLIST', $bl_items);
         }
+        $redis->exec();
       }
       catch (RedisException $e) {
         $_SESSION['return'][] = array(
@@ -348,7 +345,9 @@ function fail2ban($_action, $_data = null, $_extra = null) {
       if (is_array($_extra)) {
         $_extra = $_extra[0];
       }
-      if ($_extra != $f2b_options['banlist_id']){
+      // Timing safe, a missing or empty id never matches
+      $banlist_id = (string)($f2b_options['banlist_id'] ?? '');
+      if ($banlist_id === '' || !is_string($_extra) || !hash_equals($banlist_id, $_extra)) {
         http_response_code(404);
         return false;
       }

data/web/inc/functions.inc.php

@@ -1292,8 +1292,13 @@ function set_tfa($_data) {
       $row = $stmt->fetch(PDO::FETCH_ASSOC);
       if ($row) {
         if ($row['authsource'] == 'ldap'){
-          if (!ldap_mbox_login($username, $_data["confirm_password"], $iam_settings)) $access_denied = true;
-          else $access_denied = false;
+          try {
+            if (!ldap_mbox_login($username, $_data["confirm_password"], $iam_settings)) $access_denied = true;
+            else $access_denied = false;
+          } catch (\Mailcow\Auth\IdpUnavailable $e) {
+            $_SESSION['return'][] = array('type' => 'danger', 'log' => array(__FUNCTION__, $_data_log, $e->getMessage()), 'msg' => 'iam_unavailable');
+            return false;
+          }
         } else {
           if (!verify_hash($row['password'], $_data["confirm_password"])) $access_denied = true;
           else $access_denied = false;

data/web/inc/lib/mailcow/Auth/IdpUnavailable.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Mailcow\Auth;
+
+use RuntimeException;
+
+/**
+ * The identity provider could not answer a login (unreachable, timeout, server error). No failed login:
+ * callers report a temporary failure and must not count it as an attempt for netfilter.
+ */
+class IdpUnavailable extends RuntimeException
+{
+}

data/web/inc/lib/mailcow/Auth/Service/IdpLoginService.php

@@ -4,6 +4,10 @@ declare(strict_types=1);
 
 namespace Mailcow\Auth\Service;
 
+use LdapRecord\Auth\BindException;
+use LdapRecord\Connection;
+use LdapRecord\LdapRecordException;
+use Mailcow\Auth\IdpUnavailable;
 use Mailcow\Identity\Model\Settings;
 use Mailcow\Identity\Service\LinkService;
 use Mailcow\MailboxGroup\Service\TemplateService;
@@ -19,6 +23,37 @@ class IdpLoginService
     {
     }
 
+    /**
+     * Binds as the user to check the password, then again as the configured user. False for wrong credentials,
+     * IdpUnavailable when the server cannot answer (client side codes are negative: server down, timeout, ...).
+     */
+    public function verifyLdapPassword(Connection $ldap, string $dn, string $password): bool
+    {
+        // an empty password would be an unauthenticated bind, which succeeds
+        if ($dn === '' || $password === '') {
+            return false;
+        }
+        $guard = $ldap->auth();
+        try {
+            $guard->bind($dn, $password);
+            $bound = true;
+        } catch (BindException $e) {
+            $code = $e->getDetailedError()?->getErrorCode();
+            if ($code !== null && ($code < 0 || in_array($code, [51, 52], true))) {
+                throw new IdpUnavailable('LDAP bind failed: ' . $e->getMessage(), 0, $e);
+            }
+            $bound = false;
+        } catch (LdapRecordException $e) {
+            throw new IdpUnavailable('LDAP bind failed: ' . $e->getMessage(), 0, $e);
+        }
+        try {
+            $guard->bindAsConfiguredUser();
+        } catch (LdapRecordException $e) {
+            // the login needs no further LDAP access
+        }
+        return $bound;
+    }
+
     /** Template of the first attribute mapping matching one of the IdP attribute values, null if none */
     public function mappedTemplate(mixed $values): ?string
     {

data/web/lang/lang.cs-cz.json

@@ -395,8 +395,7 @@
         "iam_host": "Hostitel",
         "iam_host_info": "Zadejte jeden či více hostitelů, oddělte čárkou.",
         "iam_import_users": "Importovat uživatele",
-        "iam_auth_flow": "Proces autentizace",
-        "needs_restart": "potřebuje restart"
+        "iam_auth_flow": "Proces autentizace"
     },
     "danger": {
         "access_denied": "Přístup odepřen nebo jsou neplatná data ve formuláři",

data/web/lang/lang.de-de.json

@@ -352,7 +352,6 @@
         "message": "Nachricht",
         "message_size": "Nachrichtengröße",
         "nexthop": "Next Hop",
-        "needs_restart": "benötigt Neustart",
         "no": "✕",
         "no_active_bans": "Keine aktiven Banns",
         "no_new_rows": "Keine weiteren Zeilen vorhanden",
@@ -560,6 +559,7 @@
         "iam_sync_mode_invalid": "Ungültiger Sync-Modus",
         "iam_sync_unavailable": "Der Identity Provider kann nicht synchronisiert werden (nur Keycloak und LDAP)",
         "iam_test_connection": "Verbindung fehlgeschlagen",
+        "iam_unavailable": "Der Identity Provider ist nicht erreichbar, bitte später erneut versuchen.",
         "imagick_exception": "Fataler Bildverarbeitungsfehler",
         "img_dimensions_exceeded": "Grafik überschreitet die maximale Bildgröße",
         "img_invalid": "Grafik konnte nicht validiert werden",

data/web/lang/lang.en-gb.json

@@ -359,7 +359,6 @@
         "message": "Message",
         "message_size": "Message size",
         "nexthop": "Next hop",
-        "needs_restart": "needs restart",
         "no": "✕",
         "no_active_bans": "No active bans",
         "no_new_rows": "No further rows available",
@@ -561,6 +560,7 @@
         "goto_invalid": "Goto address %s is invalid",
         "ham_learn_error": "Ham learn error: %s",
         "iam_test_connection": "Connection failed",
+        "iam_unavailable": "The identity provider is not reachable, please try again later.",
         "imagick_exception": "Error: Imagick exception while reading image",
         "img_dimensions_exceeded": "Image exceeds the maximum image size",
         "img_invalid": "Cannot validate image file",

data/web/lang/lang.es-es.json

@@ -337,7 +337,6 @@
         "iam_host_info": "Introduzca uno o más hosts de LDAP, separados por comas.",
         "iam_import_users": "Importar usuarios",
         "iam_mapping": "Equivalencias de atributos",
-        "needs_restart": "necesita reinicio",
         "oauth2_apps": "Aplicaciones OAuth2",
         "oauth2_add_client": "Añadir cliente OAuth2",
         "oauth2_renew_secret": "Generar nuevo secreto de cliente",

data/web/lang/lang.fr-fr.json

@@ -395,7 +395,6 @@
         "iam_host_info": "Saisissez un ou plusieurs hôtes LDAP, séparés par des virgules.",
         "iam_import_users": "Importer des utilisateurs",
         "filter": "Filtrer",
-        "needs_restart": "nécessite un redémarrage",
         "iam": "Fournisseur d'identité"
     },
     "danger": {

data/web/lang/lang.hu-hu.json

@@ -295,8 +295,7 @@
         "validate_license_now": "GUID érvényesítése a licenszszerverrel szemben",
         "yes": "✓",
         "success": "Siker",
-        "login_page": "Belépő oldal",
-        "needs_restart": "újraindítást igényel"
+        "login_page": "Belépő oldal"
     },
     "edit": {
         "active": "Aktív",

data/web/lang/lang.nl-nl.json

@@ -376,7 +376,6 @@
         "iam_version": "Versie",
         "ignore_ssl_error": "Negeer SSL foutmeldingen",
         "ip_check_opt_in": "Kies ervoor om de externe diensten <strong>ipv4.mailcow.email</strong> en <strong>ipv6.mailcow.email</strong> te gebruiken om externe IP-adressen te bepalen.",
-        "needs_restart": "vereist een herstart",
         "no": "&#10005;",
         "password_reset_info": "Als er geen herstel-e-mailadres is opgegeven, kan deze functie niet worden gebruikt.",
         "password_reset_settings": "Instellingen voor wachtwoordherstel",

data/web/lang/lang.pl-pl.json

@@ -286,7 +286,6 @@
         "merged_vars_hint": "Wyszarzone wiersze zostały połączone z <code>vars. (local.) inc.php</code> i nie można ich modyfikować.",
         "message_size": "Rozmiar wiadomości",
         "nexthop": "Następny hop",
-        "needs_restart": "potrzebny restart",
         "no": "&#10005;",
         "no_active_bans": "Brak aktywnych banów",
         "no_new_rows": "Brak dostępnych kolejnych wierszy",

data/web/lang/lang.pt-br.json

@@ -392,7 +392,6 @@
         "iam_use_tls_info": "Se habilitar TLS, você deve usar a porta padrão para seu servidor LDAP (389). Portas SSL não podem ser usadas.",
         "iam_version": "Versão",
         "ignore_ssl_error": "Ignorar erros SSL",
-        "needs_restart": "precisa reiniciar",
         "quicklink_text": "Mostrar ou ocultar links rápidos para outras páginas de login abaixo do formulário de login",
         "task": "Tarefa",
         "user_link": "Link do usuário",

data/web/lang/lang.ru-ru.json

@@ -395,8 +395,7 @@
         "iam_host": "Хост",
         "iam_host_info": "Укажите один или несколько LDAP-хостов через запятую.",
         "iam_import_users": "Импорт пользователей",
-        "admin_quicklink": "Скрыть ссылку на вход для администраторов",
-        "needs_restart": "необходим перезапуск"
+        "admin_quicklink": "Скрыть ссылку на вход для администраторов"
     },
     "danger": {
         "access_denied": "Доступ запрещён, или указаны неверные данные",

data/web/lang/lang.si-si.json

@@ -397,7 +397,6 @@
         "task": "Naloga",
         "user_link": "Uporabniška povezava",
         "iam_realm": "Realm",
-        "needs_restart": "potreben je ponovni zagon",
         "ui_theme": "Tema",
         "ui_theme_dark": "Vključuje različico s temnim načinom",
         "ui_theme_default": "Privzeto",

data/web/lang/lang.uz-uz.json

@@ -266,7 +266,6 @@
         "message": "Xabar",
         "message_size": "Xabar oʻlchami",
         "nexthop": "Keyingi nuqta",
-        "needs_restart": "qayta ishga tushirish kerak",
         "no": "✕",
         "no_active_bans": "Faol bloklar yoʻq",
         "no_new_rows": "Boshqa satrlar mavjud emas",

data/web/lang/lang.vi-vn.json

@@ -266,7 +266,6 @@
         "message": "Tin nhắn",
         "message_size": "Kích thước tin nhắn",
         "nexthop": "Bước nhảy tiếp theo",
-        "needs_restart": "cần khởi động lại",
         "no_active_bans": "Không có lệnh cấm đang hoạt động",
         "no_new_rows": "Không có hàng nào khác",
         "no_record": "Không có bản ghi",

data/web/lang/lang.zh-cn.json

@@ -395,7 +395,6 @@
         "force_sso": "强制要求单点登录（SSO）",
         "user_link": "自定义链接",
         "app_hide": "在登入界面隐藏",
-        "needs_restart": "需要重启",
         "iam_use_tls_info": "如果使用了 TLS，必须使用 LDAP 服务器的默认端口（389）。SSL 的端口不能使用。"
     },
     "danger": {

data/web/templates/admin/tab-config-f2b.twig

@@ -119,7 +119,7 @@
             {% if active_ban.queued_for_unban == 0 %}
             <a data-action="edit_selected" data-item="{{ active_ban.network }}" data-id="f2b-quick" data-api-url='edit/fail2ban' data-api-attr='{"action":"unban"}' href="#">[{{ lang.admin.queue_unban }}]</a>
             <a data-action="edit_selected" data-item="{{ active_ban.network }}" data-id="f2b-quick" data-api-url='edit/fail2ban' data-api-attr='{"action":"whitelist"}' href="#">[allowlist]</a>
-            <a data-action="edit_selected" data-item="{{ active_ban.network }}" data-id="f2b-quick" data-api-url='edit/fail2ban' data-api-attr='{"action":"blacklist"}' href="#">[denylist (<b>{{ lang.admin.needs_restart }}</b>)]</a>
+            <a data-action="edit_selected" data-item="{{ active_ban.network }}" data-id="f2b-quick" data-api-url='edit/fail2ban' data-api-attr='{"action":"blacklist"}' href="#">[denylist]</a>
             {% else %}
             <i>{{ lang.admin.unban_pending }}</i>
             {% endif %}

docker-compose.yml

@@ -514,7 +514,7 @@ services:
             - acme
 
     netfilter-mailcow:
-      image: ghcr.io/mailcow/netfilter:1.64
+      image: ghcr.io/mailcow/netfilter:1.65
       stop_grace_period: 30s
       restart: always
       privileged: true

helper-scripts/dev_tests/run_netfilter_unit_tests.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+# Runs the unit tests of netfilter (tests/netfilter) in a throwaway container of the locally built netfilter image.
+# The container has its own network namespace, the nftables tests never touch the firewall of the host.
+# Usage: helper-scripts/dev_tests/run_netfilter_unit_tests.sh [unittest arguments, e.g. test_ban_manager]
+
+set -euo pipefail
+
+REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+IMAGE="mailcow-netfilter-tests"
+
+if [ $# -eq 0 ]; then
+  set -- discover -v
+fi
+
+docker build --quiet -t "${IMAGE}" "${REPO_DIR}/data/Dockerfiles/netfilter" > /dev/null
+
+docker run --rm --network none --cap-add NET_ADMIN \
+  -v "${REPO_DIR}/data/Dockerfiles/netfilter:/app:ro" \
+  -v "${REPO_DIR}/tests/netfilter:/tests:ro" \
+  -w /tests -e PYTHONPATH=/app -e PYTHONDONTWRITEBYTECODE=1 \
+  --entrypoint python3 "${IMAGE}" -m unittest "$@"

tests/netfilter/support.py

@@ -0,0 +1,185 @@
+import ipaddress
+
+
+class FakeLogger:
+  def __init__(self):
+    self.lines = []
+
+  def log(self, priority, message):
+    self.lines.append((priority, message))
+
+  def logWarn(self, message):
+    self.log('warn', message)
+
+  def logCrit(self, message):
+    self.log('crit', message)
+
+  def logInfo(self, message):
+    self.log('info', message)
+
+  def messages(self, priority=None):
+    return [message for level, message in self.lines if priority is None or level == priority]
+
+
+class FakePipeline:
+  def __init__(self, r):
+    self.r = r
+
+  def get(self, key):
+    return self.r.get(key)
+
+  def multi(self):
+    pass
+
+  def set(self, key, value):
+    return self.r.set(key, value)
+
+
+class FakePubSub:
+  """Hands out the queued messages; an Exception in the queue is raised, an empty queue sets the stop event"""
+
+  def __init__(self, queue, stop):
+    self.queue = queue
+    self.stop = stop
+    self.channels = []
+    self.closed = False
+
+  def subscribe(self, channel):
+    self.channels.append(channel)
+
+  def get_message(self, timeout=0.0):
+    if not self.queue:
+      self.stop.set()
+      return None
+    item = self.queue.pop(0)
+    if isinstance(item, Exception):
+      raise item
+    return {'type': 'message', 'channel': 'F2B_CHANNEL', 'data': item}
+
+  def close(self):
+    self.closed = True
+
+
+class FakeRedis:
+  def __init__(self):
+    self.data = {}
+    self.messages = []
+    self.stop = None
+    self.pubsubs = []
+
+  def get(self, key):
+    return self.data.get(key)
+
+  def set(self, key, value, nx=False):
+    if nx and key in self.data:
+      return None
+    self.data[key] = str(value)
+    return True
+
+  def exists(self, key):
+    return int(key in self.data)
+
+  def rename(self, src, dst):
+    self.data[dst] = self.data.pop(src)
+
+  def delete(self, *keys):
+    for key in keys:
+      self.data.pop(key, None)
+
+  def hset(self, key, field, value):
+    self.data.setdefault(key, {})[field] = str(value)
+
+  def hdel(self, key, *fields):
+    values = self.data.get(key, {})
+    for field in fields:
+      values.pop(field, None)
+    if not values:
+      self.data.pop(key, None)
+
+  def hkeys(self, key):
+    return list(self.data.get(key, {}))
+
+  def hgetall(self, key):
+    return dict(self.data.get(key, {}))
+
+  def lpush(self, key, value):
+    self.data.setdefault(key, []).insert(0, value)
+
+  def transaction(self, func, *watches):
+    func(FakePipeline(self))
+
+  def pubsub(self, **kwargs):
+    pubsub = FakePubSub(self.messages, self.stop)
+    self.pubsubs.append(pubsub)
+    return pubsub
+
+
+class FakeTables:
+  """Backend with one set of banned networks"""
+
+  def __init__(self):
+    self.rules = set()
+    self.failing = set()
+    self.calls = []
+    self.recreate = False
+
+  def ban(self, sources):
+    failed = []
+    for source in sources:
+      network = str(ipaddress.ip_network(source, strict=False))
+      if network in self.failing:
+        failed.append(network)
+      else:
+        self.rules.add(network)
+    self.calls.append(('ban', list(sources)))
+    return failed
+
+  def unban(self, sources):
+    removed = []
+    for source in sources:
+      network = str(ipaddress.ip_network(source, strict=False))
+      if network in self.rules:
+        self.rules.discard(network)
+        removed.append(network)
+    self.calls.append(('unban', list(sources)))
+    return removed
+
+  def clearIPv4Table(self):
+    self.rules = {rule for rule in self.rules if ipaddress.ip_network(rule).version != 4}
+    self.calls.append(('clear', 4))
+
+  def clearIPv6Table(self):
+    self.rules = {rule for rule in self.rules if ipaddress.ip_network(rule).version != 6}
+    self.calls.append(('clear', 6))
+
+  def initChainIPv4(self):
+    self.calls.append(('init', 4))
+
+  def initChainIPv6(self):
+    self.calls.append(('init', 6))
+
+  def ensure_chains(self):
+    self.calls.append(('ensure_chains',))
+    recreate, self.recreate = self.recreate, False
+    return recreate
+
+  def create_mailcow_isolation_rule(self, interface, ports, allow=''):
+    self.calls.append(('isolation', interface, tuple(ports), allow))
+    return True
+
+  def snat4(self, target, source):
+    self.calls.append(('snat4', target, source))
+
+  def snat6(self, target, source):
+    self.calls.append(('snat6', target, source))
+
+  def names(self):
+    return [call[0] for call in self.calls]
+
+
+class Clock:
+  def __init__(self, now=1000000.0):
+    self.now = now
+
+  def __call__(self):
+    return self.now

tests/netfilter/test_ban_manager.py

@@ -0,0 +1,212 @@
+import ipaddress
+import json
+import threading
+import unittest
+
+from modules.BanManager import BanManager
+from modules.Options import Options
+from support import Clock, FakeLogger, FakeRedis, FakeTables
+
+
+class BanManagerTest(unittest.TestCase):
+  def setUp(self):
+    self.r = FakeRedis()
+    self.tables = FakeTables()
+    self.logger = FakeLogger()
+    self.clock = Clock()
+    self.configure()
+
+  def configure(self, **options):
+    values = {'ban_time': 60, 'max_ban_time': 600, 'max_attempts': 3, 'retry_window': 100}
+    values.update(options)
+    self.r.set('F2B_OPTIONS', json.dumps(values))
+    self.options = Options(self.logger)
+    self.options.refresh(self.r)
+    self.bans = BanManager(self.r, self.tables, self.options, self.logger, threading.RLock(), self.clock)
+
+  def attempts(self, address, count):
+    for _ in range(count):
+      self.bans.attempt(ipaddress.ip_address(address))
+
+  def active(self):
+    return self.r.hgetall('F2B_ACTIVE_BANS')
+
+  def test_bans_after_max_attempts(self):
+    self.attempts('203.0.113.1', 2)
+    self.assertEqual(self.active(), {})
+    self.attempts('203.0.113.1', 1)
+    self.assertEqual(self.active(), {'203.0.113.1/32': str(int(self.clock.now + 60))})
+    self.assertEqual(self.tables.rules, {'203.0.113.1/32'})
+
+  def test_attempts_outside_the_retry_window_start_again(self):
+    self.attempts('203.0.113.1', 2)
+    self.clock.now += 101
+    self.attempts('203.0.113.1', 2)
+    self.assertEqual(self.active(), {})
+
+  def test_netban_prefix(self):
+    self.configure(netban_ipv4=24, netban_ipv6=64)
+    self.attempts('203.0.113.1', 1)
+    self.attempts('203.0.113.2', 2)
+    self.attempts('2001:db8::1', 3)
+    self.assertEqual(set(self.active()), {'203.0.113.0/24', '2001:db8::/64'})
+
+  def test_further_attempts_of_a_banned_network_are_ignored(self):
+    self.attempts('203.0.113.1', 3)
+    self.attempts('203.0.113.1', 5)
+    self.assertEqual(self.tables.names().count('ban'), 1)
+
+  def test_allowlisted_address_is_never_banned(self):
+    self.bans.set_allowlist(['203.0.113.0/24', 'nonsense'])
+    self.attempts('203.0.113.1', 5)
+    self.assertEqual(self.active(), {})
+    self.assertTrue(any('allowlisted' in message for message in self.logger.messages('info')))
+
+  def test_manage_external_only_writes_redis(self):
+    self.configure(manage_external=1)
+    self.attempts('203.0.113.1', 3)
+    self.assertIn('203.0.113.1/32', self.active())
+    self.assertEqual(self.tables.rules, set())
+
+  def test_failed_rule_is_logged_and_ban_kept(self):
+    self.tables.failing.add('203.0.113.1/32')
+    self.attempts('203.0.113.1', 3)
+    self.assertIn('203.0.113.1/32', self.active())
+    self.assertTrue(any('Could not add' in message for message in self.logger.messages('warn')))
+
+  def test_expired_ban_is_lifted_and_next_ban_is_longer(self):
+    self.attempts('203.0.113.1', 3)
+    self.clock.now += 61
+    self.bans.purge()
+    self.assertEqual(self.active(), {})
+    self.assertEqual(self.tables.rules, set())
+    self.attempts('203.0.113.1', 3)
+    self.assertEqual(self.active(), {'203.0.113.1/32': str(int(self.clock.now + 120))})
+
+  def test_purge_keeps_unexpired_bans(self):
+    self.attempts('203.0.113.1', 3)
+    self.clock.now += 30
+    self.bans.purge()
+    self.assertIn('203.0.113.1/32', self.active())
+    self.assertEqual(self.tables.rules, {'203.0.113.1/32'})
+
+  def test_purge_lifts_bans_with_invalid_entries(self):
+    self.r.hset('F2B_ACTIVE_BANS', '203.0.113.1/32', 'soon')
+    self.r.hset('F2B_ACTIVE_BANS', 'nonsense', str(int(self.clock.now + 600)))
+    self.tables.rules.add('203.0.113.1/32')
+    self.bans.purge()
+    self.assertEqual(self.active(), {})
+    self.assertEqual(self.tables.rules, set())
+
+  def test_purge_lifts_bans_whose_redis_entry_is_gone(self):
+    self.attempts('203.0.113.1', 3)
+    self.r.delete('F2B_ACTIVE_BANS')
+    self.bans.purge()
+    self.assertEqual(self.tables.rules, set())
+    self.assertEqual(self.bans.banned, set())
+
+  def test_unban_queue(self):
+    self.attempts('203.0.113.1', 3)
+    self.r.hset('F2B_QUEUE_UNBAN', '203.0.113.1/32', 1)
+    self.bans.purge()
+    self.assertEqual(self.active(), {})
+    self.assertEqual(self.r.hgetall('F2B_QUEUE_UNBAN'), {})
+    self.assertEqual(self.tables.rules, set())
+
+  def test_unban_queue_with_containing_network(self):
+    self.attempts('203.0.113.1', 3)
+    self.attempts('203.0.113.2', 3)
+    self.attempts('198.51.100.1', 3)
+    self.r.hset('F2B_QUEUE_UNBAN', '203.0.113.0/24', 1)
+    self.bans.purge()
+    self.assertEqual(set(self.active()), {'198.51.100.1/32'})
+    self.assertEqual(self.tables.rules, {'198.51.100.1/32'})
+
+  def test_unban_queue_without_ban_removes_left_over_rule(self):
+    self.tables.rules.add('203.0.113.1/32')
+    self.r.hset('F2B_QUEUE_UNBAN', '203.0.113.1/32', 1)
+    self.r.hset('F2B_QUEUE_UNBAN', 'nonsense', 1)
+    self.bans.purge()
+    self.assertEqual(self.tables.rules, set())
+    self.assertEqual(self.r.hgetall('F2B_QUEUE_UNBAN'), {})
+
+  def test_restore_applies_unexpired_bans(self):
+    self.r.hset('F2B_ACTIVE_BANS', '203.0.113.1/32', str(int(self.clock.now + 60)))
+    self.r.hset('F2B_ACTIVE_BANS', '203.0.113.2/32', str(int(self.clock.now - 1)))
+    self.r.hset('F2B_ACTIVE_BANS', 'nonsense', str(int(self.clock.now + 60)))
+    self.bans.restore()
+    self.assertEqual(set(self.active()), {'203.0.113.1/32'})
+    self.assertEqual(self.tables.rules, {'203.0.113.1/32'})
+    self.assertEqual(self.bans.banned, {'203.0.113.1/32'})
+    # one call for all bans
+    self.assertEqual(self.tables.names(), ['ban'])
+
+  def test_restore_with_manage_external(self):
+    self.configure(manage_external=1)
+    self.r.hset('F2B_ACTIVE_BANS', '203.0.113.1/32', str(int(self.clock.now + 60)))
+    self.bans.restore()
+    self.assertEqual(self.bans.banned, {'203.0.113.1/32'})
+    self.assertEqual(self.tables.rules, set())
+
+  def test_denylist(self):
+    self.bans.set_denylist(['198.51.100.0/24', '203.0.113.5', 'nonsense'])
+    self.assertEqual(self.tables.rules, {'198.51.100.0/24', '203.0.113.5/32'})
+    self.assertEqual(set(self.r.hgetall('F2B_PERM_BANS')), {'198.51.100.0/24', '203.0.113.5'})
+    self.bans.set_denylist(['203.0.113.5'])
+    self.assertEqual(self.tables.rules, {'203.0.113.5/32'})
+    self.assertEqual(set(self.r.hgetall('F2B_PERM_BANS')), {'203.0.113.5'})
+
+  def test_unchanged_denylist_does_nothing(self):
+    self.bans.set_denylist(['198.51.100.0/24'])
+    self.bans.set_denylist(['198.51.100.0/24'])
+    self.assertEqual(self.tables.names(), ['ban'])
+
+  def test_denylist_and_automatic_ban_of_the_same_network(self):
+    self.bans.set_denylist(['203.0.113.1/32'])
+    self.r.hset('F2B_ACTIVE_BANS', '203.0.113.1/32', str(int(self.clock.now - 1)))
+    self.bans.banned.add('203.0.113.1/32')
+    self.bans.purge()
+    # the denylist still bans it
+    self.assertEqual(self.tables.rules, {'203.0.113.1/32'})
+
+    self.attempts('198.51.100.1', 3)
+    self.bans.set_denylist(['203.0.113.1/32', '198.51.100.1/32'])
+    self.bans.set_denylist([])
+    # the automatic ban is still active
+    self.assertEqual(self.tables.rules, {'198.51.100.1/32'})
+
+  def test_reapply(self):
+    self.attempts('203.0.113.1', 3)
+    self.bans.set_denylist(['198.51.100.0/24'])
+    self.tables.rules = set()
+    self.bans.reapply()
+    self.assertEqual(self.tables.rules, {'203.0.113.1/32', '198.51.100.0/24'})
+
+  def test_idle_counters_are_dropped(self):
+    self.attempts('203.0.113.1', 1)
+    self.clock.now += 101
+    self.bans.purge()
+    self.assertEqual(self.bans.counters, {})
+
+  def test_excused_address_is_not_counted(self):
+    ip = ipaddress.ip_address('203.0.113.1')
+    self.bans.excuse(ip)
+    self.bans.excuse(ip)
+    self.attempts('203.0.113.1', 5)
+    self.assertEqual(self.active(), {})
+    self.assertEqual(len([message for message in self.logger.messages('warn') if 'Auth backend unavailable' in message]), 1)
+    # other addresses are counted
+    self.attempts('203.0.113.2', 3)
+    self.assertIn('203.0.113.2/32', self.active())
+
+  def test_excuse_ends(self):
+    self.bans.excuse(ipaddress.ip_address('203.0.113.1'))
+    self.clock.now += BanManager.EXCUSE_TIME + 1
+    self.bans.purge()
+    self.assertEqual(self.bans.excused, {})
+    self.attempts('203.0.113.1', 3)
+    self.assertIn('203.0.113.1/32', self.active())
+
+
+if __name__ == '__main__':
+  unittest.main()

tests/netfilter/test_netfilter.py

@@ -0,0 +1,178 @@
+import json
+import unittest
+from unittest import mock
+
+import dns.exception
+import dns.resolver
+
+from modules.Netfilter import Netfilter
+from support import FakeLogger, FakeRedis, FakeTables
+
+
+class NetfilterTest(unittest.TestCase):
+  ENV = {'IPV4_NETWORK': '172.22.1', 'IPV6_NETWORK': 'fd4d:6169:6c63:6f77::/64', 'MAILCOW_REPLICA_IP': ''}
+
+  def setUp(self):
+    self.r = FakeRedis()
+    self.r.set('F2B_OPTIONS', json.dumps({'max_attempts': 2}))
+    self.tables = FakeTables()
+    self.logger = FakeLogger()
+
+  def create(self, **env):
+    netfilter = Netfilter(self.r, self.tables, self.logger, dict(self.ENV, **env))
+    self.r.stop = netfilter.stop
+    netfilter.options.refresh(self.r)
+    netfilter.regexes.refresh(self.r)
+    return netfilter
+
+  def sasl(self, address):
+    return 'warning: unknown[%s]: SASL LOGIN authentication failed: UGFzc3dvcmQ6' % address
+
+  def test_setup_restores_bans_and_resets_the_firewall(self):
+    self.r.hset('F2B_ACTIVE_BANS', '1.2.3.4/32', '9999999999')
+    self.r.hset('F2B_PERM_BANS', '198.51.100.1', '1')
+    self.r.set('F2B_LOG', 'old')
+    netfilter = self.create()
+    netfilter.setup()
+    self.assertEqual(self.tables.names(), ['clear', 'clear', 'init', 'init', 'isolation', 'ban'])
+    self.assertEqual(self.tables.rules, {'1.2.3.4/32'})
+    self.assertEqual(self.r.hgetall('F2B_PERM_BANS'), {})
+    self.assertEqual(self.r.get('NETFILTER_LOG'), 'old')
+
+  def test_isolation_can_be_disabled(self):
+    self.create(DISABLE_NETFILTER_ISOLATION_RULE='y').setup()
+    self.assertNotIn('isolation', self.tables.names())
+
+  def test_isolation_allows_the_replica(self):
+    self.create(MAILCOW_REPLICA_IP='192.0.2.10').setup()
+    self.assertIn(('isolation', 'br-mailcow', (3306, 6379, 8983, 12345), '192.0.2.10'), self.tables.calls)
+
+  def test_shutdown_keeps_the_active_bans(self):
+    netfilter = self.create()
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    self.r.hset('F2B_PERM_BANS', '198.51.100.1', '1')
+    netfilter.shutdown()
+    self.assertEqual(self.tables.rules, set())
+    self.assertIn('1.2.3.4/32', self.r.hgetall('F2B_ACTIVE_BANS'))
+    self.assertEqual(self.r.hgetall('F2B_PERM_BANS'), {})
+
+  def test_handle_line_bans(self):
+    netfilter = self.create()
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    self.assertEqual(self.tables.rules, {'1.2.3.4/32'})
+    self.assertIn('1.2.3.4 matched rule id 3 (%s)' % self.sasl('1.2.3.4'), self.logger.messages('warn'))
+
+  def test_handle_line_with_trailing_colon(self):
+    netfilter = self.create()
+    netfilter.handle_line(self.sasl('1.2.3.4:'))
+    netfilter.handle_line(self.sasl('1.2.3.4:'))
+    self.assertEqual(self.tables.rules, {'1.2.3.4/32'})
+
+  def test_handle_line_ignores_private_and_invalid_addresses(self):
+    netfilter = self.create()
+    for _ in range(3):
+      netfilter.handle_line(self.sasl('192.168.1.1'))
+      netfilter.handle_line(self.sasl('::1'))
+      netfilter.handle_line(self.sasl('1.2.3'))
+    self.assertEqual(self.tables.rules, set())
+    self.assertTrue(any('invalid address 1.2.3' in message for message in self.logger.messages('warn')))
+
+  def test_handle_line_with_replaced_characters(self):
+    netfilter = self.create()
+    line = b'warning: unknown[1.2.3.4]: SASL LOGIN authentication failed: \xff\xfe'.decode('utf-8', 'replace')
+    netfilter.handle_line(line)
+    netfilter.handle_line(line)
+    self.assertEqual(self.tables.rules, {'1.2.3.4/32'})
+
+  def test_unavailable_auth_backend_is_no_failed_login(self):
+    # Dovecot logs the temporary failure, Postfix then the failed SASL login like for a wrong password
+    netfilter = self.create()
+    netfilter.handle_line('auth-worker(1): Error: conn unix:auth-worker (pid=2,uid=401): auth-worker<4>: '
+                          'lua(a@example.org,1.2.3.4): passdb-lua: Upstream unavailable')
+    for _ in range(3):
+      netfilter.handle_line('warning: unknown[1.2.3.4]: SASL LOGIN authentication failed: (reason unavailable), sasl_username=a@example.org')
+    self.assertEqual(self.tables.rules, set())
+    self.assertNotIn('1.2.3.4/32', self.r.hgetall('F2B_ACTIVE_BANS'))
+
+  def test_handle_line_survives_errors(self):
+    netfilter = self.create()
+    with mock.patch.object(netfilter.bans, 'attempt', side_effect=RuntimeError('boom')):
+      netfilter.handle_line(self.sasl('1.2.3.4'))
+    self.assertTrue(any('boom' in message for message in self.logger.messages('warn')))
+
+  def test_watch_reconnects_after_errors(self):
+    netfilter = self.create()
+    self.r.messages.extend([self.sasl('1.2.3.4'), ConnectionError('gone'), self.sasl('1.2.3.4')])
+    netfilter.loop('watch', netfilter.watch, 0)
+    self.assertEqual(self.tables.rules, {'1.2.3.4/32'})
+    self.assertEqual(len(self.r.pubsubs), 2)
+    self.assertTrue(all(pubsub.closed for pubsub in self.r.pubsubs))
+    self.assertTrue(any('gone' in message for message in self.logger.messages('warn')))
+
+  def test_worker_survives_errors(self):
+    netfilter = self.create()
+    calls = []
+
+    def task():
+      calls.append(1)
+      if len(calls) == 1:
+        raise KeyError('ban_time')
+      netfilter.stop.set()
+
+    netfilter.loop('purge', task, 0)
+    self.assertEqual(len(calls), 2)
+
+  def test_recreated_chain_gets_isolation_and_bans_again(self):
+    netfilter = self.create()
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    self.tables.rules = set()
+    self.tables.calls = []
+    self.tables.recreate = True
+    netfilter.check_chains()
+    self.assertEqual(self.tables.names(), ['ensure_chains', 'isolation', 'ban'])
+    self.assertEqual(self.tables.rules, {'1.2.3.4/32'})
+    netfilter.check_chains()
+    self.assertEqual(self.tables.names(), ['ensure_chains', 'isolation', 'ban', 'ensure_chains'])
+
+  def test_snat(self):
+    netfilter = self.create(SNAT_TO_SOURCE='192.0.2.1', SNAT6_TO_SOURCE='192.0.2.2')
+    self.assertIn('snat', netfilter.workers)
+    netfilter.snat()
+    self.assertEqual(self.tables.calls, [('snat4', '192.0.2.1', '172.22.1.0/24')])
+    self.assertTrue(any('192.0.2.2 is not a valid IPv6 address' in message for message in self.logger.messages('warn')))
+    self.assertNotIn('snat', self.create(SNAT_TO_SOURCE='n', SNAT6_TO_SOURCE='n').workers)
+
+  def test_lists(self):
+    netfilter = self.create()
+    self.r.hset('F2B_WHITELIST', '1.2.3.0/24', 1)
+    self.r.hset('F2B_BLACKLIST', '198.51.100.1', 1)
+    netfilter.update_lists()
+    self.assertEqual(self.tables.rules, {'198.51.100.1/32'})
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    netfilter.handle_line(self.sasl('1.2.3.4'))
+    self.assertNotIn('1.2.3.4/32', self.tables.rules)
+
+  def test_resolve_keeps_addresses_on_dns_errors(self):
+    netfilter = self.create()
+    answers = {('relay.example.org', 'A'): ['192.0.2.1'], ('relay.example.org', 'AAAA'): ['2001:db8::1']}
+
+    def resolve(qname, rdtype, lifetime):
+      if qname == 'gone.example.org':
+        raise dns.resolver.NXDOMAIN()
+      if (qname, rdtype) not in answers:
+        raise dns.resolver.NoAnswer()
+      return [mock.Mock(to_text=mock.Mock(return_value=address)) for address in answers[(qname, rdtype)]]
+
+    with mock.patch('dns.resolver.Resolver') as resolver:
+      resolver.return_value.resolve.side_effect = resolve
+      entries = ['relay.example.org', 'gone.example.org', '1.2.3.0/24']
+      self.assertEqual(netfilter.resolve(entries), {'192.0.2.1', '2001:db8::1', '1.2.3.0/24'})
+      resolver.return_value.resolve.side_effect = dns.exception.Timeout()
+      self.assertEqual(netfilter.resolve(entries), {'192.0.2.1', '2001:db8::1', '1.2.3.0/24'})
+
+
+if __name__ == '__main__':
+  unittest.main()

tests/netfilter/test_nftables.py

@@ -0,0 +1,188 @@
+import ipaddress
+import shutil
+import subprocess
+import unittest
+
+import nftables
+
+from modules.NFTables import NFTables
+from support import FakeLogger
+
+
+def nft(command):
+  rc, output, error = nftables.Nftables().cmd(command)
+  if rc != 0:
+    raise RuntimeError(error)
+  return output
+
+
+def kernel_available():
+  # own network namespace with CAP_NET_ADMIN (docker run --network none --cap-add NET_ADMIN)
+  try:
+    nft('add table ip mailcow_test')
+    nft('delete table ip mailcow_test')
+    return True
+  except RuntimeError:
+    return False
+
+
+class RuleParsingTest(unittest.TestCase):
+  def setUp(self):
+    self.tables = NFTables.__new__(NFTables)
+    self.tables.chain_name = 'MAILCOW'
+
+  def test_ban_rule_network(self):
+    prefix = {'expr': [{'match': {'op': '==', 'left': {'payload': {'protocol': 'ip', 'field': 'saddr'}},
+                                  'right': {'prefix': {'addr': '203.0.113.0', 'len': 24}}}},
+                       {'counter': {'packets': 0, 'bytes': 0}}, {'drop': None}]}
+    host = {'expr': [{'match': {'op': '==', 'left': {'payload': {'protocol': 'ip6', 'field': 'saddr'}},
+                                'right': '2001:db8::1'}}, {'drop': None}]}
+    accept = {'expr': [{'match': {'op': '==', 'left': {'payload': {'protocol': 'ip', 'field': 'saddr'}},
+                                  'right': '203.0.113.1'}}, {'accept': None}]}
+    isolation = dict(prefix, comment='mailcow isolation')
+    self.assertEqual(self.tables.ban_rule_network(prefix, 'ip'), ipaddress.ip_network('203.0.113.0/24'))
+    self.assertEqual(self.tables.ban_rule_network(host, 'ip6'), ipaddress.ip_network('2001:db8::1/128'))
+    self.assertIsNone(self.tables.ban_rule_network(host, 'ip'))
+    self.assertIsNone(self.tables.ban_rule_network(accept, 'ip'))
+    self.assertIsNone(self.tables.ban_rule_network(isolation, 'ip'))
+    self.assertIsNone(self.tables.ban_rule_network({'expr': [{'drop': None}]}, 'ip'))
+
+  def test_is_mailcow_jump(self):
+    self.assertTrue(self.tables.is_mailcow_jump({'expr': [{'counter': None}, {'jump': {'target': 'MAILCOW'}}]}))
+    self.assertFalse(self.tables.is_mailcow_jump({'expr': [{'jump': {'target': 'DOCKER-USER'}}], 'comment': 'mailcow'}))
+
+
+@unittest.skipUnless(kernel_available(), 'needs an own network namespace with CAP_NET_ADMIN')
+class KernelTest(unittest.TestCase):
+  def setUp(self):
+    nft('flush ruleset')
+    for family in ['ip', 'ip6']:
+      nft('add table %s filter' % family)
+      nft('add chain %s filter INPUT { type filter hook input priority 0; policy accept; }' % family)
+      nft('add chain %s filter FORWARD { type filter hook forward priority 0; policy drop; }' % family)
+      nft('add chain %s filter DOCKER-USER' % family)
+      nft('add chain %s filter DOCKER-FORWARD' % family)
+      nft('add rule %s filter FORWARD counter jump DOCKER-USER' % family)
+      nft('add rule %s filter FORWARD counter jump DOCKER-FORWARD' % family)
+    nft('add table ip nat')
+    nft('add chain ip nat POSTROUTING { type nat hook postrouting priority 100; policy accept; }')
+    self.logger = FakeLogger()
+    self.tables = NFTables('MAILCOW', self.logger)
+    self.tables.initChainIPv4()
+    self.tables.initChainIPv6()
+
+  def tearDown(self):
+    nft('flush ruleset')
+
+  def rules(self, family, chain, table='filter'):
+    return self.tables.list_rules(family, table, chain)
+
+  def jumps(self, family, chain):
+    return [position for position, rule in enumerate(self.rules(family, chain)) if self.tables.is_mailcow_jump(rule)]
+
+  def bans(self, family='ip'):
+    return [str(self.tables.ban_rule_network(rule, family)) for rule in self.rules(family, 'MAILCOW')
+            if self.tables.ban_rule_network(rule, family)]
+
+  def test_init_puts_one_jump_on_top(self):
+    for family in ['ip', 'ip6']:
+      self.assertEqual(self.jumps(family, 'INPUT'), [0])
+      self.assertEqual(self.jumps(family, 'FORWARD'), [0])
+
+  def test_init_empties_an_existing_chain(self):
+    self.tables.ban(['203.0.113.1'])
+    self.tables.initChainIPv4()
+    self.assertEqual(self.bans(), [])
+    self.assertEqual(self.jumps('ip', 'FORWARD'), [0])
+
+  def test_jump_is_moved_back_to_the_top(self):
+    # docker restarts and inserts its jumps again
+    nft('insert rule ip filter FORWARD counter jump DOCKER-FORWARD')
+    nft('insert rule ip filter FORWARD counter jump DOCKER-USER')
+    self.assertFalse(self.tables.ensure_chains())
+    self.assertEqual(self.jumps('ip', 'FORWARD'), [0])
+    self.assertEqual(len(self.rules('ip', 'FORWARD')), 5)
+
+  def test_duplicate_jumps_without_comment_are_merged(self):
+    # as left by iptables-restore
+    nft('add rule ip filter FORWARD jump MAILCOW')
+    nft('insert rule ip filter FORWARD jump MAILCOW')
+    self.tables.ensure_chains()
+    self.assertEqual(self.jumps('ip', 'FORWARD'), [0])
+    self.assertEqual(self.rules('ip', 'FORWARD')[0].get('comment'), 'mailcow')
+
+  def test_missing_chain_is_recreated(self):
+    self.tables.clearIPv4Table()
+    self.assertIsNone(self.rules('ip', 'MAILCOW'))
+    self.assertTrue(self.tables.ensure_chains())
+    self.assertEqual(self.rules('ip', 'MAILCOW'), [])
+    self.assertEqual(self.jumps('ip', 'INPUT'), [0])
+    self.assertFalse(self.tables.ensure_chains())
+
+  def test_clear_removes_jumps_without_comment(self):
+    nft('add rule ip filter INPUT jump MAILCOW')
+    nft('add rule ip filter DOCKER-USER jump MAILCOW')
+    self.tables.ban(['203.0.113.1'])
+    self.tables.clearIPv4Table()
+    self.assertIsNone(self.rules('ip', 'MAILCOW'))
+    self.assertEqual(self.jumps('ip', 'INPUT'), [])
+    self.assertEqual(self.jumps('ip', 'DOCKER-USER'), [])
+
+  def test_ban_is_idempotent(self):
+    self.assertEqual(self.tables.ban(['203.0.113.1/32', '198.51.100.0/24', '2001:db8::/64']), [])
+    self.assertEqual(self.tables.ban(['203.0.113.1', '198.51.100.0/24']), [])
+    self.assertEqual(sorted(self.bans()), ['198.51.100.0/24', '203.0.113.1/32'])
+    self.assertEqual(self.bans('ip6'), ['2001:db8::/64'])
+
+  def test_unban_removes_duplicates(self):
+    nft('add rule ip filter MAILCOW ip saddr 203.0.113.1 counter drop')
+    nft('add rule ip filter MAILCOW ip saddr 203.0.113.1 counter drop')
+    nft('add rule ip filter MAILCOW ip saddr 203.0.113.2 counter drop')
+    self.assertEqual(self.tables.unban(['203.0.113.1/32']), ['203.0.113.1/32'])
+    self.assertEqual(self.bans(), ['203.0.113.2/32'])
+    self.assertEqual(self.tables.unban(['203.0.113.1/32']), [])
+
+  def test_ban_without_ip6_table(self):
+    self.tables.clearIPv6Table()
+    nft('delete table ip6 filter')
+    self.tables.search_current_chains()
+    self.assertEqual(self.tables.ban(['2001:db8::/64', '203.0.113.1']), ['2001:db8::/64'])
+    self.assertFalse(self.tables.ensure_chains())
+
+  def test_isolation_rules(self):
+    self.tables.ban(['203.0.113.1'])
+    for _ in range(2):
+      self.assertTrue(self.tables.create_mailcow_isolation_rule('br-mailcow', [3306, 6379], '192.0.2.10'))
+    rules = self.rules('ip', 'MAILCOW')
+    self.assertEqual([rule.get('comment') for rule in rules],
+                     ['mailcow isolation allow', 'mailcow isolation allow', 'mailcow isolation', 'mailcow isolation', None])
+    self.assertEqual(self.bans(), ['203.0.113.1/32'])
+    # the ban is not lifted by unbanning the network of the replica
+    self.tables.unban(['192.0.2.10'])
+    self.assertEqual(len(self.rules('ip', 'MAILCOW')), 5)
+
+  def test_isolation_with_invalid_replica_keeps_the_drop_rules(self):
+    self.assertFalse(self.tables.create_mailcow_isolation_rule('br-mailcow', [3306], '2001:db8::1'))
+    self.assertEqual([rule.get('comment') for rule in self.rules('ip', 'MAILCOW')], ['mailcow isolation'])
+
+  @unittest.skipUnless(shutil.which('iptables-nft-save') or shutil.which('iptables-save'), 'needs iptables-save')
+  def test_isolation_rules_survive_iptables_save(self):
+    self.tables.create_mailcow_isolation_rule('br-mailcow', [3306, 6379], '192.0.2.10')
+    binary = shutil.which('iptables-nft-save') or shutil.which('iptables-save')
+    saved = subprocess.run([binary, '-t', 'filter'], capture_output=True, text=True, check=True).stdout
+    lines = [line for line in saved.splitlines() if line.startswith('-A MAILCOW')]
+    self.assertEqual(len(lines), 4)
+    for line in lines:
+      self.assertRegex(line, r'--dport (3306|6379)')
+
+  def test_snat(self):
+    nft('add rule ip nat POSTROUTING counter masquerade')
+    self.tables.snat4('192.0.2.1', '172.22.1.0/24')
+    rules = self.rules('ip', 'POSTROUTING', 'nat')
+    self.assertEqual(rules[0].get('comment'), 'mailcow')
+    self.tables.snat4('192.0.2.1', '172.22.1.0/24')
+    self.assertEqual(len(self.rules('ip', 'POSTROUTING', 'nat')), 2)
+
+
+if __name__ == '__main__':
+  unittest.main()

tests/netfilter/test_options.py

@@ -0,0 +1,103 @@
+import json
+import unittest
+
+from modules.Options import Options
+from support import FakeLogger, FakeRedis
+
+
+class OptionsTest(unittest.TestCase):
+  def setUp(self):
+    self.r = FakeRedis()
+    self.logger = FakeLogger()
+    self.options = Options(self.logger)
+
+  def stored(self):
+    return json.loads(self.r.get('F2B_OPTIONS'))
+
+  def test_first_start_stores_defaults(self):
+    self.options.refresh(self.r)
+    stored = self.stored()
+    for key, value in Options.DEFAULTS.items():
+      self.assertEqual(stored[key], value)
+    self.assertTrue(stored['banlist_id'])
+
+  def test_legacy_keys_are_migrated(self):
+    self.r.set('F2B_BAN_TIME', '900')
+    self.r.set('F2B_MAX_ATTEMPTS', '3')
+    self.options.refresh(self.r)
+    self.assertEqual(self.options.values['ban_time'], 900)
+    self.assertEqual(self.stored()['ban_time'], 900)
+    self.assertEqual(self.stored()['max_attempts'], 3)
+
+  def test_saved_options_are_never_overwritten(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 120, 'max_attempts': 'x', 'banlist_id': 'abc'}))
+    self.options.refresh(self.r)
+    stored = self.stored()
+    self.assertEqual(stored['ban_time'], 120)
+    # invalid values are only replaced in memory
+    self.assertEqual(stored['max_attempts'], 'x')
+    self.assertEqual(self.options.values['max_attempts'], 10)
+    self.assertEqual(stored['banlist_id'], 'abc')
+    # missing keys are added
+    self.assertEqual(stored['retry_window'], 600)
+
+  def test_null_values_are_filled(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 120, 'banlist_id': None}))
+    self.options.refresh(self.r)
+    self.assertTrue(self.stored()['banlist_id'])
+    self.assertEqual(self.stored()['banlist_id'], self.options.values['banlist_id'])
+
+  def test_save_of_the_ui_between_read_and_write_wins(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 120}))
+    original = self.r.transaction
+
+    def transaction(func, *watches):
+      # the UI saves while netfilter adds the missing keys
+      self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 300, 'retry_window': 60}))
+      original(func, *watches)
+
+    self.r.transaction = transaction
+    self.options.refresh(self.r)
+    stored = self.stored()
+    self.assertEqual(stored['ban_time'], 300)
+    self.assertEqual(stored['retry_window'], 60)
+    self.assertEqual(stored['max_attempts'], 10)
+
+  def test_invalid_json_keeps_current_values(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 120}))
+    self.options.refresh(self.r)
+    self.r.set('F2B_OPTIONS', '{broken')
+    self.options.refresh(self.r)
+    self.assertEqual(self.options.values['ban_time'], 120)
+    self.assertTrue(self.logger.messages('warn'))
+    self.assertEqual(self.r.get('F2B_OPTIONS'), '{broken')
+
+  def test_values_are_clamped(self):
+    values = Options.normalize({'netban_ipv4': 0, 'netban_ipv6': 200, 'ban_time': -5, 'max_attempts': 0,
+                                'manage_external': 5, 'ban_time_increment': '0'})
+    self.assertEqual(values['netban_ipv4'], 8)
+    self.assertEqual(values['netban_ipv6'], 128)
+    self.assertEqual(values['ban_time'], 1)
+    self.assertEqual(values['max_attempts'], 1)
+    self.assertEqual(values['manage_external'], 0)
+    self.assertFalse(values['ban_time_increment'])
+
+  def test_ban_time_grows_up_to_max_ban_time(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 60, 'max_ban_time': 600, 'ban_time_increment': True}))
+    self.options.refresh(self.r)
+    self.assertEqual([self.options.ban_time(counter) for counter in range(5)], [60, 120, 240, 480, 600])
+    self.assertEqual(self.options.ban_time(10 ** 6), 600)
+
+  def test_ban_time_without_increment(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 60, 'max_ban_time': 600, 'ban_time_increment': False}))
+    self.options.refresh(self.r)
+    self.assertEqual(self.options.ban_time(3), 60)
+
+  def test_ban_time_is_never_below_ban_time(self):
+    self.r.set('F2B_OPTIONS', json.dumps({'ban_time': 600, 'max_ban_time': 60}))
+    self.options.refresh(self.r)
+    self.assertEqual(self.options.ban_time(0), 600)
+
+
+if __name__ == '__main__':
+  unittest.main()

tests/netfilter/test_regexes.py

@@ -0,0 +1,79 @@
+import ipaddress
+import json
+import unittest
+
+from modules.Regexes import Regexes
+from support import FakeLogger, FakeRedis
+
+
+class RegexesTest(unittest.TestCase):
+  LINES = {
+    '1': ('mailcow UI: Invalid password for admin by 203.0.113.5', '203.0.113.5'),
+    '2': ('Rspamd UI: Invalid password by 2001:db8::5', '2001:db8::5'),
+    '3': ('warning: unknown[203.0.113.6]: SASL LOGIN authentication failed: UGFzc3dvcmQ6', '203.0.113.6'),
+    '4': ('warning: non-SMTP command from unknown[203.0.113.7]: GET / HTTP/1.1', '203.0.113.7'),
+    '5': ('NOQUEUE: reject: RCPT from [203.0.113.8]: 503 5.5.0 <a@example.org>: Protocol error; proto=ESMTP', '203.0.113.8'),
+    '6': ('auth: lua(user@example.org,203.0.113.9,<abcdef>): Password mismatch (SHA1 of given password: 0123ab)', '203.0.113.9'),
+    '7': ('auth: lua(nobody@example.org,203.0.113.10,<abcdef>): unknown user (SHA1 of given password: 0123ab)', '203.0.113.10'),
+    '8': ("SOGo[123]: Login from '203.0.113.11' for user nobody might not have worked", '203.0.113.11'),
+    '9': ('203.0.113.12 "GET /SOGo/so/ HTTP/1.1" 403 123 "-"', '203.0.113.12'),
+  }
+
+  def setUp(self):
+    self.r = FakeRedis()
+    self.logger = FakeLogger()
+    self.regexes = Regexes(self.logger)
+
+  def test_defaults_are_stored_and_match(self):
+    self.regexes.refresh(self.r)
+    self.assertEqual(set(json.loads(self.r.get('F2B_REGEX'))), {str(rule_id) for rule_id in Regexes.DEFAULTS})
+    for rule_id, (line, address) in self.LINES.items():
+      self.assertEqual(self.regexes.match(line), (rule_id, address), line)
+
+  def test_lost_connection_to_auth_server_is_no_attempt(self):
+    self.regexes.refresh(self.r)
+    line = 'warning: unknown[203.0.113.6]: SASL LOGIN authentication failed: Connection lost to authentication server'
+    self.assertIsNone(self.regexes.match(line))
+
+  def test_only_the_first_matching_rule_counts(self):
+    self.r.set('F2B_REGEX', json.dumps({'2': r'by ([0-9a-f\.:]+)', '10': r'from ([0-9a-f\.:]+)', '1': r'login from ([0-9a-f\.:]+)'}))
+    self.regexes.refresh(self.r)
+    self.assertEqual(self.regexes.match('login from 203.0.113.1 by 203.0.113.2'), ('1', '203.0.113.1'))
+    self.assertEqual([rule_id for rule_id, _ in self.regexes.rules], ['1', '2', '10'])
+
+  def test_invalid_regex_is_skipped(self):
+    self.r.set('F2B_REGEX', json.dumps({'1': r'([0-9', '2': r'by ([0-9a-f\.:]+)'}))
+    self.regexes.refresh(self.r)
+    self.assertEqual(self.regexes.match('by 203.0.113.2'), ('2', '203.0.113.2'))
+    self.assertTrue(self.logger.messages('warn'))
+
+  def test_invalid_json_keeps_current_rules(self):
+    self.regexes.refresh(self.r)
+    self.r.set('F2B_REGEX', '{broken')
+    self.regexes.refresh(self.r)
+    self.assertEqual(len(self.regexes.rules), len(Regexes.DEFAULTS))
+
+  def test_rules_without_group_do_not_match(self):
+    self.r.set('F2B_REGEX', json.dumps({'1': r'failed', '2': r'by ([0-9a-f\.:]+)?'}))
+    self.regexes.refresh(self.r)
+    self.assertIsNone(self.regexes.match('failed by nobody'))
+
+  def test_temporary_failure(self):
+    worker = 'auth-worker(3933): Error: conn unix:auth-worker (pid=3932,uid=401): auth-worker<4>: '
+    self.assertEqual(Regexes.temporary_failure(worker + 'lua(a@example.org,203.0.113.1,<GGexLdBezoasFgEB>): passdb-lua: Upstream unavailable'), '203.0.113.1')
+    self.assertEqual(Regexes.temporary_failure(worker + 'lua(a@example.org,2001:db8::1): passdb-lua: Upstream unreachable'), '2001:db8::1')
+    self.assertIsNone(Regexes.temporary_failure('auth: lua(a@example.org,203.0.113.1,<GGexLdBezoasFgEB>): unknown user (SHA1 of given password: 0123ab)'))
+    self.assertIsNone(Regexes.temporary_failure('warning: unknown[203.0.113.1]: SASL LOGIN authentication failed: (reason unavailable)'))
+
+  def test_parse_ip(self):
+    self.assertEqual(Regexes.parse_ip('203.0.113.1'), ipaddress.ip_address('203.0.113.1'))
+    self.assertEqual(Regexes.parse_ip('203.0.113.1:'), ipaddress.ip_address('203.0.113.1'))
+    self.assertEqual(Regexes.parse_ip('2001:db8::1'), ipaddress.ip_address('2001:db8::1'))
+    self.assertEqual(Regexes.parse_ip('::ffff:203.0.113.1'), ipaddress.ip_address('203.0.113.1'))
+    self.assertIsNone(Regexes.parse_ip('203.0.113'))
+    self.assertIsNone(Regexes.parse_ip('...'))
+    self.assertIsNone(Regexes.parse_ip(''))
+
+
+if __name__ == '__main__':
+  unittest.main()

tests/php/Auth/Service/IdpLoginServiceTest.php

@@ -4,11 +4,19 @@ declare(strict_types=1);
 
 namespace Mailcow\Tests\Auth\Service;
 
+use Exception;
+use LdapRecord\Auth\BindException;
+use LdapRecord\Auth\Guard;
+use LdapRecord\Connection;
+use LdapRecord\ConnectionException;
+use LdapRecord\DetailedError;
+use Mailcow\Auth\IdpUnavailable;
 use Mailcow\Auth\Service\IdpLoginService;
 use Mailcow\Identity\Model\Settings;
 use Mailcow\Identity\Service\LinkService;
 use Mailcow\MailboxGroup\Service\TemplateService;
 use PDOException;
+use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\TestCase;
 
 class IdpLoginServiceTest extends TestCase
@@ -64,4 +72,76 @@ class IdpLoginServiceTest extends TestCase
         $this->assertNull($login->mappedTemplate(null));
         $this->assertNull($login->mappedTemplate(['unknown']));
     }
+
+    public function testLdapPasswordIsCheckedWithABindAsTheUser(): void
+    {
+        $guard = $this->createMock(Guard::class);
+        $guard->expects($this->once())->method('bind')->with('uid=a,dc=example,dc=org', 'secret');
+        $guard->expects($this->once())->method('bindAsConfiguredUser');
+
+        $this->assertTrue($this->login()->verifyLdapPassword($this->ldap($guard), 'uid=a,dc=example,dc=org', 'secret'));
+    }
+
+    public function testWrongLdapPassword(): void
+    {
+        $guard = $this->createMock(Guard::class);
+        $guard->method('bind')->willThrowException($this->bindError(49, 'Invalid credentials'));
+        $guard->expects($this->once())->method('bindAsConfiguredUser');
+
+        $this->assertFalse($this->login()->verifyLdapPassword($this->ldap($guard), 'uid=a,dc=example,dc=org', 'wrong'));
+    }
+
+    public function testEmptyLdapPasswordNeverBinds(): void
+    {
+        $guard = $this->createMock(Guard::class);
+        $guard->expects($this->never())->method('bind');
+
+        $this->assertFalse($this->login()->verifyLdapPassword($this->ldap($guard), 'uid=a,dc=example,dc=org', ''));
+    }
+
+    public static function ldapOutages(): array
+    {
+        return [
+            'server down' => [-1, "Can't contact LDAP server"],
+            'timeout' => [-5, 'Timed out'],
+            'busy' => [51, 'Busy'],
+            'unavailable' => [52, 'Unavailable'],
+        ];
+    }
+
+    #[DataProvider('ldapOutages')]
+    public function testLdapOutageIsNoWrongPassword(int $code, string $message): void
+    {
+        $guard = $this->createMock(Guard::class);
+        $guard->method('bind')->willThrowException($this->bindError($code, $message));
+
+        $this->expectException(IdpUnavailable::class);
+        $this->login()->verifyLdapPassword($this->ldap($guard), 'uid=a,dc=example,dc=org', 'secret');
+    }
+
+    public function testFailedStartTlsIsAnOutage(): void
+    {
+        $guard = $this->createMock(Guard::class);
+        $guard->method('bind')->willThrowException(new ConnectionException('Unable to connect'));
+
+        $this->expectException(IdpUnavailable::class);
+        $this->login()->verifyLdapPassword($this->ldap($guard), 'uid=a,dc=example,dc=org', 'secret');
+    }
+
+    private function login(): IdpLoginService
+    {
+        return new IdpLoginService(Settings::fromArray([]), $this->createMock(LinkService::class), $this->createMock(TemplateService::class));
+    }
+
+    private function ldap(Guard $guard): Connection
+    {
+        $ldap = $this->createMock(Connection::class);
+        $ldap->method('auth')->willReturn($guard);
+        return $ldap;
+    }
+
+    private function bindError(int $code, string $message): BindException
+    {
+        return BindException::withDetailedError(new Exception($message, $code), new DetailedError($code, $message, null));
+    }
 }