#!/usr/bin/env python3 """Adversarial testing for pneumatube, run against your own deployment. Covers (see each test_* function's docstring for detail): - injection fuzzing of the Email-To / subject paths - malformed-MIME resilience of extract_text - per-message failure isolation in the smolmail -> email poll loop - username enumeration via the Postfix lookup service's own protocol - abuse/open-relay behaviour and the accept-gate mitigations for each direction (PNEUMATUBE_REQUIRE_ACCEPTED_SENDER, _RECIPIENT), including their explicit opt-out modes - 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) 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 python scripts/security_test.py uv run python scripts/security_test.py --skip-live # unit-style checks only Needs a `smolmail` checkout next to this repo for the live checks (same default pneumatube itself uses: PNEUMATUBE_SMOLMAIL_REPO, default "../smolmail"). """ from __future__ import annotations import argparse import contextlib import email as email_module import logging import os import subprocess import sys import tempfile import time from dataclasses import dataclass from email.message import EmailMessage from pathlib import Path from aiosmtpd.controller import Controller sys.path.insert(0, str(Path(__file__).resolve().parent)) from stress_test import ( # noqa: E402 REPO_ROOT, CountingSink, SmolAccount, SmolServer, free_port, make_account, send_one_lmtp, send_one_smol, start_smolmaild, wait_for_port, ) sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) from pneumatube import smtp_client, vendor # noqa: E402 from pneumatube.email_to_smol import EmailToSmolBridge, _single_line # noqa: E402 from pneumatube.lookup_server import _lookup # noqa: E402 @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) # --- injection fuzzing (no live infra needed) ----------------------------------- INJECTION_PAYLOADS = [ ("bare-lf", "evil@example.com\nX-Injected: 1"), ("bare-cr", "evil@example.com\rX-Injected: 1"), ("crlf", "evil@example.com\r\nBcc: evil@example.com"), ("nul-byte", "evil@example.com\x00X-Injected: 1"), ("unicode-line-sep", "evil@example.com
X-Injected: 1"), ("unicode-para-sep", "evil@example.com
X-Injected: 1"), ("very-long", "a" * 10_000 + "@example.com"), ] def test_address_validation_fuzz() -> None: """smtp_client.validate() is the only gate between an attacker-supplied Email-To frontmatter value and smtplib.SMTP's RCPT TO -- it must reject every CR/LF variant, not just the literal \\r\\n case the unit tests already cover.""" failures = [] for name, payload in INJECTION_PAYLOADS: try: smtp_client.validate(payload) # unicode-line-sep/para-sep and very-long aren't CR/LF and may # legitimately pass the regex; only bare-lf/bare-cr/crlf/nul-byte # must always be rejected. if name in ("bare-lf", "bare-cr", "crlf", "nul-byte"): failures.append(name) except smtp_client.InvalidAddress: pass record("address validation fuzz", "pass" if not failures else "bug", "all CR/LF/NUL variants rejected" if not failures else f"accepted dangerous payloads: {failures}") def test_single_line_fuzz() -> None: """email_to_smol._single_line() is what stops a forged email Subject from injecting extra smolmail frontmatter fields (SPEC.md S5.5: one `Key: value` per line) into an otherwise-trusted message. (MAIL FROM no longer flows into frontmatter at all -- see email_to_smol.py's module docstring -- but Subject still does.)""" failures = [] for name, payload in INJECTION_PAYLOADS: cleaned = _single_line(payload) if "\n" in cleaned or "\r" in cleaned: failures.append(name) record("frontmatter sanitizer fuzz", "pass" if not failures else "bug", "no payload left a raw CR/LF in the sanitized value" if not failures else f"raw CR/LF survived sanitization: {failures}") def test_outbound_header_injection() -> None: """Even though frontmatter values can't structurally contain a raw newline (smolmail.py's parser splits the body into single lines before a value is ever extracted), smtp_client.send() has no explicit sanitizer of its own -- it currently relies entirely on that upstream guarantee plus Python's EmailMessage rejecting embedded CR/LF. Confirm the second layer actually holds, since the first is an implicit dependency on vendored code outside this project.""" try: m = EmailMessage() m["Subject"] = "hi\r\nBcc: evil@attacker.example" rejected = False except ValueError: rejected = True record("stdlib header-injection backstop", "pass" if rejected else "bug", "EmailMessage rejects embedded CR/LF in header values (confirmed)" if rejected else "EmailMessage accepted an embedded CR/LF -- smtp_client.send() has no " "sanitizer of its own and would forward it verbatim") # --- malformed MIME resilience (no live infra needed) --------------------------- def test_malformed_mime_resilience() -> None: """extract_text() runs on every inbound email before any validation -- it must not crash or hang on adversarial MIME, since a single crash taking down the LMTP handler would be a trivial DoS against every other in-flight recipient in the same message.""" cases = { "empty": b"", "no-body": b"Subject: hi\r\n\r\n", "non-utf8-bytes": b"Subject: hi\r\nContent-Type: text/plain\r\n\r\n\xff\xfe\x00\x01", "multipart-no-text-part": ( b"Content-Type: multipart/mixed; boundary=X\r\n\r\n--X\r\n" b"Content-Type: application/octet-stream\r\n\r\nbinary\r\n--X--\r\n"), "deeply-nested-multipart": _nested_multipart(50), "declared-huge-missing-body": ( b"Content-Type: text/plain\r\nContent-Length: 999999999\r\n\r\nshort"), } failures = [] for name, raw in cases.items(): try: message = email_module.message_from_bytes(raw) t0 = time.monotonic() EmailToSmolBridge.extract_text(message) if time.monotonic() - t0 > 2.0: failures.append(f"{name} (slow: possible algorithmic DoS)") except Exception as exc: failures.append(f"{name} ({exc!r})") record("malformed MIME resilience", "pass" if not failures else "bug", "all cases handled without raising or stalling" if not failures else f"crashed or stalled on: {failures}") def _nested_multipart(depth: int) -> bytes: body = b"leaf" for i in range(depth): boundary = f"B{i}".encode() body = (b"Content-Type: multipart/mixed; boundary=" + boundary + b"\r\n\r\n--" + boundary + b"\r\n" + body + b"\r\n--" + boundary + b"--\r\n") return b"Content-Type: multipart/mixed; boundary=B0\r\n\r\n--B0\r\n" + body + b"\r\n--B0--\r\n" # --- per-message failure isolation (no live infra needed) ---------------------- def test_poll_loop_isolates_failures() -> None: """A message that makes _relay() raise anything other than InvalidAddress/OSError must not block every message behind it in the smolmail -> email direction -- see smol_to_email.py's poll_once() for why (a poisoned message is never deleted, so it sorts first forever).""" from types import SimpleNamespace from pneumatube.smol_to_email import SmolToEmailBridge class StubSmol: SmolError = RuntimeError def cmd_fetch(self, args, store): pass class StubStore: def mail(self, box): return [(b"poison", b"", 0, None), (b"ok", b"", 1, None)] bridge = SmolToEmailBridge(StubSmol(), StubStore(), account=None, config=SimpleNamespace(smolmail_key="k", smolmail_db="d", require_accepted_sender=True)) relayed = [] def fake_relay(mid, envelope): if mid == b"poison": raise ValueError("simulated") relayed.append(mid) bridge._relay = fake_relay # poll_once() logs the simulated failure via logger.exception(), which is # the whole point of the fix -- quiet it here so it doesn't look like an # unhandled crash in this script's own output. logging.getLogger("pneumatube.smol_to_email").setLevel(logging.CRITICAL) try: bridge.poll_once() record("poll loop failure isolation", "pass" if relayed == [b"ok"] else "bug", "later messages still processed after an earlier one raised" if relayed == [b"ok"] else "poll_once stopped after the first failure -- queue-jam DoS") except Exception as exc: record("poll loop failure isolation", "bug", f"poll_once raised instead of isolating the failure: {exc!r}") # --- username enumeration via the lookup service (no live infra needed) -------- def test_lookup_enumeration() -> None: """The Postfix lookup service (lookup_server.py) 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.""" import asyncio class StubBridge: def user_exists(self, username): return username in ("alice", "bob") def recipient_accepted_gateway(self, username): return True class InlineWorker: async def run(self, fn, *args): return fn(*args) from types import SimpleNamespace config = SimpleNamespace(email_domain="gateway.example", require_accepted_recipient=True) bridge, worker = StubBridge(), InlineWorker() # tcp_table(5) status is "200" either way (it means "I have an answer"); # the OK-vs-reject distinction Postfix actually acts on is in the value. results = {} for name in ("alice", "bob", "carol", "admin", "root", "postmaster"): reply = asyncio.run(_lookup(bridge, worker, config, f"{name}@gateway.example")) _code, _, value = reply.partition(" ") results[name] = value == "OK" distinguishable = results["alice"] and results["bob"] 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") # --- live tests: abuse/throttling and sender spoofing --------------------------- def _start_pneumatube(smolmail_repo: Path, gateway: SmolAccount, server: SmolServer, sink_port: int, require_accepted_sender: bool = True, require_accepted_recipient: bool = True ) -> tuple[subprocess.Popen, int]: listen_port = free_port() env = {**os.environ, "PNEUMATUBE_SMOLMAIL_REPO": str(smolmail_repo), "PNEUMATUBE_SMOLMAIL_KEY": str(gateway.key_path), "PNEUMATUBE_SMOLMAIL_DB": str(gateway.db_path), "PNEUMATUBE_SMOLMAIL_DOMAIN": f"{server.host}:{server.port}", "PNEUMATUBE_EMAIL_DOMAIN": "gateway.example", "PNEUMATUBE_FETCH_INTERVAL": "2", "PNEUMATUBE_REQUIRE_ACCEPTED_SENDER": "1" if require_accepted_sender else "0", "PNEUMATUBE_REQUIRE_ACCEPTED_RECIPIENT": "1" if require_accepted_recipient else "0", "PNEUMATUBE_LISTEN_PROTOCOL": "lmtp", "PNEUMATUBE_LISTEN_HOST": "127.0.0.1", "PNEUMATUBE_LISTEN_PORT": str(listen_port), "PNEUMATUBE_LOOKUP_HOST": "127.0.0.1", "PNEUMATUBE_LOOKUP_PORT": str(free_port()), "PNEUMATUBE_SMTP_RELAY_HOST": "127.0.0.1", "PNEUMATUBE_SMTP_RELAY_PORT": str(sink_port), "PNEUMATUBE_SMTP_RELAY_STARTTLS": "0", "PNEUMATUBE_GATEWAY_FROM_ADDRESS": "gateway@gateway.example"} proc = subprocess.Popen(["uv", "run", "pneumatube"], cwd=REPO_ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) wait_for_port("127.0.0.1", listen_port) return proc, listen_port def _stop(proc: subprocess.Popen) -> None: proc.terminate() with contextlib.suppress(Exception): proc.wait(timeout=5) def run_live_tests(smolmail_repo: Path) -> None: with tempfile.TemporaryDirectory(prefix="pneumatube-sectest-") as tmp: tmp_dir = Path(tmp) server = start_smolmaild(smolmail_repo, tmp_dir, rate_connections=6000, rate_sends=6000) smol = vendor.load(smolmail_repo) gateway = make_account(smol, 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() # -- smolmail -> email, default config (require_accepted_sender=True) -- # 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. proc, listen_port = _start_pneumatube(smolmail_repo, gateway, server, sink_port, require_accepted_sender=True) try: 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.py is not holding")) # -- email -> smolmail, default config (require_accepted_recipient=True) -- # "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 handle_RCPT/_lookup is not holding")) finally: _stop(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. proc, listen_port = _start_pneumatube(smolmail_repo, gateway, server, sink_port, require_accepted_sender=False, require_accepted_recipient=False) 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.py forwards no claim about the original sender at # all (see its module docstring) 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) sink_controller.stop() _stop(server.process) def main() -> None: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--skip-live", action="store_true", help="run only the unit-style checks (no subprocess/network)") parser.add_argument("--smolmail-repo", default="../smolmail") args = parser.parse_args() test_address_validation_fuzz() test_single_line_fuzz() test_outbound_header_injection() test_malformed_mime_resilience() test_poll_loop_isolates_failures() test_lookup_enumeration() if not args.skip_live: run_live_tests(Path(args.smolmail_repo).resolve()) 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 __name__ == "__main__": main()