tegwick
f77e2333af
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
88 lines
4.3 KiB
Python
88 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Build an explicit offline recovery kit; never include an age private identity."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
from ledger_archive import Refused, checkpoint, digest, private_directory, private_file
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def prepare(evidence, ciphertext, destination):
|
|
proof = json.loads(evidence.read_text())
|
|
transfer, recovery = proof["publication"], proof["application_recovery"]
|
|
if (recovery.get("application_restore_verified") is not True
|
|
or recovery.get("application") != "email-connect"
|
|
or recovery.get("ciphertext_sha256") != transfer.get("ciphertext_sha256")
|
|
or recovery.get("version_id") != transfer.get("version_id")):
|
|
raise Refused("verified_recovery_point_required")
|
|
private_file(ciphertext)
|
|
if (digest(ciphertext) != transfer["ciphertext_sha256"]
|
|
or ciphertext.stat().st_size != transfer["ciphertext_bytes"]):
|
|
raise Refused("protected_ciphertext_binding_changed")
|
|
destination.mkdir(mode=0o700)
|
|
private_directory(destination)
|
|
image = recovery["recovery_image"]
|
|
inspected = subprocess.run(["docker", "image", "inspect", image], capture_output=True, timeout=30)
|
|
if inspected.returncode:
|
|
raise Refused("cached_recovery_image_required")
|
|
image_id = json.loads(inspected.stdout)[0]["Id"]
|
|
saved = subprocess.run(["docker", "image", "save", "--output", str(destination / "receiver-image.tar"), image],
|
|
capture_output=True, timeout=180)
|
|
if saved.returncode:
|
|
raise Refused("recovery_image_export_failed")
|
|
(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)
|
|
# 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)
|
|
checkpoint(destination / "application-recovery.json", recovery)
|
|
files = {p.name: {"sha256": digest(p), "bytes": p.stat().st_size} for p in destination.iterdir()}
|
|
manifest = {"schema": "email-connect.ledger-recovery-kit.v1", "image": image, "image_id": image_id,
|
|
"files": files, "private_identity_included": False,
|
|
"protected_version": transfer["version_id"], "protected_ciphertext_sha256": transfer["ciphertext_sha256"]}
|
|
checkpoint(destination / "manifest.json", manifest)
|
|
with tarfile.open(destination / "kit.tar.gz", "w:gz") as archive:
|
|
for name in sorted([*files, "manifest.json"]):
|
|
archive.add(destination / name, arcname=name, recursive=False)
|
|
(destination / "kit.tar.gz").chmod(0o600)
|
|
receipt = {"status": "recovery_kit_prepared", "kit_sha256": digest(destination / "kit.tar.gz"),
|
|
"kit_bytes": (destination / "kit.tar.gz").stat().st_size,
|
|
"image": image, "image_id": image_id, "private_identity_included": False,
|
|
"provider_upload_verified": False, "protected_version": transfer["version_id"]}
|
|
checkpoint(destination / "kit-receipt.json", receipt)
|
|
return receipt
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--execute", action="store_true")
|
|
parser.add_argument("--evidence", type=Path, required=True)
|
|
parser.add_argument("--ciphertext", type=Path, required=True)
|
|
parser.add_argument("--destination", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
if not args.execute:
|
|
print(json.dumps({"status": "prepared", "image_exported": False, "provider_contacted": False}))
|
|
return
|
|
os.umask(0o077)
|
|
print(json.dumps(prepare(args.evidence, args.ciphertext, args.destination)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as error:
|
|
print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
|
|
raise SystemExit(1)
|