bongo/contrib/testing/smtp-telnet-timeout-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 paced manual SMTP input and bounded idle Telnet connections."""

from __future__ import annotations

import copy
import errno
import json
import os
import pty
import select
import shutil
import subprocess
import time


ADMIN = os.environ.get("BONGO_TEST_ADMIN", "/usr/bin/bongo-admin")

def _bongo_admin_argv(*arguments: str) -> list[str]:
    # bongo-admin's IsAdministrativeUser() (src/apps/admin/admin.c) accepts
    # either real root or a euid matching the compiled BONGO_USER -- a
    # live-instance-fixture.py instance compiles that as the invoking
    # user's own name (CTEST_BUILD_NOROOT), so sudo is never actually
    # required there.
    import os as _os
    if _os.geteuid() == 0 or _os.environ.get("BONGO_TEST_ADMIN_UNPRIVILEGED") == "1":
        return [ADMIN, *arguments]
    return ["sudo", "-n", ADMIN, *arguments]

SYSTEMCTL = os.environ.get("BONGO_TEST_SYSTEMCTL", "/usr/bin/systemctl")
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"))
SOCKET_TIMEOUT = int(os.environ.get("BONGO_TEST_SOCKET_TIMEOUT", "3"))
COMMAND_DELAY = float(os.environ.get("BONGO_TEST_COMMAND_DELAY", "2"))
STARTUP_GRACE = float(os.environ.get("BONGO_TEST_STARTUP_GRACE", "2"))
ALLOW_LIVE = os.environ.get("BONGO_ALLOW_LIVE_SMTP_TEST") == "1"


class TelnetTimeoutError(RuntimeError):
    """Raised when an SMTP Telnet session violates timeout policy."""


def run(arguments: list[str], *, input_text: str | None = None) -> str:
    completed = subprocess.run(
        arguments,
        input=input_text,
        capture_output=True,
        text=True,
        check=False,
    )
    if completed.returncode:
        raise TelnetTimeoutError(
            f"{' '.join(arguments)} failed: "
            f"{completed.stderr.strip() or completed.stdout.strip()}")
    return completed.stdout


def admin(*arguments: str, input_text: str | None = None) -> str:
    return run(
        _bongo_admin_argv(*arguments),
        input_text=input_text,
    )


def read_configuration() -> dict:
    try:
        configuration = json.loads(admin("__config-read", "smtp"))
    except json.JSONDecodeError as error:
        raise TelnetTimeoutError(
            "bongo-admin returned invalid SMTP configuration") from error
    if not isinstance(configuration, dict):
        raise TelnetTimeoutError("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 wait_port(host: str, port: int) -> None:
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline:
        master = -1
        process: subprocess.Popen[bytes] | None = None
        try:
            master, process = open_telnet(host, port)
            transcript = bytearray()
            unread = bytearray()
            read_reply(master, 220, transcript, unread, 2)
            os.write(master, b"QUIT\n")
            read_reply(master, 221, transcript, unread, 2)
            process.wait(timeout=2)
            return
        except (OSError, subprocess.SubprocessError, TelnetTimeoutError):
            time.sleep(0.1)
        finally:
            close_telnet(master, process)
    raise TelnetTimeoutError(f"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(INBOUND_HOST, INBOUND_PORT)
    wait_port(INTERNAL_HOST, INTERNAL_PORT)
    if run(
        ["sudo", "-n", SYSTEMCTL, "is-active", "bongo.service"]
    ).strip() != "active":
        raise TelnetTimeoutError("bongo.service is not active")


def open_telnet(
    host: str, port: int,
) -> tuple[int, subprocess.Popen[bytes]]:
    telnet = shutil.which("telnet")
    if telnet is None:
        raise TelnetTimeoutError("required telnet client is not installed")
    master, slave = pty.openpty()
    try:
        process = subprocess.Popen(
            [telnet, "-E", host, str(port)],
            stdin=slave,
            stdout=slave,
            stderr=slave,
            close_fds=True,
        )
    finally:
        os.close(slave)
    return master, process


def close_telnet(
    master: int, process: subprocess.Popen[bytes] | None,
) -> None:
    if process is not None and process.poll() is None:
        process.kill()
        process.wait()
    if master >= 0:
        os.close(master)


def read_reply(
    master: int,
    expected: int,
    transcript: bytearray,
    unread: bytearray,
    timeout: float,
) -> None:
    deadline = time.monotonic() + timeout
    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(prefix):
                return
        ready, _, _ = select.select(
            [master], [], [], max(0.0, deadline - time.monotonic()))
        if not ready:
            break
        try:
            data = os.read(master, 4096)
        except OSError as error:
            if error.errno == errno.EIO:
                break
            raise
        if not data:
            break
        transcript.extend(data)
        unread.extend(data)
    raise TelnetTimeoutError(
        f"Telnet did not receive SMTP {expected}: "
        f"{bytes(transcript).decode('utf-8', 'replace').strip()}")


def wait_for_remote_close(
    master: int,
    process: subprocess.Popen[bytes],
    transcript: bytearray,
) -> float:
    started = time.monotonic()
    deadline = started + SOCKET_TIMEOUT + 5
    while time.monotonic() < deadline:
        ready, _, _ = select.select(
            [master], [], [], max(0.0, deadline - time.monotonic()))
        if ready:
            try:
                data = os.read(master, 4096)
            except OSError as error:
                if error.errno == errno.EIO:
                    data = b""
                else:
                    raise
            if data:
                transcript.extend(data)
                continue
        if process.poll() is not None:
            elapsed = time.monotonic() - started
            if elapsed < max(0.5, SOCKET_TIMEOUT * 0.65):
                raise TelnetTimeoutError(
                    f"connection closed too early after {elapsed:.2f}s: "
                    f"{bytes(transcript).decode('utf-8', 'replace').strip()}")
            return elapsed
    raise TelnetTimeoutError(
        f"idle connection remained open beyond {SOCKET_TIMEOUT + 5}s: "
        f"{bytes(transcript).decode('utf-8', 'replace').strip()}")


def check_idle(host: str, port: int, name: str) -> float:
    master = -1
    process: subprocess.Popen[bytes] | None = None
    try:
        master, process = open_telnet(host, port)
        transcript = bytearray()
        unread = bytearray()
        read_reply(master, 220, transcript, unread, SOCKET_TIMEOUT + 2)
        return wait_for_remote_close(master, process, transcript)
    except (OSError, subprocess.SubprocessError) as error:
        raise TelnetTimeoutError(
            f"{name} idle Telnet check failed: {error}") from error
    finally:
        close_telnet(master, process)


def check_paced(host: str, port: int, name: str) -> float:
    master = -1
    process: subprocess.Popen[bytes] | None = None
    try:
        master, process = open_telnet(host, port)
        transcript = bytearray()
        unread = bytearray()
        read_reply(master, 220, transcript, unread, SOCKET_TIMEOUT + 2)
        for command, response in (
            (f"EHLO manual-{name}.bongo.test\n".encode("ascii"), 250),
            (b"NOOP\n", 250),
            (b"RSET\n", 250),
        ):
            time.sleep(COMMAND_DELAY)
            if process.poll() is not None:
                raise TelnetTimeoutError(
                    f"{name} closed during paced manual input")
            os.write(master, command)
            read_reply(
                master, response, transcript, unread, SOCKET_TIMEOUT + 2)
        return wait_for_remote_close(master, process, transcript)
    except (OSError, subprocess.SubprocessError) as error:
        raise TelnetTimeoutError(
            f"{name} paced Telnet check failed: {error}") from error
    finally:
        close_telnet(master, process)


def main() -> int:
    if not ALLOW_LIVE:
        raise TelnetTimeoutError(
            "set BONGO_ALLOW_LIVE_SMTP_TEST=1 for the live timeout test")
    if SOCKET_TIMEOUT < 2 or SOCKET_TIMEOUT > 15:
        raise TelnetTimeoutError(
            "BONGO_TEST_SOCKET_TIMEOUT must be between 2 and 15 seconds")
    if COMMAND_DELAY <= 0 or COMMAND_DELAY >= SOCKET_TIMEOUT:
        raise TelnetTimeoutError(
            "BONGO_TEST_COMMAND_DELAY must be positive and below the timeout")
    if COMMAND_DELAY * 2 <= SOCKET_TIMEOUT:
        raise TelnetTimeoutError(
            "two command delays must exceed the timeout to prove it resets")

    original = read_configuration()
    test_configuration = copy.deepcopy(original)
    test_configuration["socket_timeout"] = SOCKET_TIMEOUT
    results: list[str] = []
    test_error: BaseException | None = None
    restore_error: BaseException | None = None
    try:
        replace_configuration(test_configuration)
        restart_bongo()
        for name, host, port in (
            ("inbound", INBOUND_HOST, INBOUND_PORT),
            ("internal", INTERNAL_HOST, INTERNAL_PORT),
        ):
            idle = check_idle(host, port, name)
            paced = check_paced(host, port, name)
            results.append(
                f"{name}=idle:{idle:.2f}s,paced-idle:{paced:.2f}s")
    except BaseException as error:
        test_error = error
    finally:
        try:
            replace_configuration(original)
            restart_bongo()
            if read_configuration() != original:
                raise TelnetTimeoutError(
                    "SMTP configuration was not restored exactly")
        except BaseException as error:
            restore_error = error
    if restore_error is not None:
        if test_error is not None:
            raise TelnetTimeoutError(
                f"test failed ({test_error}) and SMTP configuration "
                f"restoration also failed ({restore_error})")
        raise TelnetTimeoutError(
            f"failed to restore SMTP configuration: {restore_error}")
    if test_error is not None:
        raise test_error

    print(
        "SMTP-32 PASS "
        f"configured-timeout={SOCKET_TIMEOUT}s "
        f"command-delay={COMMAND_DELAY:g}s "
        f"sessions={';'.join(results)} "
        "paced-commands=EHLO,NOOP,RSET config-restored=yes service=active"
    )
    return 0


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