#!/usr/bin/env python3 import sys, re, os, subprocess, json, requests, spf, time, random, shutil, dns.resolver import fcntl from datetime import datetime, date, timedelta, timezone from pathlib import Path from collections import defaultdict from typing import Iterator, Optional # Python 3.11+ : tomllib in der Standardlib. Auf 3.10 und aelter -> tomli. try: import tomllib # type: ignore except ModuleNotFoundError: # pragma: no cover import tomli as tomllib # type: ignore # --------------------------------------------------------------------------- # Konfiguration laden # --------------------------------------------------------------------------- # Pfad zur Config-Datei kann via ENV ueberschrieben werden. Der Default liegt # bewusst unter /etc — Sicherheits-relevante Listen gehoeren nicht ins # Working-Directory eines Daemons. CONFIG_FILE = Path(os.getenv("CONFIG_FILE", "/etc/policyguard.toml")) date_now = datetime.now() LOCAL_TZ = datetime.now().astimezone().tzinfo or timezone.utc def _fatal(msg: str) -> None: """Schreibt eine Fehlermeldung nach stderr und beendet das Programm. Wird nur beim Start verwendet — danach laeuft das Skript als Daemon und soll Fehler nur loggen, nicht crashen.""" sys.stderr.write(f"spf-policy: FATAL: {msg}\n") sys.stderr.flush() sys.exit(2) def _load_config(path: Path) -> dict: backup = Path("/t/policyguard_backup.toml") if not path.is_file(): # Haupt-Config fehlt: versuche lastknowngood-Backup if backup.is_file(): sys.stderr.write( f"spf-policy: WARN: Config-Datei {path} nicht gefunden, " f"falle zurueck auf Backup {backup}\n" ) try: with open(backup, "rb") as f: return tomllib.load(f) except tomllib.TOMLDecodeError as e: _fatal(f"Config-Backup {backup} ist kein gueltiges TOML: {e}") except OSError as e: _fatal(f"Config-Backup {backup} nicht lesbar: {e}") _fatal( f"Config-Datei nicht gefunden: {path}\n" f" Pfad via Env-Var CONFIG_FILE ueberschreibbar.\n" f" Eine Beispiel-Config liegt im Repo neben dem Skript." ) try: with open(path, "rb") as f: cfg = tomllib.load(f) except tomllib.TOMLDecodeError as e: # Config-Fehler: versuche lastknowngood-Backup if backup.is_file(): sys.stderr.write( f"spf-policy: WARN: Config-Datei {path} ist kein gueltiges TOML: {e}\n" f" falle zurueck auf Backup {backup}\n" ) try: with open(backup, "rb") as f: return tomllib.load(f) except tomllib.TOMLDecodeError as e2: _fatal(f"Config-Backup {backup} ist ebenfalls kein gueltiges TOML: {e2}") except OSError as e2: _fatal(f"Config-Backup {backup} nicht lesbar: {e2}") _fatal(f"Config-Datei {path} ist kein gueltiges TOML: {e}") except OSError as e: if backup.is_file(): sys.stderr.write( f"spf-policy: WARN: Config-Datei {path} nicht lesbar: {e}\n" f" falle zurueck auf Backup {backup}\n" ) try: with open(backup, "rb") as f: return tomllib.load(f) except tomllib.TOMLDecodeError as e2: _fatal(f"Config-Backup {backup} ist kein gueltiges TOML: {e2}") except OSError as e2: _fatal(f"Config-Backup {backup} nicht lesbar: {e2}") _fatal(f"Config-Datei {path} nicht lesbar: {e}") # Config erfolgreich eingelesen: lastknowngood-Backup anlegen, aber # nur bei geaenderter mtime der Config. copy2 uebernimmt die # Quell-mtime aufs Backup, der Vergleich reicht also als Waechter. # Tmp-Datei + os.replace(): atomar, kein Torn-File beim parallelen # Start mehrerer Policy-Worker. Der Block bleibt bewusst HINTER dem # erfolgreichen tomllib.load(), damit eine ungueltige Config nie ins # Backup durchrutscht. try: if not backup.exists() or \ backup.stat().st_mtime_ns < path.stat().st_mtime_ns: tmp = backup.with_name(f"{backup.name}.{os.getpid()}.tmp") shutil.copy2(path, tmp) os.replace(tmp, backup) except OSError as e: sys.stderr.write( f"spf-policy: WARN: lastknowngood-Backup konnte nicht " f"erstellt werden: {e}\n" ) return cfg def _cfg_section(cfg: dict, name: str) -> dict: section = cfg.get(name) if section is None: return {} if not isinstance(section, dict): _fatal(f"Config: Sektion [{name}] ist kein Tabellen-Block.") return section def _cfg_list(section: dict, key: str, section_name: str) -> list: val = section.get(key, []) if not isinstance(val, list): sys.stderr.write( f"spf-policy: WARN: [{section_name}].{key} ist kein Array, " f"wird als leer interpretiert.\n" ) return [] for item in val: if not isinstance(item, str): sys.stderr.write( f"spf-policy: WARN: [{section_name}].{key} enthaelt Nicht-String " f"{item!r}, Eintrag wird uebersprungen.\n" ) return [item for item in val if isinstance(item, str)] def _env_or_int(env_name: str, fallback: int) -> int: raw = os.getenv(env_name) if raw is None: return fallback try: return int(raw) except ValueError: sys.stderr.write( f"spf-policy: WARN: ENV {env_name}={raw!r} ist keine Zahl, " f"fallback auf Config-Wert {fallback}\n" ) return fallback def _env_or_str(env_name: str, fallback: str) -> str: return os.getenv(env_name, fallback) _CFG = _load_config(CONFIG_FILE) # --- [tunables] (ENV ueberschreibt Config) --------------------------------- _t = _cfg_section(_CFG, "tunables") DAYS = _env_or_int("DAYS", int(_t.get("days", 50))) MIN_COUNT = _env_or_int("MIN_COUNT", int(_t.get("min_count", 4))) BLOCK_QUOTA = _env_or_int("BLOCK_QUOTA", int(_t.get("block_quota", 76))) PRETEND_NON_EXISTING_MAILBOX = int(_t.get("pretend_non_existing_mailbox", 0)) GENERIC_CACHE_SIZE = int(_t.get("generic_cache_size", 4096)) INCREMENTAL_UPDATE_INTERVAL = _env_or_int("INCREMENTAL_UPDATE_INTERVAL", int(_t.get("incremental_update_interval", 30))) LOG_RATE_LIMIT_PER_SEC = _env_or_int("LOG_RATE_LIMIT_PER_SEC", int(_t.get("log_rate_limit_per_sec", 200))) INITIAL_IMPORT_FLUSH_EVERY_DAYS = _env_or_int("INITIAL_IMPORT_FLUSH_EVERY_DAYS", int(_t.get("initial_import_flush_every_days", 5))) MAX_RECIPIENTS_PER_MESSAGE = int(_t.get("max_recipients_per_message", 0)) # --- Greylisting --------------------------------------------------------------- GREYLISTING_ENABLED = _env_or_int("GREYLISTING_ENABLED", int(_t.get("greylisting_enabled", 1))) GREYLIST_EXPIRE_MAIL_S = _env_or_int("GREYLIST_EXPIRE_MAIL_SECONDS", int(_t.get("greylist_expire_mail_seconds", 86400))) if DAYS < 1: _fatal("tunables.days muss >= 1 sein") if MIN_COUNT < 1: _fatal("tunables.min_count muss >= 1 sein") if not (0 < BLOCK_QUOTA <= 100): _fatal("tunables.block_quota muss zwischen 1 und 100 liegen") # Pre-computed Block Quota fuer schnellere Division-Checks BLOCK_QUOTA_FLOAT = BLOCK_QUOTA / 100.0 # Pre-computed NON_EXISTING_MAILBOX fuer schnellere Division-Checks if not (0 <= PRETEND_NON_EXISTING_MAILBOX <= 100): _fatal("tunables.pretend_non_existing_mailbox muss zwischen 0 und 100 liegen") NON_EXISTING_MAILBOX_FLOAT = PRETEND_NON_EXISTING_MAILBOX / 100.0 # --- [paths] (ENV ueberschreibt Config) ------------------------------------ _p = _cfg_section(_CFG, "paths") TRANSPORTMAP = _env_or_str("TRANSPORTMAP", str(_p.get("transport_map", "/etc/pmg/transport"))) # Kein Journal-Cursor mehr: heutige Stats werden bei jedem Refresh frisch aus # dem Journal neu berechnet (idempotent), historische Tage kommen einmalig aus # dem CACHE. Das verhindert das Mehrfachzaehlen derselben Events, das bei # parallelen, kurzlebigen Postfix-Workern mit per-PID-Cursordateien auftrat. CACHE_FILE = Path(_env_or_str("CACHE_FILE", str(_p.get("cache_file", "/t/postfix_cache.json")))) BLOCKLIST_FILE = Path(_env_or_str("BLOCKLIST_FILE", str(_p.get("blocklist_file", "/t/blocklist.json")))) SEEN_FILE = Path(_env_or_str("SEEN_FILE", str(_p.get("seen_file", "/t/policyguard-seen.json")))) SEEN_LOCK_FILE = Path(str(SEEN_FILE) + ".lock") def _norm_domain_set(items: list) -> set[str]: return {i.lower().rstrip(".") for i in items} # --- [domains] ------------------------------------------------------------- _d = _cfg_section(_CFG, "domains") WHITELIST_AUTOBLACKLIST = set(_cfg_list(_d, "whitelist_autoblacklist", "domains")) # TLD-/Suffix-Ausnahmen fuer die Auto-Blocklist. Muessen mit "." beginnen, # damit "co.de" nicht versehentlich auch ".code" matchen wuerde. WHITELIST_AUTOBLACKLIST_TLDS_RAW = _cfg_list(_d, "whitelist_autoblacklist_tlds", "domains") for _tld in WHITELIST_AUTOBLACKLIST_TLDS_RAW: if not _tld.startswith("."): _fatal( f"Config: [domains].whitelist_autoblacklist_tlds — Eintrag {_tld!r} muss " f"mit '.' beginnen (z.B. '.de', '.co.uk')." ) WHITELIST_AUTOBLACKLIST_TLDS = tuple(t.lower() for t in WHITELIST_AUTOBLACKLIST_TLDS_RAW) BLOCKLIST_SENDER = _norm_domain_set(_cfg_list(_d, "blocklist_sender", "domains")) BLOCKLIST_CLIENT = _norm_domain_set(_cfg_list(_d, "blocklist_client", "domains")) BLOCKLIST_RECIPIENT = _norm_domain_set(_cfg_list(_d, "blocklist_recipient", "domains")) UNTRUSTED_HELO = _norm_domain_set(_cfg_list(_d, "untrusted_helo", "domains")) WHITELIST_HELO = _norm_domain_set(_cfg_list(_d, "whitelist_helo", "domains")) WHITELIST_CLIENT = _norm_domain_set(_cfg_list(_d, "whitelist_client", "domains")) WHITELIST_SENDER = _norm_domain_set(_cfg_list(_d, "whitelist_sender", "domains")) DNSBL_RAW = _cfg_list(_d, "dnsbl", "domains") # --- [regex] --------------------------------------------------------------- _r = _cfg_section(_CFG, "regex") BLOCKLIST_REGEX = _cfg_list(_r, "blocklist_sender_patterns", "regex") # Sanity: leere kritische Listen sind verdaechtig — wir bauen den Daemon # defensiv und warnen, brechen aber nicht ab (eine leere whitelist_helo # z.B. ist voellig legitim). if not UNTRUSTED_HELO: sys.stderr.write( "spf-policy: WARN: [domains].untrusted_helo ist leer — " "untrusted-HELO-spezifische Regeln greifen nicht.\n" ) # Regex-Patterns einmal kompilieren (Validierung passiert hier implizit: # ein ungueltiges Pattern crasht beim Start, nicht erst bei der ersten Mail) try: BLOCKLIST_REGEX_COMPILED = [re.compile(p) for p in BLOCKLIST_REGEX] except re.error as e: _fatal(f"Config: ungueltiger Regex in [regex].blocklist_sender_patterns: {e}") def _expand_dnsbl_entry(entry: str) -> tuple[str, set[str] | None]: if "=" not in entry: return entry.strip(), None domain, value = entry.split("=", 1) domain = domain.strip() value = value.strip() m = re.match(r"(.*)\[(\d+)\.\.(\d+)\]", value) if m: base, start, end = m.groups() return domain, {f"{base}{i}" for i in range(int(start), int(end) + 1)} return domain, {value} DNSBL = [_expand_dnsbl_entry(e) for e in DNSBL_RAW] DNSBL_CACHE: dict[str, tuple[bool, str | None, float]] = {} DNSBL_TTL = 300 def dnsbl_check(ip: str) -> tuple[bool, str | None]: if not ip or ":" in ip: return False, None now = time.time() if ip in DNSBL_CACHE: result, dnsbl_name, ts = DNSBL_CACHE[ip] if now - ts < DNSBL_TTL: return result, dnsbl_name reversed_ip = ".".join(reversed(ip.split("."))) for dnsbl_domain, valid_answers in DNSBL: query = f"{reversed_ip}.{dnsbl_domain}" try: answers = dns.resolver.resolve(query, "A") for rdata in answers: result_ip = rdata.to_text() if valid_answers is None or result_ip in valid_answers: DNSBL_CACHE[ip] = (True, dnsbl_domain, now) return True, dnsbl_domain except dns.resolver.NXDOMAIN: continue except Exception: continue DNSBL_CACHE[ip] = (False, None, now) return False, None # --------------------------------------------------------------------------- # Resolver, Globale Konstanten # --------------------------------------------------------------------------- # --- Single Source of Truth fuer Journal-Filter-Tokens --- # Diese Tokens werden an drei Stellen benoetigt: # 1. BLOCK_RE / ACCEPT_RE -> finale Regex-Klassifizierung der Message # 2. _PREFILTER_TOKENS -> billiger Substring-Pre-Filter vor json.loads() # 3. _JOURNAL_GREP -> server-seitiger Filter via journalctl --grep # # Pro Eintrag definieren wir: # - prefilter_token: die Substring-Form, die im Logtext als Indikator reicht # - regex_pattern: die genauere Form fuer BLOCK_RE/ACCEPT_RE (z.B. mit ":" # oder Praefix "rejected: ") # Die Trennung ist noetig, weil das Original den Pre-Filter laxer haelt als die # finale Regex (z.B. "Domain not found" als Pre-Filter, aber "rejected: Domain # not found" als Regex). Diese Asymmetrie wird hier 1:1 erhalten. _BLOCK_TOKENS: tuple[tuple[str, str], ...] = ( ("proxy-reject", r"proxy-reject:"), ("policyguard-500", r"policyguard-500"), ("Domain not found", r"rejected: Domain not found"), ) _ACCEPT_TOKENS: tuple[tuple[str, str], ...] = ( ("proxy-accept", r"proxy-accept:"), ) BLOCK_RE = re.compile("|".join(p for _, p in _BLOCK_TOKENS), re.IGNORECASE) ACCEPT_RE = re.compile("|".join(p for _, p in _ACCEPT_TOKENS), re.IGNORECASE) CACHE: dict = {} HUNTER_CACHE: dict[str, bool] = {} _blocklist_cache: set[str] = set() _known_domains_cache: set[str] = set() # Zeitstempel (time.monotonic()) des letzten Blocklist-Refreshs — egal ob der # Refresh aus dem Journal selbst berechnet oder von der Platte uebernommen # wurde. Steuert sowohl den In-Memory-Throttle als auch die Entscheidung, ob # ueberhaupt neu nachgeschaut werden muss. _last_incremental_update: float = 0.0 # Lazy-Load-State _history_loaded: bool = False # Log-Rate-Limit-State _log_window_start: float = 0.0 _log_window_count: int = 0 _log_dropped_count: int = 0 # --- Helfer fuer Pre-Computation --- def base_domain(dom: str) -> str: parts = dom.split(".") if len(parts) < 2: return dom tld = parts[-1] sld = parts[-2] if (tld in {"uk", "au", "nz", "za", "br", "jp"} and sld in {"co", "com", "net", "org", "gov", "ac", "edu"} and len(parts) >= 3): return ".".join(parts[-3:]) return ".".join(parts[-2:]) def normalize_domain(dom: str) -> str: return base_domain(dom.lower().rstrip(".")) # Vorab normalisierte Sets fuer O(1) Lookups WHITELIST_AUTOBLACKLIST_NORMALIZED = {normalize_domain(d) for d in WHITELIST_AUTOBLACKLIST} def domain_matches_list(domain: str, target_set: set[str]) -> bool: """O(1) Check fuer exakte Matches oder Subdomains statt O(n) String-Endswith.""" if domain in target_set: return True parts = domain.split('.') for i in range(1, len(parts)): if '.'.join(parts[i:]) in target_set: return True return False def load_transport_domains() -> set[str]: domains = set() transport_path = Path(TRANSPORTMAP) if not transport_path.is_file(): return domains with open(transport_path, "r", errors="ignore") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue key = line.split()[0].strip("[]").lower().rstrip(".") domains.add(normalize_domain(key)) return domains # --------------------------------------------------------------------------- # Aggregation (gemeinsam von stats() und der Blocklist-Persistenz genutzt, # um Domain-Statistiken einmalig zu berechnen) # --------------------------------------------------------------------------- def _aggregate_domain_stats_from(day_maps: list[dict]) -> dict: """Aggregiert beliebige Tages-Dicts zu Gesamt-Domain-Statistiken (total/blocked/accepted).""" agg = defaultdict(lambda: {"total": 0, "blocked": 0, "accepted": 0}) for day_data in day_maps: for dom, s in day_data.items(): e = agg[dom] e["total"] += s["total"] e["blocked"] += s["blocked"] e["accepted"] += s["accepted"] return {dom: dict(s) for dom, s in agg.items()} # --------------------------------------------------------------------------- # Print Stats (Optimized) # --------------------------------------------------------------------------- def stats(search_term: str = "", min_pct: int = 0, max_pct: int = 100, only_blocked: bool = False, alphabetical: bool = False): # Statistik wird nicht persistiert, sondern bei Bedarf komplett neu # berechnet: Historie aus CACHE und heutige Daten direkt aus dem Journal. _load_history_into_cache() today_stats = _collect_today_stats() transport_excludes = load_transport_domains() all_excludes = WHITELIST_AUTOBLACKLIST_NORMALIZED | transport_excludes agg_raw = _aggregate_domain_stats_from([*CACHE.values(), today_stats]) agg = { dom: s for dom, s in agg_raw.items() if dom not in all_excludes and not dom.endswith(WHITELIST_AUTOBLACKLIST_TLDS) } filtered = [] for dom, s in agg.items(): if s["total"] == 0 or s["blocked"] == 0: continue if search_term and search_term not in dom: continue pct = int(s["blocked"] / s["total"] * 100) if min_pct <= pct <= max_pct: filtered.append((dom, s, pct)) if not filtered: print("No stats available.") sys.stdout.flush() return # Sortieren greift direkt auf den vorberechneten pct-Wert (x[2]) zu if alphabetical: filtered.sort(key=lambda x: x[0]) # x[0] = domain-String else: filtered.sort(key=lambda x: (x[2], x[1]["total"]), reverse=True) print("==== TOTAL DOMAIN STATS (all days + today) ====") for dom, s, pct in filtered: meets_block_criteria = (s["total"] >= MIN_COUNT) and (pct >= BLOCK_QUOTA) if only_blocked and not meets_block_criteria: continue marker = f">>{pct}%<<" if meets_block_criteria else f" {pct}% " print(f"{dom:<40} accept: {s['accepted']:<5} blocked: {s['blocked']:<5} total: {s['total']:<5} {marker:<10}") print("==== END TOTAL DOMAIN STATS ====") sys.stdout.flush() # --------------------------------------------------------------------------- # Logging & Extraction # --------------------------------------------------------------------------- FROM_REGEX = re.compile(r"from=<[^>]*@([^> ,]+)>") def extract_domain(message: str) -> str: m = FROM_REGEX.search(message) if m: return normalize_domain(m.group(1)) return "" def log(msg: str): """Schreibt nach stderr mit Rate-Limit, um Pipe-Blocking zu vermeiden.""" global _log_window_start, _log_window_count, _log_dropped_count now = time.monotonic() if now - _log_window_start >= 1.0: # Neue Sekunde: ggf. Dropped-Counter ausgeben, dann Reset if _log_dropped_count > 0: try: sys.stderr.write(f"spf-policy: [rate-limit] {_log_dropped_count} log lines dropped\n") sys.stderr.flush() except Exception: pass _log_dropped_count = 0 _log_window_start = now _log_window_count = 0 if _log_window_count >= LOG_RATE_LIMIT_PER_SEC: _log_dropped_count += 1 return _log_window_count += 1 try: sys.stderr.write(f"spf-policy: {msg}\n") sys.stderr.flush() except Exception: pass # --------------------------------------------------------------------------- # Cache persistence # --------------------------------------------------------------------------- def _atomic_write_json(path: Path, obj) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}.{random.randrange(1_000_000)}") try: with open(tmp, "w") as f: json.dump(obj, f) tmp.replace(path) except Exception: try: tmp.unlink(missing_ok=True) except Exception: pass raise def save_cache(): today_str = date.today().isoformat() cutoff = (date.today() - timedelta(days=DAYS)).isoformat() pruned = { day: data for day, data in CACHE.items() if day != today_str and day >= cutoff } try: _atomic_write_json(CACHE_FILE, pruned) except Exception as e: log(f"save_cache error: {e}") def load_cache(): global CACHE if not CACHE_FILE.exists(): return try: with open(CACHE_FILE) as f: data = json.load(f) today_str = date.today().isoformat() cutoff = (date.today() - timedelta(days=DAYS)).isoformat() CACHE = { day: day_data for day, day_data in data.items() if day != today_str and day >= cutoff } log(f"load_cache: {len(CACHE)} days loaded from {CACHE_FILE}") except Exception as e: log(f"load_cache error: {e}") # --------------------------------------------------------------------------- # Log reading # # Strategie fuer minimalen RAM/CPU-Verbrauch: # - Ein einziger journalctl-Aufruf fuer alle benoetigten Tage am Stueck, statt # N Aufrufe (spart journalctl-Index-Walk). # - Server-side Filterung mit --grep, sodass irrelevante Zeilen gar nicht erst # serialisiert werden -> drastisch weniger I/O und weniger json.loads(). # - Reduzierte Output-Felder (--output-fields=MESSAGE,__REALTIME_TIMESTAMP). # - Pre-Filter via 'in'-Substring-Check, BEVOR json.loads() aufgerufen wird: # Wenn weder "proxy-reject" noch "proxy-accept" noch "Domain not found" in # der Rohzeile vorkommen, ist json.loads() unnoetig. Substring-Tests sind # ~50x schneller als json.loads(). # - Streaming-Generator + tagweises Flushen: Der RAM-Footprint bleibt auf der # Groessenordnung *eines* Tages, statt aller Tage gleichzeitig. # - Lokale Bindings fuer Hot-Path-Funktionen. # --------------------------------------------------------------------------- # Pre-Filter-Tokens und --grep-Pattern werden zentral aus _BLOCK_TOKENS / # _ACCEPT_TOKENS oben abgeleitet -> keine drei Stellen mehr synchron zu halten. _PREFILTER_TOKENS: tuple[str, ...] = tuple( t for t, _ in (*_BLOCK_TOKENS, *_ACCEPT_TOKENS) ) # --grep-Pattern (regex, ERE) - laeuft serverseitig in journalctl. # Reduziert die transferierte Datenmenge massiv bei stark verrauschten Logs. _JOURNAL_GREP: str = "|".join(_PREFILTER_TOKENS) def _journalctl_cmd(extra_args: list[str]) -> list[str]: """Baut den journalctl-Aufruf. Setzt voraus, dass der User Mitglied der Gruppe systemd-journal ist (siehe Hinweis am Dateianfang). Falls sudo zwingend benoetigt wird, kann man hier ['sudo', '-n', ...] voranstellen.""" return [ "sudo", "journalctl", "--no-pager", "-u", "postfix", "-o", "json", "--output-fields=MESSAGE,__REALTIME_TIMESTAMP", "--grep", _JOURNAL_GREP, ] + extra_args def _iter_journal_events(extra_args: list[str]) -> Iterator[tuple[str, str, bool, bool]]: """Generator ueber (line_date, domain, is_blocked, is_accepted) Tupel. Streamt journalctl zeilenweise und haelt keinen grossen Buffer im RAM. Schluckt Fehler und loggt einmalig. """ # Lokale Bindings fuer Performance im Hot-Path block_search = BLOCK_RE.search accept_search = ACCEPT_RE.search json_loads = json.loads fromtimestamp = datetime.fromtimestamp _extract_domain = extract_domain prefilter_tokens = _PREFILTER_TOKENS proc = None try: proc = subprocess.Popen( _journalctl_cmd(extra_args), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1 << 16, # 64 KiB Pipe-Buffer ) for raw in proc.stdout: # Schneller Pre-Filter VOR json.loads (Substring ist ~50x schneller). # journalctl --grep filtert serverseitig, aber falls --grep mal # entfernt wird, bleibt diese Zeile als Sicherheitsnetz. if not any(tok in raw for tok in prefilter_tokens): continue line = raw.strip() if not line: continue try: entry = json_loads(line) except json.JSONDecodeError: continue message = entry.get("MESSAGE", "") if not isinstance(message, str): continue is_blocked = bool(block_search(message)) is_accepted = bool(accept_search(message)) and not is_blocked if not is_blocked and not is_accepted: continue dom = _extract_domain(message) if not dom: continue ts = entry.get("__REALTIME_TIMESTAMP") if not ts: continue try: # Lokalzeit-bewusst: Tagesgrenzen muessen zu date.today() und # den --since-Parametern passen. dt = fromtimestamp(int(ts) / 1_000_000, tz=LOCAL_TZ) line_date = dt.date().isoformat() except (ValueError, TypeError, OSError): continue yield line_date, dom, is_blocked, is_accepted proc.stdout.close() proc.wait(timeout=120) except subprocess.TimeoutExpired: if proc is not None: proc.kill() log("_iter_journal_events: journalctl timed out") except Exception as e: log(f"_iter_journal_events error: {e}") finally: if proc is not None and proc.poll() is None: try: proc.kill() except Exception: pass def _bump(day_dict: dict, dom: str, is_blocked: bool, is_accepted: bool): """Inkrementiert total/blocked/accepted fuer eine Domain in einem Tagesdict.""" e = day_dict.get(dom) if e is None: e = {"total": 0, "blocked": 0, "accepted": 0} day_dict[dom] = e e["total"] += 1 if is_blocked: e["blocked"] += 1 if is_accepted: e["accepted"] += 1 # --------------------------------------------------------------------------- # Cache initialization # --------------------------------------------------------------------------- def _load_history_into_cache(): """RAM-schonender Erst-Import: - Ein einziger journalctl-Stream fuer alle fehlenden Tage gemeinsam. - Tagesweises Flushen in CACHE: sobald der Datums-Wechsel erkannt wird, ist der vorherige Tag fertig und kann persistiert werden — der RAM haelt immer nur einen aktiven Tag. - Periodisches save_cache() nach N fertigen Tagen, damit ein Crash waehrend Erst-Import nicht alle Fortschritte verliert. """ global CACHE, _history_loaded if _history_loaded: return load_cache() today = date.today() today_str = today.isoformat() needed_days = { (today - timedelta(days=i)).isoformat() for i in range(1, DAYS + 1) } - set(CACHE.keys()) if needed_days: oldest = min(needed_days) log(f"initialize_cache: need {len(needed_days)} days, streaming from {oldest}") # Streaming-Strategie: aktuellen Tag im RAM halten, beim Wechsel flushen. current_day: Optional[str] = None current_day_data: dict = {} days_done_since_flush = 0 # Tag-Set fuer O(1) Lookup beim Filtern (wir streamen ggf. auch # Tage, die schon im CACHE sind — die werden hier verworfen). needed_days_set = needed_days # Wir nehmen einen grossen Stream "since oldest" — journalctl wird die # Tage in chronologischer Reihenfolge liefern, daher reicht ein # einfacher Tageswechsel-Detektor. for line_date, dom, is_blocked, is_accepted in _iter_journal_events(["--since", oldest]): # Tage ausserhalb des needed-Fensters ueberspringen if line_date == today_str: # Heutige Daten kommen separat via Today-State rein continue if line_date not in needed_days_set: continue if current_day is None: current_day = line_date current_day_data = {} elif line_date != current_day: # Tag fertig — flushen CACHE[current_day] = current_day_data days_done_since_flush += 1 current_day_data = {} current_day = line_date if days_done_since_flush >= INITIAL_IMPORT_FLUSH_EVERY_DAYS: save_cache() days_done_since_flush = 0 _bump(current_day_data, dom, is_blocked, is_accepted) # Letzten Tag auch flushen if current_day is not None: CACHE[current_day] = current_day_data current_day_data = {} # RAM freigeben # Tage ohne Events trotzdem als leer markieren, sonst werden sie beim # naechsten Start wieder importiert for d in needed_days_set: if d not in CACHE: CACHE[d] = {} save_cache() else: log("initialize_cache: all historical days present in cache file") _history_loaded = True def _collect_today_stats() -> dict: """Berechnet die heutigen Domain-Statistiken komplett neu aus dem Journal und persistiert sie nicht global.""" today_str = date.today().isoformat() log("stats/blocklist: rebuilding today's stats from journal") today_data: dict = {} for line_date, dom, is_blocked, is_accepted in _iter_journal_events(["--since", today_str]): if line_date != today_str: continue _bump(today_data, dom, is_blocked, is_accepted) return today_data def compute_today_stats() -> dict: """Liefert die heutigen Domain-Statistiken, jedes Mal frisch aus dem Journal neu berechnet. Bewusst KEIN inkrementelles Aufaddieren mehr: fruehere Versionen haben heutige Events per Journal-Cursor additiv in einen persistenten Today-State gezaehlt. Bei mehreren parallelen, kurzlebigen Postfix-Workern (jeder mit eigener, nicht geteilter Cursor-Datei) wurde derselbe Journal-Bereich so mehrfach gezaehlt — die Blocked/Total-Absolutwerte wuchsen als ganzzahliges Vielfaches (das beobachtete "416"-Muster). Ein voller Rescan von "heute" ist guenstig (journalctl --grep + --since heute) und garantiert Idempotenz: egal wie oft und aus wie vielen Prozessen aufgerufen, das Ergebnis ist identisch und nie doppelt gezaehlt.""" return _collect_today_stats() def update_blocklist_incremental(): """Stellt sicher, dass die historische Historie einmalig geladen ist. Der Name bleibt aus Kompatibilitaetsgruenden erhalten; es wird aber nichts mehr inkrementell in einen persistenten State gezaehlt. Der eigentliche Today-Anteil wird in current_blocklist() bei jedem Refresh frisch berechnet. """ _load_history_into_cache() def initialize_cache(): global _last_incremental_update # Bewusst auf 0.0: Beim ersten Aufruf von current_blocklist() soll zuerst # die persistierte BLOCKLIST_FILE geprueft werden. Nur wenn diese nicht # frisch genug ist, wird spaeter lazy neu aggregiert. _last_incremental_update = 0.0 log("initialize_cache: lazy startup without loading historical cache") # --------------------------------------------------------------------------- # Candidate detection (Optimized) # --------------------------------------------------------------------------- def find_candidates(cache: dict, stats: dict) -> list[tuple[str, int]]: total = defaultdict(int) blocked = defaultdict(int) for day_data in [*cache.values(), stats]: for dom, s in day_data.items(): total[dom] += s["total"] blocked[dom] += s["blocked"] candidates = [ (dom, tot) for dom, tot in total.items() if tot >= MIN_COUNT and (blocked[dom] / tot) >= BLOCK_QUOTA_FLOAT ] return sorted(candidates, key=lambda x: x[1], reverse=True) # --------------------------------------------------------------------------- # Persistenz der Autoblocklist auf die Platte # --------------------------------------------------------------------------- # Ziel: Mehrere parallel laufende Policy-Worker-Prozesse (Postfix startet # davon typischerweise mehrere) sollen sich nicht gegenseitig mit teuren # journalctl-Scans blockieren. Deshalb schreibt der Prozess, der eine neue # Blocklist berechnet, das Ergebnis nach BLOCKLIST_FILE. Persistiert wird # bewusst nur die Blocklist selbst; Statistik wird separat bei Bedarf aus # Journal/CACHE aufgebaut. Andere Prozesse pruefen zuerst, ob diese Datei noch # "frisch" (juenger als INCREMENTAL_UPDATE_INTERVAL Sekunden) ist, bevor sie # selbst neu rechnen. def _write_blocklist_payload(payload: dict) -> None: try: _atomic_write_json(BLOCKLIST_FILE, payload) log(f"current_blocklist: Blocklist-Payload nach {BLOCKLIST_FILE} geschrieben") except Exception as e: log(f"_write_blocklist_payload error: {e}") def _write_policy_lists(blocklist: set[str], known_domains: set[str]) -> None: """Schreibt Blocklist + Known-Domain-Liste + Freshness-Metadaten nach BLOCKLIST_FILE (ein Refresh, eine Datei, ein journalctl-Scan).""" payload = { "generated_at": time.time(), "generated_at_iso": datetime.now(timezone.utc).isoformat(), "blocklist": sorted(blocklist), "known_domains": sorted(known_domains), } _write_blocklist_payload(payload) log(f"policy-lists: {len(blocklist)} blocklist / {len(known_domains)} known Domains geschrieben") def _refresh_policy_lists_timestamp(existing_payload: dict, blocklist: set[str], known_domains: set[str]) -> None: """Aktualisiert bei unveraenderten Listen nur die Freshness-Felder.""" payload = dict(existing_payload) payload["generated_at"] = time.time() payload["generated_at_iso"] = datetime.now(timezone.utc).isoformat() if not isinstance(payload.get("blocklist"), list): payload["blocklist"] = sorted(blocklist) if not isinstance(payload.get("known_domains"), list): payload["known_domains"] = sorted(known_domains) _write_blocklist_payload(payload) def _load_blocklist_payload() -> Optional[dict]: """Laedt die persistierte Blocklist-Datei komplett. Liefert None, wenn die Datei fehlt oder unlesbar ist.""" try: with open(BLOCKLIST_FILE) as f: data = json.load(f) if not isinstance(data, dict): return None return data except OSError: return None except Exception as e: log(f"_load_blocklist_payload error: {e}") return None def _extract_domain_set(payload: dict, key: str) -> set[str]: raw = payload.get(key, []) if not isinstance(raw, list): return set() return {item for item in raw if isinstance(item, str)} def _load_policy_lists_if_fresh() -> Optional[tuple[set[str], set[str]]]: """Liest (blocklist, known_domains) von der Platte, falls BLOCKLIST_FILE juenger als INCREMENTAL_UPDATE_INTERVAL Sekunden ist (z.B. weil ein anderer Policy-Worker-Prozess sie kuerzlich neu berechnet hat). Sonst None — dann muss der Aufrufer selbst neu aus dem Journal einlesen.""" data = _load_blocklist_payload() if data is None: return None generated_at = data.get("generated_at") if not isinstance(generated_at, (int, float)): try: generated_at = BLOCKLIST_FILE.stat().st_mtime except OSError: return None if (time.time() - float(generated_at)) >= INCREMENTAL_UPDATE_INTERVAL: return None if not isinstance(data.get("blocklist"), list): return None return _extract_domain_set(data, "blocklist"), _extract_domain_set(data, "known_domains") # --------------------------------------------------------------------------- # Autoblocklist # --------------------------------------------------------------------------- def _refresh_policy_lists() -> None: """Aktualisiert _blocklist_cache und _known_domains_cache (gedrosselt). Ablauf: 1. Solange INCREMENTAL_UPDATE_INTERVAL seit dem letzten eigenen Refresh noch nicht abgelaufen ist: nichts tun — RAM-Caches reichen. 2. Ist das Intervall abgelaufen: zuerst BLOCKLIST_FILE pruefen. Ist sie juenger als das Intervall (z.B. weil ein anderer Worker-Prozess sie bereits aktualisiert hat), werden beide Listen uebernommen — ohne erneutes Schreiben. 3. Ist auch die Datei zu alt (oder existiert nicht): selbst neu aus dem Journal berechnen. Ein Scan liefert beide Listen: - blocklist: Domains ueber der Block-Quote (find_candidates) - known_domains: Domains mit mindestens einer akzeptierten Mail (accepted > 0) im Cache-Fenster — geblockte Mails machen eine Domain bewusst NICHT bekannt (keine Selbstfuetterung) Bei unveraendertem Inhalt werden nur die Freshness-Felder erneuert. """ global _blocklist_cache, _known_domains_cache, _last_incremental_update now = time.monotonic() # 1) Eigenes Intervall noch nicht abgelaufen -> RAM-Cache reicht. if _last_incremental_update and (now - _last_incremental_update) < INCREMENTAL_UPDATE_INTERVAL: return # 2) Intervall abgelaufen -> zuerst pruefen, ob von Platte uebernehmbar. disk_lists = _load_policy_lists_if_fresh() if disk_lists is not None: _blocklist_cache, _known_domains_cache = disk_lists _last_incremental_update = now log("policy-lists: frische Listen von Platte uebernommen") return # 3) Auch die Datei ist zu alt / fehlt -> selbst neu einlesen & berechnen. update_blocklist_incremental() today_stats = compute_today_stats() transport_excludes = load_transport_domains() all_excludes = WHITELIST_AUTOBLACKLIST_NORMALIZED | transport_excludes candidates = find_candidates(CACHE, today_stats) new_blocklist = { dom for dom, _ in candidates if not dom.endswith(WHITELIST_AUTOBLACKLIST_TLDS) and dom not in all_excludes } new_known = { dom for day_data in [*CACHE.values(), today_stats] for dom, s in day_data.items() if s.get("accepted", 0) > 0 } persisted_payload = _load_blocklist_payload() persisted_blocklist = None persisted_known = None if isinstance(persisted_payload, dict): if isinstance(persisted_payload.get("blocklist"), list): persisted_blocklist = _extract_domain_set(persisted_payload, "blocklist") if isinstance(persisted_payload.get("known_domains"), list): persisted_known = _extract_domain_set(persisted_payload, "known_domains") is_initial_creation = persisted_blocklist is None or persisted_known is None has_changed = (persisted_blocklist != new_blocklist) or (persisted_known != new_known) _blocklist_cache = new_blocklist _known_domains_cache = new_known _last_incremental_update = now if has_changed: _write_policy_lists(new_blocklist, new_known) if is_initial_creation: log("policy-lists: Datei initial erstellt") else: log("policy-lists: Datei aktualisiert (Inhalt geaendert)") else: _refresh_policy_lists_timestamp(persisted_payload, new_blocklist, new_known) log("policy-lists: Datei zeitlich aktualisiert (Inhalt unveraendert)") def current_blocklist() -> set[str]: """Liefert die aktuell gueltige Autoblocklist (Refresh intern gedrosselt).""" _refresh_policy_lists() return _blocklist_cache def current_known_domains() -> set[str]: """Liefert die Known-Domain-Liste (nur Accept-Events, gedrosselt).""" _refresh_policy_lists() return _known_domains_cache # --------------------------------------------------------------------------- # Hunter IO (mit RAM Cache) # --------------------------------------------------------------------------- def check_for_hunter_io(url: str) -> bool: global HUNTER_CACHE if url in HUNTER_CACHE: return HUNTER_CACHE[url] headers = {"User-Agent": "Mozilla/5.0 (compatible; Python-Checker/1.0)"} abusestring = "abuse@hunter.io" if not re.match(r'^[a-z0-9][a-z0-9\-\.]+\.[a-z]{2,}$', url): return False result = False try: response = requests.get(f'https://{url}', headers=headers, timeout=3, allow_redirects=False) response.raise_for_status() result = abusestring in response.text except Exception: pass if len(HUNTER_CACHE) < GENERIC_CACHE_SIZE: HUNTER_CACHE[url] = result return result # --------------------------------------------------------------------------- # SPF & TXT Check (mit RAM Cache) # --------------------------------------------------------------------------- def sender_matches_regex(sender: str) -> tuple[bool, str | None]: for regex in BLOCKLIST_REGEX_COMPILED: if regex.match(sender): return True, regex.pattern return False, None # --------------------------------------------------------------------------- # Greylisting (domain-basiert, letzter Check vor dunno) # --------------------------------------------------------------------------- # Bekannte Absenderdomains kommen aus der Known-Domain-Liste in # BLOCKLIST_FILE (Domains mit Accept-Events der letzten DAYS Tage, Refresh # gedrosselt via INCREMENTAL_UPDATE_INTERVAL) — eine eigene Liste # "gesehener" Domains wird nicht gefuehrt. # # SEEN_FILE ist eine reine Retry-Pending-Liste. Aufbau: # {"": [{"sender": ..., "receiver": ..., "time": }, ...]} # # Ablauf pro Mail (Domain-Key via normalize_domain): # 1) Domain in known_domains -> dunno # 2) Domain unbekannt -> Eintrag in SEEN_FILE anlegen -> defer # 3) Domain in SEEN_FILE: # a) passender Eintrag (sender+receiver, nicht aelter als # GREYLIST_EXPIRE_MAIL_S) -> akzeptieren; Domain-Eintrag wird komplett # geloescht — die zugestellte Mail erzeugt ein proxy-accept im # Journal, die Domain wandert ueber den Cache in known_domains # b) sonst -> Eintrag hinzufuegen -> defer # Abgelaufene Eintraege und leere Domains werden beim Schreiben gepruned. # Concurrency: exklusives flock auf SEEN_LOCK_FILE um den kompletten # read-modify-write-Zyklus; geschrieben wird atomar via _atomic_write_json. # Datei-Fehler -> fail-open (akzeptieren): ein Greylisting-Problem darf # niemals legitime Mails blockieren. def _seen_load() -> dict: """Laedt die Retry-Pending-Liste. Fehler/Torn-File -> leeres Dict.""" try: with open(SEEN_FILE) as f: data = json.load(f) return data if isinstance(data, dict) else {} except OSError: return {} except Exception as e: log(f"seen load error: {e}") return {} def _seen_fresh(entries: list, now: float) -> list: """Nur gueltige, nicht abgelaufene Eintraege behalten.""" return [ e for e in entries if isinstance(e, dict) and isinstance(e.get("time"), (int, float)) and now - e["time"] <= GREYLIST_EXPIRE_MAIL_S ] def _seen_prune(data: dict, now: float) -> None: """Verwirft abgelaufene Eintraege und loescht leere Domains (in-place).""" for dom in list(data.keys()): entries = data.get(dom) if not isinstance(entries, list): del data[dom] continue fresh = _seen_fresh(entries, now) if fresh: data[dom] = fresh else: del data[dom] def greylist_check(domain: str, sender: str, recipient: str) -> bool: """Greylisting-Entscheidung fuer eine sonst akzeptierte Mail. True -> akzeptieren (dunno) False -> defern (temporaerer Fehler, ein legitimer MTA retryt) """ # Schritt 1: bekannte Domain? (Refresh intern gedrosselt — deckt damit # auch ab, dass die Domain erst kuerzlich per Accept-Event in den # Cache kam) if domain in current_known_domains(): return True # Schritt 2/3: Retry-Pending-Liste now = time.time() now_i = int(now) sender = sender.lower() try: SEEN_FILE.parent.mkdir(parents=True, exist_ok=True) lock = open(SEEN_LOCK_FILE, "a") except OSError as e: log(f"greylist: Lockfile nicht oeffenbar ({e}) - fail-open") return True with lock: fcntl.flock(lock, fcntl.LOCK_EX) data = _seen_load() entries = data.get(domain) if not isinstance(entries, list): entries = None accept = False if entries is None: # Schritt 2: Domain unbekannt -> vormerken + defer data[domain] = [{"sender": sender, "receiver": recipient, "time": now_i}] log(f"greylist: neue Domain {domain} -> defer") else: entries = _seen_fresh(entries, now) match = next( (e for e in entries if e.get("sender") == sender and e.get("receiver") == recipient), None, ) if match is not None: # 3a: Retry erkannt -> akzeptieren, Eintrag komplett loeschen del data[domain] accept = True log(f"greylist: Retry erkannt, {domain} zugelassen") else: # 3b: andere Kombination / abgelaufen -> vormerken + defer entries.append({"sender": sender, "receiver": recipient, "time": now_i}) data[domain] = entries log(f"greylist: {domain} weiterhin unbekannt, Mail vorgemerkt -> defer") _seen_prune(data, now) try: _atomic_write_json(SEEN_FILE, data) except Exception as e: log(f"seen save error: {e}") return accept def greylist_stats(arg: str = "") -> None: """Interaktiver Befehl. greylist -> ausstehende Retries (SEEN_FILE) greylist known -> prueft, ob in known_domains steht """ arg = arg.strip() if arg.lower().startswith("known"): parts = arg.split(None, 1) if len(parts) < 2 or not parts[1].strip(): print("usage: greylist [known ]") sys.stdout.flush() return dom = normalize_domain(parts[1].strip().lower()) known = dom in current_known_domains() print(f"{dom}: {'known' if known else 'unknown'}") sys.stdout.flush() return if arg: print("usage: greylist [known ]") sys.stdout.flush() return data = _seen_load() rows = [] for dom, entries in data.items(): if not isinstance(entries, list): continue for e in entries: if isinstance(e, dict) and isinstance(e.get("time"), (int, float)): rows.append((dom, str(e.get("sender", "")), str(e.get("receiver", "")), e["time"])) if not rows: print("No pending greylist entries.") sys.stdout.flush() return rows.sort(key=lambda r: (r[0], r[3])) print("==== GREYLIST PENDING RETRIES ====") print(f"{'DOMAIN':<35} {'SENDER':<35} {'RECEIVER':<25} TIME") for dom, sender, receiver, ts in rows: t = datetime.fromtimestamp(ts, tz=LOCAL_TZ).strftime("%Y-%m-%d %H:%M:%S") print(f"{dom:<35} {sender:<35} {receiver:<25} {t}") print(f"==== END GREYLIST ({len(rows)} pending, {len(data)} domains) ====") sys.stdout.flush() # --------------------------------------------------------------------------- # Request handling (Optimized O(1) lookups) # --------------------------------------------------------------------------- def parse_search_args(arg: str) -> tuple[str, int, int, bool, bool]: parts = arg.strip().split() search_term = "" numbers: list[int] = [] only_blocked = False alphabetical = False for p in parts: p_lower = p.lower() if p_lower in ("-b", "blocked"): only_blocked = True elif p_lower == "-a": alphabetical = True elif p.isdigit() and 0 <= int(p) <= 100: numbers.append(int(p)) else: search_term = p_lower if len(numbers) == 0: min_pct, max_pct = 0, 100 elif len(numbers) == 1: min_pct, max_pct = numbers[0], 100 else: min_pct, max_pct = min(numbers[0], numbers[1]), max(numbers[0], numbers[1]) return search_term, min_pct, max_pct, only_blocked, alphabetical def handle_request(attrs: dict) -> str: domain = "" try: global date_now date_now = datetime.now() sender = attrs.get("sender", "") recipient = attrs.get("recipient", "").lower() client_name = attrs.get("client_name", "").lower() helo_name = attrs.get("helo_name", "").lower() client_address = attrs.get("client_address", "") # Domain Split if "@" in sender: domain = sender.split("@", 1)[1].lower() if not domain: return "policyguard-300 empty domain" # HELO Whitelist if domain_matches_list(helo_name, WHITELIST_HELO): log("whitelist sender by helo whitelist") return "dunno" # Client Whitelist if domain_matches_list(client_name, WHITELIST_CLIENT): log("whitelist sender by client whitelist") return "dunno" # Sender Whitelist if domain and domain_matches_list(domain, WHITELIST_SENDER): log("whitelist sender by sender whitelist") return "dunno" # Block by Clientname if domain_matches_list(client_name, BLOCKLIST_CLIENT): log(f"blocked client hostname: {client_name}") return "policyguard-400 blacklist-client Your mailserver is blacklisted" # DNSBL Check listed, dnsbl_name = dnsbl_check(client_address) if listed: log(f"blocked client {client_address} via DNSBL {dnsbl_name}") dnsbl_short = ".".join(dnsbl_name.split(".")[-2:]) return f"policyguard-300 blocked based on DNSBL {dnsbl_short}" # Recipient reject if "@" in recipient: rcpt_domain_only = recipient.split("@", 1)[1] if ("@" + rcpt_domain_only) in BLOCKLIST_RECIPIENT or domain_matches_list(rcpt_domain_only, BLOCKLIST_RECIPIENT): log(f"blocked recipient by domain: {recipient}") return "policyguard-400 The email account that you tried to reach does not exist" # Whitelist bouncemail if "@" not in sender: return "dunno" # Block Sender if domain and domain_matches_list(domain, BLOCKLIST_SENDER): log(f"blocked sender domain via blocklist {domain}") return "policyguard-400 blacklist-sender Your domain is blacklisted" # Autoblocklist bl_cache = current_blocklist() if domain and domain_matches_list(domain, bl_cache): log(f"blocked sender domain via autoblocklist {domain}") return "policyguard-500 autoblacklist Your domain is autoblacklisted" # Block via SPF spf_result, spf_explanation = ( spf.check2(i=client_address, s=sender, h=helo_name) if client_address else ("none", "no client address")) if spf_result == 'fail': log(f"blocked sender {sender} due to SPF Fail ({spf_explanation})") return "policyguard-400 spf-hardfail" # Max Recipients if MAX_RECIPIENTS_PER_MESSAGE > 0: try: recipient_count = int(attrs.get("recipient_count", "1")) if recipient_count > MAX_RECIPIENTS_PER_MESSAGE: log(f"blocked: too many recipients {recipient_count}") return f"policyguard-400 too-many-recipients Maximum {MAX_RECIPIENTS_PER_MESSAGE}" except ValueError: pass # Block Untrusted Client Regex (Google Groups, Fake Facebook Spam) and SPF if domain_matches_list(helo_name, UNTRUSTED_HELO): # Block Untrusted via Regex (Google Groups, Fake Facebook Spam) matched, pattern = sender_matches_regex(sender) if matched: log(f"blocked sender by regex ({pattern}): {sender}") return "policyguard-400 untrusted-regex Regex Blocklist via Untrusted HELO" if spf_result == 'permerror': log(f"blocked sender {sender} due to SPF PermError ({spf_explanation})") return "policyguard-400 untrusted-spf-permerror SPF-Permerror via Untrusted HELO" if spf_result == 'softfail': log(f"blocked sender {sender} due to SPF SoftFail ({spf_explanation})") return f"policyguard-400 untrusted-spf-softfail SPF-Softfail via Untrusted HELO" # Block hunter.io (Cached) if "check" in sender and check_for_hunter_io(domain): log(f"blocked sender domain via hunter.io {domain}") return "policyguard-400 hunter" # Unknown client if client_name == "unknown": log(f"deferred unknown client: {client_address}") return "defer 450 4.7.1 policyguard-500 Client host rejected: temporary defer for unknown hostname" # --- Greylisting: letzter Check vor dunno --------------------------------- # Token policyguard-600 steht bewusst in KEINER Block/Accept-Tokenliste, # damit Greylist-Defers nicht in Cache/Statistik einlaufen. if (GREYLISTING_ENABLED and domain and not greylist_check(normalize_domain(domain), sender, recipient)): return "defer 450 4.7.1 policyguard-600 greylisted, please retry later" except Exception as e: log(f"error checking {domain or ''}: {e}") return "dunno" def main(): initialize_cache() while True: attrs = {} while True: line = sys.stdin.readline() if line == "": return line = line.strip() if not line: break if "=" in line: k, v = line.split("=", 1) attrs[k] = v # Interaktive Befehle direkt waehrend des Einlesens verarbeiten (Original-Verhalten) if line.startswith("greylist"): parts = line.split(None, 1) greylist_stats(parts[1] if len(parts) > 1 else "") if line.startswith("blocklist"): # Alias: blocklist [term] => stats -b -a [term] parts = line.split(None, 1) suffix = ("-b -a " + parts[1]) if len(parts) > 1 else "-b -a" search_term, min_pct, max_pct, only_blocked, alphabetical = parse_search_args(suffix) stats(search_term, min_pct, max_pct, only_blocked, alphabetical) if line.startswith("stats"): parts = line.split(None, 1) search_term, min_pct, max_pct, only_blocked, alphabetical = ( parse_search_args(parts[1]) if len(parts) > 1 else ("", 0, 100, False, False) ) stats(search_term, min_pct, max_pct, only_blocked, alphabetical) if not attrs: continue action = handle_request(attrs) # Stochastisch 5.1.1 zurueckgeben (mailbox not found), um Spam dauerhaft loszuwerden if action == "dunno" or action.startswith("defer"): DSN = "" # Montag bis Samstag zwischen 6:00 und 22:00 Uhr niemals 5.1.1 zurueckgeben elif date_now.weekday() <= 5 and 6 * 60 <= date_now.hour * 60 + date_now.minute < 22 * 60: DSN="550 5.7.1 " # Ansonsten mit x % Wahrscheinlichkeit einen 5.1.1 elif random.random() < NON_EXISTING_MAILBOX_FLOAT: DSN="550 5.1.1 " else: DSN="550 5.7.1 " print(f"action={DSN}{action}\n") sys.stdout.flush() if __name__ == "__main__": main()