"""Immutable, nonexpiring Coulomb kit publication and independent retrieval.""" from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path import primary_application_archive as archive from application_recovery_identities import ( Refused, checkpoint, private_directory, private_json, ) from application_recovery_resources import s3, status PROJECT = "06d107bc-9d84-4672-b2a0-95701cc0c07f" BUCKET = "railiance-coulomb-social-recovery" MAX_BYTES = 5 * 1024**3 def download(client, ref, output): private_directory(output.parent) archive.require_version(ref.get("version_id")) if (ref.get("project_id") != PROJECT or ref.get("bucket") != BUCKET or not re.fullmatch(r"kits/[a-f0-9]{64}/recovery-kit.tar", ref.get("object_key", "")) or not re.fullmatch(r"[a-f0-9]{64}", ref.get("sha256", "")) or type(ref.get("bytes")) is not int or not 0 < ref["bytes"] <= MAX_BYTES or output.exists() or output.is_symlink()): raise Refused("exact_kit_reference_and_fresh_output_required") result = client.get_object(Bucket=BUCKET, Key=ref["object_key"], VersionId=ref["version_id"]) if result.get("VersionId") != ref["version_id"] or result["ContentLength"] != ref["bytes"]: result["Body"].close() raise Refused("kit_version_or_length_differs") size, hashed = 0, hashlib.sha256() with result["Body"] as body, output.open("xb") as stream: output.chmod(0o600) while block := body.read(8 * 1024**2): size += len(block) if size > ref["bytes"]: raise Refused("kit_readback_exceeds_size") stream.write(block) hashed.update(block) if size != ref["bytes"] or hashed.hexdigest() != ref["sha256"]: raise Refused("kit_readback_hash_differs") return dict(ref, independently_retrieved_hash_verified=True, automatic_expiry=False) def require_protected_bucket(client): if client.get_bucket_versioning(Bucket=BUCKET).get("Status") != "Enabled": raise Refused("protected_bucket_versioning_required") try: lifecycle = client.get_bucket_lifecycle_configuration(Bucket=BUCKET).get("Rules", []) except Exception as error: if status(error) != 404: raise lifecycle = [] if lifecycle: raise Refused("protected_bucket_must_have_no_lifecycle_expiry") def protect_receipt(writer, reader, receipt): """Retain the recovery proof independently, without a self-hash cycle.""" require_protected_bucket(writer) payload = json.dumps(receipt, sort_keys=True, separators=(",", ":")).encode() if len(payload) > 1024**2 or receipt.get("status") != "application_restored_verified": raise Refused("bounded_successful_recovery_proof_required") expected = hashlib.sha256(payload).hexdigest() key = "kits/receipts/" + expected + "/recovery-proof.json" try: head = writer.head_object(Bucket=BUCKET, Key=key) except Exception as error: if status(error) != 404: raise response = writer.put_object(Bucket=BUCKET, Key=key, Body=payload, IfNoneMatch="*", Metadata={"sha256": expected}, ContentType="application/json") version = archive.require_version(response.get("VersionId")) else: if head["ContentLength"] != len(payload) or head.get("Metadata", {}).get("sha256") != expected: raise Refused("existing_proof_binding_differs") version = archive.require_version(head.get("VersionId")) response = reader.get_object(Bucket=BUCKET, Key=key, VersionId=version) with response["Body"] as stream: if response.get("VersionId") != version or stream.read(len(payload) + 1) != payload: raise Refused("independent_proof_readback_differs") return {"project_id": PROJECT, "bucket": BUCKET, "object_key": key, "version_id": version, "sha256": expected, "bytes": len(payload), "independent_readback_verified": True, "automatic_expiry": False} def publish(client, source): archive.regular_file(source) require_protected_bucket(client) size = source.stat().st_size if not 0 < size <= MAX_BYTES: raise Refused("bounded_kit_required") expected = archive.digest(source) key = "kits/" + expected + "/recovery-kit.tar" try: head = client.head_object(Bucket=BUCKET, Key=key) except Exception as error: if status(error) != 404: raise with source.open("rb") as stream: created = client.put_object(Bucket=BUCKET, Key=key, Body=stream, ContentLength=size, IfNoneMatch="*", Metadata={"sha256": expected}) version = archive.require_version(created.get("VersionId")) else: if head["ContentLength"] != size or head.get("Metadata", {}).get("sha256") != expected: raise Refused("existing_kit_binding_differs") version = archive.require_version(head.get("VersionId")) return {"schema": "platform.coulomb-protected-recovery-kit.v1", "project_id": PROJECT, "bucket": BUCKET, "object_key": key, "version_id": version, "sha256": expected, "bytes": size, "automatic_expiry": False} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("operation", choices=("publish", "retrieve")) 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("--source", type=Path) parser.add_argument("--receipt", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() if not args.execute: print(json.dumps({"status": "preview", "provider_contacted": False, "automatic_expiry": False})) return private_directory(args.directory) private_directory(args.receipt.parent) if args.operation == "publish": if not args.source or args.receipt.exists() or args.receipt.is_symlink(): raise Refused("source_and_fresh_receipt_required") ref = publish(s3(private_json(args.directory / "custodian-pair.json"), PROJECT), args.source) checkpoint(args.receipt, ref) else: ref = private_json(args.receipt) result = download(s3(private_json(args.directory / "reader-pair.json"), PROJECT), ref, args.output) print(json.dumps({"status": "protected_kit_independently_retrieved_verified", "kit": result, "application_recovery_from_kit_verified": False, "private_values_printed": False})) if __name__ == "__main__": try: main() except Exception as error: # noqa: BLE001 -- never print SDK content/credential diagnostics print(json.dumps({"status": "refused", "failure_type": type(error).__name__, "native_status": status(error)})) sys.exit(1)