bongo/contrib/testing/smtp-client-interoperability-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 local delivery through independent SMTP/MTA client programs."""

from __future__ import annotations

from email import policy
from email.parser import BytesParser
import os
from pathlib import Path
import pty
import select
import shutil
import subprocess
import sys
import tempfile
import time
from urllib.parse import quote

from bongo.store.StoreClient import StoreClient


HOST = os.environ.get("BONGO_TEST_HOST", "127.0.0.1")
PORT = int(os.environ.get("BONGO_TEST_SUBMISSION_PORT", "587"))
INBOUND_HOST = os.environ.get("BONGO_TEST_INBOUND_HOST", "127.0.0.1")
INBOUND_PORT = int(os.environ.get("BONGO_TEST_INBOUND_PORT", "25"))
INTERNAL_HOST = os.environ.get(
    "BONGO_TEST_INTERNAL_HOST", "172.16.11.190")
INTERNAL_PORT = int(os.environ.get("BONGO_TEST_INTERNAL_PORT", "26"))
STORE_PORT = int(os.environ.get("BONGO_TEST_STORE_PORT", "689"))
SENDER = os.environ.get("BONGO_TEST_USER1", "test1")
RECIPIENT = os.environ.get("BONGO_TEST_USER2", "test2")
DOMAIN = os.environ.get("BONGO_TEST_DOMAIN", "bongo.test")
PASSWORD = os.environ.get("BONGO_TEST_PASSWORD", "")
TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "90"))
QUEUE_TOOL = os.environ.get(
    "BONGO_TEST_QUEUE_TOOL", "/usr/bin/bongo-queuetool")
ALLOW_LIVE = os.environ.get("BONGO_ALLOW_LIVE_USER_TEST") == "1"
REQUESTED_CLIENTS = {
    item.strip().lower()
    for item in os.environ.get(
        "BONGO_TEST_SMTP_CLIENTS", "curl,swaks,xeams,mailutils",
    ).split(",")
    if item.strip()
}
TELNET_PLAIN = os.environ.get("BONGO_TEST_SMTP_TELNET", "1") != "0"
TELNET_TYPE_DELAY = float(os.environ.get(
    "BONGO_TEST_TELNET_TYPE_DELAY", "1"))
RUN_TOKEN = f"smtp-client-{os.getpid()}-{int(time.time())}"


class InteroperabilityError(RuntimeError):
    """Raised when a real SMTP/MTA client is incompatible with Bongo."""


def require_program(name: str) -> str:
    program = shutil.which(name)
    if program is None:
        raise InteroperabilityError(f"required client {name!r} is not installed")
    return program


def run_client(
    command: list[str],
    *,
    input_data: str | None = None,
    display_name: str,
) -> subprocess.CompletedProcess[str]:
    try:
        completed = subprocess.run(
            command,
            input=input_data,
            capture_output=True,
            text=True,
            timeout=TIMEOUT,
            check=False,
        )
    except subprocess.TimeoutExpired as error:
        raise InteroperabilityError(
            f"{display_name} exceeded the {TIMEOUT:g}-second timeout") from error
    if completed.returncode:
        detail = completed.stderr.strip() or completed.stdout.strip()
        raise InteroperabilityError(
            f"{display_name} exited with {completed.returncode}: {detail}")
    return completed


def run_queue(
    command: str, *arguments: str, binary: bool = False,
    check: bool = True,
) -> subprocess.CompletedProcess:
    completed = subprocess.run(
        [
            "sudo", "-n", "-u", "bongo", QUEUE_TOOL,
            command, *arguments,
        ],
        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 InteroperabilityError(
            f"{QUEUE_TOOL} {command} failed: "
            f"{stderr.strip() or stdout.strip()}")
    return completed


def queue_token_ids(token: str) -> list[str]:
    identifiers: list[str] = []
    for line in run_queue("list").stdout.splitlines():
        fields = line.split()
        if not fields:
            continue
        message = run_queue(
            "message", fields[0], binary=True, check=False)
        if message.returncode == 0 and token.encode("ascii") in message.stdout:
            identifiers.append(fields[0])
    return identifiers


def open_store() -> StoreClient:
    return StoreClient(
        RECIPIENT,
        RECIPIENT,
        authPassword=PASSWORD,
        host=HOST,
        port=STORE_PORT,
    )


def token_documents(
    store: StoreClient, token: str,
) -> list[tuple[str, bytes]]:
    matches: list[tuple[str, bytes]] = []
    # Store LIST is a streaming command. Consume it completely before READ.
    documents = [
        document.uid for document in store.List("/mail/INBOX")
    ]
    for document in documents:
        try:
            data = store.Read(document)
        except Exception:
            continue
        if token.encode("ascii") in data:
            matches.append((document, data))
    return matches


def delete_document(store: StoreClient, document: str) -> None:
    try:
        store.Delete(document)
    except Exception as error:
        print(
            f"SMTP client cleanup warning for Store document "
            f"{document}: {error}",
            file=sys.stderr,
        )


def cleanup_token(store: StoreClient, token: str) -> None:
    deadline = time.monotonic() + 15
    quiet_since: float | None = None
    while time.monotonic() < deadline:
        found = False
        for document, _ in token_documents(store, token):
            found = True
            delete_document(store, document)
        for queue_id in queue_token_ids(token):
            found = True
            run_queue("delete", queue_id, check=False)
        if found:
            quiet_since = None
        elif quiet_since is None:
            quiet_since = time.monotonic()
        elif time.monotonic() - quiet_since >= 1:
            return
        time.sleep(0.25)


def message_bytes(token: str, client: str, sender: str | None = None) -> bytes:
    sender = sender or f"{SENDER}@{DOMAIN}"
    recipient = f"{RECIPIENT}@{DOMAIN}"
    lines = [
        f"From: {sender}",
        f"To: {recipient}",
        f"Subject: SMTP client interoperability {client} {token}",
        f"Message-ID: <{token}@{DOMAIN}>",
        f"X-Bongo-Test: {token}",
        f"X-Bongo-Test-Client: {client}",
        "MIME-Version: 1.0",
        "Content-Type: text/plain; charset=UTF-8",
        "Content-Transfer-Encoding: 8bit",
        "",
        f"Independent {client} delivery check {token}",
        "",
    ]
    return "\r\n".join(lines).encode("utf-8")


def submit_curl(directory: Path, token: str) -> None:
    curl = require_program("curl")
    message = directory / "curl.eml"
    credentials = directory / "curl.conf"
    message.write_bytes(message_bytes(token, "curl"))
    credentials.write_text(
        f'user = "{SENDER}:{PASSWORD}"\n',
        encoding="utf-8",
    )
    credentials.chmod(0o600)
    run_client(
        [
            curl,
            "--disable",
            "--silent",
            "--show-error",
            "--fail",
            "--ssl-reqd",
            "--insecure",
            "--connect-timeout", str(TIMEOUT),
            "--max-time", str(TIMEOUT),
            "--config", str(credentials),
            "--url", f"smtp://{HOST}:{PORT}",
            "--mail-from", f"{SENDER}@{DOMAIN}",
            "--mail-rcpt", f"{RECIPIENT}@{DOMAIN}",
            "--upload-file", str(message),
        ],
        display_name="curl SMTP submission",
    )


def submit_plain_curl(
    directory: Path, token: str, host: str, port: int, sender: str, mode: str,
) -> None:
    curl = require_program("curl")
    message = directory / f"{mode}-curl.eml"
    message.write_bytes(message_bytes(token, f"{mode}-curl", sender))
    run_client(
        [
            curl,
            "--disable",
            "--silent",
            "--show-error",
            "--fail",
            "--connect-timeout", str(TIMEOUT),
            "--max-time", str(TIMEOUT),
            "--url", f"smtp://{host}:{port}",
            "--mail-from", sender,
            "--mail-rcpt", f"{RECIPIENT}@{DOMAIN}",
            "--upload-file", str(message),
        ],
        display_name=f"curl {mode} SMTP delivery",
    )


def submit_swaks(directory: Path, token: str) -> None:
    swaks = require_program("swaks")
    message = directory / "swaks.eml"
    configuration = directory / "swaks.conf"
    message.write_bytes(message_bytes(token, "swaks"))
    configuration.write_text(
        "\n".join([
            f"server {HOST}",
            f"port {PORT}",
            "tls",
            "auth LOGIN",
            f"auth-user {SENDER}",
            f"auth-password {PASSWORD}",
            f"from {SENDER}@{DOMAIN}",
            f"to {RECIPIENT}@{DOMAIN}",
            f"data @{message}",
            f"timeout {TIMEOUT:g}s",
            "silent 2",
            "",
        ]),
        encoding="utf-8",
    )
    configuration.chmod(0o600)
    run_client(
        [swaks, "--config", str(configuration)],
        display_name="swaks SMTP submission",
    )


def submit_plain_swaks(
    directory: Path, token: str, host: str, port: int, sender: str, mode: str,
) -> None:
    swaks = require_program("swaks")
    message = directory / f"{mode}-swaks.eml"
    configuration = directory / f"{mode}-swaks.conf"
    message.write_bytes(message_bytes(token, f"{mode}-swaks", sender))
    configuration.write_text(
        "\n".join([
            f"server {host}",
            f"port {port}",
            f"from {sender}",
            f"to {RECIPIENT}@{DOMAIN}",
            f"data @{message}",
            f"timeout {TIMEOUT:g}s",
            "silent 2",
            "",
        ]),
        encoding="utf-8",
    )
    configuration.chmod(0o600)
    run_client(
        [swaks, "--config", str(configuration)],
        display_name=f"swaks {mode} SMTP delivery",
    )


def submit_xeams(_directory: Path, token: str) -> None:
    xeams = require_program("xeams-email-sender")
    sender = f"{SENDER}@{DOMAIN}"
    recipient = f"{RECIPIENT}@{DOMAIN}"
    completed = run_client(
        [
            xeams,
            "-srv", f"{HOST}:{PORT}",
            "-suid", SENDER,
            "-spwd", PASSWORD,
            "-ssl", "true",
            "-subject", f"SMTP client interoperability Xeams {token}",
            "-body", f"Independent Xeams Email Sender check {token}",
            "-recipient", recipient,
            "-sender", sender,
            "-useJM", "true",
        ],
        display_name="Xeams Email Sender SMTP submission",
    )
    combined = f"{completed.stdout}\n{completed.stderr}".lower()
    if "exception" in combined or "error sending" in combined:
        raise InteroperabilityError(
            "Xeams Email Sender reported an error despite returning success")


def submit_plain_xeams(
    _directory: Path, token: str, host: str, port: int, sender: str, mode: str,
) -> None:
    xeams = require_program("xeams-email-sender")
    completed = run_client(
        [
            xeams,
            "-srv", f"{host}:{port}",
            "-ssl", "false",
            "-subject", f"SMTP client interoperability {mode} Xeams {token}",
            "-body", f"Independent {mode} Xeams check {token}",
            "-recipient", f"{RECIPIENT}@{DOMAIN}",
            "-sender", sender,
            "-useJM", "true",
        ],
        display_name=f"Xeams Email Sender {mode} SMTP delivery",
    )
    combined = f"{completed.stdout}\n{completed.stderr}".lower()
    if "exception" in combined or "error sending" in combined:
        raise InteroperabilityError(
            f"Xeams Email Sender reported an error during {mode} delivery")


def submit_mail(directory: Path, token: str) -> None:
    mail = require_program("mail")
    sender = f"{SENDER}@{DOMAIN}"
    recipient = f"{RECIPIENT}@{DOMAIN}"
    configuration = directory / "mailutils.conf"
    smtp_url = (
        f"smtp://{quote(SENDER, safe='')}:{quote(PASSWORD, safe='')}"
        f";auth=login,plain@{HOST}:{PORT}"
        f";from={quote(sender, safe='@')}"
    )
    configuration.write_text(
        "\n".join([
            "mailer {",
            f'  url "{smtp_url}";',
            "};",
            "",
        ]),
        encoding="utf-8",
    )
    configuration.chmod(0o600)
    run_client(
        [
            mail,
            "--config-file", str(configuration),
            "--return-address", sender,
            "--subject", f"SMTP client interoperability mail {token}",
            "--append", f"Message-ID: <{token}@{DOMAIN}>",
            "--append", f"X-Bongo-Test: {token}",
            "--append", "X-Bongo-Test-Client: GNU-mailutils",
            recipient,
        ],
        input_data=f"Independent GNU Mailutils SMTP delivery check {token}\n",
        display_name="GNU Mailutils SMTP submission",
    )


def submit_plain_mail(
    directory: Path, token: str, host: str, port: int, sender: str, mode: str,
) -> None:
    mail = require_program("mail")
    configuration = directory / f"{mode}-mailutils.conf"
    configuration.write_text(
        "\n".join([
            "mailer {",
            f'  url "smtp://{host}:{port}";',
            "};",
            "",
        ]),
        encoding="utf-8",
    )
    configuration.chmod(0o600)
    run_client(
        [
            mail,
            "--config-file", str(configuration),
            "--return-address", sender,
            "--subject", f"SMTP client interoperability {mode} mail {token}",
            "--append", f"Message-ID: <{token}@{DOMAIN}>",
            "--append", f"X-Bongo-Test: {token}",
            "--append", f"X-Bongo-Test-Client: {mode}-GNU-mailutils",
            f"{RECIPIENT}@{DOMAIN}",
        ],
        input_data=f"Independent {mode} GNU Mailutils check {token}\n",
        display_name=f"GNU Mailutils {mode} SMTP delivery",
    )


def submit_plain_telnet(
    _directory: Path, token: str, host: str, port: int, sender: str, mode: str,
) -> None:
    telnet = require_program("telnet")
    actions = [
        (f"EHLO {mode}.bongo.test\n".encode("ascii"), 250),
        (f"MAIL FROM:<{sender}>\n".encode("ascii"), 250),
        (f"RCPT TO:<{RECIPIENT}@{DOMAIN}>\n".encode("ascii"), 250),
        (b"DATA\n", 354),
        (message_bytes(
            token, f"{mode}-telnet", sender).replace(b"\r\n", b"\n"), None),
        (b".\n", 250),
        (b"QUIT\n", 221),
    ]
    master, slave = pty.openpty()
    process = subprocess.Popen(
        [telnet, "-E", host, str(port)],
        stdin=slave,
        stdout=slave,
        stderr=slave,
        close_fds=True,
    )
    os.close(slave)
    transcript = bytearray()
    unread = bytearray()

    def read_reply(expected: int) -> None:
        deadline = time.monotonic() + TIMEOUT
        final_prefix = f"{expected} ".encode("ascii")
        while time.monotonic() < deadline:
            while b"\n" in unread:
                line, _, remainder = unread.partition(b"\n")
                unread[:] = remainder
                if line.rstrip(b"\r").startswith(final_prefix):
                    return
            ready, _, _ = select.select(
                [master], [], [],
                max(0.0, deadline - time.monotonic()))
            if not ready:
                break
            data = os.read(master, 4096)
            if not data:
                break
            transcript.extend(data)
            unread.extend(data)
        raise InteroperabilityError(
            f"telnet {mode} did not return SMTP {expected}: "
            f"{bytes(transcript).decode('utf-8', 'replace').strip()}")

    try:
        read_reply(220)
        time.sleep(TELNET_TYPE_DELAY)
        for chunk, expected in actions:
            os.write(master, chunk)
            if expected is not None:
                read_reply(expected)
            time.sleep(TELNET_TYPE_DELAY)
        process.wait(timeout=TIMEOUT)
    except (
        BrokenPipeError,
        InteroperabilityError,
        OSError,
        subprocess.TimeoutExpired,
    ) as error:
        process.kill()
        process.wait()
        raise InteroperabilityError(
            f"telnet {mode} connection closed during paced manual input: "
            f"{bytes(transcript).decode('utf-8', 'replace').strip()}") from error
    finally:
        os.close(master)


def wait_for_delivery(
    store: StoreClient, token: str,
) -> tuple[str, bytes]:
    deadline = time.monotonic() + TIMEOUT
    while time.monotonic() < deadline:
        matches = token_documents(store, token)
        if len(matches) > 1:
            raise InteroperabilityError(
                f"{token} was delivered more than once: "
                f"{[item[0] for item in matches]!r}")
        if matches:
            return matches[0]
        time.sleep(0.25)
    raise InteroperabilityError(
        f"{token} did not reach {RECIPIENT}'s INBOX")


def validate_message(data: bytes, token: str, client: str) -> None:
    message = BytesParser(policy=policy.default).parsebytes(data)
    if not client.endswith("xeams"):
        if message["Message-ID"] != f"<{token}@{DOMAIN}>":
            raise InteroperabilityError(
                f"{client} stored Message-ID changed or is missing")
        if message["X-Bongo-Test"] != token:
            raise InteroperabilityError(
                f"{client} stored test marker changed or is missing")
    elif client.endswith("xeams") and token not in str(message["Subject"]):
        raise InteroperabilityError("Xeams stored Subject changed")
    body = message.get_body(preferencelist=("plain",))
    if body is None or token not in body.get_content():
        raise InteroperabilityError(f"{client} stored message body changed")


def client_version(program: str) -> str:
    command = [require_program(program), "--version"]
    completed = subprocess.run(
        command,
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
    )
    first_line = (completed.stdout or completed.stderr).splitlines()
    return first_line[0].strip() if first_line else "unknown"


def main() -> int:
    if not ALLOW_LIVE:
        raise InteroperabilityError(
            "set BONGO_ALLOW_LIVE_USER_TEST=1 for disposable live accounts")
    if not PASSWORD:
        raise InteroperabilityError("BONGO_TEST_PASSWORD is required")
    if TIMEOUT <= 0:
        raise InteroperabilityError("BONGO_TEST_TIMEOUT must be positive")
    if TELNET_TYPE_DELAY < 0 or TELNET_TYPE_DELAY > TIMEOUT:
        raise InteroperabilityError(
            "BONGO_TEST_TELNET_TYPE_DELAY must be between zero and timeout")

    available_clients = {
        "curl": ("curl", submit_curl, submit_plain_curl),
        "swaks": ("swaks", submit_swaks, submit_plain_swaks),
        "xeams": ("xeams-email-sender", submit_xeams, submit_plain_xeams),
        "mailutils": ("mail", submit_mail, submit_plain_mail),
    }
    unknown = REQUESTED_CLIENTS - available_clients.keys()
    if unknown:
        raise InteroperabilityError(
            f"unknown client selection: {','.join(sorted(unknown))}")
    if not REQUESTED_CLIENTS:
        raise InteroperabilityError("select at least one SMTP client")
    for name, (program, _submitter, _plain_submitter) in \
            available_clients.items():
        if name in REQUESTED_CLIENTS:
            require_program(program)

    store = open_store()
    passed: list[str] = []
    tokens: list[str] = []
    try:
        with tempfile.TemporaryDirectory(
            prefix="bongo-smtp-clients-",
        ) as temporary:
            directory = Path(temporary)
            authenticated_clients = tuple(
                (name, submitter)
                for name, (_program, submitter, _plain_submitter)
                in available_clients.items()
                if name in REQUESTED_CLIENTS
            )
            plain_clients = tuple(
                (name, submitter)
                for name, (_program, _submitter, submitter)
                in available_clients.items()
                if name in REQUESTED_CLIENTS
            )
            if TELNET_PLAIN:
                plain_clients += (("telnet", submit_plain_telnet),)
            deliveries = []
            for name, submitter in authenticated_clients:
                deliveries.append((
                    "submission-auth", name, submitter, HOST, PORT,
                    f"{SENDER}@{DOMAIN}",
                ))
            for mode, host, port, sender in (
                ("inbound-noauth", INBOUND_HOST, INBOUND_PORT,
                 "outside@sender.invalid"),
                ("internal-noauth", INTERNAL_HOST, INTERNAL_PORT,
                 "device@localdomain"),
            ):
                for name, submitter in plain_clients:
                    deliveries.append(
                        (mode, name, submitter, host, port, sender))
            for index, (
                mode, name, submitter, host, port, sender,
            ) in enumerate(deliveries, start=1):
                token = f"{RUN_TOKEN}-{index}-{mode}-{name}"
                tokens.append(token)
                cleanup_token(store, token)
                if mode == "submission-auth":
                    submitter(directory, token)
                else:
                    submitter(directory, token, host, port, sender, mode)
                document, data = wait_for_delivery(store, token)
                validate_message(data, token, f"{mode}-{name}")
                delete_document(store, document)
                cleanup_token(store, token)
                if token_documents(store, token) or queue_token_ids(token):
                    raise InteroperabilityError(
                        f"{name} test content remains after cleanup")
                passed.append(f"{mode}:{name}")
    finally:
        for token in tokens:
            cleanup_token(store, token)
        store.Quit()

    versions = "; ".join(
        client_version(program) for program in ("curl", "swaks", "mail"))
    print(
        "SMTP-30 PASS "
        f"submission={HOST}:{PORT} inbound={INBOUND_HOST}:{INBOUND_PORT} "
        f"internal={INTERNAL_HOST}:{INTERNAL_PORT} "
        f"deliveries={len(passed)} clients={','.join(passed)} "
        f"recipient={RECIPIENT}@{DOMAIN} store-inbox=yes duplicate=no "
        f"queue-clean=yes store-clean=yes versions={versions}"
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (
        InteroperabilityError,
        OSError,
        subprocess.SubprocessError,
        ValueError,
    ) as error:
        print(f"SMTP-30 FAIL: {type(error).__name__}: {error}")
        raise SystemExit(1)