Establish Coulomb scoped recovery API custody and native boundaries

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b

scripts/application_recovery_access.py

@@ -0,0 +1,89 @@
+"""Routine Coulomb recovery access; scoped custodian API, no Console session."""
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import sys
+from pathlib import Path
+
+from application_recovery_identities import Refused, private_directory, private_json
+from application_recovery_resources import API, SECRET_BASE, public_recipient, status
+
+CONTRACT = Path(__file__).resolve().parents[1] / "docs/credential-lane-designs/coulomb-recovery-api.json"
+PROJECT = "06d107bc-9d84-4672-b2a0-95701cc0c07f"
+
+
+def retrieve(api, contract, material):
+    if (contract.get("schema") != "platform.coulomb-recovery-api.v1"
+            or contract.get("project_id") != PROJECT):
+        raise Refused("exact_coulomb_contract_required")
+    reference = contract["materials"][material]
+    path = SECRET_BASE + "/" + reference["id"]
+    metadata = api("GET", path)
+    if (metadata.get("id") != reference["id"] or metadata.get("name") != reference["name"]
+            or metadata.get("project_id") != PROJECT or metadata.get("protected") is not True
+            or metadata.get("type") != "opaque" or metadata.get("ephemeral_policy") is not None
+            or metadata.get("deletion_requested_at") is not None):
+        raise Refused("protected_exact_escrow_required")
+    version = api("GET", path + "/versions/" + str(reference["revision"]))
+    properties = version.get("ephemeral_properties") or {}
+    if (version.get("status") != "enabled" or properties.get("expires_at")
+            or properties.get("expires_once_accessed")):
+        raise Refused("enabled_nonexpiring_exact_version_required")
+    response = api("GET", path + "/versions/" + str(reference["revision"]) + "/access")
+    if response.get("secret_id") != reference["id"] or response.get("revision") != reference["revision"]:
+        raise Refused("escrow_response_binding_refused")
+    payload = base64.b64decode(response["data"], validate=True)
+    if material == "age":
+        if public_recipient(payload) != contract["public_recipient"]:
+            raise Refused("public_recipient_differs")
+    else:
+        pair = json.loads(payload)
+        if set(pair) != {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"} or any(
+            not isinstance(v, str) or not v or "\n" in v for v in pair.values()
+        ):
+            raise Refused("exact_api_pair_required")
+    return payload
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--directory", type=Path, default=Path.home() / ".config/railiance-platform/coulomb-social-recovery")
+    parser.add_argument("--material", choices=("age", "reader-pair", "custodian-pair"), default="age")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    if not args.execute:
+        print(json.dumps({"status": "preview", "material": args.material, "provider_contacted": False,
+                          "personal_login_required": False}))
+        return
+    private_directory(args.directory)
+    if args.output:
+        if not args.output.is_absolute():
+            raise Refused("absolute_private_output_required")
+        private_directory(args.output.parent)
+        if args.output.exists() or args.output.is_symlink():
+            raise Refused("fresh_private_output_required")
+    contract = json.loads(CONTRACT.read_text())
+    pair = private_json(args.directory / "custodian-pair.json")
+    payload = retrieve(API(pair), contract, args.material)
+    if args.output:
+        fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        with os.fdopen(fd, "wb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+    print(json.dumps({"status": "scoped_api_retrieval_verified", "material": args.material,
+                      "private_output_created": bool(args.output), "credential_values_printed": False,
+                      "personal_session_used": False}))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:  # noqa: BLE001 -- suppress secret-bearing provider diagnostics
+        print(json.dumps({"status": "refused", "failure_type": type(error).__name__,
+                          "native_status": status(error)}))
+        sys.exit(1)

scripts/application_recovery_resources.py

@@ -7,6 +7,7 @@ from __future__ import annotations
 
 import argparse
 import base64
+import fcntl
 import hashlib
 import hmac
 import json
@@ -180,6 +181,16 @@ class Resources:
     def save(self):
         checkpoint(self.path, self.state)
 
+    def deny_s3(self, role, operation, **arguments):
+        # Some denied PUTs leave a provider keep-alive connection unusable for
+        # the next request (observed HTTP400 with an empty error). Isolate each
+        # negative probe so it cannot corrupt subsequent positive read proof.
+        client = s3(self.pairs[role], self.project)
+        try:
+            return denied(lambda: getattr(client, operation)(**arguments))
+        finally:
+            client.close()
+
     def mutate(self, label, call):
         self.state["pending"] = label
         self.save()
@@ -360,21 +371,20 @@ class Resources:
                 raise Refused("independent_version_listing_required")
             proofs[prefix + "_reader_version_get_hash"] = {"verified": True, "version_id": version}
             for role in ("reader", "custodian", "publisher"):
-                client = self.clients[role]
                 # All deletion targets are newly made synthetic canaries. No
                 # actual retained backup is ever a destructive negative probe.
-                proofs[prefix + "_" + role + "_delete"] = denied(lambda client=client, bucket=bucket, key=key, version=version: client.delete_object(
-                    Bucket=bucket, Key=key, VersionId=version))
+                proofs[prefix + "_" + role + "_delete"] = self.deny_s3(role, "delete_object",
+                    Bucket=bucket, Key=key, VersionId=version)
                 intended = self.state["buckets"][bucket]["policy"]
-                proofs[prefix + "_" + role + "_policy_write"] = denied(lambda client=client, bucket=bucket, intended=intended: client.put_bucket_policy(
-                    Bucket=bucket, Policy=json.dumps(intended)))
-                proofs[prefix + "_" + role + "_outside_prefix"] = denied(lambda client=client, bucket=bucket: client.get_object(
-                    Bucket=bucket, Key="outside-approved-prefix/synthetic-absent"))
-            proofs[prefix + "_reader_put"] = denied(lambda reader=reader, bucket=bucket, prefix=prefix: reader.put_object(
-                Bucket=bucket, Key=prefix + "/reader-write-denial.txt", Body=b"synthetic-only"))
+                proofs[prefix + "_" + role + "_policy_write"] = self.deny_s3(role, "put_bucket_policy",
+                    Bucket=bucket, Policy=json.dumps(intended))
+                proofs[prefix + "_" + role + "_outside_prefix"] = self.deny_s3(role, "get_object",
+                    Bucket=bucket, Key="outside-approved-prefix/synthetic-absent")
+            proofs[prefix + "_reader_put"] = self.deny_s3("reader", "put_object",
+                Bucket=bucket, Key=prefix + "/reader-write-denial.txt", Body=b"synthetic-only")
         kit = self.state["fixtures"]["kits"]
-        proofs["publisher_protected_kit_get"] = denied(lambda: self.clients["publisher"].get_object(
-            Bucket=kit["bucket"], Key=kit["key"], VersionId=kit["version_id"]))
+        proofs["publisher_protected_kit_get"] = self.deny_s3("publisher", "get_object",
+            Bucket=kit["bucket"], Key=kit["key"], VersionId=kit["version_id"])
         publisher = self.clients["publisher"]
         # Verify ordinary worker upload/readback in addition to its multipart
         # abort. No content is captured from the application by this helper.
@@ -425,7 +435,11 @@ def main():
         return
     import uuid
     uuid.UUID(args.owner_user_id)
-    print(json.dumps(Resources(args.directory, args.owner_user_id).run()))
+    private_directory(args.directory)
+    fd = os.open(args.directory / "resource-bootstrap.lock", os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(fd, "w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        print(json.dumps(Resources(args.directory, args.owner_user_id).run()))
 
 
 if __name__ == "__main__":