#!/usr/bin/env python3 """ Pramagent forensic bundle verifier (standalone). This file is shipped inside every forensic bundle as ``verify_bundle.py``. It depends only on the Python standard library and ``cryptography`` (for the Ed25519 bundle signature). It does NOT import pramagent. V2 inclusion proofs are checked locally; checking pinned hybrid checkpoint signatures additionally requires pqcrypto. Usage:: python verify_bundle.py bundle.zip --trusted-key exporter_public.pem --v2-keys collection_public.json python verify_bundle.py bundle.zip --hmac-key-env PRAMAGENT_AUDIT_VERIFICATION_KEY --json Checks, in order: 1. manifest signature (Ed25519 over the RFC 8785 manifest). Without ``--trusted-key`` the embedded key is used and the report says the key was not pinned: this checks internal consistency only and exits nonzero; 2. every file's SHA-256 matches the manifest, and no file is missing or extra; 3. per record: the audit-chain row names the record's hash; the chain payload matches the record field by field, with sealed values checked against their openings (salted commitments) and erased values accepted only when a receipt names the row; with ``--hmac-key`` the chain HMAC is recomputed; 4. the policy snapshot re-hashes to the record's ``policy_digest``; 5. V2 envelopes and public payload digests with independently pinned keys. Exit code 0 requires a pinned exporter AND authenticated record contents. Exit code 2 means consistent but untrusted/unattested; 1 means invalid. """ from __future__ import annotations import argparse import base64 import hashlib import hmac as _hmac import io import json import sys import time import zipfile from decimal import Decimal from typing import Any BUNDLE_DOMAIN = b"pramagent.forensic-bundle.v1\x00" SEAL_PREFIX = "sealed:v1:" SEAL_DOMAIN = b"pramagent.seal.v1\x00" POLICY_DOMAIN = b"pramagent.policy.v1\x00" CHAIN_JCS_DOMAIN = b"pramagent.audit-chain.jcs-v1\x00" STORAGE_MARKERS = {"_kid", "_canon", "_sealed", "_fss"} LINK_FIELDS = {"this_hash", "prev_hash", "anchor_tx_id", "anchor_block_number", "anchor_metadata"} # RFC 8785 (JCS), self-contained copy def _es_number(value: float) -> str: if value != value or value in (float("inf"), float("-inf")): raise ValueError("NaN/Infinity") if value == 0: return "0" if value < 0: return "-" + _es_number(-value) _sign, digits_t, exponent = Decimal(repr(value)).as_tuple() digits = "".join(str(d) for d in digits_t).rstrip("0") or "0" exponent += len(digits_t) - len(digits) k, n = len(digits), exponent + len(digits) if k <= n <= 21: return digits + "0" * (n - k) if 0 < n <= 21: return digits[:n] + "." + digits[n:] if -6 < n <= 0: return "0." + "0" * (-n) + digits e = n - 1 exp = ("+" if e >= 0 else "-") + str(abs(e)) return digits + "e" + exp if k == 1 else digits[0] + "." + digits[1:] + "e" + exp def jcs(value: Any) -> bytes: def enc(v): if v is None: return "null" if v is True: return "true" if v is False: return "false" if isinstance(v, int): if abs(v) > 9007199254740991: raise ValueError("integer out of range") return str(v) if isinstance(v, float): return _es_number(v) if isinstance(v, str): v.encode("utf-8") return json.dumps(v, ensure_ascii=False) if isinstance(v, list): return "[" + ",".join(enc(i) for i in v) + "]" if isinstance(v, dict): keys = sorted(v, key=lambda k: k.encode("utf-16-be")) return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + enc(v[k]) for k in keys) + "}" raise ValueError(f"cannot encode {type(v).__name__}") return enc(value).encode("utf-8") def chain_material(payload: dict, prev_hash: str) -> bytes: if payload.get("_canon") == "jcs-v1": return CHAIN_JCS_DOMAIN + jcs(payload) + b"|" + prev_hash.encode("utf-8") return (json.dumps(payload, sort_keys=True, separators=(",", ":")) + "|" + prev_hash).encode("utf-8") def commitment_for(value: Any, salt: bytes) -> str: try: body = jcs({"v": value}) except ValueError: body = json.dumps({"v": value}, sort_keys=True, default=str, ensure_ascii=True).encode("ascii") return SEAL_PREFIX + hashlib.sha256(SEAL_DOMAIN + salt + body).hexdigest() def _receipt_expected(record: dict) -> bool: """Input safety's allow is not a final tool authorization. An enforced tool block does not require an execution receipt. Observe-mode policy findings must not suppress missing outcomes for actions that continued. This says nothing about execution outside the governed dispatch path. """ if not record.get("tool_call") or str(record.get("pre_verdict", "")).lower() != "allow": return False if record.get("enforcement_mode") != "observe": for event in record.get("layer_events") or []: if (isinstance(event, dict) and event.get("layer") in {"ToolGuardLayer", "HostedCodingGuard"} and str(event.get("decision", "")).lower() in {"block", "blocked", "deny", "denied"}): return False return True # verification class _Report: def __init__(self) -> None: self.checks: list[dict] = [] self.records: dict[str, dict] = {} self.key_pinned = False def add(self, name: str, ok: bool | None, detail: str = "") -> None: status = "pass" if ok else ("not_checked" if ok is None else "fail") self.checks.append({"check": name, "status": status, "detail": detail}) def valid(self) -> bool: if any(c["status"] == "fail" for c in self.checks): return False return all(r["status"] != "fail" for r in self.records.values()) def _match(payload_value, record_value, openings: dict, notes: list[str], path: str) -> bool: if payload_value == record_value: return True # includes commitments stored as plain record fields if isinstance(payload_value, str) and payload_value.startswith(SEAL_PREFIX): opening = openings.get(payload_value) if opening is None: notes.append(f"{path}: sealed, no opening supplied") return record_value is None if opening.get("shredded"): # Exporter-supplied labels do not prove erasure. Without an # opening, only an undisclosed value can match this commitment. notes.append(f"{path}: sealed value unavailable; erasure not authenticated") return record_value is None try: salt = base64.b64decode(opening["salt"], validate=True) except Exception: return False if commitment_for(opening["value"], salt) != payload_value: notes.append(f"{path}: opening does not match commitment") return False return json.dumps(opening["value"], sort_keys=True) == json.dumps(record_value, sort_keys=True) if isinstance(payload_value, dict) and isinstance(record_value, dict): if set(payload_value) != set(record_value): notes.append(f"{path}: field sets differ") return False return all(_match(payload_value[k], record_value[k], openings, notes, f"{path}.{k}") for k in payload_value) if isinstance(payload_value, list) and isinstance(record_value, list): return len(payload_value) == len(record_value) and all( _match(a, b, openings, notes, f"{path}[{i}]") for i, (a, b) in enumerate(zip(payload_value, record_value))) return payload_value == record_value def _check_snapshot(commitment: str, opening, record: dict, entry: dict, fail) -> None: """Open the violation snapshot and cross-check it against the record.""" if opening is None: entry["notes"].append("violation snapshot committed; no opening in bundle") return if opening.get("shredded"): entry["notes"].append("violation snapshot erased (crypto-shredded)") return if "missing" in opening: entry["notes"].append(f"violation snapshot content unavailable: {opening['missing']}") return try: salt = base64.b64decode(opening["salt"], validate=True) except Exception: fail("violation snapshot opening is malformed") return if commitment_for(opening["value"], salt) != commitment: fail("violation snapshot does not open its commitment") return snapshot = opening["value"] prompt = snapshot.get("prompt") if prompt is not None and hashlib.sha256(prompt.encode()).hexdigest() != record.get("input_hash"): fail("snapshot prompt contradicts the record's input_hash") return recorded = {(c.get("source"), c.get("source_id")): c.get("sha256") for c in record.get("context_items") or []} for chunk in snapshot.get("context") or []: digest = hashlib.sha256((chunk.get("content") or "").encode()).hexdigest() if recorded.get((chunk.get("source"), chunk.get("source_id"))) != digest: fail(f"snapshot context {chunk.get('source_id')!r} contradicts its digest") return entry["notes"].append("violation snapshot opened and matches the record digests") def _c2sp_key_id(name: str, sig_type: int, pub: bytes) -> bytes: return hashlib.sha256(name.encode() + b"\x0a" + bytes([sig_type]) + pub).digest()[:4] def _c2sp_parse_vkey(vkey: str): name, _kid_hex, blob = vkey.split("+", 2) raw = base64.standard_b64decode(blob) return name, raw[0], raw[1:] def _c2sp_parse_note(note: str): body, _, sig_block = note.partition("\n\n") body += "\n" sigs = [] for line in sig_block.splitlines(): if line.startswith("\u2014 "): n, _, b = line[2:].partition(" ") sigs.append((n, base64.standard_b64decode(b))) return body, sigs def _c2sp_checkpoint(body: str): lines = body.split("\n") return {"origin": lines[0], "size": int(lines[1]), "root": base64.standard_b64decode(lines[2]).hex()} def verify_transparency_checkpoint(note: str, policy_text: str) -> dict: """Self-contained C2SP checkpoint check: the log signature must verify and at least ``quorum`` distinct trusted witnesses must have a valid Ed25519 cosignature over this exact checkpoint. Fails closed. Stdlib + cryptography only, so the standalone verifier keeps working.""" from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey log_vkey, witnesses, quorum = "", {}, 0 for raw in policy_text.splitlines(): parts = raw.strip().split() if not parts or parts[0].startswith("#"): continue if parts[0] == "log": log_vkey = " ".join(parts[1:]) elif parts[0] == "witness" and len(parts) >= 3: witnesses[parts[1]] = " ".join(parts[2:]) elif parts[0] == "quorum": quorum = int(parts[1]) body, sigs = _c2sp_parse_note(note) checkpoint = _c2sp_checkpoint(body) # log signature ln, lt, lpub = _c2sp_parse_vkey(log_vkey) lkid = _c2sp_key_id(ln, lt, lpub) log_ok = False for n, value in sigs: if n == ln and value[:4] == lkid: try: Ed25519PublicKey.from_public_bytes(lpub).verify(value[4:], body.encode()) log_ok = True except InvalidSignature: pass if not log_ok: return {"valid": False, "reason": "log signature does not verify", "witnesses": []} valid = set() for name, vkey in witnesses.items(): wn, wt, wpub = _c2sp_parse_vkey(vkey) wkid = _c2sp_key_id(wn, wt, wpub) for n, value in sigs: if n != wn or value[:4] != wkid or len(value) < 12: continue at = int.from_bytes(value[4:12], "big") signed = f"cosignature/v1\ntime {at}\n{body}".encode() try: Ed25519PublicKey.from_public_bytes(wpub).verify(value[12:], signed) valid.add(name) except InvalidSignature: continue ok = len(valid) >= quorum return {"valid": ok, "checkpoint": checkpoint, "witnesses": sorted(valid), "quorum": quorum, "reason": "" if ok else f"only {len(valid)} of {quorum} witnesses"} class BundleMalformed(ValueError): """The container itself is ambiguous or tampered (reject before trusting).""" def _reject_duplicate_keys(pairs): seen = {} for key, value in pairs: if key in seen: raise BundleMalformed(f"duplicate JSON key {key!r} (parser-differential risk)") seen[key] = value return seen def strict_json(raw: bytes) -> Any: """Parse JSON the way a verifier must: reject duplicate keys, NaN/Infinity and integers outside the IEEE-754 safe range (I-JSON / RFC 8785).""" def _constant(_name): raise BundleMalformed("NaN/Infinity in JSON") value = json.loads(raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys, parse_constant=_constant) def _check(node): if isinstance(node, int) and not isinstance(node, bool) and abs(node) > 2**53 - 1: raise BundleMalformed("integer outside the IEEE-754 safe range") if isinstance(node, dict): for v in node.values(): _check(v) elif isinstance(node, list): for v in node: _check(v) _check(value) return value def strict_zip(data: bytes) -> "zipfile.ZipFile": """Open a zip only if it is unambiguous: it must start at byte 0 with a local file header, carry exactly one end-of-central-directory record, and have no duplicate, absolute or traversing entry names. These are the concrete lessons of CVE-2013-4787 (duplicate names), CVE-2017-13156 (prepended data) and CVE-2025-54368 (local/central mismatch).""" if not isinstance(data, (bytes, bytearray)): raise BundleMalformed("strict_zip needs the raw bytes") if data[:4] != b"PK\x03\x04": raise BundleMalformed("bytes before the first local header (prepended data)") if data.count(b"PK\x05\x06") != 1: raise BundleMalformed("expected exactly one end-of-central-directory record") if len(data) > 64 * 1024 * 1024: raise BundleMalformed("bundle exceeds 64 MiB") archive = zipfile.ZipFile(io.BytesIO(bytes(data))) infos = archive.infolist() if len(infos) > 6000 or sum(i.file_size for i in infos) > 64 * 1024 * 1024: raise BundleMalformed("expanded bundle exceeds resource limits") names = [i.filename for i in infos] if len(names) != len(set(names)): raise BundleMalformed("duplicate entry names") for info in infos: name = info.filename if name.startswith("/") or ".." in name.split("/") or "\\" in name or "\x00" in name: raise BundleMalformed(f"unsafe entry path {name!r}") if info.flag_bits & 0x1: raise BundleMalformed("encrypted entry") # First entry must physically begin at offset 0. if min((i.header_offset for i in infos), default=0) != 0: raise BundleMalformed("first entry does not start at offset 0") return archive def _verify_bundle(source, *, trusted_public_key_pem: bytes | None = None, hmac_key: str | None = None, v2_trusted_keys=None, witness_policy: str | None = None, require_receipts: bool = False) -> dict: """Verify a bundle given as a path, bytes or file object.""" from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.serialization import ( Encoding, PublicFormat, load_pem_public_key, ) report = _Report() hybrid_verified_records = [] if isinstance(source, (bytes, bytearray)): data = bytes(source) elif isinstance(source, str): with open(source, "rb") as handle: data = handle.read() else: data = source.read() try: archive = strict_zip(data) except BundleMalformed as exc: report.add("container", False, str(exc)) return {"valid": False, "key_pinned": False, "checks": report.checks, "records": {}} names = set(archive.namelist()) for required in ("manifest.json", "manifest.sig", "public_key.pem"): if required not in names: report.add("structure", False, f"missing {required}") return {"valid": False, "key_pinned": False, "checks": report.checks, "records": {}} manifest = strict_json(archive.read("manifest.json")) ids = manifest.get("call_ids", []) if (not isinstance(ids, list) or not ids or len(ids) > 500 or len(set(ids)) != len(ids) or any(not isinstance(c, str) or not c or any(x in c for x in ("/", "\\", "..")) for c in ids)): raise BundleMalformed("invalid or empty call selection") if manifest.get("format") != "pramagent.forensic-bundle.v1": report.add("format", False, f"unknown format {manifest.get('format')!r}") # 1. signature embedded = archive.read("public_key.pem") report.key_pinned = trusted_public_key_pem is not None embedded_key = None trusted_key = None try: embedded_key = load_pem_public_key(embedded) trusted_key = (load_pem_public_key(trusted_public_key_pem) if report.key_pinned else embedded_key) trusted_key.verify(base64.b64decode(archive.read("manifest.sig")), BUNDLE_DOMAIN + jcs(manifest)) report.add("manifest_signature", True, "pinned key" if report.key_pinned else "embedded key only: internal consistency, untrusted origin") except (InvalidSignature, TypeError, ValueError) as exc: report.add("manifest_signature", False, f"{type(exc).__name__}") if report.key_pinned and embedded_key is not None and trusted_key is not None: # PEM is a transport representation: line endings and wrapping are not # key identity. Compare canonical SubjectPublicKeyInfo DER instead so # a Windows CRLF checkout verifies the same cryptographic key while a # real key substitution still fails closed. embedded_spki = embedded_key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo) trusted_spki = trusted_key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo) report.add("embedded_key_matches_pinned", embedded_spki == trusted_spki, "same public-key identity" if embedded_spki == trusted_spki else "bundle carries a different key") # 2. file digests listed = manifest.get("files", {}) extra = names - set(listed) - {"manifest.json", "manifest.sig"} bad = [p for p, digest in listed.items() if p not in names or hashlib.sha256(archive.read(p)).hexdigest() != digest] report.add("file_digests", not bad and not extra, f"altered/missing: {bad[:5]} extra: {sorted(extra)[:5]}" if bad or extra else f"{len(listed)} files") journal_authenticated = True if "case_journal.json" in names or "case_journal_anchor.json" in names: journal_authenticated = False try: journal = strict_json(archive.read("case_journal.json")) anchor = strict_json(archive.read("case_journal_anchor.json")) payload = anchor["payload"] digest = hashlib.sha256(jcs(journal)).hexdigest() annotations = strict_json(archive.read("annotations.json")) findings = strict_json(archive.read("case_findings.json")) bound = (payload["record_type"] == "pramagent.custody.v1" and payload["tenant_id"] == manifest["tenant_id"] and payload["case_id"] == manifest["case_id"] and payload["detail"]["journal_digest"] == digest and annotations == journal["annotations"] and findings == {"rule":journal["rule"],"findings":journal["findings"]}) report.add("case_journal_binding", bound, "journal, annotations and findings match audit anchor") if hmac_key: journal_authenticated = _hmac.compare_digest(_hmac.new(hmac_key.encode(), chain_material(payload, anchor["prev_hash"]), hashlib.sha256).hexdigest(), anchor["this_hash"]) report.add("case_journal_hmac", journal_authenticated, "audit anchor authentication") if "v2-case-journal.json" in names: try: envelope = strict_json(archive.read("v2-case-journal.json")) record = {**payload,"this_hash":anchor["this_hash"],"prev_hash":anchor["prev_hash"]} material = {k:v for k,v in payload.items() if k not in STORAGE_MARKERS} verify_v2_offline(envelope,record,v2_trusted_keys,chain_payload=material) journal_authenticated |= bool(v2_trusted_keys and envelope["record"].get("audit_payload_sha256")) report.add("case_journal_checkpoint", True, "selected journal anchor inclusion; not completeness") except Exception as exc: # fail closed without exposing evidence report.add("case_journal_checkpoint", False, type(exc).__name__) elif v2_trusted_keys and not hmac_key: report.add("case_journal_checkpoint", False, "public journal proof is missing") except (KeyError, TypeError, ValueError): report.add("case_journal_binding", False, "missing or malformed journal audit anchor") receipts = strict_json(archive.read("receipts.json")) if "receipts.json" in names else [] if (not isinstance(receipts, list) or any(not isinstance(row, dict) or not isinstance(row.get('payload'), dict) or not all(isinstance(row.get(k), str) for k in ('this_hash', 'prev_hash')) or len(row['this_hash']) != 64 or any(c not in '0123456789abcdef' for c in row['this_hash']) or row['payload'].get('tenant_id') != manifest.get('tenant_id') for row in receipts) or len({row['this_hash'] for row in receipts}) != len(receipts)): raise BundleMalformed('invalid receipt collection') receipt_authentication = {} for row in receipts: digest = row['this_hash'] authenticated = False if hmac_key: authenticated = _hmac.compare_digest(_hmac.new(hmac_key.encode(), chain_material(row['payload'], row['prev_hash']), hashlib.sha256).hexdigest(), digest) if not authenticated: report.add('receipt_hmac', False, digest[:16]) v2_path = f'v2-receipts/{digest}.json' if v2_path in names: try: envelope = strict_json(archive.read(v2_path)) record = {**row['payload'], 'this_hash': digest, 'prev_hash': row['prev_hash']} record.setdefault('session_id', 'default') material = {k: v for k, v in row['payload'].items() if k not in STORAGE_MARKERS} verify_v2_offline(envelope, record, v2_trusted_keys, chain_payload=material) authenticated |= bool(v2_trusted_keys and envelope['record'].get('audit_payload_sha256')) except Exception as exc: report.add('receipt_checkpoint', False, type(exc).__name__) receipt_authentication[digest] = authenticated receipt_rows = [row for row in receipts if row.get("payload", {}).get("record_type") == "pramagent.erasure_receipt.v1"] receipted = {h for row in receipt_rows for h in row["payload"].get("erased_hashes", [])} tombstones = {h: d for row in receipt_rows for h, d in (row["payload"].get("tombstones") or {}).items()} for row in receipt_rows: if not receipt_authentication[row['this_hash']]: receipted -= set(row['payload'].get('erased_hashes', [])) # 3-5. per record for call_id in manifest.get("call_ids", []): entry = {"status": "pass", "notes": [], "content_authenticated": False, "receipt_present": False, "receipt_contents_authenticated": False} report.records[call_id] = entry def fail(message: str, _entry=entry) -> None: _entry["status"] = "fail" _entry["notes"].append(message) try: record = strict_json(archive.read(f"records/{call_id}.json")) chain = strict_json(archive.read(f"chain/{call_id}.json")) except KeyError as exc: fail(f"missing file {exc}") continue except BundleMalformed as exc: fail(f"malformed record json: {exc}") continue if chain.get("missing"): fail("record hash not found in the audit chain at export") continue if record.get("tenant_id") != manifest.get("tenant_id") or record.get("call_id") != call_id: fail("record identity or tenant does not match manifest") if chain.get("prev_hash") != record.get("prev_hash"): fail("record predecessor does not match chain row") if chain["this_hash"] != record.get("this_hash"): fail("chain row does not carry the record's hash") entry['receipt_expected'] = _receipt_expected(record) openings = {} if f"openings/{call_id}.json" in names: openings = strict_json(archive.read(f"openings/{call_id}.json")) for row in receipts: result = row.get("payload", {}) if (result.get("record_type") == "pramagent.tool_result.v1" and result.get("call_id") == call_id): entry['receipt_present'] = True entry['receipt_contents_authenticated'] = receipt_authentication[row['this_hash']] bound = (result.get("trace_hash") == record.get("this_hash") and result.get("arguments_digest") == (record.get("tool_call") or {}).get("arguments_digest")) if not bound: fail("tool result is not bound to this record's hash and arguments") else: entry["notes"].append( f"host-reported tool outcome: {result.get('status')}, " f"output sha256 {str(result.get('output_sha256', ''))[:16]}") snapshot_commitment = record.get("snapshot_commitment") if snapshot_commitment: _check_snapshot(snapshot_commitment, openings.get(snapshot_commitment), record, entry, fail) payload = {k: v for k, v in chain["payload"].items() if k not in STORAGE_MARKERS} fields = {k: v for k, v in record.items() if k not in LINK_FIELDS} notes: list[str] = [] consistent = _match(payload, fields, openings, notes, "$") entry["notes"].extend(notes) erased = bool(chain["payload"].get("gdpr_erased")) if erased: expected = tombstones.get(chain["this_hash"]) actual = hashlib.sha256(json.dumps(chain["payload"], sort_keys=True, separators=(",", ":")).encode()).hexdigest() if expected is not None and expected != actual: fail("erased row was modified after its erasure receipt") elif chain["this_hash"] in receipted: entry["status"] = "erased" entry["notes"].append("row tombstoned under a receipted GDPR erasure") else: fail("row marked erased without a receipt") elif not consistent: fail("chain payload does not match the record") if hmac_key and not erased: expected = _hmac.new(hmac_key.encode(), chain_material(chain["payload"], chain["prev_hash"]), hashlib.sha256).hexdigest() if expected != chain["this_hash"]: fail("HMAC does not match (wrong key or altered row)") else: entry["notes"].append("chain HMAC recomputed") entry["content_authenticated"] = True digest = record.get("policy_digest") policy_path = f"policies/{digest}.json" if digest and policy_path not in names: fail("pinned policy snapshot missing from bundle") if digest and policy_path in names: snapshot = strict_json(archive.read(policy_path)) if hashlib.sha256(POLICY_DOMAIN + jcs(snapshot)).hexdigest() != digest: fail("policy snapshot does not hash to policy_digest") else: entry["notes"].append("policy snapshot matches policy_digest") v2_path = f"v2/{call_id}.json" if v2_path in names: try: envelope = strict_json(archive.read(v2_path)) verify_v2_offline(envelope, record, v2_trusted_keys, chain_payload=payload) if v2_trusted_keys: # verify_v2_offline requires both algorithms under pinned # keys. Never promote an embedded algorithm label to proof. hybrid_verified_records.append(call_id) if v2_trusted_keys and envelope["record"].get("audit_payload_sha256"): entry["content_authenticated"] = True entry["notes"].append("V2 inclusion and record binding verified" + (" against trusted checkpoint keys" if v2_trusted_keys else "; checkpoint signature not checked (no trusted keys)")) except Exception as exc: fail(f"V2 verification failed: {type(exc).__name__}") elif v2_trusted_keys: fail("trusted checkpoint verification requested but V2 envelope is missing") if require_receipts and entry['receipt_expected']: if not entry['receipt_present'] or not entry['receipt_contents_authenticated']: fail('authenticated execution receipt required but missing or unattested') # 6. transparency checkpoint against a pinned k-of-n witness policy. # When a policy is pinned, a missing or under-quorum checkpoint fails the # bundle (fail closed); without a pinned policy it is advisory only. if "log_checkpoint.note" in names: note = archive.read("log_checkpoint.note").decode("utf-8", "replace") policy_text = witness_policy if policy_text is None and "witness_policy.txt" in names: policy_text = archive.read("witness_policy.txt").decode("utf-8", "replace") if witness_policy is None: report.add("transparency_checkpoint", True, "checkpoint present; pin a witness policy to enforce quorum") else: try: result = verify_transparency_checkpoint(note, witness_policy) report.add("transparency_checkpoint", result["valid"], f"{len(result['witnesses'])} of {result['quorum']} witnesses" if result["valid"] else result["reason"]) except Exception as exc: # noqa: BLE001 - fail closed report.add("transparency_checkpoint", False, f"{type(exc).__name__}: {exc}") elif witness_policy is not None: report.add("transparency_checkpoint", False, "policy pinned but no checkpoint in bundle") integrity_valid = report.valid() contents_authenticated = bool(report.records) and all( entry["content_authenticated"] and entry["status"] == "pass" for entry in report.records.values()) receipts_authenticated = all(receipt_authentication.values()) trusted = integrity_valid and report.key_pinned and contents_authenticated and receipts_authenticated and journal_authenticated missing_receipts = [call for call, entry in report.records.items() if entry.get('receipt_expected') and not entry['receipt_present']] return {"valid": trusted, "integrity_valid": integrity_valid, "status": ("verified_decisions_only" if missing_receipts else "verified") if trusted else "unattested" if integrity_valid else "invalid", "record_contents_authenticated": integrity_valid and contents_authenticated, "case_journal_authenticated": journal_authenticated if "case_journal.json" in names else None, "receipt_contents_authenticated": (integrity_valid and receipts_authenticated) if receipts else None, "receipt_count": len(receipts), "receipt_authentication_status": ('none_present' if not receipts else 'authenticated' if receipts_authenticated else 'unattested'), "missing_execution_receipt_call_ids": missing_receipts, "receipt_completeness_proven": False, "execution_proven": False, "receipt_requirement_enforced": require_receipts, "key_pinned": report.key_pinned, "trusted_exporter": report.key_pinned and report.valid(), "completeness_proven": False, "checkpoint_signatures_checked": bool(v2_trusted_keys), "verified_checkpoint_signature_algorithms": ["Ed25519", "ML-DSA-65"] if hybrid_verified_records else [], "checkpoint_signature_verified_call_ids": hybrid_verified_records, "checkpoint_signatures_verified_for_all_records": (integrity_valid and bool(report.records) and len(hybrid_verified_records) == len(report.records)), "bundle_id": manifest.get("bundle_id"), "tenant_id": manifest.get("tenant_id"), "checks": report.checks, "records": report.records} def verify_v2_offline(envelope, record, trusted_keys=None, *, chain_payload=None): """Verify one V2 epoch inclusion without importing Pramagent. Checkpoint chaining/completeness and external timestamps are separate assurances. Missing optional crypto dependencies fail when keys are supplied. """ def need(condition): if not condition: raise BundleMalformed("invalid V2 evidence") def sha(value): return hashlib.sha256(jcs(value)).hexdigest() leaf = dict(envelope["leaf"]) leaf_hash = leaf.pop("leaf_hash") need(leaf["domain"] == "pramagent:evidence-leaf:v2" and leaf["schema_version"] == "2.0") need(leaf["provenance_mode"] == "native-v2" and leaf["record_hash_algorithm"] == "SHA-256") need(len(base64.b64decode(leaf["nonce_b64"], validate=True)) >= 16) need(hashlib.sha256(b"\x00" + jcs(leaf)).hexdigest() == leaf_hash) need(leaf["record_id"] == record["this_hash"] and sha(envelope["record"]) == leaf["record_digest"]) link = envelope["record"] need(link["audit_chain_hash"] == record["this_hash"] and link["audit_prev_hash"] == record["prev_hash"]) need(link["tenant_id"] == record["tenant_id"] and link["session_id"] == record["session_id"]) # A signed opaque HMAC reference alone cannot authenticate its plaintext to # an outside verifier. New checkpoints bind a public digest of the payload. if link.get("audit_payload_sha256"): material = chain_payload if chain_payload is not None else { k: v for k, v in record.items() if k not in LINK_FIELDS | STORAGE_MARKERS} need(sha(material) == link["audit_payload_sha256"]) signed = envelope["checkpoint"] cp = signed["checkpoint"] need(sha({"checkpoint": cp, "signatures": signed["signatures"]}) == signed["checkpoint_hash"]) need(cp["domain"] == "pramagent:evidence-checkpoint:v2" and cp["schema_version"] == "2.0") need(cp["hash_algorithm"] == "SHA-256" and cp["merkle_algorithm"] == "PRAMAGENT-MERKLE-SHA256-V1") need(cp["canonicalization"] == "RFC8785-JCS-INTEGER-PROFILE") size = cp["tree_size"] index = leaf["sequence"] - cp["first_sequence"] need(type(size) is int and 0 < size <= 2**53 - 1 and 0 <= index < size) need(cp["last_sequence"] - cp["first_sequence"] + 1 == size) proof = iter(envelope["inclusion_proof"]) def rebuild(position, count): if count == 1: return bytes.fromhex(leaf_hash) split = 1 << ((count - 1).bit_length() - 1) if position < split: left = rebuild(position, split) right = bytes.fromhex(next(proof)) else: right = rebuild(position - split, count - split) left = bytes.fromhex(next(proof)) need(len(left) == 32 and len(right) == 32) return hashlib.sha256(b"\x01" + left + right).digest() need(rebuild(index, size).hex() == cp["merkle_root"]) need(next(proof, None) is None) if not trusted_keys: return required = ["Ed25519", "ML-DSA-65"] need(cp["signature_policy_version"] == "pramagent-hybrid-2026-01") need(cp["required_signature_algorithms"] == required) signatures = signed["signatures"] need(len(signatures) == 2 and {s["algorithm"] for s in signatures} == set(required)) message = b"pramagent:evidence-checkpoint:v2\x00" + jcs(cp) for sig in signatures: key = trusted_keys[(sig["algorithm"], sig["key_id"])] signature = base64.b64decode(sig["signature_b64"], validate=True) if sig["algorithm"] == "Ed25519": from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey Ed25519PublicKey.from_public_bytes(key).verify(signature, message) else: from pqcrypto.sign import ml_dsa_65 # pqcrypto >= 1 raises on invalid signatures and returns None on # success. Also accept the explicit True used by earlier bindings. verified = ml_dsa_65.verify(key, message, signature) need(verified is None or verified is True) IMPORT_BUNDLE_FORMAT = "pramagent.import-bundle.v1" IMPORT_BUNDLE_DOMAIN = b"pramagent.import-bundle.v1\x00" SEAL_RECORD = "pramagent.transcript_seal.v1" _REWRITES = ("prefix_modified", "truncated") def _sealed_history(seal_rows: list, record: dict, manifest: dict, original: bytes | None, report: _Report) -> dict: """Check the sealing collector's seals of this import's source file. Each seal attests that the file's first ``size`` bytes had digest ``sha256`` when it was made. With the original embedded, every seal that fits inside it is recomputed from its bytes. Collection times describe observations, not write times: an unfinished line may predate the first pass, and changes between reads can be invisible. """ if not seal_rows: if manifest.get("seal_count"): report.add("seal_records", False, "manifest lists seals the bundle does not carry") return {"status": "invalid", "seals": 0} return {"status": "unsealed", "seals": 0} seals = [row.get("payload") or {} for row in seal_rows] path = seals[0].get("path_sha256") same_file = all(s.get("record_type") == SEAL_RECORD and s.get("tenant_id") == manifest.get("tenant_id") and s.get("namespace") == record.get("namespace") and s.get("adapter") == record.get("adapter") and s.get("path_sha256") == path and s.get("content_copied") is False for s in seals) linked = seals[0].get("previous") is None and all( isinstance(seals[i].get("previous"), dict) and seals[i]["previous"].get("chain_hash") == seal_rows[i - 1].get("this_hash") and seals[i]["previous"].get("size") == seals[i - 1].get("size") and seals[i]["previous"].get("sha256") == seals[i - 1].get("sha256") for i in range(1, len(seals))) own = next((i for i, s in enumerate(seals) if s.get("present") and s.get("import_id") == manifest.get("import_id") and s.get("sha256") == manifest.get("original_sha256") and s.get("size") == (record.get("original") or {}).get("bytes")), None) ok = same_file and linked and own is not None and len(seals) == manifest.get("seal_count") report.add("seal_records", ok, f"{len(seals)} seal(s) of one file, linked in order" if ok else "seals are foreign, unlinked, miscounted or do not cover this import") if not ok: return {"status": "invalid", "seals": len(seals)} history = seals[:own + 1] attested = [s.get("sealed_at") for s in history if s.get("status") in _REWRITES] contradicted = [] if original is not None: contradicted = [s.get("sealed_at") for s in history if s.get("present") and isinstance(s.get("size"), int) and s["size"] <= len(original) and hashlib.sha256(original[:s["size"]]).hexdigest() != s.get("sha256")] windows, last_read, last_size = [], None, 0 for s in history: if not s.get("present"): continue if s.get("status") in ("grown", "reappeared") and last_read is not None and s["size"] > last_size: windows.append(s["sealed_at"] - last_read) last_read, last_size = s.get("read_started_at"), s["size"] return {"status": "rewritten" if attested or contradicted else "append_only", "basis": "original_bytes" if original is not None else "collector_records", "seals": len(history), "seals_after_import": len(seals) - len(history), "first_sealed_at": history[0].get("sealed_at"), "bytes_at_first_seal": history[0].get("size"), "import_sealed_at": history[-1].get("sealed_at"), # Keep the old field for readers, but never manufacture a write-time bound. "max_unsealed_seconds": None, "max_observation_interval_seconds": round(max(windows), 3) if windows else None, "write_time_bound_proven": False, "clock_source": "collector_machine_clock_not_independently_timestamped", "rewrites_attested_at": attested[:10], "contradicted_seals_at": contradicted[:10], "later_source_changes": sorted({s.get("status") for s in seals[own + 1:] if s.get("status") in (*_REWRITES, "missing")})} def _describe_sealed_history(history: dict) -> str: status = history.get("status") if status == "unsealed": return "Sealed history: none; the source file was not sealed by `pramagent forensics collect`." if status == "invalid": return "Sealed history: INVALID seal records." basis = ("recomputed from the original" if history["basis"] == "original_bytes" else "as attested by the collector; embed the original to recompute") if status == "rewritten": text = f"Sealed history: REWRITTEN. Bytes sealed earlier were changed before this import ({basis})" else: stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(history["import_sealed_at"])) text = (f"Sealed history: append-only across {history['seals']} seal(s) ({basis}); " f"these exact bytes were sealed by {stamp}") if history.get("max_observation_interval_seconds") is not None: text += (f"; maximum recorded growth-observation interval " f"{history['max_observation_interval_seconds']:.0f} s") text += "; write times and changes between observations are not proven" if history["later_source_changes"]: text += f". After this import the source was: {', '.join(history['later_source_changes'])}" return text + "." def _verify_import_bundle(data: bytes, *, trusted_public_key_pem: bytes | None = None, hmac_key: str | None = None, **_ignored) -> dict: """Verify a signed transcript/hook-log import (``pramagent forensics import-bundle``). Valid means: a pinned exporter signed this exact set of files, every record's chain HMAC recomputes, and every record belongs to this import. Imported events remain host claims made before import; the HMAC authenticates them from import onward, not their truth or completeness. """ from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.serialization import ( Encoding, PublicFormat, load_pem_public_key, ) report = _Report() archive = strict_zip(data) names = set(archive.namelist()) for required in ("manifest.json", "manifest.sig", "public_key.pem", "import.json", "events.jsonl"): if required not in names: report.add("structure", False, f"missing {required}") return {"valid": False, "integrity_valid": False, "key_pinned": False, "records": {}, "checks": report.checks, "format": IMPORT_BUNDLE_FORMAT} manifest = strict_json(archive.read("manifest.json")) report.key_pinned = trusted_public_key_pem is not None try: embedded = load_pem_public_key(archive.read("public_key.pem")) trusted = load_pem_public_key(trusted_public_key_pem) if report.key_pinned else embedded trusted.verify(base64.b64decode(archive.read("manifest.sig")), IMPORT_BUNDLE_DOMAIN + jcs(manifest)) report.add("manifest_signature", True, "pinned key" if report.key_pinned else "embedded key only: untrusted origin") if report.key_pinned: same = (embedded.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo) == trusted.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)) report.add("embedded_key_matches_pinned", same, "same public-key identity" if same else "bundle carries a different key") except (InvalidSignature, TypeError, ValueError) as exc: report.add("manifest_signature", False, type(exc).__name__) listed = manifest.get("files", {}) extra = names - set(listed) - {"manifest.json", "manifest.sig"} bad = [p for p, digest in listed.items() if p not in names or hashlib.sha256(archive.read(p)).hexdigest() != digest] report.add("file_digests", not bad and not extra, f"altered/missing: {bad[:5]} extra: {sorted(extra)[:5]}" if bad or extra else f"{len(listed)} files") import_row = strict_json(archive.read("import.json")) lines = [line for line in archive.read("events.jsonl").decode("utf-8").split("\n") if line] event_rows = [strict_json(line.encode("utf-8")) for line in lines] record = import_row.get("payload", {}) identity_ok = (record.get("record_type") == "pramagent.evidence_import.v1" and record.get("import_id") == manifest.get("import_id") and record.get("tenant_id") == manifest.get("tenant_id") and (record.get("original") or {}).get("sha256") == manifest.get("original_sha256") and record.get("native_decisions_minted") == 0) report.add("import_identity", identity_ok, str(manifest.get("import_id"))) wrong = [r.get("this_hash", "")[:16] for r in event_rows if (r.get("payload") or {}).get("record_type") != "pramagent.imported_tool_event.v1" or r["payload"].get("import_id") != manifest.get("import_id") or r["payload"].get("tenant_id") != manifest.get("tenant_id") or r["payload"].get("native_decision") is not False] count_ok = len(event_rows) == manifest.get("event_count") == (record.get("summary") or {}).get("appended_events") report.add("event_identity", not wrong and count_ok, f"{len(event_rows)} events" if not wrong and count_ok else f"foreign or miscounted: {wrong[:5]}") original = archive.read("original/source.jsonl") if "original/source.jsonl" in names else None if original is not None: report.add("original_digest", hashlib.sha256(original).hexdigest() == manifest.get("original_sha256"), "original file matches the imported digest") seal_lines = archive.read("seals.jsonl").decode("utf-8").split("\n") if "seals.jsonl" in names else [] seal_rows = [strict_json(line.encode("utf-8")) for line in seal_lines if line] sealed_history = _sealed_history(seal_rows, record, manifest, original, report) authenticated = False if hmac_key: rows = [import_row, *event_rows, *seal_rows] failed = [r.get("this_hash", "")[:16] for r in rows if _hmac.new(hmac_key.encode(), chain_material(r["payload"], r["prev_hash"]), hashlib.sha256).hexdigest() != r.get("this_hash")] report.add("record_hmac", not failed, f"{len(rows)} records" if not failed else f"mismatch: {failed[:5]}") authenticated = not failed else: report.add("record_hmac", None, "supply --hmac-key to authenticate record contents") integrity = report.valid() trusted_exporter = report.key_pinned and integrity return {"format": IMPORT_BUNDLE_FORMAT, "valid": integrity and trusted_exporter and authenticated, "integrity_valid": integrity, "key_pinned": report.key_pinned, "trusted_exporter": trusted_exporter, "record_contents_authenticated": authenticated, "checks": report.checks, "records": {}, "import_id": manifest.get("import_id"), "event_count": len(event_rows), "source_trust": manifest.get("source_trust"), "completeness_proven": False, "sealed_history": sealed_history, "note": "Imported events are host claims collected after the fact; authentication starts at import, " "or at each seal for sealed bytes. A seal does not show the bytes were true when written."} def verify_bundle(source, **kwargs): try: if isinstance(source, (bytes, bytearray)): data = bytes(source) elif isinstance(source, str): with open(source, "rb") as handle: data = handle.read() else: data = source.read() try: manifest = strict_json(strict_zip(data).read("manifest.json")) except Exception: # malformed input is reported by the main verifier manifest = None if isinstance(manifest, dict) and manifest.get("format") == IMPORT_BUNDLE_FORMAT: if kwargs.get('require_receipts'): raise BundleMalformed('Imported records cannot satisfy native execution receipt requirements') return _verify_import_bundle(data, **kwargs) return _verify_bundle(data, **kwargs) except (ValueError, TypeError, KeyError, IndexError, AttributeError, OSError, RecursionError, zipfile.BadZipFile, RuntimeError, StopIteration) as exc: return {"valid": False, "key_pinned": False, "records": {}, "checks": [{"check": "structure", "status": "fail", "detail": type(exc).__name__}]} def main(argv=None) -> int: parser = argparse.ArgumentParser(description="Verify a Pramagent forensic bundle") parser.add_argument("bundle") parser.add_argument("--trusted-key", help="PEM public key of the exporter to pin") parser.add_argument("--hmac-key", help="audit-chain signing key, to recompute HMACs") parser.add_argument("--hmac-key-env", help="Environment variable containing a dedicated audit verification secret; its holder can also forge HMACs") parser.add_argument("--require-receipts", action="store_true", help="Require authenticated receipts for selected allowed tool decisions (not whole-history completeness)") parser.add_argument("--v2-keys", help="Trusted checkpoint key registry JSON") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) pinned = open(args.trusted_key, "rb").read() if args.trusted_key else None keys = None if args.v2_keys: with open(args.v2_keys, "rb") as handle: entries = strict_json(handle.read()) if isinstance(entries, dict): entries = entries["keys"] keys = {(e["algorithm"], e["key_id"]): base64.b64decode(e["public_key_b64"], validate=True) for e in entries} if not keys: parser.error("trusted key registry is empty") if args.hmac_key and args.hmac_key_env: parser.error('choose only one HMAC key source') audit_key = args.hmac_key if args.hmac_key_env: import os audit_key = os.environ.get(args.hmac_key_env) if not audit_key: parser.error('the named HMAC key environment variable is empty') result = verify_bundle(args.bundle, trusted_public_key_pem=pinned, hmac_key=audit_key, v2_trusted_keys=keys, require_receipts=args.require_receipts) if args.json: print(json.dumps(result, indent=2, sort_keys=True)) else: print(("DECISIONS VERIFIED; EXECUTION RECEIPTS MISSING" if result.get('missing_execution_receipt_call_ids') else "VERIFIED") if result["valid"] else "UNATTESTED" if result.get("integrity_valid") else "INVALID") if not result["key_pinned"]: print(" Exporter identity is not trusted: supply an independently obtained --trusted-key.") if result.get("format") == IMPORT_BUNDLE_FORMAT: if not result.get("record_contents_authenticated"): print(" Imported records are not authenticated: supply the evidence chain key with --hmac-key.") if "sealed_history" in result: print(" " + _describe_sealed_history(result["sealed_history"])) print(" " + result["note"]) elif not result.get("record_contents_authenticated"): print(" Record contents lack verified collection evidence. Supply independently trusted checkpoint keys.") if result.get('receipt_contents_authenticated') is False: print(' Receipt contents are unattested; exporter signing alone does not authenticate outcomes.') if result.get('checkpoint_signature_verified_call_ids'): print(' Checkpoint signatures verified: Ed25519 + ML-DSA-65 for ' f"{len(result['checkpoint_signature_verified_call_ids'])} of {len(result['records'])} selected records. " 'The export manifest uses Ed25519 separately.') elif result.get("format") != IMPORT_BUNDLE_FORMAT: print(' Hybrid checkpoint signatures are not authenticated by this verification.') print(' Receipt/history completeness and independently proven execution are not established.') for check in result["checks"]: print(f" [{check['status']}] {check['check']}: {check['detail']}") for call_id, entry in result["records"].items(): print(f" record {call_id}: {entry['status']}") for note in entry["notes"]: print(f" - {note}") return 0 if result["valid"] else 2 if result.get("integrity_valid") else 1 if __name__ == "__main__": sys.exit(main())