bongo/contrib/testing/protocol-smoke.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

"""Exercise a local Bongo SMTP/IMAP/POP TLS delivery path."""

from __future__ import annotations

import argparse
from email.message import EmailMessage
import imaplib
import os
import poplib
import smtplib
import socket
import ssl
import sys
import time
import urllib.request


def environment(name: str, default: str = "") -> str:
    return os.environ.get(name, default)


HOST = environment("BONGO_TEST_HOST", "127.0.0.1")
USER1 = environment("BONGO_TEST_USER1", "test1")
USER2 = environment("BONGO_TEST_USER2", "test2")
DOMAIN = environment("BONGO_TEST_DOMAIN", "bongo.test")
PASSWORD = environment("BONGO_TEST_PASSWORD")
PORTS = {
    "pop3": int(environment("BONGO_TEST_POP3_PORT", "110")),
    "imap": int(environment("BONGO_TEST_IMAP_PORT", "143")),
    "smtps": int(environment("BONGO_TEST_SMTPS_PORT", "465")),
    "submission": int(environment("BONGO_TEST_SUBMISSION_PORT", "587")),
    "sieve": int(environment("BONGO_TEST_SIEVE_PORT", "689")),
    "imaps": int(environment("BONGO_TEST_IMAPS_PORT", "993")),
    "pop3s": int(environment("BONGO_TEST_POP3S_PORT", "995")),
}
TOKEN = environment("BONGO_TEST_TOKEN", f"bongo-smoke-{int(time.time())}")
TLS = ssl.create_default_context()
TLS.check_hostname = False
TLS.verify_mode = ssl.CERT_NONE


def wait_for_listeners(timeout: float = 30.0) -> str:
    deadline = time.monotonic() + timeout
    pending = set(PORTS.values())
    while pending and time.monotonic() < deadline:
        for port in tuple(pending):
            try:
                with socket.create_connection((HOST, port), timeout=0.25):
                    pending.remove(port)
            except OSError:
                pass
        if pending:
            time.sleep(0.25)
    if pending:
        raise RuntimeError(f"listeners did not become ready: {sorted(pending)}")
    return "all protocol listeners accepted a connection"


def report(name, operation) -> bool:
    try:
        detail = operation()
        print(f"PASS {name}: {detail}")
        return True
    except Exception as error:  # integration test: report every failed leg
        print(f"FAIL {name}: {type(error).__name__}: {error}")
        return False


def web() -> str:
    url = environment("BONGO_TEST_WEB_URL")
    request = urllib.request.Request(url, method="GET")
    with urllib.request.urlopen(request, timeout=10) as response:
        return f"HTTP {response.status}, {response.geturl()}"


def smtp_submit() -> str:
    message = EmailMessage()
    message["From"] = f"{USER1}@{DOMAIN}"
    message["To"] = f"{USER2}@{DOMAIN}"
    message["Subject"] = TOKEN
    message.set_content(f"Bongo local protocol smoke test {TOKEN}\n")
    with smtplib.SMTP(HOST, PORTS["submission"], timeout=15) as client:
        code, _ = client.ehlo()
        if code != 250 or "starttls" not in client.esmtp_features:
            raise RuntimeError("submission does not advertise STARTTLS")
        client.starttls(context=TLS)
        client.ehlo()
        client.login(USER1, PASSWORD)
        refused = client.send_message(message)
        if refused:
            raise RuntimeError(f"recipients refused: {refused}")
    return TOKEN


def smtp_implicit() -> str:
    with smtplib.SMTP_SSL(HOST, PORTS["smtps"], timeout=15,
                          context=TLS) as client:
        code, _ = client.ehlo()
        if code != 250:
            raise RuntimeError(f"EHLO returned {code}")
        client.login(USER1, PASSWORD)
    return "TLS and authentication accepted"


def wait_for_imap() -> str:
    deadline = time.monotonic() + 20
    last = None
    while time.monotonic() < deadline:
        try:
            with imaplib.IMAP4_SSL(HOST, PORTS["imaps"], ssl_context=TLS,
                                   timeout=10) as client:
                client.login(USER2, PASSWORD)
                status, _ = client.select("INBOX", readonly=True)
                if status != "OK":
                    raise RuntimeError("cannot select INBOX")
                status, data = client.search(None, "SUBJECT", f'"{TOKEN}"')
                if status == "OK" and data and data[0].split():
                    identifier = data[0].split()[-1]
                    status, _ = client.fetch(
                        identifier, "(BODY.PEEK[HEADER.FIELDS (SUBJECT)])")
                    if status != "OK":
                        raise RuntimeError("message fetch failed")
                    return f"message {identifier.decode()} found"
                last = data
        except Exception as error:  # delivery can take a few seconds
            last = error
        time.sleep(1)
    raise RuntimeError(f"message did not arrive: {last}")


def imap_starttls() -> str:
    with imaplib.IMAP4(HOST, PORTS["imap"], timeout=15) as client:
        capabilities = set(client.capabilities)
        if "STARTTLS" not in capabilities and b"STARTTLS" not in capabilities:
            raise RuntimeError(f"STARTTLS not advertised: {capabilities!r}")
        client.starttls(ssl_context=TLS)
        client.login(USER2, PASSWORD)
        status, data = client.select("INBOX", readonly=True)
        if status != "OK":
            raise RuntimeError("cannot select INBOX")
        return f"INBOX contains {data[0].decode()} message(s)"


def pop3_starttls() -> str:
    client = poplib.POP3(HOST, PORTS["pop3"], timeout=15)
    try:
        capabilities = client.capa()
        if "STLS" not in capabilities and b"STLS" not in capabilities:
            raise RuntimeError(f"STLS not advertised: {capabilities!r}")
        client.stls(context=TLS)
        client.user(USER2)
        client.pass_(PASSWORD)
        count, size = client.stat()
        return f"maildrop has {count} message(s), {size} bytes"
    finally:
        try:
            client.quit()
        except Exception:
            client.close()


def pop3_implicit() -> str:
    client = poplib.POP3_SSL(HOST, PORTS["pop3s"], timeout=15, context=TLS)
    try:
        client.user(USER2)
        client.pass_(PASSWORD)
        count, size = client.stat()
        return f"maildrop has {count} message(s), {size} bytes"
    finally:
        try:
            client.quit()
        except Exception:
            client.close()


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--wait-only", action="store_true")
    arguments = parser.parse_args()
    if arguments.wait_only:
        return 0 if report("listener readiness", wait_for_listeners) else 1
    if not PASSWORD:
        parser.error("BONGO_TEST_PASSWORD must be set")
    checks = [
        ("listener readiness", wait_for_listeners),
        ("SMTP submission STARTTLS", smtp_submit),
        ("SMTPS submission", smtp_implicit),
        ("IMAPS delivery", wait_for_imap),
        ("IMAP STARTTLS", imap_starttls),
        ("POP3 STLS", pop3_starttls),
        ("POP3S", pop3_implicit),
    ]
    if environment("BONGO_TEST_WEB_URL"):
        checks.append(("Web", web))
    success = True
    for name, operation in checks:
        success = report(name, operation) and success
    return 0 if success else 1


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