#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import importlib.util import json import os import stat import sys from pathlib import Path from typing import Any, cast def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def document(path: Path, label: str) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ValueError(f"{label} is invalid") return cast(dict[str, Any], value) def load_module(path: Path, name: str) -> Any: spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise ValueError(f"module unavailable: {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def verify_modes(bundle: Path, *, allow_mutable: bool) -> None: for path in (bundle, *sorted(bundle.rglob("*"))): info = path.lstat() if stat.S_ISLNK(info.st_mode) or info.st_uid != os.geteuid(): raise ValueError(f"unsafe bundle path: {path}") if path.is_dir(): allowed = {0o500, 0o700} if allow_mutable else {0o500} elif path.is_file() and info.st_nlink == 1: allowed = {0o400, 0o600} if allow_mutable else {0o400} else: raise ValueError(f"bundle path type is invalid: {path}") if stat.S_IMODE(info.st_mode) not in allowed: raise ValueError(f"bundle mode is not frozen: {path}") def verify(bundle: Path, runtime_path: Path | None, *, allow_mutable: bool) -> dict[str, object]: bundle = bundle.absolute() verify_modes(bundle, allow_mutable=allow_mutable) candidate = document(bundle / "candidate.json", "candidate manifest") unsigned_candidate = {key: value for key, value in candidate.items() if key != "document_sha256"} if candidate.get("document_sha256") != hashlib.sha256(canonical(unsigned_candidate)).hexdigest(): raise ValueError("candidate manifest digest is invalid") candidate_digest = str(candidate.get("candidate_digest", "")) authority_module = load_module(bundle / "gateway/platforms/task26_candidate_authority.py", "sealed_task26_authority") authority = authority_module.verify_candidate_authority(bundle, candidate_digest) expected_heads = { "registry_head_sha256": candidate.get("authority_registry_head_sha256"), "ledger_head_sha256": candidate.get("authority_ledger_head_sha256"), } if any(authority.get(key) != value for key, value in expected_heads.items()): raise ValueError("candidate authority heads differ from manifest") expected = document(bundle / "sealed-expected-state.json", "sealed expected state") final_state_module = load_module( bundle / "gateway/platforms/task26_final_state.py", "sealed_task26_final_state" ) hermes_wheel = next((bundle / "artifacts").glob("hermes_agent-*.whl")) profile_wheel = next((bundle / "artifacts").glob("physique_checkin_cli-*.whl")) final_state_module.verify_expected_state( bundle, bundle / "sealed-expected-state.json", hermes_wheel=hermes_wheel, profile_wheel=profile_wheel, ) wheelhouse = final_state_module.verify_wheelhouse_inventory( bundle / "wheelhouse", expected.get("wheelhouse"), required_wheels=(profile_wheel, hermes_wheel), ) ledger_ready = document(bundle / "ledger-ready-receipt.json", "ledger-ready receipt") if ledger_ready.get("document_sha256") != hashlib.sha256(canonical({key: value for key, value in ledger_ready.items() if key != "document_sha256"})).hexdigest(): raise ValueError("ledger-ready receipt digest is invalid") if ledger_ready != { "schema": "task26-ledger-ready-receipt-v1", "candidate_digest": candidate_digest, "registry_head_sha256": authority["registry_head_sha256"], "ledger_head_sha256": authority["ledger_head_sha256"], "expected_state_sha256": sha256_file(bundle / "sealed-expected-state.json"), "document_sha256": ledger_ready["document_sha256"], }: raise ValueError("ledger-ready receipt binding is invalid") if candidate.get("expected_state_sha256") != sha256_file(bundle / "sealed-expected-state.json") or candidate.get("ledger_ready_sha256") != sha256_file(bundle / "ledger-ready-receipt.json"): raise ValueError("candidate final-state binding is invalid") excluded = {"candidate.json", "SEAL.json"} actual = [ {"path": path.relative_to(bundle).as_posix(), "sha256": sha256_file(path), "size": path.stat().st_size} for path in sorted(bundle.rglob("*")) if path.is_file() and path.relative_to(bundle).as_posix() not in excluded ] if actual != candidate.get("evidence_inventory"): raise ValueError("candidate evidence inventory differs") seal = document(bundle / "SEAL.json", "bundle seal") unsigned_seal = {key: value for key, value in seal.items() if key != "document_sha256"} if seal.get("document_sha256") != hashlib.sha256(canonical(unsigned_seal)).hexdigest() or seal.get("candidate_manifest_sha256") != sha256_file(bundle / "candidate.json") or seal.get("candidate_digest") != candidate_digest or seal.get("registry_head_sha256") != authority["registry_head_sha256"] or seal.get("ledger_head_sha256") != authority["ledger_head_sha256"]: raise ValueError("bundle seal binding is invalid") verifier_module = load_module(bundle / "verification-tools/verify_source_golden_path.py", "sealed_golden_verifier") installed_verified = verifier_module.verify(bundle / "installed-golden-bundle") contract = installed_verified.get("task26_contract") installed_receipt = document(bundle / "receipts/installed-golden-verifier-1.json", "installed Golden receipt") if contract != installed_receipt.get("task26_contract") or installed_receipt.get("candidate_digest") != candidate_digest: raise ValueError("installed seven-clause receipt differs") observer = { "backend": installed_receipt.get("observer_backend"), "directory_resource_count": installed_receipt.get("observer_directory_resource_count"), "inotify_watch_count": installed_receipt.get("observer_inotify_watch_count"), "receipt_count": installed_receipt.get("observer_receipt_count"), } if observer != { "backend": "dnotify_signalfd_v1", "directory_resource_count": 1, "inotify_watch_count": 0, "receipt_count": 5, }: raise ValueError("zero-inotify observer receipt differs") rehydrated = False if runtime_path is not None: runtime = verifier_module.read_private_json(runtime_path) verified = verifier_module.verify(bundle / "installed-golden-bundle", runtime_override=runtime) if verified.get("status") != "ACTUAL_INSTALLED_GOLDEN_PATH_PASS" or verified.get("candidate_digest") != candidate_digest or verified.get("task26_contract") != contract: raise ValueError("rehydrated installed Golden verification differs") rehydrated = True result = { "status": "TASK26_FROZEN_CANDIDATE_PASS", "candidate_digest": candidate_digest, "product_digest": candidate.get("product_digest"), "golden_runtime_digest": candidate.get("golden_runtime_digest"), "candidate_authority": authority, "runtime_mode": "installed", "task26_contract": contract, "installed_task26_contract": contract, "observer": observer, "isolated_execution": expected["runtime_portable"]["origin_proof"]["execution"], "installed_origin_proof": expected["runtime_portable"]["origin_proof"], "trust_boundary": expected["trust_boundary"], "wheelhouse_inventory_sha256": wheelhouse["inventory_sha256"], "wheelhouse_entry_count": len(wheelhouse["entries"]), "source_tree_digest": expected["candidate_parity"]["source"]["source_tree_digest"], "candidate_parity_sha256": expected["candidate_parity"]["parity_sha256"], "cross_mode_candidate_parity": expected["candidate_parity"], "frozen_modes_verified": not allow_mutable, "rehydrated_runtime_verified": rehydrated, } if runtime_path is not None and expected.get("candidate_digest") != candidate_digest: raise ValueError("sealed expected state candidate differs") return result def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("bundle", type=Path) parser.add_argument("--rehydrated-runtime", type=Path) parser.add_argument("--allow-mutable", action="store_true") args = parser.parse_args() try: result = verify(args.bundle, args.rehydrated_runtime, allow_mutable=args.allow_mutable) except (OSError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"status": "TASK26_FROZEN_CANDIDATE_FAIL", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 print(json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())