#!/usr/bin/env python3 """Independent verifier for the experimental bound-gate fixture. It does NOT import generate.py. It re-implements strict CBOR decode/encode and SHA-256 against the written wire profiles, and uses `cryptography` only for the Ed25519 signature check. It reports each check SEPARATELY: byte_conformance | signature | input_binding | material_availability definition_binding | invariant_evaluation | measurement_binding test_authority | gate_result A test-domain warrant (EXECUTE) requires ALL eight checks to PASS. A passing signature alone can never produce a warrant. Absence of material is reported as UNAVAILABLE, never PASS. The historical dimension is reported as DECISION_REPLAY_UNAVAILABLE and is never counted toward a warrant. Exit 0 iff the valid vector yields a test-domain warrant AND every negative vector fails exactly its expected check(s) and yields HOLD. """ from __future__ import annotations import hashlib import json import sys import unicodedata from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey HERE = Path(__file__).resolve().parent REQUEST_DOMAIN = b"crossing-graph:request:v1\0" CONTEXT_DOMAIN = b"crossing-graph:context:v1\0" INPUTS_DOMAIN = b"gate-inputs:v1\0" VERDICT_DOMAIN = b"crossing-graph:gate-verdict:v1\0" CHECKS = ["byte_conformance", "signature", "input_binding", "material_availability", "definition_binding", "invariant_evaluation", "measurement_binding", "test_authority"] # Registered experimental test predicates. This registry is maintained by the # verifier independently of the generator's published invariant_definitions. REGISTERED_DEFINITIONS = [ {"bit": 0, "id": "BG-I1", "name": "ClassRegistered", "predicate": "request.class_id in policy.registered_classes", "inputs": ["request.class_id", "policy.registered_classes"]}, {"bit": 1, "id": "BG-I2", "name": "EvidenceSufficient", "predicate": "request.evidence_tier >= policy.min_evidence_tier", "inputs": ["request.evidence_tier", "policy.min_evidence_tier"]}, {"bit": 2, "id": "BG-I3", "name": "AuthorityEndorsesKey", "predicate": "authority_state.status=='active' and authority_state.authority_id==policy.authority_id and verdict_signing_key in authority_state.authorized_gate_keys", "inputs": ["authority_state.status", "authority_state.authority_id", "authority_state.authorized_gate_keys", "verdict.signing_public_key", "policy.authority_id"]}, ] class ConfError(Exception): pass # --------------------------------------------------------------------------- # # Independent strict CBOR reader / canonical writer # # --------------------------------------------------------------------------- # class Cursor: def __init__(self, data: bytes): self.d, self.pos = data, 0 def _byte(self) -> int: if self.pos >= len(self.d): raise ConfError("UnexpectedEof") b = self.d[self.pos] self.pos += 1 return b def uint(self, major: int) -> int: first = self._byte() actual, ai = first >> 5, first & 31 if actual != major: raise ConfError("WrongMajorType") if ai == 31: raise ConfError("IndefiniteLength") if ai < 24: return ai if ai > 27: raise ConfError("UnsupportedAdditionalInfo") width = 1 << (ai - 24) val = 0 for _ in range(width): val = (val << 8) | self._byte() if val < {1: 24, 2: 256, 4: 65536, 8: 2**32}[width]: raise ConfError("NonMinimalInteger") return val def blob(self) -> bytes: n = self.uint(2) if self.pos + n > len(self.d): raise ConfError("UnexpectedEof") raw = self.d[self.pos:self.pos + n] self.pos += n return raw def text(self, lo: int = 1, hi: int = 256) -> str: n = self.uint(3) if self.pos + n > len(self.d): raise ConfError("UnexpectedEof") raw = self.d[self.pos:self.pos + n] self.pos += n try: s = raw.decode("utf-8") except UnicodeDecodeError: raise ConfError("InvalidUtf8") if not lo <= len(s) <= hi: raise ConfError("TextLength") if unicodedata.normalize("NFC", s) != s: raise ConfError("NonNfcText") return s def end(self): if self.pos != len(self.d): raise ConfError("TrailingBytes") def _hd(major: int, value: int) -> bytes: prefix = major << 5 if value < 24: return bytes([prefix | value]) if value < 256: return bytes([prefix | 24, value]) if value < 65536: return bytes([prefix | 25]) + value.to_bytes(2, "big") if value < 2**32: return bytes([prefix | 26]) + value.to_bytes(4, "big") return bytes([prefix | 27]) + value.to_bytes(8, "big") def _fields(cur: Cursor, count: int): if cur.uint(5) != count: raise ConfError("InvalidMapLength") previous = None for wanted in range(count): fid = cur.uint(0) if previous is not None and fid <= previous: raise ConfError("FieldOutOfOrder") previous = fid if fid != wanted: raise ConfError("UnknownField") yield wanted def decode_request(data: bytes) -> dict: cur = Cursor(data) out = {} for f in _fields(cur, 6): if f == 0: if cur.uint(0) != 1: raise ConfError("InvalidFixedField") elif f == 1: if cur.uint(0) != 1: raise ConfError("InvalidFixedField") elif f == 2: out["instance_id"] = cur.text() elif f == 3: out["class_id"] = cur.text() elif f == 4: tier = cur.uint(0) if not 1 <= tier <= 5: raise ConfError("InvalidEvidenceTier") out["evidence_tier"] = tier elif f == 5: out["payload"] = cur.blob() cur.end() return out def encode_request(r: dict) -> bytes: vals = [_hd(0, 1), _hd(0, 1), _hd(3, len(r["instance_id"].encode())) + r["instance_id"].encode(), _hd(3, len(r["class_id"].encode())) + r["class_id"].encode(), _hd(0, r["evidence_tier"]), _hd(2, len(r["payload"])) + r["payload"]] return _hd(5, 6) + b"".join(_hd(0, i) + v for i, v in enumerate(vals)) def decode_context(data: bytes) -> dict: if len(data) > 1203: raise ConfError("ContextTooLong") cur = Cursor(data) out = {} for f in _fields(cur, 10): if f == 0: if cur.uint(0) != 1: raise ConfError("InvalidFixedField") elif f == 1: if cur.uint(0) != 5: raise ConfError("InvalidFixedField") elif f in (2, 3, 7, 8): raw = cur.blob() if len(raw) != 32: raise ConfError("InvalidHashLength") out[{2: "request_hash", 3: "policy_hash", 7: "graph_head_hash", 8: "authority_state_hash"}[f]] = raw elif f == 4: out["evaluated_at_ms"] = cur.uint(0) elif f == 5: if cur.pos >= len(cur.d): raise ConfError("UnexpectedEof") if cur.d[cur.pos] == 0xF6: cur.pos += 1 out["freshness_valid_until_ms"] = None else: out["freshness_valid_until_ms"] = cur.uint(0) elif f == 6: out["graph_head_sequence"] = cur.uint(0) elif f == 9: out["gate_profile_version"] = cur.text() cur.end() if not 1 <= out["evaluated_at_ms"] <= 2**63 - 1: raise ConfError("InvalidContextTimestamp") if out["graph_head_sequence"] != 0 and out["graph_head_hash"] == bytes(32): raise ConfError("InvalidGraphHeadHash") return out def encode_context(c: dict) -> bytes: fresh = b"\xf6" if c["freshness_valid_until_ms"] is None else _hd(0, c["freshness_valid_until_ms"]) gpv = c["gate_profile_version"].encode() vals = [_hd(0, 1), _hd(0, 5), _hd(2, 32) + c["request_hash"], _hd(2, 32) + c["policy_hash"], _hd(0, c["evaluated_at_ms"]), fresh, _hd(0, c["graph_head_sequence"]), _hd(2, 32) + c["graph_head_hash"], _hd(2, 32) + c["authority_state_hash"], _hd(3, len(gpv)) + gpv] return _hd(5, 10) + b"".join(_hd(0, i) + v for i, v in enumerate(vals)) def decode_verdict(data: bytes) -> dict: cur = Cursor(data) out = {} outcome = denial = None for f in _fields(cur, 10): if f == 0: if cur.uint(0) != 1: raise ConfError("InvalidFixedField") elif f == 1: if cur.uint(0) != 4: raise ConfError("InvalidFixedField") elif f == 2: g = cur.text() if g not in ("adm", "via", "trans"): raise ConfError("InvalidGateId") out["gate_id"] = g elif f == 3: outcome = cur.uint(0) elif f == 4: if cur.pos >= len(cur.d): raise ConfError("UnexpectedEof") if cur.d[cur.pos] == 0xF6: cur.pos += 1 denial = None else: denial = cur.uint(0) elif f == 5: raw = cur.blob() if len(raw) != 1: raise ConfError("InvalidInvariantMaskLength") out["evaluated_invariants"] = raw elif f == 6: raw = cur.blob() if len(raw) != 32: raise ConfError("InvalidPolicyHashLength") out["policy_hash"] = raw elif f == 7: raw = cur.blob() if len(raw) != 32: raise ConfError("InvalidInputsHashLength") out["inputs_hash"] = raw elif f == 8: if cur.pos >= len(cur.d): raise ConfError("UnexpectedEof") if (cur.d[cur.pos] >> 5) != 0: raise ConfError("InvalidTimestampType") out["timestamp_ms"] = cur.uint(0) elif f == 9: out["key_identifier"] = cur.text() cur.end() if outcome == 0: if denial is not None: raise ConfError("InvalidOutcomeCodePairing") elif outcome == 1: if denial is None: raise ConfError("InvalidOutcomeCodePairing") if denial > 255: raise ConfError("DenialCodeOutOfRange") if not 1 <= denial <= 11: raise ConfError("UnknownDenialCode") else: raise ConfError("InvalidOutcome") out["outcome"] = outcome out["denial_code"] = denial return out def encode_verdict(v: dict) -> bytes: denial = b"\xf6" if v["denial_code"] is None else _hd(0, v["denial_code"]) gid = v["gate_id"].encode() kid = v["key_identifier"].encode() vals = [_hd(0, 1), _hd(0, 4), _hd(3, len(gid)) + gid, _hd(0, v["outcome"]), denial, _hd(2, 1) + v["evaluated_invariants"], _hd(2, 32) + v["policy_hash"], _hd(2, 32) + v["inputs_hash"], _hd(0, v["timestamp_ms"]), _hd(3, len(kid)) + kid] return _hd(5, 10) + b"".join(_hd(0, i) + v2 for i, v2 in enumerate(vals)) def sha256(*parts: bytes) -> bytes: h = hashlib.sha256() for p in parts: h.update(p) return h.digest() # --------------------------------------------------------------------------- # # The eight checks # # --------------------------------------------------------------------------- # def measure_invariants(class_id, evidence_tier, policy, authority_state, signing_key_hex): i1 = class_id in policy["registered_classes"] i2 = evidence_tier >= policy["min_evidence_tier"] i3 = (authority_state["status"] == "active" and authority_state["authority_id"] == policy["authority_id"] and signing_key_hex in authority_state["authorized_gate_keys"]) mask = (i1 << 0) | (i2 << 1) | (i3 << 2) return {"BG-I1": i1, "BG-I2": i2, "BG-I3": i3}, mask def evaluate(vector: dict, trust_root: dict, definitions: list) -> dict: """Fail closed with a complete typed check map even for malformed input.""" try: res = _evaluate(vector, trust_root, definitions) except Exception as exc: res = {"byte_conformance": ("FAIL", f"INVALID fixture input: {type(exc).__name__}: {exc}")} for check in CHECKS: res.setdefault(check, ("UNAVAILABLE", "skipped: prior input is invalid or unavailable")) return res def _evaluate(vector: dict, trust_root: dict, definitions: list) -> dict: """Return {check: (status, detail)} plus derived material. status in PASS/FAIL/UNAVAILABLE/UNMEASURABLE.""" res = {} rq, cx, vd = vector["request"], vector["context"], vector["verdict"] # ---- byte_conformance ---- try: req = decode_request(bytes.fromhex(rq["canonical_cbor_hex"])) ctx = decode_context(bytes.fromhex(cx["canonical_cbor_hex"])) ver = decode_verdict(bytes.fromhex(vd["canonical_cbor_hex"])) req_bytes = bytes.fromhex(rq["canonical_cbor_hex"]) ctx_bytes = bytes.fromhex(cx["canonical_cbor_hex"]) ver_bytes = bytes.fromhex(vd["canonical_cbor_hex"]) rr = encode_request(req) == req_bytes rc = encode_context(ctx) == ctx_bytes rv = encode_verdict(ver) == ver_bytes request_hash = sha256(REQUEST_DOMAIN, req_bytes) context_hash = sha256(CONTEXT_DOMAIN, ctx_bytes) verdict_hash = hashlib.sha256(VERDICT_DOMAIN + ver_bytes).digest() hashes_ok = (request_hash.hex() == rq["request_hash_hex"] and context_hash.hex() == cx["context_hash_hex"] and verdict_hash.hex() == vd["gate_verdict_hash_hex"]) published_ok = ( rq["instance_id"] == req["instance_id"] and rq["class_id"] == req["class_id"] and rq["evidence_tier"] == req["evidence_tier"] and rq["payload_hex"] == req["payload"].hex() and cx["policy_hash_hex"] == ctx["policy_hash"].hex() and cx["authority_state_hash_hex"] == ctx["authority_state_hash"].hex() and cx["evaluated_at_ms"] == ctx["evaluated_at_ms"] and cx["freshness_valid_until_ms"] == ctx["freshness_valid_until_ms"] and cx["graph_head_sequence"] == ctx["graph_head_sequence"] and cx["graph_head_hash_hex"] == ctx["graph_head_hash"].hex() and cx["gate_profile_version"] == ctx["gate_profile_version"] and vd["gate_id"] == ver["gate_id"] and vd["outcome"] == ("positive" if ver["outcome"] == 0 else "denied") and vd["denial_code"] == ver["denial_code"] and vd["evaluated_invariants_hex"] == ver["evaluated_invariants"].hex() and vd["policy_hash_hex"] == ver["policy_hash"].hex() and vd["inputs_hash_hex"] == ver["inputs_hash"].hex() and vd["timestamp_ms"] == ver["timestamp_ms"] and vd["key_identifier"] == ver["key_identifier"]) ok = rr and rc and rv and hashes_ok and published_ok res["byte_conformance"] = ("PASS" if ok else "FAIL", "strict decode, canonical re-encode, hashes and published fields match" if ok else "canonical bytes, hash or published field mismatch") except ConfError as e: res["byte_conformance"] = ("FAIL", f"nonconformant CBOR: {e}") return res # cannot proceed without decoded objects # ---- signature ---- ver_bytes = bytes.fromhex(vd["canonical_cbor_hex"]) message = VERDICT_DOMAIN + ver_bytes pub_hex = vd["signing_public_key_hex"] try: Ed25519PublicKey.from_public_bytes(bytes.fromhex(pub_hex)).verify( bytes.fromhex(vd["signature_hex"]), message) res["signature"] = ("PASS", f"Ed25519 verifies under key {pub_hex[:8]}.. over domain||verdict") except (InvalidSignature, ValueError): res["signature"] = ("FAIL", "Ed25519 signature does not verify") # ---- input_binding ---- request_hash = sha256(REQUEST_DOMAIN, bytes.fromhex(rq["canonical_cbor_hex"])) context_hash = sha256(CONTEXT_DOMAIN, bytes.fromhex(cx["canonical_cbor_hex"])) inputs_hash = sha256(INPUTS_DOMAIN, request_hash, context_hash) b_req = ctx["request_hash"] == request_hash b_in = ver["inputs_hash"] == inputs_hash b_pol = ver["policy_hash"] == ctx["policy_hash"] b_ts = ver["timestamp_ms"] == ctx["evaluated_at_ms"] ok = b_req and b_in and b_pol and b_ts detail = ("verdict commits to exactly this request+context" if ok else "; ".join(m for m, c in [("context.request_hash!=request_hash", not b_req), ("verdict.inputs_hash!=recomputed", not b_in), ("policy_hash mismatch", not b_pol), ("timestamp mismatch", not b_ts)] if c)) res["input_binding"] = ("PASS" if ok else "FAIL", detail) # ---- material availability ---- def material_bytes(kind): f = vector.get(f"{kind}_file") inline = vector.get(f"{kind}_bytes_hex") if f: # Fixture JSON cannot direct the verifier to arbitrary local files. expected = {"policy": "material/policy.json", "authority_state": "material/authority-state.json"}[kind] if f != expected: raise ConfError("UnapprovedMaterialPath") p = HERE / expected if not p.exists(): return None fb = p.read_bytes() if inline is not None and fb != bytes.fromhex(inline): return None # on-disk material disagrees with recorded bytes return fb if inline is not None: return bytes.fromhex(inline) return None pol_b = material_bytes("policy") auth_b = material_bytes("authority_state") if pol_b is None or auth_b is None: res["material_availability"] = ("UNAVAILABLE", "policy or authority-state material not retrievable") else: pol_ok = sha256(pol_b) == ver["policy_hash"] == ctx["policy_hash"] auth_ok = sha256(auth_b) == ctx["authority_state_hash"] ok = pol_ok and auth_ok res["material_availability"] = ("PASS" if ok else "FAIL", "content hashes of retrievable policy & authority-state match commitments" if ok else f"{'policy' if not pol_ok else 'authority-state'} content hash != committed hash") # ---- published definition binding ---- definitions_ok = definitions == REGISTERED_DEFINITIONS res["definition_binding"] = ( "PASS" if definitions_ok else "FAIL", "published definitions equal the verifier's registered test predicates" if definitions_ok else "published invariant definitions differ from registered predicates") # ---- invariant evaluation ---- if pol_b is None or auth_b is None: res["invariant_evaluation"] = ("UNAVAILABLE", "cannot measure invariants without material") else: try: policy = json.loads(pol_b) authority_state = json.loads(auth_b) measured, mask = measure_invariants(req["class_id"], req["evidence_tier"], policy, authority_state, pub_hex) claimed = ver["evaluated_invariants"][0] required = policy.get("required_invariants", []) all_required = all(measured.get(i, False) for i in required) mask_ok = mask == claimed required_ok = required == [d["id"] for d in REGISTERED_DEFINITIONS] # positive outcome must be supported by all required invariants; the mask # must carry every required bit. outcome_ok = (ver["outcome"] == 0) == all_required ok = mask_ok and outcome_ok and required_ok res["invariant_evaluation"] = ( "PASS" if ok else "FAIL", f"measured mask 0x{mask:02x} vs claimed 0x{claimed:02x}; " f"required {required} -> {'all pass' if all_required else 'NOT all pass'}; " f"outcome {'supported' if outcome_ok else 'UNSUPPORTED'}; " f"required set {'registered' if required_ok else 'INVALID'}") res["_invariant_measured"] = measured res["_invariant_mask"] = mask except (ValueError, KeyError) as e: res["invariant_evaluation"] = ("FAIL", f"invariant inputs unusable: {e}") # ---- published measurement binding ---- if "_invariant_measured" not in res: res["measurement_binding"] = ("UNAVAILABLE", "no independent measurement to compare") else: published = vector.get("invariant_measurement") expected = {"measured": res["_invariant_measured"], "measured_mask_hex": f"{res['_invariant_mask']:02x}"} match = published == expected res["measurement_binding"] = ( "PASS" if match else "FAIL", "published measurement agrees with recomputed results and mask" if match else "published measurement disagrees with recomputed results or mask") # ---- test authority ---- roots = {r["key_hex"]: r for r in trust_root.get("roots", [])} root = roots.get(pub_hex) endorsed_by_root = (root is not None and not root.get("illustrative", True) and root.get("key_identifier") == ver["key_identifier"]) endorsed_by_state = False if auth_b is not None: try: endorsed_by_state = pub_hex in json.loads(auth_b).get("authorized_gate_keys", []) except ValueError: endorsed_by_state = False ok = endorsed_by_root and endorsed_by_state res["test_authority"] = ( "PASS" if ok else "FAIL", (f"TEST authority: key endorsed by declared non-illustrative trust root " f"'{root['label']}' and by authority-state (production_identity_claimed=" f"{trust_root.get('production_identity_claimed')})") if ok else ("key not listed by a declared non-illustrative trust root" if not endorsed_by_root else "authority-state does not endorse the signing key")) return res def gate_result(res: dict) -> tuple: statuses = [res[c][0] for c in CHECKS] warrant = all(s == "PASS" for s in statuses) reasons = [c for c in CHECKS if res[c][0] != "PASS"] return ("EXECUTE" if warrant else "HOLD", reasons) def report_vector(name, res, verbose=True): if verbose: for c in CHECKS: status, detail = res[c] print(f" [{status:^12}] {c:<22} {detail}") outcome, reasons = gate_result(res) scope = "TEST-DOMAIN warrant" if outcome == "EXECUTE" else "no warrant" print(f" => gate_result: {outcome} ({scope})" + (f"; HOLD reasons: {', '.join(reasons)}" if reasons else "")) return outcome, reasons def main() -> int: fixture = json.loads((HERE / "bound-gate.v1.json").read_text(encoding="utf-8")) trust_root = json.loads((HERE / "trust-root.json").read_text(encoding="utf-8")) fixture_bytes = (HERE / "bound-gate.v1.json").read_bytes() print(f"bound-gate.v1.json sha256: {hashlib.sha256(fixture_bytes).hexdigest()}") print(f"decision_replay: {fixture['decision_replay']}") print(f" {fixture['decision_replay_note']}") failures = [] print("\n[valid vector] " + fixture["valid_vector"]["name"]) definitions = fixture.get("invariant_definitions") res = evaluate(fixture["valid_vector"], trust_root, definitions) outcome, reasons = report_vector(fixture["valid_vector"]["name"], res) if outcome != "EXECUTE": failures.append(f"valid vector did not warrant: {reasons}") # A passing signature must never be sufficient on its own. if res["signature"][0] == "PASS" and any(res[c][0] != "PASS" for c in CHECKS if c != "signature"): if outcome == "EXECUTE": failures.append("signature-only warrant leaked") print("\n[negative vectors]") for neg in fixture["negative_vectors"]: print(f"\n - {neg['name']}: {neg['description']}") res = evaluate(neg, trust_root, definitions) outcome, reasons = report_vector(neg["name"], res) expected = set(neg["expected_failed_checks"]) actual_failed = {c for c in CHECKS if res[c][0] != "PASS"} if outcome == "EXECUTE": failures.append(f"{neg['name']}: negative vector produced a warrant") if expected != actual_failed: failures.append(f"{neg['name']}: expected failed {expected}, got {actual_failed}") # Signature-alone guard: signature PASS must not rescue a HOLD. if res["signature"][0] == "PASS" and outcome == "EXECUTE": failures.append(f"{neg['name']}: passing signature produced a warrant") print("\n" + "=" * 70) if failures: print("VERIFICATION FAILED:") for f in failures: print(" -", f) return 1 print(f"OK: valid vector warrants (test domain); {len(fixture['negative_vectors'])} " f"negatives each held with the expected failing check.") print("Note: a valid signature is necessary but never sufficient; " "DECISION_REPLAY_UNAVAILABLE stands for the historical dimension.") return 0 if __name__ == "__main__": sys.exit(main())