diff --git a/scripts/application_recovery_kit.py b/scripts/application_recovery_kit.py new file mode 100644 index 0000000..d3149fc --- /dev/null +++ b/scripts/application_recovery_kit.py @@ -0,0 +1,108 @@ +"""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 publish(client, source): + archive.regular_file(source) + 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) diff --git a/tests/test_application_recovery_kit.py b/tests/test_application_recovery_kit.py new file mode 100644 index 0000000..40e6895 --- /dev/null +++ b/tests/test_application_recovery_kit.py @@ -0,0 +1,91 @@ +import hashlib +import io +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +import application_recovery_kit as kit + + +class Missing(Exception): + def __init__(self): + self.response = {"ResponseMetadata": {"HTTPStatusCode": 404}} + + +class Provider: + def __init__(self): + self.data, self.calls = None, [] + self.version = "synthetic-kit-version" + self.corrupt = False + + def head_object(self, **args): + self.calls.append("head") + if self.data is None: + raise Missing() + return {"ContentLength": len(self.data), "VersionId": self.version, + "Metadata": {"sha256": hashlib.sha256(self.data).hexdigest()}} + + def put_object(self, **args): + self.calls.append("put") + assert args["Bucket"] == kit.BUCKET and args["IfNoneMatch"] == "*" + self.data = args["Body"].read() + return {"VersionId": self.version} + + def get_object(self, **args): + self.calls.append("get") + assert args["Bucket"] == kit.BUCKET and args["VersionId"] == self.version + body = b"X" * len(self.data) if self.corrupt else self.data + return {"VersionId": self.version, "ContentLength": len(body), "Body": io.BytesIO(body)} + + +@pytest.fixture +def fixture(tmp_path, monkeypatch): + tmp_path.chmod(0o700) + # Real outside-Git enforcement is covered in identity tests; isolate the + # synthetic provider paths from a workstation's possible /tmp Git ancestor. + monkeypatch.setattr(kit, "private_directory", kit.archive.private_directory) + source = tmp_path / "kit.tar" + source.write_bytes(b"synthetic-public-kit-no-private-keys") + source.chmod(0o600) + return source, Provider() + + +def test_immutable_versioned_kit_readback_and_idempotent_publication(fixture, tmp_path): + source, client = fixture + ref = kit.publish(client, source) + assert ref["automatic_expiry"] is False + assert ref["object_key"] == "kits/" + hashlib.sha256(source.read_bytes()).hexdigest() + "/recovery-kit.tar" + assert kit.download(client, ref, tmp_path / "fetched.tar")["independently_retrieved_hash_verified"] + assert kit.publish(client, source) == ref + assert client.calls.count("put") == 1 + + +def test_corruption_is_not_accepted_as_independent_kit(fixture, tmp_path): + source, client = fixture + ref = kit.publish(client, source) + client.corrupt = True + with pytest.raises(kit.Refused, match="kit_readback_hash_differs"): + kit.download(client, ref, tmp_path / "fetched.tar") + + +@pytest.mark.parametrize("field,value", [("project_id", "foreign"), ("bucket", "foreign"), + ("object_key", "kits/foreign.tar")]) +def test_foreign_reference_refused_before_get(fixture, tmp_path, field, value): + source, client = fixture + ref = kit.publish(client, source) + ref[field] = value + with pytest.raises(kit.Refused): + kit.download(client, ref, tmp_path / "fetched.tar") + assert "get" not in client.calls + + +def test_existing_output_is_never_overwritten(fixture, tmp_path): + source, client = fixture + ref = kit.publish(client, source) + output = tmp_path / "fetched.tar" + output.write_bytes(b"keep") + with pytest.raises(kit.Refused): + kit.download(client, ref, output) + assert output.read_bytes() == b"keep"