diff --git a/scripts/application_recovery_drill.py b/scripts/application_recovery_drill.py new file mode 100644 index 0000000..f4de1b8 --- /dev/null +++ b/scripts/application_recovery_drill.py @@ -0,0 +1,175 @@ +"""Timed Coulomb recovery using only the provider kit, escrow and held API callers.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +import tarfile +import time +from datetime import UTC, datetime +from pathlib import Path + +from application_recovery_access import retrieve +from application_recovery_identities import ( + Refused, + checkpoint, + private_directory, + private_json, +) +from application_recovery_kit import PROJECT, download +from application_recovery_resources import API, s3, status + + +def extract_verified(archive, destination): + if destination.exists() or destination.is_symlink(): + raise Refused("fresh_kit_extraction_required") + with tarfile.open(archive, "r:") as bundle: + members = bundle.getmembers() + names = [m.name for m in members] + if (not 1 < len(members) < 10000 or len(set(names)) != len(names) + or any(not m.isfile() or m.sparse is not None or Path(m.name).is_absolute() + or ".." in Path(m.name).parts for m in members) + or sum(m.size for m in members) > 5 * 1024**3): + raise Refused("bounded_regular_unique_kit_members_required") + metadata = bundle.getmember("manifest.json") + if not 0 < metadata.size <= 1024**2: + raise Refused("bounded_kit_manifest_required") + manifest = json.loads(bundle.extractfile(metadata).read()) + if manifest.get("schema") != "platform.coulomb-recovery-kit-manifest.v1": + raise Refused("exact_kit_manifest_required") + files = manifest["files"] + if set(files) | {"manifest.json"} != set(names): + raise Refused("complete_manifest_members_required") + if manifest.get("private_identity_included") is not False or manifest.get("api_pairs_included") is not False: + raise Refused("separate_credential_custody_required") + for member in members: + if member.name == "manifest.json": + continue + expected = files[member.name] + if (member.size != expected["bytes"] or not re.fullmatch(r"[a-f0-9]{64}", expected["sha256"])): + raise Refused("kit_member_binding_differs") + with bundle.extractfile(member) as stream: + if hashlib.file_digest(stream, "sha256").hexdigest() != expected["sha256"]: + raise Refused("kit_member_hash_differs") + # No executable or image is used until every member is verified. + destination.mkdir(mode=0o700) + for member in members: + target = destination / member.name + target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + with bundle.extractfile(member) as source, os.fdopen(fd, "wb") as output: + while block := source.read(8 * 1024**2): + output.write(block) + return manifest + + +class Drill: + def __init__(self, directory, destination, reference): + private_directory(directory) + private_directory(destination.parent) + if destination.exists() or destination.is_symlink(): + raise Refused("fresh_private_drill_destination_required") + destination.mkdir(mode=0o700) + self.directory, self.root, self.reference = directory, destination, reference + self.stages = {} + + def stage(self, name, call): + began = time.monotonic() + value = call() + self.stages[name] = round(time.monotonic() - began, 3) + checkpoint(self.root / "stage-receipt.json", {"stages": self.stages, "last_completed": name}) + return value + + def command(self, name, arguments): + result = subprocess.run(arguments, capture_output=True, timeout=300, check=False) + if result.returncode: + fd = os.open(self.root / (name + "-private-diagnostics.txt"), + os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(result.stdout + result.stderr) + raise Refused("private_drill_child_refused") + return result.stdout + + def run(self): + began, started = time.monotonic(), datetime.now(UTC).isoformat() + kit = self.root / "retrieved-kit.tar" + reader = s3(private_json(self.directory / "reader-pair.json"), PROJECT) + self.stage("provider_kit_retrieval", lambda: download(reader, self.reference, kit)) + contents = self.root / "kit" + manifest = self.stage("all_kit_member_hashes_and_extraction", lambda: extract_verified(kit, contents)) + contract = json.loads((contents / "tools/docs/credential-lane-designs/coulomb-recovery-api.json").read_text()) + custodian = API(private_json(self.directory / "custodian-pair.json")) + identity = self.stage("provider_age_escrow_retrieval", lambda: retrieve(custodian, contract, "age")) + identity_path = self.root / "identity.agekey" + fd = os.open(identity_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(identity) + del identity + for kind in ("application", "postgres"): + self.stage(kind + "_image_load", lambda kind=kind: self.command(kind + "-image-load", + ["docker", "load", "-i", str(contents / (kind + "-image.tar"))])) + image_id = manifest["application_image_id" if kind == "application" else "postgres_image_id"] + actual = self.command(kind + "-image-inspect", ["docker", "image", "inspect", image_id, + "--format", "{{.Id}}"]) + if actual.decode().strip() != image_id: + raise Refused("retained_image_id_differs") + tools = contents / "tools" + pair = self.root / "pair" + unpacked = self.stage("decrypt_and_validate_pair", lambda: self.command("decrypt", [sys.executable, + str(tools / "recover-paired-backup.py"), "--execute", "--archive", str(contents / "packet.tar.age"), + "--transfer-receipt", str(contents / "primary-receipt.json"), "--identity", str(identity_path), + "--destination", str(pair)])) + unpacked = json.loads(unpacked) + restored = self.stage("isolated_database_content_restore", lambda: self.command("restore", [sys.executable, + str(tools / "restore-paired-snapshot.py"), str(pair), "--keep-instance", "--postgres-image", manifest["postgres_image_id"]])) + restored = json.loads(restored) + app = self.stage("packaged_app_views_and_permissions", lambda: self.command("application", [sys.executable, + str(tools / "verify-restored-application.py"), "--snapshot", str(pair), "--restore-receipt", + str(Path(restored["restored_directory"]) / "restore-receipt.json"), "--kit-manifest", str(contents / "manifest.json"), + "--output", str(self.root / "application-receipt.json")])) + app = json.loads(app) + baseline = json.loads((contents / "application-recovery.json").read_text()) + for key in ("actual_pages_rendered_through_view", "same_tenant_ungranted_page_denials", "content_files_unchanged"): + if app.get(key) != baseline.get(key): + raise Refused("kit_application_baseline_differs") + if app["status"] != "isolated_application_recovery_verified": + raise Refused("actual_application_recovery_required") + result = {"schema": "platform.coulomb-provider-kit-recovery.v1", "status": "application_restored_verified", + "started_at": started, "completed_at": datetime.now(UTC).isoformat(), "stages_seconds": self.stages, + "recovery_seconds": round(time.monotonic() - began, 3), "protected_kit": self.reference, + "kit_members_verified": len(manifest["files"]) + 1, "application": "coulomb-social", + "captured_at": manifest["captured_at"], "application_restore_verified": True, + "all_data_images_and_tools_from_independent_provider_kit": True, + "age_identity_from_separate_provider_escrow": True, "original_node_or_plaintext_used": False, + "personal_session_used": False, "registry_pulls_performed": False, "fresh_docker_daemon": False, + "operator_account_recovery_time_included": False, "real_browser_journey_verified": False, + "unpack": unpacked, "restore": restored, "application_checks": app, "timers_activated": False} + checkpoint(self.root / "recovery-receipt.json", result) + return result + + +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("--kit-reference", type=Path, required=True) + parser.add_argument("--destination", type=Path, required=True) + args = parser.parse_args() + if not args.execute: + print(json.dumps({"status": "preview", "provider_contacted": False, "production_restore": False})) + return + result = Drill(args.directory, args.destination, private_json(args.kit_reference)).run() + print(json.dumps({k: result[k] for k in ("status", "recovery_seconds", "kit_members_verified", "stages_seconds")})) + + +if __name__ == "__main__": + try: + main() + except Exception as error: # noqa: BLE001 -- child diagnostics stay owner-private + print(json.dumps({"status": "refused", "failure_type": type(error).__name__, "native_status": status(error), + "reason": str(error) if isinstance(error, Refused) else "private_review_required"})) + sys.exit(1) diff --git a/tests/test_application_recovery_drill.py b/tests/test_application_recovery_drill.py new file mode 100644 index 0000000..9c3cb83 --- /dev/null +++ b/tests/test_application_recovery_drill.py @@ -0,0 +1,65 @@ +"""Kit integrity failures must refuse before extracting executable material.""" +import hashlib +import io +import json +import sys +import tarfile +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +import application_recovery_drill as drill + + +def kit(path, mutation=None): + payload = b"synthetic recovery code" + manifest = {"schema": "platform.coulomb-recovery-kit-manifest.v1", + "private_identity_included": False, "api_pairs_included": False, + "files": {"tools/recover.py": {"bytes": len(payload), + "sha256": hashlib.sha256(payload).hexdigest()}}} + name = "../escape" if mutation == "traversal" else "tools/recover.py" + if mutation == "credential": + manifest["private_identity_included"] = True + with tarfile.open(path, "w") as archive: + data = json.dumps(manifest).encode() + meta = tarfile.TarInfo("manifest.json") + meta.size = len(data) + archive.addfile(meta, io.BytesIO(data)) + item = tarfile.TarInfo(name) + item.size = len(payload) + if mutation == "link": + item.type, item.linkname, item.size = tarfile.SYMTYPE, "/etc/passwd", 0 + actual = b"X" * len(payload) if mutation == "hash" else payload + archive.addfile(item, io.BytesIO(actual)) + if mutation == "duplicate": + archive.addfile(item, io.BytesIO(actual)) + + +def test_every_member_is_verified_before_extraction(tmp_path): + source = tmp_path / "kit.tar" + kit(source) + result = drill.extract_verified(source, tmp_path / "extracted") + assert (tmp_path / "extracted/tools/recover.py").read_bytes() == b"synthetic recovery code" + assert result["api_pairs_included"] is False + + +@pytest.mark.parametrize("mutation", ["traversal", "link", "duplicate", "hash", "credential"]) +def test_unsafe_or_corrupt_kit_never_extracts_code(tmp_path, mutation): + source = tmp_path / "kit.tar" + kit(source, mutation) + output = tmp_path / "extracted" + with pytest.raises(drill.Refused): + drill.extract_verified(source, output) + assert not output.exists() + + +def test_existing_extraction_is_preserved(tmp_path): + source = tmp_path / "kit.tar" + kit(source) + output = tmp_path / "existing" + output.mkdir() + (output / "keep").write_text("keep") + with pytest.raises(drill.Refused): + drill.extract_verified(source, output) + assert (output / "keep").read_text() == "keep"