#!/usr/bin/env python3 """Independent verifier for the sealed DualCoach activation-cutover successor.""" from __future__ import annotations import argparse import hashlib import json import re import stat import zipfile from pathlib import Path from typing import Any CONTROLS = {"candidate-manifest.json", "candidate-checkpoint.json", "hash-inventory.json", "verifier-input.json"} def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False).encode() def sha(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def file_sha(path: Path) -> str: return sha(path.read_bytes()) def obj(path: Path) -> dict[str, Any]: value = json.loads(path.read_bytes()) if not isinstance(value, dict): raise AssertionError(f"JSON object required: {path}") return value def list_sha(value: object) -> str: if not isinstance(value, list): raise AssertionError("list required") return sha(canonical(value)) def safe_tree(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 root.rglob("*"): info = path.lstat() if path.is_symlink(): raise AssertionError(f"symlink forbidden: {path}") if stat.S_ISDIR(info.st_mode): if stat.S_IMODE(info.st_mode) != 0o500: raise AssertionError(f"directory mode drift: {path}") elif stat.S_ISREG(info.st_mode): if stat.S_IMODE(info.st_mode) != 0o400 or info.st_nlink != 1: raise AssertionError(f"file mode/link drift: {path}") else: raise AssertionError(f"unsupported entry: {path}") def verify_wheel(path: Path, expected: dict[str, Any]) -> None: if file_sha(path) != expected["sha256"] or path.stat().st_size != expected["bytes"]: raise AssertionError("wheel byte drift") with zipfile.ZipFile(path) as archive: members = [] for item in sorted(archive.infolist(), key=lambda row: row.filename): if not item.is_dir(): raw = archive.read(item.filename) members.append({"path": item.filename, "bytes": len(raw), "sha256": sha(raw)}) timestamps = {item.date_time for item in archive.infolist()} for source, member in expected["production_parity"].items(): if archive.read(member) != (path.parent.parent / "source-delta/after" / source).read_bytes(): raise AssertionError(f"source/wheel parity drift: {source}") if len(members) != expected["member_count"] or list_sha(members) != expected["members_sha256"]: raise AssertionError("wheel member closure drift") if timestamps != {(2000, 1, 1, 0, 0, 0)}: raise AssertionError("wheel timestamps are not reproducible") def verify(input_path: Path) -> dict[str, object]: verifier_input = obj(input_path) if verifier_input.get("schema") != "task26-activation-cutover-successor-verifier-input-v1": raise AssertionError("verifier input schema mismatch") root = Path(str(verifier_input["candidate_root"])) safe_tree(root) manifest_path = root / "candidate-manifest.json" checkpoint_path = root / "candidate-checkpoint.json" inventory_path = root / "hash-inventory.json" manifest = obj(manifest_path) checkpoint = obj(checkpoint_path) inventory = obj(inventory_path) if file_sha(manifest_path) != verifier_input["manifest_sha256"]: raise AssertionError("manifest hash mismatch") if file_sha(checkpoint_path) != verifier_input["checkpoint_sha256"]: raise AssertionError("checkpoint hash mismatch") if file_sha(inventory_path) != manifest["hash_inventory"]["sha256"]: raise AssertionError("inventory hash mismatch") full = manifest.pop("full_candidate_digest", None) core = manifest.pop("core_candidate_digest", None) calculated_core = sha(canonical(manifest)) calculated_full = sha(canonical({"core_candidate_digest": calculated_core, "manifest_core": manifest})) if core != calculated_core or full != calculated_full or full != verifier_input["full_candidate_digest"]: raise AssertionError("candidate digest mismatch") manifest["core_candidate_digest"] = core manifest["full_candidate_digest"] = full entries = inventory.get("entries") if not isinstance(entries, list) or list_sha(entries) != inventory.get("entries_sha256"): raise AssertionError("inventory digest mismatch") indexed = {row["path"] for row in entries} actual = {str(path.relative_to(root)) for path in root.rglob("*") if path.is_file()} if actual != indexed | CONTROLS or indexed & CONTROLS: extra = sorted(actual - indexed - CONTROLS) missing = sorted(indexed - actual) raise AssertionError(f"unindexed or missing candidate bytes: extra={extra}, missing={missing}") for row in entries: path = root / row["path"] if file_sha(path) != row["sha256"] or path.stat().st_size != row["bytes"]: raise AssertionError(f"indexed byte drift: {row['path']}") closure = obj(root / manifest["executable_closure"]["path"]) old_closure = obj(root / manifest["lineage"]["predecessor_closure_path"]) if list_sha(closure["entries"]) != closure["entries_sha256"]: raise AssertionError("successor closure digest mismatch") old = {row["path"]: row for row in old_closure["entries"]} new = {row["path"]: row for row in closure["entries"]} delta = [{"path": p, "before": old.get(p), "after": new.get(p)} for p in sorted(set(old) | set(new)) if old.get(p) != new.get(p)] declared = obj(root / manifest["source_delta"]["path"]) if delta != declared["executable_delta"] or list_sha(delta) != declared["executable_delta_sha256"]: raise AssertionError("unexpected executable source delta") if len(delta) != 5 or declared["source_delta_count"] != 6: raise AssertionError("source delta count mismatch") expected_paths = set(manifest["source_delta"]["expected_paths"]) if {row["path"] for row in declared["source_delta"]} != expected_paths or len(expected_paths) != 6: raise AssertionError("unexpected declared source delta") for row in declared["source_delta"]: after = root / "source-delta/after" / row["path"] if file_sha(after) != row["after"]["sha256"] or after.stat().st_size != row["after"]["bytes"]: raise AssertionError(f"source snapshot drift: {row['path']}") before = row["before"] if before is not None and old.get(row["path"]) != before: raise AssertionError(f"predecessor source pin drift: {row['path']}") predecessor_manifest = obj(root / manifest["lineage"]["predecessor_manifest_path"]) if predecessor_manifest.get("full_candidate_digest") != manifest["lineage"]["predecessor_full_digest"]: raise AssertionError("predecessor lineage mismatch") for field in ("predecessor_manifest_sha256", "predecessor_checkpoint_sha256", "predecessor_closure_sha256"): path_field = field.replace("_sha256", "_path") if file_sha(root / manifest["lineage"][path_field]) != manifest["lineage"][field]: raise AssertionError(f"predecessor artifact drift: {field}") for name in ("pre.nul", "post.nul"): status_path = root / "status" / name if file_sha(status_path) != manifest["repository_status"]["sha256"]: raise AssertionError("repository status drift") if (root / "status/pre.nul").read_bytes() != (root / "status/post.nul").read_bytes(): raise AssertionError("repository changed during build") verify_wheel(root / manifest["wheel"]["path"], manifest["wheel"]) focused = (root / "verification/focused-49-tests.txt").read_text() installed = (root / "verification/wheel-installed-tests.txt").read_text() if "49 passed" not in focused or "40 passed" not in installed: raise AssertionError("test gate receipt mismatch") if (root / "verification/ruff.txt").read_text().strip() != "All checks passed!": raise AssertionError("Ruff gate receipt mismatch") new_types = obj(root / "verification/basedpyright-new-files.json")["summary"] full_types = obj(root / "verification/basedpyright-full.json")["summary"] if (new_types["errorCount"], new_types["warningCount"]) != (0, 0): raise AssertionError("new-file type gate mismatch") if (full_types["errorCount"], full_types["warningCount"]) != (4, 96): raise AssertionError("parent type-error classification mismatch") help_text = (root / "verification/wheel-installed-cli-help.txt").read_text() if "usage: dualcoach_admin customer activate" not in help_text or "--bootstrap-session" not in help_text: raise AssertionError("installed CLI command missing") expected_checkpoint = { "schema": "task26-activation-cutover-successor-checkpoint-v1", "status": "PASS_SEALED_SUCCESSOR_CANDIDATE_NOT_DEPLOYED", "full_candidate_digest": full, "core_candidate_digest": core, "manifest_sha256": verifier_input["manifest_sha256"], "wheel_sha256": manifest["wheel"]["sha256"], "predecessor_full_digest": manifest["lineage"]["predecessor_full_digest"], "source_delta_count": 6, "executable_delta_count": 5, "inventory_entry_count": len(entries), "focused_test_count": 49, "wheel_test_count": 40, } if checkpoint != expected_checkpoint: raise AssertionError("checkpoint content mismatch") return { "status": "PASS", "full_candidate_digest": full, "core_candidate_digest": core, "wheel_sha256": manifest["wheel"]["sha256"], "source_delta_count": 6, "executable_delta_count": 5, "inventory_entry_count": len(entries), "focused_test_count": 49, "wheel_test_count": 40, "unindexed_byte_count": 0, "predecessor_full_digest": manifest["lineage"]["predecessor_full_digest"], } 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())