849f3947dc
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
89 lines
4.2 KiB
Python
89 lines
4.2 KiB
Python
"""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)
|