#!/usr/bin/env python3 """Independently recompute the frozen DualCoach candidate evidence.""" from __future__ import annotations import base64 import hashlib import json import subprocess from pathlib import Path from typing import cast EVIDENCE = Path(__file__).resolve().parent MANIFEST = EVIDENCE / "dualcoach-candidate-manifest.json" FREEZE = EVIDENCE / "dualcoach-owner-v1-freeze-receipt.json" PRE_APPROVAL_FREEZE = EVIDENCE / "dualcoach-owner-v1-freeze-receipt-pre-approval.json" POST_APPROVAL = EVIDENCE / "dualcoach-task-22-post-approval-evidence.json" PRE_SUPPLEMENTAL = EVIDENCE / "dualcoach-task-22-pre-supplemental-evidence.json" HISTORICAL_FREEZE = Path("/tmp/dualcoach-task22-freeze-receipt.json") PROFILE_ROOT = Path("/home/cube/.hermes/profiles/dualcoachtest") MIGRATION_JOURNAL = PROFILE_ROOT / ( "data/customers/task22_dm_rehearsal/nutrition-onboarding/" "migrations/owner-v1.json" ) def canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def run(command: list[str], *, cwd: Path) -> subprocess.CompletedProcess[bytes]: return subprocess.run(command, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) def main() -> int: manifest_bytes = MANIFEST.read_bytes() manifest = cast(dict[str, object], json.loads(manifest_bytes)) roots_raw = cast(dict[str, object], manifest["roots"]) roots = {name: Path(cast(str, value)) for name, value in roots_raw.items()} leaves = cast(list[object], manifest["candidate_files"]) candidate = hashlib.sha256() package = hashlib.sha256() diff = hashlib.sha256() previous: tuple[bytes, bytes] | None = None category_counts: dict[str, int] = {} for raw_leaf in leaves: leaf = cast(dict[str, object], raw_leaf) root_name = cast(str, leaf["root"]) relative = cast(str, leaf["path"]) category = cast(str, leaf["diff_category"]) key = (root_name.encode(), relative.encode()) if previous is not None and key <= previous: raise ValueError("candidate leaves are not in strict canonical order") previous = key content = (roots[root_name] / relative).read_bytes() if leaf["bytes"] != len(content) or leaf["sha256"] != hashlib.sha256(content).hexdigest(): raise ValueError(f"candidate leaf mismatch: {root_name}:{relative}") candidate.update(key[0] + b"\0" + key[1] + b"\0" + content + b"\0") if root_name == "profile_package": package.update(key[1] + b"\0" + content + b"\0") if root_name == "gateway": tracked = run( ["git", "ls-files", "--error-unmatch", "--", relative], cwd=roots[root_name], ).returncode == 0 expected_category = "gateway-tracked" if tracked else "gateway-untracked" else: expected_category = "profile-package-external" if category != expected_category: raise ValueError(f"diff category mismatch: {root_name}:{relative}") if category == "gateway-tracked": result = run( [ "git", "diff", "--binary", "--no-ext-diff", "--no-textconv", "HEAD", "--", relative, ], cwd=roots[root_name], ) if result.returncode != 0: raise ValueError(result.stderr.decode(errors="replace")) else: result = run( [ "git", "diff", "--no-index", "--binary", "--no-ext-diff", "--", "/dev/null", f"./{relative}", ], cwd=roots[root_name], ) if result.returncode not in {0, 1}: raise ValueError(result.stderr.decode(errors="replace")) for frame in (key[0], key[1], category.encode(), result.stdout): diff.update(frame + b"\0") category_counts[category] = category_counts.get(category, 0) + 1 if candidate.hexdigest() != manifest["candidate_digest"]: raise ValueError("candidate aggregate mismatch") if package.hexdigest() != manifest["profile_package_digest"]: raise ValueError("package aggregate mismatch") if diff.hexdigest() != manifest["candidate_diff_sha256"]: raise ValueError("candidate diff aggregate mismatch") if category_counts != manifest["candidate_diff_category_counts"]: raise ValueError("candidate diff category counts mismatch") status = run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=roots["gateway"], ) if status.returncode != 0: raise ValueError(status.stderr.decode(errors="replace")) status_records = [record for record in status.stdout.split(b"\0") if record] if ( hashlib.sha256(status.stdout).hexdigest() != manifest["gateway_status_snapshot_sha256"] or len(status.stdout) != manifest["gateway_status_byte_length"] or len(status_records) != manifest["gateway_status_entry_count"] ): raise ValueError("gateway status evidence mismatch") freeze = cast(dict[str, object], json.loads(FREEZE.read_bytes())) unsigned = {key: value for key, value in freeze.items() if key != "receipt_digest"} if hashlib.sha256(canonical(unsigned)).hexdigest() != freeze["receipt_digest"]: raise ValueError("freeze receipt seal mismatch") if ( freeze["release_candidate_digest"] != manifest["candidate_digest"] or freeze["profile_package_digest"] != manifest["profile_package_digest"] or freeze["manifest_digest"] != hashlib.sha256(manifest_bytes).hexdigest() ): raise ValueError("freeze receipt release binding mismatch") historical = cast(dict[str, object], json.loads(HISTORICAL_FREEZE.read_bytes())) historical_paths = cast(dict[str, object], historical["files"]) historical_fingerprints = cast(dict[str, object], historical["fingerprints"]) current_fingerprints = cast(dict[str, object], freeze["files"]) journal = cast(dict[str, object], json.loads(MIGRATION_JOURNAL.read_bytes())) journal_unsigned = { key: value for key, value in journal.items() if key != "journal_digest" } if hashlib.sha256(canonical(journal_unsigned)).hexdigest() != journal["journal_digest"]: raise ValueError("migration journal seal mismatch") pre_approval = cast( dict[str, object], json.loads(PRE_APPROVAL_FREEZE.read_bytes()) ) pre_approval_unsigned = { key: value for key, value in pre_approval.items() if key != "receipt_digest" } journal_pins = cast(dict[str, object], journal["pins"]) if ( hashlib.sha256(canonical(pre_approval_unsigned)).hexdigest() != pre_approval["receipt_digest"] or journal_pins["candidate_digest"] != pre_approval["release_candidate_digest"] or journal_pins["package_digest"] != pre_approval["profile_package_digest"] or journal_pins["manifest_digest"] != pre_approval["manifest_digest"] or journal_pins["freeze_receipt_digest"] != pre_approval["receipt_digest"] ): raise ValueError("pre-approval release binding mismatch") retained = cast(dict[str, object], journal["source_files"]) retained_digests = cast(dict[str, object], journal["source_file_digests"]) if set(retained) != set(historical_paths): raise ValueError("retained frozen file set mismatch") for name in historical_paths: content = base64.b64decode(cast(str, retained[name]), validate=True) historical_record = historical_fingerprints[name] expected = ( cast(str, cast(dict[str, object], historical_record)["sha256"]) if isinstance(historical_record, dict) else cast(str, historical_record) ) current = cast(dict[str, object], current_fingerprints[name]) if ( hashlib.sha256(content).hexdigest() != expected or retained_digests[name] != expected or current["sha256"] != expected ): raise ValueError(f"retained frozen byte mismatch: {name}") if current["bytes"] != len(content): raise ValueError(f"retained frozen length mismatch: {name}") post = cast(dict[str, object], json.loads(POST_APPROVAL.read_bytes())) post_unsigned = {key: value for key, value in post.items() if key != "evidence_digest"} facts = cast(dict[str, object], post["facts"]) if hashlib.sha256(canonical(post_unsigned)).hexdigest() != post["evidence_digest"]: raise ValueError("post-approval evidence seal mismatch") if ( facts.get("owner_update_id") != 629525050 or facts.get("owner_update_commit_count") != 1 or facts.get("owner_review_generation") != 25 or facts.get("owner_review_message_id") != 120 or facts.get("ready_generation") != 26 or facts.get("ready_message_id") != 121 or facts.get("handler_exception_count") != 1 or facts.get("business_commit_reconciled_count") != 1 or facts.get("ingress_receipt_schema_version") != 2 or facts.get("ingress_terminal_receipt_count") != 1 or facts.get("ingress_pending_receipt_count") != 0 or facts.get("owner_reviewed") is not True or facts.get("gateway_healthy") is not True or facts.get("customer_enabled") is not False or facts.get("activation_enabled") is not False or facts.get("delivery_enabled") is not False or facts.get("feature_epoch_present") is not False ): raise ValueError("post-approval evidence binding mismatch") post_files = cast(dict[str, object], post["files"]) for name, raw_record in post_files.items(): record = cast(dict[str, object], raw_record) content = (PROFILE_ROOT / cast(str, record["relative_path"])).read_bytes() if ( record["bytes"] != len(content) or record["sha256"] != hashlib.sha256(content).hexdigest() ): raise ValueError(f"post-approval file mismatch: {name}") pre = cast(dict[str, object], json.loads(PRE_SUPPLEMENTAL.read_bytes())) pre_unsigned = {key: value for key, value in pre.items() if key != "evidence_digest"} pre_facts = cast(dict[str, object], pre["facts"]) if ( hashlib.sha256(canonical(pre_unsigned)).hexdigest() != pre["evidence_digest"] or pre["release_candidate_digest"] != manifest["candidate_digest"] or pre["profile_package_digest"] != manifest["profile_package_digest"] or pre["manifest_digest"] != hashlib.sha256(manifest_bytes).hexdigest() or pre["freeze_receipt_digest"] != freeze["receipt_digest"] or pre_facts.get("generation_25_classification") != "historical_provenance_incomplete" or pre_facts.get("generation_25_committed_count") != 1 or pre_facts.get("generation_25_receipt_valid") is not True or pre_facts.get("generation_26_classification") != "historical_ready" or pre_facts.get("generation_26_committed_count") != 1 or pre_facts.get("generation_26_receipt_valid") is not True or pre_facts.get("supplemental_journal_present") is not False or pre_facts.get("supplemental_lock_present") is not False or pre_facts.get("callback_envelope_present") is not False or pre_facts.get("customer_enabled") is not False or pre_facts.get("activation_enabled") is not False or pre_facts.get("delivery_enabled") is not False or pre_facts.get("activation_count") != 0 or pre_facts.get("delivery_count") != 0 or pre_facts.get("later_publication_count") != 0 or pre_facts.get("gateway_active_state") != "active" or pre_facts.get("gateway_sub_state") != "running" or pre_facts.get("gateway_profile_bound") is not True ): raise ValueError("pre-supplemental evidence binding mismatch") absent = cast(dict[str, object], pre["absent_authorities"]) for name, relative in absent.items(): if (PROFILE_ROOT / cast(str, relative)).exists(): raise ValueError(f"pre-supplemental absent authority appeared: {name}") pre_files = cast(dict[str, object], pre["files"]) for name, raw_record in pre_files.items(): record = cast(dict[str, object], raw_record) content = (PROFILE_ROOT / cast(str, record["relative_path"])).read_bytes() if ( record["bytes"] != len(content) or record["sha256"] != hashlib.sha256(content).hexdigest() ): raise ValueError(f"pre-supplemental file mismatch: {name}") print(json.dumps({ "candidate": candidate.hexdigest(), "diff": diff.hexdigest(), "freeze": freeze["receipt_digest"], "frozen_files": len(historical_paths), "leaves": len(leaves), "manifest": hashlib.sha256(manifest_bytes).hexdigest(), "package": package.hexdigest(), "post_approval": post["evidence_digest"], "pre_supplemental": pre["evidence_digest"], "status": hashlib.sha256(status.stdout).hexdigest(), }, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())