#!/usr/bin/env python3 """RANKIGI chain verifier. Dependencies: Mode A (hash-chain consistency, no signatures): stdlib only. Mode B (full verification with Ed25519 signatures, RFC 3161 TSA, Rekor inclusion, and --check-seals): requires the `cryptography` package. Install with: pip install cryptography Without `cryptography` installed only LOCAL-CONSISTENT (Mode A) is achievable; the verifier will not produce a VERIFIED verdict. Post-quantum passport verification (ML-DSA-65) is available as a server-side endpoint on the Enterprise tier: GET https://rankigi.com/api/agents/{agent_id}/verify-pq Bearer auth (read scope or dashboard session). The ML-DSA-65 signature is captured for every agent at registration time regardless of tier; only the verification endpoint is enterprise-gated. This verifier (verify.py) covers the chain + closure path; PQ passport verification stays server-side because it would require pulling in pyca/cryptography for ML-DSA support, which is not yet stable across platforms as of 2026-06. Integrity: this script is served over HTTPS from rankigi.com. For maximum assurance, verify the file's SHA-256 digest against the value published in the matching release notes: https://github.com/Rankigi-Inc/rankigi-core/releases Compute locally with: shasum -a 256 verify.py # macOS / BSD sha256sum verify.py # Linux Cosign signing of verify.py (so the hash is itself signed by a known key) is on the roadmap for v2. Until then, cross-check the release-notes hash from a second source (a colleague, your git clone of the rankigi-core repo, or a copy held outside rankigi.com) before relying on a verdict from this file for high-stakes attestation. Exit codes: 0 = VERIFIED (Mode B: chain + closure + anchors all confirmed) 0 = LOCAL-CONSISTENT (Mode A: chain only; no --rekor-required passed) 1 = BROKEN (tamper or signature mismatch detected) 1 = ANCHOR_UNCONFIRMED when --rekor-required is passed and the public transparency log entry could not be confirmed 2 = ERROR (unexpected runtime failure) Modes: Mode A (events-only bundle): produces LOCAL-CONSISTENT, not VERIFIED. Events-only bundles carry no closure_signature, no TSA anchor, and no Rekor inclusion proof, so the strongest verdict reachable is local hash-chain consistency + signature validity against the supplied key. For VERIFIED, use a closure-export bundle (mode B). Mode B (closure-export bundle, rankigi_schema 1.4/1.5/1.6): produces VERIFIED only when closure_signature, TSA, and (optionally) Rekor anchor all check out. When any external anchor (TSA or Rekor) cannot be confirmed the verdict is downgraded to ANCHOR_UNCONFIRMED with exit code 1 -- callers MUST treat this as not-passing. ANCHOR_ UNCONFIRMED means the local chain and closure signature verified but the external time/anchor binding failed; it is NOT a VERIFIED result. Usage: python3 verify.py chain.json python3 verify.py chain.json --rekor python3 verify.py chain.json --check-seals The --check-seals flag, when supplied alongside a bundle that carries a top-level or per-event "seals" array, fetches each seal issuer's public trust record from https://rankigi.com/api/public/deployers/{id} (stdlib urllib only, 5s timeout) and verifies the seal's ed25519 issuer_signature against the issuer's ed25519_public_key. Output: per-seal status block plus one overall verdict line. Network failures are reported and do not abort the chain verdict. Default behavior (without --check-seals) is unchanged. The --rekor flag, when supplied alongside a closure bundle that carries a sigstore_log_index, fetches the matching entry from rekor.sigstore.dev and confirms that the hash anchored in the public transparency log matches the chain head this bundle reports. If the network is unavailable, verification falls through with a SKIPPED notice and exit code 0; local verification proceeds independently. Multi-version verifier: hash_version = 1 (legacy) Hash input is pipe-concatenated: prev_hash | chain_ts | org_id | agent_id | payload_canonical [| intent_hash] [| decision_token_hash] chain_ts is COALESCE(server_received_at, occurred_at) cast to Postgres timestamptz::text. Chains are per-agent: prev_hash continuity and chain_index monotonicity are checked per agent_id. hash_version = 2 Hash input is canonical JSON of: { "action": , "agent_id": , "chain_id": , "occurred_at": , "org_id": , "payload": , "prev_hash": , "severity": , "tool": } Keys sorted lexicographically, no whitespace. prev_hash is scoped to chain_id when present, else to the agent global chain. Continuity and monotonicity are checked per (agent_id, chain_id) when chain_id is present, else per agent_id. hash_version = 3 (current) v2 input plus canon_id and canon_version, slotted alphabetically between agent_id and chain_id: { "action": ..., "agent_id": ..., "canon_id": , "canon_version": , "chain_id": ..., ... } Same canonicalization rules as v2. Existing v2 rows continue to verify under their own branch. Per-event `hash_version` defaults to 1 when absent so legacy exports keep verifying unchanged. """ import base64 import hashlib import json import sys import unicodedata import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone try: from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PublicKey, ) from cryptography.hazmat.primitives.asymmetric.ec import ( EllipticCurvePublicKey, ECDSA, ) from cryptography.hazmat.primitives.serialization import ( load_pem_public_key, load_der_public_key, ) from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 from cryptography.exceptions import InvalidSignature from cryptography import x509 CRYPTO_AVAILABLE = True except ImportError: CRYPTO_AVAILABLE = False # Phase 2 encrypted-evidence helpers. These imports are optional in the same # way Ed25519 is optional: when cryptography is absent we still report # encrypted bundles honestly, we just cannot decrypt them. try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives.hashes import SHA256 PAYLOAD_CRYPTO_AVAILABLE = True except ImportError: PAYLOAD_CRYPTO_AVAILABLE = False # Sentinel raised when an event carries payload_canonical_encrypted but no # --payload-key was supplied. Callers catch and report rather than fail. class EncryptedPayloadUnavailable(Exception): pass PAYLOAD_HKDF_SALT = b"rankigi-payload-v1" PAYLOAD_ENVELOPE_VERSION = 1 PAYLOAD_ALG = "aes-256-gcm" # RANKIGI root public key (Ed25519, key_id = rankigi-root-2026). # This is the out-of-band trust anchor for passport attestations and closure # envelope signatures. Pinning the key in source means tampering with the # verifier itself is the only attack vector against the trust chain; that is # the model. Independent verifiers should cross-check this value against # https://rankigi.com/.well-known/rankigi-root-key.json. RANKIGI_ROOT_PUBKEY_PEM = b"""-----BEGIN PUBLIC KEY----- MCowBQYDK2VwAyEAoI8+bGiHpg5AFVf01Tbs8/I9WY+T0+U+7Ma7p1CjvrQ= -----END PUBLIC KEY----- """ RANKIGI_ROOT_KEY_ID = "rankigi-root-2026" # Sprint 10A: pin the root key version + content fingerprint so an offline # verifier can prove the embedded key matches what RANKIGI published. The # fingerprint is the SHA-256 of the PEM bytes above (newlines included). # Computed at module load to avoid drift if the constant is hand-edited. RANKIGI_PUBKEY_VERSION = "2026-06-01" RANKIGI_PUBKEY_FINGERPRINT = hashlib.sha256( RANKIGI_ROOT_PUBKEY_PEM ).hexdigest() # RANKIGI witness public key (Ed25519, key_id = rankigi-witness-2026). Used # only as a fallback signer for unattested CCAP receipts when the root key is # not configured server-side. Sourced from public/.well-known/rankigi-witness-key.pem # and src/lib/ccap/witness.ts at build time. Independent verifiers can cross # check against https://rankigi.com/.well-known/rankigi-witness-key.json. RANKIGI_WITNESS_PUBKEY_PEM = b"""-----BEGIN PUBLIC KEY----- MCowBQYDK2VwAyEAKyxZz7WkojokBZa0Tl/Kd4q75fTrqPmDoFM/Kk25h1k= -----END PUBLIC KEY----- """ # --- L1-H2: RANKIGI-internal witness-signed events ------------------------- # Server-originated events (passport_revoked, agent_archived, wallet_frozen, # federation/arbitration records, ...) are written by writeInternalChainEvent # (src/lib/ingest/internal.ts), which signs the portable-regime input with # the RANKIGI root key (key id rankigi-root-2026; legacy fallback # rankigi-witness-2026) because no agent passport key exists for server # events. Such events carry attested_by == "rankigi-witness" INSIDE the # signed payload. Verification policy: # signature present + valid against a pinned RANKIGI key -> verified # signature present + INVALID -> COMPROMISED # signature absent (witness key unconfigured at write) -> verdict capped # at SELF-ATTESTED # Count of witness-marked events that carried no signature. Read by the # verdict printer to cap VERIFIED down to SELF-ATTESTED. Module-global by # the same pattern as _PAYLOAD_ENCRYPTED_SEEN. _WITNESS_UNSIGNED_COUNT = 0 def _event_witness_marker(e): """True when the event payload declares attested_by == 'rankigi-witness'. Mirrors event_closure_binding's payload / payload_canonical fallback so hash-only-retention bundles still route correctly. Never raises.""" if not isinstance(e, dict): return False payload = e.get("payload") if isinstance(payload, dict) and payload.get("attested_by") == "rankigi-witness": return True pc = e.get("payload_canonical") if isinstance(pc, str) and '"attested_by":"rankigi-witness"' in pc: return True return False def _load_rankigi_witness_keys(): """Load the pinned RANKIGI signing keys (root preferred, legacy witness fallback) for internal-event verification. Returns [] when the cryptography package is unavailable.""" if not CRYPTO_AVAILABLE: return [] keys = [] for pem in (RANKIGI_ROOT_PUBKEY_PEM, RANKIGI_WITNESS_PUBKEY_PEM): try: keys.append(load_pem_public_key(pem)) except Exception: continue return keys def _verify_witness_signature(sig, signing_input): """Verify an internal event's witness signature against the pinned RANKIGI keys. True on the first key that verifies.""" for key in _load_rankigi_witness_keys(): try: key.verify(sig, signing_input.encode("utf-8")) return True except Exception: continue return False # --- L1-H1: coverage attestation (Step 8) ---------------------------------- def check_coverage_commitment(doc, events): """Compare the customer's pre-closure commitment against the bundle. Two independent checks, both completeness signals -- a coverage gap means events were emitted by the SDK but never reached the ledger (or vice versa). It is NOT tamper evidence and never downgrades the verdict on its own: 1. expected_events_commitment (SDK commitEvents): recompute the nonce-Merkle root over SDK-session events (nonce-bearing, excluding chain_closure and RANKIGI-internal witness events) and compare root and count against the customer-signed commitment. The Merkle algorithm mirrors sdk-node-v4 computeNonceMerkleRoot: leaves = sha256(nonce), pairwise sha256(left + right), odd levels duplicate the last node. 2. client_seq contiguity: events stamped with the SDK's monotonic client_seq must be gap-free (1, 2, 4 means event 3 was dropped). Returns True when no gap was found (or nothing to check). """ ok = True events = events if isinstance(events, list) else [] commitment = doc.get("expected_events_commitment") if isinstance(doc, dict) else None if isinstance(commitment, dict) and commitment.get("root"): expected_root = str(commitment.get("root") or "") expected_count = commitment.get("count") nonces = [] for ev in events: if not isinstance(ev, dict): continue if ev.get("action") == "chain_closure": continue if _event_witness_marker(ev): continue n = ev.get("nonce") if isinstance(n, str) and n: nonces.append(n) level = [hashlib.sha256(n.encode("utf-8")).hexdigest() for n in nonces] if not level: root = "" else: while len(level) > 1: nxt = [] for i in range(0, len(level), 2): left = level[i] right = level[i + 1] if i + 1 < len(level) else left nxt.append( hashlib.sha256((left + right).encode("utf-8")).hexdigest() ) level = nxt root = level[0] found = len(nonces) if isinstance(expected_count, int) and found != expected_count: print("COVERAGE GAP: expected %d events, found %d" % (expected_count, found)) ok = False elif root != expected_root: print("COVERAGE GAP: nonce-Merkle root mismatch over %d events" % found) print(" expected root: %s..." % expected_root[:32]) print(" recomputed: %s..." % (root[:32] if root else "(empty)")) ok = False else: print( "Coverage commitment: OK (%d events match the customer-signed root)" % found ) seqs = sorted( ev.get("client_seq") for ev in events if isinstance(ev, dict) and isinstance(ev.get("client_seq"), int) ) if seqs: gap_found = False prev_seq = None for s in seqs: if prev_seq is not None and s > prev_seq + 1: print( "COVERAGE GAP: client_seq jump %d -> %d (%d event(s) missing)" % (prev_seq, s, s - prev_seq - 1) ) gap_found = True prev_seq = s if gap_found: ok = False else: print( "client_seq contiguity: OK (%d sequenced events, no gaps)" % len(seqs) ) if not (isinstance(commitment, dict) and commitment.get("root")) and not seqs: print("Coverage commitment: NONE (no expected_events_commitment or client_seq in bundle)") return ok RANKIGI_WITNESS_KEY_ID = "rankigi-witness-2026" # FreeTSA root CA SHA-256 fingerprint pin. The authoritative source for this # fingerprint is https://freetsa.org/files/cacert.pem. The pin IS committed # below and is enforced at verification time. When the in-bundle TSA cert # chain validates back to a root whose DER SHA-256 matches this pin, the # TSA status is reported as VERIFIED. When the chain is present but the # root does not match the pin, the status falls back to PARTIAL so the # operator can see the trust gap. # # SHA-256 of the DER encoding of the FreeTSA root cert published at # https://freetsa.org/files/cacert.pem (subject C=DE, ST=Bayern, L=Wuerzburg, # O=Free TSA, OU=Root CA, CN=www.freetsa.org). FREETSA_ROOT_SHA256_FINGERPRINT = ( "a6379e7cecc05faa3cbf076013d745e327bbbaa38c0b9af22469d4701d18aabc" ) # Default base URL for the --pubkey-from-network fetch path. Override with # --base-url so the network path is testable against staging. DEFAULT_BASE_URL = "https://rankigi.com" def parse_payload_master_key(b64_str): """Decode a base64-encoded 32-byte master key. Returns bytes or raises.""" if not b64_str: return None raw = base64.b64decode(b64_str) if len(raw) != 32: raise ValueError("--payload-key must decode to 32 bytes (got %d)" % len(raw)) return raw def derive_org_payload_key(master, org_id): """HKDF-SHA256: salt='rankigi-payload-v1', info=org_id utf8, length=32. Mirrors src/lib/payload/encrypt.ts deriveOrgPayloadKey. The salt and length are fixed for v1 envelopes; a future v2 envelope would change both. """ if not PAYLOAD_CRYPTO_AVAILABLE: raise RuntimeError("cryptography library required for payload decryption") hkdf = HKDF( algorithm=SHA256(), length=32, salt=PAYLOAD_HKDF_SALT, info=str(org_id).encode("utf-8"), ) return hkdf.derive(master) def decrypt_payload_envelope(envelope_b64, master, org_id): """Reverse src/lib/payload/encrypt.ts encryptPayloadCanonical. envelope_b64: base64 of canonical JSON {v, alg, iv, tag, ct}. Returns the decrypted canonical event record string. Raises on any schema / version / algorithm / tag mismatch. Tag mismatch is the only signal of envelope tampering; the caller must not swallow it. """ if not PAYLOAD_CRYPTO_AVAILABLE: raise RuntimeError("cryptography library required for payload decryption") try: env = json.loads(base64.b64decode(envelope_b64).decode("utf-8")) except Exception as e: raise ValueError("envelope is not valid base64 JSON: %s" % e) if env.get("v") != PAYLOAD_ENVELOPE_VERSION: raise ValueError("unsupported envelope version %r" % env.get("v")) if env.get("alg") != PAYLOAD_ALG: raise ValueError("unsupported envelope alg %r" % env.get("alg")) iv = base64.b64decode(env["iv"]) tag = base64.b64decode(env["tag"]) ct = base64.b64decode(env["ct"]) if len(iv) != 12: raise ValueError("bad IV length %d" % len(iv)) if len(tag) != 16: raise ValueError("bad tag length %d" % len(tag)) key = derive_org_payload_key(master, org_id) aes = AESGCM(key) # AESGCM in cryptography expects ciphertext || tag as a single buffer. return aes.decrypt(iv, ct + tag, associated_data=None).decode("utf-8") # Module-level switch: set by main() when --payload-key is parsed. Read by # resolve_canonical_payload via the per-call _payload_master_key argument so # unit tests can pass an override without poking globals. _PAYLOAD_MASTER_KEY = None _PAYLOAD_ENCRYPTED_SEEN = 0 _PAYLOAD_ENCRYPTED_DECRYPTED = 0 _PAYLOAD_ENCRYPTED_SKIPPED = 0 # ── v1 helpers ──────────────────────────────────────────────────────────── def pg_timestamptz_text(iso): """Convert an ISO 8601 timestamp string to Postgres timestamptz::text.""" if not iso: return "" s = str(iso).strip() if s.endswith("Z"): s = s[:-1] + "+00:00" dt = datetime.fromisoformat(s).astimezone(timezone.utc) base = dt.strftime("%Y-%m-%d %H:%M:%S") micro = dt.microsecond if micro == 0: return base + "+00" digits = ("%06d" % micro).rstrip("0") return base + "." + digits + "+00" if digits else base + "+00" def event_hash_v1(e): chain_ts_iso = e.get("server_received_at") or e.get("occurred_at") or "" chain_ts_text = pg_timestamptz_text(chain_ts_iso) parts = [ e.get("prev_hash", ""), chain_ts_text, str(e.get("org_id", "")), str(e.get("agent_id", "")), e.get("payload_canonical", ""), ] if e.get("intent_hash"): parts.append(e["intent_hash"]) dt = e.get("decision_token") if dt: parts.append(hashlib.sha256(dt.encode()).hexdigest()) elif e.get("decision_token_hash"): parts.append(e["decision_token_hash"]) return hashlib.sha256("|".join(parts).encode()).hexdigest() # ── v2 helpers ──────────────────────────────────────────────────────────── def normalize_occurred_at(s): """Reproduce the v2 SQL to_char shape: YYYY-MM-DDTHH:MM:SS.mmmZ (UTC, ms). Matches JS Date.prototype.toISOString and the SQL RPC's to_char format, so the canonical bytes are identical regardless of input timezone or fractional precision. Python 3.10's datetime.fromisoformat rejects fractional seconds that are not 3 or 6 digits, so we pad first. """ if not s: raise ValueError("normalize_occurred_at: empty timestamp") text = s.replace("Z", "+00:00") # Pad the fractional-seconds field to 6 digits so fromisoformat accepts # it on Python 3.10. The fractional region is whatever sits between the # last "." and the next "+" or "-" (timezone), if any. if "." in text: dot = text.rfind(".") tz_idx = max(text.rfind("+"), text.rfind("-")) if tz_idx > dot: frac = text[dot + 1: tz_idx] rest = text[tz_idx:] else: frac = text[dot + 1:] rest = "" # Keep only digits; pad to 6 if shorter, truncate if longer. frac_digits = "".join(ch for ch in frac if ch.isdigit()) if len(frac_digits) > 6: frac_digits = frac_digits[:6] else: frac_digits = frac_digits.ljust(6, "0") text = text[:dot + 1] + frac_digits + rest dt = datetime.fromisoformat(text) dt = dt.astimezone(timezone.utc) ms = dt.microsecond // 1000 return dt.strftime("%Y-%m-%dT%H:%M:%S.") + ("%03d" % ms) + "Z" class _RawCanonical: """Wrapper that carries an already-canonical JSON fragment verbatim. When _canonical() encounters one of these it emits the wrapped string as is, without re-serializing it. This is how the verifier splices a stored payload_canonical string into the canonical event object at the "payload" slot, reproducing the exact preimage the ingest RPC hashed (the RPC writes '"payload":' || payload_canonical into the hash input verbatim). Re- canonicalizing the raw payload would JSON-escape the string and break the match, so the verbatim splice is required. """ __slots__ = ("text",) def __init__(self, text): self.text = text def resolve_canonical_payload(e): """Resolve the canonical payload fragment for an event. Precedence: 1. e["payload_canonical_encrypted"] when present (Phase 2). Decrypted with the --payload-key master if supplied, else raises EncryptedPayloadUnavailable so the caller can mark the event as unverifiable rather than silently falsifying the hash. 2. e["payload_canonical"] when present (a STRING already in canonical form). Used verbatim - this is the exact preimage the ingest RPC hashed, and it survives even after the raw payload is nulled in the database. NOTE: on Phase 2 encrypted rows the column stores the sha256_hex of the canonical, not the canonical itself, so it must NOT be spliced into the hash preimage. The encrypted branch above takes precedence to prevent that mistake. 3. Fall back to canonicalizing e["payload"] for older bundles that were exported before payload_canonical was carried. Returns a _RawCanonical wrapper so _canonical() splices the fragment in without re-escaping it. """ global _PAYLOAD_ENCRYPTED_SEEN, _PAYLOAD_ENCRYPTED_DECRYPTED, _PAYLOAD_ENCRYPTED_SKIPPED enc = e.get("payload_canonical_encrypted") if isinstance(enc, str) and enc != "": _PAYLOAD_ENCRYPTED_SEEN += 1 if _PAYLOAD_MASTER_KEY is None: _PAYLOAD_ENCRYPTED_SKIPPED += 1 raise EncryptedPayloadUnavailable( "event carries payload_canonical_encrypted but no --payload-key was provided" ) org_id = e.get("org_id") or "" decrypted = decrypt_payload_envelope(enc, _PAYLOAD_MASTER_KEY, org_id) _PAYLOAD_ENCRYPTED_DECRYPTED += 1 return _RawCanonical(decrypted) pc = e.get("payload_canonical") if isinstance(pc, str) and pc != "": return _RawCanonical(pc) return _RawCanonical(_canonical(e.get("payload"))) def event_closure_binding(e): """Read the closure_binding label for an event without crashing on a nulled payload. Precedence: 1. The structured payload dict, when present. 2. The stored payload_canonical string, parsed back to a dict, for bundles whose raw payload was nulled at rest but whose canonical preimage was retained for hash recomputation. Returns the closure_binding string, or None when it cannot be resolved. Never raises. """ if not isinstance(e, dict): return None payload = e.get("payload") if isinstance(payload, dict): cb = payload.get("closure_binding") if isinstance(cb, str): return cb # Fall through: under hash-only retention the payload dict is empty # but the canonical preimage carries the real closure_binding value. pc = e.get("payload_canonical") if isinstance(pc, str) and pc != "": try: parsed = json.loads(pc) except (ValueError, TypeError): return None if isinstance(parsed, dict): cb = parsed.get("closure_binding") return cb if isinstance(cb, str) else None return None def _canonical(value): """Canonical JSON, matching src/lib/crypto/canonical-json.ts rules. Object keys sorted lex; no whitespace; RFC 8259 string escaping; -0 -> 0; integers without decimal point; no scientific notation; null for missing-undefined positions (the producer should not emit undefined). """ if isinstance(value, _RawCanonical): return value.text if value is None: return "null" if value is True: return "true" if value is False: return "false" if isinstance(value, str): return _escape_string(value) if isinstance(value, int) and not isinstance(value, bool): return str(value) if isinstance(value, float): if value != value or value in (float("inf"), float("-inf")): raise ValueError("non-finite number not canonicalizable") if value == 0: return "0" if value.is_integer(): return str(int(value)) s = repr(value) if "e" in s or "E" in s: s = ("%.20f" % value).rstrip("0").rstrip(".") return s if isinstance(value, list): return "[" + ",".join(_canonical(v) for v in value) + "]" if isinstance(value, dict): keys = sorted(k for k, v in value.items() if v is not None or _explicit_null(value, k)) # Drop undefined-equivalent (Python has no undefined, so include all) keys = sorted(value.keys()) parts = [] for k in keys: parts.append(_escape_string(k) + ":" + _canonical(value[k])) return "{" + ",".join(parts) + "}" raise TypeError("cannot canonicalize type %s" % type(value).__name__) def _explicit_null(_d, _k): return True def _escape_string(s): # Stage 2 (2A): NFC normalization, matching the server's canonicalJson # (src/lib/crypto/canonical-json.ts escapeString). Visually identical # strings in different Unicode composition forms must hash identically, # or honest chains false-flag as tampered. s = unicodedata.normalize("NFC", s) out = ['"'] for ch in s: cp = ord(ch) if ch == '"': out.append("\\\"") elif ch == "\\": out.append("\\\\") elif cp == 0x08: out.append("\\b") elif cp == 0x0c: out.append("\\f") elif cp == 0x0a: out.append("\\n") elif cp == 0x0d: out.append("\\r") elif cp == 0x09: out.append("\\t") elif cp < 0x20: out.append("\\u%04x" % cp) else: out.append(ch) out.append('"') return "".join(out) def event_hash_v2(e): obj = { "action": e.get("action"), "agent_id": e.get("agent_id"), "chain_id": e.get("chain_id"), "occurred_at": normalize_occurred_at(e.get("occurred_at")), "org_id": e.get("org_id"), "payload": resolve_canonical_payload(e), "prev_hash": e.get("prev_hash"), "severity": e.get("severity"), "tool": e.get("tool"), } return hashlib.sha256(_canonical(obj).encode("utf-8")).hexdigest() # - v3 helpers ------------------------------------------------------------ def event_hash_v3(e): """v3 canonical input adds canon_id and canon_version in alphabetical key order (between agent_id and chain_id). """ obj = { "action": e.get("action"), "agent_id": e.get("agent_id"), "canon_id": e.get("canon_id"), "canon_version": e.get("canon_version"), "chain_id": e.get("chain_id"), "occurred_at": normalize_occurred_at(e.get("occurred_at")), "org_id": e.get("org_id"), "payload": resolve_canonical_payload(e), "prev_hash": e.get("prev_hash"), "severity": e.get("severity"), "tool": e.get("tool"), } return hashlib.sha256(_canonical(obj).encode("utf-8")).hexdigest() # - v4 helpers ------------------------------------------------------------ def event_hash_v4(e): """v4 canonical input adds passport_fingerprint and signature_hash in alphabetical key order. passport_fingerprint slots between org_id and payload; signature_hash slots between severity and tool. Both are emitted as JSON null when missing so the canonical shape is stable across signed and unsigned events. """ obj = { "action": e.get("action"), "agent_id": e.get("agent_id"), "canon_id": e.get("canon_id"), "canon_version": e.get("canon_version"), "chain_id": e.get("chain_id"), "occurred_at": normalize_occurred_at(e.get("occurred_at")), "org_id": e.get("org_id"), "passport_fingerprint": e.get("passport_fingerprint"), "payload": resolve_canonical_payload(e), "prev_hash": e.get("prev_hash"), "severity": e.get("severity"), "signature_hash": e.get("signature_hash"), "tool": e.get("tool"), } return hashlib.sha256(_canonical(obj).encode("utf-8")).hexdigest() def compute_event_hash_v5_canonical(e): """v5 (Phase B): the v4 canonical input plus proof_extensions_hash as the 14th preimage key, sorted alphabetically between prev_hash and severity. Byte-identical to ingest_event_with_hash_v5's v_canonical (migration 20260614000002). proof_extensions_hash is emitted as JSON null when the row carries none -- the key is ALWAYS present, which is what makes a v5 row hash differently from the same v4 row. """ obj = { "action": e.get("action"), "agent_id": e.get("agent_id"), "canon_id": e.get("canon_id"), "canon_version": e.get("canon_version"), "chain_id": e.get("chain_id"), "occurred_at": normalize_occurred_at(e.get("occurred_at")), "org_id": e.get("org_id"), "passport_fingerprint": e.get("passport_fingerprint"), "payload": resolve_canonical_payload(e), "prev_hash": e.get("prev_hash"), "proof_extensions_hash": e.get("proof_extensions_hash"), "severity": e.get("severity"), "signature_hash": e.get("signature_hash"), "tool": e.get("tool"), } return hashlib.sha256(_canonical(obj).encode("utf-8")).hexdigest() def event_hash(e): """Dispatch to the version-specific hash recomputation. hash_version is read per-event. v4 events use v4 recomputation; v2/v3 use their respective recomputations; v1 explicit values use the legacy pipe-concatenated form. When hash_version is missing or null on an event, we default to v3 (the current canonical-JSON form). Legacy v1 exports must set hash_version explicitly to 1. Stage 4 (F1): UNKNOWN versions return None. The old behavior fell through to the v1 pipe format, guaranteeing a mismatch and branding valid future-version rows as COMPROMISED. None means "this verifier cannot recompute this row" -- callers MUST report such rows as UNVERIFIABLE, never COMPROMISED, and keep processing other events. """ raw_version = e.get("hash_version") version = int(raw_version) if raw_version is not None else 3 if version == 5: return compute_event_hash_v5_canonical(e) if version == 4: return event_hash_v4(e) if version == 3: return event_hash_v3(e) if version == 2: return event_hash_v2(e) if version == 1: return event_hash_v1(e) # Unknown version -- cannot verify, must not claim COMPROMISED. return None # === AGENT-A-BLOCK: TSA VERIFICATION === # RFC 3161 TSA per-batch verification. Surfaces tsa_tokens[] from the # closure bundle and, when tsr_token_base64 and tsa_cert_chain_pem are # present, verifies the TimeStampToken's messageImprint and signing # certificate chain offline. Honest behaviour rule: never print VERIFIED # without performing the check; PARTIAL is used when only some pillars # can be confirmed. # # ASN.1 decoder choice: we use a small stdlib-only DER walker for the TSR # (extracting messageImprint and the embedded TSTInfo). Certificate parsing # and signature verification use the `cryptography` library because rolling # DER X.509 parsing in stdlib is error prone. cryptography is already a # hard requirement for Ed25519 chain verification, so no new dep. def _der_read_tag_len(data, off): """Read a DER tag and length starting at offset off. Returns (tag_class, constructed, tag_number, content_off, content_len, next_off). Only handles single-byte tag numbers (tag < 31), which covers every tag emitted by RFC 3161 TSR and X.509 structures we need to walk. Raises ValueError on malformed input. """ if off >= len(data): raise ValueError("DER read past end") first = data[off] tag_class = (first >> 6) & 0x03 constructed = bool((first >> 5) & 0x01) tag_number = first & 0x1F if tag_number == 0x1F: raise ValueError("multi-byte DER tags not supported") off += 1 if off >= len(data): raise ValueError("DER length past end") length = data[off] off += 1 if length & 0x80: n_bytes = length & 0x7F if n_bytes == 0: raise ValueError("DER indefinite length not supported") if off + n_bytes > len(data): raise ValueError("DER long length past end") length = 0 for i in range(n_bytes): length = (length << 8) | data[off + i] off += n_bytes content_off = off return (tag_class, constructed, tag_number, content_off, length, content_off + length) def _der_iter_sequence(data, off): """Iterate top-level children of a DER SEQUENCE / SET at offset off. Yields tuples (tag_class, constructed, tag_number, content_off, content_len, child_total_off). The sequence header MUST start at off. """ tag_class, constructed, tag_number, c_off, c_len, _ = _der_read_tag_len( data, off ) if tag_number != 0x10 and tag_number != 0x11: raise ValueError("expected DER SEQUENCE/SET at offset %d" % off) end = c_off + c_len p = c_off while p < end: child = _der_read_tag_len(data, p) yield child p = child[5] def _extract_tst_info_from_tsr(tsr_bytes): """Walk a DER-encoded TimeStampResp and pull the embedded TSTInfo. Returns (tst_info_bytes, signer_cert_der_or_none, signed_attrs_off, signed_attrs_len, raw_signature, signature_alg_oid_bytes_or_none). Structure (simplified, RFC 3161 + CMS): TimeStampResp ::= SEQUENCE { status PKIStatusInfo, timeStampToken ContentInfo OPTIONAL } ContentInfo ::= SEQUENCE { contentType OID, content [0] EXPLICIT ANY } SignedData ::= SEQUENCE { version, digestAlgorithms, encapContentInfo, certificates [0] IMPLICIT OPTIONAL, crls [1] IMPLICIT OPTIONAL, signerInfos SET OF SignerInfo } encapContentInfo ::= SEQUENCE { eContentType OID, eContent [0] EXPLICIT OCTET STRING } -- eContent OCTET STRING wraps the DER TSTInfo bytes. We do not enforce strict tag checks on every layer because real-world FreeTSA tokens are stable; we extract just enough to (a) recompute messageImprint and (b) hand cert and signature bytes to `cryptography` for cryptographic verification. Returns Nones on any structural surprise rather than raising. """ try: # TimeStampResp SEQUENCE children = list(_der_iter_sequence(tsr_bytes, 0)) if len(children) < 2: return (None, None, None, None, None, None) # children[0] = PKIStatusInfo, children[1] = timeStampToken (ContentInfo) _, _, _, ci_off, ci_len, _ = children[1] # ContentInfo SEQUENCE ci_children = list(_der_iter_sequence(tsr_bytes, ci_off - _der_header_len(tsr_bytes, ci_off, ci_len))) except Exception: return (None, None, None, None, None, None) # Simpler approach: locate the SignedData by scanning. The ContentInfo # header sits immediately before ci_off; rewind to find its own SEQUENCE # start. We can also just parse ContentInfo as its own SEQUENCE since # children[1] gave us the inner content_off/len; reparse from the # SEQUENCE position of timeStampToken. try: tst_seq_off = children[1][3] - 0 # children[1] tuple is (tc, c, tn, c_off, c_len, next). We need the # header start, which is c_off minus the header length. Recompute by # binary searching backward is messy; instead, we re-walk from the # raw bytes: the timeStampToken is a SEQUENCE whose header started at # some offset <= ci_off. Easiest: build ContentInfo SEQUENCE by # wrapping it. We will re-derive by reading at children[1] header # start, which we never stored. Workaround: walk children of # TimeStampResp again but remember header starts. pass except Exception: return (None, None, None, None, None, None) # Re-walk top-level recording header starts. try: tag_class, constructed, tag_number, c_off, c_len, _ = _der_read_tag_len( tsr_bytes, 0 ) if tag_number != 0x10: return (None, None, None, None, None, None) end = c_off + c_len p = c_off # PKIStatusInfo _, _, _, _, _, p_next = _der_read_tag_len(tsr_bytes, p) p = p_next if p >= end: return (None, None, None, None, None, None) # timeStampToken ContentInfo (SEQUENCE) ts_hdr_off = p tc, _, tn, ts_c_off, ts_c_len, _ = _der_read_tag_len(tsr_bytes, p) if tn != 0x10: return (None, None, None, None, None, None) # ContentInfo children: contentType OID + [0] EXPLICIT content cp = ts_c_off ts_end = ts_c_off + ts_c_len # contentType OID _, _, oid_tn, _, _, cp_next = _der_read_tag_len(tsr_bytes, cp) cp = cp_next if cp >= ts_end: return (None, None, None, None, None, None) # [0] EXPLICIT content _, _, tn0, sd_outer_off, sd_outer_len, _ = _der_read_tag_len(tsr_bytes, cp) # Inside [0] EXPLICIT is the SignedData SEQUENCE _, _, sd_tn, sd_c_off, sd_c_len, _ = _der_read_tag_len( tsr_bytes, sd_outer_off ) if sd_tn != 0x10: return (None, None, None, None, None, None) # Walk SignedData children sp = sd_c_off sd_end = sd_c_off + sd_c_len # version INTEGER _, _, _, _, _, sp_next = _der_read_tag_len(tsr_bytes, sp) sp = sp_next # digestAlgorithms SET _, _, _, _, _, sp_next = _der_read_tag_len(tsr_bytes, sp) sp = sp_next # encapContentInfo SEQUENCE _, _, _, eci_c_off, eci_c_len, sp_next = _der_read_tag_len( tsr_bytes, sp ) sp = sp_next # encapContentInfo: eContentType OID, eContent [0] EXPLICIT OCTET STRING ep = eci_c_off eci_end = eci_c_off + eci_c_len _, _, _, _, _, ep_next = _der_read_tag_len(tsr_bytes, ep) ep = ep_next if ep >= eci_end: return (None, None, None, None, None, None) # [0] EXPLICIT _, _, ec_tn, ec_outer_off, ec_outer_len, _ = _der_read_tag_len( tsr_bytes, ep ) # Inside is OCTET STRING containing DER TSTInfo _, _, os_tn, os_c_off, os_c_len, _ = _der_read_tag_len( tsr_bytes, ec_outer_off ) if os_tn != 0x04: return (None, None, None, None, None, None) tst_info_bytes = tsr_bytes[os_c_off: os_c_off + os_c_len] # certificates [0] IMPLICIT (optional). The signer cert is the first # certificate in this SET-equivalent. cryptography can parse a single # X.509 cert from DER bytes. signer_cert_der = None # Optional fields after encapContentInfo: certificates [0], # crls [1], then signerInfos SET. Loop until we find SET (signerInfos). signer_info_off = None cur = sp while cur < sd_end: tc2, _, tn2, c_off2, c_len2, nxt2 = _der_read_tag_len( tsr_bytes, cur ) if tc2 == 2 and tn2 == 0: # certificates [0] IMPLICIT SET OF Certificate. Take first cert. inner_off = c_off2 _, _, _, cert_c_off, cert_c_len, _ = _der_read_tag_len( tsr_bytes, inner_off ) # cert_total = header + content cert_total_start = inner_off cert_total_end = cert_c_off + cert_c_len signer_cert_der = tsr_bytes[cert_total_start: cert_total_end] elif tc2 == 0 and tn2 == 0x11: # signerInfos SET signer_info_off = cur break cur = nxt2 # If we didn't find signerInfos via SET (tag 0x11), bail out. if signer_info_off is None: return (tst_info_bytes, signer_cert_der, None, None, None, None) # SignerInfo: walk to extract signedAttrs [0] IMPLICIT and signature # OCTET STRING. SignerInfo SEQUENCE inside the SET. _, _, _, si_set_off, si_set_len, _ = _der_read_tag_len( tsr_bytes, signer_info_off ) _, _, _, si_off, si_len, _ = _der_read_tag_len(tsr_bytes, si_set_off) si_end = si_off + si_len sip = si_off # version _, _, _, _, _, sip = _der_read_tag_len(tsr_bytes, sip) # sid _, _, _, _, _, sip = _der_read_tag_len(tsr_bytes, sip) # digestAlgorithm _, _, _, _, _, sip = _der_read_tag_len(tsr_bytes, sip) # signedAttrs [0] IMPLICIT (optional) signed_attrs_off = None signed_attrs_len = None tc3, _, tn3, c3_off, c3_len, nxt3 = _der_read_tag_len(tsr_bytes, sip) if tc3 == 2 and tn3 == 0: signed_attrs_off = sip signed_attrs_len = nxt3 - sip sip = nxt3 # signatureAlgorithm sig_alg_start = sip _, _, _, _, _, sip = _der_read_tag_len(tsr_bytes, sip) # signature OCTET STRING _, _, sig_tn, sig_c_off, sig_c_len, _ = _der_read_tag_len( tsr_bytes, sip ) if sig_tn != 0x04: return (tst_info_bytes, signer_cert_der, signed_attrs_off, signed_attrs_len, None, None) raw_signature = tsr_bytes[sig_c_off: sig_c_off + sig_c_len] return (tst_info_bytes, signer_cert_der, signed_attrs_off, signed_attrs_len, raw_signature, None) except Exception: return (None, None, None, None, None, None) def _der_header_len(data, c_off, c_len): # Returns the byte count of the DER header that precedes c_off given # that the content length is c_len. Unused safety helper kept for the # broken early branch above; harmless. if c_len < 0x80: return 2 n = 1 v = c_len while v: v >>= 8 n += 1 return 1 + n def _extract_tst_message_imprint(tst_info_bytes): """From DER TSTInfo, extract messageImprint.hashedMessage bytes. TSTInfo ::= SEQUENCE { version INTEGER, policy OBJECT IDENTIFIER, messageImprint MessageImprint, ... } MessageImprint ::= SEQUENCE { hashAlgorithm AlgorithmIdentifier, hashedMessage OCTET STRING } """ try: tc, _, tn, c_off, c_len, _ = _der_read_tag_len(tst_info_bytes, 0) if tn != 0x10: return None p = c_off # version INTEGER _, _, _, _, _, p = _der_read_tag_len(tst_info_bytes, p) # policy OID _, _, _, _, _, p = _der_read_tag_len(tst_info_bytes, p) # messageImprint SEQUENCE _, _, _, mi_c_off, mi_c_len, _ = _der_read_tag_len( tst_info_bytes, p ) mp = mi_c_off # hashAlgorithm SEQUENCE (skip) _, _, _, _, _, mp = _der_read_tag_len(tst_info_bytes, mp) # hashedMessage OCTET STRING _, _, h_tn, h_c_off, h_c_len, _ = _der_read_tag_len( tst_info_bytes, mp ) if h_tn != 0x04: return None return tst_info_bytes[h_c_off: h_c_off + h_c_len] except Exception: return None def verify_tsa_tokens(doc): def _mark_tsa_failed(): # Fix 4: every TSA NOT-VERIFIED branch must propagate tsa_failed # for v1.5 / v4-bound bundles so the final verdict downgrades to # VERIFIED (TSA_UNCONFIRMED). Previously only the cert-chain-walk # failure path (C2) flipped the flag, leaving parse/decode/ # missing-chain branches as silent forgery vectors. if isinstance(doc, dict) and ( doc.get("rankigi_schema") in ("1.5", "1.6") or doc.get("passport_key_fingerprint") ): doc.setdefault("_verify_state", {})["tsa_failed"] = True if not isinstance(doc, dict): print("RFC 3161 TSA: NOT YET ANCHORED") return tokens = doc.get("tsa_tokens") # Fallback path: pre-Day-1 bundles (and chains whose SDK has not yet # populated event_hash_chain.tsa_token_ref) carry a single chain-level # proof under rfc3161_timestamp instead of a tsa_tokens list. When that # envelope carries a TSR token plus the backfilled tsa_cert_chain_pem, # synthesize a one-entry tokens list so the chain-pem branch below runs # and the FreeTSA root can be pinned offline. if not isinstance(tokens, list) or len(tokens) == 0: rfc = doc.get("rfc3161_timestamp") if ( isinstance(rfc, dict) and rfc.get("tsr_token_base64") and rfc.get("tsa_cert_chain_pem") ): tokens = [ { "event_id": doc.get("closure_event_hash"), "tsa_url": rfc.get("tsa_url", "unknown"), "verified_at": rfc.get("tsa_timestamp", "unknown"), "token_present": True, "tsr_token_base64": rfc.get("tsr_token_base64"), "tsa_cert_chain_pem": rfc.get("tsa_cert_chain_pem"), "message_imprint_hash": rfc.get("message_imprint_hash"), } ] else: print("RFC 3161 TSA: NOT YET ANCHORED") return for tok in tokens: if not isinstance(tok, dict): continue tsa_url = tok.get("tsa_url", "unknown") verified_at = tok.get("verified_at", "unknown") message_imprint_hex = tok.get("message_imprint_hash") or tok.get( "message_imprint" ) or tok.get("hash") token_b64 = tok.get("tsr_token_base64") chain_pem = tok.get("tsa_cert_chain_pem") token_present = bool(tok.get("token_present") or tok.get("tsa_token_present")) print("TSA token: %s" % tsa_url) print(" Message imprint: %s" % (message_imprint_hex or "unknown")) print(" Verified at: %s" % verified_at) print(" Token: %s" % ("PRESENT" if token_present else "PENDING")) if not token_b64: print(" TSR token bytes: NOT IN BUNDLE (cannot verify offline)") print(" TSA signature: NOT VERIFIED (no token)") print(" Cert chain: NOT VERIFIED (chain not in bundle)") _mark_tsa_failed() continue try: tsr_bytes = base64.b64decode(token_b64) except Exception as e: print(" TSR token decode: FAILED (%s)" % type(e).__name__) print(" TSA signature: NOT VERIFIED (token undecodable)") print(" Cert chain: NOT VERIFIED (chain not in bundle)") _mark_tsa_failed() continue tst_info, signer_cert_der, signed_attrs_off, signed_attrs_len, \ raw_sig, _ = _extract_tst_info_from_tsr(tsr_bytes) if tst_info is None: print(" TSR token parse: FAILED (structure)") print(" MessageImprint: NOT VERIFIED (parse error)") print(" TSA signature: NOT VERIFIED (parse error)") print(" Cert chain: NOT VERIFIED (parse error)") _mark_tsa_failed() continue # Message imprint check. The bundle field message_imprint_hash holds # the SHA-256 hex of the content the TSA stamped. Recompute from the # bundle's recorded content (the hash itself, hex-decoded, is the # imprint the TSA signed over an octet string of the hash bytes when # the producer hashed the content first; FreeTSA emits hashedMessage # = sha256(content_hash). Here we trust the bundle-recorded # message_imprint_hash as the expected value and compare verbatim to # the TST's hashedMessage hex. token_imprint_bytes = _extract_tst_message_imprint(tst_info) if token_imprint_bytes is None: print(" MessageImprint: NOT VERIFIED (TSTInfo decode)") else: token_imprint_hex = token_imprint_bytes.hex() expected = (message_imprint_hex or "").lower() if expected and token_imprint_hex.lower() == expected: print(" MessageImprint: VERIFIED (FreeTSA root pinned)") elif expected: print(" MessageImprint: NOT VERIFIED (mismatch)") print(" expected: %s..." % expected[:32]) print(" token: %s..." % token_imprint_hex[:32]) _mark_tsa_failed() continue else: print(" MessageImprint: REPORTED (token=%s..., no expected hash in bundle)" % token_imprint_hex[:32]) if not chain_pem: print(" TSA signature: NOT VERIFIED (cert chain not in bundle)") print(" Cert chain: NOT VERIFIED (chain not in bundle)") _mark_tsa_failed() continue if not CRYPTO_AVAILABLE: print(" TSA signature: NOT VERIFIED (cryptography library missing)") print(" Cert chain: NOT VERIFIED (cryptography library missing)") _mark_tsa_failed() continue # Parse cert chain. Accept multi-cert PEM bundle. try: chain_certs = [] pem_text = chain_pem if isinstance(chain_pem, str) else chain_pem.decode( "utf-8", errors="replace" ) blocks = [] cur = [] in_cert = False for line in pem_text.splitlines(): if "-----BEGIN" in line: in_cert = True cur = [line] elif "-----END" in line and in_cert: cur.append(line) blocks.append("\n".join(cur) + "\n") cur = [] in_cert = False elif in_cert: cur.append(line) for b in blocks: chain_certs.append(x509.load_pem_x509_certificate(b.encode("utf-8"))) except Exception as e: print(" Cert chain: NOT VERIFIED (PEM parse failed, %s)" % type(e).__name__) print(" TSA signature: NOT VERIFIED (no usable chain)") _mark_tsa_failed() continue if not chain_certs: print(" Cert chain: NOT VERIFIED (empty chain)") print(" TSA signature: NOT VERIFIED (empty chain)") _mark_tsa_failed() continue # Verify TSR signerInfo signature against the leaf cert (chain_certs[0]). # signedAttrs path: the actual signed bytes are the DER re-encoding of # signedAttrs with the tag rewritten from [0] IMPLICIT (0xA0) to SET # (0x31). Without signedAttrs the signature covers eContent directly. # Phase B check-12 fix: the cert that signed the TST is the TSA's # SIGNING cert, embedded in the token's SignedData.certificates -- # NOT chain_certs[0]. The bundle pem carries the pinned ROOT only # (freetsa.org cacert.pem), so the old code verified the signature # against the root key and reported InvalidSignature on every honest # token. Verify against the token-embedded signer cert (fallback: # each bundle chain cert), then cryptographically bind that signer # cert to a bundle chain cert so an attacker cannot smuggle their own # signing cert inside a forged token. sig_ok = False sig_reason = "structure" signer_cert = None if raw_sig is None: sig_reason = "no signature octet string" else: if signed_attrs_off is not None and signed_attrs_len is not None: attrs = bytearray( tsr_bytes[signed_attrs_off: signed_attrs_off + signed_attrs_len] ) # Flip [0] IMPLICIT (0xA0) to SET (0x31) for the hash input. attrs[0] = 0x31 signed_input = bytes(attrs) else: signed_input = tst_info candidates = [] if signer_cert_der: try: candidates.append( x509.load_der_x509_certificate(signer_cert_der) ) except Exception: pass candidates.extend(chain_certs) # FreeTSA historically signed RSA+SHA-256; its current signing # cert is ECDSA P-384 with SHA-512. Try each digest for both key # types -- the signature must verify under exactly one. digests = (hashes.SHA256(), hashes.SHA384(), hashes.SHA512()) for cand in candidates: try: pubkey = cand.public_key() except Exception as e: sig_reason = type(e).__name__ continue for digest in digests: try: pubkey.verify(raw_sig, signed_input, PKCS1v15(), digest) sig_ok = True break except InvalidSignature: sig_reason = "invalid" continue except TypeError: # Not an RSA key; try ECDSA with the same digest. try: pubkey.verify(raw_sig, signed_input, ECDSA(digest)) sig_ok = True break except InvalidSignature: sig_reason = "invalid" continue except Exception as e2: sig_reason = type(e2).__name__ continue except Exception as e: sig_reason = type(e).__name__ continue if sig_ok: signer_cert = cand break # Bind the token-embedded signer cert to the pinned chain: its own # certificate signature must verify against one of the bundle chain # certs (the FreeTSA root signs its TSA signing cert directly). # Without this, signature validity alone would trust whatever cert # the token shipped. if sig_ok and signer_cert is not None and signer_cert not in chain_certs: bound = False for issuer in chain_certs: try: issuer.public_key().verify( signer_cert.signature, signer_cert.tbs_certificate_bytes, PKCS1v15(), signer_cert.signature_hash_algorithm, ) bound = True break except InvalidSignature: continue except TypeError: try: issuer.public_key().verify( signer_cert.signature, signer_cert.tbs_certificate_bytes, ECDSA(signer_cert.signature_hash_algorithm), ) bound = True break except Exception: continue except Exception: continue if not bound: sig_ok = False sig_reason = "signer cert not issued by pinned chain" if sig_ok: print(" TSA signature: VERIFIED (FreeTSA root pinned)") else: print(" TSA signature: NOT VERIFIED (%s)" % sig_reason) _mark_tsa_failed() # Walk chain: each non-root cert's signature should verify against # the next cert in the chain. The final cert is root. chain_ok = True chain_reason = "" try: for i in range(len(chain_certs) - 1): child = chain_certs[i] parent = chain_certs[i + 1] try: parent.public_key().verify( child.signature, child.tbs_certificate_bytes, PKCS1v15(), child.signature_hash_algorithm, ) except InvalidSignature: chain_ok = False chain_reason = "link %d invalid" % i break except TypeError: # EC parent: ECDSA over hash try: parent.public_key().verify( child.signature, child.tbs_certificate_bytes, ECDSA(child.signature_hash_algorithm), ) except InvalidSignature: chain_ok = False chain_reason = "link %d invalid" % i break except Exception as e: chain_ok = False chain_reason = type(e).__name__ if not chain_ok: # Explicit chain walk from leaf to pinned FreeTSA root failed. # Report as WARNING and continue: TSA chain failure does not # invalidate the local hash chain. print(" WARNING: Cert chain not verified (%s); continuing" % chain_reason) # Propagate failure to final verdict for v1.5/v1.6 / v4-bound bundles. if isinstance(doc, dict) and ( doc.get("rankigi_schema") in ("1.5", "1.6") or doc.get("passport_key_fingerprint") ): doc.setdefault("_verify_state", {})["tsa_failed"] = True continue # Root pin check. root_cert = chain_certs[-1] try: root_der = root_cert.public_bytes( __import__("cryptography").hazmat.primitives.serialization.Encoding.DER ) except Exception: root_der = None root_fp = hashlib.sha256(root_der).hexdigest() if root_der else None if FREETSA_ROOT_SHA256_FINGERPRINT and root_fp: if root_fp.lower() == FREETSA_ROOT_SHA256_FINGERPRINT.lower(): print(" Cert chain: VERIFIED (FreeTSA root pinned)") else: print(" Cert chain: NOT VERIFIED (root fingerprint mismatch)") print(" expected: %s..." % FREETSA_ROOT_SHA256_FINGERPRINT[:32]) print(" actual: %s..." % (root_fp or "")[:32]) _mark_tsa_failed() else: # No pin available in this verifier build. if root_fp: print(" Cert chain: PARTIAL (chain present but root not pinned)") print(" root fingerprint: sha256:%s..." % root_fp[:32]) else: print(" Cert chain: PARTIAL (chain present but root not extractable)") # === END AGENT-A-BLOCK === # === AGENT-C-BLOCK: REKOR VERIFICATION === # Day 1 Agent C will harden this section. Day 0 prep moved the comparison to # anchor_payload_hash (instead of last_event_hash). L1-N1 fix (2026-06-11): # anchor_payload_hash is sha256 over the raw 32 bytes of the snapshot hash -- # exactly the hashedrekord data.hash.value that RekorProvider.submit() # anchors. The earlier JCS-object formula was never what was anchored, so # every --rekor run reported MISMATCH. Pre-fix bundles (no # anchor_payload_formula marker) get a deprecation warning instead of a hard # exit; post-fix bundles compare strictly. # - Rekor binding (optional, --rekor flag) --------------------------------- REKOR_BASE = "https://rekor.sigstore.dev/api/v1/log/entries" # Pinned Rekor public key used to verify the signedEntryTimestamp on entries # fetched from the public log. If the entry signature does not verify, we # print a WARNING and continue: offline chain verification is authoritative. REKOR_PUBLIC_KEY_PEM = b"""-----BEGIN PUBLIC KEY----- MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE2ryMB3Y4rDy5z9/QH13TRYBl4tNB2cNQ+8FJVhsSDz0AiG9tMrmvNSL3GHrqgHJFoxLhGz8N2wf3N0x8ADGOSQ== -----END PUBLIC KEY----- """ def _verify_rekor_entry_signature(entry): """Best-effort verification of a Rekor entry's signedEntryTimestamp against the pinned Rekor public key. Returns (ok, reason). Never raises. The signedEntryTimestamp is an ECDSA signature (P-256, SHA-256) over the canonical JSON of {body, integratedTime, logID, logIndex}. A failure here is reported as WARNING and does not fail overall verification. """ if not CRYPTO_AVAILABLE: return False, "cryptography library missing" if not isinstance(entry, dict): return False, "entry not dict" verification = entry.get("verification") if not isinstance(verification, dict): return False, "no verification block" set_b64 = verification.get("signedEntryTimestamp") if not isinstance(set_b64, str) or not set_b64: return False, "no signedEntryTimestamp" try: sig = base64.b64decode(set_b64) except Exception: return False, "set base64 decode" payload = { "body": entry.get("body"), "integratedTime": entry.get("integratedTime"), "logID": entry.get("logID"), "logIndex": entry.get("logIndex"), } try: canonical = _canonical(payload).encode("utf-8") except Exception as e: return False, "canonical: %s" % type(e).__name__ try: rekor_key = load_pem_public_key(REKOR_PUBLIC_KEY_PEM) rekor_key.verify(sig, canonical, ECDSA(hashes.SHA256())) return True, "ok" except InvalidSignature: return False, "invalid signature" except Exception as e: return False, type(e).__name__ def fetch_rekor_hash(log_index, offline=False): """Fetch the hash anchored at a specific Rekor log index. Returns (rekor_hash_hex, None) on success, (None, reason) on any failure. Never raises; offline verification is authoritative. When offline=True, no network call is attempted and the caller receives reason="offline mode". """ if offline: return None, "offline mode" try: url = "%s?logIndex=%d" % (REKOR_BASE, int(log_index)) req = urllib.request.Request( url, headers={"Accept": "application/json"} ) with urllib.request.urlopen(req, timeout=5) as resp: raw = resp.read() except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError) as e: return None, "network unavailable (%s)" % type(e).__name__ except Exception as e: return None, "fetch failed (%s)" % type(e).__name__ try: entries = json.loads(raw.decode("utf-8")) except Exception: return None, "Rekor returned non-JSON" if not isinstance(entries, dict) or not entries: return None, "Rekor returned empty entry set" first_key = next(iter(entries)) entry = entries[first_key] body_b64 = entry.get("body") if isinstance(entry, dict) else None if not isinstance(body_b64, str): return None, "Rekor entry has no body" # Verify the entry signature against the pinned Rekor key. On failure, # refuse to compare body hashes: an MITM that served a fabricated entry # JSON could otherwise trick the body-hash compare. Treat as Rekor failed. sig_ok, sig_reason = _verify_rekor_entry_signature(entry) if sig_ok: print("Rekor entry signature: VERIFIED (pinned Rekor key) (Sigstore transparency log)") else: print("Rekor entry signature: NOT VERIFIED (signedEntryTimestamp invalid)") print("Rekor anchor: SKIPPED (entry signature required for trusted comparison)") return None, "ENTRY_SIG_FAILED:%s" % sig_reason try: decoded = json.loads(base64.b64decode(body_b64).decode("utf-8")) except Exception: return None, "Rekor body decode failed" value = ( decoded.get("spec", {}).get("data", {}).get("hash", {}).get("value", "") if isinstance(decoded, dict) else "" ) if not isinstance(value, str) or not value: return None, "Rekor body missing hash.value" return value.lower(), None def verify_rekor_binding(log_index, expected_hash, rekor_requested=False, offline=False, doc=None): """Compare expected_hash against the Rekor entry at log_index. Prints a structured block describing the third-party transparency-log witness and returns one of 'verified', 'legacy_unbound', 'mismatch', 'skipped', 'required_unreachable', 'failed'. Network failures are SKIPPED (offline verification remains authoritative) and never raise UNLESS rekor_requested=True, in which case the unreachable case is marked on doc._verify_state.rekor_required_unreachable so the final verdict can downgrade to VERIFIED (REKOR_UNCONFIRMED). L1-N1 (2026-06-11): expected_hash is sha256 over the raw snapshot-hash bytes -- the exact data.hash.value submit() anchors. Bundles exported before the fix carry the deprecated JCS-object digest, which can never match a real Rekor entry; those bundles are recognised by the ABSENCE of anchor_payload_formula == "snapshot_bytes_v2" on the doc and downgrade to a deprecation warning ('legacy_unbound') instead of a hard exit. Post-fix bundles declare the formula marker and compare strictly. """ expected = (expected_hash or "").lower() public_url = "https://search.sigstore.dev/?logIndex=%s" % log_index if offline: print("Rekor anchor: SKIPPED (offline mode)") return "skipped" rekor_hash, reason = fetch_rekor_hash(log_index, offline=offline) if rekor_hash is None: if isinstance(reason, str) and reason.startswith("ENTRY_SIG_FAILED:"): # signedEntryTimestamp did not verify against pinned Rekor key. # Treat as Rekor failed, not skipped. print("Rekor anchor: logIndex %s" % log_index) print("Independent re-verify: %s" % public_url) return "failed" # Network/availability failure. Offline chain verification is # authoritative, so we report SKIPPED rather than failing. # Fix 5: when --rekor was explicitly requested, flag the bundle so # the verdict path downgrades to VERIFIED (REKOR_UNCONFIRMED) instead # of fail-open SKIPPED. Without --rekor, behaviour is unchanged. if rekor_requested: print("Rekor anchor: REQUIRED BUT UNREACHABLE (network failure)") print("Reason: %s" % reason) print("Independent re-verify: %s" % public_url) if isinstance(doc, dict): doc.setdefault("_verify_state", {})["rekor_required_unreachable"] = True return "required_unreachable" print("Rekor anchor: logIndex %s" % log_index) print("Anchor hash: sha256:%s..." % expected[:32] if expected else "Anchor hash: (none)") print("Verified at search.sigstore.dev: SKIPPED (%s)" % reason) print("Independent re-verify: %s" % public_url) return "skipped" if rekor_hash == expected: print("Rekor anchor: logIndex %s" % log_index) print("Anchor hash: sha256:%s..." % expected[:32]) print("Verified at search.sigstore.dev: YES (Sigstore transparency log)") print("Independent re-verify: %s" % public_url) return "verified" # Dual-compare window for pre-L1-N1 bundles. Bundles exported before the # 2026-06-11 fix store anchor_payload_hash under the deprecated JCS-object # formula sha256(canonicalJson({canonicalization_version, org_id, # previous_sealed_head, snapshot_date, snapshot_hash})), which by # construction can never equal the snapshot-bytes digest the Rekor entry # actually holds. Those bundles do not carry the snapshot hash itself, so # the binding cannot be recomputed offline; rather than report a false # MISMATCH (tamper signal), print a deprecation warning and continue. # Post-fix bundles declare anchor_payload_formula == "snapshot_bytes_v2" # and take the strict path below. formula = (doc or {}).get("anchor_payload_formula") if formula != "snapshot_bytes_v2": print("Rekor anchor: logIndex %s" % log_index) print("Anchor hash: sha256:%s... (deprecated pre-2026-06-11 JCS formula)" % expected[:32]) print("Rekor entry hash: sha256:%s..." % rekor_hash[:32]) print("DEPRECATION WARNING: this bundle predates the anchor-formula fix") print(" (L1-N1). Its anchor_payload_hash uses the retired JCS-object") print(" formula and cannot be bound to the Rekor entry offline.") print(" Re-export the bundle to obtain a strictly verifiable anchor.") print("Independent re-verify: %s" % public_url) return "legacy_unbound" print("Rekor anchor: logIndex %s" % log_index) print("Anchor hash: sha256:%s..." % expected[:32] if expected else "Anchor hash: (none)") print("Verified at search.sigstore.dev: MISMATCH") print(" expected: %s..." % expected[:32]) print(" rekor: %s..." % rekor_hash[:32]) print("Independent re-verify: %s" % public_url) sys.exit(1) # === END AGENT-C-BLOCK === # === AGENT-E-BLOCK: CCAP VERIFICATION === # CCAP (Counterparty Closure Attestation Protocol) receipt verification. # # Phase 3 week 3 (2026-05-28): when --ccap-keys is supplied, this block # verifies each receipt's signature against the named public key using the # canonical pre-image the producer signed (signed_payload_canonical from # src/lib/ccap/receipts.ts). Without --ccap-keys, every receipt is reported # as REPORTED rather than VERIFIED. The hash chain and dossier verdicts are # unaffected by CCAP outcome; this is a per-receipt attestation layer. # # The HMAC algorithms (Stripe webhook signatures, Svix headers) require the # shared secret which RANKIGI does not hold offline. They are explicitly # labelled REPORTED rather than verified. def load_ccap_keys(path): """Load CCAP signing keys from a JWKS file, flat JSON dict, or directory. Returns dict mapping kid -> cryptography public_key object. Returns {} when the path is None or the file is unreadable. """ if not path: return {} if not CRYPTO_AVAILABLE: print("WARNING: --ccap-keys supplied but cryptography library missing.") return {} import os keys = {} try: if os.path.isdir(path): for fname in sorted(os.listdir(path)): if not fname.endswith(".pem"): continue kid = fname[:-4] with open(os.path.join(path, fname), "rb") as f: pem = f.read() try: keys[kid] = load_pem_public_key(pem) except Exception as e: print("WARNING: ccap key %s failed to load (%s)" % (kid, type(e).__name__)) return keys with open(path, "rb") as f: raw = f.read() data = json.loads(raw.decode("utf-8")) if isinstance(data, dict) and isinstance(data.get("keys"), list): # JWKS form: each entry has kid + x5c (PEM cert chain) OR x (Ed25519). # We support the simpler embedded-PEM convention { kid, pem }. for jwk in data["keys"]: if not isinstance(jwk, dict): continue kid = jwk.get("kid") pem = jwk.get("pem") or jwk.get("public_key_pem") if not kid or not pem: continue try: keys[kid] = load_pem_public_key( pem.encode("utf-8") if isinstance(pem, str) else pem ) except Exception as e: print("WARNING: ccap key %s failed to load (%s)" % (kid, type(e).__name__)) return keys if isinstance(data, dict): # Flat dict: { kid: pem_string }. for kid, pem in data.items(): if not isinstance(pem, str): continue try: keys[kid] = load_pem_public_key(pem.encode("utf-8")) except Exception as e: print("WARNING: ccap key %s failed to load (%s)" % (kid, type(e).__name__)) return keys except Exception as e: print("WARNING: --ccap-keys load failed (%s)" % type(e).__name__) return {} return keys def _verify_signature_with_alg(public_key, signature, signed_input, sig_alg): """Verify signature using the algorithm hint. Returns (ok, reason).""" alg = (sig_alg or "").lower() try: if alg in ("ed25519",): public_key.verify(signature, signed_input) return True, "Ed25519" if alg in ("es256", "ecdsa-p256-sha256"): public_key.verify(signature, signed_input, ECDSA(hashes.SHA256())) return True, "ES256" # Default attempt: Ed25519 if Ed25519PublicKey, else ECDSA-P256. if isinstance(public_key, Ed25519PublicKey): public_key.verify(signature, signed_input) return True, "Ed25519" if isinstance(public_key, EllipticCurvePublicKey): public_key.verify(signature, signed_input, ECDSA(hashes.SHA256())) return True, "ES256" return False, "unsupported alg %r" % sig_alg except InvalidSignature: return False, "invalid signature" except Exception as e: return False, type(e).__name__ def verify_ccap_receipts(doc, ccap_keys=None): """Verify CCAP receipt signatures using a supplied keymap. Returns (ok, summary_string) where ok is False if any receipt that COULD have been verified failed. REPORTED receipts (HMAC algorithms or no key supplied) do NOT count as failures; they degrade to REPORTED. """ receipts = doc.get("counterparty_receipts") if isinstance(doc, dict) else None if not isinstance(receipts, list) or len(receipts) == 0: return True, "no receipts" event_hashes = doc.get("event_hashes") or [] total_events = len(event_hashes) if isinstance(event_hashes, list) else 0 receipted = doc.get("receipt_count") if not isinstance(receipted, int): receipted = len(receipts) print("Counterparty receipts: %d" % len(receipts)) if total_events > 0: print("Receipt coverage: %d of %d events" % (receipted, total_events)) else: print("Receipt coverage: %d events" % receipted) if ccap_keys is None: ccap_keys = {} all_ok = True HMAC_ALGS = {"stripe-hmac-sha256", "svix-hmac-sha256", "hmac-sha256"} for r in receipts: if not isinstance(r, dict): continue provider = r.get("provider", "unknown") witness_class = r.get("witness_class", "unattested") status = r.get("status", "unknown") sig_alg = r.get("sig_alg") sig_b64 = r.get("signature") signed_canonical = r.get("signed_payload_canonical") signer_kid = r.get("signer_key_id") or r.get("counterparty_key_id") print(" Provider: %s" % provider) print(" Witness class: %s" % witness_class) print(" Status: %s" % status) # HMAC algorithms: cannot verify offline without the shared secret. if sig_alg and sig_alg.lower() in HMAC_ALGS: print(" %s: REPORTED (HMAC; offline verify not supported)" % provider) continue # No signature: report only. if not sig_b64 or not signed_canonical: print(" %s (%s): REPORTED (no signature fields)" % (provider, witness_class)) continue if not CRYPTO_AVAILABLE: print(" %s (%s): REPORTED (cryptography library missing)" % (provider, witness_class)) continue # Resolve a key. public_key = None key_label = None if witness_class == "unattested" and signer_kid in ( RANKIGI_ROOT_KEY_ID, RANKIGI_WITNESS_KEY_ID, ): try: if signer_kid == RANKIGI_ROOT_KEY_ID: public_key = load_pem_public_key(RANKIGI_ROOT_PUBKEY_PEM) key_label = "root" else: public_key = load_pem_public_key(RANKIGI_WITNESS_PUBKEY_PEM) key_label = "witness" except Exception: public_key = None elif signer_kid and signer_kid in ccap_keys: public_key = ccap_keys[signer_kid] key_label = signer_kid if public_key is None: if not ccap_keys: print(" %s (%s): REPORTED (not cryptographically verified offline, supply --ccap-keys)" % (provider, witness_class)) else: print(" %s (%s): REPORTED (no matching key for kid=%s)" % (provider, witness_class, signer_kid or "unknown")) continue try: sig_bytes = base64.b64decode(sig_b64) except Exception: print(" %s: FAILED (signature base64 decode)" % provider) all_ok = False continue signed_input = ( signed_canonical.encode("utf-8") if isinstance(signed_canonical, str) else signed_canonical ) ok, reason = _verify_signature_with_alg( public_key, sig_bytes, signed_input, sig_alg, ) if ok: if witness_class == "unattested" and key_label in ("root", "witness"): print(" rankigi-%s: VERIFIED (Ed25519, %s)" % (key_label, key_label)) else: print(" %s: VERIFIED (%s, kid=%s)" % (provider, reason, signer_kid or "unknown")) else: print(" %s: FAILED (%s)" % (provider, reason)) all_ok = False return all_ok, "checked %d receipts" % len(receipts) # === END AGENT-E-BLOCK === # === AGENT-D-BLOCK: DOSSIER OUTPUT === # Action Dossier availability output. Renders the FRE 902(14) certification # block summary and download pointer in a court-receivable text format. def report_action_dossier(doc): if not isinstance(doc, dict): return version = doc.get("dossier_version") if not version: return chain_id = doc.get("chain_id") or doc.get("chainId") or "{chainId}" print("Action Dossier: AVAILABLE") print(" Version: %s" % version) print(" FRE applicable: %s" % doc.get("fre_applicable", False)) print(" UAR completeness: %s%%" % doc.get("uar_completeness", 0)) print(" Download: POST /api/chains/%s/dossier" % chain_id) def report_fre_applicability(doc, local_verified): """Print the FRE 902(14) applicability block immediately after the final PASS/FAIL/UNVERIFIED verdict. NEVER prints TRUE unless the bundle's dossier claims TRUE AND verify.py's own independent checks all passed. Disagreement collapses to FALSE with a disclosure that the bundle's claim was overridden locally. Phase 3 week 3: local_verified is required to be False whenever the key_source is bundle-self-reported. Callers enforce this; the function trusts its argument. """ if not isinstance(doc, dict): return # Prefer the merged dossier block; fall back to top-level legacy fields. dossier = doc.get("dossier") if isinstance(doc.get("dossier"), dict) else {} bundle_claim = dossier.get("fre_applicable") if bundle_claim is None: bundle_claim = doc.get("fre_applicable") basis = dossier.get("fre_basis") if basis is None: basis = doc.get("fre_basis") if bundle_claim is None and basis is None: # Pre-Day-2 bundle without a dossier block; emit nothing rather than # implying applicability either way. return if bundle_claim is True and local_verified: print("FRE 902(14): TRUE") elif bundle_claim is True and not local_verified: print("FRE 902(14): FALSE (bundle claimed TRUE, but key source is self-reported or local verification disagreed, independent verification required)") else: print("FRE 902(14): FALSE") if isinstance(basis, list) and len(basis) > 0: print("Basis:") for line in basis: print(" - %s" % line) # === END AGENT-D-BLOCK === # === AGENT-B-BLOCK: AUTHORIZATIONS === def _b64url_decode(s): """RFC 4648 base64url decode tolerant of missing padding.""" if isinstance(s, bytes): s = s.decode("ascii", errors="replace") s = s.replace("-", "+").replace("_", "/") pad = (-len(s)) % 4 if pad: s = s + ("=" * pad) return base64.b64decode(s) def _verify_webauthn_assertion(a): # WebAuthn verification not yet available. The bundle may still carry # webauthn_* fields from earlier issuances; the offline verifier no # longer attempts to replay the ceremony. return "WebAuthn assertion: not yet available" # --- legacy implementation retained for reference (unreachable) --- cdj = a.get("webauthn_client_data_json") auth_data_b64 = a.get("webauthn_authenticator_data") sig_b64 = a.get("webauthn_signature") spki_b64 = a.get("webauthn_credential_public_key_spki") challenge_field = a.get("webauthn_challenge") if not (cdj and auth_data_b64 and sig_b64 and spki_b64): return "WebAuthn assertion: NOT VERIFIED (missing assertion fields)" if not CRYPTO_AVAILABLE: return "WebAuthn assertion: NOT VERIFIED (cryptography library missing)" try: cdj_bytes = cdj.encode("utf-8") if isinstance(cdj, str) else cdj client_data = json.loads(cdj_bytes.decode("utf-8")) except Exception: return "WebAuthn assertion: NOT VERIFIED (clientDataJSON parse)" if client_data.get("type") != "webauthn.get": return "WebAuthn assertion: NOT VERIFIED (wrong type %r)" % client_data.get("type") allowed_origins = {"https://rankigi.com", "https://www.rankigi.com"} origin = client_data.get("origin") if origin not in allowed_origins: return "WebAuthn assertion: NOT VERIFIED (origin %s)" % origin # Challenge binding: compare base64url-decoded values. try: client_challenge = _b64url_decode(client_data.get("challenge", "")) bundle_challenge = _b64url_decode(challenge_field or "") if not client_challenge or client_challenge != bundle_challenge: return "WebAuthn assertion: NOT VERIFIED (challenge mismatch)" except Exception: return "WebAuthn assertion: NOT VERIFIED (challenge decode)" try: auth_data = base64.b64decode(auth_data_b64) sig = base64.b64decode(sig_b64) spki = base64.b64decode(spki_b64) except Exception: return "WebAuthn assertion: NOT VERIFIED (base64 decode)" try: public_key = load_der_public_key(spki) except Exception as e: return "WebAuthn assertion: NOT VERIFIED (SPKI parse %s)" % type(e).__name__ cdj_hash = hashlib.sha256(cdj_bytes).digest() signed_bytes = auth_data + cdj_hash try: if isinstance(public_key, Ed25519PublicKey): public_key.verify(sig, signed_bytes) return "WebAuthn assertion: VERIFIED (Ed25519, RP rankigi.com)" if isinstance(public_key, EllipticCurvePublicKey): public_key.verify(sig, signed_bytes, ECDSA(hashes.SHA256())) return "WebAuthn assertion: VERIFIED (ES256, RP rankigi.com)" return "WebAuthn assertion: NOT VERIFIED (unsupported key type)" except InvalidSignature: return "WebAuthn assertion: NOT VERIFIED (invalid signature)" except Exception as e: return "WebAuthn assertion: NOT VERIFIED (%s)" % type(e).__name__ def report_authorizations(doc): """Stage 6 (6B): real authorization verification, replacing the old unconditional "Authorization hashes: PRESENT" claim. Per grant: 1. Recompute authorization_hash from the bundle fields (sha256(canonicalJson({scope_hash, agent_id, granted_at, granted_by}))) -- the exact preimage POST /api/authorizations computes. Supersedure rows (approval rows carrying approved_at, and revocation tombstones) use DIFFERENT preimages that reference the original grant's hash, which the bundle record does not carry, so they report NOT RECOMPUTABLE rather than a false MISMATCH. 2. Verify grant_signature against the pinned RANKIGI keys (root preferred, witness fallback -- the same keys internal events use). 3. Cross-check each event's action against its grant's allowed_actions (the grant is located via proof_extensions.authorization_hash on v5 events, or a top-level authorization_hash when present). """ if not isinstance(doc, dict): print("Authorization records: NONE") print("Authorization hashes: NONE") return auths = doc.get("authorizations") if not isinstance(auths, list) or len(auths) == 0: print("Authorization records: NONE") print("Authorization hashes: NONE") return print("Authorization records: %d" % len(auths)) any_webauthn_present = False any_webauthn_failed = False recomputed_ok = 0 recomputed_mismatch = 0 not_recomputable = 0 grants_by_hash = {} for a in auths: if not isinstance(a, dict): continue auth_hash = a.get("authorization_hash") if isinstance(auth_hash, str): grants_by_hash[auth_hash] = a print(" Type: %s" % a.get("authority_type", "unknown")) print(" Granted by: %s" % a.get("granted_by", "unknown")) print(" Scope hash: %s" % a.get("scope_hash", "unknown")) print(" Hash: %s" % a.get("authorization_hash", "unknown")) # 1. authorization_hash recompute. label = str(auth_hash)[:8] if isinstance(auth_hash, str) else "????????" if a.get("authority_type") == "revocation": not_recomputable += 1 print(" Grant %s...: hash NOT RECOMPUTABLE (revocation tombstone; preimage references the original grant)" % label) elif a.get("approved_at"): not_recomputable += 1 print(" Grant %s...: hash NOT RECOMPUTABLE (approval supersedure; preimage references the original grant)" % label) else: try: granted_at = a.get("granted_at") try: granted_at = normalize_occurred_at(granted_at) except (ValueError, TypeError): pass recomputed = hashlib.sha256(_canonical({ "scope_hash": a.get("scope_hash"), "agent_id": a.get("agent_id"), "granted_at": granted_at, "granted_by": a.get("granted_by") or "api_key", }).encode("utf-8")).hexdigest() except Exception as e: recomputed = None not_recomputable += 1 print(" Grant %s...: hash NOT RECOMPUTABLE (%s)" % (label, type(e).__name__)) if recomputed is not None: if recomputed == auth_hash: recomputed_ok += 1 print(" Grant %s...: hash VERIFIED" % label) else: recomputed_mismatch += 1 print(" Grant %s...: hash MISMATCH" % label) print(" expected: %s" % auth_hash) print(" computed: %s" % recomputed) # A mismatch means the bundle fields do not reproduce the # stored hash -- a bundle integrity issue, not necessarily # tampering (e.g. a pre-Stage-3 export missing agent_id). # 2. grant_signature: RANKIGI witnessed this grant at creation. The # signing input mirrors src/app/api/authorizations/route.ts # grantCanonical exactly. sig_b64 = a.get("grant_signature") if isinstance(sig_b64, str) and sig_b64: try: granted_at_sig = a.get("granted_at") try: granted_at_sig = normalize_occurred_at(granted_at_sig) except (ValueError, TypeError): pass signing_input = _canonical({ "authorization_hash": a.get("authorization_hash"), "scope_hash": a.get("scope_hash"), "agent_id": a.get("agent_id") if a.get("agent_id") is not None else None, "granted_at": granted_at_sig, "granted_by": a.get("granted_by"), "authority_type": a.get("authority_type"), "allowed_actions": a.get("allowed_actions") or [], }) sig = base64.b64decode(sig_b64) if _verify_witness_signature(sig, signing_input): print(" Grant signature: VERIFIED (RANKIGI witnessed this grant)") else: print(" Grant signature: INVALID") except Exception as e: print(" Grant signature: INVALID (%s)" % type(e).__name__) else: print(" Grant signature: NOT PRESENT") # WebAuthn block: only report when at least one of the four fields is # present; pre-Phase-3 records without WebAuthn slots stay silent. webauthn_fields_present = any( a.get(k) for k in ( "webauthn_client_data_json", "webauthn_authenticator_data", "webauthn_signature", "webauthn_credential_public_key_spki", ) ) if webauthn_fields_present: any_webauthn_present = True verdict = _verify_webauthn_assertion(a) print(" %s" % verdict) if "NOT VERIFIED" in verdict: any_webauthn_failed = True # 3. Cross-check event actions vs grant allowed_actions. events = doc.get("events") if isinstance(events, list) and grants_by_hash: for ev in events: if not isinstance(ev, dict): continue ev_auth_hash = ev.get("authorization_hash") if not ev_auth_hash: pe = ev.get("proof_extensions") if isinstance(pe, dict): ev_auth_hash = pe.get("authorization_hash") if not ev_auth_hash: continue grant = grants_by_hash.get(ev_auth_hash) if not grant: continue allowed = grant.get("allowed_actions") if isinstance(allowed, list) and len(allowed) > 0: if ev.get("action") not in allowed: print(" WARNING: event %s action '%s' is outside grant scope %s" % (ev.get("chain_index"), ev.get("action"), allowed)) # Summary: never claim more than what was actually checked. if recomputed_mismatch > 0: print("Authorization hashes: MISMATCH (see above)") elif recomputed_ok > 0: print("Authorization hashes: VERIFIED") elif not_recomputable > 0: print("Authorization hashes: NOT RECOMPUTABLE (supersedure rows only)") else: print("Authorization hashes: NONE") if any_webauthn_present and any_webauthn_failed: # Caller (the verdict block) reads this flag via doc-level cache below. doc.setdefault("_verify_state", {})["webauthn_failed"] = True # === END AGENT-B-BLOCK === # === DATA-LINEAGE-BLOCK === def verify_data_lineage(bundle): """Report the data_lineage section (L5 Gap 2): which data_chain receipts the bundle's events consumed, from which registered sources. Presence reporting only -- full cryptographic verification of data receipts requires the data chain export (separate from the event chain bundle). """ if not isinstance(bundle, dict): print("Data lineage: NOT PRESENT in bundle") return lineage = bundle.get('data_lineage') if not lineage or not isinstance(lineage, dict): print("Data lineage: NOT PRESENT in bundle") return total = lineage.get('total_receipts', 0) sourced = lineage.get('total_events_with_receipts', 0) print("Data lineage: %s events linked to %s receipts" % (sourced, total)) receipts = lineage.get('receipts') if isinstance(receipts, list): for receipt in receipts: if not isinstance(receipt, dict): continue # Verify receipt hash is present before reporting the receipt. if receipt.get('payload_hash'): print(" Receipt %s...: source %s..., classification: %s" % ( str(receipt.get('receipt_id', 'unknown'))[:8], str(receipt.get('source_id', 'unknown'))[:8], receipt.get('classification', 'unknown'), )) # Note: full cryptographic verification of data receipts requires the # data chain export (separate from the event chain bundle). print("Data lineage: PRESENCE VERIFIED (cryptographic binding requires Phase B)") # === END DATA-LINEAGE-BLOCK === # === AGENT-F-BLOCK: CITATIONS === def report_citations(doc): if not isinstance(doc, dict): print("Citation provenance: NONE") return citations = doc.get("citations") if citations is None: print("Citation provenance: NONE") return lock_status = doc.get("citation_lock_status") or doc.get("lock_status") or "unknown" if isinstance(citations, list): items = citations elif isinstance(citations, dict): items = citations.get("items") or [] lock_status = citations.get("lock_status", lock_status) else: items = [] print("Citation lock: %s" % lock_status) print("Citations: %d" % len(items)) for c in items: if not isinstance(c, dict): continue print(" Source: %s" % c.get("source_name", "unknown")) print(" Type: %s" % c.get("citation_type", "unknown")) print(" Used in output: %s" % c.get("used_in_output", False)) print(" Hash: %s" % c.get("citation_hash", "unknown")) # === END AGENT-F-BLOCK === # === AGENT-CROSSING-BLOCK === def verify_crossings(doc): """Report world receipts (universal crossing layer) bound to this chain. A crossing is a world receipt (Stripe webhook, GitHub event, DKIM-signed email, Rekor entry, CloudTrail digest, etc.) bound to a chain event. This is world-down-to-agent provenance: proof the world acted, tied to the agent action that caused it. The crossings array is hash-bound by the envelope content_hash, so no RANKIGI access is needed to trust it. """ crossings = doc.get("crossings", []) if not crossings: print("World receipts: NONE") return total = len(crossings) verified = sum( 1 for c in crossings if c.get("witness_class") not in ("unattested", "sidecar_observed") ) print("World receipts: %d" % total) print("Counterparty signed: %d/%d" % (verified, total)) for c in crossings: provider = c.get("provider", "unknown") strength = c.get("claim_strength", "unknown") witness = c.get("witness_class", "unknown") bound = c.get("bound_event_id") print(" %s: %s (%s)" % (provider, strength, witness)) if bound: print(" Bound to event: %s..." % str(bound)[:8]) else: print(" WARNING: unbound world receipt") # === END AGENT-CROSSING-BLOCK === # === AGENT-G-BLOCK: XACT === # Pinned RANKIGI XACT witness public key (Ed25519). Cross-check against # https://rankigi.com/.well-known/xact-key.json (field public_key_pem) and # public/.well-known/rankigi-xact-key.json. RANKIGI_XACT_PUBKEY_PEM = b"""-----BEGIN PUBLIC KEY----- MCowBQYDK2VwAyEA7UpQ1GcjWW1YzHNPGIy7cja2C42LuAsvV+D+TI+maHI= -----END PUBLIC KEY----- """ _XACT_MAX_DEPTH = 12 def xact_canonical(value, depth=0): """Strict XACT canonical JSON, matching src/lib/xact/canonical.ts. Differences from the chain _canonical: floats are rejected (integer minor units only) and nesting is capped at depth 12. For ASCII-key objects this produces identical bytes to the chain canonicalizer, but XACT money-adjacent data must never silently round a float. """ if depth > _XACT_MAX_DEPTH: raise ValueError("XACT canonical JSON exceeds max nesting depth 12") if value is None: return "null" if value is True: return "true" if value is False: return "false" if isinstance(value, str): return _escape_string(value) if isinstance(value, bool): return "true" if value else "false" if isinstance(value, int): return str(value) if isinstance(value, float): raise ValueError("XACT canonical JSON does not allow floats. Use integer minor units.") if isinstance(value, list): return "[" + ",".join(xact_canonical(v, depth + 1) for v in value) + "]" if isinstance(value, dict): keys = sorted(value.keys()) return "{" + ",".join( _escape_string(k) + ":" + xact_canonical(value[k], depth + 1) for k in keys ) + "}" raise TypeError("cannot canonicalize type %s" % type(value).__name__) def xact_canonical_hash(obj): return hashlib.sha256(xact_canonical(obj).encode("utf-8")).hexdigest() def xact_commitment_root(transcript_hash, fragment_a_hash, fragment_b_hash, party_a_fp, party_b_fp): """Mirror src/lib/xact/crypto.ts computeCommitmentRoot.""" return xact_canonical_hash({ "schema": "rankigi.xact.commitment_root.v0", "fragment_a_hash": fragment_a_hash, "fragment_b_hash": fragment_b_hash, "party_a_pubkey_fingerprint": party_a_fp, "party_b_pubkey_fingerprint": party_b_fp, "transcript_hash": transcript_hash, }) def _xact_verify_ed25519(pem, message_hex, signature_b64): """Verify Ed25519 over the UTF-8 bytes of message_hex (matching signWithXactKey). Returns False when crypto is unavailable or on any failure.""" if not CRYPTO_AVAILABLE: return False try: if isinstance(pem, str): pem = pem.encode("utf-8") key = load_pem_public_key(pem) if not isinstance(key, Ed25519PublicKey): return False key.verify(base64.b64decode(signature_b64), message_hex.encode("utf-8")) return True except Exception: return False def verify_xact_commitments(doc): xact = doc.get("xact_commitments") or {} if isinstance(xact, list): commitments = xact else: commitments = xact.get("xact_commitments", []) if isinstance(xact, dict) else [] if not commitments: print("XACT commitments: NONE") return count = len(commitments) bilateral = sum(1 for c in commitments if c.get("party_b_attestation_class") == "bilateral_signed") print(f"XACT commitments: {count}") print(f"Bilateral signed: {bilateral} of {count}") if not CRYPTO_AVAILABLE: print(" (cryptography library missing -- hash recomputation only, signatures NOT checked)") for c in commitments: cid = (c.get("commitment_id") or "unknown")[:8] state = c.get("state", "unknown") attestation = c.get("party_b_attestation_class", "unknown") print(f" [{cid}] {state} | {attestation}") checks = [] problems = [] transcript = c.get("transcript") commitment_root = c.get("commitment_root") or c.get("transcript_hash") # 1. Recompute transcript_hash and bind it to the proposal root. if isinstance(transcript, dict) and commitment_root: try: recomputed = xact_canonical_hash(transcript) except Exception as e: recomputed = None problems.append("transcript canonicalization failed: %s" % e) if recomputed is not None: if recomputed == commitment_root: checks.append("transcript_hash: OK") else: problems.append("transcript_hash: MISMATCH") else: problems.append("transcript: MISSING (cannot recompute)") # 2. Verify A's proposal signature over commitment_root. prop_sig = c.get("party_a_proposal_signature") or c.get("party_a_signature") if commitment_root and prop_sig: if _xact_verify_ed25519(RANKIGI_XACT_PUBKEY_PEM, commitment_root, prop_sig): checks.append("party_a proposal sig: VALID") elif CRYPTO_AVAILABLE: problems.append("party_a proposal sig: INVALID") elif CRYPTO_AVAILABLE: problems.append("party_a proposal sig: ABSENT") # 3-5. Accepted commitments: final root + counter-signatures. final_root = c.get("final_commitment_root") if final_root: fa = c.get("fragment_a_hash") fb = c.get("fragment_b_hash") pa_fp = c.get("witness_key_fingerprint") pb_fp = c.get("party_b_pubkey_fingerprint") if commitment_root and fa and fb and pa_fp and pb_fp: try: recomputed_final = xact_commitment_root(commitment_root, fa, fb, pa_fp, pb_fp) except Exception as e: recomputed_final = None problems.append("final root canonicalization failed: %s" % e) if recomputed_final is not None: if recomputed_final == final_root: checks.append("final_commitment_root: OK") else: problems.append("final_commitment_root: MISMATCH") else: problems.append("final root inputs MISSING") final_sig = c.get("party_a_final_signature") if final_sig: if _xact_verify_ed25519(RANKIGI_XACT_PUBKEY_PEM, final_root, final_sig): checks.append("party_a counter-sig: VALID") elif CRYPTO_AVAILABLE: problems.append("party_a counter-sig: INVALID") elif CRYPTO_AVAILABLE: problems.append("party_a counter-sig: ABSENT") b_sig = c.get("party_b_signature") b_pem = c.get("party_b_pubkey_pem") if b_sig: if b_pem: if _xact_verify_ed25519(b_pem, final_root, b_sig): checks.append("party_b sig: VALID") elif CRYPTO_AVAILABLE: problems.append("party_b sig: INVALID") else: problems.append("party_b sig present but party_b_pubkey_pem MISSING") elif attestation == "bilateral_signed": problems.append("bilateral_signed but party_b signature ABSENT") # Provider-HMAC labeling (non-negotiable). if attestation == "provider_hmac_bridge": print(" WARNING: Counterparty did not sign.") print(" RANKIGI witnessed provider HMAC only.") note = c.get("attestation_note") if note: print(f" Note: {note}") # Five proof layers. layers = c.get("five_layers") if not isinstance(layers, dict) and isinstance(transcript, dict): layers = { "action": "bound" if transcript.get("action_event_hash") else "absent", "policy": "bound" if transcript.get("policy_eval_hash") else "absent", "authorization": "bound" if transcript.get("authorization_hash") else "absent", "model_state": "bound" if transcript.get("model_state_hash") else "absent", "data_lineage": "bound" if transcript.get("data_lineage_hash") else "absent", } if isinstance(layers, dict): order = ["action", "policy", "authorization", "model_state", "data_lineage"] print(" Layers: " + " ".join("%s=%s" % (k, layers.get(k, "absent")) for k in order)) for ok in checks: print(f" {ok}") for p in problems: print(f" PROBLEM: {p}") if c.get("anchored_at"): idx = c.get("rekor_log_index") print(f" Anchored: YES (Rekor #{idx})") else: print(" Anchored: PENDING") # Per-commitment verdict. hard = any( ("MISMATCH" in p) or ("INVALID" in p) or ("MISSING" in p) for p in problems ) if hard: verdict = "UNVERIFIED" elif problems: verdict = "PARTIAL" elif not CRYPTO_AVAILABLE: verdict = "PARTIAL (hashes only)" elif not final_root: verdict = "PROPOSAL VERIFIED (not yet accepted)" else: verdict = "VERIFIED" print(f" => {verdict}") # === END AGENT-G-BLOCK === # === AGENT-INTEGRITY-BLOCK === def verify_integrity_object(path): with open(path) as f: doc = json.load(f) if doc.get('schema') != 'rankigi.integrity.v1': print("Not an integrity object") return print("Chain: " + str(doc['chain_id'])) print("Events: " + str(doc['event_count'])) print("Sealed: " + str(doc['sealed_at'])) recomputed = hashlib.sha256(json.dumps( {k: v for k, v in doc.items() if k != 'integrity_hash'}, sort_keys=True, separators=(',', ':') ).encode()).hexdigest() if recomputed == doc['integrity_hash']: print("Integrity hash: VERIFIED (bundle-self-attested; no external trust anchor)") else: print("Integrity hash: FAILED") if doc.get('rekor_log_index'): print("Rekor: #" + str(doc['rekor_log_index'])) else: print("Rekor: not anchored") # === END AGENT-INTEGRITY-BLOCK === def report_v4_binding(events): """When any event carries v4 binding fields, print the fingerprint and signature_hash previews so the operator can see what was checked.""" bound = [ e for e in events if isinstance(e, dict) and (e.get("passport_fingerprint") or e.get("signature_hash")) ] if not bound: return sample = bound[0] fp = sample.get("passport_fingerprint") or "" sh = sample.get("signature_hash") or "" if fp: print("Passport fingerprint: %s..." % fp[:16]) if sh: print("Signature hash: %s..." % sh[:16]) print("v4-bound events: %d of %d" % (len(bound), len(events))) if len(bound) < len(events): print("Note: events recorded before v4 SDK was installed are v3-bound.") print("All new events use v4 passport binding.") closure_event = next( ( e for e in events if isinstance(e, dict) and event_closure_binding(e) in ("server_attested", "v2_fallback") ), None, ) if closure_event is not None: binding = event_closure_binding(closure_event) if binding == "server_attested": print("Closure attestation: server-attested. RANKIGI sealed the closure event;") print(" event hashes verified by passport fingerprint.") elif binding == "v2_fallback": print("Closure attestation: v2 fallback (no passport binding)") def verify_passport_binding(envelope, events): """FIX A. Cross-check the envelope's declared passport fingerprint against every v4 event's passport_fingerprint. Returns one of 'verified', 'mismatch', 'no_binding' and prints a verdict line. """ envelope_fp = envelope.get("passport_key_fingerprint") if isinstance(envelope, dict) else None v4_events = [ e for e in events if isinstance(e, dict) and e.get("passport_fingerprint") ] if not v4_events and not envelope_fp: return "no_binding" if not v4_events: print("Passport binding: NO V4 EVENTS") print("Envelope declares passport %s... but no events carry v4 binding." % (envelope_fp or "")[:16]) return "mismatch" if not envelope_fp: sample = v4_events[0].get("passport_fingerprint") or "" print("Passport binding: ENVELOPE MISSING FINGERPRINT") print("Events declare passport %s... but the envelope has no key." % sample[:16]) return "mismatch" distinct = sorted({(e.get("passport_fingerprint") or "") for e in v4_events}) if len(distinct) != 1: print("Passport binding: MIXED FINGERPRINTS ACROSS EVENTS") for fp in distinct: print(" fingerprint: %s..." % fp[:16]) return "mismatch" event_fp = distinct[0] if event_fp.lower() != envelope_fp.lower(): print("Passport binding: MISMATCH") print("Envelope: %s..." % envelope_fp[:16]) print("Events: %s..." % event_fp[:16]) return "mismatch" print("Passport fingerprint binding: VERIFIED (bundle-self-reported; not externally pinned)") print("Passport: %s..." % envelope_fp[:16]) print("Events bound to envelope passport: %d of %d" % (len(v4_events), len(events))) return "verified" def fingerprint_pem(pem_text, hash_version=None): """SHA-256 hex fingerprint of a public key. Gated on hash_version: - hash_version >= 4: strip PEM armor, base64-decode body to DER bytes, SHA-256 the DER. Matches the v4 server-side fingerprint over SubjectPublicKeyInfo DER. - hash_version < 4 or None: legacy behavior -- SHA-256 the PEM string bytes verbatim. Preserves compatibility with pre-v4 bundles. """ if isinstance(pem_text, bytes): raw = pem_text text = pem_text.decode("utf-8", errors="replace") else: raw = pem_text.encode("utf-8") text = pem_text try: hv = int(hash_version) if hash_version is not None else 0 except (TypeError, ValueError): hv = 0 if hv >= 4: # Strip PEM armor and whitespace, base64-decode body to DER. body_lines = [] in_body = False for line in text.splitlines(): stripped = line.strip() if stripped.startswith("-----BEGIN"): in_body = True continue if stripped.startswith("-----END"): in_body = False continue if in_body: body_lines.append(stripped) body_b64 = "".join(body_lines) try: der = base64.b64decode(body_b64) return hashlib.sha256(der).hexdigest() except Exception: # Fall back to legacy PEM-string hash so callers always get a # deterministic string rather than an exception. return hashlib.sha256(raw).hexdigest() return hashlib.sha256(raw).hexdigest() def load_pinned_pubkey(path): """Load a pinned PEM public key from disk. Returns (pem_text, public_key) or raises on any error; the caller is expected to handle failure. """ with open(path, "rb") as f: pem_bytes = f.read() if not CRYPTO_AVAILABLE: return (pem_bytes.decode("utf-8", errors="replace"), None) public_key = load_pem_public_key(pem_bytes) return (pem_bytes.decode("utf-8", errors="replace"), public_key) def fetch_pubkey_from_server(agent_id, base_url=DEFAULT_BASE_URL): """Fetch a root-attested public key for agent_id and verify the attestation against the pinned RANKIGI root key. Endpoint: GET {base_url}/api/proof/pubkey?agent_id= Response (Phase 3 week 3): { agent_id, public_key_pem, key_attestation_signature, key_attestation_signer_kid, key_attestation_alg, issued_at, revoked_at } Returns (pem_text, public_key, attestation_status, attested_org_id) where attestation_status is one of: "attested" signature verified against root, revoked_at null "revoked" signature verified but revoked_at is set "unattested" endpoint returned no attestation fields (legacy) "bad-sig" signature failed to verify against root "unreachable" network failure or non-200 "offline" --offline mode; no network call attempted attested_org_id is the org_id covered by the attestation tuple (or None for legacy/unattested responses). """ if not agent_id: return None, None, "unreachable", None try: url = "%s/api/proof/pubkey?agent_id=%s" % ( base_url.rstrip("/"), urllib.parse.quote(str(agent_id), safe=""), ) req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=5) as resp: raw = resp.read() data = json.loads(raw.decode("utf-8")) except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError): return None, None, "unreachable", None except Exception: return None, None, "unreachable", None if not isinstance(data, dict): return None, None, "unreachable", None # The route returns the canonical apiOk envelope: # { ok: true, data: { ... }, request_id: "..." }. Unwrap it. Fall back to # a flat shape for forward/backward compatibility with any future or # legacy server that returns fields at the top level. inner = data.get("data") if isinstance(data.get("data"), dict) else data pem = inner.get("public_key_pem") if not isinstance(pem, str) or not pem.strip(): return None, None, "unreachable", None attested_org_id = inner.get("org_id") if not CRYPTO_AVAILABLE: return pem, None, "unattested", attested_org_id # agent_passports.public_key may be stored as either a full PEM (with # BEGIN/END markers) or a bare SPKI base64 string (compact form). Detect # the compact form and wrap it before handing to load_pem_public_key. if "-----BEGIN" not in pem: wrapped = ( "-----BEGIN PUBLIC KEY-----\n" + pem.strip() + "\n-----END PUBLIC KEY-----\n" ) else: wrapped = pem try: public_key = load_pem_public_key(wrapped.encode("utf-8")) except Exception: return None, None, "unreachable", None # Preserve the original bytes the server returned in `pem` so the # canonical attestation input below matches what the server signed. # The wrapped form is only used to parse the key into KeyObject form. sig_b64 = inner.get("key_attestation_signature") kid = inner.get("key_attestation_signer_kid") revoked_at = inner.get("revoked_at") issued_at = inner.get("issued_at") if not sig_b64 or kid != RANKIGI_ROOT_KEY_ID: return pem, public_key, "unattested", attested_org_id # Rebuild the canonical input the route signed. Route key order in # src/app/api/proof/pubkey/route.ts: # canonicalJson({ agent_id, org_id, public_key_pem, issued_at, revoked_at }) # Our _canonical() sorts keys lex, matching canonicalJson's behaviour. canonical = _canonical({ "agent_id": str(agent_id), "org_id": attested_org_id, "public_key_pem": pem, "issued_at": issued_at, "revoked_at": revoked_at, }) try: root_key = load_pem_public_key(RANKIGI_ROOT_PUBKEY_PEM) root_key.verify(base64.b64decode(sig_b64), canonical.encode("utf-8")) except InvalidSignature: return pem, public_key, "bad-sig", attested_org_id except Exception: return pem, public_key, "bad-sig", attested_org_id if revoked_at: return pem, public_key, "revoked", attested_org_id return pem, public_key, "attested", attested_org_id def _envelope_declares_v4(doc): """True if the bundle envelope declares hash_version >= 4 via any of: - rankigi_schema is a dict with hash_version field >= 4 - top-level hash_version >= 4 - passport_key_fingerprint present (v4 passport binding marker) """ if not isinstance(doc, dict): return False schema = doc.get("rankigi_schema") if isinstance(schema, dict): hv = schema.get("hash_version") try: if hv is not None and int(hv) >= 4: return True except (TypeError, ValueError): pass top_hv = doc.get("hash_version") try: if top_hv is not None and int(top_hv) >= 4: return True except (TypeError, ValueError): pass if doc.get("passport_key_fingerprint"): return True return False def verify_closure_envelope_signature(doc): """Verify the bundle's closure_signature against the pinned RANKIGI root. Returns one of: "verified", "failed", "warning", "absent". Prints the appropriate status line. Canonical input: canonical_json(envelope) with the three signature fields (closure_signature, closure_signature_alg, closure_signing_key_id) and the warning field (closure_signature_warning) removed. Matches the server-side construction in src/app/api/chains/[chainId]/closure-export/route.ts. """ if not isinstance(doc, dict): print("Closure signature: not present (pre-Phase-3 bundle)") return "absent" sig_b64 = doc.get("closure_signature") warning = doc.get("closure_signature_warning") envelope_is_v4 = _envelope_declares_v4(doc) if sig_b64 is None and not warning and "closure_signature" not in doc: if envelope_is_v4: print("Closure signature: FAILED (v4 envelope missing closure_signature)") return "failed" print("Closure signature: not present (pre-Phase-3 bundle)") return "absent" if sig_b64 is None and warning: if envelope_is_v4: print("Closure signature: FAILED (v4 envelope unsigned: %s)" % warning) return "failed" print("Closure signature: NOT SIGNED (%s)" % warning) return "warning" if sig_b64 is None: if envelope_is_v4: print("Closure signature: FAILED (v4 envelope has no closure_signature)") return "failed" print("Closure signature: NOT SIGNED") return "warning" alg = doc.get("closure_signature_alg") kid = doc.get("closure_signing_key_id") if alg != "Ed25519" or kid != RANKIGI_ROOT_KEY_ID: print("Closure signature: NOT VERIFIED (unexpected alg=%r kid=%r)" % (alg, kid)) return "failed" if not CRYPTO_AVAILABLE: print("Closure signature: NOT VERIFIED (cryptography library missing)") return "failed" # Strip signature + warning fields and re-canonicalize. redacted = { k: v for k, v in doc.items() if k not in ( "closure_signature", "closure_signature_alg", "closure_signing_key_id", "closure_signature_warning", # closure_export_id / terminal_status / terminal_status_computed_at # are appended to the HTTP response but are not part of the # envelope canonical input (matches the content_hash strip list). "closure_export_id", "terminal_status", "terminal_status_computed_at", ) } canonical = _canonical(redacted) try: root_key = load_pem_public_key(RANKIGI_ROOT_PUBKEY_PEM) root_key.verify(base64.b64decode(sig_b64), canonical.encode("utf-8")) except InvalidSignature: print("Closure signature: NOT VERIFIED") return "failed" except Exception as e: print("Closure signature: NOT VERIFIED (%s)" % type(e).__name__) return "failed" print("Closure signature: VERIFIED (against pinned rankigi-root key)") return "verified" def verify_proof_extensions(event, proof_extensions, proof_extensions_hash): """Verify the proof_extensions_hash for a v5 event. Phase B complete: 'hash_verified' means the stored proof_extensions object is intact (its canonical-JSON hash recomputes to proof_extensions_hash). The binding of proof_extensions_hash INTO the chain is verified separately by the per-event hash recomputation -- compute_event_hash_v5_canonical folds the same proof_extensions_hash in as the 14th preimage key, so a v5 event whose event hash verifies AND whose proof_extensions hash_verified has its full policy/authorization/ lineage posture cryptographically committed. """ if proof_extensions is None and proof_extensions_hash is None: # v4 event in a v5-schema bundle -- acceptable. return { "status": "not_applicable", "reason": "v4 event has no proof_extensions", } if proof_extensions is None and proof_extensions_hash is not None: # Hash present but the object missing from the bundle -- incomplete. return { "status": "incomplete", "reason": "proof_extensions_hash present but proof_extensions missing from bundle", } if proof_extensions is not None and proof_extensions_hash is None: return { "status": "incomplete", "reason": "proof_extensions present but proof_extensions_hash missing", } # Both present -- verify the hash. try: recomputed = hashlib.sha256( _canonical(proof_extensions).encode("utf-8") ).hexdigest() if recomputed == proof_extensions_hash: return { "status": "hash_verified", "reason": "proof_extensions_hash matches recomputed value", "pe_version": proof_extensions.get("pe_version"), "policy_eval_hash": proof_extensions.get("policy_eval_hash"), "authorization_scope_check": proof_extensions.get("authorization_scope_check"), "authorization_issues": proof_extensions.get("authorization_issues", []), "data_receipt_ids_hash": proof_extensions.get("data_receipt_ids_hash"), } return { "status": "hash_mismatch", "reason": "expected %s, got %s" % (proof_extensions_hash, recomputed), "verdict_impact": "COMPROMISED", } except Exception as e: return { "status": "error", "reason": str(e), } def report_proof_extensions(events, schema_version): """Print the --- Proof Extensions --- section and return the number of hash mismatches (each one is COMPROMISED-grade). The section renders for schema 1.6 bundles, and for any bundle whose events actually carry a non-null proof_extensions or proof_extensions_hash. Stage 3 bundles (schema 1.5) carry proof_extensions: null on every event; those stay silent so 1.5 output is unchanged. """ relevant = str(schema_version) == "1.6" or any( isinstance(ev, dict) and ( ev.get("proof_extensions") is not None or ev.get("proof_extensions_hash") is not None ) for ev in events ) if not relevant: return 0 print() print("--- Proof Extensions ---") mismatches = 0 for ev in events: if not isinstance(ev, dict): continue pe = ev.get("proof_extensions") peh = ev.get("proof_extensions_hash") result = verify_proof_extensions(ev, pe, peh) status = result.get("status") ci = ev.get("chain_index") if status == "hash_mismatch": print("Event %s: proof_extensions COMPROMISED" % ci) print(" %s" % result.get("reason")) mismatches += 1 elif status == "hash_verified": print("Event %s: proof_extensions OK" % ci) print(" scope_check: %s" % result.get("authorization_scope_check")) issues = result.get("authorization_issues") or [] if issues: print(" authorization issues: %s" % issues) # Stage 9 (9C): RANKIGI countersignature over the # proof_extensions_hash hex string, verified against the pinned # witness/root keys. ABSENT is acceptable (legacy v5 events # predate it) and never downgrades the verdict; INVALID is # tamper evidence (COMPROMISED-grade, counted like a mismatch). pe_sig = ev.get("proof_extensions_signature") if isinstance(pe_sig, str) and pe_sig: try: sig_bytes = base64.b64decode(pe_sig) if _verify_witness_signature(sig_bytes, peh): print(" proof_extensions signature: VERIFIED (RANKIGI witnessed)") else: print(" proof_extensions signature: INVALID") mismatches += 1 except Exception as e: print(" proof_extensions signature: INVALID (%s)" % type(e).__name__) mismatches += 1 else: print(" proof_extensions signature: NOT PRESENT") elif status in ("incomplete", "not_applicable"): print("Event %s: proof_extensions %s" % (ci, status)) if status == "incomplete": print(" %s" % result.get("reason")) elif status == "error": print("Event %s: proof_extensions ERROR (%s)" % (ci, result.get("reason"))) return mismatches def check_nonce_uniqueness(events): """In-bundle nonce uniqueness check. Catches replay regardless of server-side state. Returns (ok, reason). """ seen = {} for i, ev in enumerate(events): if not isinstance(ev, dict): continue n = ev.get("nonce") if not n: continue if n in seen: return False, "Duplicate nonce %r at events[%d] and events[%d]" % (n, seen[n], i) seen[n] = i return True, None def verify_event_signatures(envelope, events, pinned_key=None, network_mode=False, base_url=DEFAULT_BASE_URL, offline=False): """Optional Ed25519 signature verification. When pinned_key is provided, it is used as the trust anchor (the envelope's public key is ignored for verification). Otherwise the envelope's self-reported passport_public_key_pem is used. If neither is present, a network fetch is attempted against the RANKIGI proof endpoint using the agent_id from the first signed event. All three fallback paths degrade gracefully: missing key prints a WARNING and skips signature verification; missing cryptography library prints a WARNING and skips; network failure prints a WARNING and skips. Signing object (portable regime, matching gate.ts and sdk-node-v4): {action, agent_id, chain_id, nonce, occurred_at, payload, tool} Keys sorted lexicographically; null included; undefined dropped. This is exactly the canonicalJson() input the SDK signs client-side. Fields NOT covered by the signature: signature, hash, event_hash, prev_hash, org_id, severity, passport_fingerprint, signature_hash, chain_index, server_received_at, and all envelope-level fields. Returns a tuple (status, verified_count, failed_idx, signed_total, server_attested_count, key_source). status is one of: 'no_key' no usable public key (envelope, pin, and fetch all absent) 'skipped' cryptography library not installed 'no_signatures' no events carried a signature field 'passed' all signed events verified successfully 'failed' an InvalidSignature was encountered at failed_idx server_attested_count counts chain_closure events that legitimately lack a signature because the closure was server-attested (RANKIGI does not hold the passport private key). Callers use it to relax envelope-level 'no_signatures' hard-fails when every unsigned event is a server-attested closure. Per-event output: "Signature verified for event N" -- verification passed "Signature FAILED for event N" -- verification failed (also sets status=failed) """ global _WITNESS_UNSIGNED_COUNT if not CRYPTO_AVAILABLE: print("WARNING: Ed25519 signature verification skipped.") print(" The 'cryptography' package is not installed.") print(" Install it with: pip install cryptography") print(" Hash-chain integrity verification is unaffected.") return ("skipped", 0, 0, 0, 0, None) # If the bundle claims a v4 passport binding (envelope carries # passport_key_fingerprint), every event must carry signature + nonce. # Missing signatures on a claimed-v4 bundle are treated as tampering and # hard-fail rather than silently skipping verification. claims_v4 = bool( isinstance(envelope, dict) and envelope.get("passport_key_fingerprint") ) # Resolve the public key. Trust hierarchy (highest first): # A. --pubkey PATH (pinned). key_source A # B. --pubkey-from-network (root-attested fetch). key_source B # C. passport_public_key_pem self-reported in bundle. key_source C # Only sources A and B can carry the verdict to VERIFIED. Source C is # SELF-ATTESTED at best. public_key = pinned_key key_source = "--pubkey (pinned)" if pinned_key is not None else None attested_org_id = None if public_key is None and network_mode: agent_id_for_fetch = None for ev in events: if isinstance(ev, dict) and ev.get("agent_id"): agent_id_for_fetch = ev["agent_id"] break if agent_id_for_fetch: if offline: print("Pubkey network fetch: SKIPPED (offline mode)") else: print("Fetching attested key from %s/api/proof/pubkey ..." % base_url) fetched_pem, fetched_key, attestation, attested_org_id = fetch_pubkey_from_server( agent_id_for_fetch, base_url=base_url, ) if attestation == "attested" and fetched_key is not None: public_key = fetched_key key_source = "network (root-attested)" print("Network key attestation: VERIFIED (root)") elif attestation == "revoked": print("Network key attestation: FAILED (key revoked)") return ("no_key", 0, 0, 0, 0, None) elif attestation == "bad-sig": print("Network key attestation: FAILED (signature did not verify against root)") # Fall through to bundle key (source C) below. elif attestation == "unattested": print("Network key attestation: ABSENT (endpoint returned no signature)") else: print("Network key attestation: UNREACHABLE") else: print("Network key fetch skipped (no agent_id in events).") # Fix 2: enforce org_id binding when network-attested key was accepted. # The attestation now covers { agent_id, org_id, public_key_pem, issued_at, # revoked_at }. Every signed event must claim the same org_id; otherwise # an attacker who registered org X could forge events bearing victim # org Y while presenting their own root-attested pubkey. if key_source == "network (root-attested)" and attested_org_id is not None: for ev in events: if not isinstance(ev, dict): continue ev_org = ev.get("org_id") if ev_org is None: continue if str(ev_org) != str(attested_org_id): print( "org_id mismatch: event claims org %s but attestation is for org %s" % (ev_org, attested_org_id) ) sys.exit(1) if public_key is None: pem = envelope.get("passport_public_key_pem") if isinstance(envelope, dict) else None if pem: try: public_key = load_pem_public_key( pem.encode("utf-8") if isinstance(pem, str) else pem ) key_source = "bundle (self-reported)" except Exception: public_key = None if public_key is None: return ("no_key", 0, 0, 0, 0, None) verified_count = 0 signed_seen = 0 server_attested_count = 0 first_failed_idx = -1 for idx, ev in enumerate(events): if not isinstance(ev, dict): continue sig_b64 = ev.get("signature") nonce = ev.get("nonce") if not sig_b64 or not nonce: # L1-H2: an internal event carrying the witness marker but NO # signature means the server's witness key was not configured at # write time. Not tamper evidence (nothing was forged), but an # unsigned server-originated event cannot support VERIFIED: # count it and cap the final verdict at SELF-ATTESTED. Checked # before the claims_v4 gate -- the marker lives on the event # itself, independent of envelope-level v4 claims. if _event_witness_marker(ev): print( "Event %d: RANKIGI-internal event lacks a witness signature; verdict capped at SELF-ATTESTED." % idx ) _WITNESS_UNSIGNED_COUNT += 1 continue if claims_v4: action = ev.get("action") closure_binding = event_closure_binding(ev) if action == "chain_closure" and closure_binding == "server_attested": print("Closure event %d: server-attested by RANKIGI (no agent signature)." % idx) server_attested_count += 1 continue if action == "chain_closure" and not closure_binding: print("ERROR: Closure event %d lacks both signature and closure_binding label." % idx) sys.exit(1) # System-attested heuristic: the event has no agent signature # in the bundle, but the server attested it at ingest. We # recognise four signals: # - actor field marked "__system__" # - action explicitly server_attested # - passport_fingerprint missing or null # - signature_hash present (server recorded the verified # signature digest at ingest; raw signature not retained # in this bundle, common under hash-only retention) # The closure envelope itself is root-signed by RANKIGI, so # trusting the server's at-ingest attestation chains to the # pinned root key. actor = ev.get("actor") raw_hv = ev.get("hash_version") try: hv = int(raw_hv) if raw_hv is not None else 3 except (TypeError, ValueError): hv = 3 if hv >= 4: # Strict v4 allowlist: only these unsigned actions may be # accepted as system-attested. chain_closure is handled # earlier via the closure_binding check; any other unsigned # v4 event is UNVERIFIED, not system-attested. SYSTEM_ATTEST_ACTIONS = { "agent_offline_start", "agent_offline_end", "system_annotation", } system_attested = ( action in SYSTEM_ATTEST_ACTIONS and actor == "__system__" and closure_binding == "server_attested" ) else: # Fix 6: v3 tightened to match v4. Previous code accepted # any of (actor=__system__, action=server_attested, missing # passport_fp, sig_hash present) as system-attested, which # let attackers strip envelope v4 markers and forge entire # chains of unsigned "__system__" events. Now require all # three v4 signals: allowlisted action AND actor=__system__ # AND closure_binding=server_attested. v3 events failing # this gate fall through to normal signature verification # below; events with no signature increment no_signatures. SYSTEM_ATTEST_ACTIONS = { "agent_offline_start", "agent_offline_end", "system_annotation", } system_attested = ( action in SYSTEM_ATTEST_ACTIONS and actor == "__system__" and closure_binding == "server_attested" ) if system_attested: print( "Event %d: server-attested by RANKIGI (no passport signature)." % idx ) server_attested_count += 1 continue if hv >= 4: # v4: any unsigned event outside the system-attest allowlist # is UNVERIFIED. Record as a failed signature so the overall # verdict reflects the gap; do not silently accept. print( "Event %d: UNVERIFIED (v4 event lacks signature and is not in the system-attest allowlist)." % idx ) if first_failed_idx < 0: first_failed_idx = idx signed_seen += 1 continue print("SIGNATURE MISSING: Event %d claims v4 binding but has no signature or nonce. This bundle may have been tampered with." % idx) sys.exit(1) continue signed_seen += 1 # Rebuild the exact portable canonical JSON the SDK signed. # Fields: action, agent_id, chain_id, nonce, occurred_at, payload, tool. # Matches canonicalJson() in sdk-node-v4/src/index.ts lines 641-649 # and the portableInput in src/lib/ingest/gate.ts lines 652-659. # null values are included; undefined is absent (Python has no undefined # so all keys are always present). Keys are sorted lexicographically by # _canonical(), matching canonical-json.ts Object.keys().sort(). # payload is taken from payload_canonical when present (verbatim, the # exact bytes the SDK signed) so signature verification still works # after the raw payload is nulled at rest. Older bundles that carry # only payload fall back to canonicalizing it. A missing payload yields # the literal "null" fragment rather than raising. # Phase B check-12 fix: the SDK signs occurred_at in JS # toISOString() ms-Z form, but closure bundles carry the PostgREST # "+00:00" microsecond serialization of events.occurred_at. The hash # recompute path already normalizes (normalize_occurred_at); the # signing path must normalize identically or every honest signature # in a closure bundle fails. Fall back to the raw string when the # value cannot be parsed (matches the SDK-sent bytes for legacy # bundles that carried it verbatim). raw_occurred = ev.get("occurred_at") try: signing_occurred = normalize_occurred_at(raw_occurred) except (ValueError, TypeError): signing_occurred = raw_occurred signing_obj = { "action": ev.get("action"), "agent_id": ev.get("agent_id"), "chain_id": ev.get("chain_id"), "nonce": nonce, "occurred_at": signing_occurred, "payload": resolve_canonical_payload(ev), "tool": ev.get("tool"), } signing_input = _canonical(signing_obj) try: sig = base64.b64decode(sig_b64) except Exception: print("Signature FAILED for event %d (base64 decode error)" % idx) if first_failed_idx < 0: first_failed_idx = idx # Continue checking remaining events before returning. continue # L1-H2: witness-signed internal events verify against the pinned # RANKIGI keys (root preferred, legacy witness fallback) instead of # the agent passport key -- server events have no passport key. The # marker lives INSIDE the signed payload, so an attacker cannot strip # or add it without invalidating the signature. An invalid witness # signature is treated exactly like an invalid passport signature: # status 'failed' -> COMPROMISED. if _event_witness_marker(ev): if _verify_witness_signature(sig, signing_input): print( "Event %d: witness-signed by RANKIGI (verified against pinned RANKIGI key)." % idx ) verified_count += 1 else: print( "Signature FAILED for event %d (RANKIGI witness signature invalid)" % idx ) if first_failed_idx < 0: first_failed_idx = idx continue try: public_key.verify(sig, signing_input.encode("utf-8")) print("Signature verified for event %d" % idx) verified_count += 1 except InvalidSignature: print("Signature FAILED for event %d" % idx) if first_failed_idx < 0: first_failed_idx = idx except Exception as exc: print("Signature FAILED for event %d (%s)" % (idx, type(exc).__name__)) if first_failed_idx < 0: first_failed_idx = idx if signed_seen == 0: return ("no_signatures", 0, 0, 0, server_attested_count, key_source) # Summary line: shows key trust source alongside pass/fail count so the # overall PASS/FAIL verdict accounts for actual signature verification. if first_failed_idx >= 0: print("Signature summary: %d of %d verified, %d FAILED [key: %s]" % (verified_count, signed_seen, signed_seen - verified_count, key_source or "unknown")) return ("failed", verified_count, first_failed_idx, signed_seen, server_attested_count, key_source) print("Signature summary: %d of %d verified [key: %s]" % (verified_count, signed_seen, key_source or "unknown")) return ("passed", verified_count, 0, signed_seen, server_attested_count, key_source) def _key_source_suffix(key_source): """Return a bracketed trust-source label for the Ed25519 VERIFIED line. Maps key_source strings produced by verify_event_signatures() to one of: [pinned] -- supplied via --pubkey [network-attested] -- fetched + root-attested via --pubkey-from-network [bundle-self-reported] -- extracted from passport_public_key_pem in bundle Anything else returns an empty string. """ if not key_source: return "" if key_source == "--pubkey (pinned)": return " [pinned]" if key_source == "network (root-attested)": return " [network-attested]" if key_source == "bundle (self-reported)": return " [bundle-self-reported]" return "" # ── output banner / disclaimer ──────────────────────────────────────────── BANNER = "=== RANKIGI CHAIN VERIFICATION ===" DISCLAIMER = ( "DISCLAIMER: This verifier checks hash chain integrity, passport fingerprint\n" "binding, Ed25519 event-signature validity, RFC 3161 TSA anchoring, Rekor\n" "inclusion, and CCAP receipts when those fields are present. It does not\n" "verify agent conduct or regulatory compliance. The closure event is\n" "server-attested (RANKIGI-signed), not passport-signed, by design.\n" "VERIFIED requires a pinned (--pubkey) or root-attested\n" "(--pubkey-from-network) key source. Bundle-self-reported keys collapse\n" "the verdict to SELF-ATTESTED (UNVERIFIED)." ) PUBKEY_NOTICE = ( "Note: passport public key is self-reported in this bundle.\n" "For maximum assurance use: python3 verify.py bundle.json --pubkey path/to/key.pem\n" "Download from: https://rankigi.com/.well-known/rankigi-public-key.json" ) def print_banner(pinned_via_flag): print(BANNER) print() print(DISCLAIMER) print() if not pinned_via_flag: print(PUBKEY_NOTICE) print() # ── main ────────────────────────────────────────────────────────────────── def verify_closure_envelope(doc, check_rekor=False, pinned_pem=None, pinned_key=None, force_ccap=False, force_xact=False, network_mode=False, base_url=DEFAULT_BASE_URL, ccap_keys=None, offline=False, rekor_requested=False): """Closure envelope verifier (rankigi_schema 1.3+). Performs three checks in order: 1. content_hash self-seal: recompute SHA-256 of canonical JSON of the envelope minus content_hash and confirm it matches. 2. Per-event hash recomputation over events[] using the v1/v2 dispatch already shipped for raw events. 3. Computed hash list equals event_hashes[] in order. Per-event records in the closure envelope use the v2 field names (action, agent_id, chain_id, occurred_at, org_id, payload, prev_hash, severity, tool) plus hash and chain_index. event_hash is taken from the 'hash' field for closure records. """ envelope_fp = doc.get("passport_key_fingerprint") if isinstance(doc, dict) else None if pinned_pem is not None: envelope_hv = doc.get("hash_version") if isinstance(doc, dict) else None if envelope_hv is None and isinstance(doc, dict): schema = doc.get("rankigi_schema") if isinstance(schema, dict): envelope_hv = schema.get("hash_version") pinned_fp = fingerprint_pem(pinned_pem, hash_version=envelope_hv) if envelope_fp and pinned_fp.lower() != str(envelope_fp).lower(): print("--pubkey SPKI fingerprint does not match passport_key_fingerprint in bundle") print("Pinned key fingerprint: %s..." % pinned_fp[:16]) print("Envelope key fingerprint: %s..." % str(envelope_fp)[:16]) sys.exit(1) if not envelope_fp: # No envelope fingerprint to compare against. Fall through; # the closure_signature check below will gate the verdict. pass # Fix 3: --pubkey requires the bundle to carry a closure_signature. # Without one, the pinned key is self-referential -- it only proves # the events were signed by the same key the user supplied, with no # external chain of trust to RANKIGI root. if isinstance(doc, dict) and not doc.get("closure_signature"): print("--pubkey was supplied but bundle has no closure_signature.") print("A valid closure_signature signed by RANKIGI root is required to reach VERIFIED.") print("Without it the supplied key is self-referential and proves nothing externally.") print("=== RESULT: SELF-ATTESTED ===") sys.exit(1) # Stage 4 (4B): 1.6 added BEFORE any 1.6 bundle exists (verifiers update # before producers). The 1.6 handler is the 1.5 handler plus the # proof_extensions section below; v5 event rows report UNVERIFIABLE until # the Phase B preimage spec lands in compute_event_hash_v5_canonical. # Unknown future versions still exit UNSUPPORTED -- never a false # VERIFIED or COMPROMISED. schema_version = doc.get("rankigi_schema") if schema_version is not None and str(schema_version) not in ("1.4", "1.5", "1.6"): print("SCHEMA VERSION UNSUPPORTED") print("Bundle declares rankigi_schema=%s; this verifier supports 1.4, 1.5, and 1.6." % schema_version) return 1 # Key source preview. The final source is determined inside # verify_event_signatures (it knows whether network attestation succeeded), # but we surface the operator's intent up front. print("--- Key Source ---") if pinned_pem is not None: print("Resolved: --pubkey (pinned, highest trust)") elif network_mode: print("Resolved: --pubkey-from-network (will require root attestation)") print("Base URL: %s" % base_url) elif isinstance(doc, dict) and doc.get("passport_public_key_pem"): print("Resolved: bundle (self-reported; VERDICT will be SELF-ATTESTED)") print("Re-run with --pubkey or --pubkey-from-network for VERIFIED.") else: print("Resolved: NONE (no pinned, no network, no bundle key)") print() print("--- Hash Chain ---") expected_content_hash = doc.get("content_hash") if not isinstance(expected_content_hash, str): print("Hash chain integrity: COMPROMISED") print("Missing content_hash on envelope") return 1 # The server computes content_hash over draftWithContributions BEFORE # the closure-signature fields are attached and BEFORE closure_export_id # is appended to the HTTP response. To recompute, strip: # content_hash itself # closure_signature, closure_signature_alg, closure_signing_key_id # closure_signature_warning (set when signing was skipped) # closure_export_id (added to the HTTP response after persistence) # terminal_status, terminal_status_computed_at (P0 audit 2026-06-10: # also appended to the HTTP response AFTER content_hash; leaving them # in branded every fresh bundle COMPROMISED with a false # content_hash mismatch) envelope_without_seal = { k: v for k, v in doc.items() if k not in ( "content_hash", "closure_signature", "closure_signature_alg", "closure_signing_key_id", "closure_signature_warning", "closure_export_id", "terminal_status", "terminal_status_computed_at", ) } computed_content_hash = hashlib.sha256( _canonical(envelope_without_seal).encode("utf-8") ).hexdigest() if computed_content_hash != expected_content_hash: print("Hash chain integrity: COMPROMISED") print("content_hash mismatch") print("Expected: %s..." % expected_content_hash[:16]) print("Computed: %s..." % computed_content_hash[:16]) return 1 events = doc.get("events") event_hashes = doc.get("event_hashes") if not isinstance(events, list) or not isinstance(event_hashes, list): print("Hash chain integrity: COMPROMISED") print("Envelope missing events[] or event_hashes[]") return 1 if len(event_hashes) != len(events): print("EVENT COUNT MISMATCH: envelope declares %d hashes but bundle contains %d events" % (len(event_hashes), len(events))) sys.exit(1) # FIX 2: Block v4-to-v3 downgrade. If the envelope declares v4 via any # of (rankigi_schema.hash_version, top-level hash_version, # passport_key_fingerprint), every event must claim v4 or higher; # otherwise the bundle is a downgrade attack. has_v4_binding = _envelope_declares_v4(doc) if has_v4_binding: for n, ev in enumerate(events): if not isinstance(ev, dict): continue raw_v = ev.get("hash_version") try: v = int(raw_v) if raw_v is not None else 3 except (TypeError, ValueError): v = 3 if v < 4: print("VERSION DOWNGRADE DETECTED: envelope claims v4 binding but event %d is v%d" % (n, v)) sys.exit(1) # FIX B (defense in depth): For v4 bundles, any chain_closure event must # carry payload.closure_binding == "server_attested". Server-attested is # the only sanctioned way a v4 closure can lack an Ed25519 signature; any # other closure_binding value (or a missing label) on a v4 chain is a # forged or downgraded closure attempt and must hard-fail. Non-v4 bundles # are unaffected. if has_v4_binding: for n, ev in enumerate(events): if not isinstance(ev, dict): continue if ev.get("action") != "chain_closure": continue cb = event_closure_binding(ev) if cb != "server_attested": print("CLOSURE BINDING INVALID: v4 bundle chain_closure at event %d has closure_binding=%r (expected 'server_attested')" % (n, cb)) sys.exit(1) computed_hashes = [] encrypted_unverified = 0 unknown_version_unverifiable = 0 for i, ev in enumerate(events): # Closure records carry the per-event hash under 'hash'; map it to # 'event_hash' so the shared event_hash() dispatch can consume it. normalized = dict(ev) if "event_hash" not in normalized and "hash" in normalized: normalized["event_hash"] = normalized["hash"] if has_v4_binding: # v4 envelope: never let an event silently fall back to v3 hashing. raw_v = normalized.get("hash_version") try: vv = int(raw_v) if raw_v is not None else 0 except (TypeError, ValueError): vv = 0 if vv < 4: normalized["hash_version"] = 4 try: computed = event_hash(normalized) except EncryptedPayloadUnavailable: # Phase 2 encrypted event with no master key. Skip hash # recomputation for this row; do NOT fail the bundle. The honest # summary at the end reports the count. encrypted_unverified += 1 computed_hashes.append(normalized.get("event_hash", "")) continue if computed is None: # Stage 4 (F1): unknown/future hash_version. This verifier cannot # recompute the row -- UNVERIFIABLE, never COMPROMISED. The # recorded hash is carried so prev_hash continuity for later # events is still checked against it. unknown_version_unverifiable += 1 computed_hashes.append(normalized.get("event_hash", "")) continue recorded = normalized.get("event_hash", "") if computed != recorded: print("Hash chain integrity: COMPROMISED") print("First broken event at index %d (chain_index %s)" % (i, normalized.get("chain_index"))) print("Expected hash: %s..." % recorded[:16]) print("Computed hash: %s..." % computed[:16]) return 1 if computed != event_hashes[i]: print("Hash chain integrity: COMPROMISED") print("event_hashes[%d] does not match recomputed hash" % i) print("Expected: %s..." % event_hashes[i][:16]) print("Computed: %s..." % computed[:16]) return 1 computed_hashes.append(computed) if "chain_verified" in doc: print("chain_verified field present in bundle -- ignored (field is bundle-supplied and not cryptographically binding; hash recomputation is the authoritative check)") if encrypted_unverified > 0 or unknown_version_unverifiable > 0: print("Hash chain integrity: PARTIAL") if encrypted_unverified > 0: print("Payload storage: encrypted (rankigi-managed)") print("Content verification: NOT AVAILABLE (supply --payload-key for full verification)") print("Events with encrypted canonical not recomputed: %d of %d" % (encrypted_unverified, len(events))) if unknown_version_unverifiable > 0: print("%d events with unknown hash version -- UNVERIFIABLE (this verify.py cannot recompute them; fetch the latest from https://rankigi.com/verify.py)" % unknown_version_unverifiable) else: print("Hash chain integrity: VERIFIED (recomputed locally)") print("Schema: %s" % doc.get("rankigi_schema")) print("Events: %d" % len(computed_hashes)) print("Content hash: %s" % expected_content_hash) # In-bundle nonce uniqueness (replay defense) nonce_ok, nonce_reason = check_nonce_uniqueness(events) if not nonce_ok: print() print("DUPLICATE NONCE DETECTED") print(nonce_reason) return 1 # Stage 4 (4C): per-event proof_extensions integrity. Renders for 1.6 # bundles (and any bundle carrying non-null pe fields); a hash mismatch # is tamper evidence and hard-fails the bundle. pe_mismatches = report_proof_extensions(events, schema_version) if pe_mismatches > 0: print() print("=== RESULT: COMPROMISED ===") print("%d event(s) failed proof_extensions hash verification." % pe_mismatches) return 1 print() print("--- Passport Binding ---") report_v4_binding(events) binding_result = verify_passport_binding(doc, events) if binding_result == "mismatch": print("Hash chain integrity: COMPROMISED") print("Passport binding cross-check failed.") return 1 pinned_via = "--pubkey" if pinned_pem is not None else "envelope (self-reported)" print("Pinned via: %s" % pinned_via) print() print("--- Merkle Inclusion ---") _merkle_proof = doc.get("merkle_inclusion_proof") if isinstance(doc, dict) else None _merkle_status, _ = verify_merkle_inclusion_proof(_merkle_proof) if _merkle_status == "SKIPPED": print("Merkle inclusion proof: SKIPPED (no proof in bundle)") print() print("--- Closure Signature ---") closure_sig_status = verify_closure_envelope_signature(doc) print() print("--- Signature Verification ---") sig_status, sig_count, sig_failed_idx, signed_total, server_attested_total, key_source = verify_event_signatures( doc, events, pinned_key=pinned_key, network_mode=network_mode, base_url=base_url, offline=offline, ) total_events = len(events) if sig_status == "passed": print("Ed25519 signatures: VERIFIED (against passport key)%s" % _key_source_suffix(key_source)) unsigned = total_events - signed_total if unsigned > 0: print("Verified: %d of %d events (others lack signature field)" % (sig_count, total_events)) else: print("Verified: %d of %d events" % (sig_count, total_events)) elif sig_status == "no_signatures": if isinstance(doc, dict) and doc.get("passport_key_fingerprint"): if server_attested_total > 0: print("Ed25519 signatures: NOT REQUIRED") print("Note: every event lacking a signature is a server-attested chain_closure.") else: print("ERROR: Bundle envelope claims v4 passport binding but no events carry signatures. Hard fail.") sys.exit(1) else: print("Ed25519 signatures: SKIPPED") print("Reason: no events carry signature/nonce fields (legacy or v3 bundle)") elif sig_status == "skipped": print("Ed25519 signatures: SKIPPED") print("Reason: cryptography library not installed (pip install cryptography)") elif sig_status == "no_key": print("Ed25519 signatures: SKIPPED") print("Reason: no public key available (envelope omits passport_public_key_pem and no --pubkey provided)") elif sig_status == "failed": print("Ed25519 signatures: FAILED") print("SIGNATURE INVALID at event %d" % sig_failed_idx) print() print("=== RESULT: COMPROMISED ===") sys.exit(1) print() print("--- External Anchor ---") rekor = doc.get("sigstore_log_index") anchor_printed = False if rekor is not None: if check_rekor: # Compare against anchor_payload_hash if present. L1-N1 fix # (2026-06-11): the Rekor entry's data.hash.value is sha256 over # the raw snapshot-hash bytes, and anchor_payload_hash now stores # exactly that digest (backfilled by migration 20260611000001). # The earlier comment here claimed the entry stored a JCS-object # digest -- that described intent, not reality; submit() never # anchored it. Bundles exported before Day 0 prep have no # anchor_payload_hash and cannot be compared safely; SKIP rather # than print MISMATCH. Bundles exported before the L1-N1 fix # (no anchor_payload_formula marker) downgrade to a deprecation # warning inside verify_rekor_binding. expected = doc.get("anchor_payload_hash") if expected: verify_rekor_binding( rekor, expected, rekor_requested=rekor_requested, offline=offline, doc=doc, ) else: print("Rekor binding: SKIPPED (pre-Day-0 bundle has no anchor_payload_hash)") else: print("Rekor logIndex: %s (PENDING --rekor flag)" % rekor) anchor_printed = True elif check_rekor: print("Rekor logIndex: NOT YET ANCHORED (daily snapshot pending)") anchor_printed = True rfc = doc.get("rfc3161_timestamp") if isinstance(rfc, dict) and rfc.get("tsa_timestamp"): print("TSA token present: YES (offline verification follows)") anchor_printed = True if not anchor_printed: print("Rekor logIndex: SKIPPED (none in bundle)") print("TSA timestamp: SKIPPED (none in bundle)") print() print("--- RFC 3161 TSA ---") verify_tsa_tokens(doc) print() print("--- Authorizations ---") report_authorizations(doc) print() print("--- Data Lineage ---") verify_data_lineage(doc) print() print("--- Coverage Attestation ---") check_coverage_commitment(doc, doc.get("events")) print() print("--- Counterparty Receipts ---") ccap_ok = True if isinstance(doc.get("counterparty_receipts"), list) and len(doc.get("counterparty_receipts")) > 0: ccap_ok, _ = verify_ccap_receipts(doc, ccap_keys=ccap_keys) elif force_ccap: print("Counterparty receipts: NONE (--ccap requested)") else: print("Counterparty receipts: NONE") # Crossings and XACT are CCAP-adjacent attestations from the universal # crossing layer; print them as sub-blocks under Counterparty Receipts # rather than introducing new top-level sections (Phase 3 spec order). print() print("Crossings (world receipts):") verify_crossings(doc) print() print("XACT commitments:") if "xact_commitments" in doc or force_xact: verify_xact_commitments(doc) else: print("XACT commitments: NONE") print() print("--- Action Dossier ---") # The dossier block lives at doc["dossier"] (the contributor merges it as # { dossier: {...} }). Older bundles emitted top-level dossier_version; # support both for back-compat. dossier = doc.get("dossier") if isinstance(doc.get("dossier"), dict) else None if dossier and dossier.get("dossier_version"): # Promote dossier fields to the top so report_action_dossier and # report_fre_applicability can read them without code duplication. for k in ("dossier_version", "generated_at", "fre_applicable", "uar_completeness_score", "fre_basis", "authority_chain_count"): if k in dossier and k not in doc: doc[k] = dossier[k] if doc.get("dossier_version"): report_action_dossier(doc) # Citations subsection lives under Action Dossier semantically. print() print("Citations:") report_citations(doc) else: print("Action Dossier: NOT AVAILABLE") print() print("Citations:") report_citations(doc) print() if unknown_version_unverifiable > 0: # Stage 4 (F1): surface unknown-version rows in the verdict block. print("%d events with unknown hash version -- UNVERIFIABLE" % unknown_version_unverifiable) # Verdict for closure envelope (mode B). # sig_status is set above from verify_event_signatures(). Failures and # COMPROMISED exits already happened before this point, so remaining # sig_status values are: passed, no_signatures, no_key, skipped. # # Special case: server_attested_total > 0 means every unsigned event is a # legitimately server-attested chain_closure. When sig_status is # no_signatures AND all unsigned events are server-attested, the closure is # still considered VERIFIED because RANKIGI sealed it without holding the # passport private key -- that is the designed, documented behaviour. # Phase B check-12 fix: every closed chain is N passport-signed events # plus ONE server-attested chain_closure (RANKIGI does not hold the # passport private key -- documented design). Server-attested events are # strictly allowlisted inside verify_event_signatures (forged unsigned # events either sys.exit or count as FAILED before this point), so they # count toward signature coverage. The old `signed_total == total_events` # made the standard closure-bundle shape land on PARTIAL forever. all_signed_closure = ( signed_total > 0 and signed_total + server_attested_total == total_events ) some_signed_closure = ( signed_total > 0 and signed_total + server_attested_total < total_events ) server_attested_only = ( sig_status == "no_signatures" and server_attested_total > 0 and server_attested_total == total_events ) # Trust hierarchy: VERIFIED is reserved for key_source in the pinned or # network-attested classes. A bundle whose key came from its own # self-reported envelope cannot reach VERIFIED no matter how clean the # crypto is locally; that path collapses to SELF-ATTESTED. trusted_key = key_source in ("--pubkey (pinned)", "network (root-attested)") self_attested_key = key_source == "bundle (self-reported)" closure_blocking = closure_sig_status == "failed" closure_warning = closure_sig_status == "warning" webauthn_failed = bool( doc.get("_verify_state", {}).get("webauthn_failed") ) tsa_failed = bool( doc.get("_verify_state", {}).get("tsa_failed") ) rekor_required_unreachable = bool( doc.get("_verify_state", {}).get("rekor_required_unreachable") ) sigs_pass_or_server_attested = ( (sig_status == "passed" and all_signed_closure) or server_attested_only ) if closure_blocking: print("=== RESULT: UNVERIFIED ===") print("Closure envelope signature did not verify against the RANKIGI root key.") report_fre_applicability(doc, local_verified=False) return 1 if not ccap_ok: print("=== RESULT: UNVERIFIED ===") print("At least one Counterparty receipt signature FAILED.") report_fre_applicability(doc, local_verified=False) return 1 if webauthn_failed: print("=== RESULT: UNVERIFIED ===") print("WebAuthn assertion did not verify.") report_fre_applicability(doc, local_verified=False) return 1 if sigs_pass_or_server_attested and trusted_key: # Fix 5: require non-null closure_signature for VERIFIED. closure_sig_present = bool(doc.get("closure_signature")) schema = doc.get("rankigi_schema") if not closure_sig_present: if schema in ("1.5", "1.6"): print("=== RESULT: FAILED ===") print("closure_signature required for schema %s" % schema) report_fre_applicability(doc, local_verified=False) return 1 if schema == "1.4": print("=== RESULT: SELF-ATTESTED (closure unsigned) ===") print("Schema 1.4 bundle has no closure_signature; cannot reach VERIFIED.") report_fre_applicability(doc, local_verified=False) return 1 # Any other case without closure_signature: refuse VERIFIED. print("=== RESULT: SELF-ATTESTED (closure unsigned) ===") print("closure_signature missing; cannot reach VERIFIED.") report_fre_applicability(doc, local_verified=False) return 1 # Fix 4 / Fix 5: combine TSA_UNCONFIRMED and REKOR_UNCONFIRMED into # a single verdict when both flags are set. Either flag alone still # downgrades verdict + returns 1. caveats = [] if tsa_failed: caveats.append("TSA_UNCONFIRMED") if rekor_required_unreachable: caveats.append("REKOR_UNCONFIRMED") if caveats: # Security sprint: do NOT print "VERIFIED" when an external anchor # could not be confirmed. ANCHOR_UNCONFIRMED is a distinct verdict # paired with exit code 1 so CI/automation cannot mistakenly treat # the output as a passing run. print( "=== RESULT: ANCHOR_UNCONFIRMED (chain verified, external timestamp failed: %s) ===" % ", ".join(caveats) ) if tsa_failed: print("Hash chain and closure signature verified, but the RFC 3161 TSA") print("certificate chain did not verify against the pinned FreeTSA root.") if rekor_required_unreachable: print("--rekor was requested but Rekor was unreachable.") print("Chain integrity and signatures verified. External anchor unconfirmed.") report_fre_applicability(doc, local_verified=True) return 1 # L1-H2: unsigned internal events (witness marker, no signature) cap # the verdict. The chain verified locally and nothing was forged, but # server-originated events without a witness signature cannot support # an externally trustable VERIFIED. if _WITNESS_UNSIGNED_COUNT > 0: print( "=== RESULT: SELF-ATTESTED (unsigned internal events) ===" ) print( "%d RANKIGI-internal event(s) carry the witness marker but no witness" % _WITNESS_UNSIGNED_COUNT ) print("signature (server witness key unconfigured at write time). The chain") print("verifies locally, but unsigned server-originated events cannot reach") print("VERIFIED.") report_fre_applicability(doc, local_verified=True) return 1 if closure_warning: print("=== RESULT: VERIFIED (closure envelope unsigned) ===") print("Hash chain and event signatures verified, but the closure envelope") print("itself was not signed (warning: %s)." % (doc.get("closure_signature_warning") or "unsigned")) else: print("=== RESULT: VERIFIED ===") print("The hash chain was recomputed independently and every event " "signature checked out. Nothing in this bundle has been altered " "since it was sealed; you did not need to trust RANKIGI to confirm it.") report_fre_applicability(doc, local_verified=True) return 0 if sigs_pass_or_server_attested and self_attested_key: print("=== RESULT: SELF-ATTESTED (UNVERIFIED) ===") print("Every event signature checks against the key shipped inside the bundle,") print("but that key has no out-of-band trust anchor. Re-run with --pubkey or") print("--pubkey-from-network for independent verification.") report_fre_applicability(doc, local_verified=False) return 1 if sig_status == "passed" and some_signed_closure: print("=== RESULT: PARTIAL ===") print("Hash chain integrity verified. %d of %d events carry verified Ed25519 signatures." % (sig_count, total_events)) print("%d events have no signature field." % (total_events - signed_total)) report_fre_applicability(doc, local_verified=False) return 1 if sig_status in ("no_signatures", "no_key", "skipped"): print("=== RESULT: UNVERIFIED ===") print("WARNING: This bundle contains unsigned events or no usable key.") print("Signature verification was not possible; the bundle cannot be treated as tamper-evident proof.") report_fre_applicability(doc, local_verified=False) return 1 print("=== RESULT: PARTIAL ===") print("Hash chain integrity verified. Signature status: %s." % sig_status) report_fre_applicability(doc, local_verified=False) return 1 def verify_proof_bundle_entries(doc, check_rekor=False): """Mode C: proof/bundle schema 1.0 -- chain.entries[] only. The public proof bundle (from /api/proof/bundle) contains a chain.entries[] array with fields: event_id, chain_index, hash, prev_hash, hash_version, passport_fingerprint. It does NOT include action, agent_id, org_id, occurred_at, payload, or tool -- the fields needed to recompute event hashes from scratch. Hash recomputation is therefore NOT possible and this mode does NOT attempt it. What this mode DOES verify: - prev_hash link continuity: each entry's prev_hash equals the preceding entry's hash (ordered by chain_index ascending). - chain_index monotonicity: indexes are strictly increasing. - The head hash is reported for out-of-band comparison. What this mode does NOT verify: - Individual hash recomputation (fields absent). - Ed25519 signatures (no signature or nonce fields in entries). - Rekor anchor (no anchor_payload_hash in this bundle shape). This path is ONLY taken when the document has no top-level events[] and no rankigi_schema key. The closure-export path (rankigi_schema 1.4/1.5) is unaffected and continues to use verify_closure_envelope(). """ chain = doc.get("chain") if isinstance(doc, dict) else None entries = chain.get("entries") if isinstance(chain, dict) else None if not isinstance(entries, list) or len(entries) == 0: print("Hash chain integrity: NO DATA") print("Bundle contains no chain.entries and no events.") print() print("=== RESULT: UNVERIFIED ===") print("WARNING: This bundle contains unsigned events. Signature verification was not possible. This bundle cannot be treated as tamper-evident proof.") return 1 print("--- Hash Chain (proof bundle -- link continuity only) ---") print("Note: This is a public proof bundle (schema %s)." % doc.get("rankigi_proof_bundle_schema", "unknown")) print("Hash recomputation is NOT possible: event fields (action, agent_id, payload, etc.)") print("are not included in this bundle. Only prev_hash link continuity is checked.") print("For full hash recomputation, use the authenticated closure export:") print(" POST /api/chains/{chainId}/closure-export") print() # Group entries by chain_id so multi-chain bundles do not falsely report # COMPROMISED. Each chain_id is its own monotonic scope with its own # ZERO prev_hash for the first entry. Legacy single-chain bundles have # no chain_id and group under the None key. ZERO = "0" * 64 groups = {} for entry in entries: if not isinstance(entry, dict): continue cid = entry.get("chain_id") groups.setdefault(cid, []).append(entry) broken = False heads_by_chain = {} total_entries = 0 for cid, group_entries in groups.items(): sorted_group = sorted( group_entries, key=lambda e: e.get("chain_index", 0) or 0 ) prev_hash = None prev_index = -1 for entry in sorted_group: idx = entry.get("chain_index", 0) or 0 h = entry.get("hash", "") ph = entry.get("prev_hash", "") expected_prev = ZERO if prev_hash is None else prev_hash if idx <= prev_index: print("Hash chain integrity: COMPROMISED") print( "chain_index not monotonic at %s (previous %s) in chain %s" % (idx, prev_index, cid) ) broken = True break if ph != expected_prev: print("Hash chain integrity: COMPROMISED") print( "prev_hash mismatch at chain_index %s in chain %s" % (idx, cid) ) print("Expected: %s..." % expected_prev[:16]) print("Found: %s..." % ph[:16]) broken = True break prev_hash = h prev_index = idx if broken: break heads_by_chain[cid] = prev_hash or "" total_entries += len(sorted_group) if broken: return 1 print("Hash chain link continuity: VERIFIED (prev_hash links only — hash recomputation not possible in this bundle format)") print("Entries: %d" % total_entries) print("Chains: %d" % len(heads_by_chain)) for cid in sorted(heads_by_chain, key=lambda x: ("" if x is None else str(x))): label = cid if cid is not None else "legacy" print("Head[%s]: %s..." % (label, heads_by_chain[cid][:16])) passport = doc.get("passport") if isinstance(doc, dict) else None if isinstance(passport, dict) and passport.get("fingerprint"): print() print("--- Passport ---") print("Algorithm: %s" % passport.get("algorithm", "unknown")) print("Fingerprint: %s..." % str(passport.get("fingerprint", ""))[:16]) print("Well-known: %s" % passport.get("well_known_url", "unknown")) print("Note: signature verification NOT possible (no signature fields in entries).") print() print("--- External Anchor ---") rekor = doc.get("rekor_log_index") if isinstance(doc, dict) else None if rekor is not None: if check_rekor: anchor_hash = doc.get("anchor_payload_hash", "") verify_rekor_binding( rekor, anchor_hash, rekor_requested=check_rekor, doc=doc, ) else: print("Rekor logIndex: %s (PENDING --rekor flag)" % rekor) else: print("Rekor logIndex: SKIPPED (none in bundle)") print() print("=== RESULT: UNVERIFIED ===") print("WARNING: This bundle contains unsigned events. Signature verification was not possible. This bundle cannot be treated as tamper-evident proof.") return 1 # === SEAL VERIFICATION === # Optional, gated behind --check-seals. Surfaces each seal in the bundle, # fetches the issuer's public deployer record, verifies the ed25519 # issuer_signature over the snapshot_hash bytes, and reports an overall # verdict line. Network and crypto errors are reported per seal and never # raise; the chain verdict is independent of seal status. SEALS_ENDPOINT_PATH = "/api/public/deployers/" def _collect_bundle_seals(doc): """Return a flat list of seal dicts from the bundle. Looks for a top-level `seals` array first, then any `seals` arrays embedded inside events[]. Returns an empty list when nothing is found. Never raises. """ out = [] if not isinstance(doc, dict): return out top = doc.get("seals") if isinstance(top, list): for s in top: if isinstance(s, dict): out.append(s) events = doc.get("events") if isinstance(events, list): for ev in events: if not isinstance(ev, dict): continue ev_seals = ev.get("seals") if isinstance(ev_seals, list): for s in ev_seals: if isinstance(s, dict): out.append(s) return out def _fetch_deployer_record(base_url, deployer_id): """GET {base_url}/api/public/deployers/{id} with a 5s timeout. Returns (record_dict, None) on success or (None, reason) on any failure. Stdlib urllib only, no new dependencies. """ if not deployer_id: return None, "missing deployer id" url = base_url.rstrip("/") + SEALS_ENDPOINT_PATH + str(deployer_id) req = urllib.request.Request(url, headers={"Accept": "application/json"}) try: with urllib.request.urlopen(req, timeout=5) as resp: raw = resp.read() except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError) as e: return None, "network unavailable (%s)" % type(e).__name__ except Exception as e: return None, "fetch failed (%s)" % type(e).__name__ try: return json.loads(raw.decode("utf-8")), None except Exception as e: return None, "non-JSON response (%s)" % type(e).__name__ def _verify_seal_signature(public_key_b64_or_pem, signature_str, snapshot_hash): """Verify an Ed25519 seal signature. public_key_b64_or_pem: PEM string OR raw 32-byte base64-encoded key. signature_str: "ed25519:" OR raw base64 string. snapshot_hash: "sha256:" OR raw hex string. The signing input is the hex part, UTF-8 encoded (matches the spec in the task brief). Returns (ok, reason). ok=True means the signature mathematically verified. """ if not CRYPTO_AVAILABLE: return False, "signature check skipped (no crypto lib available)" if not public_key_b64_or_pem or not signature_str or not snapshot_hash: return False, "missing key, signature, or snapshot_hash" # Strip the algorithm prefix from the signature if present. sig_raw = signature_str if isinstance(sig_raw, str) and sig_raw.startswith("ed25519:"): sig_raw = sig_raw[len("ed25519:"):] try: sig_bytes = base64.b64decode(sig_raw) except Exception as e: return False, "signature decode failed (%s)" % type(e).__name__ # Hash input: hex part of snapshot_hash (after "sha256:") as UTF-8 bytes. hash_input = snapshot_hash if isinstance(hash_input, str) and hash_input.startswith("sha256:"): hash_input = hash_input[len("sha256:"):] signed_input = str(hash_input).encode("utf-8") # Load the public key. Accept PEM, or raw base64 of the 32-byte Ed25519 key. pubkey = None pem_candidate = public_key_b64_or_pem if isinstance(pem_candidate, str) and "BEGIN PUBLIC KEY" in pem_candidate: try: pubkey = load_pem_public_key(pem_candidate.encode("utf-8")) except Exception as e: return False, "PEM load failed (%s)" % type(e).__name__ else: try: raw = base64.b64decode(public_key_b64_or_pem) pubkey = Ed25519PublicKey.from_public_bytes(raw) except Exception as e: return False, "raw key load failed (%s)" % type(e).__name__ try: pubkey.verify(sig_bytes, signed_input) return True, "ed25519" except InvalidSignature: return False, "invalid" except Exception as e: return False, type(e).__name__ def verify_seals(doc, base_url=DEFAULT_BASE_URL): """Run --check-seals verification over the bundle. Returns one of 'SEAL VERIFIED', 'SEAL_ISSUER_REVOKED', 'SEAL_INVALID', or 'SEAL SKIPPED' (no seals in bundle). Prints structured output per seal. """ seals = _collect_bundle_seals(doc) if not seals: print("Seals: NONE IN BUNDLE") return "SEAL SKIPPED" overall = "SEAL VERIFIED" any_revoked = False any_invalid = False for i, seal in enumerate(seals): deployer_id = ( seal.get("deployer_id") or seal.get("issuer_id") or seal.get("deployer", {}).get("id") if isinstance(seal.get("deployer"), dict) else None ) if not deployer_id and isinstance(seal.get("deployer"), dict): deployer_id = seal["deployer"].get("id") snapshot_hash = seal.get("snapshot_hash") or seal.get("seal_hash") issuer_signature = ( seal.get("issuer_signature") or seal.get("signature") ) inline_pubkey = ( seal.get("ed25519_public_key") or seal.get("public_key") or seal.get("issuer_public_key") ) canon_id = seal.get("canon_id") canon_version = seal.get("canon_version") issued = seal.get("issued_at") or seal.get("created_at") or "unknown" expires = seal.get("expires_at") or "unknown" print("SEAL %d" % i) record = None if inline_pubkey is None and deployer_id: record, reason = _fetch_deployer_record(base_url, deployer_id) if record is None: print(" seal issuer fetch failed: %s" % reason) continue # Resolve issuer display attributes from the fetched record when # available, else fall back to whatever the seal embeds. issuer_name = ( (record or {}).get("display_name") if record else seal.get("deployer_name") ) or "unknown" issuer_status = ( (record or {}).get("status") if record else seal.get("issuer_status") ) or "unknown" record_pubkey = (record or {}).get("ed25519_public_key") if record else None revoked_at = (record or {}).get("revoked_at") if record else seal.get("revoked_at") canon_slug = ( seal.get("canon_slug") or seal.get("canon_name") or canon_id or "unknown" ) verified_label = "verified" if record is not None else "self-reported" print(" issuer: %s (%s)" % (issuer_name, verified_label)) print(" issuer status: %s" % str(issuer_status).upper()) print(" issued: %s" % issued) print(" expires: %s" % expires) print(" canon: %s %s" % (canon_slug, canon_version or "")) sig_key = record_pubkey or inline_pubkey ok, reason = _verify_seal_signature( sig_key, issuer_signature, snapshot_hash ) if reason == "signature check skipped (no crypto lib available)": print(" signature check skipped (no crypto lib available)") elif ok: print(" signature: VERIFIED (against issuer key: %s)" % issuer_name) else: print(" signature: NOT VERIFIED (%s)" % reason) any_invalid = True if str(issuer_status).lower() == "revoked": any_revoked = True print(" WARNING: seal issuer %s has been revoked as of %s." % (issuer_name, revoked_at or "unknown")) print(" Seal cryptography is valid but issuer trust is withdrawn.") if any_invalid: overall = "SEAL_INVALID" elif any_revoked: overall = "SEAL_ISSUER_REVOKED" return overall # === END SEAL VERIFICATION === # === GOVERNANCE SEALS (presentation block) === # Additive presentation layer over the chain events. Unlike --check-seals, this # block does not perform cryptographic verification of the seal signature: it # surfaces operator-visible status (ACTIVE/EXPIRED/REVOKED) based on the seal's # expires_at, revoked_at, and the live deployer status from the public # rankigi.com endpoint. Always emitted (no CLI flag) for bundles whose events # carry hash_version >= 4. Never alters the overall verdict. GOVERNANCE_SEALS_ENDPOINT = "/api/public/deployers/" def _bundle_has_v4_events(doc): """Return True when any event in the bundle declares hash_version >= 4.""" if not isinstance(doc, dict): return False candidates = [] if isinstance(doc.get("events"), list): candidates.extend(doc["events"]) chain = doc.get("chain") if isinstance(chain, dict) and isinstance(chain.get("entries"), list): candidates.extend(chain["entries"]) for ev in candidates: if not isinstance(ev, dict): continue raw = ev.get("hash_version") try: if raw is not None and int(raw) >= 4: return True except (TypeError, ValueError): continue return False def _collect_seal_issued_events(doc): """Return chain events whose action == 'seal_issued'. Inspects both top-level events[] (closure bundles) and chain.entries[] (public proof bundles) so the same surface area works for either mode. """ out = [] if not isinstance(doc, dict): return out pools = [] if isinstance(doc.get("events"), list): pools.append(doc["events"]) chain = doc.get("chain") if isinstance(chain, dict) and isinstance(chain.get("entries"), list): pools.append(chain["entries"]) for pool in pools: for ev in pool: if isinstance(ev, dict) and ev.get("action") == "seal_issued": out.append(ev) return out def _fetch_deployer_status(base_url, deployer_slug, offline): """GET {base_url}/api/public/deployers/{slug}. Returns one of 'ACTIVE', 'REVOKED', 'UNKNOWN_OFFLINE'. Never raises. The public endpoint returns 200 only for active deployers; non-200 is treated as UNKNOWN_OFFLINE rather than asserting REVOKED so the verifier does not slander a deployer over a transient lookup failure. Explicit 'revoked' status in the JSON body, when surfaced, is honored. """ if offline or not deployer_slug: return "UNKNOWN_OFFLINE" url = base_url.rstrip("/") + GOVERNANCE_SEALS_ENDPOINT + str(deployer_slug) try: req = urllib.request.Request( url, headers={"Accept": "application/json"} ) with urllib.request.urlopen(req, timeout=5) as resp: raw = resp.read() except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError): return "UNKNOWN_OFFLINE" except Exception: return "UNKNOWN_OFFLINE" try: record = json.loads(raw.decode("utf-8")) except Exception: return "UNKNOWN_OFFLINE" status = ( record.get("status") if isinstance(record, dict) else None ) if isinstance(status, str) and status.lower() == "active": return "ACTIVE" if isinstance(status, str) and status.lower() == "revoked": return "REVOKED" return "UNKNOWN_OFFLINE" def _iso_date(s): if not s: return "n/a" try: text = str(s).replace("Z", "+00:00") dt = datetime.fromisoformat(text) return dt.astimezone(timezone.utc).strftime("%Y-%m-%d") except Exception: return str(s)[:10] def render_governance_seals_block(doc, base_url=DEFAULT_BASE_URL, offline=False): """Render the --- GOVERNANCE SEALS --- presentation block. No-op when no v4 events exist or no seal_issued events are found. """ if not _bundle_has_v4_events(doc): return seal_events = _collect_seal_issued_events(doc) if not seal_events: return today = datetime.now(timezone.utc).date() print() print("--- GOVERNANCE SEALS ---") for ev in seal_events: payload = ev.get("payload") if isinstance(ev, dict) else None if not isinstance(payload, dict): # Fall back to payload_canonical when payload is hash-only nulled. pc = ev.get("payload_canonical") if isinstance(ev, dict) else None if isinstance(pc, str) and pc: try: payload = json.loads(pc) except Exception: payload = {} else: payload = {} if not isinstance(payload, dict): payload = {} canon_id = payload.get("canon_id") or "unknown" canon_name = ( payload.get("canon_name") or payload.get("canon_display_name") or canon_id ) canon_version = payload.get("canon_version") or "unknown" deployer_slug = payload.get("deployer_slug") or "unknown" trust_tier = payload.get("deployer_trust_tier") or "unknown" expires_at = payload.get("expires_at") issued_at = ( payload.get("issued_at") or ev.get("occurred_at") ) seal_hash = payload.get("seal_hash") or payload.get("snapshot_hash") or "" seal_short = "" if isinstance(seal_hash, str): cleaned = seal_hash if cleaned.startswith("sha256:"): cleaned = cleaned[len("sha256:"):] seal_short = cleaned[:16] # Compute expiry. is_expired = False if isinstance(expires_at, str) and expires_at: try: exp_dt = datetime.fromisoformat( expires_at.replace("Z", "+00:00") ).astimezone(timezone.utc).date() is_expired = today > exp_dt except Exception: is_expired = False deployer_status = _fetch_deployer_status( base_url, deployer_slug, offline ) if deployer_status == "REVOKED": status_word = "REVOKED" elif is_expired: status_word = "EXPIRED" else: status_word = "ACTIVE" if is_expired: result = "SEAL_EXPIRED" elif deployer_status == "REVOKED": result = "SEAL_ISSUER_REVOKED" else: result = "SEAL VERIFIED" seal_label = ( str(canon_name) if canon_name and canon_name != "unknown" else str(deployer_slug) ) print("Seal: %s Seal" % seal_label) print("Issuer: %s (%s)" % (deployer_slug, trust_tier)) print("Canon: %s@%s" % (canon_id, canon_version)) print("Status: %s" % status_word) print("Issued: %s" % _iso_date(issued_at)) print("Expires: %s" % _iso_date(expires_at)) print("Deployer status: %s" % deployer_status) print("Seal hash: %s..." % (seal_short or "n/a")) print("Result: %s" % result) if ( result == "SEAL_ISSUER_REVOKED" and deployer_status == "REVOKED" ): print("(cryptography valid, trust withdrawn)") print() # === END GOVERNANCE SEALS === # ============================================================================= # Merkle inclusion proof verification (audit 2026-06-09, P1-1) # ============================================================================= # # Closure bundles published from 2026-06-09 forward carry # `merkle_inclusion_proof.merkle_algo_version` identifying the algorithm used # to build the snapshot root + proof. We support two: # # legacy-v1 = pre-RFC-6962 (Bitcoin-style: concat left+right hex, sha256). # Single leaf root is the leaf itself. No domain separation. # Vulnerable to CVE-2012-2459-class second-preimage attacks. # Kept for backward compatibility with pre-2026-06-09 bundles. # # rfc6962-v2 = RFC 6962 §2.1 (0x00-prefixed leaves, 0x01-prefixed internal # nodes). Odd nodes carry up; no Bitcoin-style duplication. # # The bundle's `merkle_algo_version` selects the verifier. Bundles missing the # field are treated as legacy-v1 with a printed downgrade warning. import hashlib as _hashlib def _merkle_verify_legacy_v1(leaf_hex, path, expected_root): """Pre-RFC-6962 Merkle proof verification. Hex-concatenated sha256.""" cur = leaf_hex.lower() for step in path: sibling = step.get("sibling", "").lower() side = step.get("side") if side == "L": combined = sibling + cur else: combined = cur + sibling cur = _hashlib.sha256(combined.encode("ascii")).hexdigest() return cur == (expected_root or "").lower() def _merkle_verify_rfc6962_v2(leaf_hex, path, expected_root): """RFC 6962 §2.1: leaf = sha256(0x00 || data), internal = sha256(0x01 || L || R).""" try: leaf_bytes = bytes.fromhex(leaf_hex) except (TypeError, ValueError): return False cur = _hashlib.sha256(b"\x00" + leaf_bytes).digest() for step in path: sibling_hex = step.get("sibling", "") side = step.get("side") try: sibling_bytes = bytes.fromhex(sibling_hex) except (TypeError, ValueError): return False if side == "L": cur = _hashlib.sha256(b"\x01" + sibling_bytes + cur).digest() else: cur = _hashlib.sha256(b"\x01" + cur + sibling_bytes).digest() return cur.hex() == (expected_root or "").lower() def verify_merkle_inclusion_proof(merkle_proof): """ Verify a merkle_inclusion_proof object pulled from a closure bundle. Returns a (status, detail) tuple where status is one of: "VERIFIED" - proof verifies under the advertised algorithm "BROKEN" - proof does not verify "SKIPPED" - missing fields; nothing to verify "ALGO_DOWNGRADE" - algo not advertised; defaulted to legacy-v1 Prints a human-readable result line. Does NOT abort verification of the rest of the closure bundle on a Merkle failure -- the caller decides whether to bake the result into the final verdict. """ if not isinstance(merkle_proof, dict): return ("SKIPPED", "no merkle_inclusion_proof in bundle") leaf = merkle_proof.get("leaf") root = merkle_proof.get("merkle_root") path = merkle_proof.get("path") if not leaf or not root or not isinstance(path, list): return ("SKIPPED", "incomplete merkle_inclusion_proof") algo = merkle_proof.get("merkle_algo_version") downgrade_warning = False if not algo: algo = "legacy-v1" downgrade_warning = True if algo == "rfc6962-v2": ok = _merkle_verify_rfc6962_v2(leaf, path, root) algo_label = "rfc6962-v2 (RFC 6962, domain-separated)" elif algo == "legacy-v1": ok = _merkle_verify_legacy_v1(leaf, path, root) algo_label = "legacy-v1 (pre-RFC-6962, NOT domain-separated)" else: print("Merkle: UNKNOWN algorithm '%s' -- not verified" % algo) return ("SKIPPED", "unknown merkle_algo_version: %s" % algo) if downgrade_warning: print( "Merkle: WARNING bundle has no merkle_algo_version; " "assuming legacy-v1 (pre-RFC-6962)." ) if ok: print("Merkle inclusion proof: VERIFIED (%s)" % algo_label) if algo == "legacy-v1": print( " Note: this bundle's Merkle snapshot uses the legacy " "algorithm and is vulnerable to CVE-2012-2459-class " "second-preimage attacks. Snapshots taken on 2026-06-09 " "or later use rfc6962-v2." ) return ("VERIFIED", "%s root=%s" % (algo, root[:16])) print("Merkle inclusion proof: BROKEN (%s)" % algo_label) print(" leaf=%s..." % (leaf[:16] if leaf else "")) print(" root=%s..." % (root[:16] if root else "")) return ("BROKEN", "%s root mismatch" % algo) def main(path, check_rekor=False, pubkey_path=None, force_ccap=False, force_xact=False, payload_key_b64=None, pubkey_from_network=False, base_url=DEFAULT_BASE_URL, ccap_keys_path=None, check_seals=False, offline=False): global _PAYLOAD_MASTER_KEY, _PAYLOAD_ENCRYPTED_SEEN, _PAYLOAD_ENCRYPTED_DECRYPTED, _PAYLOAD_ENCRYPTED_SKIPPED _PAYLOAD_ENCRYPTED_SEEN = 0 _PAYLOAD_ENCRYPTED_DECRYPTED = 0 _PAYLOAD_ENCRYPTED_SKIPPED = 0 if payload_key_b64 is not None: try: _PAYLOAD_MASTER_KEY = parse_payload_master_key(payload_key_b64) except Exception as e: print("PAYLOAD KEY INVALID: %s" % e) return 1 if not PAYLOAD_CRYPTO_AVAILABLE: print("WARNING: --payload-key supplied but cryptography library is not installed.") print(" Payload decryption will be skipped. Install with: pip install cryptography") _PAYLOAD_MASTER_KEY = None else: _PAYLOAD_MASTER_KEY = None with open(path) as f: doc = json.load(f) pinned_pem = None pinned_key = None if pubkey_path is not None: try: pinned_pem, pinned_key = load_pinned_pubkey(pubkey_path) except Exception as e: print("PASSPORT PIN LOAD FAILED") print("Path: %s" % pubkey_path) print("Reason: %s" % e) return 1 print_banner(pinned_via_flag=pubkey_path is not None) # Closure envelope (mode B) is identified by the rankigi_schema key. # Anything else falls through to the events-only mode (mode A). ccap_keys = load_ccap_keys(ccap_keys_path) if ccap_keys_path else {} if isinstance(doc, dict) and "rankigi_schema" in doc: return verify_closure_envelope( doc, check_rekor=check_rekor, pinned_pem=pinned_pem, pinned_key=pinned_key, force_ccap=force_ccap, force_xact=force_xact, network_mode=pubkey_from_network, base_url=base_url, ccap_keys=ccap_keys, offline=offline, rekor_requested=check_rekor, ) # Mode C: public proof bundle (rankigi_proof_bundle_schema) -- chain.entries[] # only. No top-level events[] and no rankigi_schema. Hash recomputation is # not possible; only link continuity is verified. This does NOT affect the # closure-export path (mode B). if ( isinstance(doc, dict) and "rankigi_proof_bundle_schema" in doc and not isinstance(doc.get("events"), list) ): return verify_proof_bundle_entries(doc, check_rekor=check_rekor) events = doc.get("events", doc) if isinstance(doc, dict) else doc print("--- Hash Chain ---") print("Verifying %d events..." % len(events)) # Group by chain scope. v2 events with chain_id chain per (agent_id, # chain_id). Everything else chains per agent_id. def scope_key(ev): raw = ev.get("hash_version") version = int(raw) if raw is not None else 3 # v2+ events with chain_id chain per (agent_id, chain_id). v1 chains # per agent_id only. if version >= 2 and ev.get("chain_id"): return (ev.get("agent_id", ""), ev.get("chain_id")) return (ev.get("agent_id", ""), None) events_sorted = sorted( events, key=lambda e: (scope_key(e), e.get("chain_index", 0) or 0), ) ZERO = "0" * 64 last_by_scope = {} versions_seen = set() encrypted_unverified_a = 0 unknown_version_a = 0 for e in events_sorted: raw_v = e.get("hash_version") version = int(raw_v) if raw_v is not None else 3 versions_seen.add(version) key = scope_key(e) idx = e.get("chain_index", 0) or 0 recorded = e.get("hash", e.get("event_hash", "")) prev_hash, prev_index = last_by_scope.get(key, (None, -1)) expected_prev = ZERO if prev_hash is None else prev_hash if idx <= prev_index: print("Hash chain integrity: COMPROMISED") print("First broken link at chain_index: %s (hash_version %s)" % (idx, version)) print("Index not monotonic for scope %s (previous %s)" % (key, prev_index)) return 1 if e.get("prev_hash", "") != expected_prev: print("Hash chain integrity: COMPROMISED") print("First broken link at chain_index: %s (scope %s, hash_version %s)" % (idx, key, version)) print("Expected prev_hash: %s..." % expected_prev[:16]) print("Found prev_hash: %s..." % e.get("prev_hash", "")[:16]) return 1 try: computed = event_hash(e) except EncryptedPayloadUnavailable: encrypted_unverified_a += 1 last_by_scope[key] = (recorded, idx) continue if computed is None: # Stage 4 (F1): unknown/future hash_version -- UNVERIFIABLE, # never COMPROMISED. Link continuity above already ran; carry the # recorded hash so later prev_hash links still verify. unknown_version_a += 1 last_by_scope[key] = (recorded, idx) continue if computed != recorded: print("Hash chain integrity: COMPROMISED") print("First broken link at chain_index: %s (scope %s, hash_version %s)" % (idx, key, version)) print("Expected hash: %s..." % recorded[:16]) print("Computed hash: %s..." % computed[:16]) return 1 last_by_scope[key] = (recorded, idx) if encrypted_unverified_a > 0 or unknown_version_a > 0: print("Hash chain integrity: PARTIAL") if encrypted_unverified_a > 0: print("Payload storage: encrypted (rankigi-managed)") print("Content verification: NOT AVAILABLE (supply --payload-key for full verification)") print("Events with encrypted canonical not recomputed: %d of %d" % (encrypted_unverified_a, len(events))) if unknown_version_a > 0: print("%d events with unknown hash version -- UNVERIFIABLE (this verify.py cannot recompute them; fetch the latest from https://rankigi.com/verify.py)" % unknown_version_a) elif len(events) == 0: # Fix 9: do not print VERIFIED when there are no events to verify. print("Hash chain integrity: NO EVENTS (nothing to verify)") else: print("Hash chain integrity: VERIFIED (recomputed locally)") print("Events: %d" % len(events)) print("Hash versions seen: %s" % sorted(versions_seen)) for k in sorted(last_by_scope, key=lambda x: (x[0], x[1] or "")): h, _ = last_by_scope[k] print("Head[%s]: %s..." % (k, h[:16])) # In-bundle nonce uniqueness (replay defense) nonce_ok, nonce_reason = check_nonce_uniqueness(events) if not nonce_ok: print() print("DUPLICATE NONCE DETECTED") print(nonce_reason) return 1 print() print("--- Passport Binding ---") report_v4_binding(events) pinned_via = "--pubkey" if pinned_pem is not None else "envelope (self-reported)" print("Pinned via: %s" % pinned_via) print() print("--- Signature Verification ---") envelope_for_sig = doc if isinstance(doc, dict) else {} network_mode_a = pubkey_from_network base_url_a = base_url sig_status, sig_count, sig_failed_idx, signed_total, server_attested_total, key_source = verify_event_signatures( envelope_for_sig, events, pinned_key=pinned_key, network_mode=network_mode_a, base_url=base_url_a, offline=offline, ) total_events = len(events) if sig_status == "passed": unsigned = total_events - signed_total print("Ed25519 signatures: VERIFIED (against passport key)%s" % _key_source_suffix(key_source)) if unsigned > 0: print("Verified: %d of %d events (others lack signature field)" % (sig_count, total_events)) else: print("Verified: %d of %d events" % (sig_count, total_events)) elif sig_status == "no_signatures": if isinstance(doc, dict) and doc.get("passport_key_fingerprint"): if server_attested_total > 0: print("Ed25519 signatures: NOT REQUIRED") print("Note: every event lacking a signature is a server-attested chain_closure.") else: print("ERROR: Bundle envelope claims v4 passport binding but no events carry signatures. Hard fail.") sys.exit(1) else: print("Ed25519 signatures: SKIPPED") print("Reason: no events carry signature/nonce fields (legacy or v3 bundle)") elif sig_status == "skipped": print("Ed25519 signatures: SKIPPED") print("Reason: cryptography library not installed (pip install cryptography)") elif sig_status == "no_key": print("Ed25519 signatures: SKIPPED") print("Reason: no public key available (no envelope key and no --pubkey provided)") elif sig_status == "failed": print("Ed25519 signatures: FAILED") print("SIGNATURE INVALID at event %d" % sig_failed_idx) print() print("=== RESULT: COMPROMISED ===") sys.exit(1) print() print("--- External Anchor ---") rekor = doc.get("rekor_log_index") if isinstance(doc, dict) else None if rekor is not None and not check_rekor: print("Rekor logIndex: %s (PENDING --rekor flag)" % rekor) elif rekor is None and not check_rekor: print("Rekor logIndex: SKIPPED (none in bundle)") else: # check_rekor was requested: Mode A cannot compare safely. The bundle # does not carry the snapshot anchor hash, so comparing the head event # hash to a Rekor entry would either false-mismatch real chains or # false-match forgeries (see audit C3). print("Rekor anchor: SKIPPED (events-only mode does not include snapshot anchor hash)") print(" Use closure-export bundle for Rekor verification") print("TSA timestamp: SKIPPED (events-only mode)") print() print("--- Data Lineage ---") verify_data_lineage(doc if isinstance(doc, dict) else {}) print() print("--- Coverage Attestation ---") check_coverage_commitment( doc if isinstance(doc, dict) else {}, doc.get("events") if isinstance(doc, dict) else [], ) print() if unknown_version_a > 0: # Stage 4 (F1): surface unknown-version rows in the verdict block. print("%d events with unknown hash version -- UNVERIFIABLE" % unknown_version_a) # Verdict: VERIFIED only when every signed event passed Ed25519 verification. # PARTIAL when some events signed+verified and others unsigned. # UNVERIFIED when no key, no signatures, or skipped (cannot confirm tamper-evidence). all_signed = signed_total == total_events and total_events > 0 some_signed = signed_total > 0 and signed_total < total_events trusted_key = key_source in ("--pubkey (pinned)", "network (root-attested)") self_attested_key = key_source == "bundle (self-reported)" # L1-H2: unsigned internal events (witness marker, no signature) cap the # Mode A verdict the same way they cap Mode B: not tamper evidence, but # not independently attestable either. if sig_status == "passed" and _WITNESS_UNSIGNED_COUNT > 0: print("=== RESULT: SELF-ATTESTED (unsigned internal events) ===") print( "%d RANKIGI-internal event(s) carry the witness marker but no witness" % _WITNESS_UNSIGNED_COUNT ) print("signature (server witness key unconfigured at write time). The chain") print("verifies locally, but unsigned server-originated events cannot be") print("independently attested.") return 1 if sig_status == "passed" and all_signed and trusted_key: print("=== RESULT: LOCAL-CONSISTENT ===") print(" Signatures verified against supplied key.") print(" However: events-only bundles carry no closure_signature, no TSA anchor,") print(" and no Rekor inclusion proof. This result proves local hash-chain") print(" consistency and signature validity only.") print(" For externally-anchored proof use a closure-export bundle (mode B).") return 1 if sig_status == "passed" and all_signed and self_attested_key: print("=== RESULT: SELF-ATTESTED (UNVERIFIED) ===") print("Every event signature checks against the key shipped inside the bundle,") print("but that key has no out-of-band trust anchor. Re-run with --pubkey or") print("--pubkey-from-network for independent verification.") return 1 if sig_status == "passed" and some_signed: print("=== RESULT: PARTIAL ===") print("Hash chain integrity verified. %d of %d events carry verified Ed25519 signatures." % (sig_count, total_events)) print("%d events have no signature field." % (total_events - signed_total)) return 1 if sig_status in ("no_signatures", "no_key", "skipped"): print("=== RESULT: UNVERIFIED ===") print("WARNING: This bundle contains unsigned events or no usable key.") print("Signature verification was not possible; the bundle cannot be treated as tamper-evident proof.") return 1 print("=== RESULT: PARTIAL ===") print("Hash chain integrity verified. Signature status: %s." % sig_status) return 1 def run_seal_check_if_requested(path, check_seals, base_url, offline=False): """Re-load the bundle and emit the seal verification block. Invoked from the __main__ wrapper so that every chain-verdict exit path (including sys.exit calls inside verify_closure_envelope) gets the seal block appended. Stays a no-op when --check-seals is not set. Security sprint: --offline suppresses the seal check because seal verification requires fetching issuer trust records over the network; silently running it in offline mode would either contact the network (violating the flag) or print a misleading SKIPPED line that looks indistinguishable from a missing-seals case. """ if not check_seals: return if offline: print() print("--- Agent Seals ---") print("Seals: SKIPPED (--offline mode -- seal verification requires network)") print("=== SEAL SKIPPED ===") return try: with open(path) as f: doc = json.load(f) except Exception as e: print() print("--- Agent Seals ---") print("Seals: SKIPPED (bundle reload failed: %s)" % type(e).__name__) print("=== SEAL SKIPPED ===") return print() print("--- Agent Seals ---") verdict = verify_seals(doc, base_url=base_url) print("=== %s ===" % verdict) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( prog="verify.py", description="RANKIGI chain verifier. Checks hash chain integrity, " "passport fingerprint binding, and Ed25519 signatures. " "Mode A (events-only) produces LOCAL-CONSISTENT, not " "VERIFIED. For VERIFIED use a closure-export bundle " "(mode B).", ) parser.add_argument("bundle", help="Path to chain.json or closure bundle.") parser.add_argument( "--rekor", action="store_true", help="Cross-check the chain head against the Rekor transparency log.", ) parser.add_argument( "--rekor-required", action="store_true", help="Treat Rekor unreachable / mismatch as a hard failure. Without this flag, ANCHOR_UNCONFIRMED previously exited 1; with it, the failure mode is explicit in the exit code semantics and a missing Rekor connection cannot be confused with a passing local-only run.", ) parser.add_argument( "--pubkey", help="Path to expected passport public key PEM. Pins trust out-of-band so the verifier does not have to trust the key shipped inside the bundle.", ) parser.add_argument( "--ccap", action="store_true", help="Force CCAP (Counterparty Closure Attestation Protocol) receipt reporting even when no receipts are present in the bundle.", ) parser.add_argument( "--xact", action="store_true", help="Force XACT bilateral commitment reporting even when no xact_commitments are present in the bundle.", ) parser.add_argument( "--integrity", action="store_true", help="Verify a standalone closure integrity object (schema rankigi.integrity.v1) instead of a full closure bundle.", ) parser.add_argument( "--payload-key", help="Base64-encoded 32-byte RANKIGI payload master key. When supplied, events carrying payload_canonical_encrypted are decrypted (HKDF-SHA256 per org_id, AES-256-GCM) and their canonical event records are recomputed for full hash verification. Without this flag, encrypted events are reported as 'content verification NOT AVAILABLE' rather than verified.", ) parser.add_argument( "--pubkey-from-network", action="store_true", help="Fetch the passport public key from the RANKIGI proof endpoint and require it to carry a root-attested signature. Trust level: HIGH. Required for VERIFIED when --pubkey is not supplied.", ) parser.add_argument( "--base-url", default=DEFAULT_BASE_URL, help="Override the proof endpoint base URL (default: %s). Used by --pubkey-from-network." % DEFAULT_BASE_URL, ) parser.add_argument( "--ccap-keys", help="Path to a JWKS file, flat-dict JSON, or directory of *.pem files containing CCAP signing keys. When supplied, counterparty receipt signatures are cryptographically verified. Without this flag receipts are REPORTED but not verified.", ) parser.add_argument( "--check-seals", action="store_true", help="Verify Agent Seal issuer signatures by fetching the issuer's public deployer record from the RANKIGI public endpoint. Stdlib urllib only, 5s timeout, never aborts the chain verdict.", ) parser.add_argument( "--offline", action="store_true", help="Suppress all outbound network fetches. Governance-seal deployer-status lookups (the additive --- GOVERNANCE SEALS --- block) will report UNKNOWN_OFFLINE instead of contacting rankigi.com.", ) ns = parser.parse_args() # Sprint 10A: surface the embedded root key provenance up-front so an # operator can see at-a-glance that the trust anchor is the version they # expected, not whatever the network returns later. Printed unconditionally # because it has no network dependency and is the only trust statement the # verifier can make without reading a single bundle byte. print( "Root key: VERIFIED (hardcoded v%s, fingerprint %s)" % (RANKIGI_PUBKEY_VERSION, RANKIGI_PUBKEY_FINGERPRINT[:16]) ) if ns.offline: print("OFFLINE MODE: all network verification disabled. Results reflect local bundle consistency only.") if ns.rekor: print("Rekor: SKIPPED (offline mode overrides --rekor)") if ns.pubkey_from_network: print("Pubkey network fetch: SKIPPED (offline mode overrides --pubkey-from-network)") elif ns.pubkey_from_network: # Cross-check: if the network root-key endpoint disagrees with the # embedded constant, the embedded constant wins. A malicious or stale # network response cannot silently change the trust anchor. try: req = urllib.request.Request( ns.base_url.rstrip("/") + "/.well-known/rankigi-root-key.json", headers={"Accept": "application/json", "User-Agent": "rankigi-verify/1"}, ) with urllib.request.urlopen(req, timeout=5) as resp: doc = json.loads(resp.read().decode("utf-8")) net_pem = (doc.get("public_key_pem") or "").encode("utf-8") if net_pem and net_pem != RANKIGI_ROOT_PUBKEY_PEM: print( "Root key mismatch warning: network endpoint returned a " "different root key than the one embedded in this " "verifier (v%s). Using the embedded constant. If you " "trust the embedded version, this is fine; if you " "expected a rotation, fetch the latest verify.py." % RANKIGI_PUBKEY_VERSION ) else: print( "Root key cross-check: OK (network matches embedded v%s)" % RANKIGI_PUBKEY_VERSION ) except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError, KeyError) as _e: print( "Root key cross-check: SKIPPED (network unreachable, " "using embedded v%s)" % RANKIGI_PUBKEY_VERSION ) if ns.integrity: verify_integrity_object(ns.bundle) sys.exit(0) # Offline mode overrides --rekor (suppress network call entirely; do not # mark the bundle as REQUIRED-BUT-UNREACHABLE since the user explicitly # opted out of network calls). _effective_rekor = ns.rekor and not ns.offline _rc = main( ns.bundle, check_rekor=_effective_rekor, pubkey_path=ns.pubkey, force_ccap=ns.ccap, force_xact=ns.xact, payload_key_b64=ns.payload_key, pubkey_from_network=ns.pubkey_from_network and not ns.offline, base_url=ns.base_url, ccap_keys_path=ns.ccap_keys, check_seals=ns.check_seals, offline=ns.offline, ) run_seal_check_if_requested(ns.bundle, ns.check_seals, ns.base_url, offline=ns.offline) # Always-on additive presentation block for v4 bundles. Independent of # --check-seals and never alters the chain verdict. try: with open(ns.bundle) as _f: _gs_doc = json.load(_f) render_governance_seals_block( _gs_doc, base_url=ns.base_url, offline=ns.offline ) except Exception: pass sys.exit(_rc)