#!/usr/bin/env python3 """Verify capital1000 A->B linkage + historical package genesis binding.""" from __future__ import annotations import hashlib, json, sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) def h(b: bytes)->str: return hashlib.sha256(b).hexdigest() def main()->int: fails=[] binding="FAIL" gsha=(ROOT/"prospective/genesis/GENESIS.sha256").read_text().strip().split()[0] if h((ROOT/"prospective/genesis/GENESIS.canonical").read_bytes())!=gsha: fails.append("genesis sha") genesis=json.loads((ROOT/"prospective/genesis/GENESIS.json").read_text(encoding="utf-8")) pkg=(ROOT/"historical/PACKAGE.sha256").read_text().strip().split()[0] if h((ROOT/"historical/PACKAGE.canonical").read_bytes())!=pkg: fails.append("historical package sha") bound=genesis.get("historical_capital1000_package_sha256") if bound == pkg: binding="PASS" else: binding="FAIL" fails.append("historical_package_genesis_binding_mismatch") print("HISTORICAL_PACKAGE_GENESIS_BINDING="+binding) tsr = ROOT/"prospective/genesis/GENESIS.tsr" if not tsr.exists(): fails.append("genesis tsr missing") else: try: from rfc3161_lib import verify_tsr trust = HERE/"TSA_TRUST" if trust.exists(): v = verify_tsr( tsr, expected_digest_hex=gsha, trust_dir=trust, receipt_json=ROOT/"prospective/genesis/GENESIS_rfc3161_receipt.json", expect_authority="freetsa", ) for k in ("RFC3161_PARSE","RFC3161_IMPRINT","RFC3161_TIMESTAMP","RFC3161_SIGNATURE"): if str(v.get(k,"")).upper()!="PASS": fails.append("genesis_tsr_"+k) except Exception as e: fails.append("genesis_tsr_verify_error:"+str(e)[:80]) latest=json.loads((ROOT/"prospective/latest.json").read_text(encoding="utf-8")) if latest.get("missing_decisions",1)!=0: fails.append("missing_decisions!=0") if latest.get("duplicate_decisions",1)!=0: fails.append("duplicate_decisions!=0") src = int(latest.get("source_signals_after_genesis") or 0) acc = int(latest.get("accepted_count") or 0) rej = int(latest.get("rejected_capital_count") or 0) if src != acc + rej: fails.append("signal/decision count mismatch") dec_path = ROOT/"prospective/chain/DECISIONS.jsonl" decisions=[] if dec_path.exists() and dec_path.read_text(encoding="utf-8").strip(): for ln in dec_path.read_text(encoding="utf-8").splitlines(): if ln.strip(): decisions.append(json.loads(ln)) if len(decisions) != src: fails.append("decisions_count!=source_signals") seen=set() for d in decisions: sid=d.get("source_signal_id") or d.get("source_signal_hash") if sid in seen: fails.append("duplicate_capital_decision") seen.add(sid) if d.get("decision") not in ("ACCEPT","REJECT_CAPITAL"): fails.append("invalid_decision") a2=sum(1 for d in decisions if d.get("decision")=="ACCEPT") r2=sum(1 for d in decisions if d.get("decision")=="REJECT_CAPITAL") if a2!=acc or r2!=rej: fails.append("decision_counts_vs_latest") if decisions and decisions[0].get("previous_executable_root")!=gsha: fails.append("first_previous_root!=genesis") prev=gsha for d in decisions: if d.get("previous_executable_root")!=prev: fails.append("broken_previous_root_chain") break if d.get("decision_commitment_sha256"): chk=dict(d) er=chk.pop("executable_root", None) commit=chk.pop("decision_commitment_sha256", None) can=(json.dumps(chk, sort_keys=True, separators=(",", ":"))+"\n").encode("utf-8") if h(can)!=commit: fails.append("decision_commitment_mismatch:"+str(d.get("event_id"))) if er!=h(can+prev.encode("utf-8")): fails.append("executable_root_mismatch:"+str(d.get("event_id"))) prev=d.get("executable_root") if decisions and latest.get("executable_root")!=decisions[-1].get("executable_root"): fails.append("latest_root!=last_decision_root") if float(latest.get("equity_eur", 1000)) != 1000.0: life=ROOT/"prospective/chain/LIFECYCLE.jsonl" has_result=False if life.exists(): for ln in life.read_text(encoding="utf-8").splitlines(): if ln.strip() and "RESULT_OBSERVED" in ln: has_result=True break if not has_result: fails.append("equity_tamper_without_result") for day_json in sorted((ROOT/"prospective/days").glob("*/day.json")): daily=json.loads(day_json.read_text(encoding="utf-8")) if daily.get("source_signals_seen") is not None and daily.get("accepted_count") is not None: if int(daily.get("accepted_count") or 0)+int(daily.get("rejected_capital_count") or 0) != int(daily.get("source_signals_seen") or 0): fails.append("daily_count_mismatch:"+day_json.parent.name) if decisions and daily.get("daily_executable_root") not in {None, latest.get("executable_root"), decisions[-1].get("executable_root")}: if daily.get("daily_executable_root") != latest.get("executable_root"): fails.append("daily_root_mismatch:"+day_json.parent.name) pol_sha=(ROOT/"policy/CAPITAL_POLICY_COMMITMENT.sha256").read_text().strip().split()[0] if h((ROOT/"policy/CAPITAL_POLICY_COMMITMENT.canonical").read_bytes())!=pol_sha: fails.append("policy sha") # EPOCH-1 history integrity if present e1 = ROOT/"prospective/history/TOX-DEMO-1000-EPOCH-1" if e1.exists(): e1sha=(e1/"GENESIS.sha256").read_text().strip().split()[0] if h((e1/"GENESIS.canonical").read_bytes())!=e1sha: fails.append("epoch1_history_genesis_sha") st=json.loads((e1/"EPOCH_STATUS.json").read_text(encoding="utf-8")) if st.get("status")!="SUPERSEDED_BEFORE_FIRST_DECISION": fails.append("epoch1_status_unexpected") # archived epoch1 must FAIL the binding invariant vs final package e1g=json.loads((e1/"GENESIS.json").read_text(encoding="utf-8")) if e1g.get("historical_capital1000_package_sha256")==pkg: fails.append("epoch1_unexpectedly_matches_final_package") if fails: print("CAPITAL1000_LINKAGE_VERIFY=FAIL") for f in fails: print("FAIL="+f) return 1 print("CAPITAL1000_LINKAGE_VERIFY=PASS") print("GENESIS_SHA256="+gsha) print("ACTIVE_EPOCH="+str(genesis.get("epoch_id"))) print("MISSING_DECISIONS=0") print("DUPLICATE_DECISIONS=0") return 0 if __name__=="__main__": raise SystemExit(main())