Add a signed DANE and MTA-STS live fixture

contrib/testing/README.md

@@ -65,9 +65,19 @@ one-off files in `/tmp`:
   globally required TLS never falls back after absent or failed STARTTLS. A
   second phase injects RFC 8689 `REQUIRETLS` over an encrypted trusted
   listener and verifies its Queue trace header, outbound `MAIL FROM`
-  parameter, encrypted delivery to a capable peer, and deferral when a peer
-  lacks TLS or `REQUIRETLS`. It has the same resolver, Queue, configuration,
-  and service restoration guarantees as the opportunistic check.
+  parameter, encrypted delivery to a capable peer, and non-delivery when a
+  peer lacks TLS or `REQUIRETLS`. It has the same resolver, Queue,
+  configuration, and service restoration guarantees as the opportunistic
+  check.
+- `smtp-dane-mta-sts-check.py` creates a disposable ECDSA-signed DNS zone,
+  authoritative BIND instance, validating Unbound resolver, HTTPS policy
+  host, and eight loopback MX fixtures. It proves DANE-EE delivery with a
+  certificate outside the Web PKI, TLSA mismatch deferral, DANE precedence,
+  MTA-STS enforce/testing behavior, exact and wildcard MX policy matching,
+  and durable cached-policy use while HTTPS is unavailable. The root-only
+  fixture backs up and atomically restores `resolv.conf`, the DNS trust-anchor
+  file, SMTP Store configuration, Queue entries, generated certificates, and
+  the original service state.
 - `sendmail-local-submission-check.py` runs as an unprivileged Unix account
   and verifies both `/usr/sbin/sendmail` and GNU `mail` through Bongo's
   restricted, rate-limited local SMTP submission path. It confirms that no

contrib/testing/smtp-dane-mta-sts-check.py

@@ -0,0 +1,853 @@
+#!/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 DANE and MTA-STS with a disposable signed DNS zone."""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import copy
+from dataclasses import dataclass, field
+import hashlib
+import http.server
+import importlib.util
+import os
+from pathlib import Path
+import pwd
+import shutil
+import signal
+import smtplib
+import socket
+import socketserver
+import ssl
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+
+TOKEN = f"smtp27-{os.getpid()}-{int(time.time())}"
+ZONE = "smtp27.test"
+DNS_RESOLVER = "127.0.0.53"
+DNS_AUTHORITY = "127.0.0.54"
+DNS_AUTHORITY_PORT = 5353
+HTTPS_ADDRESS = "127.0.0.9"
+TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "75"))
+STARTUP_SETTLE = float(os.environ.get("BONGO_TEST_STARTUP_SETTLE", "3"))
+RESOLV_BACKUP = Path("/run/bongo-smtp27-resolv.conf.backup")
+ANCHOR_FILE = Path("/etc/dnssec/root-anchors.txt")
+ANCHOR_BACKUP = Path("/run/bongo-smtp27-root-anchors.backup")
+PKI_ROOT = Path("/var/lib/bongo-test-pki")
+PKI_TOOL = Path(__file__).with_name("local-test-pki.py")
+WORK_ROOT = Path("/var/lib/bongo/work")
+
+
+class SMTP27Error(RuntimeError):
+    """Raised when a DANE or MTA-STS invariant is violated."""
+
+
+def load_smtp07_helpers():
+    path = Path(__file__).with_name(
+        "smtp-outbound-opportunistic-tls-check.py")
+    specification = importlib.util.spec_from_file_location(
+        "bongo_smtp27_helpers", path)
+    if specification is None or specification.loader is None:
+        raise SMTP27Error(f"cannot load fixture helpers from {path}")
+    module = importlib.util.module_from_spec(specification)
+    sys.modules[specification.name] = module
+    specification.loader.exec_module(module)
+    module.TOKEN = TOKEN
+    module.RESOLV_BACKUP = RESOLV_BACKUP
+    return module
+
+
+SMTP07 = load_smtp07_helpers()
+
+
+def run(
+    arguments: list[str | Path],
+    *,
+    cwd: Path | None = None,
+    input_text: str | None = None,
+    check: bool = True,
+) -> subprocess.CompletedProcess[str]:
+    command = [str(argument) for argument in arguments]
+    completed = subprocess.run(
+        command,
+        cwd=cwd,
+        input=input_text,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if check and completed.returncode:
+        raise SMTP27Error(
+            f"{' '.join(command)} failed: "
+            f"{completed.stderr.strip() or completed.stdout.strip()}")
+    return completed
+
+
+@dataclass(frozen=True)
+class Route:
+    name: str
+    address: str
+    tls_kind: str | None
+    sts_mode: str | None = None
+    sts_mx: str | None = None
+    tlsa: str | None = None
+
+    @property
+    def domain(self) -> str:
+        return f"{self.name}.{ZONE}"
+
+    @property
+    def mx(self) -> str:
+        return f"mx-{self.name}.{ZONE}"
+
+    @property
+    def policy_host(self) -> str:
+        return f"mta-sts.{self.domain}"
+
+
+ROUTES = (
+    Route("dane-good", "127.0.0.2", "untrusted", tlsa="valid"),
+    Route("dane-bad", "127.0.0.3", "untrusted", tlsa="invalid"),
+    Route("sts-enforce", "127.0.0.4", "trusted",
+          sts_mode="enforce"),
+    Route("sts-testing", "127.0.0.5", None,
+          sts_mode="testing"),
+    Route("sts-mismatch", "127.0.0.6", "trusted",
+          sts_mode="enforce", sts_mx="other.smtp27.test"),
+    Route("dane-precedence", "127.0.0.7", "trusted",
+          sts_mode="enforce", tlsa="invalid"),
+    Route("sts-cache", "127.0.0.8", "trusted",
+          sts_mode="enforce", sts_mx="*.smtp27.test"),
+    Route("sts-plain", "127.0.0.10", None,
+          sts_mode="enforce"),
+)
+
+
+@dataclass
+class Capture:
+    route: Route
+    tls_context: ssl.SSLContext | None
+    lock: threading.Lock = field(default_factory=threading.Lock)
+    connections: int = 0
+    starttls: int = 0
+    messages: list[tuple[bool, bytes]] = field(default_factory=list)
+
+    def connected(self) -> None:
+        with self.lock:
+            self.connections += 1
+
+    def tls_started(self) -> None:
+        with self.lock:
+            self.starttls += 1
+
+    def delivered(self, encrypted: bool, message: bytes) -> None:
+        with self.lock:
+            self.messages.append((encrypted, message))
+
+    def snapshot(self) -> tuple[int, int, list[tuple[bool, bytes]]]:
+        with self.lock:
+            return self.connections, self.starttls, list(self.messages)
+
+
+class SMTPHandler(socketserver.BaseRequestHandler):
+    def handle(self) -> None:
+        capture: Capture = self.server.capture  # type: ignore[attr-defined]
+        capture.connected()
+        connection = self.request
+        connection.settimeout(15)
+        encrypted = False
+        data_mode = False
+        message = bytearray()
+        buffer = bytearray()
+
+        def send(line: bytes) -> None:
+            connection.sendall(line + b"\r\n")
+
+        def read_line() -> bytes | None:
+            while b"\n" not in buffer:
+                chunk = connection.recv(8192)
+                if not chunk:
+                    return None
+                buffer.extend(chunk)
+                if len(buffer) > 4 * 1024 * 1024:
+                    raise SMTP27Error("fixture SMTP input exceeded limit")
+            end = buffer.index(b"\n") + 1
+            line = bytes(buffer[:end])
+            del buffer[:end]
+            return line.rstrip(b"\r\n")
+
+        send(f"220 {capture.route.mx} ESMTP smtp27".encode("ascii"))
+        while True:
+            try:
+                line = read_line()
+            except (OSError, SMTP27Error):
+                return
+            if line is None:
+                return
+            if data_mode:
+                if line == b".":
+                    capture.delivered(encrypted, bytes(message))
+                    send(b"250 2.0.0 captured")
+                    data_mode = False
+                    message.clear()
+                else:
+                    if line.startswith(b".."):
+                        line = line[1:]
+                    message.extend(line + b"\r\n")
+                continue
+
+            command, _, _argument = line.partition(b" ")
+            command = command.upper()
+            if command in (b"EHLO", b"HELO"):
+                if command == b"EHLO":
+                    send(f"250-{capture.route.mx}".encode("ascii"))
+                    if not encrypted and capture.tls_context is not None:
+                        send(b"250-STARTTLS")
+                    send(b"250 SIZE 10485760")
+                else:
+                    send(f"250 {capture.route.mx}".encode("ascii"))
+            elif command == b"STARTTLS":
+                capture.tls_started()
+                if encrypted or capture.tls_context is None:
+                    send(b"454 4.7.0 TLS unavailable")
+                    continue
+                send(b"220 2.0.0 begin TLS")
+                try:
+                    connection = capture.tls_context.wrap_socket(
+                        connection, server_side=True)
+                    connection.settimeout(15)
+                    buffer.clear()
+                    encrypted = True
+                except ssl.SSLError:
+                    return
+            elif command == b"MAIL":
+                send(b"250 2.1.0 sender accepted")
+            elif command == b"RCPT":
+                send(b"250 2.1.5 recipient accepted")
+            elif command == b"DATA":
+                send(b"354 end with <CRLF>.<CRLF>")
+                data_mode = True
+                message.clear()
+            elif command == b"RSET":
+                data_mode = False
+                message.clear()
+                send(b"250 2.0.0 reset")
+            elif command == b"NOOP":
+                send(b"250 2.0.0 ok")
+            elif command == b"QUIT":
+                send(b"221 2.0.0 bye")
+                return
+            else:
+                send(b"502 5.5.1 command not implemented")
+
+
+class ThreadingTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+
+class PolicyServer(http.server.ThreadingHTTPServer):
+    allow_reuse_address = True
+
+
+class PolicyHandler(http.server.BaseHTTPRequestHandler):
+    server_version = "BongoSMTP27/1"
+
+    def do_GET(self) -> None:
+        host = self.headers.get("Host", "").split(":", 1)[0].lower()
+        policy = self.server.policies.get(host)  # type: ignore[attr-defined]
+        if self.path != "/.well-known/mta-sts.txt" or policy is None:
+            self.send_error(404)
+            return
+        self.server.requests.append(host)  # type: ignore[attr-defined]
+        payload = policy.encode("utf-8")
+        self.send_response(200)
+        self.send_header("Content-Type", "text/plain; charset=utf-8")
+        self.send_header("Content-Length", str(len(payload)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def log_message(self, _format: str, *_arguments) -> None:
+        return
+
+
+def certificate_context(directory: Path) -> ssl.SSLContext:
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.load_cert_chain(directory / "fullchain.pem",
+                            directory / "key.pem")
+    return context
+
+
+def certificate_sha256(certificate: Path) -> str:
+    pem = certificate.read_text(encoding="ascii")
+    der = ssl.PEM_cert_to_DER_cert(pem)
+    return hashlib.sha256(der).hexdigest()
+
+
+def issue_certificates() -> tuple[Path, Path, list[Path]]:
+    trusted_name = f"smtp27-trusted-{os.getpid()}"
+    untrusted_name = f"smtp27-untrusted-{os.getpid()}"
+    trusted_names = [
+        route.mx for route in ROUTES if route.tls_kind == "trusted"
+    ] + [
+        route.policy_host for route in ROUTES if route.sts_mode is not None
+    ]
+    untrusted_names = [
+        route.mx for route in ROUTES if route.tls_kind == "untrusted"
+    ]
+    commands = (
+        ["issue", "--ca", "trusted", "--name", trusted_name],
+        ["issue", "--ca", "untrusted", "--name", untrusted_name],
+    )
+    for command, names in zip(commands, (trusted_names, untrusted_names)):
+        arguments: list[str | Path] = [PKI_TOOL, *command]
+        for name in names:
+            arguments.extend(["--dns-name", name])
+        run(arguments)
+    trusted = PKI_ROOT / "issued" / trusted_name
+    untrusted = PKI_ROOT / "issued" / untrusted_name
+    return trusted, untrusted, [trusted, untrusted]
+
+
+def write_zone(directory: Path, valid_tlsa: str) -> str:
+    lines = [
+        f"$ORIGIN {ZONE}.",
+        "$TTL 30",
+        (f"@ IN SOA ns.{ZONE}. hostmaster.{ZONE}. "
+         f"( {int(time.time())} 30 30 300 30 )"),
+        f"  IN NS ns.{ZONE}.",
+        f"ns IN A {DNS_AUTHORITY}",
+    ]
+    for route in ROUTES:
+        lines.extend([
+            f"{route.name} IN MX 10 {route.mx}.",
+            f"mx-{route.name} IN A {route.address}",
+        ])
+        if route.sts_mode is not None:
+            lines.extend([
+                (f'_mta-sts.{route.name} IN TXT '
+                 f'"v=STSv1; id={TOKEN.replace("-", "")};"'),
+                f"mta-sts.{route.name} IN A {HTTPS_ADDRESS}",
+            ])
+        if route.tlsa is not None:
+            digest = valid_tlsa if route.tlsa == "valid" else "00" * 32
+            lines.append(
+                f"_25._tcp.mx-{route.name} IN TLSA 3 0 1 {digest}")
+    (directory / "zone").write_text("\n".join(lines) + "\n",
+                                    encoding="ascii")
+    run([
+        "dnssec-keygen", "-q", "-K", ".", "-a", "ECDSAP256SHA256",
+        "-n", "ZONE", ZONE,
+    ], cwd=directory)
+    run([
+        "dnssec-keygen", "-q", "-K", ".", "-a", "ECDSAP256SHA256",
+        "-f", "KSK", "-n", "ZONE", ZONE,
+    ], cwd=directory)
+    run([
+        "dnssec-signzone", "-S", "-K", ".", "-o", ZONE,
+        "-f", "zone.signed", "zone",
+    ], cwd=directory)
+    for key_file in directory.glob(f"K{ZONE}.*.key"):
+        for line in key_file.read_text(encoding="ascii").splitlines():
+            if " DNSKEY 257 " in line:
+                return " ".join(line.split())
+    raise SMTP27Error("dnssec-keygen did not produce a KSK trust anchor")
+
+
+def write_dns_configs(directory: Path, trust_anchor: str) -> None:
+    (directory / "named.conf").write_text(
+        "options {\n"
+        f'  directory "{directory}";\n'
+        f"  listen-on port {DNS_AUTHORITY_PORT} {{ {DNS_AUTHORITY}; }};\n"
+        "  listen-on-v6 { none; };\n"
+        "  recursion no;\n"
+        "  dnssec-validation no;\n"
+        f'  pid-file "{directory / "named.pid"}";\n'
+        f'  session-keyfile "{directory / "session.key"}";\n'
+        "};\n"
+        "controls {};\n"
+        f'zone "{ZONE}" {{ type primary; file "zone.signed"; }};\n',
+        encoding="ascii",
+    )
+    (directory / "anchor").write_text(trust_anchor + "\n",
+                                      encoding="ascii")
+    (directory / "unbound.conf").write_text(
+        "server:\n"
+        f"  interface: {DNS_RESOLVER}@53\n"
+        "  do-ip4: yes\n"
+        "  do-ip6: no\n"
+        '  username: ""\n'
+        f'  directory: "{directory}"\n'
+        f'  pidfile: "{directory / "unbound.pid"}"\n'
+        f'  logfile: "{directory / "unbound.log"}"\n'
+        "  use-syslog: no\n"
+        f'  trust-anchor-file: "{directory / "anchor"}"\n'
+        "stub-zone:\n"
+        f'  name: "{ZONE}."\n'
+        f"  stub-addr: {DNS_AUTHORITY}@{DNS_AUTHORITY_PORT}\n",
+        encoding="ascii",
+    )
+
+
+def start_process(arguments: list[str], log: Path) -> subprocess.Popen:
+    stream = log.open("wb")
+    try:
+        process = subprocess.Popen(
+            arguments,
+            stdout=stream,
+            stderr=subprocess.STDOUT,
+            start_new_session=True,
+        )
+    finally:
+        stream.close()
+    return process
+
+
+def stop_process(process: subprocess.Popen | None) -> None:
+    if process is None or process.poll() is not None:
+        return
+    process.terminate()
+    try:
+        process.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait(timeout=5)
+
+
+def wait_for_dns() -> None:
+    deadline = time.monotonic() + 15
+    while time.monotonic() < deadline:
+        result = run([
+            "dig", f"@{DNS_RESOLVER}", "+dnssec", "+adflag",
+            "+time=1", "+tries=1", "MX", f"dane-good.{ZONE}",
+        ], check=False)
+        if (result.returncode == 0 and "status: NOERROR" in result.stdout
+                and " ad;" in result.stdout):
+            return
+        time.sleep(0.2)
+    raise SMTP27Error("local validating DNSSEC resolver did not become ready")
+
+
+def start_mail_servers(
+    trusted_context: ssl.SSLContext,
+    untrusted_context: ssl.SSLContext,
+) -> tuple[list[ThreadingTCPServer], dict[str, Capture]]:
+    servers: list[ThreadingTCPServer] = []
+    captures: dict[str, Capture] = {}
+    for route in ROUTES:
+        context = {
+            "trusted": trusted_context,
+            "untrusted": untrusted_context,
+            None: None,
+        }[route.tls_kind]
+        capture = Capture(route, context)
+        server = ThreadingTCPServer((route.address, 25), SMTPHandler)
+        server.capture = capture  # type: ignore[attr-defined]
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        servers.append(server)
+        captures[route.name] = capture
+    return servers, captures
+
+
+def stop_server(server: socketserver.BaseServer | None) -> None:
+    if server is None:
+        return
+    server.shutdown()
+    server.server_close()
+
+
+def policies() -> dict[str, str]:
+    result = {}
+    for route in ROUTES:
+        if route.sts_mode is None:
+            continue
+        mx = route.sts_mx or route.mx
+        result[route.policy_host] = (
+            "version: STSv1\n"
+            f"mode: {route.sts_mode}\n"
+            f"mx: {mx}\n"
+            "max_age: 86400\n"
+        )
+    return result
+
+
+def start_policy_server(
+    trusted_context: ssl.SSLContext,
+) -> PolicyServer:
+    server = PolicyServer((HTTPS_ADDRESS, 443), PolicyHandler)
+    server.socket = trusted_context.wrap_socket(
+        server.socket, server_side=True)
+    server.policies = policies()  # type: ignore[attr-defined]
+    server.requests = []  # type: ignore[attr-defined]
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    return server
+
+
+def marker(route: Route, phase: str) -> str:
+    return f"{TOKEN}-{phase}-{route.name}"
+
+
+def submit(route: Route, phase: str, internal_host: str) -> None:
+    recipient = f"capture@{route.domain}"
+    token = marker(route, phase)
+    message = (
+        f"From: SMTP-27 <smtp27@bongo.test>\r\n"
+        f"To: {recipient}\r\n"
+        f"Subject: {token}\r\n"
+        f"Message-ID: <{token}@bongo.test>\r\n"
+        "\r\n"
+        f"{token}\r\n"
+    ).encode("ascii")
+    with smtplib.SMTP(internal_host, SMTP07.INTERNAL_PORT,
+                      timeout=10) as client:
+        client.ehlo("smtp27-client.bongo.test")
+        refused = client.sendmail(
+            "smtp27@bongo.test", [recipient], message)
+        if refused:
+            raise SMTP27Error(f"Bongo refused {recipient}: {refused}")
+
+
+def matching_messages(
+    capture: Capture, route: Route, phase: str
+) -> list[tuple[bool, bytes]]:
+    token = marker(route, phase).encode("ascii")
+    return [
+        item for item in capture.snapshot()[2]
+        if token in item[1]
+    ]
+
+
+def queued_tokens(expected: set[str]) -> set[str]:
+    found: set[str] = set()
+    for queue_id in SMTP07.queue_ids():
+        message = SMTP07.queue("message", queue_id, check=False)
+        if message.returncode:
+            continue
+        for token in expected:
+            if token in message.stdout:
+                found.add(token)
+    return found
+
+
+def wait_until(predicate, description: str) -> None:
+    deadline = time.monotonic() + TIMEOUT
+    while time.monotonic() < deadline:
+        if predicate():
+            time.sleep(1)
+            return
+        time.sleep(0.2)
+    raise SMTP27Error(f"timeout waiting for {description}")
+
+
+def validate_initial(
+    captures: dict[str, Capture],
+) -> set[str]:
+    success = ("dane-good", "sts-enforce", "sts-testing", "sts-cache")
+    failure = ("dane-bad", "sts-mismatch", "dane-precedence", "sts-plain")
+    queued = {
+        marker(next(route for route in ROUTES if route.name == name),
+               "initial")
+        for name in failure
+    }
+    wait_until(
+        lambda: (
+            all(matching_messages(
+                captures[name],
+                next(route for route in ROUTES if route.name == name),
+                "initial") for name in success)
+            and queued_tokens(queued) == queued
+        ),
+        "DANE and MTA-STS initial outcomes",
+    )
+    for name in success:
+        route = next(route for route in ROUTES if route.name == name)
+        messages = matching_messages(captures[name], route, "initial")
+        if len(messages) != 1:
+            raise SMTP27Error(
+                f"{name}: expected one delivery, got {len(messages)}")
+        expected_tls = name != "sts-testing"
+        if messages[0][0] is not expected_tls:
+            raise SMTP27Error(
+                f"{name}: expected encrypted={expected_tls}, "
+                f"got {messages[0][0]}")
+    for name in failure:
+        route = next(route for route in ROUTES if route.name == name)
+        if matching_messages(captures[name], route, "initial"):
+            raise SMTP27Error(f"{name}: policy failure delivered a message")
+    return queued
+
+
+def validate_cached_policy(
+    captures: dict[str, Capture], route: Route
+) -> None:
+    wait_until(
+        lambda: len(matching_messages(
+            captures[route.name], route, "cached")) == 1,
+        "MTA-STS cached-policy delivery",
+    )
+    messages = matching_messages(captures[route.name], route, "cached")
+    if messages[0][0] is not True:
+        raise SMTP27Error("cached MTA-STS policy delivered without TLS")
+
+
+def append_trust_anchor(anchor: str) -> None:
+    if ANCHOR_BACKUP.exists():
+        raise SMTP27Error(
+            f"stale DNS trust-anchor backup at {ANCHOR_BACKUP}")
+    metadata = SMTP07.regular_file_metadata(ANCHOR_FILE)
+    original = ANCHOR_FILE.read_bytes()
+    backup_file(ANCHOR_BACKUP, original)
+    payload = original
+    if payload and not payload.endswith(b"\n"):
+        payload += b"\n"
+    payload += (anchor + "\n").encode("ascii")
+    SMTP07.atomic_write(ANCHOR_FILE, payload, metadata)
+
+
+def backup_file(path: Path, payload: bytes) -> None:
+    descriptor = os.open(
+        path,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+        0o600,
+    )
+    with os.fdopen(descriptor, "wb") as stream:
+        stream.write(payload)
+        stream.flush()
+        os.fsync(stream.fileno())
+
+
+def restore_file(path: Path, backup: Path) -> None:
+    if not backup.exists():
+        return
+    metadata = SMTP07.regular_file_metadata(path)
+    SMTP07.atomic_write(path, backup.read_bytes(), metadata)
+    backup.unlink()
+
+
+def perform_test() -> None:
+    if os.geteuid() != 0:
+        raise SMTP27Error("the live DNSSEC fixture must run as root")
+    if RESOLV_BACKUP.exists() or ANCHOR_BACKUP.exists():
+        raise SMTP27Error(
+            "stale SMTP-27 backup exists; run --restore before testing")
+    if not SMTP07.service_active():
+        raise SMTP27Error("bongo.service must be active before this test")
+    if not ANCHOR_FILE.is_file():
+        raise SMTP27Error(f"DNS root trust-anchor file missing: {ANCHOR_FILE}")
+    if not (PKI_ROOT / "trusted-ca.crt").is_file():
+        raise SMTP27Error(
+            "local test PKI is not prepared; run local-test-pki.py install")
+
+    original_smtp = SMTP07.read_smtp_configuration()
+    if original_smtp.get("use_relay_host"):
+        raise SMTP27Error(
+            "SMTP-27 requires direct MX delivery, but relay_host is enabled")
+    internal_host = str(original_smtp.get(
+        "internal_relay_bind_address", "")).strip()
+    if internal_host in ("", "0.0.0.0", "::"):
+        raise SMTP27Error(
+            "set BONGO_TEST_INTERNAL_HOST for a wildcard trusted listener")
+    SMTP07.INTERNAL_HOST = internal_host
+    original_resolver = SMTP07.RESOLV_CONF.read_bytes()
+    resolver_metadata = SMTP07.regular_file_metadata(SMTP07.RESOLV_CONF)
+    original_active = SMTP07.service_active()
+    smtp_changed = False
+    resolver_changed = False
+    anchor_changed = False
+    dns_directory: Path | None = None
+    certificate_directories: list[Path] = []
+    named_process: subprocess.Popen | None = None
+    unbound_process: subprocess.Popen | None = None
+    mail_servers: list[ThreadingTCPServer] = []
+    policy_server: PolicyServer | None = None
+    primary_error: BaseException | None = None
+
+    def terminate(_signum: int, _frame) -> None:
+        raise KeyboardInterrupt
+
+    signal.signal(signal.SIGTERM, terminate)
+    signal.signal(signal.SIGHUP, terminate)
+    try:
+        trusted, untrusted, certificate_directories = issue_certificates()
+        trusted_context = certificate_context(trusted)
+        untrusted_context = certificate_context(untrusted)
+        valid_tlsa = certificate_sha256(untrusted / "cert.pem")
+
+        bongo_user = pwd.getpwnam("bongo")
+        dns_directory = Path(tempfile.mkdtemp(
+            prefix="smtp27-", dir=WORK_ROOT))
+        dns_directory.chmod(0o755)
+        cache_directory = dns_directory / "mta-sts-cache"
+        cache_directory.mkdir(mode=0o750)
+        os.chown(cache_directory, bongo_user.pw_uid, bongo_user.pw_gid)
+        trust_anchor = write_zone(dns_directory, valid_tlsa)
+        write_dns_configs(dns_directory, trust_anchor)
+
+        test_smtp = copy.deepcopy(original_smtp)
+        test_smtp["port"] = SMTP07.TEMPORARY_PUBLIC_PORT
+        test_smtp["outbound_tls_verify"] = False
+        test_smtp["outbound_tls_required"] = False
+        test_smtp["outbound_dane_enabled"] = True
+        test_smtp["outbound_mta_sts_enabled"] = True
+        test_smtp["outbound_mta_sts_cache_directory"] = str(cache_directory)
+        SMTP07.replace_smtp_configuration(test_smtp)
+        smtp_changed = True
+        run([SMTP07.SYSTEMCTL, "stop", "bongo.service"])
+
+        SMTP07.save_resolver(original_resolver)
+        SMTP07.atomic_write(
+            SMTP07.RESOLV_CONF,
+            (
+                "# Temporary Bongo SMTP-27 validating resolver\n"
+                f"nameserver {DNS_RESOLVER}\n"
+                "options timeout:1 attempts:2\n"
+            ).encode("ascii"),
+            resolver_metadata,
+        )
+        resolver_changed = True
+        append_trust_anchor(trust_anchor)
+        anchor_changed = True
+
+        named_process = start_process(
+            ["named", "-g", "-c", str(dns_directory / "named.conf")],
+            dns_directory / "named.log",
+        )
+        unbound_process = start_process(
+            ["unbound", "-d", "-c", str(dns_directory / "unbound.conf")],
+            dns_directory / "unbound-process.log",
+        )
+        wait_for_dns()
+        mail_servers, captures = start_mail_servers(
+            trusted_context, untrusted_context)
+        policy_server = start_policy_server(trusted_context)
+
+        run([SMTP07.SYSTEMCTL, "start", "bongo.service"])
+        SMTP07.wait_for_bongo()
+        time.sleep(STARTUP_SETTLE)
+        for route in ROUTES:
+            submit(route, "initial", internal_host)
+        failed_tokens = validate_initial(captures)
+
+        cache_route = next(
+            route for route in ROUTES if route.name == "sts-cache")
+        requests_before = list(policy_server.requests)  # type: ignore[attr-defined]
+        stop_server(policy_server)
+        policy_server = None
+        submit(cache_route, "cached", internal_host)
+        validate_cached_policy(captures, cache_route)
+        if requests_before.count(cache_route.policy_host) != 1:
+            raise SMTP27Error(
+                "MTA-STS cache route did not fetch exactly one policy")
+        if queued_tokens(failed_tokens) != failed_tokens:
+            raise SMTP27Error("a deferred policy failure disappeared from Queue")
+    except BaseException as error:
+        primary_error = error
+    finally:
+        restoration_errors: list[str] = []
+        if policy_server is not None:
+            with contextlib.suppress(OSError):
+                stop_server(policy_server)
+        for server in mail_servers:
+            with contextlib.suppress(OSError):
+                stop_server(server)
+        stop_process(unbound_process)
+        stop_process(named_process)
+
+        if not SMTP07.service_active():
+            try:
+                run([SMTP07.SYSTEMCTL, "start", "bongo.service"])
+                SMTP07.wait_for_bongo()
+            except (OSError, RuntimeError) as error:
+                restoration_errors.append(
+                    f"could not start Bongo for cleanup: {error}")
+        if SMTP07.service_active():
+            try:
+                SMTP07.cleanup_queue()
+            except (OSError, RuntimeError) as error:
+                restoration_errors.append(f"queue cleanup failed: {error}")
+            if smtp_changed:
+                try:
+                    SMTP07.replace_smtp_configuration(original_smtp)
+                    smtp_changed = False
+                except (OSError, RuntimeError) as error:
+                    restoration_errors.append(
+                        f"SMTP configuration restore failed: {error}")
+        if run([SMTP07.SYSTEMCTL, "stop", "bongo.service"],
+               check=False).returncode:
+            restoration_errors.append("could not stop bongo.service")
+        try:
+            if resolver_changed:
+                restore_file(SMTP07.RESOLV_CONF, RESOLV_BACKUP)
+                resolver_changed = False
+            if anchor_changed:
+                restore_file(ANCHOR_FILE, ANCHOR_BACKUP)
+                anchor_changed = False
+        except OSError as error:
+            restoration_errors.append(f"DNS configuration restore failed: {error}")
+        if dns_directory is not None:
+            shutil.rmtree(dns_directory, ignore_errors=True)
+        for directory in certificate_directories:
+            shutil.rmtree(directory, ignore_errors=True)
+        if original_active:
+            try:
+                run([SMTP07.SYSTEMCTL, "start", "bongo.service"])
+                SMTP07.wait_for_bongo()
+            except (OSError, RuntimeError) as error:
+                restoration_errors.append(
+                    f"service state restore failed: {error}")
+        if restoration_errors:
+            cleanup_error = "; ".join(restoration_errors)
+            if primary_error is not None:
+                raise SMTP27Error(
+                    f"{primary_error}; cleanup: {cleanup_error}"
+                ) from primary_error
+            raise SMTP27Error(cleanup_error)
+
+    if primary_error is not None:
+        raise primary_error
+    print(
+        "SMTP-27 PASS "
+        "dane=untrusted-ee/mismatch-deferred/precedence "
+        "mta-sts=enforce/testing/mx-mismatch/cache "
+        "dnssec=validated queue-clean=yes "
+        f"service-restored={'active' if original_active else 'inactive'}"
+    )
+
+
+def restore_only() -> None:
+    if os.geteuid() != 0:
+        raise SMTP27Error("restore must run as root")
+    restore_file(SMTP07.RESOLV_CONF, RESOLV_BACKUP)
+    restore_file(ANCHOR_FILE, ANCHOR_BACKUP)
+    print("restored SMTP-27 resolver and trust-anchor files")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    action = parser.add_mutually_exclusive_group(required=True)
+    action.add_argument("--live", action="store_true")
+    action.add_argument("--restore", action="store_true")
+    arguments = parser.parse_args()
+    try:
+        if arguments.restore:
+            restore_only()
+        else:
+            perform_test()
+    except (OSError, RuntimeError, smtplib.SMTPException) as error:
+        print(f"smtp-dane-mta-sts-check: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())