#!/usr/bin/env python3 """Independently verify an immutable repaired-source Task26 successor candidate.""" from __future__ import annotations import argparse import hashlib import json import stat import zipfile from pathlib import Path from typing import Any def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False).encode() def sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def sha256_file(path: Path) -> str: return sha256(path.read_bytes()) def load_object(path: Path) -> dict[str, Any]: value = json.loads(path.read_bytes()) if not isinstance(value, dict): raise AssertionError(f"{path} is not a JSON object") return value def assert_file(path: Path) -> None: info = path.lstat() if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: raise AssertionError(f"unsafe candidate file: {path}") if stat.S_IMODE(info.st_mode) != 0o400: raise AssertionError(f"candidate file mode drift: {path}") def assert_tree_modes(root: Path) -> None: info = root.lstat() if root.is_symlink() or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o500: raise AssertionError("candidate root is unsafe or mutable") for path in sorted(root.rglob("*")): info = path.lstat() if path.is_symlink(): raise AssertionError(f"candidate contains symlink: {path}") if stat.S_ISDIR(info.st_mode): if stat.S_IMODE(info.st_mode) != 0o500: raise AssertionError(f"candidate directory mode drift: {path}") elif stat.S_ISREG(info.st_mode): assert_file(path) else: raise AssertionError(f"unsupported candidate entry: {path}") def list_digest(value: object) -> str: if not isinstance(value, list): raise AssertionError("expected a list") return sha256(canonical(value)) def verify_wheel(path: Path, expected: dict[str, Any]) -> None: assert_file(path) if sha256_file(path) != expected.get("sha256") or path.stat().st_size != expected.get("bytes"): raise AssertionError("wheel bytes drifted") with zipfile.ZipFile(path) as archive: members = [ {"path": item.filename, "bytes": len(raw), "sha256": sha256(raw)} for item in sorted(archive.infolist(), key=lambda value: value.filename) if not item.is_dir() for raw in (archive.read(item.filename),) ] timestamps = {item.date_time for item in archive.infolist()} if list_digest(members) != expected.get("members_sha256"): raise AssertionError("wheel member closure drifted") if timestamps != {(2000, 1, 1, 0, 0, 0)}: raise AssertionError("wheel metadata is not reproducible") def verify(input_path: Path) -> dict[str, object]: assert_file(input_path) verifier_input = load_object(input_path) if verifier_input.get("schema") != "task26-repaired-archive-successor-verifier-input-v1": raise AssertionError("verifier input schema mismatch") root = Path(str(verifier_input["candidate_root"])) assert_tree_modes(root) manifest_path = root / "candidate-manifest.json" checkpoint_path = root / "candidate-checkpoint.json" manifest = load_object(manifest_path) checkpoint = load_object(checkpoint_path) if sha256_file(manifest_path) != verifier_input.get("manifest_sha256"): raise AssertionError("manifest hash mismatch") if sha256_file(checkpoint_path) != verifier_input.get("checkpoint_sha256"): raise AssertionError("checkpoint hash mismatch") full_digest = manifest.pop("full_candidate_digest", None) core_digest = manifest.pop("core_candidate_digest", None) calculated_core = sha256(canonical(manifest)) calculated_full = sha256(canonical({"core_candidate_digest": calculated_core, "manifest_core": manifest})) if core_digest != calculated_core or full_digest != calculated_full: raise AssertionError("candidate digest mismatch") if verifier_input.get("full_candidate_digest") != calculated_full: raise AssertionError("verifier input candidate mismatch") manifest["core_candidate_digest"] = core_digest manifest["full_candidate_digest"] = full_digest expected_checkpoint = { "schema": "task26-repaired-archive-successor-checkpoint-v1", "status": "SEALED_SUCCESSOR_CANDIDATE_NOT_TASK26_PASS", "full_candidate_digest": full_digest, "core_candidate_digest": core_digest, "manifest_sha256": verifier_input["manifest_sha256"], "wheel_sha256": manifest["wheel"]["sha256"], "repository_status_sha256": manifest["repository_status"]["sha256"], "repository_status_entry_count": manifest["repository_status"]["entry_count"], "archive_binding_sha256": manifest["historical_archive_binding"]["binding_sha256"], } supersession = manifest.get("supersession") if isinstance(supersession, dict): expected_checkpoint["predecessor_full_digest"] = supersession["predecessor_full_digest"] expected_checkpoint["exact_source_delta_sha256"] = supersession["exact_source_delta_sha256"] trainer_binding_manifest = manifest.get("trainer_free_v4_successor_input") if isinstance(trainer_binding_manifest, dict): expected_checkpoint["trainer_free_v4_binding_sha256"] = trainer_binding_manifest["binding_sha256"] if checkpoint != expected_checkpoint: raise AssertionError("checkpoint content mismatch") for name in ("pre.nul", "post.nul"): path = root / "status" / name assert_file(path) if sha256_file(path) != manifest["repository_status"]["sha256"]: raise AssertionError("status snapshot mismatch") if (root / "status/pre.nul").read_bytes() != (root / "status/post.nul").read_bytes(): raise AssertionError("pre/post status snapshots differ") bindings = [ (root / "bindings/executable-closure.json", manifest["executable_closure"]), (root / "bindings/historical-archive-binding.json", manifest["historical_archive_binding"]), (root / "bindings/task22-25-reconciliation-index.json", manifest["task22_25_reconciliation"]), ] if isinstance(trainer_binding_manifest, dict): bindings.append((root / "bindings/trainer-free-v4-successor-binding.json", trainer_binding_manifest)) for path, section in bindings: if sha256_file(path) != section["artifact_sha256"]: raise AssertionError(f"binding artifact drifted: {path.name}") closure = load_object(bindings[0][0]) if list_digest(closure.get("entries")) != closure.get("entries_sha256"): raise AssertionError("executable closure digest mismatch") hashes = {entry["path"]: entry["sha256"] for entry in closure["entries"]} repaired = manifest["repaired_source"] if hashes.get(repaired["controller_path"]) != repaired["controller_sha256"]: raise AssertionError("repaired controller is absent from closure") if hashes.get(repaired["test_path"]) != repaired["test_sha256"]: raise AssertionError("repaired test is absent from closure") if isinstance(supersession, dict): predecessor_manifest_path = root / str(supersession["predecessor_manifest_path"]) predecessor_checkpoint_path = root / str(supersession["predecessor_checkpoint_path"]) predecessor_closure_path = root / str(supersession["predecessor_closure_path"]) if sha256_file(predecessor_manifest_path) != supersession["predecessor_manifest_sha256"]: raise AssertionError("predecessor manifest copy mismatch") if sha256_file(predecessor_checkpoint_path) != supersession["predecessor_checkpoint_sha256"]: raise AssertionError("predecessor checkpoint copy mismatch") if sha256_file(predecessor_closure_path) != supersession["predecessor_closure_sha256"]: raise AssertionError("predecessor closure copy mismatch") predecessor_manifest = load_object(predecessor_manifest_path) predecessor_checkpoint = load_object(predecessor_checkpoint_path) predecessor_closure = load_object(predecessor_closure_path) predecessor_digest = supersession["predecessor_full_digest"] if predecessor_manifest.get("full_candidate_digest") != predecessor_digest or predecessor_checkpoint.get("full_candidate_digest") != predecessor_digest: raise AssertionError("predecessor digest mismatch") old_entries = predecessor_closure.get("entries") new_entries = closure.get("entries") if not isinstance(old_entries, list) or not isinstance(new_entries, list): raise AssertionError("successor closure entries are malformed") old = {entry["path"]: entry for entry in old_entries} new = {entry["path"]: entry for entry in new_entries} calculated_delta = [ {"path": path, "before": old.get(path), "after": new.get(path)} for path in sorted(set(old) | set(new)) if old.get(path) != new.get(path) ] if calculated_delta != supersession.get("exact_source_delta"): raise AssertionError("declared successor delta is not exact") if sha256(canonical(calculated_delta)) != supersession.get("exact_source_delta_sha256"): raise AssertionError("successor delta digest mismatch") expected_count = supersession.get("expected_source_delta_count") if expected_count is None: if len(calculated_delta) != 1 or calculated_delta[0]["path"] != supersession.get("expected_delta_path"): raise AssertionError("successor has additional source drift") if hashes.get(supersession["status_test_path"]) != supersession.get("status_test_sha256"): raise AssertionError("status test pin drift") elif len(calculated_delta) != expected_count: raise AssertionError("successor source delta count mismatch") if isinstance(trainer_binding_manifest, dict): trainer_path = root / "bindings/trainer-free-v4-successor-binding.json" trainer = load_object(trainer_path) trainer_core = dict(trainer) trainer_digest = trainer_core.pop("binding_sha256", None) if trainer_digest != sha256(canonical(trainer_core)) or trainer_digest != trainer_binding_manifest.get("binding_sha256"): raise AssertionError("trainer-free v4 binding digest mismatch") copied_entries = trainer.get("copied_entries") if not isinstance(copied_entries, list) or list_digest(copied_entries) != trainer.get("copied_entries_sha256"): raise AssertionError("trainer-free copied entry inventory mismatch") copied_root = root / str(trainer_binding_manifest["copied_root"]) expected_names = {entry["path"] for entry in copied_entries} observed_names = {path.name for path in copied_root.iterdir() if path.is_file()} if expected_names != observed_names or len(copied_entries) != trainer_binding_manifest.get("copied_leaf_count"): raise AssertionError("trainer-free copied leaf set mismatch") for entry in copied_entries: path = copied_root / entry["path"] if sha256_file(path) != entry["sha256"] or path.stat().st_size != entry["bytes"]: raise AssertionError("trainer-free copied leaf drift") required = { "bundle-manifest-v4-authoritative.json": trainer["bundle_manifest_sha256"], "candidate-checkpoint-v3-ready.json": trainer["checkpoint_sha256"], "final-reaudit-receipt-v4-ready.json": trainer["reaudit_sha256"], "v1-runtime-closure-manifest-v2.json": trainer["runtime_closure_sha256"], "sealer-input-index-v3.json": trainer["sealer_input_index_sha256"], "authoritative-inventory-v2.json": trainer["authoritative_inventory_sha256"], "inventory-v2-supersession-receipt.json": trainer["inventory_supersession_sha256"], } if any(sha256_file(copied_root / name) != digest for name, digest in required.items()): raise AssertionError("trainer-free authoritative control mismatch") bundle_manifest = load_object(copied_root / "bundle-manifest-v4-authoritative.json") if list_digest(bundle_manifest.get("entries")) != bundle_manifest.get("bundle_digest") or bundle_manifest.get("bundle_digest") != trainer.get("bundle_digest"): raise AssertionError("trainer-free bundle digest mismatch") index = load_object(copied_root / "sealer-input-index-v3.json") index_digest = index.pop("input_index_digest", None) if index_digest != trainer.get("sealer_input_index_digest") or sha256(canonical(index)) != index_digest: raise AssertionError("trainer-free input index digest mismatch") reaudit = load_object(copied_root / "final-reaudit-receipt-v4-ready.json") fields = reaudit.get("field_results") if not isinstance(fields, list) or len(fields) != 23 or reaudit.get("field_result_count") != 23: raise AssertionError("trainer-free checkpoint fields missing") if any(row.get("status") != "PASS_INPUT" for row in fields) or list_digest(fields) != trainer.get("checkpoint_fields_sha256"): raise AssertionError("trainer-free checkpoint field result mismatch") closure_v2 = load_object(copied_root / "v1-runtime-closure-manifest-v2.json") absence = closure_v2.get("required_absence_controls") if absence != trainer.get("required_absence_controls") or list_digest(absence) != trainer.get("required_absence_controls_sha256"): raise AssertionError("trainer-free absence controls mismatch") archive_binding = load_object(bindings[1][0]) copied_manifest_path = root / "historical/archive-manifest.json" copied_receipt_path = root / "historical/archive-receipt.json" copied_manifest = load_object(copied_manifest_path) copied_receipt = load_object(copied_receipt_path) if sha256_file(copied_manifest_path) != archive_binding["manifest_sha256"]: raise AssertionError("archive manifest copy mismatch") if sha256_file(copied_receipt_path) != archive_binding["receipt_sha256"]: raise AssertionError("archive receipt copy mismatch") if sha256(canonical(copied_manifest)) != copied_receipt.get("archive_digest"): raise AssertionError("archive canonical digest mismatch") if len(copied_manifest.get("scopes", [])) != 23: raise AssertionError("archive does not contain 23 scopes") if sha256(canonical(copied_manifest["scopes"])) != archive_binding["scope_digest_sha256"]: raise AssertionError("23-scope binding mismatch") binding_core = dict(archive_binding) binding_digest = binding_core.pop("binding_sha256", None) if binding_digest != sha256(canonical(binding_core)): raise AssertionError("archive binding digest mismatch") reconciliation = load_object(bindings[2][0]) if reconciliation.get("historical_execution_retargeted_to_successor") is not False: raise AssertionError("historical evidence was impermissibly retargeted") for entry in reconciliation.get("indexes", []): copied = root / "historical/reconciliation-indexes" / Path(entry["path"]).name if sha256_file(copied) != entry["sha256"]: raise AssertionError("historical index mismatch") verify_wheel(root / str(manifest["wheel"]["path"]), manifest["wheel"]) return { "status": "PASS", "full_candidate_digest": full_digest, "core_candidate_digest": core_digest, "manifest_sha256": verifier_input["manifest_sha256"], "checkpoint_sha256": verifier_input["checkpoint_sha256"], "wheel_sha256": manifest["wheel"]["sha256"], "status_sha256": manifest["repository_status"]["sha256"], "status_entry_count": manifest["repository_status"]["entry_count"], "archive_binding_sha256": manifest["historical_archive_binding"]["binding_sha256"], "historical_execution_retargeted_to_successor": False, "predecessor_full_digest": supersession.get("predecessor_full_digest") if isinstance(supersession, dict) else None, "exact_source_delta_sha256": supersession.get("exact_source_delta_sha256") if isinstance(supersession, dict) else None, "trainer_free_v4_binding_sha256": trainer_binding_manifest.get("binding_sha256") if isinstance(trainer_binding_manifest, dict) else None, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("input", type=Path) args = parser.parse_args() print(json.dumps(verify(args.input), sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())