Prepare independent ledger backup and isolated no-send recovery

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d

scripts/backup-ledger.py

@@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+"""Node ledger backup or attended recovery; default performs preview only."""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+import re
+import shutil
+import signal
+import sys
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+
+from ledger_archive import (
+    PREFIX,
+    Refused,
+    checkpoint,
+    command,
+    digest,
+    private_directory,
+    private_file,
+    recover,
+    seal,
+)
+
+NAME = re.compile(r"\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age")
+PRIMARY_TOOL = Path("/opt/railiance-platform/scripts/primary_application_archive.py")
+PRIMARY_PYTHON = Path("/opt/railiance-platform/backup-venv/bin/python")
+RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"
+
+
+def publish(encrypted, fetched, receipt_path, captured, tool, python):
+    command(
+        [
+            str(python),
+            str(tool),
+            "transfer",
+            "--application",
+            "email-connect",
+            "--source",
+            str(encrypted),
+            "--output",
+            str(fetched),
+            "--receipt",
+            str(receipt_path),
+            "--captured-at",
+            captured["captured_at"],
+            "--execute",
+        ]
+    )
+    private_file(fetched)
+    receipt = json.loads(receipt_path.read_text())
+    if (
+        receipt.get("schema") != "platform.application-primary-archive.v1"
+        or receipt.get("status") != "offsite_fetched_verified"
+        or receipt.get("application") != "email-connect"
+        or receipt.get("captured_at") != captured["captured_at"]
+        or receipt.get("remote_retrieval_verified") is not True
+        or receipt.get("immutable_create") is not True
+        or not receipt.get("version_id")
+        or receipt["version_id"] == "null"
+        or receipt.get("ciphertext_sha256") != digest(encrypted)
+        or receipt.get("ciphertext_sha256") != digest(fetched)
+        or receipt.get("ciphertext_bytes") != encrypted.stat().st_size
+        or not receipt.get("destination", "").startswith(PREFIX)
+    ):
+        raise Refused("primary_receipt_not_verified")
+    return receipt
+
+
+def backup(state, primary_tool, primary_python, recipient, resume=False):
+    state.mkdir(mode=0o700, parents=True, exist_ok=True)
+    private_directory(state)
+    fd = os.open(state / "backup.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(fd, "w") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            raise Refused("backup_already_running") from None
+        pending = list(state.glob("pending-*"))
+        if resume:
+            if len(pending) != 1:
+                raise Refused("one_pending_capture_required")
+            stage = pending[0]
+            private_directory(stage)
+        else:
+            if pending:
+                raise Refused("pending_backup_requires_review")
+            stage = Path(tempfile.mkdtemp(prefix="pending-", dir=state))
+            command(
+                [
+                    sys.executable,
+                    str(Path(__file__).with_name("capture-ledger.py")),
+                    "--execute",
+                    "--destination",
+                    str(stage / "capture"),
+                ],
+                timeout=1800,
+            )
+        encrypted, captured = seal(stage / "capture", stage, recipient)
+        fd, name = tempfile.mkstemp(prefix="readback-", dir=stage)
+        os.close(fd)
+        fetched = Path(name)
+        fetched.unlink()
+        transfer = publish(
+            encrypted,
+            fetched,
+            stage / "primary-transfer.json",
+            captured,
+            primary_tool,
+            primary_python,
+        )
+        filename = transfer["object_key"].rsplit("/", 1)[-1]
+        if not NAME.fullmatch(filename):
+            raise Refused("invalid_archive_name")
+        receipt = dict(
+            transfer,
+            image=captured["image"],
+            database_sha256=captured["database_sha256"],
+            database_bytes=captured["database_bytes"],
+            ledger_fingerprint=captured["ledger_fingerprint"],
+            pause_and_capture_seconds=captured["elapsed_seconds"],
+            local_archive=filename,
+            application_restore_verified=False,
+            completed_at=datetime.now(UTC).isoformat(),
+        )
+        output = state / filename
+        if output.exists() or output.is_symlink():
+            private_file(output)
+            if digest(output) != receipt["ciphertext_sha256"]:
+                raise Refused("local_archive_collision")
+        else:
+            # Create private file from the start, including manual invocations.
+            fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+            with os.fdopen(fd, "wb") as dst, fetched.open("rb") as src:
+                shutil.copyfileobj(src, dst)
+                dst.flush()
+                os.fsync(dst.fileno())
+        inventory_path = state / "inventory.json"
+        inventory = json.loads(inventory_path.read_text()) if inventory_path.exists() else []
+        if not any(item["object_key"] == receipt["object_key"] for item in inventory):
+            inventory.append(receipt)
+        checkpoint(inventory_path, inventory)
+        checkpoint(state / "last-success.json", receipt)
+        shutil.rmtree(stage)
+        return receipt
+
+
+def freshness(state, now=None, hours=26):
+    now = now or datetime.now(UTC)
+    private_directory(state)
+    private_file(state / "last-success.json")
+    receipt = json.loads((state / "last-success.json").read_text())
+    stamp = datetime.fromisoformat(receipt["captured_at"].replace("Z", "+00:00"))
+    age = (now - stamp).total_seconds()
+    if not 0 <= age <= hours * 3600:
+        raise Refused("backup_stale")
+    name = receipt.get("local_archive", "")
+    if (
+        receipt.get("status") != "offsite_fetched_verified"
+        or receipt.get("application") != "email-connect"
+        or receipt.get("remote_retrieval_verified") is not True
+        or not receipt.get("version_id")
+        or receipt["version_id"] == "null"
+        or not NAME.fullmatch(name)
+    ):
+        raise Refused("verified_backup_required")
+    private_file(state / name)
+    if digest(state / name) != receipt["ciphertext_sha256"]:
+        raise Refused("local_ciphertext_mismatch")
+    return {"status": "fresh", "age_hours": round(age / 3600, 2), "remote_retrieval_verified": True}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("action", choices=["backup", "recover", "freshness"])
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--resume-pending", action="store_true")
+    parser.add_argument("--state-dir", type=Path, default=Path("/var/lib/email-connect-backup"))
+    parser.add_argument("--primary-tool", type=Path, default=PRIMARY_TOOL)
+    parser.add_argument("--primary-python", type=Path, default=PRIMARY_PYTHON)
+    parser.add_argument("--age-recipient", default=RECIPIENT)
+    parser.add_argument("--archive", type=Path)
+    parser.add_argument("--transfer-receipt", type=Path)
+    parser.add_argument("--identity", type=Path)
+    parser.add_argument("--destination", type=Path)
+    args = parser.parse_args()
+    if args.action == "freshness":
+        print(json.dumps(freshness(args.state_dir)))
+    elif args.action == "recover":
+        if not args.execute:
+            print(json.dumps({"status": "reviewed", "decryption_performed": False}))
+            return 0
+        if not all([args.archive, args.transfer_receipt, args.identity, args.destination]):
+            raise Refused("recovery_arguments_required")
+        print(
+            json.dumps(
+                recover(args.archive, args.transfer_receipt, args.identity, args.destination)
+            )
+        )
+    elif args.execute:
+        if (
+            not args.state_dir.is_absolute()
+            or not args.primary_tool.is_file()
+            or not args.primary_python.is_file()
+        ):
+            raise Refused("invalid_backup_configuration")
+        command(["age", "--version"])
+        command(
+            [
+                str(args.primary_python),
+                str(args.primary_tool),
+                "check",
+                "--application",
+                "email-connect",
+            ]
+        )
+        print(
+            json.dumps(
+                backup(
+                    args.state_dir,
+                    args.primary_tool,
+                    args.primary_python,
+                    args.age_recipient,
+                    args.resume_pending,
+                )
+            )
+        )
+    else:
+        if args.resume_pending:
+            raise Refused("resume_requires_execute")
+        command(
+            [
+                sys.executable,
+                str(Path(__file__).with_name("capture-ledger.py")),
+                "--destination",
+                "/tmp/email-ledger-capture-review",
+            ]
+        )
+        print(
+            json.dumps({"status": "reviewed", "live_changes": False, "provider_contacted": False})
+        )
+    return 0
+
+
+if __name__ == "__main__":
+
+    def terminate(signum, frame):
+        raise Refused("backup_interrupted")
+
+    signal.signal(signal.SIGTERM, terminate)
+    try:
+        raise SystemExit(main())
+    except Exception as error:
+        print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
+        raise SystemExit(1)

scripts/capture-ledger.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""Node-only ledger capture, preview by default; execution briefly pauses email."""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import fcntl
+import json
+import os
+import secrets
+import signal
+import subprocess
+import time
+from pathlib import Path
+
+from ledger_archive import MAX_BYTES, Refused, digest, fingerprint, private_directory, write_private
+
+NS = "email-connect"
+CLUSTER_UID = "a553c742-0115-43d4-99a4-a5ca56fe0786"
+IMAGE = (
+    "forgejo.coulomb.social/coulomb/email-connect@sha256:"
+    "1e613d43734c44c5cd32c6083fda8b4da5d01ef43d10c6ca9c29a33c788cf75c"
+)
+LOCK_PATH = Path("/var/lib/email-connect-backup/capture.lock")
+SELECTOR = "app.kubernetes.io/name=email-connect"
+
+
+def k(*args, body=None, timeout=180):
+    result = subprocess.run(["kubectl", *args], input=body, capture_output=True, timeout=timeout)
+    if result.returncode:
+        raise Refused("cluster_command_failed")
+    return result.stdout
+
+
+def document(*args, body=None):
+    return json.loads(k(*args, "-o", "json", body=body, timeout=30))
+
+
+def helper(name):
+    return {
+        "apiVersion": "v1",
+        "kind": "Pod",
+        "metadata": {"name": name, "namespace": NS, "labels": {"email-ledger-capture": name}},
+        "spec": {
+            "restartPolicy": "Never",
+            "automountServiceAccountToken": False,
+            "securityContext": {
+                "runAsNonRoot": True,
+                "runAsUser": 10001,
+                "runAsGroup": 10001,
+                "fsGroup": 10001,
+                "fsGroupChangePolicy": "OnRootMismatch",
+                "seccompProfile": {"type": "RuntimeDefault"},
+            },
+            "containers": [
+                {
+                    "name": "capture",
+                    "image": IMAGE,
+                    "command": ["python", "-c", "import time;time.sleep(1800)"],
+                    "resources": {
+                        "requests": {"cpu": "10m", "memory": "32Mi"},
+                        "limits": {"cpu": "100m", "memory": "256Mi"},
+                    },
+                    "securityContext": {
+                        "allowPrivilegeEscalation": False,
+                        "readOnlyRootFilesystem": True,
+                        "capabilities": {"drop": ["ALL"]},
+                    },
+                    "volumeMounts": [
+                        {"name": "data", "mountPath": "/data", "readOnly": True},
+                        {"name": "tmp", "mountPath": "/tmp"},
+                    ],
+                }
+            ],
+            "volumes": [
+                {
+                    "name": "data",
+                    "persistentVolumeClaim": {"claimName": "email-connect-data", "readOnly": True},
+                },
+                {"name": "tmp", "emptyDir": {"sizeLimit": "160Mi"}},
+            ],
+        },
+    }
+
+
+def replica_patch(template, expected, desired):
+    return json.dumps(
+        [
+            {"op": "test", "path": "/spec/template", "value": template},
+            {"op": "test", "path": "/spec/replicas", "value": expected},
+            {"op": "replace", "path": "/spec/replicas", "value": desired},
+        ]
+    ).encode()
+
+
+def selected(selector, labels):
+    if any(labels.get(key) != value for key, value in selector.get("matchLabels", {}).items()):
+        return False
+    for expression in selector.get("matchExpressions", []):
+        key, op, values = expression["key"], expression["operator"], expression.get("values", [])
+        if op == "In" and labels.get(key) not in values:
+            return False
+        if op == "NotIn" and labels.get(key) in values:
+            return False
+        if op == "Exists" and key not in labels:
+            return False
+        if op == "DoesNotExist" and key in labels:
+            return False
+        if op not in {"In", "NotIn", "Exists", "DoesNotExist"}:
+            raise Refused("unknown_network_selector")
+    return True
+
+
+def require_no_helper_egress(policies, labels):
+    applicable = [
+        p["spec"]
+        for p in policies
+        if "Egress" in p["spec"].get("policyTypes", [])
+        and selected(p["spec"].get("podSelector", {}), labels)
+    ]
+    if not applicable or any(spec.get("egress") for spec in applicable):
+        raise Refused("helper_egress_not_denied")
+
+
+def preflight():
+    if document("get", "namespace", "kube-system")["metadata"]["uid"] != CLUSTER_UID:
+        raise Refused("wrong_cluster")
+    before = document("get", "deployment", "email-connect", "-n", NS)
+    spec = before["spec"]
+    containers = spec["template"]["spec"]["containers"]
+    if (
+        spec["replicas"] != 1
+        or before["status"].get("readyReplicas") != 1
+        or len(containers) != 1
+        or containers[0]["image"] != IMAGE
+    ):
+        raise Refused("unexpected_release")
+    if document("get", "pvc", "email-connect-data", "-n", NS)["status"]["phase"] != "Bound":
+        raise Refused("pvc_not_bound")
+    config = document("get", "configmap", "email-connect-config", "-n", NS)["data"]
+    if config.get("EMAIL_CONNECT_DATABASE_PATH") != "/data/email-connect.db":
+        raise Refused("unexpected_database_path")
+    name = "email-ledger-capture-" + secrets.token_hex(8)
+    pod = helper(name)
+    policies = document("get", "networkpolicies", "-n", NS)["items"]
+    require_no_helper_egress(policies, pod["metadata"]["labels"])
+    body = json.dumps(pod).encode()
+    k("create", "--dry-run=server", "-f", "-", body=body)
+    projected = document(
+        "patch",
+        "deployment",
+        "email-connect",
+        "-n",
+        NS,
+        "--type=json",
+        "--patch-file=/dev/stdin",
+        "--dry-run=server",
+        body=replica_patch(spec["template"], 1, 0),
+    )
+    expected = copy.deepcopy(spec)
+    expected["replicas"] = 0
+    if projected["spec"] != expected:
+        raise Refused("unexpected_pause_spec")
+    return before, body
+
+
+def capture_database(name, destination):
+    # No app startup, mounted secrets, SMTP import or provider request in helper.
+    code = """import os,sqlite3,sys
+source=sqlite3.connect('file:/data/email-connect.db?mode=ro',uri=True)
+pages=source.execute('PRAGMA page_count').fetchone()[0]
+size=source.execute('PRAGMA page_size').fetchone()[0]
+assert pages*size<=LIMIT
+os.umask(0o077)
+output=sqlite3.connect('/tmp/ledger.sqlite3')
+source.backup(output,pages=256)
+assert output.execute('PRAGMA integrity_check').fetchall()==[('ok',)]
+output.close();source.close()
+with open('/tmp/ledger.sqlite3','rb') as stream:sys.stdout.buffer.write(stream.read())
+""".replace("LIMIT", str(MAX_BYTES))
+    data = k("exec", "-n", NS, name, "--", "python", "-c", code)
+    if not data.startswith(b"SQLite format 3\x00") or len(data) > MAX_BYTES:
+        raise Refused("invalid_ledger_backup")
+    path = destination / "database.sqlite3"
+    write_private(path, data)
+    return {
+        "database_sha256": digest(path),
+        "database_bytes": len(data),
+        "ledger_fingerprint": fingerprint(path),
+        "sqlite_integrity": "ok",
+        "writers_quiesced": True,
+        "source_pvc_read_only": True,
+    }
+
+
+def capture(before, body, destination):
+    if not destination.is_absolute() or destination.exists() or destination.is_symlink():
+        raise Refused("fresh_absolute_destination_required")
+    destination.mkdir(mode=0o700)
+    private_directory(destination)
+    receipt = {
+        "status": "incomplete",
+        "application": "email-connect",
+        "image": IMAGE,
+        "captured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+    }
+    started, paused, created, captured = time.monotonic(), False, False, False
+    name = json.loads(body)["metadata"]["name"]
+    cleanup_error = None
+    try:
+        paused = True  # A failed response can still have changed replica count.
+        k(
+            "patch",
+            "deployment",
+            "email-connect",
+            "-n",
+            NS,
+            "--type=json",
+            "--patch-file=/dev/stdin",
+            body=replica_patch(before["spec"]["template"], 1, 0),
+        )
+        k("wait", "--for=delete", "pod", "-n", NS, "-l", SELECTOR, "--timeout=120s")
+        current = document("get", "deployment", "email-connect", "-n", NS)
+        expected = copy.deepcopy(before["spec"])
+        expected["replicas"] = 0
+        if current["spec"] != expected:
+            raise Refused("writers_not_quiesced")
+        # Require no other pod mounting the ledger PVC, including a prior helper.
+        pods = document("get", "pods", "-n", NS)["items"]
+        if any(
+            any(
+                v.get("persistentVolumeClaim", {}).get("claimName") == "email-connect-data"
+                for v in pod["spec"].get("volumes", [])
+            )
+            for pod in pods
+        ):
+            raise Refused("other_ledger_pvc_consumer")
+        created = True
+        k("create", "-f", "-", body=body)
+        k("wait", "--for=condition=Ready", "pod/" + name, "-n", NS, "--timeout=120s")
+        receipt.update(capture_database(name, destination))
+        captured = True
+    except BaseException as error:
+        receipt.update(failure_type=type(error).__name__)
+        raise
+    finally:
+        if created:
+            try:
+                k(
+                    "delete",
+                    "pod",
+                    name,
+                    "-n",
+                    NS,
+                    "--ignore-not-found=true",
+                    "--wait=true",
+                    "--timeout=120s",
+                )
+                receipt["helper_removed"] = True
+            except BaseException as error:
+                receipt["helper_cleanup_failed"] = True
+                cleanup_error = error
+        if paused:
+            try:
+                current = document("get", "deployment", "email-connect", "-n", NS)
+                expected = copy.deepcopy(before["spec"])
+                expected["replicas"] = current["spec"]["replicas"]
+                if current["spec"] != expected or current["spec"]["replicas"] not in {0, 1}:
+                    raise Refused("foreign_release_requires_receiver_check")
+                if current["spec"]["replicas"] == 0:
+                    k(
+                        "patch",
+                        "deployment",
+                        "email-connect",
+                        "-n",
+                        NS,
+                        "--type=json",
+                        "--patch-file=/dev/stdin",
+                        body=replica_patch(before["spec"]["template"], 0, 1),
+                    )
+                k(
+                    "rollout",
+                    "status",
+                    "deployment/email-connect",
+                    "-n",
+                    NS,
+                    "--timeout=180s",
+                    timeout=240,
+                )
+                current = document("get", "deployment", "email-connect", "-n", NS)
+                if current["spec"] != before["spec"] or current["status"].get("readyReplicas") != 1:
+                    raise Refused("receiver_resume_not_verified")
+                receipt["receiver_resumed"] = True
+                receipt["unchanged_receiver_spec"] = True
+            except BaseException as error:
+                receipt["receiver_state_requires_check"] = True
+                cleanup_error = cleanup_error or error
+        receipt["status"] = (
+            "captured"
+            if captured and receipt.get("receiver_resumed") and not cleanup_error
+            else "failed"
+        )
+        receipt["elapsed_seconds"] = round(time.monotonic() - started, 2)
+        write_private(destination / "receipt.json", json.dumps(receipt, indent=2).encode())
+        if cleanup_error:
+            raise Refused("capture_cleanup_requires_check") from None
+    return receipt
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--destination", type=Path, required=True)
+    args = parser.parse_args()
+    os.environ["KUBECONFIG"] = "/etc/rancher/k3s/k3s.yaml"
+    before, body = preflight()
+    if not args.execute:
+        print(
+            json.dumps(
+                {
+                    "status": "reviewed",
+                    "live_changes": False,
+                    "server_dry_run": True,
+                    "image_pinned": True,
+                    "pvc_read_only": True,
+                    "receiver_spec_changed": False,
+                }
+            )
+        )
+        return 0
+    LOCK_PATH.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    private_directory(LOCK_PATH.parent)
+    fd = os.open(LOCK_PATH, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(fd, "w") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            raise Refused("capture_already_running") from None
+        # Recheck after locking; do not use a stale preflight across another run.
+        before, body = preflight()
+        print(json.dumps(capture(before, body, args.destination)))
+    return 0
+
+
+if __name__ == "__main__":
+
+    def terminate(signum, frame):
+        raise Refused("capture_interrupted")
+
+    signal.signal(signal.SIGTERM, terminate)
+    try:
+        raise SystemExit(main())
+    except Exception as error:
+        print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
+        raise SystemExit(1)

scripts/ledger_archive.py

@@ -0,0 +1,284 @@
+#!/usr/bin/env python3
+"""Private transactional-ledger archive primitives; no mail/provider code."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import stat
+import subprocess
+import tarfile
+import tempfile
+from contextlib import closing
+from pathlib import Path
+
+MAX_BYTES = 128 * 1024**2
+TABLES = {"deliveries", "verifications", "suppressions", "send_attempts", "verification_attempts"}
+PREFIX = "s3://railiance-platform-pg-backup/platform-pg/application-archives/email-connect/"
+
+
+class Refused(RuntimeError):
+    pass
+
+
+def digest(path):
+    with path.open("rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def private_directory(path):
+    info = path.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+        raise Refused("private_owner_directory_required")
+
+
+def private_file(path):
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+        raise Refused("private_owner_file_required")
+
+
+def write_private(path, value):
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(fd, "wb") as stream:
+        stream.write(value)
+        stream.flush()
+        os.fsync(stream.fileno())
+
+
+def checkpoint(path, value):
+    fd, name = tempfile.mkstemp(prefix="receipt-", dir=path.parent)
+    with os.fdopen(fd, "w") as stream:
+        json.dump(value, stream, indent=2)
+        stream.flush()
+        os.fsync(stream.fileno())
+    Path(name).replace(path)
+
+
+def fingerprint(path):
+    private_file(path)
+    if not 0 < path.stat().st_size <= MAX_BYTES:
+        raise Refused("ledger_size_outside_contract")
+    with closing(sqlite3.connect(path.absolute().as_uri() + "?mode=ro", uri=True)) as database:
+        database.execute("PRAGMA trusted_schema=OFF")
+        if database.execute("PRAGMA integrity_check").fetchall() != [("ok",)]:
+            raise Refused("ledger_integrity_failed")
+        tables = {
+            row[0]
+            for row in database.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
+            )
+        }
+        if tables != TABLES:
+            raise Refused("unexpected_ledger_schema")
+        schema = list(
+            database.execute("SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name")
+        )
+        result = {
+            "schema_sha256": hashlib.sha256(json.dumps(schema).encode()).hexdigest(),
+            "user_version": database.execute("PRAGMA user_version").fetchone()[0],
+            "tables": {},
+        }
+        for name in sorted(TABLES):
+            rows = [
+                json.dumps(
+                    row, separators=(",", ":"), default=lambda value: {"bytes_hex": value.hex()}
+                ).encode()
+                for row in database.execute('SELECT * FROM "' + name + '"')
+            ]
+            result["tables"][name] = {
+                "rows": len(rows),
+                "sha256": hashlib.sha256(b"\n".join(sorted(rows))).hexdigest(),
+            }
+        return result
+
+
+def sqlite_backup(source, output):
+    """Use SQLite backup rather than copying a file that may still have WAL."""
+    private_directory(output.parent)
+    if source.is_symlink() or not source.is_file() or output.exists() or output.is_symlink():
+        raise Refused("invalid_sqlite_backup_paths")
+    # No source write or checkpoint; a read-only connection includes committed WAL.
+    with closing(sqlite3.connect(source.absolute().as_uri() + "?mode=ro", uri=True)) as src:
+        if (
+            src.execute("PRAGMA page_count").fetchone()[0]
+            * src.execute("PRAGMA page_size").fetchone()[0]
+            > MAX_BYTES
+        ):
+            raise Refused("ledger_size_outside_contract")
+        write_private(output, b"")
+        with closing(sqlite3.connect(output)) as dst:
+            src.backup(dst, pages=256)
+    return fingerprint(output)
+
+
+def validate_capture(directory):
+    private_directory(directory)
+    private_file(directory / "receipt.json")
+    receipt = json.loads((directory / "receipt.json").read_text())
+    if (
+        receipt.get("status") != "captured"
+        or receipt.get("application") != "email-connect"
+        or receipt.get("writers_quiesced") is not True
+        or receipt.get("receiver_resumed") is not True
+        or receipt.get("source_pvc_read_only") is not True
+    ):
+        raise Refused("capture_incomplete")
+    path = directory / "database.sqlite3"
+    actual = fingerprint(path)
+    if (
+        digest(path) != receipt.get("database_sha256")
+        or path.stat().st_size != receipt.get("database_bytes")
+        or actual != receipt.get("ledger_fingerprint")
+    ):
+        raise Refused("capture_binding_mismatch")
+    return receipt
+
+
+def command(argv, timeout=900):
+    # Keep private child diagnostics out of both memory pipes and the journal.
+    with tempfile.TemporaryFile() as output:
+        child = subprocess.Popen(argv, stdout=output, stderr=output)
+        try:
+            result = child.wait(timeout=timeout)
+        except BaseException:
+            child.terminate()
+            try:
+                child.wait(timeout=800)
+            except subprocess.TimeoutExpired:
+                child.kill()
+                child.wait()
+            raise
+        if result:
+            raise Refused("archive_command_failed")
+
+
+def seal(directory, stage, recipient):
+    captured = validate_capture(directory)
+    encrypted, sealed_path = stage / "packet.tar.age", stage / "sealed.json"
+    binding = {
+        key: captured[key]
+        for key in ["database_sha256", "captured_at", "image", "ledger_fingerprint"]
+    }
+    if sealed_path.exists():
+        private_file(sealed_path)
+        private_file(encrypted)
+        receipt = json.loads(sealed_path.read_text())
+        if receipt.get("binding") != binding or receipt.get("ciphertext_sha256") != digest(
+            encrypted
+        ):
+            raise Refused("sealed_packet_binding_mismatch")
+        return encrypted, captured
+    if encrypted.exists() or encrypted.is_symlink():
+        raise Refused("partial_encryption_requires_review")
+    archive = stage / "packet.tar"
+    with archive.open("xb") as stream:
+        os.chmod(archive, 0o600)
+        with tarfile.open(fileobj=stream, mode="w") as packet:
+            for name in ["database.sqlite3", "receipt.json"]:
+                packet.add(directory / name, arcname=name, recursive=False)
+    command(["age", "-r", recipient, "-o", str(encrypted), str(archive)])
+    encrypted.chmod(0o600)
+    private_file(encrypted)
+    if not 0 < encrypted.stat().st_size <= MAX_BYTES + 1024**2:
+        raise Refused("encrypted_packet_size_outside_contract")
+    checkpoint(sealed_path, {"binding": binding, "ciphertext_sha256": digest(encrypted)})
+    return encrypted, captured
+
+
+def unpack(packet, destination):
+    if packet.stat().st_size > MAX_BYTES + 1024**2:
+        raise Refused("packet_too_large")
+    with tarfile.open(packet, "r:") as archive:
+        entries = archive.getmembers()
+        if (
+            len(entries) != 2
+            or {entry.name for entry in entries} != {"database.sqlite3", "receipt.json"}
+            or any(
+                not entry.isfile() or entry.sparse is not None or entry.size <= 0
+                for entry in entries
+            )
+            or sum(entry.size for entry in entries) > MAX_BYTES + 1024**2
+            or any(entry.size > 1024**2 for entry in entries if entry.name == "receipt.json")
+        ):
+            raise Refused("invalid_packet_members")
+        for entry in entries:
+            with (
+                archive.extractfile(entry) as source,
+                (destination / entry.name).open("xb") as target,
+            ):
+                os.chmod(target.name, 0o600)
+                remaining = entry.size
+                while remaining:
+                    block = source.read(min(remaining, 8 * 1024**2))
+                    if not block:
+                        raise Refused("truncated_packet_member")
+                    target.write(block)
+                    remaining -= len(block)
+    return validate_capture(destination)
+
+
+def recover(encrypted, transfer_path, identity, destination):
+    for path in [encrypted, transfer_path, identity]:
+        private_file(path)
+    transfer = json.loads(transfer_path.read_text())
+    key = transfer.get("object_key", "")
+    suffix = key.removeprefix("platform-pg/application-archives/email-connect/")
+    version = transfer.get("version_id")
+    if (
+        not isinstance(version, str)
+        or version in {"", "null"}
+        or any(char in version for char in "\r\n")
+        or not key.startswith("platform-pg/application-archives/email-connect/")
+        or not re.fullmatch(r"\d{4}/\d{2}/\d{2}/\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age", suffix)
+        or transfer.get("destination") != "s3://railiance-platform-pg-backup/" + key
+    ):
+        raise Refused("verified_transfer_required")
+    if (
+        transfer.get("schema") != "platform.application-primary-archive.v1"
+        or transfer.get("application") != "email-connect"
+        or transfer.get("status") != "offsite_fetched_verified"
+        or transfer.get("remote_retrieval_verified") is not True
+        or not transfer.get("version_id")
+        or transfer["version_id"] == "null"
+        or not transfer.get("destination", "").startswith(PREFIX)
+        or transfer.get("ciphertext_sha256") != digest(encrypted)
+        or transfer.get("ciphertext_bytes") != encrypted.stat().st_size
+    ):
+        raise Refused("verified_transfer_required")
+    if not destination.is_absolute() or destination.exists() or destination.is_symlink():
+        raise Refused("fresh_absolute_destination_required")
+    destination.mkdir(mode=0o700)
+    packet = destination / "packet.tar"
+    # age must create plaintext with private modes even in an interactive shell.
+    result = subprocess.run(
+        ["age", "-d", "-i", str(identity), "-o", str(packet), str(encrypted)],
+        capture_output=True,
+        timeout=900,
+        umask=0o077,
+    )
+    if result.returncode:
+        raise Refused("decryption_failed")
+    private_file(packet)
+    captured = unpack(packet, destination)
+    for key in ["captured_at", "image", "database_sha256", "ledger_fingerprint"]:
+        if transfer.get(key) != captured.get(key):
+            raise Refused("recovery_capture_binding_mismatch")
+    packet.unlink()
+    result = {
+        "status": "decrypted_ledger_verified",
+        "application": "email-connect",
+        "object_key": transfer["object_key"],
+        "version_id": transfer["version_id"],
+        "ciphertext_sha256": transfer["ciphertext_sha256"],
+        "database_sha256": captured["database_sha256"],
+        "ledger_fingerprint": captured["ledger_fingerprint"],
+        "mail_sent": False,
+        "production_restored": False,
+        "application_restore_verified": False,
+    }
+    checkpoint(destination / "unpack-receipt.json", result)
+    return result

scripts/rehearse-ledger.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Attended recovered-ledger rehearsal in the cached receiver image, no network."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import secrets
+import subprocess
+import time
+from pathlib import Path
+
+from ledger_archive import (
+    Refused,
+    checkpoint,
+    private_directory,
+    private_file,
+    validate_capture,
+)
+
+IMAGE = (
+    "forgejo.coulomb.social/coulomb/email-connect@sha256:"
+    "1e613d43734c44c5cd32c6083fda8b4da5d01ef43d10c6ca9c29a33c788cf75c"
+)
+
+CODE = """import io,json,os,sys
+from pathlib import Path
+sys.path.insert(0,'/opt/ledger-rehearsal')
+from ledger_archive import fingerprint
+from email_connect.transactional import SQLiteDeliveryStore,TransactionalApplication,ProviderError
+assert os.getuid()==10001
+path=Path('/tmp/recovered.sqlite3')
+os.umask(0o077)
+path.write_bytes(sys.stdin.buffer.read())
+path.chmod(0o600)
+before=fingerprint(path)
+store=SQLiteDeliveryStore(str(path))
+assert fingerprint(path)==before
+class ForbiddenProvider:
+    def send(self,*args):raise AssertionError('recovery_provider_call_forbidden')
+app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token','https://issuer.example.test')
+checks={'accepted_verifications':0,'unknown_verifications':0,'invalidated_handles':0,
+        'suppressions':0,'accepted_deliveries':0,'unknown_sends':0}
+query=("SELECT v.registration_id,v.email,v.username,v.client_id,v.tenant,v.display_name,a.state "
+       "FROM verifications v JOIN verification_attempts a ON a.request_id=v.request_id "
+       "WHERE a.state IN ('accepted','in_flight')")
+for row in store.db.execute(query).fetchall():
+    payload=dict(zip(['registration_id','normalized_email','preferred_username','client_id','tenant','display_name'],row[:6]))
+    payload['correlation_id']='synthetic-recovery-probe'
+    raw=json.dumps(payload).encode();response={}
+    body=json.loads(b''.join(app({'PATH_INFO':'/v1/registration-verifications','REQUEST_METHOD':'POST',
+        'CONTENT_LENGTH':str(len(raw)),'wsgi.input':io.BytesIO(raw),
+        'HTTP_AUTHORIZATION':'Bearer synthetic-recovery-token'},
+        lambda status,headers:response.update(status=status))))
+    if row[6]=='accepted':
+        assert response['status'].startswith('202') and body['accepted']
+        checks['accepted_verifications']+=1
+    else:
+        assert response['status'].startswith('422') and body['error']=='delivery_outcome_unknown'
+        checks['unknown_verifications']+=1
+query=("SELECT handle_hash FROM verifications "
+       "WHERE consumed_at IS NOT NULL OR canceled_at IS NOT NULL")
+for row in store.db.execute(query).fetchall():
+    try:store.consume_verification(row[0]);raise AssertionError('invalid_handle_reactivated')
+    except ValueError as error:assert str(error)=='verification_invalid'
+    checks['invalidated_handles']+=1
+for row in store.db.execute('SELECT email FROM suppressions').fetchall():
+    assert store.is_suppressed(row[0]);checks['suppressions']+=1
+query=("SELECT d.event_id,a.fingerprint FROM deliveries d "
+       "LEFT JOIN send_attempts a ON a.event_id=d.event_id")
+for row in store.db.execute(query).fetchall():
+    assert store.delivery_status(row[0])['state']=='provider_accepted'
+    assert store.reserve_send(row[0],row[1] or 'synthetic-legacy-probe') is False
+    checks['accepted_deliveries']+=1
+query=("SELECT event_id,fingerprint FROM send_attempts WHERE state='in_flight' "
+       "AND event_id NOT IN (SELECT event_id FROM deliveries)")
+for row in store.db.execute(query).fetchall():
+    assert store.delivery_status(row[0])['state']=='unknown'
+    try:store.reserve_send(row[0],row[1]);raise AssertionError('unknown_send_retried')
+    except ProviderError as error:assert error.code=='delivery_outcome_unknown'
+    checks['unknown_sends']+=1
+store.db.close()
+assert fingerprint(path)==before
+print(json.dumps({'status':'isolated_ledger_verified','ledger_fingerprint':before,
+                  'checks':checks,'mail_sent':False,'uid':os.getuid()}))
+"""
+
+
+def rehearse(recovered, receipt_path):
+    private_directory(recovered)
+    private_directory(receipt_path.parent)
+    if receipt_path.exists() or receipt_path.is_symlink():
+        raise Refused("fresh_receipt_required")
+    capture = validate_capture(recovered)
+    private_file(recovered / "unpack-receipt.json")
+    unpacked = json.loads((recovered / "unpack-receipt.json").read_text())
+    if (
+        capture["image"] != IMAGE
+        or unpacked.get("status") != "decrypted_ledger_verified"
+        or unpacked.get("ledger_fingerprint") != capture["ledger_fingerprint"]
+    ):
+        raise Refused("pinned_recovered_ledger_required")
+    # Never pull a recovery image from the production forge during this rehearsal.
+    inspected = subprocess.run(
+        ["docker", "image", "inspect", IMAGE], capture_output=True, timeout=30
+    )
+    if inspected.returncode:
+        raise Refused("cached_pinned_recovery_image_required")
+    started = time.monotonic()
+    module = Path(__file__).with_name("ledger_archive.py").resolve()
+    name = "email-ledger-rehearsal-" + secrets.token_hex(8)
+    try:
+        with (recovered / "database.sqlite3").open("rb") as source:
+            result = subprocess.run(
+                [
+                    "docker",
+                    "run",
+                    "--rm",
+                    "--name",
+                    name,
+                    "-i",
+                    "--network",
+                    "none",
+                    "--read-only",
+                    "--user",
+                    "10001:10001",
+                    "--cap-drop",
+                    "ALL",
+                    "--security-opt",
+                    "no-new-privileges",
+                    "--memory",
+                    "512m",
+                    "--cpus",
+                    "0.5",
+                    "--tmpfs",
+                    "/tmp:rw,noexec,nosuid,size=160m,uid=10001,gid=10001,mode=0700",
+                    "--mount",
+                    "type=bind,source="
+                    + str(module)
+                    + ",target=/opt/ledger-rehearsal/ledger_archive.py,readonly",
+                    "-e",
+                    "PYTHONDONTWRITEBYTECODE=1",
+                    "--entrypoint",
+                    "python",
+                    IMAGE,
+                    "-c",
+                    CODE,
+                ],
+                stdin=source,
+                capture_output=True,
+                timeout=120,
+            )
+    finally:
+        subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=30)
+    if result.returncode:
+        raise Refused("isolated_rehearsal_failed")
+    verification = json.loads(result.stdout)
+    if verification.get("ledger_fingerprint") != capture["ledger_fingerprint"]:
+        raise Refused("isolated_fingerprint_mismatch")
+    receipt = dict(
+        unpacked,
+        status="application_restored_verified",
+        application_restore_verified=True,
+        recovery_image=IMAGE,
+        network="disabled",
+        production_restored=False,
+        checks=verification["checks"],
+        uid=verification["uid"],
+        mail_sent=False,
+        evidence_ceiling="isolated_ledger_rehearsal",
+        elapsed_seconds=round(time.monotonic() - started, 2),
+    )
+    checkpoint(receipt_path, receipt)
+    return receipt
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--recovered", required=True, type=Path)
+    parser.add_argument("--receipt", required=True, type=Path)
+    parser.add_argument("--execute", action="store_true")
+    args = parser.parse_args()
+    if not args.execute:
+        print(
+            json.dumps(
+                {"status": "reviewed", "container_started": False, "production_restored": False}
+            )
+        )
+        return 0
+    print(json.dumps(rehearse(args.recovered, args.receipt)))
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except Exception as error:
+        print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
+        raise SystemExit(1)