mailsystem/files/srs/srs-alias-decision.py

#!/usr/bin/python3

import subprocess
from collections import deque
import sys

ALIAS_MAPS = [
    "pgsql:/etc/postfix/pgsql_virtual_alias_maps.cf",
    "pgsql:/etc/postfix/pgsql_virtual_alias_domain_maps.cf",
    "pgsql:/etc/postfix/pgsql_virtual_alias_address_rewriting.cf",
    "pgsql:/etc/postfix/pgsql_virtual_alias_domain_catchall_maps.cf",
    "btree:/etc/postfix/virtual_alias_maps",
]

MAILBOX_MAPS = [
    "pgsql:/etc/postfix/pgsql_virtual_mailbox_maps.cf",
    "pgsql:/etc/postfix/pgsql_virtual_alias_domain_mailbox_maps.cf",
]

DOMAIN_MAPS = [
    "pgsql:/etc/postfix/pgsql_virtual_domains_maps.cf",
    "pgsql:/etc/postfix/pgsql_relay-domain-maps.cf",
    "btree:/etc/postfix/virtual_alias_domains",
]

TRANSPORT_MAPS = [
    "pgsql:/etc/postfix/pgsql_transport_maps.cf",
    "btree:/etc/postfix/transport",
    "btree:/etc/postfix/relay_domains",
]


def lookup(key, table):
    result = subprocess.run(
        ["/usr/sbin/postmap", "-q", key, table],
        capture_output=True,
        text=True,
        timeout=10,
    )

    if result.returncode == 0:
        return result.stdout.strip() or None

    if result.returncode == 1 and not result.stderr.strip():
        return None

    raise RuntimeError(
        f"{table}: {result.stderr.strip() or 'Lookup fehlgeschlagen'}"
    )


def lookup_any(key, tables):
    for table in tables:
        value = lookup(key, table)
        if value is not None:
            return value
    return None


def postconf(name):
    result = subprocess.run(
        ["/usr/sbin/postconf", "-c", "/etc/postfix", "-x", "-h", name],
        capture_output=True,
        text=True,
        timeout=10,
    )

    if result.returncode:
        raise RuntimeError(result.stderr.strip())

    return result.stdout.strip()


def transport_keys(address):
    localpart, domain = address.rsplit("@", 1)

    keys = [address, f"{localpart}@"]

    parts = domain.split(".")

    for i in range(len(parts)):
        suffix = ".".join(parts[i:])
        keys.extend([suffix, "." + suffix])

    return list(dict.fromkeys(keys))


def classify_destination(address):
    domain = address.rsplit("@", 1)[1]

    if lookup_any(address, MAILBOX_MAPS):
        return "LOCAL"

    if domain in postconf("mydestination").lower().split():
        return "LOCAL"

    if lookup_any(domain, DOMAIN_MAPS):
        return "UNKNOWN"

    for key in transport_keys(address):
        transport = lookup_any(key, TRANSPORT_MAPS)

        if transport:
            transport_type = transport.split(":", 1)[0].lower()

            if transport_type in ("smtp", "srs-forward"):
                return "EXTERNAL"

            return "UNKNOWN"

    default_transport = postconf("default_transport").strip().lower()
    if (
        default_transport in ("smtp", "srs-forward")
        or default_transport.startswith(("smtp:", "srs-forward:"))
    ):
        return "EXTERNAL"

    return "UNKNOWN"


MAX_ALIAS_DEPTH = 10
MAX_ALIAS_NODES = 200


def alias_reaches(original, recipient):
    """Find alias paths with BFS and detect cycles in the explored graph."""
    pending = deque([(original, 0)])
    visited = {original}
    edges = {}
    nodes = 0
    depth_exceeded = False

    while pending:
        current, depth = pending.popleft()
        nodes += 1

        if nodes > MAX_ALIAS_NODES:
            return None, "Zu viele Alias-Knoten"

        aliases = lookup_any(current, ALIAS_MAPS)
        if not aliases:
            edges[current] = []
            continue

        targets = [
            x.strip().lower()
            for x in aliases.split(",")
            if (
                x.strip()
                and x.strip().count("@") == 1
                and x.strip().lower() != current
            )
        ]

        edges[current] = targets

        for target in targets:
            if target == recipient:
                if depth + 1 <= MAX_ALIAS_DEPTH:
                    return True, "Alias-Kette erkannt"
                depth_exceeded = True
                continue

            if target in visited:
                continue

            if depth + 1 >= MAX_ALIAS_DEPTH:
                depth_exceeded = True
                continue

            visited.add(target)
            pending.append((target, depth + 1))

    if depth_exceeded:
        return None, "Maximale Alias-Tiefe erreicht"

    # Detect cycles in the explored directed graph.
    # 0 = unseen, 1 = active, 2 = completed
    state = {}

    def has_cycle(node):
        state[node] = 1

        for target in edges.get(node, []):
            if target not in edges:
                continue

            status = state.get(target, 0)

            if status == 1:
                return True

            if status == 0 and has_cycle(target):
                return True

        state[node] = 2
        return False

    for node in edges:
        if state.get(node, 0) == 0 and has_cycle(node):
            return None, "Alias-Schleife erkannt"

    return False, "Empfaenger nicht in direkter Alias-Kette"


def decide(sender, recipient, original):
    sender = sender.strip().lower()
    recipient = recipient.strip().lower()
    original = original.strip().lower()

    if not sender or sender == "<>":
        return "NO_SRS", "Null-Absender"

    if sender.count("@") != 1 or recipient.count("@") != 1:
        return "UNKNOWN", "Ungueltige Empfaengeradresse"

    if original.count("@") == 0:
        return "NO_SRS", "Lokaler Alias"

    if original.count("@") != 1:
        return "UNKNOWN", "Ungueltige Empfaengeradresse"

    if recipient == original:
        return "NO_SRS", "Keine Empfaengerumschreibung"

    reaches, reason = alias_reaches(original, recipient)
    if reaches is not True:
        return "UNKNOWN", reason

    destination = classify_destination(recipient)
    if destination == "LOCAL":
        return "NO_SRS", "Lokales Postfach"
    if destination == "EXTERNAL":
        return "SRS", "PostfixAdmin-Alias-Kette mit SMTP-Zustellung"
    return "UNKNOWN", "Zustellweg nicht eindeutig"


def main():
    if len(sys.argv) != 4:
        print(
            "Aufruf: srs-alias-decision.py "
            "<sender> <recipient> <original_recipient>",
            file=sys.stderr,
        )
        return 2

    try:
        decision, reason = decide(*sys.argv[1:])
    except (RuntimeError, subprocess.TimeoutExpired) as exc:
        print(f"UNKNOWN: Lookup-Fehler: {exc}")
        return 1

    print(f"{decision}: {reason}")
    return 1 if decision == "UNKNOWN" else 0


if __name__ == "__main__":
    sys.exit(main())