From b7d24881b11acf03e684d3b942b2ad0fdd43f5be Mon Sep 17 00:00:00 2001
From: FreddleSpl0it <75116288+FreddleSpl0it@users.noreply.github.com>
Date: Tue, 27 Oct 2026 12:54:45 +0100
Subject: [PATCH] [Netfilter][Tests] Add netfilter unit tests and identity
 provider outage tests

---
 .../dev_tests/run_netfilter_unit_tests.sh     |  21 ++
 tests/netfilter/support.py                    | 185 +++++++++++++++
 tests/netfilter/test_ban_manager.py           | 212 ++++++++++++++++++
 tests/netfilter/test_netfilter.py             | 178 +++++++++++++++
 tests/netfilter/test_nftables.py              | 188 ++++++++++++++++
 tests/netfilter/test_options.py               | 103 +++++++++
 tests/netfilter/test_regexes.py               |  79 +++++++
 .../php/Auth/Service/IdpLoginServiceTest.php  |  80 +++++++
 8 files changed, 1046 insertions(+)
 create mode 100755 helper-scripts/dev_tests/run_netfilter_unit_tests.sh
 create mode 100644 tests/netfilter/support.py
 create mode 100644 tests/netfilter/test_ban_manager.py
 create mode 100644 tests/netfilter/test_netfilter.py
 create mode 100644 tests/netfilter/test_nftables.py
 create mode 100644 tests/netfilter/test_options.py
 create mode 100644 tests/netfilter/test_regexes.py

diff --git a/helper-scripts/dev_tests/run_netfilter_unit_tests.sh b/helper-scripts/dev_tests/run_netfilter_unit_tests.sh
new file mode 100755
index 000000000..8fecc6aa4
--- /dev/null
+++ b/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 "$@"
diff --git a/tests/netfilter/support.py b/tests/netfilter/support.py
new file mode 100644
index 000000000..f007bebf6
--- /dev/null
+++ b/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
diff --git a/tests/netfilter/test_ban_manager.py b/tests/netfilter/test_ban_manager.py
new file mode 100644
index 000000000..ebacea568
--- /dev/null
+++ b/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()
diff --git a/tests/netfilter/test_netfilter.py b/tests/netfilter/test_netfilter.py
new file mode 100644
index 000000000..dc702335b
--- /dev/null
+++ b/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()
diff --git a/tests/netfilter/test_nftables.py b/tests/netfilter/test_nftables.py
new file mode 100644
index 000000000..d62d284ca
--- /dev/null
+++ b/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()
diff --git a/tests/netfilter/test_options.py b/tests/netfilter/test_options.py
new file mode 100644
index 000000000..db8dab4d4
--- /dev/null
+++ b/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()
diff --git a/tests/netfilter/test_regexes.py b/tests/netfilter/test_regexes.py
new file mode 100644
index 000000000..e55cac4f5
--- /dev/null
+++ b/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()
diff --git a/tests/php/Auth/Service/IdpLoginServiceTest.php b/tests/php/Auth/Service/IdpLoginServiceTest.php
index dbc5acd23..7b406cac2 100644
--- a/tests/php/Auth/Service/IdpLoginServiceTest.php
+++ b/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));
+    }
 }