bongo/contrib/testing/smtp-internal-normalization-check.py

#!/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 internal SMTP sender normalization in the real Queue envelope."""

from __future__ import annotations

import copy
import json
import os
import re
import socket
import subprocess
import sys
import time


ADMIN = os.environ.get("BONGO_TEST_ADMIN", "/usr/bin/bongo-admin")
QUEUE_TOOL = os.environ.get(
    "BONGO_TEST_QUEUE_TOOL", "/usr/bin/bongo-queuetool")
SYSTEMCTL = os.environ.get("BONGO_TEST_SYSTEMCTL", "/usr/bin/systemctl")
HOST = os.environ.get("BONGO_TEST_INTERNAL_HOST", "172.16.11.190")
PORT = int(os.environ.get("BONGO_TEST_INTERNAL_PORT", "26"))
SOURCE = os.environ.get("BONGO_TEST_ALLOWED_SOURCE", HOST)
DOMAIN = os.environ.get("BONGO_TEST_DOMAIN", "bongo.test")
DEVICE = os.environ.get("BONGO_TEST_DEVICE", "matrix")
RECIPIENT = os.environ.get(
    "BONGO_TEST_REMOTE_RECIPIENT", "capture@normalization.invalid")
TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "10"))
STARTUP_GRACE = float(os.environ.get("BONGO_TEST_STARTUP_GRACE", "2"))
ALLOW_LIVE = os.environ.get("BONGO_ALLOW_LIVE_SMTP_TEST") == "1"
RESPONSE = re.compile(rb"^([0-9]{3})([- ])(.*)\r\n$")
QUEUE_ID = re.compile(r"^[0-9]{3}-[0-9a-f]+$")
TOKEN = f"smtp05-{os.getpid()}-{int(time.time())}"


class NormalizationCheckError(RuntimeError):
    """Raised when live sender normalization differs from its contract."""


def run(
    arguments: list[str],
    *,
    input_text: str | None = None,
    binary: bool = False,
    check: bool = True,
) -> subprocess.CompletedProcess:
    completed = subprocess.run(
        arguments,
        input=input_text,
        capture_output=True,
        text=not binary,
        check=False,
    )
    if check and completed.returncode:
        stdout = completed.stdout
        stderr = completed.stderr
        if binary:
            stdout = stdout.decode("utf-8", "replace")
            stderr = stderr.decode("utf-8", "replace")
        raise NormalizationCheckError(
            f"{' '.join(arguments)} failed: "
            f"{stderr.strip() or stdout.strip()}")
    return completed


def admin(*arguments: str, input_text: str | None = None) -> str:
    return run(
        ["sudo", "-n", ADMIN, *arguments],
        input_text=input_text,
    ).stdout


def queue(*arguments: str, binary: bool = False,
          check: bool = True) -> subprocess.CompletedProcess:
    return run(
        ["sudo", "-n", "-u", "bongo", QUEUE_TOOL, *arguments],
        binary=binary,
        check=check,
    )


def read_configuration() -> dict:
    try:
        configuration = json.loads(admin("__config-read", "smtp"))
    except json.JSONDecodeError as error:
        raise NormalizationCheckError(
            "bongo-admin returned invalid SMTP configuration") from error
    if not isinstance(configuration, dict):
        raise NormalizationCheckError("SMTP configuration is not an object")
    return configuration


def replace_configuration(configuration: dict) -> None:
    admin(
        "__config-replace", "smtp",
        input_text=json.dumps(configuration, separators=(",", ":")),
    )


def queue_ids() -> set[str]:
    identifiers: set[str] = set()
    for line in queue("list").stdout.splitlines():
        fields = line.split()
        if fields and QUEUE_ID.fullmatch(fields[0]):
            identifiers.add(fields[0])
    return identifiers


def token_queue_ids(baseline: set[str]) -> list[str]:
    identifiers: list[str] = []
    for queue_id in queue_ids() - baseline:
        message = queue("message", queue_id, binary=True, check=False)
        if message.returncode == 0 and TOKEN.encode("ascii") in message.stdout:
            identifiers.append(queue_id)
    return identifiers


def cleanup_entries(baseline: set[str]) -> None:
    for queue_id in token_queue_ids(baseline):
        queue("delete", queue_id, check=False)


def wait_port() -> None:
    deadline = time.monotonic() + 30
    consecutive = 0
    while time.monotonic() < deadline:
        client = None
        try:
            client = SMTPConnection()
            client.response(220)
            client.command("QUIT", 221)
            consecutive += 1
            if consecutive == 2:
                return
            time.sleep(0.25)
        except (OSError, NormalizationCheckError):
            consecutive = 0
            time.sleep(0.1)
        finally:
            if client is not None:
                client.close()
    raise NormalizationCheckError(
        f"internal SMTP did not become ready at {HOST}:{PORT}")


def restart_bongo() -> None:
    run(["sudo", "-n", SYSTEMCTL, "restart", "bongo.service"])
    time.sleep(STARTUP_GRACE)
    wait_port()
    if run(
        ["sudo", "-n", SYSTEMCTL, "is-active", "bongo.service"]
    ).stdout.strip() != "active":
        raise NormalizationCheckError("bongo.service is not active")


class SMTPConnection:
    def __init__(self) -> None:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(TIMEOUT)
        try:
            sock.bind((SOURCE, 0))
            sock.connect((HOST, PORT))
        except BaseException:
            sock.close()
            raise
        self.socket = sock
        self.reader = sock.makefile("rb")

    def close(self) -> None:
        try:
            self.reader.close()
        finally:
            self.socket.close()

    def response(self, expected: int) -> list[bytes]:
        lines: list[bytes] = []
        response_code = -1
        while True:
            line = self.reader.readline(8194)
            if not line:
                raise NormalizationCheckError(
                    f"connection closed while waiting for SMTP {expected}")
            match = RESPONSE.match(line)
            if match is None:
                raise NormalizationCheckError(
                    f"malformed SMTP response: {line!r}")
            code = int(match.group(1))
            if response_code == -1:
                response_code = code
            elif code != response_code:
                raise NormalizationCheckError(
                    f"multiline SMTP response changed code: {line!r}")
            if code != expected:
                raise NormalizationCheckError(
                    f"expected SMTP {expected}, received {line!r}")
            lines.append(match.group(3))
            if match.group(2) == b" ":
                return lines

    def command(self, command: str, expected: int) -> list[bytes]:
        self.socket.sendall(command.encode("ascii") + b"\r\n")
        return self.response(expected)


def submit(sender: str | None, label: str) -> None:
    client = SMTPConnection()
    try:
        client.response(220)
        client.command("EHLO smtp05-device.bongo.test", 250)
        if sender is None:
            client.command("MAIL FROM:<>", 250)
        else:
            client.command(f"MAIL FROM:<{sender}>", 250)
        client.command(f"RCPT TO:<{RECIPIENT}>", 250)
        client.command("DATA", 354)
        message = (
            f"From: device@{DOMAIN}\r\n"
            f"To: {RECIPIENT}\r\n"
            f"Subject: SMTP-05 {label} {TOKEN}\r\n"
            f"Message-ID: <{TOKEN}-{label}@{DOMAIN}>\r\n"
            "\r\n"
            f"Internal sender normalization check {TOKEN} {label}\r\n"
        ).encode("ascii")
        client.socket.sendall(message + b".\r\n")
        client.response(250)
        client.command("QUIT", 221)
    finally:
        client.close()


def wait_envelope(baseline: set[str], expected_sender: str) -> str:
    deadline = time.monotonic() + 20
    while time.monotonic() < deadline:
        matches = token_queue_ids(baseline)
        if len(matches) > 1:
            raise NormalizationCheckError(
                f"test created multiple Queue entries: {matches!r}")
        if matches:
            held = queue("hold", matches[0], check=False)
            if held.returncode:
                time.sleep(0.1)
                continue
            queue_id = held.stdout.strip()
            if not QUEUE_ID.fullmatch(queue_id) or not queue_id.startswith(
                "999-"
            ):
                raise NormalizationCheckError(
                    f"invalid held Queue ID: {queue_id!r}")
            envelope = queue("show", queue_id, binary=True).stdout
            from_lines = [
                line[1:].decode("utf-8", "strict").split()
                for line in envelope.splitlines() if line.startswith(b"F")
            ]
            if len(from_lines) != 1 or not from_lines[0]:
                raise NormalizationCheckError(
                    f"Queue entry {queue_id} has invalid FROM envelope")
            actual_sender = from_lines[0][0]
            if actual_sender != expected_sender:
                raise NormalizationCheckError(
                    f"expected envelope sender {expected_sender!r}, "
                    f"received {actual_sender!r}")
            return queue_id
        time.sleep(0.1)
    raise NormalizationCheckError(
        f"no deferred Queue entry containing {TOKEN} appeared")


def check_case(
    baseline: set[str], label: str, sender: str | None, expected: str
) -> None:
    submit(sender, label)
    queue_id = wait_envelope(baseline, expected)
    queue("delete", queue_id)


def main() -> int:
    if not ALLOW_LIVE:
        raise NormalizationCheckError(
            "set BONGO_ALLOW_LIVE_SMTP_TEST=1 for this temporary "
            "live-configuration test")
    if STARTUP_GRACE < 0:
        raise NormalizationCheckError(
            "BONGO_TEST_STARTUP_GRACE cannot be negative")

    original = read_configuration()
    if not original.get("internal_relay_enabled"):
        raise NormalizationCheckError("internal SMTP is not enabled")
    if original.get("internal_relay_bind_address") != HOST:
        raise NormalizationCheckError(
            "BONGO_TEST_INTERNAL_HOST does not match live configuration")
    if int(original.get("internal_relay_port", 0)) != PORT:
        raise NormalizationCheckError(
            "BONGO_TEST_INTERNAL_PORT does not match live configuration")
    if original.get("internal_relay_domain") != DOMAIN:
        raise NormalizationCheckError(
            "BONGO_TEST_DOMAIN does not match internal relay domain")

    baseline = queue_ids()
    test_configuration = copy.deepcopy(original)
    test_configuration["internal_relay_device_mappings"] = [
        f"{SOURCE}={DEVICE}"]
    test_configuration["internal_relay_dns_suffixes"] = []
    test_configuration["use_relay_host"] = True
    test_configuration["relay_host"] = "127.0.0.2"
    test_configuration["internal_relay_messages_per_minute"] = max(
        100, int(original.get("internal_relay_messages_per_minute", 0)))
    test_configuration["internal_relay_recipients_per_minute"] = max(
        100, int(original.get("internal_relay_recipients_per_minute", 0)))

    restored = False
    try:
        replace_configuration(test_configuration)
        restart_bongo()
        cases = [
            ("device-root", "root@localhost",
             f"root+{DEVICE}@{DOMAIN}"),
            ("local-daemon", "daemon@localdomain",
             f"daemon@{DOMAIN}"),
            ("localhost-fqdn", "ROOT@LOCALHOST.LOCALDOMAIN",
             f"ROOT+{DEVICE}@{DOMAIN}"),
            ("subdomain", f"root@matrix.{DOMAIN}",
             f"root+matrix@{DOMAIN}"),
            ("nested-subdomain", f"alerts@rack3.matrix.{DOMAIN}",
             f"alerts+rack3.matrix@{DOMAIN}"),
            ("send-only-subdomain", f"rsync@backup.{DOMAIN}",
             f"rsync+backup@{DOMAIN}"),
            ("hosted-domain", f"blog@{DOMAIN}",
             f"blog@{DOMAIN}"),
            ("external-boundary", f"root@attacker{DOMAIN}",
             f"root+{DEVICE}@{DOMAIN}"),
            ("external", "sender@example.net", f"sender@{DOMAIN}"),
            ("null-sender", None, "-"),
        ]
        for label, sender, expected in cases:
            print(f"SMTP-05: {label}", file=sys.stderr, flush=True)
            check_case(baseline, label, sender, expected)

        cleanup_entries(baseline)
        replace_configuration(original)
        restart_bongo()
        restored = read_configuration() == original
        if not restored:
            raise NormalizationCheckError(
                "SMTP configuration did not round-trip after restoration")
    finally:
        cleanup_entries(baseline)
        if not restored:
            replace_configuration(original)
            restart_bongo()

    if token_queue_ids(baseline):
        raise NormalizationCheckError("test Queue entries remain after cleanup")
    print(
        "SMTP-05 PASS "
        f"host={HOST}:{PORT} source={SOURCE} device={DEVICE} cases=10 "
        "queue-envelope=yes queue-held=yes fcrdns=unit-tested "
        "external-boundary=yes "
        "internet-delivery=no restored=yes"
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (
        NormalizationCheckError,
        json.JSONDecodeError,
        OSError,
        UnicodeError,
        ValueError,
    ) as error:
        print(f"SMTP-05 FAIL: {error}")
        raise SystemExit(1)