"""Only synthetic ledger records; no production URLs, custody or SMTP sends.""" import copy import hashlib import importlib.util import io import json import sqlite3 import subprocess import sys import tarfile from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from email_connect.transactional import SQLiteDeliveryStore, TransactionalApplication @pytest.fixture def modules(monkeypatch): scripts = Path(__file__).resolve().parents[1] / "scripts" monkeypatch.syspath_prepend(str(scripts)) result = [] for name in ["ledger_archive", "capture-ledger", "backup-ledger"]: spec = importlib.util.spec_from_file_location( name.replace("-", "_"), scripts / (name + ".py") ) module = importlib.util.module_from_spec(spec) monkeypatch.setitem(sys.modules, name.replace("-", "_"), module) spec.loader.exec_module(module) result.append(module) return result def synthetic_store(path): path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) store = SQLiteDeliveryStore(str(path)) path.chmod(0o600) accepted = { "registration_id": "accepted", "normalized_email": "accepted@example.test", "preferred_username": "synthetic", "client_id": "synthetic-client", "tenant": "synthetic-tenant", "correlation_id": "trace-1", } expires = (datetime.now(UTC) + timedelta(days=1)).isoformat() def reserve(payload, handle): return store.reserve_verification( payload, hashlib.sha256(handle.encode()).hexdigest(), expires )[0] request = reserve(accepted, "accepted-handle") store.verification_accepted(request, "synthetic-provider-reference") reserve(dict(accepted, registration_id="unknown"), "unknown-handle") consumed = reserve(dict(accepted, registration_id="consumed"), "consumed-handle") store.verification_accepted(consumed, "synthetic-consumed-reference") store.consume_verification(hashlib.sha256(b"consumed-handle").hexdigest()) canceled = reserve(dict(accepted, registration_id="canceled"), "canceled-handle") store.verification_accepted(canceled, "synthetic-canceled-reference") store.cancel_verification(hashlib.sha256(b"canceled-handle").hexdigest()) store.reserve_send("accepted-invitation", "synthetic-accepted-fingerprint") store.record("accepted-invitation", "synthetic-invitation-reference") store.reserve_send("unknown-invitation", "synthetic-unknown-fingerprint") store.suppress("suppressed@example.test", "synthetic complaint") store.db.close() return accepted def captured(directory, archive, image="synthetic@sha256:fixture"): directory.mkdir(mode=0o700) synthetic_store(directory / "database.sqlite3") receipt = { "status": "captured", "application": "email-connect", "image": image, "captured_at": "2026-10-07T20:00:00Z", "elapsed_seconds": 30, "writers_quiesced": True, "receiver_resumed": True, "source_pvc_read_only": True, "database_sha256": archive.digest(directory / "database.sqlite3"), "database_bytes": (directory / "database.sqlite3").stat().st_size, "ledger_fingerprint": archive.fingerprint(directory / "database.sqlite3"), } archive.write_private(directory / "receipt.json", json.dumps(receipt).encode()) return receipt def invoke(app, payload, path): result = {} raw = json.dumps(payload).encode() environ = { "PATH_INFO": path, "REQUEST_METHOD": "POST", "CONTENT_LENGTH": str(len(raw)), "wsgi.input": io.BytesIO(raw), "HTTP_AUTHORIZATION": "Bearer synthetic-token", } body = json.loads(b"".join(app(environ, lambda status, headers: result.update(status=status)))) return result["status"], body def test_sqlite_backup_includes_committed_wal_without_source_checkpoint(modules, tmp_path): archive, _, _ = modules source = tmp_path / "source.sqlite3" synthetic_store(source) with sqlite3.connect(source) as writer: writer.execute("PRAGMA journal_mode=WAL") writer.execute("INSERT INTO deliveries VALUES ('wal-event','wal-reference')") writer.commit() original_bytes = source.read_bytes() before = archive.fingerprint(source) output = tmp_path / "private" / "database.sqlite3" output.parent.mkdir(mode=0o700) assert archive.sqlite_backup(source, output) == before assert source.read_bytes() == original_bytes with sqlite3.connect(output) as restored: assert restored.execute( "SELECT provider_ref FROM deliveries WHERE event_id='wal-event'" ).fetchone() == ("wal-reference",) writer.execute("INSERT INTO deliveries VALUES ('uncommitted','private')") output2 = output.parent / "another.sqlite3" archive.sqlite_backup(source, output2) with sqlite3.connect(output2) as restored: assert ( restored.execute("SELECT 1 FROM deliveries WHERE event_id='uncommitted'").fetchone() is None ) writer.rollback() def test_real_encryption_recovery_preserves_challenge_and_delivery_behavior(modules, tmp_path): archive, _, _ = modules stage = tmp_path / "stage" stage.mkdir(mode=0o700) receipt = captured(stage / "capture", archive) identity = tmp_path / "synthetic-identity" subprocess.run(["age-keygen", "-o", str(identity)], capture_output=True, check=True) public = ( subprocess.run(["age-keygen", "-y", str(identity)], capture_output=True, check=True) .stdout.decode() .strip() ) encrypted, _ = archive.seal(stage / "capture", stage, public) first_ciphertext = encrypted.read_bytes() assert ( archive.seal(stage / "capture", stage, "unused-retry-recipient")[0].read_bytes() == first_ciphertext ) transfer = dict( receipt, schema="platform.application-primary-archive.v1", status="offsite_fetched_verified", remote_retrieval_verified=True, version_id="synthetic-version", object_key="platform-pg/application-archives/email-connect/2026/10/07/20261007T200000Z-" + "a" * 24 + ".tar.age", destination=archive.PREFIX + "2026/10/07/20261007T200000Z-" + "a" * 24 + ".tar.age", ciphertext_sha256=archive.digest(encrypted), ciphertext_bytes=encrypted.stat().st_size, ) transfer_path = tmp_path / "transfer.json" archive.write_private(transfer_path, json.dumps(transfer).encode()) destination = tmp_path / "recovered" result = archive.recover(encrypted, transfer_path, identity, destination) assert result["status"] == "decrypted_ledger_verified" and not result["production_restored"] assert not (destination / "packet.tar").exists() assert all(p.stat().st_mode & 0o777 == 0o600 for p in destination.iterdir()) path = destination / "database.sqlite3" before = archive.fingerprint(path) store = SQLiteDeliveryStore(str(path)) assert archive.fingerprint(path) == before class ForbiddenProvider: def send(self, *args): raise AssertionError("recovery must not resend") app = TransactionalApplication( store, ForbiddenProvider(), "synthetic-token", "https://issuer.example.test" ) payload = { "registration_id": "accepted", "normalized_email": "accepted@example.test", "preferred_username": "synthetic", "client_id": "synthetic-client", "tenant": "synthetic-tenant", "correlation_id": "another-trace", } status, response = invoke(app, payload, "/v1/registration-verifications") assert status.startswith("202") and response["accepted"] status, response = invoke( app, dict(payload, registration_id="unknown"), "/v1/registration-verifications" ) assert status.startswith("422") and response["error"] == "delivery_outcome_unknown" for handle in ["consumed-handle", "canceled-handle"]: _, response = invoke(app, {"handle": handle}, "/v1/registration-verifications/consume") assert response["error"] == "verification_invalid" assert store.reference("accepted-invitation") == "synthetic-invitation-reference" assert store.is_suppressed("suppressed@example.test") store.db.close() assert archive.fingerprint(path) == before @pytest.mark.parametrize("bad", ["duplicate", "traversal", "link", "extra"]) def test_packet_refuses_all_unsafe_members_before_extract(modules, tmp_path, bad): archive, _, _ = modules packet = tmp_path / "packet.tar" with tarfile.open(packet, "w") as output: for name in ["database.sqlite3", "receipt.json"]: entry = tarfile.TarInfo(name) entry.size = 1 if bad == "link" and name == "database.sqlite3": entry.type, entry.linkname, entry.size = tarfile.SYMTYPE, "/tmp/private", 0 output.addfile(entry, io.BytesIO(b"x")) if bad != "link": entry = tarfile.TarInfo( {"duplicate": "database.sqlite3", "traversal": "../escape", "extra": "smtp.env"}[ bad ] ) entry.size = 1 output.addfile(entry, io.BytesIO(b"x")) destination = tmp_path / "out" destination.mkdir(mode=0o700) with pytest.raises(archive.Refused, match="invalid_packet_members"): archive.unpack(packet, destination) assert not list(destination.iterdir()) def deployment(capture): return { "spec": {"replicas": 1, "template": {"spec": {"containers": [{"image": capture.IMAGE}]}}}, "status": {"readyReplicas": 1}, } @pytest.mark.parametrize( "failure", [None, "capture", "interrupt", "helper-cleanup", "foreign-image"] ) def test_capture_resumes_unchanged_receiver_after_failures(modules, monkeypatch, tmp_path, failure): archive, capture, _ = modules before = deployment(capture) current = copy.deepcopy(before) calls = [] def k(*args, **kwargs): calls.append(args) if args[0] == "patch": patches = json.loads(kwargs["body"]) assert patches[0]["value"] == before["spec"]["template"] assert patches[1]["value"] == current["spec"]["replicas"] current["spec"]["replicas"] = patches[2]["value"] if args[0] == "delete" and failure == "helper-cleanup": raise capture.Refused("cleanup failed") return b"" def document(*args, **kwargs): return {"items": []} if args[1] == "pods" else copy.deepcopy(current) def capture_database(name, destination): if failure == "capture": raise capture.Refused("capture failed") if failure == "interrupt": raise KeyboardInterrupt() if failure == "foreign-image": current["spec"]["template"]["spec"]["containers"][0]["image"] = "foreign-image" synthetic_store(destination / "database.sqlite3") path = destination / "database.sqlite3" return { "database_sha256": archive.digest(path), "database_bytes": path.stat().st_size, "ledger_fingerprint": archive.fingerprint(path), "writers_quiesced": True, "source_pvc_read_only": True, } monkeypatch.setattr(capture, "k", k) monkeypatch.setattr(capture, "document", document) monkeypatch.setattr(capture, "capture_database", capture_database) destination = tmp_path / "capture" body = json.dumps(capture.helper("synthetic-capture")).encode() if failure: with pytest.raises((archive.Refused, KeyboardInterrupt)): capture.capture(before, body, destination) else: assert capture.capture(before, body, destination)["status"] == "captured" receipt = json.loads((destination / "receipt.json").read_text()) assert any(args[0] == "delete" for args in calls) if failure == "foreign-image": assert current["spec"]["replicas"] == 0 and receipt["receiver_state_requires_check"] assert not receipt.get("receiver_resumed") else: assert current == before and receipt["receiver_resumed"] assert receipt["status"] == ("captured" if failure is None else "failed") @pytest.mark.parametrize("failure", [None, "publish", "encrypt", "mirror"]) def test_backup_success_requires_readback_and_preserves_one_pending_capture( modules, monkeypatch, tmp_path, failure ): archive, _, runner = modules captures, publications = [], [] original_seal = runner.seal state = tmp_path / "state" def command(argv, **kwargs): if "--primary-receipt" in argv: if failure == "mirror": raise archive.Refused("synthetic mirror failure") archive.checkpoint(Path(argv[-1]), {"hash_readback_verified": True}) return captures.append(argv) captured(Path(argv[-1]), archive) def seal(directory, stage, recipient): if failure == "encrypt": raise archive.Refused("synthetic encryption failure") return original_seal(directory, stage, recipient) def publish(encrypted, fetched, receipt_path, capture, tool, python): publications.append(encrypted.read_bytes()) if failure == "publish": raise archive.Refused("synthetic readback failure") archive.write_private(fetched, encrypted.read_bytes()) key = ( "platform-pg/application-archives/email-connect/2026/10/07/20261007T200000Z-" + "b" * 24 + ".tar.age" ) return { "schema": "platform.application-primary-archive.v1", "application": "email-connect", "status": "offsite_fetched_verified", "remote_retrieval_verified": True, "version_id": "synthetic-version", "object_key": key, "destination": archive.PREFIX + key.rsplit("/", 1)[-1], "captured_at": capture["captured_at"], "ciphertext_sha256": archive.digest(encrypted), "ciphertext_bytes": encrypted.stat().st_size, "immutable_create": True, } identity = tmp_path / "synthetic-identity" subprocess.run(["age-keygen", "-o", str(identity)], capture_output=True, check=True) public = ( subprocess.run(["age-keygen", "-y", str(identity)], capture_output=True, check=True) .stdout.decode() .strip() ) monkeypatch.setattr(runner, "command", command) monkeypatch.setattr(runner, "seal", seal) monkeypatch.setattr(runner, "publish", publish) if failure: with pytest.raises(archive.Refused): runner.backup(state, Path("synthetic-tool"), Path("synthetic-python"), public, mirror_required=failure == "mirror") assert not (state / "last-success.json").exists() assert len(list(state.glob("pending-*"))) == 1 with pytest.raises(archive.Refused, match="pending_backup_requires_review"): runner.backup(state, Path("synthetic-tool"), Path("synthetic-python"), public) if failure in {"publish", "mirror"}: mirror_required = failure == "mirror" ciphertext = publications[0] failure = None result = runner.backup( state, Path("synthetic-tool"), Path("synthetic-python"), public, resume=True, mirror_required=mirror_required, ) assert len(captures) == 1 and publications == [ciphertext, ciphertext] assert result["application_restore_verified"] is False else: result = runner.backup(state, Path("synthetic-tool"), Path("synthetic-python"), public) assert not list(state.glob("pending-*")) assert not list(state.rglob("database.sqlite3")) assert json.loads((state / "inventory.json").read_text()) == [result] assert runner.freshness(state, datetime(2026, 10, 8, 20, tzinfo=UTC))["status"] == "fresh" with pytest.raises(archive.Refused, match="backup_stale"): runner.freshness(state, datetime(2026, 10, 9, 20, tzinfo=UTC)) (state / result["local_archive"]).write_bytes(b"corruption") with pytest.raises(archive.Refused, match="local_ciphertext_mismatch"): runner.freshness(state, datetime(2026, 10, 8, 20, tzinfo=UTC)) @pytest.mark.parametrize( "field,value", [ ("receiver_resumed", False), ("database_sha256", "bad"), ("ledger_fingerprint", {}), ("source_pvc_read_only", False), ], ) def test_invalid_capture_cannot_be_sealed(modules, tmp_path, field, value): archive, _, _ = modules stage = tmp_path / "stage" stage.mkdir(mode=0o700) receipt = captured(stage / "capture", archive) receipt[field] = value (stage / "capture" / "receipt.json").write_text(json.dumps(receipt)) with pytest.raises(archive.Refused): archive.seal(stage / "capture", stage, "unused public recipient") assert not (stage / "packet.tar").exists() def test_helper_has_no_secret_or_token_mount_and_requires_no_egress(modules): _, capture, _ = modules pod = capture.helper("synthetic-helper") container = pod["spec"]["containers"][0] assert not pod["spec"]["automountServiceAccountToken"] assert "env" not in container and "envFrom" not in container assert container["volumeMounts"][0]["readOnly"] labels = pod["metadata"]["labels"] deny = {"spec": {"policyTypes": ["Ingress", "Egress"], "podSelector": {}}} scoped = { "spec": { "policyTypes": ["Egress"], "podSelector": {"matchLabels": {"app.kubernetes.io/name": "email-connect"}}, "egress": [{}], } } capture.require_no_helper_egress([deny, scoped], labels) with pytest.raises(capture.Refused, match="helper_egress_not_denied"): capture.require_no_helper_egress([], labels) broad = copy.deepcopy(scoped) broad["spec"]["podSelector"] = {} with pytest.raises(capture.Refused, match="helper_egress_not_denied"): capture.require_no_helper_egress([deny, broad], labels) assert capture.selected( {"matchExpressions": [{"key": "email-ledger-capture", "operator": "Exists"}]}, labels ) assert not capture.selected( { "matchExpressions": [ {"key": "email-ledger-capture", "operator": "In", "values": ["other"]} ] }, labels, ) def test_scoped_ledger_rehearsal_preserves_ownership_and_budgets(modules, tmp_path, monkeypatch, capsys): archive, _, _ = modules path = tmp_path / "source.sqlite3" synthetic_store(path) store = SQLiteDeliveryStore(str(path)) payload = { "registration_id": "scoped", "normalized_email": "scoped@example.test", "preferred_username": "synthetic", "client_id": "synthetic-client", "tenant": "synthetic-tenant", "correlation_id": "trace-scoped", } expires = (datetime.now(UTC) + timedelta(days=1)).isoformat() request, _ = store.reserve_verification( payload, hashlib.sha256(b"scoped-handle").hexdigest(), expires, caller_id="caller-a" ) store.verification_accepted(request, "synthetic-scoped-reference") store.reserve_verification( dict(payload, registration_id="scoped-unknown"), hashlib.sha256(b"scoped-unknown-handle").hexdigest(), expires, caller_id="caller-b" ) assert store.request_budget("caller-a", rate=2, burst=1, now=100)[0] store.db.close() before = archive.fingerprint(path) assert before["tables"]["caller_buckets"]["rows"] == 1 script = Path(__file__).resolve().parents[1] / "scripts/rehearse-ledger.py" spec = importlib.util.spec_from_file_location("rehearsal", script) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) recovered = tmp_path / "recovered.sqlite3" code = module.CODE.replace("Path('/tmp/recovered.sqlite3')", "Path(" + repr(str(recovered)) + ")") monkeypatch.setattr(sys, "stdin", io.TextIOWrapper(io.BytesIO(path.read_bytes()))) monkeypatch.setattr(__import__("os"), "getuid", lambda: 10001) exec(compile(code, str(script), "exec"), {}) receipt = json.loads(capsys.readouterr().out) assert receipt["checks"]["accepted_verifications"] == 4 assert receipt["checks"]["unknown_verifications"] == 2 assert receipt["mail_sent"] is False assert archive.fingerprint(recovered) == before restarted = SQLiteDeliveryStore(str(recovered)) assert not restarted.request_budget("caller-a", rate=2, burst=1, now=100)[0] restarted.db.close()