bongo/contrib/testing/smtp4dev-fixture.py

#!/usr/bin/env python3
# /****************************************************************************
#  * <Novell-copyright>
#  * Copyright (c) 2001 Novell, Inc. All Rights Reserved.
#  *
#  * This program is free software; you can redistribute it and/or
#  * modify it under the terms of version 2 of the GNU General Public License
#  * as published by the Free Software Foundation.
#  *
#  * This program is distributed in the hope that it will be useful,
#  * but WITHOUT ANY WARRANTY; without even the implied warranty of
#  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  * GNU General Public License for more details.
#  *
#  * You should have received a copy of the GNU General Public License
#  * along with this program; if not, contact Novell, Inc.
#  *
#  * To contact Novell about this file by physical or electronic mail, you
#  * may find current contact information at www.novell.com.
#  * </Novell-copyright>
#  ****************************************************************************/

"""Manage loopback-only authenticated smtp4dev provider fixtures."""

from __future__ import annotations

import argparse
from email.message import EmailMessage
import hashlib
import json
import os
from pathlib import Path
import shutil
import signal
import smtplib
import socket
import ssl
import stat
import subprocess
import sys
import time
import urllib.request
import zipfile


DEFAULT_ROOT = Path("/tmp/bongo-smtp4dev-fixture")
DEFAULT_BINARY = "Rnwood.Smtp4dev"
DEFAULT_USERNAME = "provider"
DEFAULT_PASSWORD = "BongoProvider-Test-2026"
FIXTURE_HOSTNAME = "smtp-provider.bongo.test"
PORTS = {
    "starttls": 12587,
    "smtps": 12465,
    "starttls_api": 15080,
    "smtps_api": 15081,
}
PINNED_ARCHIVE = (
    "Rnwood.Smtp4dev-linux-x64-3.16.0-ci20260724100_pr2100.zip"
)
PINNED_SHA256 = (
    "4bcff7d9ecbb62d3abe44888ba38502674b9a634c6b87453e43f1af272bd4249"
)
CERTTOOL = "/usr/bin/certtool"


def checked_root(value: str) -> Path:
    root = Path(value).resolve()
    if root == Path("/") or not str(root).startswith("/tmp/bongo-smtp4dev-"):
        raise argparse.ArgumentTypeError(
            "fixture root must be below /tmp and start with bongo-smtp4dev-"
        )
    return root


def paths(root: Path) -> dict[str, Path]:
    return {
        "tool": root / "tool",
        "binary": root / "tool" / DEFAULT_BINARY,
        "run": root / "run",
        "tls": root / "tls",
        "ca_key": root / "tls" / "ca.key",
        "ca_cert": root / "tls" / "ca.crt",
        "ca_template": root / "tls" / "ca.tmpl",
        "server_key": root / "tls" / "server.key",
        "server_cert": root / "tls" / "server.crt",
        "server_template": root / "tls" / "server.tmpl",
        "dotnet": root / "dotnet",
        "config": root / "config",
    }


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        for block in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def extract_archive(
    archive: Path,
    destination: Path,
    expected_sha256: str | None,
) -> None:
    if archive.name == PINNED_ARCHIVE:
        expected_sha256 = PINNED_SHA256
    if expected_sha256 is None:
        raise RuntimeError(
            "an unpinned smtp4dev archive requires --sha256"
        )
    actual = sha256(archive)
    if actual.lower() != expected_sha256.lower():
        raise RuntimeError(
            f"smtp4dev archive digest mismatch: {actual}"
        )
    destination.mkdir(parents=True, exist_ok=True)
    destination_root = destination.resolve()
    with zipfile.ZipFile(archive) as bundle:
        for member in bundle.infolist():
            target = (destination / member.filename).resolve()
            if target != destination_root and destination_root not in target.parents:
                raise RuntimeError(
                    f"unsafe smtp4dev archive path: {member.filename!r}"
                )
            mode = member.external_attr >> 16
            if stat.S_ISLNK(mode):
                raise RuntimeError(
                    f"smtp4dev archive contains symlink: {member.filename!r}"
                )
        bundle.extractall(destination)


def run_certtool(arguments: list[str]) -> None:
    completed = subprocess.run(
        [CERTTOOL, *arguments],
        check=False,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if completed.returncode != 0:
        detail = completed.stderr.decode("utf-8", "replace").strip()
        raise RuntimeError(f"certtool failed: {detail}")


def generate_certificate(root: Path) -> None:
    values = paths(root)
    values["ca_template"].write_text(
        """cn = Bongo smtp4dev Fixture CA
expiration_days = 3
ca
cert_signing_key
crl_signing_key
""",
        encoding="ascii",
    )
    values["server_template"].write_text(
        f"""cn = {FIXTURE_HOSTNAME}
dns_name = {FIXTURE_HOSTNAME}
dns_name = localhost
ip_address = 127.0.0.1
expiration_days = 2
signing_key
encryption_key
tls_www_server
""",
        encoding="ascii",
    )
    run_certtool(
        [
            "--generate-privkey",
            "--sec-param",
            "high",
            "--outfile",
            str(values["ca_key"]),
        ]
    )
    run_certtool(
        [
            "--generate-self-signed",
            "--load-privkey",
            str(values["ca_key"]),
            "--template",
            str(values["ca_template"]),
            "--outfile",
            str(values["ca_cert"]),
        ]
    )
    run_certtool(
        [
            "--generate-privkey",
            "--sec-param",
            "high",
            "--outfile",
            str(values["server_key"]),
        ]
    )
    run_certtool(
        [
            "--generate-certificate",
            "--load-privkey",
            str(values["server_key"]),
            "--load-ca-certificate",
            str(values["ca_cert"]),
            "--load-ca-privkey",
            str(values["ca_key"]),
            "--template",
            str(values["server_template"]),
            "--outfile",
            str(values["server_cert"]),
        ]
    )
    os.chmod(values["ca_key"], 0o600)
    os.chmod(values["server_key"], 0o600)
    os.chmod(values["ca_cert"], 0o644)
    os.chmod(values["server_cert"], 0o644)


def prepare(
    root: Path,
    archive: Path | None,
    archive_sha256: str | None = None,
) -> None:
    values = paths(root)
    for name in ("run", "tls", "dotnet", "config"):
        values[name].mkdir(parents=True, exist_ok=True)
    if archive is not None:
        extract_archive(
            archive.resolve(),
            values["tool"],
            archive_sha256,
        )
    binary = Path(
        os.environ.get("SMTP4DEV_BIN", str(values["binary"]))
    ).resolve()
    if not binary.is_file():
        raise RuntimeError(
            "smtp4dev binary not found; pass --archive or set SMTP4DEV_BIN"
        )
    os.chmod(binary, os.stat(binary).st_mode | stat.S_IXUSR)
    if not values["ca_cert"].is_file() or not values["server_key"].is_file():
        if not Path(CERTTOOL).is_file():
            raise RuntimeError(f"certtool not found at {CERTTOOL}")
        generate_certificate(root)
    print(f"prepared smtp4dev fixture in {root}")
    print(f"CA certificate: {values['ca_cert']}")


def instance_paths(root: Path, name: str) -> tuple[Path, Path, Path]:
    return (
        root / name,
        root / "run" / f"{name}.pid",
        root / "run" / f"{name}.log",
    )


def read_pid(pid_path: Path) -> int | None:
    try:
        pid = int(pid_path.read_text(encoding="ascii").strip())
        os.kill(pid, 0)
        command_line = Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")
        name = pid_path.stem
        expected_data = (
            f"--baseappdatapath={pid_path.parent.parent / name}".encode()
        )
        if (
            not command_line
            or Path(os.fsdecode(command_line[0])).name != DEFAULT_BINARY
            or expected_data not in command_line
        ):
            return None
        return pid
    except (
        FileNotFoundError,
        ProcessLookupError,
        PermissionError,
        ValueError,
    ):
        return None


def wait_for_port(port: int, timeout: float = 20.0) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.25):
                return
        except OSError:
            time.sleep(0.1)
    raise RuntimeError(f"smtp4dev did not open 127.0.0.1:{port}")


def command(
    root: Path,
    name: str,
    tls_mode: str,
    smtp_port: int,
    api_port: int,
    username: str,
    password: str,
) -> tuple[list[str], dict[str, str]]:
    values = paths(root)
    data, _, _ = instance_paths(root, name)
    binary = Path(
        os.environ.get("SMTP4DEV_BIN", str(values["binary"]))
    ).resolve()
    arguments = [
        str(binary),
        f"--baseappdatapath={data}",
        f"--installpath={binary.parent}",
        f"--hostname={FIXTURE_HOSTNAME}",
        "--bindaddress=127.0.0.1",
        "--disableipv6",
        f"--smtpport={smtp_port}",
        "--db=messages.db",
        "--messagestokeep=1000",
        "--sessionstokeep=1000",
        f"--tlsmode={tls_mode}",
        f"--tlscertificate={values['server_cert']}",
        f"--tlscertificateprivatekey={values['server_key']}",
        "--authenticationrequired",
        "--secureconnectionrequired",
        "--smtpallowanycredentials-",
        "--SmtpAuthTypesSecure=PLAIN,LOGIN",
        "--SmtpAuthTypesNotSecure=",
        "--imapport=",
        "--pop3port=",
        "--maxmessagesize=33554432",
        "--locksettings",
        "--nousersettings",
        f"--urls=http://127.0.0.1:{api_port}",
    ]
    environment = os.environ.copy()
    environment.update(
        {
            "DOTNET_BUNDLE_EXTRACT_BASE_DIR": str(values["dotnet"]),
            "XDG_CONFIG_HOME": str(values["config"]),
            "ServerOptions__Users__0__Username": username,
            "ServerOptions__Users__0__Password": password,
        }
    )
    return arguments, environment


def start_instance(
    root: Path,
    name: str,
    tls_mode: str,
    smtp_port: int,
    api_port: int,
    username: str,
    password: str,
) -> None:
    data, pid_path, log_path = instance_paths(root, name)
    if read_pid(pid_path) is not None:
        raise RuntimeError(f"smtp4dev {name} fixture is already running")
    data.mkdir(parents=True, exist_ok=True)
    arguments, environment = command(
        root, name, tls_mode, smtp_port, api_port, username, password
    )
    with log_path.open("ab") as log:
        process = subprocess.Popen(
            arguments,
            stdin=subprocess.DEVNULL,
            stdout=log,
            stderr=subprocess.STDOUT,
            env=environment,
            start_new_session=True,
        )
    pid_path.write_text(f"{process.pid}\n", encoding="ascii")
    try:
        wait_for_port(smtp_port)
        wait_for_port(api_port)
    except Exception:
        stop_instance(root, name)
        raise


def start(root: Path, username: str, password: str) -> None:
    prepare(root, None)
    started: list[str] = []
    try:
        start_instance(
            root,
            "starttls",
            "StartTls",
            PORTS["starttls"],
            PORTS["starttls_api"],
            username,
            password,
        )
        started.append("starttls")
        start_instance(
            root,
            "smtps",
            "ImplicitTls",
            PORTS["smtps"],
            PORTS["smtps_api"],
            username,
            password,
        )
        started.append("smtps")
    except Exception:
        for name in reversed(started):
            stop_instance(root, name)
        raise
    print(
        "started smtp4dev fixtures: "
        f"STARTTLS=127.0.0.1:{PORTS['starttls']} "
        f"SMTPS=127.0.0.1:{PORTS['smtps']}"
    )


def stop_instance(root: Path, name: str) -> None:
    _, pid_path, _ = instance_paths(root, name)
    pid = read_pid(pid_path)
    if pid is None:
        pid_path.unlink(missing_ok=True)
        return
    try:
        os.killpg(pid, signal.SIGTERM)
    except ProcessLookupError:
        pass
    deadline = time.monotonic() + 10.0
    while time.monotonic() < deadline:
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            break
        time.sleep(0.1)
    else:
        try:
            os.killpg(pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
    pid_path.unlink(missing_ok=True)


def stop(root: Path) -> None:
    for name in ("smtps", "starttls"):
        stop_instance(root, name)
    print("stopped smtp4dev fixtures")


def tls_context(root: Path) -> ssl.SSLContext:
    context = ssl.create_default_context(cafile=str(paths(root)["ca_cert"]))
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    return context


def authenticate(
    root: Path,
    mode: str,
    username: str,
    password: str,
) -> str:
    context = tls_context(root)
    if mode == "starttls":
        with smtplib.SMTP("127.0.0.1", PORTS[mode], timeout=10) as client:
            client.ehlo()
            if "auth" in client.esmtp_features:
                raise RuntimeError("smtp4dev advertised AUTH before STARTTLS")
            if "starttls" not in client.esmtp_features:
                raise RuntimeError("smtp4dev did not advertise STARTTLS")
            client.starttls(context=context)
            client.ehlo()
            client.login(username, password)
            return client.sock.version()
    with smtplib.SMTP_SSL(
        "127.0.0.1",
        PORTS[mode],
        timeout=10,
        context=context,
    ) as client:
        client.ehlo()
        client.login(username, password)
        return client.sock.version()


def api_request(
    mode: str,
    path: str,
    method: str = "GET",
) -> bytes:
    request = urllib.request.Request(
        f"http://127.0.0.1:{PORTS[mode + '_api']}{path}",
        method=method,
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        return response.read()


def message_summaries(mode: str) -> list[dict[str, object]]:
    document = json.loads(
        api_request(mode, "/api/messages?pageSize=1000").decode("utf-8")
    )
    if isinstance(document, dict):
        results = document.get("results")
    else:
        results = document
    if not isinstance(results, list):
        raise RuntimeError(f"{mode} returned malformed message list")
    return results


def submit_capture(
    root: Path,
    mode: str,
    username: str,
    password: str,
    token: str,
) -> None:
    message = EmailMessage()
    message["From"] = f"{username}@{FIXTURE_HOSTNAME}"
    message["To"] = "capture@example.test"
    message["Subject"] = token
    message.set_content(f"Local Bongo smtp4dev fixture capture {token}\n")
    context = tls_context(root)
    if mode == "starttls":
        with smtplib.SMTP("127.0.0.1", PORTS[mode], timeout=10) as client:
            client.ehlo()
            client.starttls(context=context)
            client.ehlo()
            client.login(username, password)
            client.send_message(message)
    else:
        with smtplib.SMTP_SSL(
            "127.0.0.1",
            PORTS[mode],
            timeout=10,
            context=context,
        ) as client:
            client.ehlo()
            client.login(username, password)
            client.send_message(message)


def verify_capture(
    root: Path,
    mode: str,
    username: str,
    password: str,
) -> None:
    token = f"bongo-smtp4dev-{mode}-{time.time_ns()}"
    submit_capture(root, mode, username, password, token)
    deadline = time.monotonic() + 10.0
    found: dict[str, object] | None = None
    while time.monotonic() < deadline:
        found = next(
            (
                item
                for item in message_summaries(mode)
                if item.get("subject") == token
            ),
            None,
        )
        if found is not None:
            break
        time.sleep(0.1)
    if found is None or not isinstance(found.get("id"), str):
        raise RuntimeError(f"{mode} did not expose the captured message")
    identifier = found["id"]
    source = api_request(mode, f"/api/messages/{identifier}/source")
    if token.encode("ascii") not in source:
        raise RuntimeError(f"{mode} raw captured message omitted its token")
    api_request(mode, f"/api/messages/{identifier}", method="DELETE")


def verify(root: Path, username: str, password: str) -> None:
    versions = {
        mode: authenticate(root, mode, username, password)
        for mode in ("starttls", "smtps")
    }
    wrong = password + "-wrong"
    for mode in ("starttls", "smtps"):
        try:
            authenticate(root, mode, username, wrong)
        except smtplib.SMTPAuthenticationError as error:
            if error.smtp_code != 535:
                raise RuntimeError(
                    f"{mode} rejected invalid credentials with "
                    f"SMTP {error.smtp_code}, expected 535"
                ) from error
        else:
            raise RuntimeError(f"{mode} accepted invalid credentials")
    for mode in ("starttls", "smtps"):
        verify_capture(root, mode, username, password)
    print(
        "smtp4dev fixture PASS "
        f"starttls={PORTS['starttls']}/{versions['starttls']} "
        f"smtps={PORTS['smtps']}/{versions['smtps']} "
        "auth=valid invalid=535 capture-api=yes"
    )


def clear(root: Path) -> None:
    del root
    for mode in ("starttls", "smtps"):
        for message in message_summaries(mode):
            identifier = message.get("id")
            if isinstance(identifier, str):
                api_request(
                    mode,
                    f"/api/messages/{identifier}",
                    method="DELETE",
                )
    print("cleared smtp4dev messages")


def status(root: Path) -> int:
    running = True
    for name in ("starttls", "smtps"):
        _, pid_path, log_path = instance_paths(root, name)
        pid = read_pid(pid_path)
        print(
            f"{name}: "
            f"{'running pid=' + str(pid) if pid is not None else 'stopped'} "
            f"log={log_path}"
        )
        running = running and pid is not None
    return 0 if running else 1


def reset(root: Path) -> None:
    stop(root)
    if root.exists():
        shutil.rmtree(root)
    print(f"removed {root}")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=checked_root, default=DEFAULT_ROOT)
    parser.add_argument(
        "--username",
        default=os.environ.get("BONGO_SMTP4DEV_USER", DEFAULT_USERNAME),
    )
    parser.add_argument(
        "--password",
        default=os.environ.get("BONGO_SMTP4DEV_PASSWORD", DEFAULT_PASSWORD),
    )
    subparsers = parser.add_subparsers(dest="action", required=True)
    prepare_parser = subparsers.add_parser("prepare")
    prepare_parser.add_argument("--archive", type=Path)
    prepare_parser.add_argument("--sha256")
    subparsers.add_parser("start")
    subparsers.add_parser("stop")
    subparsers.add_parser("status")
    subparsers.add_parser("verify")
    subparsers.add_parser("clear")
    subparsers.add_parser("reset")
    arguments = parser.parse_args()

    if arguments.action == "prepare":
        prepare(arguments.root, arguments.archive, arguments.sha256)
    elif arguments.action == "start":
        start(arguments.root, arguments.username, arguments.password)
    elif arguments.action == "stop":
        stop(arguments.root)
    elif arguments.action == "status":
        return status(arguments.root)
    elif arguments.action == "verify":
        verify(arguments.root, arguments.username, arguments.password)
    elif arguments.action == "clear":
        clear(arguments.root)
    elif arguments.action == "reset":
        reset(arguments.root)
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (
        OSError,
        RuntimeError,
        smtplib.SMTPException,
        subprocess.CalledProcessError,
        zipfile.BadZipFile,
    ) as error:
        print(f"smtp4dev-fixture: {error}", file=sys.stderr)
        raise SystemExit(1)