pneumatube/scripts/security_test.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["aiosmtpd>=1.4", "pynacl>=1.5", "noiseprotocol>=0.3.1"]
# ///
"""Adversarial testing for pneumatube, run against your own deployment.

Covers (see each check's docstring for detail):
  - abuse/open-relay behaviour and the accept-gate mitigations for each
    direction (PNEUMATUBE_REQUIRE_ACCEPTED_SENDER, _RECIPIENT), including
    their explicit opt-out modes
  - username enumeration via the Postfix lookup service's own protocol
  - sender spoofing: confirms a forged MAIL FROM is never forwarded into
    the relayed smolmail message (email -> smolmail carries no Email-From
    field at all, precisely because MAIL FROM is unauthenticated)

The unit-style checks the Python version used to run here (address
validation fuzz, the frontmatter sanitizer, the CR/LF header backstop,
malformed-MIME resilience, poll-loop failure isolation) live in the Rust
implementation's `cargo test` suite now -- see the tests in
src/smtp_client.rs, src/email_to_smol.rs and src/smol_to_email.rs.

Each check prints PASS (a defense verified), RISK (confirmed, accepted
design tradeoff -- not a bug, but worth knowing it's real), or BUG
(unexpected, worth fixing).

Usage:
    uv run scripts/security_test.py

Needs a `bunshin` and a `fumi` binary on PATH (or --server-bin/--fumi-bin),
a built pneumatube (--pneumatube-bin, default target/release/pneumatube),
and a `smolmail` checkout next to this repo for the load-driving client
(--smolmail-repo, default "../smolmail").
"""
from __future__ import annotations

import argparse
import contextlib
import os
import socket
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import unquote

from aiosmtpd.controller import Controller

from stress_test import (
    REPO_ROOT, CountingSink, free_port, gateway_env,
    load_smolmail, make_account, make_gateway_account, send_one_lmtp,
    send_one_smol, start_bunshin, start_pneumatube, stop_proc,
)


@dataclass
class Finding:
    check: str
    kind: str  # "pass" (defense verified), "risk" (confirmed accepted tradeoff), "bug" (unexpected)
    detail: str

    def print(self) -> None:
        label = {"pass": "PASS", "risk": "RISK", "bug": "BUG "}[self.kind]
        print(f"[{label}] {self.check}: {self.detail}")


FINDINGS: list[Finding] = []


def record(check: str, kind: str, detail: str) -> None:
    f = Finding(check, kind, detail)
    f.print()
    FINDINGS.append(f)


def lookup_get(host: str, port: int, key: str) -> str:
    with socket.create_connection((host, port), timeout=10) as s:
        s.sendall(f"get {key}\n".encode())
        return s.recv(4096).decode().strip()


def run_live_tests(args: argparse.Namespace) -> None:
    smolmail_repo = Path(args.smolmail_repo).resolve()
    smol = load_smolmail(smolmail_repo)
    with tempfile.TemporaryDirectory(prefix="pneumatube-sectest-") as tmp:
        tmp_dir = Path(tmp)
        server = start_bunshin(args.server_bin, tmp_dir, rate_connections=6000, rate_sends=6000)
        gateway = make_gateway_account(args.fumi_bin, tmp_dir, "gatewayuser", server)
        victim = make_account(smol, tmp_dir, "victim", server)
        spoof_victim = make_account(smol, tmp_dir, "spoofvictim", server)

        sink = CountingSink()
        sink_port = free_port()
        sink_controller = Controller(sink, hostname="127.0.0.1", port=sink_port)
        sink_controller.start()

        # -- default config: both accept gates on --
        listen_port, lookup_port = free_port(), free_port()
        env = gateway_env(gateway, server, require_accepted_sender=True,
                          require_accepted_recipient=True)
        proc = start_pneumatube(args.pneumatube_bin, env, listen_port, lookup_port, sink_port)
        try:
            # 15 messages from 15 DIFFERENT, never-before-seen sender
            # identities (no accept token, no prior relationship), each
            # addressed to a different external destination. None should
            # reach the sink.
            senders = [make_account(smol, tmp_dir, f"stranger{i}", server) for i in range(15)]
            for i, sender in enumerate(senders):
                send_one_smol(smol, sender, gateway.address, f"dest{i}@external.example",
                              f"relay test {i}")
            time.sleep(4)  # >= one PNEUMATUBE_FETCH_INTERVAL
            delivered = sink.count()
            record("open relay blocked by default (smolmail -> email)",
                   "pass" if delivered == 0 else "bug",
                   f"{delivered}/{len(senders)} first-contact messages reached the outbound "
                   "sink with PNEUMATUBE_REQUIRE_ACCEPTED_SENDER at its default (1) -- "
                   + ("none did, as intended" if delivered == 0 else
                      "some did; the accept gate in smol_to_email is not holding"))

            # "victim" is registered on the smolmail server (so user_exists
            # passes) but has never accepted the gateway (no token held) --
            # the all-or-nothing gate should reject this live, not queue it.
            outcome, _lat = send_one_lmtp(
                "127.0.0.1", listen_port, "sender@external.example",
                "victim@gateway.example", b"Subject: hi\r\n\r\nbody\r\n")
            record("relay blocked for unaccepted recipient (email -> smolmail)",
                   "pass" if outcome == "rejected" else "bug",
                   f"outcome={outcome} with PNEUMATUBE_REQUIRE_ACCEPTED_RECIPIENT at its "
                   "default (1) for a registered-but-never-accepted recipient -- " +
                   ("rejected live at RCPT TO, as intended" if outcome == "rejected" else
                    "was not rejected; the gate in the LMTP handler and lookup service "
                    "is not holding"))

            # The lookup service answers definitively per username with no
            # authentication and no rate limiting of its own -- by design,
            # it's meant to be reachable only from Postfix on a private
            # network. Confirm that property so the deployment docs'
            # "loopback only" requirement is understood as load-bearing,
            # not cosmetic. Both accept-gate replies ("550 5.7.1 ...") and
            # "550 5.1.1 no such user" reveal which usernames exist, just
            # as an "OK" would; the OK-vs-reject distinction Postfix acts
            # on is a different axis from the exists-or-not distinction an
            # enumerator wants.
            results = {}
            for name in ("victim", "spoofvictim", "carol", "admin", "root", "postmaster"):
                reply = lookup_get("127.0.0.1", lookup_port, f"{name}@gateway.example")
                # The reply's action text is %XX-quoted on the wire; a
                # known username answers anything but "no such user".
                action = unquote(reply.split(" ", 1)[1]) if " " in reply else reply
                results[name] = action != "550 5.1.1 no such user"
            distinguishable = (results["victim"] and results["spoofvictim"]
                               and not any(results[n] for n in
                                           ("carol", "admin", "root", "postmaster")))
            record("lookup service enumeration", "risk" if distinguishable else "bug",
                   "the lookup protocol unambiguously reveals which usernames exist -- "
                   "expected, it must for check_recipient_access to work, but this is "
                   "only safe because the service is loopback-only; exposing it beyond "
                   "Postfix would be a username enumeration oracle" if distinguishable else
                   "lookup responses did not distinguish known from unknown usernames as "
                   "expected -- check_recipient_access would misbehave")
        finally:
            stop_proc(proc)

        # -- both directions, explicit opt-in to the open-bridge mode --
        # Confirms both flags at 0 still do what they document: restore
        # "relay anyone" for an operator who deliberately wants it. victim
        # and spoof_victim never accepted the gateway, so this also doubles
        # as the opt-out confirmation for the recipient gate.
        listen_port, lookup_port = free_port(), free_port()
        env = gateway_env(gateway, server, require_accepted_sender=False,
                          require_accepted_recipient=False)
        proc = start_pneumatube(args.pneumatube_bin, env, listen_port, lookup_port, sink_port)
        try:
            senders = [make_account(smol, tmp_dir, f"openbridge{i}", server) for i in range(15)]
            for i, sender in enumerate(senders):
                send_one_smol(smol, sender, gateway.address, f"dest2-{i}@external.example",
                              f"open bridge test {i}")
            deadline = time.monotonic() + 15
            while sink.count() < len(senders) and time.monotonic() < deadline:
                time.sleep(0.5)
            delivered = sink.count()
            record("open relay via explicit opt-in (smolmail -> email)", "risk",
                   f"{delivered}/{len(senders)} first-contact senders were relayed to "
                   "arbitrary external addresses with PNEUMATUBE_REQUIRE_ACCEPTED_SENDER=0 "
                   "-- confirms the opt-in open-bridge mode works as documented; an operator "
                   "choosing it should understand this is what they're choosing")

            # -- open relay / no throttling (email -> smolmail) --
            ok = 0
            total = 15
            for i in range(total):
                outcome, _lat = send_one_lmtp("127.0.0.1", listen_port,
                                               f"stranger{i}@external.example",
                                               "victim@gateway.example",
                                               f"Subject: relay {i}\r\n\r\nbody {i}\r\n".encode())
                ok += outcome == "ok"
            record("open relay via explicit opt-in (email -> smolmail)", "risk",
                   f"{ok}/{total} messages from {total} distinct external senders were all "
                   "relayed to an unaccepted recipient with PNEUMATUBE_REQUIRE_ACCEPTED_"
                   "RECIPIENT=0 -- confirms the opt-out works as documented; an operator "
                   "choosing it should understand this is what they're choosing. There is "
                   "also no per-sender throttling regardless of this flag's setting.")

            # -- sender spoofing --
            # Own recipient account, untouched by the open-relay tests above,
            # so finding the message isn't a race against unrelated mail.
            # email_to_smol forwards no claim about the original sender at
            # all (see its module docs) specifically because MAIL FROM is
            # unauthenticated; confirm a forged one doesn't leak through
            # into any frontmatter field of the relayed message.
            forged = "security-team@trusted-bank.example"
            outcome, _lat = send_one_lmtp(
                "127.0.0.1", listen_port, forged, "spoofvictim@gateway.example",
                b"Subject: Urgent: verify your account\r\n\r\nClick here...\r\n")
            opened = None
            deadline = time.monotonic() + 15
            account = smol.load_account(argparse.Namespace(key=str(spoof_victim.key_path)),
                                         smol.Store(str(spoof_victim.db_path)))
            while time.monotonic() < deadline and opened is None:
                store = smol.Store(str(spoof_victim.db_path))
                smol.cmd_fetch(argparse.Namespace(key=str(spoof_victim.key_path),
                                                    db=str(spoof_victim.db_path),
                                                    reset=False, keep=False), store)
                rows = store.mail("all")
                if rows:
                    mid, envelope, _at, _r = rows[0]
                    opened = smol.describe(store, envelope, account.keys)
                else:
                    time.sleep(0.5)

            if opened is None:
                record("sender spoofing (no Email-From claim forwarded)", "bug",
                       f"the message never arrived (outcome={outcome}) -- could not confirm")
            else:
                leaked = forged in opened["fields"].values() or "email-from" in opened["fields"]
                record("sender spoofing (no Email-From claim forwarded)",
                       "pass" if not leaked else "bug",
                       "the relayed message carries no field reflecting the forged MAIL "
                       "FROM -- nothing here for a recipient to mistake for a verified "
                       "sender claim" if not leaked else
                       f"the forged MAIL FROM {forged!r} leaked into the relayed message's "
                       f"frontmatter: {opened['fields']}")
        finally:
            stop_proc(proc)
            sink_controller.stop()
            stop_proc(server.process)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__,
                                      formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--smolmail-repo", default="../smolmail")
    parser.add_argument("--server-bin", default="bunshin")
    parser.add_argument("--fumi-bin", default="fumi")
    parser.add_argument("--pneumatube-bin", default=str(REPO_ROOT / "target/release/pneumatube"))
    args = parser.parse_args()

    run_live_tests(args)

    print()
    bugs = [f for f in FINDINGS if f.kind == "bug"]
    risks = [f for f in FINDINGS if f.kind == "risk"]
    print(f"{len(FINDINGS)} check(s): {len(FINDINGS) - len(bugs) - len(risks)} pass, "
          f"{len(risks)} confirmed accepted risk(s), {len(bugs)} bug(s)")
    if bugs:
        raise SystemExit(1)


if __name__ == "__main__":
    main()