Add the MAS offsite restore rehearsal for RPF-WP-0070-T04.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 4541@bnt-lap001
Assistant-Session: 2099b3aa-d7d9-49a9-ab4e-fa8b7d0e39e8

scripts/mas_evidence_restore_rehearsal.py

@@ -0,0 +1,208 @@
+#!/usr/bin/env python3
+"""Rehearse MAS evidence recovery from the offsite copy on the workstation.
+
+Run only inside the attended OpenBao platform-admin login:
+
+    warden access openbao-platform-admin-login --exec -- \
+        python3 scripts/mas_evidence_restore_rehearsal.py --identity /abs/path/to/identity
+
+Independent credentials: the founder-held supplier key on the MAS IAM
+application, not the node's workload pair. The object is fetched at the exact
+version named in the T03 evidence receipt and checked against its hash.
+The identity must derive to the recorded recipient before it is used. The
+packet is decrypted into a private scratch directory, every file is checked
+against the packet's own hashes, SQLite integrity is checked, and every event
+line must parse. Nothing is restored onto the facility, no receiver is
+enabled, and no mail is sent. Plaintext scratch files are removed after the
+checks. Stdout and stderr stay silent; the receipt records names, counts,
+hashes and timings only.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tarfile
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+
+ROOT = Path("/home/worsch/railiance-platform")
+EVIDENCE = ROOT / "docs/evidence/mas-evidence-first-offsite-2026-11-03.json"
+RECIPIENT = (ROOT / "deploy/mas-backup/age-recipient").read_text().strip() if (
+    ROOT / "deploy/mas-backup/age-recipient").exists() else ""
+RECEIPT_DIR = Path("/home/worsch/.local/state/railiance/receipts")
+SCRATCH_ROOT = Path("/home/worsch/.local/state/railiance/mas-restore-rehearsal")
+RECEIPT_NAME = "mas-evidence-offsite-rehearsal.json"
+
+
+def _load(name, path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+sys.path.insert(0, str(ROOT / "scripts"))
+gate = _load("mas_evidence_publisher_gate", ROOT / "scripts/mas_evidence_publisher_gate.py")
+archive = _load("mas_evidence_archive", ROOT / "scripts/mas_evidence_archive.py")
+primary = _load("primary_application_archive", ROOT / "scripts/primary_application_archive.py")
+
+
+class RehearsalError(RuntimeError):
+    def __init__(self, code, detail=None):
+        super().__init__(code)
+        self.code, self.detail = code, detail
+
+
+def identity_matches(identity, recipient=None):
+    recipient = recipient or RECIPIENT
+    path = Path(identity)
+    if not path.is_absolute() or not path.is_file() or path.stat().st_mode & 0o077:
+        raise RehearsalError("private_absolute_identity_required")
+    result = subprocess.run(["age-keygen", "-y", str(path)], capture_output=True, timeout=30)
+    public = result.stdout.decode().strip().splitlines()
+    if result.returncode or not recipient or public[:1] != [recipient]:
+        raise RehearsalError("identity_recipient_mismatch")
+    return True
+
+
+def offsite_receipt(path=EVIDENCE):
+    body = json.loads(path.read_text(encoding="utf-8"))
+    receipt = dict(body["receipt"])
+    if receipt.get("schema") != "platform.application-primary-archive.v1":
+        raise RehearsalError("offsite_receipt_rejected")
+    return receipt
+
+
+def verify_restore(encrypted, identity, scratch):
+    """Decrypt, extract, and check. Returns counts and hashes; never values."""
+    scratch.mkdir(mode=0o700, parents=False)
+    packet = scratch / "packet.tar"
+    result = subprocess.run(
+        ["age", "-d", "-i", str(identity), "-o", str(packet), str(encrypted)],
+        capture_output=True, timeout=120,
+    )
+    if result.returncode:
+        raise RehearsalError("decryption_failed")
+    os.chmod(packet, 0o600)
+    with tarfile.open(packet, "r:") as tar:
+        entries = tar.getmembers()
+        if "receipt.json" not in {e.name for e in entries} or any(
+            not e.isfile() or "/" in e.name or e.name.startswith(".") for e in entries
+        ):
+            raise RehearsalError("invalid_packet_members")
+        for entry in entries:
+            target = scratch / entry.name
+            with tar.extractfile(entry) as source, target.open("xb") as out:
+                os.chmod(target, 0o600)
+                out.write(source.read())
+    inner = json.loads((scratch / "receipt.json").read_text())
+    if inner.get("application") != "mas" or inner.get("mail_sent") is not False:
+        raise RehearsalError("capture_receipt_rejected")
+    files = {}
+    for name, meta in inner["files"].items():
+        path = scratch / name
+        if not meta.get("present"):
+            files[name] = {"present": False, "restored": path.exists()}
+            continue
+        if archive.digest(path) != meta["sha256"] or path.stat().st_size != meta["bytes"]:
+            raise RehearsalError("restored_hash_mismatch", {"file": name})
+        entry = {"present": True, "bytes": meta["bytes"], "sha256_matches_capture": True}
+        if name.endswith(".sqlite3"):
+            fingerprint = archive._sqlite_fingerprint(path)  # integrity_check must be ok
+            entry["integrity_check"] = "ok"
+            entry["tables"] = {t: v["rows"] for t, v in fingerprint["tables"].items()}
+            if "tables" in meta:
+                entry["fingerprint_matches_capture"] = fingerprint["tables"] == meta["tables"]
+        else:
+            count = 0
+            with path.open("rb") as stream:
+                for line in stream:
+                    if line.strip():
+                        json.loads(line)
+                        count += 1
+            entry["events"] = count
+        files[name] = entry
+    return {"captured_at": inner.get("captured_at"), "image": inner.get("image"), "files": files}
+
+
+def rehearse(identity, client, receipt, scratch_root=SCRATCH_ROOT, clock=time.monotonic):
+    started = clock()
+    started_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+    identity_matches(identity)
+    scratch_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    run = scratch_root / started_at.replace(":", "")
+    run.mkdir(mode=0o700)
+    try:
+        encrypted = run / "fetched.tar.age"
+        fetched = primary.retrieve(client, "mas", encrypted, receipt)
+        fetched_seconds = round(clock() - started, 2)
+        restored = verify_restore(encrypted, identity, run / "restored")
+    finally:
+        shutil.rmtree(run, ignore_errors=True)
+    elapsed = round(clock() - started, 2)
+    sqlite_ok = all(
+        meta.get("integrity_check") == "ok" and meta.get("fingerprint_matches_capture", True)
+        for name, meta in restored["files"].items() if name.endswith(".sqlite3") and meta.get("present")
+    )
+    return {
+        "schema": "platform.mas-evidence-offsite-rehearsal.v1",
+        "status": "passed" if sqlite_ok else "failed",
+        "started_at": started_at,
+        "object_key": fetched["object_key"],
+        "version_id": fetched["version_id"],
+        "ciphertext_sha256": fetched["ciphertext_sha256"],
+        "ciphertext_bytes": fetched["ciphertext_bytes"],
+        "fetch_verified": fetched["status"] == "retrieved_verified",
+        "credential": "supplier key on IAM application 25a4cbca (independent of the node workload pair)",
+        "identity_matches_recipient": True,
+        "restored": restored,
+        "fetch_seconds": fetched_seconds,
+        "elapsed_seconds": elapsed,
+        "rto_minutes_provisional": 60,
+        "within_rto": elapsed <= 3600,
+        "scratch_removed": not run.exists(),
+        "production_restored": False,
+        "receiver_enabled": False,
+        "mail_sent": False,
+        "provider_mutations": False,
+        "values_recorded": False,
+    }
+
+
+def _supplier_client():
+    pair = gate.supplier_pair(gate._kv_data(gate.SUPPLIER), gate.SUPPLIER)
+    scoped, _ = gate.scoped_supplier_pair(pair, gate.PROJECT_ID)
+    return gate._client(scoped)
+
+
+def main(argv=None):
+    args = list(sys.argv[1:] if argv is None else argv)
+    sys.stdout = open(os.devnull, "w")
+    sys.stderr = open(os.devnull, "w")
+    path = RECEIPT_DIR / RECEIPT_NAME
+    try:
+        if len(args) != 2 or args[0] != "--identity":
+            raise RehearsalError("usage_refused")
+        gate._ensure_boto()
+        RECEIPT_DIR.mkdir(mode=0o700, exist_ok=True)
+        result = rehearse(args[1], _supplier_client(), offsite_receipt())
+    except (RehearsalError, gate.GateError) as exc:
+        result = {"status": "failed", "code": exc.code, "values_recorded": False}
+        if exc.detail:
+            result["detail"] = exc.detail
+    except Exception as exc:
+        result = {"status": "failed", "code": "contained_rehearsal_failure",
+                  "error_type": type(exc).__name__, "values_recorded": False}
+    gate._write_json(path, result)
+    return 0 if result.get("status") == "passed" else 71
+
+
+if __name__ == "__main__":
+    sys.exit(main())