tests: cover trusted SMTP proxy metadata

2026-07-28 13:21:10 +02:00
parent bc675e75ba
commit 74f044a24e
1 changed files with 493 additions and 0 deletions
@@ -0,0 +1,493 @@
+#!/usr/bin/env python3
+# This program is free software, licensed under the terms of the GNU GPL.
+# See the Bongo COPYING file for full details.
+# Copyright (c) 2026 Bongo Project contributors
+
+"""Verify trusted outbound XCLIENT and XFORWARD metadata negotiation."""
+
+from __future__ import annotations
+
+import copy
+from dataclasses import dataclass, field
+import importlib.util
+import json
+import os
+from pathlib import Path
+import smtplib
+import socketserver
+import sys
+import threading
+import time
+
+
+TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "45"))
+FIXTURE_HOST = os.environ.get("BONGO_TEST_SMTP22_HOST", "127.0.0.1")
+FIXTURE_PORT = int(os.environ.get("BONGO_TEST_SMTP22_PORT", "25"))
+TOKEN = f"smtp22-{os.getpid()}-{time.time_ns()}"
+
+
+class SMTP22Error(RuntimeError):
+    """Raised when outbound connection metadata violates its contract."""
+
+
+def load_helpers():
+    path = Path(__file__).with_name(
+        "smtp-outbound-opportunistic-tls-check.py"
+    )
+    specification = importlib.util.spec_from_file_location(
+        "bongo_smtp22_helpers", path
+    )
+    if specification is None or specification.loader is None:
+        raise SMTP22Error(f"cannot load SMTP fixture helpers from {path}")
+    module = importlib.util.module_from_spec(specification)
+    sys.modules[specification.name] = module
+    specification.loader.exec_module(module)
+    return module
+
+
+SMTP07 = load_helpers()
+
+
+def privileged(arguments: list[str]) -> list[str]:
+    if os.geteuid() == 0:
+        return arguments
+    return ["/usr/bin/sudo", "-n", *arguments]
+
+
+@dataclass
+class Delivery:
+    marker: str
+    commands: list[str]
+    message: bytes
+
+
+@dataclass
+class Capture:
+    lock: threading.Lock = field(default_factory=threading.Lock)
+    deliveries: list[Delivery] = field(default_factory=list)
+    sessions: list[list[str]] = field(default_factory=list)
+
+    def started(self, commands: list[str]) -> None:
+        with self.lock:
+            self.sessions.append(commands)
+
+    def delivered(
+        self, marker: str, commands: list[str], message: bytes
+    ) -> None:
+        with self.lock:
+            self.deliveries.append(
+                Delivery(marker, list(commands), message)
+            )
+
+    def matching(self, marker: str) -> list[Delivery]:
+        with self.lock:
+            return [
+                delivery for delivery in self.deliveries
+                if delivery.marker == marker
+            ]
+
+    def session_snapshot(self) -> list[list[str]]:
+        with self.lock:
+            return [list(commands) for commands in self.sessions]
+
+
+class MetadataSMTPHandler(socketserver.BaseRequestHandler):
+    """Minimal ESMTP peer implementing Postfix-style proxy extensions."""
+
+    def handle(self) -> None:
+        capture: Capture = self.server.capture  # type: ignore[attr-defined]
+        connection = self.request
+        connection.settimeout(15)
+        buffer = bytearray()
+        commands: list[str] = []
+        sender = b""
+        recipients: list[bytes] = []
+        message = bytearray()
+        data_mode = False
+        greeted = False
+        capture.started(commands)
+
+        def send(line: bytes) -> None:
+            connection.sendall(line + b"\r\n")
+
+        def read_line() -> bytes | None:
+            while b"\n" not in buffer:
+                chunk = connection.recv(8192)
+                if not chunk:
+                    return None
+                buffer.extend(chunk)
+                if len(buffer) > 4 * 1024 * 1024:
+                    raise SMTP22Error("SMTP-22 fixture input exceeded limit")
+            end = buffer.index(b"\n") + 1
+            line = bytes(buffer[:end])
+            del buffer[:end]
+            return line.rstrip(b"\r\n")
+
+        send(b"220 smtp22.fixture ESMTP")
+        while True:
+            try:
+                line = read_line()
+            except (OSError, SMTP22Error):
+                return
+            if line is None:
+                return
+            if data_mode:
+                if line == b".":
+                    source = bytes(message)
+                    marker = next(
+                        (
+                            value.decode("ascii")
+                            for value in source.split()
+                            if value.startswith(TOKEN.encode("ascii") + b"-")
+                        ),
+                        "",
+                    ).rstrip(">\r\n")
+                    capture.delivered(marker, commands, source)
+                    send(b"250 2.0.0 queued")
+                    data_mode = False
+                    sender = b""
+                    recipients.clear()
+                    message.clear()
+                else:
+                    if line.startswith(b".."):
+                        line = line[1:]
+                    message.extend(line + b"\r\n")
+                continue
+
+            decoded = line.decode("ascii", "replace")
+            commands.append(decoded)
+            command, _, argument = line.partition(b" ")
+            command = command.upper()
+            if command == b"EHLO":
+                greeted = True
+                send(b"250-smtp22.fixture")
+                send(b"250-XCLIENT HELO PROTO ADDR")
+                send(b"250-XFORWARD NAME ADDR PORT IDENT SOURCE")
+                send(b"250 SIZE 10485760")
+            elif command == b"HELO":
+                greeted = True
+                send(b"250 smtp22.fixture")
+            elif command == b"XCLIENT":
+                if not greeted or not argument:
+                    send(b"503 5.5.1 EHLO and attributes required")
+                else:
+                    greeted = False
+                    sender = b""
+                    recipients.clear()
+                    send(b"220 smtp22.fixture ESMTP after XCLIENT")
+            elif command == b"XFORWARD":
+                if not greeted or not argument:
+                    send(b"503 5.5.1 EHLO and attributes required")
+                else:
+                    send(b"250 2.0.0 XFORWARD accepted")
+            elif command == b"MAIL":
+                if not greeted:
+                    send(b"503 5.5.1 send EHLO after XCLIENT")
+                else:
+                    sender = argument
+                    send(b"250 2.1.0 sender accepted")
+            elif command == b"RCPT":
+                if not sender:
+                    send(b"503 5.5.1 MAIL required")
+                else:
+                    recipients.append(argument)
+                    send(b"250 2.1.5 recipient accepted")
+            elif command == b"DATA":
+                if not sender or not recipients:
+                    send(b"503 5.5.1 envelope required")
+                else:
+                    message.clear()
+                    data_mode = True
+                    send(b"354 end with <CRLF>.<CRLF>")
+            elif command == b"RSET":
+                sender = b""
+                recipients.clear()
+                message.clear()
+                data_mode = False
+                send(b"250 2.0.0 reset")
+            elif command == b"NOOP":
+                send(b"250 2.0.0 ok")
+            elif command == b"QUIT":
+                send(b"221 2.0.0 bye")
+                return
+            else:
+                send(b"502 5.5.1 command not implemented")
+
+
+class ThreadingSMTPServer(
+    socketserver.ThreadingMixIn, socketserver.TCPServer
+):
+    daemon_threads = True
+    allow_reuse_address = True
+
+
+def read_configuration(name: str) -> dict:
+    completed = SMTP07.run(
+        privileged([SMTP07.ADMIN, "__config-read", name])
+    )
+    try:
+        value = json.loads(completed.stdout)
+    except json.JSONDecodeError as error:
+        raise SMTP22Error(
+            f"Bongo returned invalid {name} configuration"
+        ) from error
+    if not isinstance(value, dict):
+        raise SMTP22Error(f"Bongo {name} configuration is not an object")
+    return value
+
+
+def replace_configuration(name: str, value: dict) -> None:
+    SMTP07.run(
+        privileged([SMTP07.ADMIN, "__config-replace", name]),
+        input_text=json.dumps(value, separators=(",", ":")),
+    )
+
+
+def restart_bongo() -> None:
+    SMTP07.run(
+        privileged([SMTP07.SYSTEMCTL, "restart", "bongo.service"])
+    )
+    SMTP07.wait_for_bongo()
+
+
+def direct_configuration(
+    original: dict,
+    *,
+    xclient: list[str],
+    xforward: list[str],
+) -> dict:
+    configuration = copy.deepcopy(original)
+    configuration.update(
+        {
+            "use_relay_host": False,
+            "xclient_trusted_destinations": xclient,
+            "xforward_trusted_destinations": xforward,
+            "proxy_header_trusted_destinations": [],
+        }
+    )
+    return configuration
+
+
+def submit(host: str, port: int, case: str) -> str:
+    marker = f"{TOKEN}-{case}"
+    message = (
+        "From: SMTP-22 fixture <smtp22@bongo.test>\r\n"
+        f"To: capture-{case}@example.net\r\n"
+        f"Subject: {marker}\r\n"
+        f"Message-ID: <{marker}@bongo.test>\r\n"
+        "\r\n"
+        f"Outbound metadata fixture {marker}\r\n"
+    ).encode("ascii")
+    with smtplib.SMTP(host, port, timeout=10) as client:
+        client.ehlo("smtp22-source.bongo.test")
+        refused = client.sendmail(
+            "smtp22@bongo.test",
+            [f"capture-{case}@[{FIXTURE_HOST}]"],
+            message,
+        )
+        if refused:
+            raise SMTP22Error(
+                f"Bongo refused SMTP-22 ingress recipient: {refused}")
+    return marker
+
+
+def wait_for_delivery(capture: Capture, marker: str) -> Delivery:
+    deadline = time.monotonic() + TIMEOUT
+    while time.monotonic() < deadline:
+        matching = capture.matching(marker)
+        if len(matching) == 1:
+            return matching[0]
+        if len(matching) > 1:
+            raise SMTP22Error(f"duplicate delivery for {marker}")
+        time.sleep(0.2)
+    raise SMTP22Error(
+        f"metadata fixture did not receive {marker}; "
+        f"sessions={capture.session_snapshot()!r}, "
+        f"deliveries={[item.marker for item in capture.deliveries]!r}"
+    )
+
+
+def command_of(delivery: Delivery, name: str) -> list[str]:
+    prefix = f"{name} "
+    return [
+        command for command in delivery.commands
+        if command.upper().startswith(prefix)
+    ]
+
+
+def validate_untrusted(delivery: Delivery) -> None:
+    if command_of(delivery, "XCLIENT") or command_of(delivery, "XFORWARD"):
+        raise SMTP22Error(
+            "Bongo disclosed proxy metadata to an untrusted destination")
+
+
+def validate_xclient(delivery: Delivery, *, also_xforward: bool) -> None:
+    commands = command_of(delivery, "XCLIENT")
+    if len(commands) != 1:
+        raise SMTP22Error(
+            f"expected one XCLIENT command, received {commands!r}")
+    command = commands[0]
+    for attribute in ("HELO=smtp22-source.bongo.test", "PROTO=ESMTP",
+                      "ADDR="):
+        if attribute not in command:
+            raise SMTP22Error(
+                f"XCLIENT omitted advertised attribute {attribute}")
+    for attribute in ("NAME=", "PORT=", "LOGIN=", "DESTADDR=", "DESTPORT="):
+        if attribute in command:
+            raise SMTP22Error(
+                f"XCLIENT sent unadvertised attribute {attribute}")
+    xclient_index = delivery.commands.index(command)
+    mail_index = next(
+        index for index, value in enumerate(delivery.commands)
+        if value.upper().startswith("MAIL ")
+    )
+    ehlo_after = [
+        index for index, value in enumerate(delivery.commands)
+        if value.upper().startswith("EHLO ") and
+        xclient_index < index < mail_index
+    ]
+    if len(ehlo_after) != 1:
+        raise SMTP22Error("XCLIENT was not followed by exactly one fresh EHLO")
+    if also_xforward:
+        forward = command_of(delivery, "XFORWARD")
+        if len(forward) != 1:
+            raise SMTP22Error("XFORWARD was not sent after XCLIENT/EHLO")
+        if not (ehlo_after[0] < delivery.commands.index(forward[0]) <
+                mail_index):
+            raise SMTP22Error("XFORWARD was sent in the wrong SMTP state")
+    elif command_of(delivery, "XFORWARD"):
+        raise SMTP22Error("unexpected XFORWARD command")
+
+
+def validate_xforward(delivery: Delivery) -> None:
+    if command_of(delivery, "XCLIENT"):
+        raise SMTP22Error("unexpected XCLIENT command")
+    commands = command_of(delivery, "XFORWARD")
+    if len(commands) != 1:
+        raise SMTP22Error(
+            f"expected one XFORWARD command, received {commands!r}")
+    command = commands[0]
+    for attribute in ("ADDR=", "PORT=", "IDENT=", "SOURCE=LOCAL"):
+        if attribute not in command:
+            raise SMTP22Error(
+                f"XFORWARD omitted advertised attribute {attribute}")
+    for attribute in ("PROTO=", "HELO=", "LOGIN=", "DESTADDR=", "DESTPORT="):
+        if attribute in command:
+            raise SMTP22Error(
+                f"XFORWARD sent unadvertised attribute {attribute}")
+    if delivery.commands.index(command) > next(
+        index for index, value in enumerate(delivery.commands)
+        if value.upper().startswith("MAIL ")
+    ):
+        raise SMTP22Error("XFORWARD was sent after MAIL")
+
+
+def cleanup_queue() -> None:
+    for identifier in SMTP07.queue_ids():
+        message = SMTP07.queue("message", identifier, check=False)
+        if message.returncode == 0 and TOKEN in message.stdout:
+            SMTP07.queue("delete", identifier)
+
+
+def perform_test() -> None:
+    original_smtp = read_configuration("smtp")
+    original_global = read_configuration("global")
+    internal_host = str(
+        original_smtp.get("internal_relay_bind_address", "")
+    ).strip()
+    if not internal_host or internal_host in {"0.0.0.0", "::"}:
+        internal_host = str(original_global.get("hostaddr", "")).strip()
+    if not internal_host or internal_host in {"0.0.0.0", "::"}:
+        internal_host = "127.0.0.1"
+    internal_port = int(original_smtp.get("internal_relay_port", 26))
+    SMTP07.INTERNAL_HOST = internal_host
+    SMTP07.INTERNAL_PORT = internal_port
+
+    capture = Capture()
+    server = ThreadingSMTPServer(
+        (FIXTURE_HOST, FIXTURE_PORT), MetadataSMTPHandler
+    )
+    server.capture = capture  # type: ignore[attr-defined]
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    configuration_changed = False
+    try:
+        cases = (
+            ("untrusted", ["192.0.2.0/24"], ["198.51.100.0/24"]),
+            ("xclient", ["127.0.0.1"], []),
+            ("xforward", [], ["127.0.0.0/8"]),
+            ("both", ["127.0.0.0/8"], ["127.0.0.1"]),
+        )
+        for case, xclient, xforward in cases:
+            replace_configuration(
+                "smtp",
+                direct_configuration(
+                    original_smtp, xclient=xclient, xforward=xforward
+                ),
+            )
+            configuration_changed = True
+            restart_bongo()
+            marker = submit(internal_host, internal_port, case)
+            delivery = wait_for_delivery(capture, marker)
+            if case == "untrusted":
+                validate_untrusted(delivery)
+            elif case == "xclient":
+                validate_xclient(delivery, also_xforward=False)
+            elif case == "xforward":
+                validate_xforward(delivery)
+            else:
+                validate_xclient(delivery, also_xforward=True)
+    finally:
+        restoration_errors: list[str] = []
+        if configuration_changed:
+            try:
+                replace_configuration("smtp", original_smtp)
+            except (OSError, SMTP07.SMTP07Error) as error:
+                restoration_errors.append(
+                    f"SMTP configuration restore failed: {error}")
+        try:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=5)
+        except OSError as error:
+            restoration_errors.append(
+                f"SMTP fixture shutdown failed: {error}")
+        try:
+            restart_bongo()
+            cleanup_queue()
+        except (OSError, SMTP07.SMTP07Error) as error:
+            restoration_errors.append(
+                f"service/Queue restore failed: {error}")
+        if restoration_errors:
+            raise SMTP22Error("; ".join(restoration_errors))
+
+    if read_configuration("smtp") != original_smtp:
+        raise SMTP22Error("SMTP configuration was not restored exactly")
+    print(
+        "SMTP-22 PASS metadata=xclient/xforward "
+        "trust=untrusted-blocked/ip/cidr attributes=advertised-only "
+        "xclient=220/fresh-ehlo xforward=250/in-transaction "
+        "combined=ordered duplicate=no queue-clean=yes "
+        "config-restored=yes service-restored=active"
+    )
+
+
+def main() -> int:
+    try:
+        perform_test()
+        return 0
+    except (
+        json.JSONDecodeError,
+        OSError,
+        SMTP22Error,
+        SMTP07.SMTP07Error,
+        smtplib.SMTPException,
+        StopIteration,
+        ValueError,
+    ) as error:
+        print(f"smtp-forward-metadata-check: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())