From c0c322185538fb46698c654c83f71e0a4988c90d Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 31 Oct 2026 17:02:48 +0100 Subject: [PATCH] Prepare native sender adapter and verify snapshot integration Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3 --- docs/sender-native-adapter-2026-10-31.md | 78 +++ docs/sender-native-preflight-2026-10-31.json | 43 ++ ...-native-snapshot-rehearsal-2026-10-31.json | 136 ++++++ docs/sender-transition-protocol-2026-10-31.md | 5 + scripts/rehearse-sender-native.py | 137 ++++++ scripts/sender_native.py | 453 ++++++++++++++++++ tests/test_sender_native.py | 377 +++++++++++++++ ...MAIL-WP-0010-platform-sender-transition.md | 25 +- 8 files changed, 1248 insertions(+), 6 deletions(-) create mode 100644 docs/sender-native-adapter-2026-10-31.md create mode 100644 docs/sender-native-preflight-2026-10-31.json create mode 100644 docs/sender-native-snapshot-rehearsal-2026-10-31.json create mode 100644 scripts/rehearse-sender-native.py create mode 100644 scripts/sender_native.py create mode 100644 tests/test_sender_native.py diff --git a/docs/sender-native-adapter-2026-10-31.md b/docs/sender-native-adapter-2026-10-31.md new file mode 100644 index 0000000..0d4d4f1 --- /dev/null +++ b/docs/sender-native-adapter-2026-10-31.md @@ -0,0 +1,78 @@ +# Native sender adapter preparation + +EMAIL-WP-0010-T02 remains in progress. `scripts/sender_native.py` connects the +sender transition protocol to the existing Kubernetes, ledger capture and +isolated replay tools. It has no CLI; default construction permits reads and +server dry-runs only. The attended entry point and final source-bound release +packet still need preparation. No production sender change occurred. + +## Implemented native checks + +The adapter pins cluster, namespace, deployment and ConfigMap identities. It +compares complete deployment specifications and ConfigMap data with UID and +resource-version preconditions before patches. It preserves caller policy, +existing projections, Service and network policies. It verifies the ledger PVC's +PV claim UID and derives the single backing node from that PV's node affinity. + +Secret requests negotiate Kubernetes PartialObjectMetadata; a full Secret +response is refused. The dedicated SMTP projection must match the reviewed +foundation contract, have Ready resources, and retain the expected controller +ownership. Admission requires the exact successful foundation proposal receipt +and its authentication, scope, denial, TTL and workload-token revocation checks. +The adapter creates no credential grant. The independent RPF-WP-0068-T04 owner +return, including the approved credential lane and cleanup, remains a prerequisite +for the final execution packet. + +Mutation methods require an execution flag, durable journal and exclusive existing +capture lock. Pausing waits for receiver pods to disappear and rejects any other +PVC consumer except exact journal-owned helpers. Helper creation records an intent +before the API call and the returned UID before use. A lost creation response +without a recorded UID stops for identity inspection. Cleanup checks both UID and +complete spec and uses a UID deletion precondition; it cannot delete a replacement +pod merely because its name matches. + +Snapshots reuse the existing read-only SQLite backup path. The quiesced snapshot +also runs the existing no-provider replay checks in helper memory. Every snapshot +is private, fingerprinted, and checked for unexpected PVC consumers before and +after capture. Postflight permits only the independently verified receiver UID as +a live writer. Runtime verification checks the actual process sender, username, +caller-file mode and password presence, plus health/readiness and unauthenticated +request rejection. It prints no credential value and issues no accepted send. + +## Evidence and its limits + +`sender-native-preflight-2026-10-31.json` records five successful native server +dry-runs: pause, paused environment change, resume, sender ConfigMap change and +helper creation. The observed production configuration was unchanged after the +checks. PVC/PV/node identity and helper egress denial passed. Full proposal +preimages remain in the operator's private plan; the public receipt binds their +hash. This did not admit credentials, capture production data or alter Kubernetes +resources. Dry-run success does not prove a future rollout will succeed. + +`sender-native-snapshot-rehearsal-2026-10-31.json` records the actual snapshot and +replay paths running in the pinned receiver image with a synthetic ledger. The +container had no network, a read-only root, a read-only source mount, no published +ports and UID 10001. All six tables survived both captures. Accepted and unknown +invitation/verification outcomes, consumed/canceled links and suppression were +retained. The disposable container was removed. No Kubernetes API, production +data or SMTP provider was used. This is isolated snapshot integration evidence, +not a live writer-fence or end-to-end production transition proof. + +Thirty new adapter cases pass, covering identity/spec drift, private exclusive +locking, lost helper replies, UID-bound cleanup, unexpected PVC consumers, +metadata-only Secret responses and owner projection/admission refusal. The runtime +probe is exercised against the actual scoped WSGI application on local loopback +with a synthetic provider that records zero sends. Owner default suite: +**249 passed, 21 skipped in 23.51s**; skips are existing optional mail-harness +tests. Ruff passes. Log: `/tmp/email-sender-native-check-20261031.log`. + +## Remaining T02 work + +Bind the complete executor and dependencies into the final capsule, implement the +attended entry point with exact proposal digest and owner-return verification, +and prove the assembled forward/recovery path. Prepare the concrete execution and +recovery packet against fresh native preimages. Preserve the existing refusal of +blind forward replay and configuration-only recovery using the current ledger. +Only then can T02 finish; attended execution and independent return remain +EMAIL-WP-0010-T03/T04. Productive mail delivery, inbound handling and user verdicts +remain CSOC-WP-0015 obligations. diff --git a/docs/sender-native-preflight-2026-10-31.json b/docs/sender-native-preflight-2026-10-31.json new file mode 100644 index 0000000..1a9911a --- /dev/null +++ b/docs/sender-native-preflight-2026-10-31.json @@ -0,0 +1,43 @@ +{ + "schema": "email-connect.sender-native-preflight.v1", + "observed_at": "2026-10-31T15:54:10.156289+00:00", + "native_server_dry_runs": 5, + "cluster_identity_verified": true, + "namespace_identity_verified": true, + "deployment_identity_verified": true, + "ledger_pvc_pv_node_binding_verified": true, + "helper_read_only_and_egress_denied": true, + "only_reviewed_sender_environment_delta": true, + "application_changed": false, + "mail_sent": false, + "secret_values_requested": false, + "foundation_admission_executed": false, + "execution_ready": false, + "private_plan_sha256": "e660c5bb912d907c2a9dc2fc195f39f6d4507c1de912ec8d2514f4437c0d121d", + "sources": [ + { + "path": "scripts/sender_native.py", + "sha256": "41722d23f7d04a41c7fb428cb25a24438a144684cfbc5dcd875d5299ec6ffc71" + }, + { + "path": "scripts/sender_transition.py", + "sha256": "fa80b428b6231b74578baf23ff5977b6ce871a84321718adb2ba15d314f7abe1" + }, + { + "path": "scripts/capture-ledger.py", + "sha256": "7994fe7b4fbf72e28a4296212005a88d0ef9fcb626b066e6bcf2885500bc69d6" + }, + { + "path": "scripts/ledger_archive.py", + "sha256": "92d9d5e1908c7d433eb3f4ada6460f7d9b639017a4c2311cb7ffaacd262f33c3" + }, + { + "path": "scripts/inspect-sender-ledger.py", + "sha256": "262d6d52d5c10fb8583e0828ed34945b647f1d53bf9de5286374fef85cf76a8d" + }, + { + "path": "scripts/rehearse-ledger.py", + "sha256": "5c43b655149d8ae577fab80e76358be35f90cb0357df9b726ac7cb23346946a0" + } + ] +} diff --git a/docs/sender-native-snapshot-rehearsal-2026-10-31.json b/docs/sender-native-snapshot-rehearsal-2026-10-31.json new file mode 100644 index 0000000..c8c4553 --- /dev/null +++ b/docs/sender-native-snapshot-rehearsal-2026-10-31.json @@ -0,0 +1,136 @@ +{ + "schema": "email-connect.sender-native-snapshot-rehearsal.v1", + "image": "forgejo.coulomb.social/coulomb/email-connect@sha256:5d79fd1fec15d3a279383820f170d263de3a55f5308df01a0a34732ebaea6d3d", + "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": { + "accepted_invitations": 1, + "unknown_invitations": 1, + "failed_invitations": 0, + "accepted_verifications": 3, + "unknown_verifications": 1, + "failed_verifications": 0, + "unlinked_verifications": 0, + "unrecognized_attempt_states": 0, + "orphan_verification_attempts": 0, + "consumed_challenges": 1, + "canceled_challenges": 1, + "suppressed_recipients": 1 + }, + "continuity": { + "all_six_tables_checked": true, + "schema_unchanged": true, + "tables": { + "caller_buckets": { + "preserved_rows": 0, + "new_rows": 0, + "permitted_updates": 0 + }, + "deliveries": { + "preserved_rows": 1, + "new_rows": 0, + "permitted_updates": 0 + }, + "send_attempts": { + "preserved_rows": 2, + "new_rows": 0, + "permitted_updates": 0 + }, + "suppressions": { + "preserved_rows": 1, + "new_rows": 0, + "permitted_updates": 0 + }, + "verification_attempts": { + "preserved_rows": 4, + "new_rows": 0, + "permitted_updates": 0 + }, + "verifications": { + "preserved_rows": 4, + "new_rows": 0, + "permitted_updates": 0 + } + }, + "ledger_restored": false, + "snapshot_bytes_equal": false + }, + "rehearsal": { + "status": "isolated_ledger_verified", + "ledger_fingerprint": { + "schema_sha256": "a548e86495322b41fc4587cc6bbc03b182607d82dcde91a06f9085871566941f", + "user_version": 0, + "tables": { + "caller_buckets": { + "rows": 0, + "sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + }, + "deliveries": { + "rows": 1, + "sha256": "f46352824895fc3a207a5fb807303c407440095f605cd3242e880e202972461d" + }, + "send_attempts": { + "rows": 2, + "sha256": "bd63b2377bdf2372bc3faaa779c3fd997a2e15d0c200605790942086302caaef" + }, + "suppressions": { + "rows": 1, + "sha256": "f79a6f80ce153a87e543066fa17822c4b85d7728c2bfd8e576923171c7b0d7a3" + }, + "verification_attempts": { + "rows": 4, + "sha256": "b4f0b6403f63379755df9b0925b8e468a9533baa568cd1416f208ecbc7263f66" + }, + "verifications": { + "rows": 4, + "sha256": "2473b17fa0c12c0d7da91a0c81fbd894b7e9b28136e9728cac3e89fbca4bd245" + } + } + }, + "checks": { + "accepted_verifications": 3, + "unknown_verifications": 1, + "invalidated_handles": 2, + "suppressions": 1, + "accepted_deliveries": 1, + "unknown_sends": 1 + }, + "mail_sent": false, + "uid": 10001 + }, + "container_removed": true, + "sources": [ + { + "path": "scripts/rehearse-sender-native.py", + "sha256": "fbdff7070ad511214aa898602dc2626957900caa71728070395ac0b120248233" + }, + { + "path": "scripts/sender_native.py", + "sha256": "41722d23f7d04a41c7fb428cb25a24438a144684cfbc5dcd875d5299ec6ffc71" + }, + { + "path": "scripts/sender_transition.py", + "sha256": "fa80b428b6231b74578baf23ff5977b6ce871a84321718adb2ba15d314f7abe1" + }, + { + "path": "scripts/capture-ledger.py", + "sha256": "7994fe7b4fbf72e28a4296212005a88d0ef9fcb626b066e6bcf2885500bc69d6" + }, + { + "path": "scripts/ledger_archive.py", + "sha256": "92d9d5e1908c7d433eb3f4ada6460f7d9b639017a4c2311cb7ffaacd262f33c3" + }, + { + "path": "scripts/rehearse-ledger.py", + "sha256": "5c43b655149d8ae577fab80e76358be35f90cb0357df9b726ac7cb23346946a0" + }, + { + "path": "scripts/inspect-sender-ledger.py", + "sha256": "262d6d52d5c10fb8583e0828ed34945b647f1d53bf9de5286374fef85cf76a8d" + } + ] +} diff --git a/docs/sender-transition-protocol-2026-10-31.md b/docs/sender-transition-protocol-2026-10-31.md index 8b2f6e9..bea9c76 100644 --- a/docs/sender-transition-protocol-2026-10-31.md +++ b/docs/sender-transition-protocol-2026-10-31.md @@ -71,6 +71,11 @@ Ruff passes for both new Python files. Log: ## Required native completion +Update: the adapter and its native dry-run/isolated snapshot evidence are now +documented in [the native adapter review](sender-native-adapter-2026-10-31.md). +The requirements below remain the overall integration contract; the capsule, +attended entry point and assembled transition proof are still outstanding. + The adapter must implement complete UID/spec compare-and-swap operations, verify the RPF-WP-0068 custody return and current projection, hold the existing capture lock, prove the absence of other PVC writers, and bind each helper creation and diff --git a/scripts/rehearse-sender-native.py b/scripts/rehearse-sender-native.py new file mode 100644 index 0000000..880e80b --- /dev/null +++ b/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"]})) diff --git a/scripts/sender_native.py b/scripts/sender_native.py new file mode 100644 index 0000000..18200b6 --- /dev/null +++ b/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"]) diff --git a/tests/test_sender_native.py b/tests/test_sender_native.py new file mode 100644 index 0000000..e634d61 --- /dev/null +++ b/tests/test_sender_native.py @@ -0,0 +1,377 @@ +"""Adapter transport/fault tests use synthetic API objects, never a live cluster.""" + +import copy +import io +import json +import os +import subprocess +import sys +import threading +from pathlib import Path +from types import SimpleNamespace +from wsgiref.simple_server import make_server + +import pytest +from test_callers import application +from test_sender_transition import initial + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +import sender_native as native +from sender_transition import Journal, Refused + + +def obj(kind, name, *, spec=None, data=None, uid=None): + value = {"apiVersion": "v1", "kind": kind, + "metadata": {"name": name, "uid": uid or name + "-uid", "resourceVersion": "123"}} + if spec is not None: + value["spec"] = spec + if data is not None: + value["data"] = data + return value + + +class API(native.Native): + def __init__(self): + super().__init__() + before = initial() + before["deployment_spec"]["template"]["spec"]["volumes"] = [ + {"name": "caller-policy", "configMap": {"name": "email-connect-callers-user-engine-276257849bf2"}}] + self.objects = { + ("namespace", "kube-system"): obj("Namespace", "kube-system", uid=native.CLUSTER), + ("namespace", native.NS): obj("Namespace", native.NS, uid=native.NAMESPACE), + ("deployment", native.NS): obj("Deployment", native.NS, spec=before["deployment_spec"], uid=native.DEPLOYMENT), + ("configmap", "email-connect-config"): obj("ConfigMap", "email-connect-config", data=before["config_data"], uid=native.CONFIG), + ("pvc", native.PVC): obj("PersistentVolumeClaim", native.PVC, spec={"volumeName": "fixture-volume"}), + ("pv", "fixture-volume"): obj("PersistentVolume", "fixture-volume", spec={ + "claimRef": {"uid": native.PVC + "-uid"}, "nodeAffinity": {"required": { + "nodeSelectorTerms": [{"matchExpressions": [{"key": "kubernetes.io/hostname", + "operator": "In", "values": ["fixture-host"]}]}]}}}), + ("service", native.NS): obj("Service", native.NS, spec={"selector": {"app": native.NS}}), + ("externalsecret", "email-connect-runtime"): obj("ExternalSecret", "email-connect-runtime", spec={"retained": True}), + ("configmap", "email-connect-callers-user-engine-276257849bf2"): + obj("ConfigMap", "policy", data={"callers.json": "synthetic-policy"}), + } + dep = self.objects[("deployment", native.NS)] + dep["metadata"]["generation"] = 74 + dep["status"] = {"readyReplicas": 1, "observedGeneration": 74} + self.objects[("pvc", native.PVC)]["status"] = {"phase": "Bound"} + self.pods = {} + self.policies = [obj("NetworkPolicy", "deny-all", spec={"podSelector": {}, "policyTypes": ["Egress"]})] + self.calls = [] + self.loss = None + self.runtime_proof = None + + def metadata(self, name): + return {"uid": name + "-uid", "owners": []} + + def get(self, kind, name=None, *, namespace=native.NS): + if kind == "nodes": + node = obj("Node", "fixture-node") + node["metadata"]["labels"] = {"kubernetes.io/hostname": "fixture-host"} + return {"items": [node]} + if kind == "networkpolicies": + return {"items": copy.deepcopy(self.policies)} + if kind == "pods": + return {"items": copy.deepcopy(list(self.pods.values()))} + if kind == "pod": + return copy.deepcopy(self.pods.get(name)) + return copy.deepcopy(self.objects.get((kind, name))) + + def k(self, *args, body=None, timeout=180): + self.calls.append((args, body)) + if args[0] == "patch": + kind, name = args[1:3] + patch = json.loads(body) + value = copy.deepcopy(self.objects[(kind, name)]) + assert patch[0]["value"] == value["metadata"]["uid"] + assert patch[1]["value"] == value["metadata"]["resourceVersion"] + field = patch[2]["path"][1:] + assert value[field] == patch[2]["value"] + value[field] = patch[3]["value"] + if "--dry-run=server" not in args: + self.objects[(kind, name)] = value + return json.dumps(value).encode() + if args[0] == "create": + value = json.loads(body) + value["metadata"]["uid"] = "created-helper-uid" + value["metadata"]["resourceVersion"] = "456" + if "--dry-run=server" not in args: + self.pods[value["metadata"]["name"]] = value + if self.loss == "create": + raise OSError("lost create response") + return json.dumps(value).encode() + if args[0] == "delete": + name = next(a for a in args if a.startswith("--raw=")).rsplit("/", 1)[1] + assert json.loads(body)["preconditions"]["uid"] == self.pods[name]["metadata"]["uid"] + self.pods.pop(name) + return b"{}" + if args[0] == "exec": + code = args[-1] + compile(code, "runtime-probe", "exec") + self.runtime_proof = code + return b'{"mail_sent":false,"private_values_printed":false,"unauthenticated_denial":401}' + if args[0] in {"wait", "rollout"}: + return b"" + raise AssertionError(args) + + +@pytest.fixture +def api(): + return API() + + +@pytest.fixture +def armed(api, tmp_path): + private = tmp_path / "private" + private.mkdir(mode=0o700) + before = api.observe() + journal = Journal(private / "journal.jsonl", proposal={"before": before, "after": native.desired(before)}) + api.execute = True + api.journal = journal + api.lock_fd = 999 # Synthetic adapter only; real lock behavior has its own test. + try: + yield api, journal + finally: + journal.close() + + +def test_preparation_is_five_dry_runs_and_preserves_whole_spec(api): + before = copy.deepcopy(api.objects) + plan = api.prepare() + assert plan["native_server_dry_runs"] == 5 + assert api.objects == before and not api.pods + assert all("--dry-run=server" in args for args, _ in api.calls) + assert plan["before"]["protected"]["node"]["name"] == "fixture-node" + assert plan["after"]["deployment_spec"]["template"]["spec"]["containers"][0]["image"] == native.IMAGE + + +@pytest.mark.parametrize("drift", ["cluster", "namespace", "deployment", "config", "volume", "node"]) +def test_native_identity_or_volume_scope_drift_refuses(api, drift): + keys = {"cluster": ("namespace", "kube-system"), "namespace": ("namespace", native.NS), + "deployment": ("deployment", native.NS), "config": ("configmap", "email-connect-config")} + if drift in keys: + api.objects[keys[drift]]["metadata"]["uid"] = "foreign" + elif drift == "volume": + api.objects[("pv", "fixture-volume")]["spec"]["claimRef"]["uid"] = "other-pvc" + else: + api.objects[("pv", "fixture-volume")]["spec"]["nodeAffinity"]["required"]["nodeSelectorTerms"] = [] + with pytest.raises(Refused): + api.prepare() + assert not api.calls + + +def test_mutation_needs_armed_lock_and_admission_preparation_cannot_mutate(api): + current = api.observe() + api.execute = True + with pytest.raises(Refused, match="locked_execution"): + api.patch_deployment(current["deployment_spec"], dict(current["deployment_spec"], replicas=0)) + with pytest.raises(Refused, match="read_only"): + api.prepare() + assert not api.calls + + +def test_capture_lock_is_exclusive_and_private(api, tmp_path, monkeypatch): + private = tmp_path / "private" + private.mkdir(mode=0o700) + path = private / "capture.lock" + monkeypatch.setattr(native, "CAPTURE_LOCK", path) + api.execute = True + api.journal = SimpleNamespace() + api.lock_path = path + api.acquire() + other = API() + other.execute, other.journal, other.lock_path = True, SimpleNamespace(), path + try: + with pytest.raises(BlockingIOError): + other.acquire() + assert other.lock_fd is None + assert path.stat().st_mode & 0o777 == 0o600 + finally: + api.close() + + +def test_lost_helper_create_does_not_authorize_adoption_or_deletion(armed): + api, journal = armed + api.loss = "create" + with pytest.raises(OSError): + api.ensure_helper(journal) + assert not api.helpers() and len(api.pods) == 1 + with pytest.raises(Refused, match="unconfirmed_helper"): + api.ensure_helper(journal) + api.cleanup(journal) + assert len(api.pods) == 1 + assert not any(args[0] == "delete" for args, _ in api.calls) + + +def test_helper_cleanup_uses_recorded_uid_and_refuses_replacement(armed): + api, journal = armed + name = api.ensure_helper(journal) + assert api.helpers()[name]["uid"] == "created-helper-uid" + api.pods[name]["metadata"]["uid"] = "replacement" + with pytest.raises(Refused, match="helper_identity"): + api.cleanup(journal) + assert name in api.pods + api.pods[name]["metadata"]["uid"] = "created-helper-uid" + api.cleanup(journal) + assert not api.pods + assert journal.entries[-1]["phase"] == "native_helper_deleted" + + +@pytest.mark.parametrize("change", ["foreign_mount", "writable_helper", "egress"]) +def test_fence_refuses_extra_writer_or_changed_helper(armed, change): + api, journal = armed + api.objects[("deployment", native.NS)]["spec"]["replicas"] = 0 + name = api.ensure_helper(journal) + if change == "foreign_mount": + api.pods["other"] = obj("Pod", "other", spec={"volumes": [{"persistentVolumeClaim": {"claimName": native.PVC}}]}) + elif change == "writable_helper": + api.pods[name]["spec"]["containers"][0]["volumeMounts"][0]["readOnly"] = False + else: + api.policies[0]["spec"]["egress"] = [{}] + with pytest.raises((Refused, api.archive.Refused, api.capture.Refused)): + api.fence(journal) + + +def test_unrecorded_receiver_cannot_use_the_pvc_during_postflight(armed): + api, _ = armed + api.pods["receiver"] = obj("Pod", "receiver", uid="other-uid", spec={ + "volumes": [{"persistentVolumeClaim": {"claimName": native.PVC}}]}) + api.receiver_uid = "verified-uid" + with pytest.raises(Refused, match="unowned_ledger"): + api.check_consumers(allow_receiver=True) + api.pods["receiver"]["metadata"]["uid"] = "verified-uid" + api.check_consumers(allow_receiver=True) + + +def test_runtime_probe_checks_actual_owner_app_without_provider_send(armed, tmp_path): + api, journal = armed + target = journal.entries[0]["proposal"]["before"] + pod = obj("Pod", "receiver", uid="receiver-uid", spec={"containers": copy.deepcopy( + target["deployment_spec"]["template"]["spec"]["containers"])}) + pod["metadata"].update(labels={"app.kubernetes.io/name": native.NS}, + ownerReferences=[{"kind": "ReplicaSet", "name": "rs", "uid": "rs-uid", "controller": True}]) + pod["status"] = {"phase": "Running", "containerStatuses": [{"ready": True, "imageID": native.IMAGE}]} + rs = obj("ReplicaSet", "rs", uid="rs-uid") + rs["metadata"]["ownerReferences"] = [{"uid": native.DEPLOYMENT, "controller": True}] + api.objects[("replicaset", "rs")] = rs + api.pods["receiver"] = pod + api.verify(target, journal) + app, store, provider = application(tmp_path) + server = make_server("127.0.0.1", 0, app) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + code = api.runtime_proof.replace("127.0.0.1:8080", f"127.0.0.1:{server.server_port}") + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, + env={**os.environ, "EMAIL_CONNECT_SENDER": "forgejo@coulomb.social", + "EMAIL_CONNECT_SMTP_USERNAME": "forgejo@coulomb.social", + "EMAIL_CONNECT_SMTP_PASSWORD": "synthetic-private-password", + "EMAIL_CONNECT_CALLERS_FILE": "/etc/email-connect-callers/callers.json"}, + timeout=15, check=False) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["unauthenticated_denial"] == 401 + assert "synthetic-private-password" not in result.stdout + result.stderr + assert not provider.calls + assert not store.db.execute("SELECT * FROM deliveries").fetchall() + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + store.db.close() + + +@pytest.mark.parametrize("response_kind", ["metadata", "full_secret", "missing"]) +def test_metadata_content_negotiation_never_falls_back_to_secret_values(tmp_path, monkeypatch, response_kind): + calls, cleanup = [], [] + class Process: + def poll(self): return None + def terminate(self): cleanup.append("terminated") + def wait(self, timeout): return 0 + class Socket: + def settimeout(self, value): pass + def connect(self, path): pass + def close(self): pass + class Connection: + def __init__(self, *args, **kwargs): pass + def request(self, method, path, headers): calls.append((method, path, headers)) + def getresponse(self): + data = {"kind": "PartialObjectMetadata", "apiVersion": "meta.k8s.io/v1", "metadata": {"uid": "fixture"}} + if response_kind == "full_secret": data = {"kind": "Secret", "data": {"password": "private"}} + response = io.BytesIO(json.dumps(data).encode()) + response.status = 404 if response_kind == "missing" else 200 + return response + def close(self): pass + monkeypatch.setattr(native.subprocess, "Popen", lambda *a, **kw: Process()) + monkeypatch.setattr(native.socket, "socket", lambda *a: Socket()) + monkeypatch.setattr(native.http.client, "HTTPConnection", Connection) + monkeypatch.setattr(native.time, "sleep", lambda _: None) + api = native.Native() + if response_kind == "full_secret": + with pytest.raises(Refused, match="metadata_only_response"): + api.metadata("email-connect-platform-smtp") + else: + value = api.metadata("email-connect-platform-smtp") + assert value == (None if response_kind == "missing" else {"uid": "fixture", "owners": []}) + assert len(calls) == 1 and "PartialObjectMetadata" in calls[0][2]["Accept"] + assert cleanup == ["terminated"] + + +def test_secret_full_queries_and_unknown_metadata_names_refuse(): + api = native.Native() + with pytest.raises(Refused, match="outside_reviewed_scope"): + api.get("secrets", "anything") + with pytest.raises(Refused, match="metadata_scope_drift"): + api.metadata("unrelated") + + +@pytest.mark.parametrize("failure", [None, "status", "digest", "scope", "identity"]) +def test_admission_requires_complete_owner_receipt_and_matching_native_identity(armed, monkeypatch, failure): + api, journal = armed + receipt = {"schema": "platform.email-connect-smtp-foundation.v1", + "proposal_digest": native.FOUNDATION, + "status": "foundation_admitted_projection_ready", + "application_sender_changed": False, "service_account_uid": "admitted-sa"} + 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"): + receipt[key] = True + if failure == "status": receipt["status"] = "started" + if failure == "digest": receipt["proposal_digest"] = "other" + if failure == "scope": receipt["exact_read_verified"] = False + if failure == "identity": receipt["service_account_uid"] = "different-sa" + pins = {"ServiceAccount": "admitted-sa", "Secret": "new-secret"} + monkeypatch.setattr(api, "projection", lambda proposal: pins) + if failure: + with pytest.raises(Refused): + api.admission({"foundation_receipt": receipt}) + assert not api.calls and api.foundation_pins is None + else: + api.admission({"foundation_receipt": receipt}) + assert api.foundation_pins == pins + assert journal.entries[-1]["phase"] == "native_foundation_verified" + + +@pytest.mark.parametrize("failure", [None, "spec", "ready", "owner", "account"]) +def test_projection_checks_exact_specs_readiness_and_metadata_owner(api, monkeypatch, failure): + foundation = {"resources": [ + {"kind": "ServiceAccount", "apiVersion": "v1", "metadata": {"name": "sa"}}, + {"kind": "SecretStore", "apiVersion": "external-secrets.io/v1", "metadata": {"name": "store"}, "spec": {"scope": "fixed"}}, + {"kind": "ExternalSecret", "apiVersion": "external-secrets.io/v1", "metadata": {"name": "projection"}, "spec": {"scope": "fixed"}}, + ]} + monkeypatch.setattr(native, "FOUNDATION_SPEC", native.digest(foundation)) + for wanted in foundation["resources"]: + value = obj(wanted["kind"], wanted["metadata"]["name"], spec=wanted.get("spec")) + value["apiVersion"] = wanted["apiVersion"] + if wanted["kind"] == "ServiceAccount": value["automountServiceAccountToken"] = False + else: value["status"] = {"conditions": [{"type": "Ready", "status": "True"}], "refreshTime": "fixture"} + api.objects[(wanted["kind"], wanted["metadata"]["name"])] = value + metadata = {"uid": "secret-uid", "owners": [{"uid": "projection-uid", "controller": True}]} + monkeypatch.setattr(api, "metadata", lambda name: metadata) + if failure == "spec": api.objects[("SecretStore", "store")]["spec"] = {"scope": "expanded"} + if failure == "ready": api.objects[("ExternalSecret", "projection")]["status"]["conditions"] = [] + if failure == "owner": metadata["owners"][0]["uid"] = "someone-else" + if failure == "account": api.objects[("ServiceAccount", "sa")]["automountServiceAccountToken"] = True + if failure: + with pytest.raises(Refused): api.projection({"foundation": foundation}) + else: + assert api.projection({"foundation": foundation})["Secret"] == "secret-uid" diff --git a/workplans/EMAIL-WP-0010-platform-sender-transition.md b/workplans/EMAIL-WP-0010-platform-sender-transition.md index 54d50f4..d27a668 100644 --- a/workplans/EMAIL-WP-0010-platform-sender-transition.md +++ b/workplans/EMAIL-WP-0010-platform-sender-transition.md @@ -86,14 +86,27 @@ state, source/configuration drift, absent admission, journal privacy/truncation, unexpected writer resume and preservation of consumed links, quotas and new rows. All 27 new cases pass; owner default suite 219 passed / 21 existing optional harness skips. -This is the protocol core, not a production executable or completed review packet. -The native adapter, identity/projection/admission verification, shared capture lock, -PVC fencing, source-bound helper capture/cleanup, runtime checks, capsule/digest, -attended entry point and native server dry-runs remain in T02. Existing capture -and projection tools are the integration surfaces; no credential/deployment/source -packet or native configuration was changed. Review: +The protocol-core review is `docs/sender-transition-protocol-2026-10-31.md`. +2026-10-31 native adapter follow-through: `scripts/sender_native.py` now implements +UID/spec compare-and-swap, metadata-only Secret queries, exact foundation receipt +and projection checks, the existing exclusive capture lock, PVC/PV/node binding, +writer fencing, durable helper identities, UID-preconditioned cleanup and actual +runtime/no-send checks. Five native server dry-runs passed without a configuration +change. The actual snapshot/replay path passed in the pinned receiver image with +a synthetic ledger, networking disabled and the source mounted read-only; its +disposable container was removed. Thirty new adapter cases pass; full owner suite +249 passed / 21 existing optional harness skips, Ruff clean. Public receipts bind +the source hashes and scope of each proof. Review: +`docs/sender-native-adapter-2026-10-31.md`. + +T02 remains progress: the final source-bound capsule, attended entry point, +assembled forward/recovery proof and exact execution/recovery packet are still +required. Native dry-runs and isolated snapshot integration do not establish a +production transition. Existing backup and foundation source packets remain +unchanged; no live grant, sender change, capture or mail send occurred. + ## T03 — Execute the admitted sender switch ```task