Prepare native sender adapter and verify snapshot integration

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3

scripts/rehearse-sender-native.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Explicit disposable-image rehearsal; synthetic ledger, no Kubernetes or SMTP."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import secrets
+import subprocess
+import tempfile
+from pathlib import Path
+
+from sender_native import IMAGE, Native
+from sender_transition import Journal, continuity, require
+
+from email_connect.transactional import SQLiteDeliveryStore
+
+
+def docker(*args, body=None):
+    result = subprocess.run(["docker", *args], input=body, capture_output=True,
+                            timeout=120, check=False)
+    require(result.returncode == 0, "isolated_container_command_failed")
+    return result.stdout
+
+
+def synthetic(path):
+    store = SQLiteDeliveryStore(str(path))
+    payload = {"tenant": "fixture", "client_id": "fixture", "registration_id": "accepted",
+               "normalized_email": "fixture@example.test", "preferred_username": "fixture"}
+    for kind in ("accepted", "unknown", "consumed", "canceled"):
+        payload["registration_id"] = kind
+        request, _ = store.reserve_verification(payload, "hash-" + kind, "2030-01-01T00:00:00+00:00")
+        if kind != "unknown":
+            store.verification_accepted(request, "fixture-reference-" + kind)
+        if kind == "consumed":
+            store.consume_verification("hash-consumed")
+        if kind == "canceled":
+            store.cancel_verification("hash-canceled")
+    store.reserve_send("accepted-invitation", "accepted-fingerprint")
+    store.record("accepted-invitation", "fixture-receipt")
+    store.reserve_send("unknown-invitation", "unknown-fingerprint")
+    store.suppress("suppressed@example.test", "fixture")
+    store.db.close()
+    path.chmod(0o444)  # Synthetic fixture is mounted read-only for UID10001.
+
+
+class Isolated(Native):
+    def __init__(self, journal, container):
+        super().__init__(execute=True, journal=journal)
+        self.container = container
+        self.lock_fd = -1  # No host capture lock: transport is Docker-only below.
+
+    def k(self, *args, body=None, timeout=180):
+        require(args[0] == "exec" and self.container in args and "--" in args,
+                "fixture_transport_scope_changed")
+        command = list(args[args.index("--") + 1:])
+        require(command[:2] == ["python", "-c"], "fixture_command_scope_changed")
+        return docker("exec", "-i", self.container, *command, body=body)
+
+    def fence(self, journal):
+        self.check_consumers(allow_receiver=False)
+
+    def check_consumers(self, *, allow_receiver):
+        data = json.loads(docker("inspect", self.container))[0]
+        require(data["HostConfig"]["NetworkMode"] == "none"
+                and data["HostConfig"]["ReadonlyRootfs"] is True
+                and not data["HostConfig"].get("PortBindings")
+                and data["Config"]["User"] == "10001:10001", "fixture_isolation_changed")
+        binds = [m for m in data["Mounts"] if m["Type"] == "bind"]
+        require(len(binds) == 1 and binds[0]["Destination"] == "/data" and not binds[0]["RW"],
+                "fixture_mount_scope_changed")
+
+    def ensure_helper(self, journal):
+        return self.container
+
+
+def rehearse(output):
+    require(not output.exists(), "fresh_output_required")
+    image = json.loads(docker("image", "inspect", IMAGE))[0]
+    require(image["Id"] == IMAGE.split("@", 1)[1], "cached_image_mismatch")
+    container = None
+    with tempfile.TemporaryDirectory(prefix="sender-native-rehearsal-") as directory:
+        root = Path(directory)
+        source = root / "source"
+        source.mkdir(mode=0o755)
+        synthetic(source / "email-connect.db")
+        journal = Journal(root / "journal.jsonl", proposal={"synthetic_fixture": True})
+        try:
+            name = "sender-rehearsal-" + secrets.token_hex(6)
+            container = docker("run", "--detach", "--rm", "--pull=never", "--name", name,
+                "--network=none", "--read-only", "--user=10001:10001",
+                "--tmpfs=/tmp:rw,noexec,nosuid,size=160m,mode=1777",
+                "--mount", "type=bind,src=" + str(source) + ",dst=/data,readonly",
+                IMAGE, "python", "-c", "import time;time.sleep(600)").decode().strip()
+            require(len(container) == 64 and all(c in "0123456789abcdef" for c in container),
+                    "fixture_container_identity_missing")
+            adapter = Isolated(journal, container)
+            first, outcomes = adapter.snapshot(journal, quiesced=True)
+            second, _ = adapter.snapshot(journal, quiesced=False)
+            compared = continuity(first, second)
+            proof = {"schema": "email-connect.sender-native-snapshot-rehearsal.v1",
+                "image": IMAGE, "network": "none", "source_mount_read_only": True,
+                "native_capture_and_rehearsal_code_used": True,
+                "kubernetes_api_used": False, "production_data_used": False,
+                "smtp_provider_called": False, "outcomes": outcomes["outcomes"],
+                "continuity": compared,
+                "rehearsal": next(e["proof"] for e in journal.entries
+                                  if e["phase"] == "native_snapshot_rehearsed")}
+        finally:
+            journal.close()
+            if container:
+                docker("rm", "--force", container)
+    proof["container_removed"] = True
+    files = [Path(__file__), Path(__file__).with_name("sender_native.py"),
+             Path(__file__).with_name("sender_transition.py"),
+             Path(__file__).with_name("capture-ledger.py"),
+             Path(__file__).with_name("ledger_archive.py"),
+             Path(__file__).with_name("rehearse-ledger.py"),
+             Path(__file__).with_name("inspect-sender-ledger.py")]
+    proof["sources"] = [{"path": "scripts/" + p.name, "sha256": hashlib.sha256(p.read_bytes()).hexdigest()}
+                        for p in files]
+    fd = os.open(output, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+    with os.fdopen(fd, "w") as stream:
+        json.dump(proof, stream, indent=2)
+        stream.write("\n")
+    return proof
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    result = rehearse(args.output)
+    print(json.dumps({"status": "isolated_snapshot_adapter_verified",
+                      "container_removed": result["container_removed"], "image": result["image"]}))

scripts/sender_native.py

@@ -0,0 +1,453 @@
+"""Node-side adapter for sender_transition; read/dry-run unless explicitly armed.
+
+The attended capsule/entry point must bind these sources and the proposal. This
+module deliberately has no CLI. Secret reads use PartialObjectMetadata only.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import http.client
+import importlib.util
+import json
+import os
+import socket
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+
+from sender_transition import IMAGE, desired, digest, require
+
+NS = "email-connect"
+CLUSTER = "a553c742-0115-43d4-99a4-a5ca56fe0786"
+NAMESPACE = "5678c5d7-e4f5-4a62-8f24-1c6e1d906319"
+DEPLOYMENT = "2e98e52a-a2d1-4c52-9ac8-6efbc0a18e72"
+CONFIG = "558d3338-c4c2-490c-a5e5-af206c39eb51"
+FOUNDATION = "sha256:f8da6c831160e1c6354ce54f936809d2d4fb48eb45eb23484dc0bcdfe0bb52f0"
+FOUNDATION_SPEC = "cf5e1a5782243ebb72790ac7e9b868f44df027c1608ad37325264d290e70f537"
+PVC = "email-connect-data"
+CAPTURE_LOCK = Path("/var/lib/email-connect-backup/capture.lock")
+ROOT = Path(__file__).resolve().parent
+
+
+def module(name):
+    spec = importlib.util.spec_from_file_location(name.replace("-", "_"), ROOT / (name + ".py"))
+    value = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(value)
+    return value
+
+
+def stable(obj):
+    require(obj is not None and not obj["metadata"].get("deletionTimestamp"), "object_missing_or_deleting")
+    return {"uid": obj["metadata"]["uid"], "spec": obj.get("spec"), "data": obj.get("data")}
+
+
+class Native:
+    def __init__(self, *, execute=False, journal=None, lock_path=CAPTURE_LOCK):
+        self.kube = ["k3s", "kubectl", "--kubeconfig=/etc/rancher/k3s/k3s.yaml"]
+        self.execute = execute
+        self.journal = journal
+        self.lock_path = lock_path
+        self.lock_fd = None
+        self.foundation_pins = None
+        self.capture = module("capture-ledger")
+        self.archive = module("ledger_archive")
+        self.observer = module("inspect-sender-ledger")
+        self.rehearsal = module("rehearse-ledger")
+        self.capture.k = self.k
+        self.receiver_uid = None
+        self.node_name = None
+
+    def k(self, *args, body=None, timeout=180):
+        result = subprocess.run(self.kube + list(args), input=body, capture_output=True,
+                                timeout=timeout, check=False)
+        require(result.returncode == 0, "native_command_failed")
+        return result.stdout
+
+    def get(self, kind, name=None, *, namespace=NS):
+        require(kind.lower() in {"namespace", "deployment", "configmap", "pvc", "pv", "nodes",
+                "externalsecret", "service", "networkpolicies", "serviceaccount", "secretstore",
+                "pods", "pod", "replicaset"}, "resource_query_outside_reviewed_scope")
+        args = ["get", kind]
+        if name:
+            args += [name, "--ignore-not-found"]
+        if namespace:
+            args += ["-n", namespace]
+        raw = self.k(*args, "-o", "json", timeout=30)
+        return json.loads(raw) if raw.strip() else None
+
+    def metadata(self, name):
+        require(name in {"email-connect-runtime", "email-connect-caller-user-engine",
+                         "email-connect-platform-smtp"}, "secret_metadata_scope_drift")
+        path = f"/api/v1/namespaces/{NS}/secrets/{name}"
+        with tempfile.TemporaryDirectory(prefix="sender-metadata-") as directory:
+            sockpath = str(Path(directory) / "api.sock")
+            proc = subprocess.Popen(self.kube + [
+                "proxy", "--unix-socket=" + sockpath, "--accept-paths=^" + path + "$",
+                "--reject-methods=^(POST|PUT|PATCH|DELETE)$",
+            ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            conn = None
+            try:
+                for _ in range(50):
+                    if Path(sockpath).exists():
+                        break
+                    require(proc.poll() is None, "metadata_proxy_failed")
+                    time.sleep(0.1)
+                sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                sock.settimeout(10)
+                try:
+                    sock.connect(sockpath)
+                except BaseException:
+                    sock.close()
+                    raise
+                conn = http.client.HTTPConnection("localhost", timeout=10)
+                conn.sock = sock
+                conn.request("GET", path, headers={
+                    "Accept": "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1"})
+                response = conn.getresponse()
+                if response.status == 404:
+                    return None
+                require(response.status == 200, "metadata_query_failed")
+                value = json.load(response)
+                require(value.get("kind") == "PartialObjectMetadata"
+                        and value.get("apiVersion") == "meta.k8s.io/v1"
+                        and "data" not in value and "stringData" not in value,
+                        "metadata_only_response_required")
+                meta = value["metadata"]
+                require(not meta.get("deletionTimestamp"), "secret_deleting")
+                return {"uid": meta["uid"], "owners": meta.get("ownerReferences", [])}
+            finally:
+                if conn:
+                    conn.close()
+                proc.terminate()
+                try:
+                    proc.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    proc.kill()
+                    proc.wait(timeout=5)
+
+    def observe(self):
+        require(stable(self.get("namespace", "kube-system", namespace=None))["uid"] == CLUSTER,
+                "cluster_identity_changed")
+        require(stable(self.get("namespace", NS, namespace=None))["uid"] == NAMESPACE,
+                "namespace_identity_changed")
+        deployment = self.get("deployment", "email-connect")
+        config = self.get("configmap", "email-connect-config")
+        require(stable(deployment)["uid"] == DEPLOYMENT and stable(config)["uid"] == CONFIG,
+                "application_identity_changed")
+        pvc = self.get("pvc", PVC)
+        require(pvc.get("status", {}).get("phase") == "Bound", "ledger_pvc_not_bound")
+        pv = self.get("pv", pvc["spec"]["volumeName"], namespace=None)
+        require(pv["spec"]["claimRef"]["uid"] == pvc["metadata"]["uid"], "ledger_volume_binding_changed")
+        terms = pv["spec"]["nodeAffinity"]["required"]["nodeSelectorTerms"]
+        require(len(terms) == 1 and not terms[0].get("matchFields"), "volume_node_scope_changed")
+        expressions = terms[0]["matchExpressions"]
+        require(len(expressions) == 1 and expressions[0]["key"] == "kubernetes.io/hostname"
+                and expressions[0]["operator"] == "In" and len(expressions[0]["values"]) == 1,
+                "volume_node_scope_changed")
+        hostname = expressions[0]["values"][0]
+        nodes = [n for n in self.get("nodes", namespace=None)["items"]
+                 if n["metadata"].get("labels", {}).get("kubernetes.io/hostname") == hostname]
+        require(len(nodes) == 1, "volume_node_identity_ambiguous")
+        node = nodes[0]
+        self.node_name = node["metadata"]["name"]
+        volumes = deployment["spec"]["template"]["spec"]["volumes"]
+        policy = next(v["configMap"]["name"] for v in volumes if v["name"] == "caller-policy")
+        require(policy == "email-connect-callers-user-engine-276257849bf2", "caller_policy_changed")
+        projection = self.get("externalsecret", "email-connect-runtime")
+        old_secret = self.metadata("email-connect-runtime")
+        caller_secret = self.metadata("email-connect-caller-user-engine")
+        require(old_secret and caller_secret, "existing_custody_missing")
+        policies = self.get("networkpolicies")["items"]
+        if self.foundation_pins is not None:
+            require(self.projection(self.journal.entries[0]["proposal"])
+                    == self.foundation_pins, "new_projection_identity_changed")
+        return {
+            "identity": {"cluster": CLUSTER, "namespace": NAMESPACE,
+                         "deployment": DEPLOYMENT, "config": CONFIG},
+            "deployment_spec": deployment["spec"], "config_data": config["data"],
+            "protected": {
+                "pvc": stable(pvc), "pv": stable(pv),
+                "node": {"name": self.node_name, "uid": node["metadata"]["uid"], "hostname": hostname},
+                "service": stable(self.get("service", "email-connect")),
+                "caller_policy": stable(self.get("configmap", policy)),
+                "old_projection": stable(projection), "old_secret": old_secret,
+                "caller_secret": caller_secret,
+                "network_policies": sorted((stable(p) for p in policies), key=lambda x: x["uid"]),
+            },
+        }
+
+    def prepare(self):
+        require(not self.execute, "preparation_must_be_read_only")
+        before = self.observe()
+        dep = self.get("deployment", "email-connect")
+        require(dep["status"].get("readyReplicas") == 1
+                and dep["status"].get("observedGeneration") == dep["metadata"]["generation"],
+                "receiver_not_ready")
+        after = desired(before)
+        for spec in (dict(before["deployment_spec"], replicas=0),
+                     dict(after["deployment_spec"], replicas=0), after["deployment_spec"]):
+            self.patch_deployment(before["deployment_spec"], spec)
+        self.patch_config(before["config_data"], after["config_data"])
+        body = self.helper_manifest("sender-preflight-helper")
+        self.helper_egress(body)
+        projected = json.loads(self.k("create", "--dry-run=server", "-f", "-", "-o", "json",
+                                      body=json.dumps(body).encode(), timeout=30))
+        require(projected["spec"]["automountServiceAccountToken"] is False,
+                "helper_token_mount_changed")
+        require(self.observe() == before, "preparation_changed")
+        return {"before": before, "after": after, "native_server_dry_runs": 5}
+
+    def armed(self):
+        require(self.execute and self.journal is not None and self.lock_fd is not None,
+                "attended_locked_execution_required")
+
+    def acquire(self):
+        require(self.execute and self.journal is not None, "attended_execution_required")
+        require(self.lock_path == CAPTURE_LOCK, "capture_lock_path_changed")
+        self.archive.private_directory(self.lock_path.parent)
+        self.lock_fd = os.open(self.lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+        try:
+            self.archive.private_file(self.lock_path)
+            fcntl.flock(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BaseException:
+            os.close(self.lock_fd)
+            self.lock_fd = None
+            raise
+
+    def close(self):
+        if self.lock_fd is not None:
+            os.close(self.lock_fd)
+            self.lock_fd = None
+
+    def patch(self, kind, name, field, expected, value, uid):
+        if self.execute:
+            self.armed()
+        obj = self.get(kind, name)
+        require(stable(obj)["uid"] == uid and obj[field] == expected, "compare_and_swap_drift")
+        patch = [{"op": "test", "path": "/metadata/uid", "value": uid},
+                 {"op": "test", "path": "/metadata/resourceVersion",
+                  "value": obj["metadata"]["resourceVersion"]},
+                 {"op": "test", "path": "/" + field, "value": expected},
+                 {"op": "replace", "path": "/" + field, "value": value}]
+        args = ["patch", kind, name, "-n", NS, "--type=json", "--patch-file=/dev/stdin", "-o", "json"]
+        if not self.execute:
+            args.append("--dry-run=server")
+        result = json.loads(self.k(*args, body=json.dumps(patch).encode(), timeout=30))
+        require(result["metadata"]["uid"] == uid and result[field] == value,
+                "patch_result_not_verified")
+
+    def patch_deployment(self, expected, value):
+        self.patch("deployment", "email-connect", "spec", expected, value, DEPLOYMENT)
+
+    def patch_config(self, expected, value):
+        require(self.get("deployment", "email-connect")["spec"]["replicas"] == 0
+                or not self.execute, "receiver_not_paused_for_config")
+        self.patch("configmap", "email-connect-config", "data", expected, value, CONFIG)
+
+    def projection(self, proposal):
+        foundation = proposal["foundation"]
+        require(digest(foundation) == FOUNDATION_SPEC, "foundation_contract_changed")
+        pins = {}
+        for wanted in foundation["resources"]:
+            kind, name = wanted["kind"], wanted["metadata"]["name"]
+            obj = self.get(kind, name)
+            require(obj is not None and obj.get("apiVersion") == wanted["apiVersion"],
+                    "foundation_resource_missing")
+            pins[kind] = stable(obj)["uid"]
+            if kind == "ServiceAccount":
+                require(obj.get("automountServiceAccountToken") is False
+                        and not obj.get("secrets") and not obj.get("imagePullSecrets"),
+                        "service_account_changed")
+            else:
+                require(obj["spec"] == wanted["spec"], "foundation_spec_changed")
+                require(any(c.get("type") == "Ready" and c.get("status") == "True"
+                            for c in obj.get("status", {}).get("conditions", [])),
+                        "foundation_not_ready")
+            if kind == "ExternalSecret":
+                require(obj.get("status", {}).get("refreshTime"), "projection_refresh_missing")
+        meta = self.metadata("email-connect-platform-smtp")
+        require(meta is not None and any(o.get("uid") == pins["ExternalSecret"]
+                and o.get("controller") is True for o in meta["owners"]), "projection_owner_changed")
+        pins["Secret"] = meta["uid"]
+        return pins
+
+    def admission(self, proposal):
+        self.armed()
+        receipt = proposal["foundation_receipt"]
+        require(receipt.get("schema") == "platform.email-connect-smtp-foundation.v1"
+                and receipt.get("proposal_digest") == FOUNDATION
+                and receipt.get("status") == "foundation_admitted_projection_ready"
+                and receipt.get("application_sender_changed") is False,
+                "verified_owner_foundation_receipt_required")
+        for key in ("native_binding_verified", "native_consumer_auth_verified", "exact_read_verified",
+                    "mutation_sibling_metadata_denied", "bounded_ttl", "other_service_account_denied",
+                    "wrong_audience_denied", "workload_token_revocation_verified", "projection_ready"):
+            require(receipt.get(key) is True, "foundation_proof_incomplete")
+        pins = self.projection(proposal)
+        require(pins["ServiceAccount"] == receipt["service_account_uid"], "admitted_identity_changed")
+        self.journal.record("native_foundation_verified", pins=pins)
+        self.foundation_pins = pins
+
+    def helper_manifest(self, name):
+        body = self.capture.helper(name)
+        # Read-only PVC, no credentials, explicit scheduling on the existing node.
+        require(self.node_name is not None, "volume_node_not_observed")
+        body["spec"]["nodeName"] = self.node_name
+        return body
+
+    def helper_egress(self, body):
+        self.capture.require_no_helper_egress(self.get("networkpolicies")["items"],
+                                             body["metadata"]["labels"])
+
+    def helpers(self):
+        require(self.journal is not None, "journal_required")
+        return {e["name"]: e for e in self.journal.entries if e["phase"] == "native_helper_created"}
+
+    def exact_helper(self, obj, record):
+        require(stable(obj)["uid"] == record["uid"] and obj["spec"] == record["spec"],
+                "helper_identity_or_spec_changed")
+        self.helper_egress(obj)
+
+    def fence(self, journal):
+        self.armed()
+        self.k("wait", "--for=delete", "pod", "-n", NS,
+               "-l", "app.kubernetes.io/name=email-connect", "--timeout=120s")
+        require(self.get("deployment", "email-connect")["spec"]["replicas"] == 0,
+                "receiver_not_quiesced")
+        self.check_consumers(allow_receiver=False)
+        journal.record("native_writers_fenced", deployment_uid=DEPLOYMENT)
+
+    def check_consumers(self, *, allow_receiver):
+        known_helpers = self.helpers()
+        for pod in self.get("pods")["items"]:
+            if any(v.get("persistentVolumeClaim", {}).get("claimName") == PVC
+                   for v in pod["spec"].get("volumes", [])):
+                record = known_helpers.get(pod["metadata"]["name"])
+                if allow_receiver and pod["metadata"]["uid"] == self.receiver_uid:
+                    require(not pod["metadata"].get("deletionTimestamp"), "receiver_is_terminating")
+                    continue
+                require(record is not None, "unowned_ledger_pvc_consumer")
+                self.exact_helper(pod, record)
+
+    def ensure_helper(self, journal):
+        self.armed()
+        for name, record in reversed(list(self.helpers().items())):
+            obj = self.get("pod", name)
+            if obj is not None:
+                self.exact_helper(obj, record)
+                self.k("wait", "--for=condition=Ready", "pod/" + name, "-n", NS, "--timeout=120s")
+                return name
+        # A create with no durable returned UID cannot authorize adoption/deletion.
+        for entry in journal.entries:
+            if entry["phase"] == "native_helper_create_intent" and entry["name"] not in self.helpers():
+                require(self.get("pod", entry["name"]) is None, "unconfirmed_helper_requires_inspection")
+        name = "sender-ledger-" + journal.entries[0]["proposal_digest"][:12] + "-" + str(len(journal.entries))
+        body = self.helper_manifest(name)
+        self.helper_egress(body)
+        require(self.get("pod", name) is None, "helper_name_occupied")
+        projected = json.loads(self.k("create", "--dry-run=server", "-f", "-", "-o", "json",
+                                      body=json.dumps(body).encode(), timeout=30))
+        journal.record("native_helper_create_intent", name=name, spec=projected["spec"])
+        obj = json.loads(self.k("create", "-f", "-", "-o", "json", body=json.dumps(body).encode()))
+        require(obj["spec"] == projected["spec"], "helper_create_spec_changed")
+        journal.record("native_helper_created", name=name, uid=obj["metadata"]["uid"], spec=obj["spec"])
+        self.k("wait", "--for=condition=Ready", "pod/" + name, "-n", NS, "--timeout=120s")
+        self.exact_helper(self.get("pod", name), self.helpers()[name])
+        return name
+
+    def snapshot(self, journal, *, quiesced):
+        self.armed()
+        if quiesced:
+            self.fence(journal)
+        name = self.ensure_helper(journal)
+        self.check_consumers(allow_receiver=not quiesced)
+        destination = journal.path.parent / ("snapshot-" + str(len(journal.entries)))
+        journal.record("native_snapshot_intent", name=name, directory=destination.name,
+                       writers_quiesced=quiesced)
+        destination.mkdir(mode=0o700)
+        captured = self.capture.capture_database(name, destination)
+        captured["writers_quiesced"] = quiesced
+        path = destination / "database.sqlite3"
+        outcomes = self.observer.inspect(path)
+        self.check_consumers(allow_receiver=not quiesced)
+        require(captured["ledger_fingerprint"] == outcomes["ledger_fingerprint"], "snapshot_drift")
+        if quiesced:
+            # Run the existing source-bound no-provider rehearsal in helper memory.
+            archive_source = (ROOT / "ledger_archive.py").read_text()
+            prefix = ("import types,sys\nif not __debug__:raise RuntimeError('assertions_required')\n"
+                      "m=types.ModuleType('ledger_archive')\n"
+                      "sys.modules['ledger_archive']=m\nexec(" + repr(archive_source) + ",m.__dict__)\n")
+            code = prefix + self.rehearsal.CODE
+            proof = json.loads(self.k("exec", "-i", "-n", NS, name, "--", "python", "-c", code,
+                                      body=path.read_bytes()))
+            require(proof.get("status") == "isolated_ledger_verified" and proof.get("mail_sent") is False
+                    and proof["ledger_fingerprint"] == captured["ledger_fingerprint"],
+                    "isolated_rehearsal_failed")
+            journal.record("native_snapshot_rehearsed", proof=proof)
+        journal.record("native_snapshot_captured", name=name, directory=destination.name, proof=captured)
+        return path, outcomes
+
+    def verify(self, target, journal):
+        self.armed()
+        self.k("rollout", "status", "deployment/email-connect", "-n", NS, "--timeout=180s", timeout=240)
+        require(self.observe() == target, "receiver_spec_not_verified")
+        pods = [p for p in self.get("pods")["items"]
+                if p["metadata"].get("labels", {}).get("app.kubernetes.io/name") == NS]
+        require(len(pods) == 1, "receiver_pod_count_changed")
+        pod = pods[0]
+        require(not pod["metadata"].get("deletionTimestamp") and pod["status"]["phase"] == "Running",
+                "receiver_not_running")
+        statuses = pod["status"].get("containerStatuses", [])
+        require(len(statuses) == 1 and statuses[0].get("ready") is True
+                and statuses[0]["imageID"].endswith(IMAGE.split("@", 1)[1]), "receiver_image_not_verified")
+        wanted = target["deployment_spec"]["template"]["spec"]["containers"][0]
+        actual = pod["spec"]["containers"]
+        require(len(actual) == 1 and all(actual[0].get(k) == wanted.get(k)
+                for k in ("image", "env", "envFrom")), "receiver_environment_references_changed")
+        owner = next(o for o in pod["metadata"]["ownerReferences"] if o.get("controller"))
+        require(owner["kind"] == "ReplicaSet", "receiver_owner_changed")
+        rs = self.get("replicaset", owner["name"])
+        require(rs["metadata"]["uid"] == owner["uid"] and any(
+            o.get("uid") == DEPLOYMENT and o.get("controller") for o in rs["metadata"]["ownerReferences"]),
+            "receiver_deployment_owner_changed")
+        sender = target["config_data"]["EMAIL_CONNECT_SENDER"]
+        code = '''import json,os,urllib.request,urllib.error
+if not __debug__:raise RuntimeError('assertions_required')
+assert os.environ['EMAIL_CONNECT_SENDER']==__EXPECTED_SENDER__
+assert os.environ['EMAIL_CONNECT_SMTP_USERNAME']==__EXPECTED_SENDER__
+assert os.environ['EMAIL_CONNECT_CALLERS_FILE']=='/etc/email-connect-callers/callers.json'
+assert os.environ.get('EMAIL_CONNECT_SMTP_PASSWORD')
+for path in ['/healthz','/readyz']:
+ with urllib.request.urlopen('http://127.0.0.1:8080'+path,timeout=5) as r:assert r.status==200
+request=urllib.request.Request('http://127.0.0.1:8080/v1/send',data=b'{}',method='POST',
+ headers={'Content-Type':'application/json','Authorization':'Bearer synthetic-invalid'})
+try:
+ urllib.request.urlopen(request,timeout=5);raise RuntimeError('unauthenticated_request_accepted')
+except urllib.error.HTTPError as e:
+ assert e.code==401 and json.load(e).get('error')=='unauthorized'
+print(json.dumps({'health_ready':True,'public_sender_verified':True,'scoped_caller_mode':True,
+ 'smtp_password_present':True,'unauthenticated_denial':401,'mail_sent':False,'private_values_printed':False}))
+'''.replace("__EXPECTED_SENDER__", repr(sender))
+        proof = json.loads(self.k("exec", "-n", NS, pod["metadata"]["name"], "-c", NS,
+                                  "--", "python", "-c", code))
+        self.receiver_uid = pod["metadata"]["uid"]
+        journal.record("native_receiver_verified", pod_uid=self.receiver_uid, proof=proof)
+        return proof
+
+    def cleanup(self, journal):
+        self.armed()
+        for name, record in self.helpers().items():
+            obj = self.get("pod", name)
+            if obj is None:
+                continue
+            self.exact_helper(obj, record)
+            journal.record("native_helper_delete_intent", name=name, uid=record["uid"])
+            options = {"apiVersion": "v1", "kind": "DeleteOptions",
+                       "preconditions": {"uid": record["uid"]}}
+            self.k("delete", "--raw=/api/v1/namespaces/" + NS + "/pods/" + name,
+                   "-f", "/dev/stdin", body=json.dumps(options).encode())
+            self.k("wait", "--for=delete", "pod/" + name, "-n", NS, "--timeout=120s")
+            require(self.get("pod", name) is None, "helper_cleanup_unverified")
+            journal.record("native_helper_deleted", name=name, uid=record["uid"])