diff --git a/data/Dockerfiles/netfilter/docker-entrypoint.sh b/data/Dockerfiles/netfilter/docker-entrypoint.sh index 98cab0a72..e58bcc0b7 100755 --- a/data/Dockerfiles/netfilter/docker-entrypoint.sh +++ b/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 diff --git a/data/Dockerfiles/netfilter/main.py b/data/Dockerfiles/netfilter/main.py index 5b718a94b..019f185d6 100644 --- a/data/Dockerfiles/netfilter/main.py +++ b/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) \ No newline at end of file +#!/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) diff --git a/data/Dockerfiles/netfilter/modules/BanManager.py b/data/Dockerfiles/netfilter/modules/BanManager.py new file mode 100644 index 000000000..77818dd4a --- /dev/null +++ b/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) diff --git a/data/Dockerfiles/netfilter/modules/IPTables.py b/data/Dockerfiles/netfilter/modules/IPTables.py index 3d3d43974..92491fbaf 100644 --- a/data/Dockerfiles/netfilter/modules/IPTables.py +++ b/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 \ No newline at end of file + return False diff --git a/data/Dockerfiles/netfilter/modules/NFTables.py b/data/Dockerfiles/netfilter/modules/NFTables.py index 7740fa59f..db3585e2a 100644 --- a/data/Dockerfiles/netfilter/modules/NFTables.py +++ b/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 [] + # nft (add | create) chain []
_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 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 nat - _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 \ No newline at end of file + # 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 diff --git a/data/Dockerfiles/netfilter/modules/Netfilter.py b/data/Dockerfiles/netfilter/modules/Netfilter.py new file mode 100644 index 000000000..bd75938c8 --- /dev/null +++ b/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 diff --git a/data/Dockerfiles/netfilter/modules/Options.py b/data/Dockerfiles/netfilter/modules/Options.py new file mode 100644 index 000000000..bab32c9ec --- /dev/null +++ b/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) diff --git a/data/Dockerfiles/netfilter/modules/Regexes.py b/data/Dockerfiles/netfilter/modules/Regexes.py new file mode 100644 index 000000000..8818d4ee3 --- /dev/null +++ b/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)) diff --git a/docker-compose.yml b/docker-compose.yml index fda28e8ea..cd50604f9 100644 --- a/docker-compose.yml +++ b/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