Prepare immutable caller rollout and independent Scaleway recovery custody

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

scripts/admit-scoped-caller.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Silent, attended exact-plan caller custody. Preview is the default."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+import re
+import secrets
+import subprocess
+import uuid
+from pathlib import Path
+from urllib.request import urlopen
+
+from ledger_archive import Refused, checkpoint, private_directory
+
+ROOT = Path(__file__).resolve().parents[1]
+PLAN = ROOT / "deploy/closure/plan.json"
+KV = "platform/workloads/email-connect/callers/user-engine"
+WRITER = "email-connect-scoped-caller-writer"
+READER = "email-connect-scoped-caller-reader"
+READERS = [
+    ("email-connect", "email-connect-scoped-caller-eso"),
+    ("user-engine", "user-engine-scoped-mail-eso"),
+]
+
+
+def role(namespace, account):
+    return {
+        "bound_service_account_names": [account],
+        "bound_service_account_namespaces": [namespace], "audience": "openbao",
+        "policies": [READER], "ttl": "15m", "token_max_ttl": "15m",
+        "token_explicit_max_ttl": "15m", "token_no_default_policy": True,
+        "token_type": "service",
+    }
+
+
+def bao(args, payload=None, token=None, missing=False):
+    env = os.environ.copy()
+    if token:
+        env.update(BAO_TOKEN=token, VAULT_TOKEN=token)
+    raw = payload if isinstance(payload, bytes) else None if payload is None else json.dumps(payload).encode()
+    result = subprocess.run(["bao", *args], input=raw, env=env, capture_output=True, timeout=60)
+    if result.returncode:
+        if missing and (b"Code: 404" in result.stderr or b"No value found" in result.stderr):
+            return None
+        raise Refused("attended_bao_operation_failed")
+    return result.stdout
+
+
+def approved_plan(plan, reviewed, decision, base):
+    expected = hashlib.sha256(PLAN.read_bytes()).hexdigest()
+    if reviewed != expected:
+        raise Refused("reviewed_plan_hash_required")
+    for name, expected_hash in plan["artifacts"].items():
+        path = ROOT / name
+        if not path.resolve().is_relative_to(ROOT) or hashlib.sha256(path.read_bytes()).hexdigest() != expected_hash:
+            raise Refused("reviewed_artifact_changed")
+    uuid.UUID(decision)
+    with urlopen(base.rstrip("/") + "/decisions/" + decision, timeout=15) as response:
+        record = json.loads(response.read(65536))
+    if (record.get("status") != "resolved" or record.get("decision_type") != "made"
+            or record.get("decided_by") != "User"
+            or expected not in (record.get("description", "") or "")):
+        raise Refused("founder_exact_plan_decision_required")
+    custody = plan["caller_custody"]
+    if (custody["kv_path"] != KV or custody["field"] != "TOKEN"
+            or custody["writer_policy"] != WRITER or custody["reader_policy"] != READER
+            or [(r["namespace"], r["service_account"]) for r in custody["readers"]] != READERS):
+        raise Refused("exact_caller_contract_required")
+    return expected
+
+
+def install_policy(name, path):
+    intended = path.read_bytes()
+    raw = bao(["read", "-format=json", "sys/policies/acl/" + name], missing=True)
+    observed = None if raw is None else json.loads(raw)["data"]["policy"].encode()
+    if observed is not None and observed.strip() != intended.strip():
+        raise Refused("existing_policy_differs")
+    if observed is None:
+        bao(["write", "sys/policies/acl/" + name, "-"], {"policy": intended.decode()})
+    observed = json.loads(bao(["read", "-format=json", "sys/policies/acl/" + name]))["data"]["policy"]
+    if observed.strip() != intended.decode().strip():
+        raise Refused("policy_readback_differs")
+
+
+def provision(transport=bao):
+    auth = json.loads(transport([
+        "token", "create", "-format=json", "-policy=" + WRITER,
+        "-no-default-policy", "-ttl=5m", "-explicit-max-ttl=5m", "-use-limit=8",
+    ]))["auth"]
+    token = auth["client_token"]
+    try:
+        if (auth.get("policies") != [WRITER] or auth.get("identity_policies", [])
+                or not 0 < auth.get("lease_duration", 0) <= 300):
+            raise Refused("bounded_writer_required")
+        metadata = transport(["read", "-format=json", "platform/metadata/workloads/email-connect/callers/user-engine"],
+                             token=token, missing=True)
+        if metadata is None:
+            value = secrets.token_urlsafe(48)
+            written = json.loads(transport(["write", "-format=json", "platform/data/workloads/email-connect/callers/user-engine", "-"],
+                {"options": {"cas": 0}, "data": {"TOKEN": value}}, token=token))
+            version = written["data"]["version"]
+            if version != 1:
+                raise Refused("initial_cas_write_required")
+            created = True
+        else:
+            version = json.loads(metadata)["data"]["current_version"]
+            value = None
+            created = False
+        readback = json.loads(transport(["read", "-format=json", "platform/data/workloads/email-connect/callers/user-engine"],
+                                       token=token))["data"]
+        if set(readback["data"]) != {"TOKEN"} or readback["metadata"]["version"] != version:
+            raise Refused("caller_record_changed")
+        stored = readback["data"]["TOKEN"]
+        if not isinstance(stored, str) or not re.fullmatch(r"[!-~]{16,512}", stored):
+            raise Refused("caller_credential_invalid")
+        if value is not None and not hmac.compare_digest(value, stored):
+            raise Refused("caller_readback_differs")
+        return {"kv_version": version, "created": created, "readback_equal": True,
+                "credential_values_printed": False}
+    finally:
+        transport(["write", "auth/token/revoke-self"], token=token)
+
+
+def execute(plan, reviewed, decision, base):
+    binding = approved_plan(plan, reviewed, decision, base)
+    for name, filename in ((WRITER, "caller-writer.hcl"), (READER, "caller-reader.hcl")):
+        install_policy(name, ROOT / "deploy/closure" / filename)
+    for namespace, account in READERS:
+        wanted = role(namespace, account)
+        observed = bao(["read", "-format=json", "auth/kubernetes/role/" + account], missing=True)
+        if observed is None:
+            bao(["write", "auth/kubernetes/role/" + account, "-"], wanted)
+            observed = bao(["read", "-format=json", "auth/kubernetes/role/" + account])
+        data = json.loads(observed)["data"]
+        if (data.get("bound_service_account_names") != [account]
+                    or data.get("bound_service_account_namespaces") != [namespace]
+                    or data.get("audience") != "openbao"
+                    or (data.get("token_policies") or data.get("policies")) != [READER]
+                    or data.get("token_no_default_policy") is not True
+                    or data.get("token_max_ttl") not in (900, "15m")
+                    or data.get("token_explicit_max_ttl") not in (900, "15m")):
+            raise Refused("existing_reader_role_differs")
+    return dict(provision(), status="caller_custody_prepared", plan_sha256=binding,
+                decision_id=decision, workloads_changed=False)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--reviewed-sha256")
+    parser.add_argument("--decision-id")
+    parser.add_argument("--api-base", default="http://127.0.0.1:8000")
+    parser.add_argument("--receipt", type=Path)
+    args = parser.parse_args()
+    plan = json.loads(PLAN.read_text())
+    if not args.execute:
+        print(json.dumps({"status": "prepared", "plan_sha256": hashlib.sha256(PLAN.read_bytes()).hexdigest(),
+                          "credential_generated": False, "workloads_changed": False}))
+        return
+    if not args.receipt or not args.receipt.is_absolute():
+        raise Refused("private_receipt_path_required")
+    private_directory(args.receipt.parent)
+    if args.receipt.exists() or args.receipt.is_symlink():
+        raise Refused("fresh_receipt_required")
+    try:
+        receipt = execute(plan, args.reviewed_sha256, args.decision_id, args.api_base)
+    except Exception as error:
+        checkpoint(args.receipt, {"status": "failed", "failure_type": type(error).__name__})
+        raise
+    checkpoint(args.receipt, receipt)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception:
+        raise SystemExit(1) from None

scripts/backup-ledger.py

@@ -30,9 +30,32 @@ from ledger_archive import (
 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")
+MINIMUM_FREE_BYTES = 8 * 1024**3
+LOCAL_BUDGET_BYTES = 8 * 1024**3
 RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"
 
 
+def capacity(state, minimum_free=MINIMUM_FREE_BYTES, budget=LOCAL_BUDGET_BYTES):
+    ancestor = state
+    while not ancestor.exists():
+        ancestor = ancestor.parent
+    if shutil.disk_usage(ancestor).free < minimum_free:
+        raise Refused("backup_disk_headroom_required")
+    used = 0
+    if state.exists():
+        private_directory(state)
+        for path in state.rglob("*"):
+            if path.is_symlink():
+                raise Refused("backup_state_symlink_requires_review")
+            if path.is_file():
+                used += path.stat().st_size
+        # Reserve room for capture, ciphertext, readback and a failure packet.
+        if used + 512 * 1024**2 > budget:
+            raise Refused("backup_local_retention_review_required")
+    return {"status": "headroom_verified", "local_bytes": used,
+            "minimum_free_bytes": minimum_free, "local_budget_bytes": budget}
+
+
 def publish(encrypted, fetched, receipt_path, captured, tool, python):
     command(
         [
@@ -209,6 +232,7 @@ def main():
             or not args.primary_python.is_file()
         ):
             raise Refused("invalid_backup_configuration")
+        capacity(args.state_dir)
         command(["age", "--version"])
         command(
             [

scripts/prepare-ledger-recovery-kit.py

@@ -0,0 +1,86 @@
+#!/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)
+        (destination / name).chmod(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)

scripts/publish-ledger-recovery-kit.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""Attended immutable kit upload and read-only recovery-caller readback."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+
+from ledger_archive import Refused, checkpoint, digest, private_directory, private_file
+
+BUCKET = "railiance-email-connect-recovery"
+ENDPOINT = "https://s3.nl-ams.scw.cloud"
+
+
+def client(credentials):
+    import boto3
+    from botocore.config import Config
+    return boto3.client("s3", endpoint_url=ENDPOINT, region_name="nl-ams",
+                        aws_access_key_id=credentials["AWS_ACCESS_KEY_ID"],
+                        aws_secret_access_key=credentials["AWS_SECRET_ACCESS_KEY"],
+                        config=Config(signature_version="s3v4", connect_timeout=15, read_timeout=180,
+                                      s3={"addressing_style": "path"}))
+
+
+def credentials(path):
+    private_file(path)
+    value = json.loads(path.read_text())
+    if (set(value) != {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"}
+            or any(not isinstance(v, str) or not v.strip() for v in value.values())):
+        raise Refused("exact_private_s3_pair_required")
+    return value
+
+
+def version(value):
+    if not isinstance(value, str) or not value or value == "null" or any(c in value for c in "\r\n"):
+        raise Refused("versioned_kit_required")
+    return value
+
+
+def publish(owner, reader, kit, expected, output):
+    if (expected.get("status") != "recovery_kit_prepared"
+            or expected.get("private_identity_included") is not False
+            or digest(kit) != expected.get("kit_sha256")
+            or kit.stat().st_size != expected.get("kit_bytes")
+            or not 0 < kit.stat().st_size <= 1024**3):
+        raise Refused("verified_bounded_kit_required")
+    if owner.get_bucket_versioning(Bucket=BUCKET).get("Status") != "Enabled":
+        raise Refused("versioned_recovery_bucket_required")
+    try:
+        lifecycle = owner.get_bucket_lifecycle_configuration(Bucket=BUCKET)
+    except Exception as error:
+        if getattr(error, "response", {}).get("Error", {}).get("Code") != "NoSuchLifecycleConfiguration":
+            raise
+    else:
+        if lifecycle.get("Rules"):
+            raise Refused("nonexpiring_recovery_bucket_required")
+    key = "kits/" + expected["kit_sha256"] + ".tar.gz"
+    private_directory(output.parent)
+    if output.exists() or output.is_symlink():
+        raise Refused("fresh_kit_readback_required")
+    # Conditional creation preserves an existing immutable kit. Readback must
+    # use a separately admitted recovery identity, never the uploader.
+    try:
+        with kit.open("rb") as source:
+            created = owner.put_object(Bucket=BUCKET, Key=key, Body=source, IfNoneMatch="*",
+                                       Metadata={"sha256": expected["kit_sha256"]})
+        selected = version(created.get("VersionId"))
+    except Exception as error:
+        if getattr(error, "response", {}).get("ResponseMetadata", {}).get("HTTPStatusCode") != 412:
+            raise
+        head = owner.head_object(Bucket=BUCKET, Key=key)
+        if (head.get("Metadata", {}).get("sha256") != expected["kit_sha256"]
+                or head.get("ContentLength") != expected["kit_bytes"]):
+            raise Refused("existing_kit_binding_changed")
+        selected = version(head.get("VersionId"))
+    response = reader.get_object(Bucket=BUCKET, Key=key, VersionId=selected)
+    if response.get("VersionId") != selected or response.get("ContentLength") != expected["kit_bytes"]:
+        response["Body"].close()
+        raise Refused("kit_readback_version_changed")
+    hashed, size = hashlib.sha256(), 0
+    with response["Body"] as body, output.open("xb") as target:
+        output.chmod(0o600)
+        while block := body.read(8 * 1024**2):
+            size += len(block)
+            if size > expected["kit_bytes"]:
+                raise Refused("kit_readback_exceeds_expected_size")
+            hashed.update(block)
+            target.write(block)
+    if size != expected["kit_bytes"] or hashed.hexdigest() != expected["kit_sha256"]:
+        raise Refused("kit_readback_hash_changed")
+    return {"status": "scaleway_recovery_kit_readback_verified", "bucket": BUCKET,
+            "object_key": key, "version_id": selected, "kit_sha256": hashed.hexdigest(), "kit_bytes": size,
+            "separate_reader_readback_verified": True, "private_identity_included": False,
+            "reader_write_denial_verified": False, "worker_access_denial_verified": False,
+            "reader_application_binding_verified": False, "restored_from_provider_kit": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    for name in ("kit", "kit-receipt", "owner-credentials", "reader-credentials", "output", "receipt"):
+        parser.add_argument("--" + name, type=Path)
+    args = parser.parse_args()
+    if not args.execute:
+        print(json.dumps({"status": "prepared", "bucket": BUCKET, "provider_contacted": False,
+                          "caller_credentials_read": False, "private_identity_included": False}))
+        return
+    if not all((args.kit, args.kit_receipt, args.owner_credentials, args.reader_credentials, args.output, args.receipt)):
+        raise Refused("attended_kit_arguments_required")
+    private_directory(args.receipt.parent)
+    if args.receipt.exists() or args.receipt.is_symlink():
+        raise Refused("fresh_receipt_required")
+    private_file(args.kit)
+    private_file(args.kit_receipt)
+    owner, reader = credentials(args.owner_credentials), credentials(args.reader_credentials)
+    if owner["AWS_ACCESS_KEY_ID"] == reader["AWS_ACCESS_KEY_ID"]:
+        raise Refused("separate_recovery_identity_required")
+    os.umask(0o077)
+    try:
+        result = publish(client(owner), client(reader), args.kit,
+                         json.loads(args.kit_receipt.read_text()), args.output)
+    except Exception as error:
+        checkpoint(args.receipt, {"status": "failed", "failure_type": type(error).__name__})
+        raise
+    checkpoint(args.receipt, result)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception:
+        raise SystemExit(1) from None

scripts/rehearse-ledger.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import re
 import secrets
 import subprocess
 import time
@@ -13,6 +14,7 @@ from pathlib import Path
 from ledger_archive import (
     Refused,
     checkpoint,
+    digest,
     private_directory,
     private_file,
     validate_capture,
@@ -89,7 +91,26 @@ print(json.dumps({'status':'isolated_ledger_verified','ledger_fingerprint':befor
 """
 
 
-def rehearse(recovered, receipt_path):
+def recovery_image(offline_binding=None):
+    if offline_binding is None:
+        return IMAGE
+    private_file(offline_binding)
+    binding = json.loads(offline_binding.read_text())
+    image_id = binding.get("image_id", "")
+    if (binding.get("schema") != "email-connect.ledger-recovery-kit.v1"
+            or binding.get("image") != IMAGE
+            or not isinstance(image_id, str)
+            or not re.fullmatch(r"sha256:[a-f0-9]{64}", image_id)):
+        raise Refused("offline_image_binding_required")
+    archive = offline_binding.parent / "receiver-image.tar"
+    private_file(archive)
+    expected = binding["files"]["receiver-image.tar"]
+    if archive.stat().st_size != expected["bytes"] or digest(archive) != expected["sha256"]:
+        raise Refused("offline_image_archive_changed")
+    return image_id
+
+
+def rehearse(recovered, receipt_path, offline_binding=None):
     private_directory(recovered)
     private_directory(receipt_path.parent)
     if receipt_path.exists() or receipt_path.is_symlink():
@@ -104,8 +125,9 @@ def rehearse(recovered, receipt_path):
     ):
         raise Refused("pinned_recovered_ledger_required")
     # Never pull a recovery image from the production forge during this rehearsal.
+    selected_image = recovery_image(offline_binding)
     inspected = subprocess.run(
-        ["docker", "image", "inspect", IMAGE], capture_output=True, timeout=30
+        ["docker", "image", "inspect", selected_image], capture_output=True, timeout=30
     )
     if inspected.returncode:
         raise Refused("cached_pinned_recovery_image_required")
@@ -119,6 +141,7 @@ def rehearse(recovered, receipt_path):
                     "docker",
                     "run",
                     "--rm",
+                    "--pull=never",
                     "--name",
                     name,
                     "-i",
@@ -145,7 +168,7 @@ def rehearse(recovered, receipt_path):
                     "PYTHONDONTWRITEBYTECODE=1",
                     "--entrypoint",
                     "python",
-                    IMAGE,
+                    selected_image,
                     "-c",
                     CODE,
                 ],
@@ -171,6 +194,7 @@ def rehearse(recovered, receipt_path):
         uid=verification["uid"],
         mail_sent=False,
         evidence_ceiling="isolated_ledger_rehearsal",
+        offline_image_binding_verified=offline_binding is not None,
         elapsed_seconds=round(time.monotonic() - started, 2),
     )
     checkpoint(receipt_path, receipt)
@@ -182,6 +206,7 @@ def main():
     parser.add_argument("--recovered", required=True, type=Path)
     parser.add_argument("--receipt", required=True, type=Path)
     parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--offline-image-binding", type=Path)
     args = parser.parse_args()
     if not args.execute:
         print(
@@ -190,7 +215,7 @@ def main():
             )
         )
         return 0
-    print(json.dumps(rehearse(args.recovered, args.receipt)))
+    print(json.dumps(rehearse(args.recovered, args.receipt, args.offline_image_binding)))
     return 0
 
 

scripts/scaleway-recovery-custody.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""Attended age escrow to a protected Scaleway Secret Manager secret, no key output."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hmac
+import json
+import os
+import subprocess
+import uuid
+from pathlib import Path
+from urllib.request import Request, urlopen
+
+from ledger_archive import Refused, checkpoint, private_directory, private_file
+
+NAME = "email-connect-ledger-age-recovery"
+REGION = "nl-ams"
+RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"
+BASE = "https://api.scaleway.com/secret-manager/v1beta1/regions/nl-ams/secrets/"
+
+
+def request(secret_id, suffix="", payload=None):
+    key = os.environ.get("SCW_SECRET_KEY", "")
+    if not key:
+        raise Refused("attended_scaleway_identity_required")
+    req = Request(BASE + secret_id + suffix,
+                  data=None if payload is None else json.dumps(payload).encode(),
+                  headers={"X-Auth-Token": key, "Content-Type": "application/json"},
+                  method="GET" if payload is None else "POST")
+    with urlopen(req, timeout=30) as response:
+        return json.loads(response.read(262144))
+
+
+def recipient(identity):
+    result = subprocess.run(["age-keygen", "-y", "/dev/stdin"], input=identity,
+                            capture_output=True, timeout=30)
+    if result.returncode:
+        raise Refused("age_identity_invalid")
+    return result.stdout.decode().strip()
+
+
+def escrow(secret_id, project_id, identity, transport=request):
+    uuid.UUID(secret_id)
+    uuid.UUID(project_id)
+    if recipient(identity) != RECIPIENT or not 0 < len(identity) <= 65536:
+        raise Refused("configured_age_recipient_required")
+    metadata = transport(secret_id)
+    if (metadata.get("name") != NAME or metadata.get("project_id") != project_id
+            or metadata.get("region") != REGION or metadata.get("type") != "opaque"
+            or metadata.get("protected") is not True
+            or metadata.get("ephemeral_policy") is not None
+            or metadata.get("deletion_requested_at") is not None):
+        raise Refused("protected_nonexpiring_exact_secret_required")
+    versions = transport(secret_id, "/versions?page_size=100")
+    count = versions.get("total_count")
+    if type(count) is not int or count != len(versions.get("versions", [])) or count > 1:
+        raise Refused("existing_versions_require_custodian_review")
+    if count == 0:
+        written = transport(secret_id, "/versions", {
+            "data": base64.b64encode(identity).decode(),
+            "description": "Existing email-connect age identity; attended recovery only.",
+            "disable_previous": False,
+        })
+        revision = written["revision"]
+        if revision != 1:
+            raise Refused("initial_secret_version_required")
+    else:
+        version = versions["versions"][0]
+        properties = version.get("ephemeral_properties") or {}
+        if (version.get("status") != "enabled" or properties.get("expires_at") is not None
+                or properties.get("expires_once_accessed") is True
+                or version.get("revision") != 1):
+            raise Refused("nonexpiring_enabled_version_required")
+        revision = 1
+    returned = transport(secret_id, "/versions/" + str(revision) + "/access")
+    restored = base64.b64decode(returned["data"], validate=True)
+    if (returned.get("secret_id") != secret_id or returned.get("revision") != revision
+            or not hmac.compare_digest(restored, identity) or recipient(restored) != RECIPIENT):
+        raise Refused("independent_escrow_readback_differs")
+    return {"status": "scaleway_age_escrow_verified", "secret_id": secret_id,
+            "revision": revision, "region": REGION, "public_recipient_matches": True,
+            "readback_equal": True, "private_identity_printed": False,
+            "independent_account_login_verified": False,
+            "worker_access_denial_verified": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--secret-id")
+    parser.add_argument("--project-id")
+    parser.add_argument("--identity", type=Path)
+    parser.add_argument("--receipt", type=Path)
+    args = parser.parse_args()
+    if not args.execute:
+        print(json.dumps({"status": "prepared", "name": NAME, "region": REGION,
+                          "protected": True, "ephemeral_policy": None,
+                          "provider_contacted": False, "private_identity_read": False}))
+        return
+    if not args.identity or not args.receipt or not args.receipt.is_absolute():
+        raise Refused("private_identity_and_receipt_paths_required")
+    private_directory(args.receipt.parent)
+    private_file(args.identity)
+    if args.receipt.exists() or args.receipt.is_symlink():
+        raise Refused("fresh_receipt_required")
+    try:
+        result = escrow(args.secret_id, args.project_id, args.identity.read_bytes())
+    except Exception as error:
+        checkpoint(args.receipt, {"status": "failed", "failure_type": type(error).__name__})
+        raise
+    checkpoint(args.receipt, result)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception:
+        raise SystemExit(1) from None