Retain independent ledger recovery custody and activate verified backups

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2

scripts/backup-ledger.py

@@ -95,7 +95,7 @@ def publish(encrypted, fetched, receipt_path, captured, tool, python):
     return receipt
 
 
-def backup(state, primary_tool, primary_python, recipient, resume=False):
+def backup(state, primary_tool, primary_python, recipient, resume=False, mirror_required=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)
@@ -137,6 +137,17 @@ def backup(state, primary_tool, primary_python, recipient, resume=False):
             primary_tool,
             primary_python,
         )
+        if mirror_required:
+            transfer.update(image=captured["image"], database_sha256=captured["database_sha256"],
+                            ledger_fingerprint=captured["ledger_fingerprint"])
+            checkpoint(stage / "primary-transfer.json", transfer)
+            command([str(primary_python), str(Path(__file__).with_name("mirror-ledger.py")),
+                     "--archive", str(encrypted), "--primary-receipt", str(stage / "primary-transfer.json"),
+                     "--output", str(stage / "mirror.json")])
+            mirrored = json.loads((stage / "mirror.json").read_text())
+            if mirrored.get("hash_readback_verified") is not True:
+                raise Refused("recovery_mirror_required")
+            transfer["recovery_mirror"] = mirrored
         filename = transfer["object_key"].rsplit("/", 1)[-1]
         if not NAME.fullmatch(filename):
             raise Refused("invalid_archive_name")
@@ -173,7 +184,7 @@ def backup(state, primary_tool, primary_python, recipient, resume=False):
         return receipt
 
 
-def freshness(state, now=None, hours=26):
+def freshness(state, now=None, hours=26, mirror_required=False):
     now = now or datetime.now(UTC)
     private_directory(state)
     private_file(state / "last-success.json")
@@ -195,6 +206,14 @@ def freshness(state, now=None, hours=26):
     private_file(state / name)
     if digest(state / name) != receipt["ciphertext_sha256"]:
         raise Refused("local_ciphertext_mismatch")
+    if mirror_required:
+        mirror = receipt.get("recovery_mirror", {})
+        if (mirror.get("hash_readback_verified") is not True
+                or mirror.get("bucket") != "railiance-email-connect-archives"
+                or mirror.get("project_id") != "1d37ea63-8c8f-44e1-80e1-2de862579ba7"
+                or not mirror.get("version_id") or mirror["version_id"] == "null"
+                or not mirror.get("receipt_version_id") or mirror["receipt_version_id"] == "null"):
+            raise Refused("verified_recovery_mirror_required")
     return {"status": "fresh", "age_hours": round(age / 3600, 2), "remote_retrieval_verified": True}
 
 
@@ -203,6 +222,7 @@ def main():
     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("--mirror-required", 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)
@@ -213,7 +233,7 @@ def main():
     parser.add_argument("--destination", type=Path)
     args = parser.parse_args()
     if args.action == "freshness":
-        print(json.dumps(freshness(args.state_dir)))
+        print(json.dumps(freshness(args.state_dir, mirror_required=args.mirror_required)))
     elif args.action == "recover":
         if not args.execute:
             print(json.dumps({"status": "reviewed", "decryption_performed": False}))
@@ -251,6 +271,7 @@ def main():
                     args.primary_python,
                     args.age_recipient,
                     args.resume_pending,
+                    args.mirror_required,
                 )
             )
         )

scripts/mirror-ledger.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Copy ciphertext into the isolated recovery project; never read an age key."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+
+from ledger_archive import Refused, checkpoint, digest, private_file
+
+PROJECT = "1d37ea63-8c8f-44e1-80e1-2de862579ba7"
+BUCKET = "railiance-email-connect-archives"
+
+
+def mirror(client, archive, receipt):
+    private_file(archive)
+    expected = digest(archive)
+    size = archive.stat().st_size
+    if expected != receipt["ciphertext_sha256"] or size != receipt["ciphertext_bytes"]:
+        raise Refused("mirror_source_differs")
+    key = "archives/" + receipt["object_key"]
+    try:
+        with archive.open("rb") as source:
+            created = client.put_object(Bucket=BUCKET, Key=key, Body=source,
+                                        IfNoneMatch="*", Metadata={"sha256": expected})
+        version = created.get("VersionId")
+    except Exception as error:
+        if getattr(error, "response", {}).get("ResponseMetadata", {}).get("HTTPStatusCode") != 412:
+            raise
+        current = client.head_object(Bucket=BUCKET, Key=key)
+        if current.get("Metadata", {}).get("sha256") != expected or current.get("ContentLength") != size:
+            raise Refused("mirror_existing_object_differs")
+        version = current.get("VersionId")
+    if not version or version == "null":
+        raise Refused("versioned_mirror_required")
+    result = client.get_object(Bucket=BUCKET, Key=key, VersionId=version)
+    hashed, count = hashlib.sha256(), 0
+    with result["Body"] as stream:
+        while block := stream.read(1024**2):
+            count += len(block)
+            if count > size:
+                raise Refused("mirror_readback_exceeds_size")
+            hashed.update(block)
+    if (result.get("VersionId") != version or count != size
+            or hashed.hexdigest() != expected):
+        raise Refused("mirror_readback_differs")
+    # Recovery needs a receipt even if both node and workstation are lost.
+    record = dict(receipt, recovery_mirror={"bucket": BUCKET, "object_key": key,
+                                          "version_id": version, "project_id": PROJECT,
+                                          "hash_readback_verified": True})
+    payload = json.dumps(record, sort_keys=True).encode()
+    receipt_key = key + "." + version + "." + hashlib.sha256(payload).hexdigest() + ".receipt.json"
+    try:
+        saved = client.put_object(Bucket=BUCKET, Key=receipt_key, Body=payload, IfNoneMatch="*")
+        receipt_version = saved.get("VersionId")
+    except Exception as error:
+        if getattr(error, "response", {}).get("ResponseMetadata", {}).get("HTTPStatusCode") != 412:
+            raise
+        receipt_version = client.head_object(Bucket=BUCKET, Key=receipt_key).get("VersionId")
+    if not receipt_version or receipt_version == "null":
+        raise Refused("versioned_mirror_receipt_required")
+    saved = client.get_object(Bucket=BUCKET, Key=receipt_key, VersionId=receipt_version)
+    with saved["Body"] as stream:
+        if stream.read(len(payload) + 1) != payload:
+            raise Refused("mirror_receipt_readback_differs")
+    return dict(record["recovery_mirror"], receipt_key=receipt_key,
+                receipt_version_id=receipt_version)
+
+
+def main():
+    import boto3
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--archive", type=Path, required=True)
+    parser.add_argument("--primary-receipt", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    private_file(args.primary_receipt)
+    key = os.environ["AWS_ACCESS_KEY_ID"].split("@", 1)[0] + "@" + PROJECT
+    client = boto3.client("s3", endpoint_url="https://s3.nl-ams.scw.cloud", region_name="nl-ams",
+                          aws_access_key_id=key, aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"])
+    checkpoint(args.output, mirror(client, args.archive, json.loads(args.primary_receipt.read_text())))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
+        raise SystemExit(1) from None

scripts/prepare-ledger-recovery-kit.py

@@ -42,7 +42,9 @@ def prepare(evidence, ciphertext, destination):
     (destination / "receiver-image.tar").chmod(0o600)
     for name in ("ledger_archive.py", "backup-ledger.py", "rehearse-ledger.py"):
         shutil.copyfile(ROOT / "scripts" / name, destination / name)
-        (destination / name).chmod(0o600)
+        # This public module is bind-mounted into the UID10001 rehearsal.
+        # The enclosing kit directory remains private; data stays mode 0600.
+        (destination / name).chmod(0o644 if name == "ledger_archive.py" else 0o600)
     shutil.copyfile(ciphertext, destination / "ledger.tar.age")
     (destination / "ledger.tar.age").chmod(0o600)
     checkpoint(destination / "transfer-receipt.json", transfer)