"""VeilEngine Receipt Verifier -- standalone auditor-side CLI. SPDX-License-Identifier: MIT Customer-facing single-file tool. The third-party-auditability property of VeilEngine evidence rests on the customer being able to verify an artifact WITHOUT trusting the same codebase that signed it. This script imports no Vertical Edge AI source and makes no project-layout assumptions. It performs no network I/O: every input is a local file. Dependencies: cryptography Ed25519 verification (always required) rfc8785 RFC 8785 JSON Canonicalization, required only for artifacts stamped canonicalization_version rfc8785-jcs-v1 pip install cryptography rfc8785 Modes: python verifier_cli.py --receipt receipt.json --public-key-file public_key.json python verifier_cli.py --manifest manifest.json --public-key-file public_key.json python verifier_cli.py --chain session_export.json --public-key-file public_key.json python verifier_cli.py --session-export session_export.json --public-key-file public_key.json python verifier_cli.py --session-export session_export.json --public-keyring-file public_keyring.json python verifier_cli.py --self-test Exit codes: 0 PASS | 1 FAIL | 2 argument or environment error. Input handling. Every mode reads its input through one guarded loader and then asserts the SHAPE of what it got, before any verification runs. A missing file, unreadable bytes, non-UTF-8 bytes, malformed JSON, a JSON object that names one key twice, or a root that is not what the mode verifies is an ENVIRONMENT fault: `[ENV]` on stderr and exit 2, never exit 1 and never a traceback. Exit 1 means an artifact was checked and did not verify, and nothing else. --receipt a JSON object declaring a non-empty `receipt_id` --manifest a JSON object whose `artifacts` is a NON-EMPTY list of {path, sha256, bytes} descriptors, each path inside the MANIFEST-PATH GRAMMAR below --chain a JSON array of events, an {events: [...]} wrapper, or a session export carrying chain.events --session-export a JSON object carrying a `chain`, optionally inside the {"veilengine_session_export": {...}} wrapper Duplicate JSON keys are refused BY KEY NAME. Python's parser is last-wins, first-wins parsers and human readers take the first, and a signature covers one of them: a file whose text opens with a different session id than the one that was signed is two documents, not one, and this verifier will not choose between them. `--self-test` cannot be combined with any artifact mode: it checks the verifier, not an artifact, and one verdict cannot answer both questions. Manifest-path grammar. One grammar, enforced identically here and in the site's build (`scripts/build-render-publish.js`), so a package that builds is a package that verifies. A descriptor path is RELATIVE and `/`-separated; NFC-normalized; every segment matches `^[A-Za-z0-9][A-Za-z0-9._-]*$`; no `.` or `..` segment; no segment ending in `.` or a space; no `:` anywhere; and no Windows reserved device name (CON, PRN, AUX, NUL, COM1-COM9, LPT1-LPT9, in any case, with or without an extension). The rules past "relative" exist because STRING-LEVEL UNIQUENESS IS NOT FILESYSTEM-OBJECT UNIQUENESS: Win32 strips a trailing dot or space, so `README.md.` and `README.md` are one file addressed by two descriptors; `README.md:x` is an NTFS stream of README.md; `NUL.txt` opens the null device; and two Unicode spellings of one name are one file on macOS. Because a grammar can only close the aliases decidable from the string, the manifest check ALSO requires unique resolved file identity -- (st_dev, st_ino) across every descriptor target, after the no-symlink walk -- so a hard link that gives one file two in-grammar names is refused by name too. Canonicalization. A signed object may carry a `canonicalization_version` field. Absent means the frozen historical format `ve-canonical-json-v1` (json.dumps with sort_keys=True, separators=(",", ":"), ensure_ascii=False). `rfc8785-jcs-v1` means RFC 8785 JCS. Any other identifier is REJECTED by name. This verifier never silently falls back to another format: a silent fallback would check a signature against bytes the signer never produced. Chain. `--chain` walks prev_event_hash from the 64-zero genesis sentinel, recomputes each event's metadata_hash, and verifies each event signature, halting on the FIRST broken link so the diagnostic names the exact event. Binding. Each layer of a session export self-verifies, but self-verification alone does not stop a validly signed chain, event or receipt from ANOTHER session being mixed into this one. `--session-export` therefore also binds the layers, under one explicit, fail-closed schema contract: EVERY LAYER THAT CAN CARRY IDENTITY MUST CARRY IT, AND EVERY IDENTITY IT CARRIES MUST MATCH. Concretely: `bundle.session_id`, `bundle.tenant_id`, `chain.session_id` and `chain.tenant_id` are all REQUIRED; every event must name the bundle's session and the bundle's tenant; every receipt must name both as well. An absent identity on any of those layers is a rejection, not a skipped comparison: an export bound to a tenant that some layer never declared is an export whose tenant claim is weaker than it reads. This is deliberately stricter than the schemas that merely permit the field, and it is stricter than the gateway-side verifier this tool is a subset of, which leaves `bundle.tenant_id` and the chain wrapper's tenant unrequired. The explicit-key `signing_key_id` binding that this tool used to omit is now aligned with that reference (see Key identity below), so the remaining divergences are the two stricter ones and the conditional derived-id rule under Keyring. Identity paths are declared in RECEIPT_SESSION_PATHS and RECEIPT_TENANT_PATHS below: nested under the receipt's own request object in the schema shipped here, or at the top level in the gateway schema. EVERY declared path is read and they must AGREE. There is no first-non-empty selection: a receipt that names one session nested and a different session at the top level is refused by name rather than admitted under whichever path this verifier happened to consult first. A receipt that names neither is likewise rejected rather than skipped -- an unbindable object inside a bound export is exactly the cross-session mix this layer exists to catch. Identity is TYPED, and presence is decided by key membership rather than by a null-valued lookup. An identity is a NON-EMPTY STRING. A path that is present but holds a number, an array, an object, a boolean, null or the empty string is a schema violation, not an absence: it is refused by name, printing the layer, the path and the offending type, on EVERY identity-bearing layer -- bundle, chain wrapper, event and receipt alike. Treating a non-string declaration as absence let a receipt whose own request block named `["sess_other"]` be admitted under a planted top-level string while the binding line reported that every object declared one session. The success diagnostic names only the layers and the identities that were actually compared, so a reader cannot infer a comparison that did not happen. Key identity. The key id a signed object names is held to the SAME contract as its session and tenant: `signing_key_id`, `signer_key_id` and `verification.signer_key_id` are all read, every one present must agree, and a conflict is refused by name. Selecting the first non-empty path let a receipt name the active key at one path and a REVOKED key at another, verify, and print neither. Explicit single-key mode binds that declaration to the key it was handed, as the gateway-side reference does: a declared id in the DERIVED SHAPE must equal the supplied key's fingerprint. This package's ids are human-readable LABELS, which no derivation can produce and no computation can bind to a raw key, so a label id is instead PRINTED on every layer line -- the `[STATUS] component` lines of `--session-export` and the result line of `--receipt`, `--manifest` and `--chain` -- and `--expect-key-id` turns that reading into an assertion enforced on every layer, in both modes. Under a pinned keyring the label is already the selector, so it is not repeated. Chain anchor. `chain.anchor` is the signed commitment to the chain TIP: the algorithm, the event count, and the last event hash. Without it an export can be truncated to a shorter prefix of its own chain and every remaining layer still verifies. Keyring verification therefore REQUIRES a structured anchor and recomputes it. Explicit single-key verification keeps a narrow, documented legacy exception: bundles frozen before the anchor existed stay verifiable, and the exception is reported as SKIP rather than silently passed. When an anchor is present it is ALWAYS recomputed and must match exactly, in both modes. Chain events may declare their own `event_hash`. Declaring it is optional; declaring it falsely is not. When present it is RECOMPUTED -- sha256 over the canonical event minus its own hash and its signature, the same value that links the next event -- and must match exactly. It sits inside the signed body, so a false one is a signed lie about the event, and nothing compared it before 2026-09-02. A session export delivered inside the `{"veilengine_session_export": {...}}` wrapper may carry NOTHING beside that key. Sibling fields sit outside the signed envelope: no signature covers them, so a wrapper could carry its own `session_id` and an `audit_opinion` of "UNQUALIFIED PASS" next to a genuine export and the run would report PASS without mentioning them. They are now refused by name. Keyring. `--public-keyring-file` is an externally pinned trust input. This tool checks its shape, status, validity window and uniqueness and binds each signed object to the one key it names, but it does NOT establish the keyring's own authenticity. Pin it out of band; `--verbose` prints each key's fingerprint. Descriptor validation mirrors the gateway loader: `valid_from` is a non-empty RFC 3339 timestamp, `valid_until` is an RFC 3339 timestamp or null, a window may not run backwards, private-key-shaped fields are rejected by name, key ids are unique, and no two descriptors may name the same public key. VALIDITY WINDOW means COMPARED, not merely shaped. `valid_from` / `valid_until` are parsed at load and ENFORCED at key resolution, following the gateway's attestation-keyring pattern: an object that predates or postdates the window of the key it names is refused, naming the key id and the bound it crossed. The comparison is PER SIGNED OBJECT, against that object's OWN declared instant, never against the instant the bundle was packaged. An event signed before a key rotation is older than the export that carries it, so measuring it by the export's date would refuse exactly the historical verification the retired key exists for. `bundle.export_generated_utc` dates the envelope and is REQUIRED under a pinned keyring; `request.timestamp_utc` dates a receipt; `payload.received_utc` (or occurred_utc / timestamp_utc) dates an event. An object that declares no instant has nothing to compare, and the run SAYS so: `--session-export` prints a `validity` line naming which objects were compared, against how many, and which declared no timestamp; a bare `--manifest` or `--chain` under a keyring says the same on its result line. Before 2026-09-02 the two fields were type-checked and then never compared to anything at all: an active key whose window read 2099-01-01..2099-01-02 verified a 2026 export, and a `valid_from` of "not-a-date" passed the shape check. One documented divergence: the gateway mints key ids BY DERIVATION (the first 16 hex of the SHA-256 of the raw public key) and so requires every descriptor's id to equal that derivation. This package's keyring uses human-readable key ids (`vke_synthetic_2026-05`), which no derivation can produce, so the derived-id binding is enforced conditionally: an id in the derived SHAPE (exactly 16 lowercase hex characters) MUST equal the derivation for its key, and an id that is not in that shape is treated as a label. The same-public-key check closes the aliasing hole the derivation rule otherwise covers. A keyring whose labels map to the wrong keys is a compromised pinned input, which this tool documents rather than claims to detect. DIVERGENCES FROM THE GATEWAY-SIDE REFERENCE, in full. This tool is a customer-facing subset of the internal VeilEngine receipt verifier. Where its behaviour differs from that reference, it is listed here and nowhere else, so "documented superset" is a claim a reader can check rather than take: 1. STRICTER -- required bundle and chain-wrapper tenant. `bundle.tenant_id` and the chain wrapper's `tenant_id` are REQUIRED here; the reference leaves both unrequired. An export asserted to hold one tenant that some layer never declared is an export whose tenant claim is weaker than it reads. See Binding above. 2. CONDITIONAL -- derived key-id binding. The reference mints every key id BY DERIVATION and requires each descriptor's id to equal it. This package's keyring uses human-readable labels, which no derivation can produce, so the rule is enforced conditionally: an id in the derived SHAPE must equal the derivation, and any other id is treated as a label. The same-public-key check closes the aliasing hole the derivation rule otherwise covers. See Keyring above. 3. ALIGNED as of 2026-09-02 -- explicit-key `signing_key_id` binding. This tool used to ignore the declared key id entirely in explicit-key mode, which made it LOOSER than the reference (veilengine_receipt_verifier.py:458-462). It now binds a derived-shape declared id to the supplied key, prints a label id on every layer line, and honours --expect-key-id. This is no longer a divergence and is listed so the record of what changed is not lost. 4. STRICTER -- an export carrying no receipts FAILS here. The reference reports SKIP ("empty receipts array; legitimate for chain-only sessions"). This package publishes a receipt-bearing export and a binding layer that reads the receipts, so a receipts array that arrives empty is more likely to be an export that lost them than a chain-only session, and it is refused rather than skipped. Do not modify the verification logic without re-publishing a documented diff: this tool's value rests on byte-for-byte alignment with the signing path. """ from __future__ import annotations import argparse import copy import hashlib import json import os import re import sys import unicodedata from datetime import datetime from pathlib import Path, PurePosixPath from typing import Any, Dict, List, Optional, Sequence, Tuple LEGACY_CANON = "ve-canonical-json-v1" JCS_CANON = "rfc8785-jcs-v1" SUPPORTED_CANON = (LEGACY_CANON, JCS_CANON) CHAIN_HASH_ALGORITHMS = { LEGACY_CANON: "sha256-ve-canonical-json-v1", JCS_CANON: "sha256-rfc8785-jcs-v1", } CHAIN_HEAD_SENTINEL = "0" * 64 # The Windows reparse tags that make one path mean another path: a mount point # (what `mklink /J` creates) and a symbolic link. Used only on interpreters # older than the one that added `Path.is_junction()`. REDIRECTING_REPARSE_TAGS = frozenset({0xA0000003, 0xA000000C}) # The self-test derives its throwaway keys from this seed instead of generating # them randomly, so a self-test run is byte-reproducible and the recorded # transcript can be hashed into the package manifest. The seed is SYNTHETIC and # NON-SECRET for exactly the reason the package's signing key is: it protects # nothing, and it is published so anyone can re-derive the same run. SELF_TEST_KEY_SEED = "ve-sample-package-self-test-v1" KEYRING_SCHEMA_VERSION = "1.0.0" MAX_KEYRING_BYTES = 1024 * 1024 MAX_KEYRING_KEYS = 128 UNSUPPORTED = "canonicalization_version {0!r} not supported by this verifier; expected one of {1}" KEYRING_REQUIRED_FIELDS = {"alg", "key_id", "public_key_hex", "status", "valid_from", "valid_until"} KEYRING_STATUSES = {"active", "retired", "revoked"} # A public keyring must never carry secret material. These names are rejected # by name so the diagnostic says WHY, ahead of the generic unexpected-field # check that would otherwise swallow them. PRIVATE_KEY_SHAPED_FIELDS = { "private_key", "private_key_hex", "private_key_bytes", "privatekey", "secret_key", "secret_key_hex", "secret", "seed", "seed_hex", "d", "sk", } # The gateway's key-id derivation shape: sha256(raw public key).hexdigest()[:16]. DERIVED_KEY_ID_SHAPE = re.compile(r"^[0-9a-f]{16}$") # A chain anchor is exactly these three fields; see compute_chain_anchor. CHAIN_ANCHOR_FIELDS = {"algorithm", "event_count", "last_event_hash"} # Where a receipt names the session and tenant it belongs to. The receipt # schema shipped in this package nests request identity under `request`; the # gateway schema carries it at the top level. EVERY path present is read and # they must agree -- see declared_identity below. Order carries no precedence. RECEIPT_SESSION_PATHS = (("request", "session_id"), ("session_id",)) RECEIPT_TENANT_PATHS = (("request", "tenant_id"), ("tenant_id",)) # Where a signed object names the key that signed it. Held to the same # contract as session and tenant identity: EVERY path present is read and they # must agree. Order carries no precedence. KEY_ID_PATHS = (("signing_key_id",), ("signer_key_id",), ("verification", "signer_key_id")) # Where a signed object declares the instant it was produced. A key's validity # window is compared against the object's OWN instant, never against the # instant the bundle was packaged: an event signed before a key rotation is # older than the export that carries it, and comparing it to the export's date # would refuse exactly the historical verification the retired key exists for. EXPORT_TIME_PATHS = (("export_generated_utc",),) RECEIPT_TIME_PATHS = (("request", "timestamp_utc"), ("timestamp_utc",)) EVENT_TIME_PATHS = ( ("payload", "received_utc"), ("payload", "occurred_utc"), ("payload", "timestamp_utc"), ("timestamp_utc",), ) class CanonicalizationError(ValueError): """Raised when an artifact names no exact supported canonicalization.""" def resolve_canonicalization_version(payload: Dict[str, Any]) -> str: """Absent field means the frozen legacy format; every other value is exact.""" if "canonicalization_version" not in payload: return LEGACY_CANON version = payload["canonicalization_version"] if not isinstance(version, str) or version not in SUPPORTED_CANON: raise CanonicalizationError(UNSUPPORTED.format(version, list(SUPPORTED_CANON))) return version def canonical_json(payload: Any, canonicalization_version: str) -> bytes: if canonicalization_version == LEGACY_CANON: return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") if canonicalization_version != JCS_CANON: raise CanonicalizationError(UNSUPPORTED.format(canonicalization_version, list(SUPPORTED_CANON))) try: import rfc8785 except ImportError as exc: raise CanonicalizationError( f"canonicalization_version {JCS_CANON!r} requires the `rfc8785` package. " f"Install: pip install rfc8785" ) from exc try: return rfc8785.dumps(payload) except Exception as exc: raise CanonicalizationError(f"payload is not representable under {JCS_CANON}: {exc}") from exc def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def is_link_or_reparse_point(candidate: Path) -> bool: """True when this path redirects to somewhere else on the filesystem. A SYMBOLIC LINK is not the only way a directory name stops meaning the directory it sits in. On Windows a JUNCTION is a different reparse tag (IO_REPARSE_TAG_MOUNT_POINT) and `Path.is_symlink()` answers False for one, so a walk that asks only that question steps straight through a junction and hashes bytes from wherever it points -- outside the package root included -- while every string-level check still holds. `Path.is_junction()` answers the question directly and arrived in Python 3.12; this file supports 3.9+, so it is used when present. On an older interpreter the reparse tag is read off the lstat result instead (Windows, 3.8+) and compared against the two tags that redirect a path: a mount point and a symbolic link. Tags that do not redirect -- cloud placeholders, deduplication, WIM-backed files -- are not this function's subject and a file carrying one is a perfectly ordinary file to read. Any error reaching the path is answered False: this function decides REDIRECTION, and a path that cannot be stat-ed is handled by the caller's existence and regular-file checks, which produce a better diagnostic than "it might have been a link". """ try: if candidate.is_symlink(): return True except OSError: return False probe = getattr(candidate, "is_junction", None) if probe is not None: try: return bool(probe()) except OSError: return False try: tag = getattr(candidate.lstat(), "st_reparse_tag", 0) except OSError: return False return tag in REDIRECTING_REPARSE_TAGS # --- timestamps ----------------------------------------------------------- # Mirrors the gateway's attestation-keyring pattern # (veilengine/gateway/provider_attestation.py): a validity bound is parsed as # an RFC 3339 instant at LOAD time and compared against the artifact's own # timestamp at RESOLVE time. This verifier used to type-check valid_from and # valid_until and then never compare them to anything, so an active key whose # window read 2099-01-01..2099-01-02 verified a 2026 export, and a valid_from # of "not-a-date" passed the shape check. RFC3339_RE = re.compile( r"^\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}:\d{2}(\.\d+)?([Zz]|[+-]\d{2}:\d{2})$" ) def parse_rfc3339(value: Any, field: str) -> datetime: """Parse one RFC 3339 instant, or raise ValueError naming the field.""" if not isinstance(value, str) or not RFC3339_RE.match(value): raise ValueError(f"{field} must be an RFC 3339 timestamp; got {value!r}") text = value.replace("Z", "+00:00").replace("z", "+00:00").replace(" ", "T") # Python 3.9's fromisoformat accepts only 3 or 6 fractional digits; pad or # trim so a conformant timestamp is not rejected by the standard library. match = re.match(r"^(.*?)\.(\d+)([+-]\d{2}:\d{2})$", text) if match: head, fraction, offset = match.groups() text = f"{head}.{fraction[:6].ljust(6, '0')}{offset}" try: parsed = datetime.fromisoformat(text) except ValueError as exc: raise ValueError(f"{field} must be an RFC 3339 timestamp; got {value!r}") from exc if parsed.tzinfo is None: # pragma: no cover - the regex requires an offset raise ValueError(f"{field} must include a timezone; got {value!r}") return parsed def artifact_timestamp(obj: Any, paths: Sequence[Sequence[str]], label: str) -> Tuple[ Optional[datetime], Optional[str] ]: """The instant an artifact declares it was produced. Returns (when, error).""" for path in paths: value = dig(obj, path) if value is ABSENT: continue try: return parse_rfc3339(value, f"{label}.{'.'.join(path)}"), None except ValueError as exc: return None, str(exc) return None, None # --- input handling ------------------------------------------------------- class InputError(Exception): """An argument or environment fault: the run never reached verification. Exit code 2, never 1. Telling a buyer who typo'd a filename that verification FAILED is a false negative about their evidence, and any script gating on exit codes -- this package's own release gate included -- cannot tell a rejection from a crash if both come back as 1. Ten malformed/missing-input classes used to raise a raw Python traceback and exit 1, against a documented contract of "0 PASS | 1 FAIL | 2 argument or environment error" and against the gateway-side reference, which handles all of them with [ENV] and exit 2. """ class DependencyError(Exception): """A dependency this tool needs could not be loaded. Exit code 2, like every other environment fault. A broken cryptography backend -- an installed package whose `_cffi_backend` will not import -- used to surface as a traceback and exit 1, which reads as "your evidence failed to verify" when what actually happened is that the verifier never ran. """ def _reject_duplicate_keys(pairs: List[Tuple[str, Any]]) -> Dict[str, Any]: """json object_pairs_hook: refuse duplicate keys, naming the key. Python's parser is last-wins. A first-wins parser, and every human reading the file, sees a different object than the one whose bytes were signed, so a receipt whose text opens with `"session_id": "sess_first_declared"` verified clean while the signature covered the second, honest value. An object that names one key twice is not a document with a value; it is two documents, and this verifier will not choose between them. """ seen: Dict[str, Any] = {} for key, value in pairs: if key in seen: raise ValueError(f"duplicate JSON key {key!r}") seen[key] = value return seen def load_json_document(path_value: str, flag: str) -> Any: """Read and parse one JSON input, or raise InputError. Never a traceback.""" path = Path(path_value) try: text = path.read_text(encoding="utf-8") except FileNotFoundError: raise InputError(f"{flag} file not found: {path_value}") from None except IsADirectoryError: raise InputError(f"{flag} is a directory, not a file: {path_value}") from None except UnicodeDecodeError as exc: raise InputError(f"{flag} is not valid UTF-8: {exc}") from None except OSError as exc: raise InputError(f"{flag} could not be read: {exc}") from None try: return json.loads(text, object_pairs_hook=_reject_duplicate_keys) except ValueError as exc: raise InputError(f"{flag} JSON parse failed: {exc}") from None # --- the manifest-path grammar --------------------------------------------- # # ONE PORTABLE GRAMMAR, ENFORCED IN BOTH THE BUILDER AND THE VERIFIER, so a # package that BUILDS is a package that VERIFIES. A manifest path is: # # * RELATIVE and '/'-separated. No drive letter, no leading separator, no # backslash -- '\' is a legal filename character on POSIX, so accepting it # as a separator means one string names two different files depending on # the host reading it. # * NFC-NORMALIZED. Two Unicode spellings of one name are one file on macOS # and two strings everywhere. # * Every segment matches SEGMENT_GRAMMAR: ASCII letters and digits, '.', # '_' and '-', starting with a letter or digit. No '.' or '..' segments. # * No segment ending in '.' or ' '. Win32 silently strips both, so # 'README.md.' and 'README.md ' open 'README.md'. This is the alias class # that puts two descriptors over one physical file and reports "11 # artifact hashes match" over ten. # * No ':' anywhere. On NTFS 'README.md:hidden' names an alternate data # stream of README.md, not a file called 'README.md:hidden'. # * No Windows reserved device name -- CON, PRN, AUX, NUL, COM1-COM9, # LPT1-LPT9 -- in any case, with or without an extension. 'NUL.txt' opens # the null device, which reads as an empty file rather than as a missing # one. # # String-level uniqueness is not filesystem-object uniqueness, so the grammar # is only half the rule; verify_manifest_file also requires unique RESOLVED # FILE IDENTITY across descriptor targets. The grammar closes the aliases that # are decidable from the string alone; the identity check closes the rest # (hard links, and anything this host aliases that the grammar does not name). SEGMENT_GRAMMAR = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") RESERVED_DEVICE_NAMES = frozenset( ["CON", "PRN", "AUX", "NUL"] + [f"COM{digit}" for digit in range(1, 10)] + [f"LPT{digit}" for digit in range(1, 10)] ) def manifest_path_problem(path_value: str, where: str) -> Optional[str]: """Why this descriptor path is outside the manifest-path grammar, or None. `where` is the caller's label for the descriptor, so every rejection names which descriptor broke which rule. """ if re.match(r"^[A-Za-z]:", path_value) or path_value.startswith(("/", "\\")): return f"{where}.path {path_value!r} is absolute; artifacts must live inside the package" # Traversal is decided on a separator-tolerant reading, BEFORE the # backslash rule refuses the string: '..\\outside.txt' is a traversal # attempt on the host that matters, and naming it as a punctuation problem # would report the smaller of the two truths. if ".." in path_value.replace("\\", "/").split("/"): return ( f"{where}.path {path_value!r} resolves outside the package directory; a signed " f"descriptor is not a licence to read outside it" ) if "\\" in path_value: return ( f"{where}.path {path_value!r} contains a backslash; manifest paths are relative " f"and '/'-separated, and '\\' is an ordinary filename character on POSIX, so one " f"such string names two different files depending on the host reading it" ) if unicodedata.normalize("NFC", path_value) != path_value: return ( f"{where}.path {path_value!r} is not NFC-normalized; two Unicode spellings of one " f"name are one file on some hosts and two strings on others" ) segments = path_value.split("/") if path_value.rstrip("/") in ("", "."): return ( f"{where}.path {path_value!r} names the package directory itself, not a file " f"inside it; a directory has no bytes to hash" ) if segments[-1] == "": return ( f"{where}.path {path_value!r} ends in a separator, so it names a directory " f"rather than a file" ) if "." in segments or "" in segments: return ( f"{where}.path {path_value!r} is not in canonical form; a descriptor must name " f"its target one way, or two descriptors can address one file and still look " f"like two" ) for segment in segments: if segment.endswith("."): return ( f"{where}.path {path_value!r} has a segment ending in '.'; Win32 strips it, " f"so this string and the same name without the dot open ONE file while " f"reading as two descriptors" ) if segment.endswith(" "): return ( f"{where}.path {path_value!r} has a segment ending in a space; Win32 strips " f"it, so this string and the same name without the space open ONE file while " f"reading as two descriptors" ) if ":" in segment: return ( f"{where}.path {path_value!r} contains ':'; on NTFS that is alternate-stream " f"syntax naming a stream of another file, not a file of its own" ) if segment.split(".")[0].upper() in RESERVED_DEVICE_NAMES: return ( f"{where}.path {path_value!r} names the reserved Windows device " f"{segment.split('.')[0].upper()}; opening it reads a device rather than a " f"package artifact" ) if not SEGMENT_GRAMMAR.match(segment): return ( f"{where}.path {path_value!r} has a segment outside the manifest-path " f"grammar; each segment must match {SEGMENT_GRAMMAR.pattern} -- ASCII " f"letters, digits, '.', '_' and '-', starting with a letter or digit" ) return None def manifest_root_problem(obj: Any) -> Optional[str]: """Why this object is not a package manifest, or None. Artifact descriptors used to be read straight out of a signed manifest with no shape check at all, so `path` could be missing (KeyError traceback), `artifacts` could be an object (TypeError traceback), the list could be EMPTY and still report "0 artifact hashes match", and a path could resolve outside the package directory: `"../_atk_outside.txt"` was read, hashed and reported as a passing artifact. A signature over a descriptor does not make the descriptor safe to follow. """ if not isinstance(obj, dict): return f"a manifest must be a JSON object; got {type(obj).__name__}" artifacts = obj.get("artifacts", ABSENT) if artifacts is ABSENT: return "a manifest must carry an 'artifacts' list" if not isinstance(artifacts, list): return f"manifest 'artifacts' must be a list; got {type(artifacts).__name__}" if not artifacts: return ( "manifest 'artifacts' is empty; a manifest that commits to nothing cannot be " "reported as covering the package" ) # MEMBERSHIP IS EXACT. Descriptors used to be checked # one at a time, so a manifest with ten descriptors and one duplicated path # passed and printed "10 artifact hashes match" while naming nine files -- # more coverage than it had -- and a descriptor whose path was "." passed # the shape check and then raised an uncaught filesystem exception when the # verifier tried to read a directory as bytes. Uniqueness and file-shape # are decided HERE, before anything is opened. declared_paths: Dict[str, int] = {} folded_paths: Dict[str, int] = {} for index, artifact in enumerate(artifacts): where = f"manifest artifacts[{index}]" if not isinstance(artifact, dict): return f"{where} must be an object; got {type(artifact).__name__}" missing = sorted({"path", "sha256", "bytes"} - set(artifact)) if missing: return f"{where} is missing {missing}" path_value = artifact["path"] if not isinstance(path_value, str) or not path_value: return ( f"{where}.path must be a non-empty string; got " f"{type(path_value).__name__} {path_value!r}" ) problem = manifest_path_problem(path_value, where) if problem: return problem canonical = path_value if canonical in declared_paths: return ( f"{where}.path {path_value!r} is already declared at " f"artifacts[{declared_paths[canonical]}]; a manifest that names one file twice " f"reports more coverage than it holds" ) folded = canonical.casefold() if folded in folded_paths: return ( f"{where}.path {path_value!r} differs from " f"artifacts[{folded_paths[folded]}] only by case; on a case-insensitive " f"filesystem both descriptors address one file" ) declared_paths[canonical] = index folded_paths[folded] = index digest = artifact["sha256"] if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", digest): return f"{where}.sha256 must be 64 hex characters; got {digest!r}" size = artifact["bytes"] if not isinstance(size, int) or isinstance(size, bool) or size < 0: return f"{where}.bytes must be a non-negative integer; got {size!r}" # SIGNATURE TYPE, AT THE ROOT. Same rule, two # verdicts: as the INPUT this is a shape fault (exit 2), as an artifact # named by an export the manifest layer reports it as a FAIL. if "manifest_signature" in obj and obj["manifest_signature"]: problem = signature_problem(obj["manifest_signature"], "manifest_signature") if problem: return problem return None def receipt_root_problem(obj: Any) -> Optional[str]: """Why this object is not a receipt, or None. No mode validated the shape of its input, so `--receipt session_export.json` printed "Receipt: VERIFIED" and "Overall: PASS" over a whole session export whose chain, binding, manifest and anchor were never looked at. An assurance that names a layer must name a layer that ran. """ if not isinstance(obj, dict): return f"a receipt must be a JSON object; got {type(obj).__name__}" receipt_id = obj.get("receipt_id", ABSENT) if receipt_id is ABSENT: return ( "a receipt must declare 'receipt_id'; this object does not, so it is not a " "receipt and no receipt-layer verdict can be reported for it" ) if not isinstance(receipt_id, str) or not receipt_id: return ( f"receipt 'receipt_id' must be a non-empty string; got " f"{type(receipt_id).__name__} {receipt_id!r}" ) # SIGNATURE TYPE, AT THE ROOT. See # session_export_root_problem: as the INPUT this is a shape fault (exit 2); # as an embedded artifact the receipt layer reports it as a FAIL. verification = obj.get("verification") if isinstance(verification, dict) and "signature" in verification: problem = signature_problem(verification["signature"], "receipt verification.signature") if problem: return problem elif "signature" in obj: problem = signature_problem(obj["signature"], "receipt signature") if problem: return problem return None def unwrap_session_export(raw: Any) -> Tuple[Any, Optional[str]]: """Return (bundle, unsigned wrapper sibling names) for a session export. Operators who curl the endpoint save the {"veilengine_session_export": {...}} wrapper; operators who extract the bundle save the inner object. Both are accepted. """ if isinstance(raw, dict) and "veilengine_session_export" in raw: siblings = sorted(k for k in raw if k != "veilengine_session_export") return raw["veilengine_session_export"], (", ".join(siblings) if siblings else None) return raw, None def session_export_root_problem(raw: Any) -> Optional[str]: """Why this object is not a session export, or None.""" if not isinstance(raw, dict): return f"a session export must be a JSON object; got {type(raw).__name__}" bundle, _ = unwrap_session_export(raw) if not isinstance(bundle, dict): return ( f"the veilengine_session_export wrapper must hold an object; got " f"{type(bundle).__name__}" ) if "chain" not in bundle: return ( "a session export must carry a 'chain'; this object does not, so it is not a " "session export and no export-layer verdict can be reported for it" ) # SIGNATURE TYPE, AT THE ROOT. Reached as the INPUT, # a wrong-typed signature is a shape fault: the run never got to a verdict, # so it exits 2. Reached as an artifact NAMED BY something else, the per # layer checks report it as that layer's FAIL. Same rule, two verdicts -- # the pattern this file already uses for root shape. if "signature" in bundle: problem = signature_problem(bundle["signature"], "session export bundle.signature") if problem: return problem chain = bundle.get("chain") if isinstance(chain, dict) and isinstance(chain.get("events"), list): for index, event in enumerate(chain["events"]): if isinstance(event, dict) and "event_signature" in event: problem = signature_problem(event["event_signature"], f"event[{index}] event_signature") if problem: return problem receipts = bundle.get("receipts") if isinstance(receipts, list): for index, receipt in enumerate(receipts): if not isinstance(receipt, dict): continue verification = receipt.get("verification") if isinstance(verification, dict) and "signature" in verification: problem = signature_problem(verification["signature"], f"receipt[{index}] verification.signature") if problem: return problem elif "signature" in receipt: problem = signature_problem(receipt["signature"], f"receipt[{index}] signature") if problem: return problem return None # --- key material --------------------------------------------------------- def key_fingerprint(public_key_hex: str) -> str: """The gateway's derived key id: first 16 hex of SHA-256 over the raw key.""" raw = bytes.fromhex(public_key_hex) if len(raw) != 32: raise ValueError(f"public key must decode to 32 bytes; got {len(raw)}") if raw == b"\x00" * 32: raise ValueError("public key must not be the all-zero key") return sha256_hex(raw)[:16] def parse_public_keyring(obj: Any) -> Dict[str, Dict[str, Any]]: """Validate a parsed keyring object; return descriptors by key_id. Trust metadata is validated BEFORE any key is resolved, so a keyring that cannot be believed fails closed rather than being consulted. Mirrors the gateway loader's descriptor validation; the one documented divergence is the conditional derived-id binding described in the module docstring. """ if not isinstance(obj, dict) or sorted(obj) != ["keys", "schema_version"]: raise ValueError("--public-keyring-file must be an object of exactly {schema_version, keys}") if obj["schema_version"] != KEYRING_SCHEMA_VERSION: raise ValueError(f"--public-keyring-file schema_version must be {KEYRING_SCHEMA_VERSION!r}") keys = obj["keys"] if not isinstance(keys, list) or not keys: raise ValueError("--public-keyring-file 'keys' must be a non-empty array") if len(keys) > MAX_KEYRING_KEYS: raise ValueError(f"--public-keyring-file has {len(keys)} keys; maximum is {MAX_KEYRING_KEYS}") by_id: Dict[str, Dict[str, Any]] = {} by_public_key: Dict[str, int] = {} for i, entry in enumerate(keys): if not isinstance(entry, dict): raise ValueError(f"--public-keyring-file keys[{i}] must be an object") leaked = sorted(name for name in entry if name.lower() in PRIVATE_KEY_SHAPED_FIELDS) if leaked: raise ValueError( f"--public-keyring-file keys[{i}] carries private-key-shaped field(s) {leaked}; " f"a public keyring must never contain secret material" ) unexpected = sorted(set(entry) - KEYRING_REQUIRED_FIELDS) missing = sorted(KEYRING_REQUIRED_FIELDS - set(entry)) if unexpected: raise ValueError(f"--public-keyring-file keys[{i}] unexpected fields: {unexpected}") if missing: raise ValueError(f"--public-keyring-file keys[{i}] missing fields: {missing}") if entry["alg"] != "ed25519": raise ValueError(f"--public-keyring-file keys[{i}].alg must be 'ed25519'") if entry["status"] not in KEYRING_STATUSES: raise ValueError(f"--public-keyring-file keys[{i}].status must be active, retired, or revoked") valid_from = entry["valid_from"] if not isinstance(valid_from, str) or not valid_from: raise ValueError(f"--public-keyring-file keys[{i}].valid_from must be a non-empty string") valid_until = entry["valid_until"] if valid_until is not None and (not isinstance(valid_until, str) or not valid_until): raise ValueError(f"--public-keyring-file keys[{i}].valid_until must be a string or null") # The window is PARSED here and COMPARED in resolve_key. Shape-checking # it and stopping there is what let "not-a-date" through and left a # 2099 window binding a 2026 export. try: starts = parse_rfc3339(valid_from, f"--public-keyring-file keys[{i}].valid_from") ends = ( parse_rfc3339(valid_until, f"--public-keyring-file keys[{i}].valid_until") if valid_until is not None else None ) except ValueError as exc: raise ValueError(str(exc)) from exc if ends is not None and ends < starts: raise ValueError( f"--public-keyring-file keys[{i}] validity window is reversed: " f"valid_until {valid_until} precedes valid_from {valid_from}" ) key_id = entry["key_id"] if not isinstance(key_id, str) or not key_id: raise ValueError(f"--public-keyring-file keys[{i}].key_id must be a non-empty string") public_key_hex = entry["public_key_hex"] if not isinstance(public_key_hex, str): raise ValueError(f"--public-keyring-file keys[{i}].public_key_hex must be a string") public_key_hex = public_key_hex.strip().lower() try: derived = key_fingerprint(public_key_hex) except ValueError as exc: raise ValueError(f"--public-keyring-file keys[{i}].public_key_hex: {exc}") from exc if DERIVED_KEY_ID_SHAPE.match(key_id) and key_id != derived: raise ValueError( f"--public-keyring-file keys[{i}].key_id mismatch: " f"declared {key_id!r}, derived {derived!r}" ) if key_id in by_id: raise ValueError(f"--public-keyring-file duplicate key_id {key_id!r}") if derived in by_public_key: raise ValueError( f"--public-keyring-file keys[{i}] and keys[{by_public_key[derived]}] name the same " f"public key (fingerprint {derived}) under different key ids" ) by_public_key[derived] = i entry = dict(entry) entry["public_key_hex"] = public_key_hex entry["fingerprint"] = derived entry["valid_from_at"] = starts entry["valid_until_at"] = ends by_id[key_id] = entry return by_id def load_public_keyring_file(path_value: str) -> Dict[str, Dict[str, Any]]: """Load a bounded, public-only keyring; return descriptors by key_id.""" path = Path(path_value) if not path.exists(): raise InputError(f"--public-keyring-file file not found: {path_value}") if path.stat().st_size > MAX_KEYRING_BYTES: raise ValueError(f"--public-keyring-file exceeds {MAX_KEYRING_BYTES} bytes") return parse_public_keyring(load_json_document(path_value, "--public-keyring-file")) def load_key_material(args: argparse.Namespace) -> Tuple[Optional[str], Optional[Dict[str, Dict[str, Any]]]]: """Resolve exactly one single-key source or one pinned keyring file.""" provided = [ (flag, value) for flag, value in ( ("--public-key-hex", args.public_key_hex), ("--public-key-file", args.public_key_file), ("--public-keyring-file", args.public_keyring_file), ) if value ] if len(provided) != 1: raise ValueError("exactly one of --public-key-hex, --public-key-file, --public-keyring-file required") flag, value = provided[0] if flag == "--public-keyring-file": return None, load_public_keyring_file(value) if flag == "--public-key-hex": return value.strip().lower(), None key_file = load_json_document(value, "--public-key-file") if not isinstance(key_file, dict): raise InputError( f"--public-key-file must be a JSON object carrying 'public_key_hex'; got " f"{type(key_file).__name__}" ) hex_str = key_file.get("public_key_hex") if not hex_str: raise ValueError("--public-key-file missing 'public_key_hex'") # str() coercion turned a truthy non-string into "['ab']" and reported it # as a hex-decode failure of the KEY MATERIAL, which is a verification # verdict about the wrong thing. The keyring path already typed this field; # the single-key path did not. if isinstance(hex_str, bool) or not isinstance(hex_str, str): raise ValueError( f"--public-key-file 'public_key_hex' must be a hex string; got " f"{type(hex_str).__name__} {hex_str!r}" ) return hex_str.strip().lower(), None def declared_key_id( obj: Any, label: str = "object" ) -> Tuple[Optional[str], Optional[str]]: """The key id a signed object names. Returns (key_id, error). EVERY path in KEY_ID_PATHS that is present is read and they must AGREE. Selecting the first non-empty path -- which is what this did before 2026-09-02 -- is the same defect class the identity layer closed for session and tenant, left in place for key identity: a receipt could carry `signing_key_id` naming the active key and `verification.signer_key_id` naming a REVOKED one, be verified and reported under the first, and print nothing about the second. Key ids are typed the same way identities are: a present path holding anything but a non-empty string is refused by name. """ found: List[Tuple[str, str]] = [] for path in KEY_ID_PATHS: value = dig(obj, path) if value is ABSENT: continue rendered_path = ".".join(path) if not isinstance(value, str) or not value: return None, ( f"{label} declares a key id at {rendered_path} as " f"{type(value).__name__} {value!r}; a key id must be a non-empty string" ) found.append((rendered_path, value)) if not found: return None, None if len({value for _, value in found}) > 1: rendered = ", ".join(f"{name}={value!r}" for name, value in found) return None, ( f"{label} declares conflicting key ids ({rendered}); an object that names " f"two signing keys cannot be bound to one" ) return found[0][1], None def resolve_key( obj: Dict[str, Any], public_key_hex: Optional[str], keyring: Optional[Dict[str, Dict[str, Any]]], label: str, expect_key_id: Optional[str] = None, at: Optional[datetime] = None, ) -> Tuple[Optional[str], Optional[str], Optional[str]]: """Bind one signed object to exactly one public key. Returns (public_key_hex, declared_key_id, error). `at` is the instant the artifact declares it was produced. When it is known, the pinned keyring's validity window for the resolved key is ENFORCED against it. When it is not known -- a bare `--manifest` or `--chain` carries no timestamp -- the window is not compared, and the caller SAYS so on the layer line rather than leaving the reader to assume it was. Explicit-key mode used to ignore `declared_key_id` entirely, which made this tool strictly LOOSER than the gateway-side reference on the mode the README leads with: an export whose every layer named a key id that was not the supplied key -- a retired one, a nonexistent one -- verified clean. It is now held to the reference's contract, with one conditional the reference does not need: this package's key ids are human-readable LABELS, which no derivation can produce, so a declared id in the DERIVED SHAPE must equal the supplied key's fingerprint, and a label id is printed on every layer line and can be pinned with --expect-key-id. """ key_id, error = declared_key_id(obj, label) if error: return None, None, error if expect_key_id is not None: if key_id is None: return None, None, ( f"{label}: --expect-key-id {expect_key_id!r} was supplied but this object " f"declares no key id" ) if key_id != expect_key_id: return None, key_id, ( f"{label}: declares key id {key_id!r}, which is not the pinned " f"--expect-key-id {expect_key_id!r}" ) if keyring is None: if public_key_hex is None: return None, key_id, f"{label}: no public-key source" if key_id is not None and DERIVED_KEY_ID_SHAPE.match(key_id): try: supplied = key_fingerprint(public_key_hex) except ValueError as exc: return None, key_id, f"{label}: supplied public key is invalid: {exc}" if key_id != supplied: return None, key_id, ( f"{label}: declares key id {key_id!r}, which does not match the supplied " f"public key's id {supplied!r}" ) return public_key_hex, key_id, None if key_id is None: return None, None, f"{label}: names no key id; keyring verification cannot select a key" descriptor = keyring.get(key_id) if descriptor is None: return None, key_id, f"{label}: key id {key_id!r} is not in the pinned keyring" if descriptor["status"] == "revoked": return None, key_id, f"{label}: key id {key_id!r} is revoked in the pinned keyring" if at is not None: if at < descriptor["valid_from_at"]: return None, key_id, ( f"{label}: dated {at.isoformat()}, which PREDATES the pinned keyring's validity " f"window for key id {key_id!r} (valid_from {descriptor['valid_from']})" ) if descriptor["valid_until_at"] is not None and at > descriptor["valid_until_at"]: return None, key_id, ( f"{label}: dated {at.isoformat()}, which POSTDATES the pinned keyring's validity " f"window for key id {key_id!r} (valid_until {descriptor['valid_until']})" ) return descriptor["public_key_hex"], key_id, None def validity_note(at: Optional[datetime], keyring_mode: bool) -> str: """Say when a pinned keyring's validity windows were NOT compared. A bare --manifest or --chain carries no timestamp of its own, so there is nothing to compare the window against. That is not a reason to leave the reader assuming it was compared: the layer line says so, and points at the mode that does enforce it. """ if not keyring_mode or at is not None: return "" return ( "; keyring validity windows NOT compared (this input declares no timestamp; " "verify it inside its session export, which does)" ) def key_id_note(ids: Sequence[Optional[str]], keyring_mode: bool) -> str: """The declared-key-id suffix a layer line carries in explicit-key mode. A LABEL key id cannot be bound to a supplied raw public key by any computation -- that is what the pinned keyring is for -- so instead of silently ignoring it the way this tool did before 2026-09-02, explicit-key mode PRINTS every label the layer declared. A reader who knows which key they handed the tool sees the mismatch on the line that claims the layer passed. --expect-key-id turns that reading into an assertion. """ if keyring_mode: return "" labels = sorted({i for i in ids if i and not DERIVED_KEY_ID_SHAPE.match(i)}) if not labels: return "" return f"; declares key id {', '.join(labels)}" def signature_problem(value: Any, where: str) -> Optional[str]: """Why this value cannot be a signature, or None. Every signature-bearing layer used to test its field for TRUTH and then hand it straight to `.strip()`. A truthy non-string -- `123`, `["sig"]`, `{"hex": "..."}`, `True` -- passed the truth test and raised AttributeError out of the verifier, against a documented contract that input faults leave by the exit-2 door without a traceback. The type and the encoding are decided here, once, and every layer names itself. """ if isinstance(value, bool) or not isinstance(value, str): return ( f"{where} must be a hex string, optionally prefixed 'ed25519:'; got " f"{type(value).__name__} {value!r}" ) if not value.strip(): return f"{where} is empty" return None def ed25519_verify(public_key_hex: str, canonical: bytes, signature_value: str) -> Tuple[bool, str]: try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey except ImportError as exc: raise DependencyError( "the `cryptography` package is required. Install: pip install cryptography" ) from exc # Defensive: every caller validates its own field and names its own layer, # but a signature that reached here untyped would raise rather than judge. problem = signature_problem(signature_value, "signature") if problem: return False, problem sig_hex = signature_value.strip() if sig_hex.startswith("ed25519:"): sig_hex = sig_hex[len("ed25519:"):] if not sig_hex or sig_hex.startswith("PLACEHOLDER"): return False, "signature is empty or PLACEHOLDER; run `sign_and_verify.py --sign` first" try: sig_bytes, key_bytes = bytes.fromhex(sig_hex), bytes.fromhex(public_key_hex) except ValueError as exc: return False, f"hex decode failed: {exc}" if len(sig_bytes) != 64: return False, f"signature is {len(sig_bytes)} bytes; Ed25519 expects 64" if len(key_bytes) != 32: return False, f"public key is {len(key_bytes)} bytes; Ed25519 expects 32" try: Ed25519PublicKey.from_public_bytes(key_bytes).verify(sig_bytes, canonical) except InvalidSignature: return False, "signature verification failed: content was modified OR signed by a different key" except Exception as exc: return False, f"signature verification raised: {exc}" return True, "signature verified" def verify_signed_object(body: Dict[str, Any], signature_value: str, public_key_hex: str) -> Tuple[bool, str]: """Canonicalize under the object's own declared version, then verify.""" try: version = resolve_canonicalization_version(body) canonical = canonical_json(body, version) except CanonicalizationError as exc: return False, f"canonicalization rejected: {exc}" ok, message = ed25519_verify(public_key_hex, canonical, signature_value) return ok, f"{message} [{version}]" # --- receipts ------------------------------------------------------------- def split_receipt_signature(receipt: Dict[str, Any]) -> Tuple[Dict[str, Any], str]: """Return (body-as-signed, signature). Supports both receipt schemas.""" verification = receipt.get("verification") if isinstance(verification, dict) and "signature" in verification: # str() coercion used to hide the type: a list became "['abc']" and # failed as a hex-decode error, which names the wrong defect. problem = signature_problem(verification["signature"], "receipt verification.signature") if problem: raise ValueError(problem) body = copy.deepcopy(receipt) body["verification"]["signature"] = "" return body, verification["signature"] if "signature" in receipt: problem = signature_problem(receipt["signature"], "receipt signature") if problem: raise ValueError(problem) return {k: v for k, v in receipt.items() if k != "signature"}, receipt["signature"] raise ValueError("receipt carries no signature field") def verify_receipt_obj( receipt: Dict[str, Any], public_key_hex: Optional[str], keyring: Optional[Dict[str, Dict[str, Any]]], label: str = "receipt", expect_key_id: Optional[str] = None, at: Optional[datetime] = None, ) -> Tuple[bool, str, Optional[str]]: """Verify one receipt. Returns (ok, message, the key id it declared).""" try: body, signature = split_receipt_signature(receipt) except ValueError as exc: return False, f"{label}: {exc}", None resolved, key_id, error = resolve_key( receipt, public_key_hex, keyring, label, expect_key_id, at ) if error or resolved is None: return False, error or f"{label}: key resolution failed", key_id ok, message = verify_signed_object(body, signature, resolved) return ok, f"{label} {receipt.get('receipt_id', '')}: {message}", key_id # --- manifest ------------------------------------------------------------- def verify_manifest_file( manifest_path: Path, public_key_hex: Optional[str], keyring: Optional[Dict[str, Dict[str, Any]]], verbose: bool, expect_key_id: Optional[str] = None, at: Optional[datetime] = None, ) -> Tuple[bool, str]: manifest = load_json_document(str(manifest_path), "manifest") # The shape rule is one rule with two verdicts. Reached as the INPUT # (--manifest), a malformed manifest is an environment fault: exit 2, no # verification claimed. Reached as an artifact NAMED BY a session export, # the export is the input under test and its manifest layer FAILs. problem = manifest_root_problem(manifest) if problem: return False, problem signature = manifest.get("manifest_signature", "") if not signature: return False, "manifest_signature is missing" problem = signature_problem(signature, "manifest_signature") if problem: return False, problem resolved, key_id, error = resolve_key( manifest, public_key_hex, keyring, "manifest", expect_key_id, at ) if error or resolved is None: return False, error or "manifest: key resolution failed" note = key_id_note([key_id], keyring is not None) # The signer BLANKS this field rather than removing it; canonicalize the # same shape it signed, or the bytes differ and every manifest fails. body = dict(manifest, manifest_signature="") ok, message = verify_signed_object(body, signature, resolved) if not ok: return False, f"manifest signature: {message}" problems: List[str] = [] artifacts = manifest.get("artifacts", []) # THE PACKAGE ROOT is the directory holding manifest.json, resolved, and it # is the same boundary in every mode: --manifest resolves its own argument, # --session-export/keyring reaches this function through the manifest the # bundle names beside the export. Resolving here rather than trusting the # caller means membership is defined against one boundary and not against # whatever working directory the run happened to start in. package_dir = manifest_path.resolve().parent # EXACT MEMBERSHIP. The published README says the # manifest covers every artifact in the package except itself. That was # true of the shipped fixture and enforced nowhere, so a package could # carry a file no descriptor named and still report a clean manifest. The # declared set and the present set are now compared by name, in both # directions, and every descriptor's target must be a REGULAR FILE reached # without following a symlink -- decided before any bytes are read, because # `open()` on a directory raises rather than judges. # # UNIQUE RESOLVED FILE IDENTITY, NOT ONLY UNIQUE STRINGS. # manifest_root_problem has already refused every alias decidable # from the string -- trailing dots and spaces, ':' streams, reserved device # names, non-NFC spellings, case-only twins. That is not the whole class: a # HARD LINK gives one file two grammar-legal names, and any aliasing this # host does that the grammar does not name has the same shape. So after the # no-symlink walk each descriptor's target is identified by (st_dev, # st_ino) and two descriptors resolving to ONE object are refused by name. # A host that reports no inode (st_ino == 0) cannot answer the question; # the check says nothing there rather than passing it off as unique, and # the gate reports that control as unavailable by name. declared = [str(artifact["path"]) for artifact in artifacts] identities: Dict[Tuple[int, int], str] = {} for artifact in artifacts: relative = str(artifact["path"]) target = package_dir / relative walked = package_dir link_at: Optional[Path] = None for segment in PurePosixPath(relative).parts: walked = walked / segment if is_link_or_reparse_point(walked): link_at = walked break if link_at is not None: problems.append( f"LINK OR REPARSE POINT: {relative} (reached through {link_at.name}, which " f"redirects to another location; a signed descriptor is not a licence to " f"follow one)" ) continue if not target.exists(): problems.append(f"MISSING: {relative}") continue if not target.is_file(): problems.append( f"NOT A REGULAR FILE: {relative} (a descriptor must name a file with bytes " f"to hash)" ) continue try: status = target.stat() except OSError as exc: problems.append(f"UNREADABLE: {relative} ({type(exc).__name__}: {exc})") continue if status.st_ino: identity = (status.st_dev, status.st_ino) twin = identities.get(identity) if twin is not None and twin != relative: problems.append( f"SAME FILE: {relative} and {twin} are two descriptors resolving to ONE " f"file object (device {status.st_dev}, inode {status.st_ino}); a manifest " f"that names one file twice reports more coverage than it holds" ) continue identities[identity] = relative artifact = dict(artifact, path=relative) data = target.read_bytes() actual = sha256_hex(data) if actual != artifact.get("sha256"): problems.append( f"HASH MISMATCH: {artifact['path']} " f"(expected {str(artifact.get('sha256'))[:16]}..., got {actual[:16]}...)" ) elif len(data) != artifact.get("bytes"): problems.append(f"SIZE MISMATCH: {artifact['path']}") elif verbose: print(f" {artifact['path']}: PASS ({len(data)} bytes)") # The other direction: a file present in the package that no descriptor # names. Reported by name, because "something here is not covered" is not # an answer an auditor can act on. # # MEMBERSHIP IS EXACT AGAINST THE PACKAGE ROOT. # EXACTLY ONE file is exempt: manifest.json itself, which cannot carry its # own hash. That is the one documented gap and there is no other. # # This used to also subtract every document named on the command line, on # the theory that a file you asked the verifier to CHECK is the subject # under test rather than inventory. The theory does not survive contact # with the boundary: membership is defined against the PACKAGE ROOT -- the # directory holding manifest.json -- and that is also the only directory # walked, so an external trust input (an export a counterparty sent you, a # keyring you were told to pin) sitting outside the package was never in # the present set and never needed exempting. The exemption could only ever # fire for a file INSIDE the package -- and every such file is one of the # ten artifacts README.md says the manifest must cover. Naming # session_export.json or public_keyring.json on the command line therefore # bought that file a pass out of the exact set it was supposed to be in, # and the whole-export mode, which names both, exempted both. exempt = {manifest_path.name} present: List[str] = [] for directory, _subdirectories, filenames in os.walk(package_dir): for filename in filenames: present.append((Path(directory) / filename).relative_to(package_dir).as_posix()) unlisted = sorted(set(present) - set(declared) - exempt) for name in unlisted: problems.append( f"UNLISTED: {name} (present in the package and named by no descriptor; the " f"manifest covers every artifact except itself)" ) if problems: return False, "manifest membership failed:\n " + "\n ".join(problems) return True, f"{message}; {len(artifacts)} artifact hashes match{note}" # --- chain ---------------------------------------------------------------- def verify_chain( events: List[Dict[str, Any]], public_key_hex: Optional[str], keyring: Optional[Dict[str, Dict[str, Any]]], expect_key_id: Optional[str] = None, ) -> Tuple[bool, List[str], List[Optional[str]], List[Tuple[str, Optional[datetime]]]]: """Walk prev_event_hash links and per-event signatures in submitted order. Returns (ok, diagnostics, the key id each event declared, the instant each event declared). Each event's key window is compared against ITS OWN instant; an event that declares none has no window comparison, and the caller says so rather than substituting the export's date for it. """ diagnostics: List[str] = [] key_ids: List[Optional[str]] = [] stamps: List[Tuple[str, Optional[datetime]]] = [] if not events: return False, ["empty chain: an evidence chain with no events is not evidence"], key_ids, stamps prev_event_hash = CHAIN_HEAD_SENTINEL for index, event in enumerate(events): label = f"event[{index}]" if event.get("sequence") != index: diagnostics.append( f"{label}: sequence={event.get('sequence')!r}; expected {index} " f"(events are not in submitted order)" ) return False, diagnostics, key_ids, stamps payload = event.get("payload") if payload is None: diagnostics.append(f"{label}: missing 'payload'") return False, diagnostics, key_ids, stamps try: version = resolve_canonicalization_version(event) recomputed = sha256_hex(canonical_json(payload, version)) except CanonicalizationError as exc: diagnostics.append(f"{label}: canonicalization rejected: {exc}") return False, diagnostics, key_ids, stamps if event.get("metadata_hash") != recomputed: diagnostics.append( f"{label}: metadata_hash mismatch -- declared={event.get('metadata_hash')!r} " f"recomputed={recomputed!r} (payload was modified)" ) return False, diagnostics, key_ids, stamps if event.get("prev_event_hash") != prev_event_hash: diagnostics.append( f"{label}: prev_event_hash={event.get('prev_event_hash')!r} expected={prev_event_hash!r} " f"(chain link broken here; the previous event was modified or removed)" ) return False, diagnostics, key_ids, stamps # The event's OWN hash: sha256 over the event minus its own hash and # its signature. This value was already computed to link the NEXT # event and was never compared to what the event DECLARED, so an event # could carry `event_hash: "dead...be"` inside its correctly signed # body and every layer still passed. Declaring it is # optional; declaring it falsely is not. Checked here, alongside # metadata_hash and the chain link, so a hash lie is reported as a hash # lie rather than reaching the signature first. body_for_hash = {k: v for k, v in event.items() if k not in ("event_hash", "event_signature")} own_hash = sha256_hex(canonical_json(body_for_hash, version)) declared_hash = dig(event, ("event_hash",)) checked = "metadata hash + chain link + signature" if declared_hash is not ABSENT: if declared_hash != own_hash: diagnostics.append( f"{label}: event_hash mismatch -- declared={declared_hash!r} " f"recomputed={own_hash!r} (the event declares a hash of itself that is not " f"the hash of itself)" ) return False, diagnostics, key_ids, stamps checked += " + declared event_hash" signature = event.get("event_signature") if not signature: diagnostics.append(f"{label}: missing 'event_signature'; integrity cannot be checked") return False, diagnostics, key_ids, stamps signature_fault = signature_problem(signature, f"{label} event_signature") if signature_fault: diagnostics.append(f"{label}: {signature_fault}") return False, diagnostics, key_ids, stamps body = {k: v for k, v in event.items() if k != "event_signature"} event_at, timestamp_error = artifact_timestamp(event, EVENT_TIME_PATHS, label) if timestamp_error: diagnostics.append(f"{label}: {timestamp_error}") return False, diagnostics, key_ids, stamps stamps.append((label, event_at)) resolved, key_id, error = resolve_key( body, public_key_hex, keyring, label, expect_key_id, event_at ) key_ids.append(key_id) if error or resolved is None: diagnostics.append(f"{label}: {error or 'key resolution failed'}") return False, diagnostics, key_ids, stamps ok, message = verify_signed_object(body, signature, resolved) if not ok: diagnostics.append(f"{label}: {message}") return False, diagnostics, key_ids, stamps diagnostics.append(f"{label} sequence={index}: PASS ({checked})") prev_event_hash = own_hash return True, diagnostics, key_ids, stamps def compute_chain_anchor(events: List[Dict[str, Any]]) -> Dict[str, Any]: tip = events[-1] version = resolve_canonicalization_version(tip) body_for_hash = {k: v for k, v in tip.items() if k not in ("event_hash", "event_signature")} return { "algorithm": CHAIN_HASH_ALGORITHMS[version], "event_count": len(events), "last_event_hash": sha256_hex(canonical_json(body_for_hash, version)), } def extract_events(raw: Any) -> Optional[List[Dict[str, Any]]]: """Accept a raw array, an {events:[...]} wrapper, or a session export.""" if isinstance(raw, list): return raw if isinstance(raw, dict): if isinstance(raw.get("events"), list): return raw["events"] chain = raw.get("chain") if isinstance(chain, dict) and isinstance(chain.get("events"), list): return chain["events"] return None # --- cross-layer binding and chain anchor --------------------------------- class _Absent: """Sentinel for a path that is not present at all. See dig().""" __slots__ = () def __repr__(self) -> str: # pragma: no cover - diagnostic only return "" ABSENT = _Absent() def dig(obj: Any, path: Sequence[str]) -> Any: """Walk a dotted path through nested dicts; ABSENT if any hop is missing. Returning None for a missing hop -- which is what this did before 2026-09-02 -- makes `{"session_id": null}` indistinguishable from `{}`. They are not the same thing: the first is a DECLARATION whose value is not an identity, and dropping it silently is how a non-string declaration got treated as absence. Presence is decided by key membership, and the caller types the value. """ current = obj for segment in path: if not isinstance(current, dict) or segment not in current: return ABSENT current = current[segment] return current def typed_identity_error( value: Any, layer: str, field: str, path: str ) -> Optional[str]: """The rejection reason for a present-but-untyped identity, or None. An identity is a non-empty string. Anything else AT A PRESENT PATH is a schema violation that must be refused by name -- naming the layer, the path and the offending type -- not silently skipped so that the remaining paths compare clean and the success line reports agreement that was never tested. """ if isinstance(value, str) and value: return None return ( f"{layer} declares {field} at {path} as {type(value).__name__} {value!r}; " f"identity must be a non-empty string" ) def keyed_identity( obj: Any, key: str, layer: str, field: str, absent_reason: str ) -> Tuple[Optional[str], Optional[str]]: """One directly-keyed identity. Returns (value, error). Absence gets the caller's own required-field reason; presence is typed. """ value = dig(obj, (key,)) if value is ABSENT: return None, absent_reason error = typed_identity_error(value, layer, field, key) if error: return None, error return value, None def declared_identity( obj: Any, paths: Sequence[Sequence[str]], layer: str, field: str, ) -> Tuple[Optional[str], Optional[str]]: """Resolve ONE identity across every declared path. Returns (value, error). Every path that is PRESENT is read and they must AGREE. Returning the first non-empty path instead -- which is what this did before 2026-09-02 -- lets a signed object declare two different identities and be admitted under whichever one the verifier happened to consult first, while the success line reports only the winner. That is a cross-session mix, not a schema variation, so it is refused by name and both offending paths are printed. Presence is a declaration and type is a schema violation, not absence. A path holding a number, an array, an object, a boolean, null or the empty string is REFUSED by name rather than skipped: skipping it left the remaining path comparing against itself and the binding line then reported agreement across an object whose own request block named a different session. (None, None) means the object declared the identity nowhere; the caller decides whether that layer is allowed to be silent. In this verifier, none of them is. """ found: List[Tuple[str, str]] = [] for path in paths: value = dig(obj, path) if value is ABSENT: continue rendered_path = ".".join(path) error = typed_identity_error(value, layer, field, rendered_path) if error: return None, error found.append((rendered_path, value)) if not found: return None, None distinct = {value for _, value in found} if len(distinct) > 1: rendered = ", ".join(f"{name}={value!r}" for name, value in found) return None, ( f"{layer} declares conflicting {field} values ({rendered}); an object that " f"names two identities cannot be bound to one" ) return found[0][1], None def receipt_identity( receipt: Dict[str, Any], layer: str = "receipt" ) -> Tuple[Optional[str], Optional[str], Optional[str]]: """The (session_id, tenant_id, error) a receipt names, across both schemas.""" session, session_error = declared_identity(receipt, RECEIPT_SESSION_PATHS, layer, "session_id") if session_error: return None, None, session_error tenant, tenant_error = declared_identity(receipt, RECEIPT_TENANT_PATHS, layer, "tenant_id") if tenant_error: return None, None, tenant_error return session, tenant, None def check_binding( bundle: Dict[str, Any], chain: Dict[str, Any], events: List[Dict[str, Any]], receipts: List[Dict[str, Any]], ) -> Tuple[bool, str]: """Bind every signed object in the export to ONE session and ONE tenant. Each layer self-verifies, but a validly signed chain, event or receipt from session B can be mixed into session A's export and every layer still passes. This is the layer that catches that, and it must read the RECEIPTS as well as the events: a receipt that is not compared is a receipt that can be swapped. The contract is stated once in the module docstring and enforced here: every layer that can carry identity MUST carry it, and every identity it carries must match. Absence is a rejection on every one of those layers, and a layer that declares an identity twice must declare it the same way both times. Identity is TYPED on every layer: a path that is present but holds anything other than a non-empty string is refused by name, with the layer, the path and the offending type printed, rather than compared with != or dropped. The returned success message lists only the layers actually compared. """ session_id, error = keyed_identity( bundle, "session_id", "bundle", "session_id", "bundle.session_id is required to bind the layers", ) if error: return False, error bundle_tenant, error = keyed_identity( bundle, "tenant_id", "bundle", "tenant_id", "bundle.tenant_id is required to bind the layers; without it the export is " "asserted to hold one tenant that the bundle itself never declared", ) if error: return False, error chain_session, error = keyed_identity( chain, "session_id", "chain wrapper", "session_id", "chain.session_id is required to bind the layers", ) if error: return False, error if chain_session != session_id: return False, ( f"chain.session_id={chain_session!r} != bundle.session_id {session_id!r} " f"(cross-session wrapper mix)" ) chain_tenant, error = keyed_identity( chain, "tenant_id", "chain wrapper", "tenant_id", "chain.tenant_id is required to bind the layers; a chain wrapper that names no " "tenant cannot be reported as declaring one", ) if error: return False, error if chain_tenant != bundle_tenant: return False, ( f"mixed tenant_id across signed objects: the chain wrapper declares " f"{chain_tenant!r} but bundle.tenant_id is {bundle_tenant!r} " f"(cross-tenant wrapper mix)" ) for index, event in enumerate(events): event_session, error = keyed_identity( event, "session_id", f"event[{index}]", "session_id", f"event[{index}].session_id is required to bind the layers", ) if error: return False, error if event_session != session_id: return False, ( f"event[{index}].session_id={event_session!r} != bundle session " f"{session_id!r} (cross-session mix)" ) tenant, error = keyed_identity( event, "tenant_id", f"event[{index}]", "tenant_id", f"event[{index}].tenant_id is required to bind tenants", ) if error: return False, error if tenant != bundle_tenant: return False, ( f"mixed tenant_id across signed objects: event[{index}] declares {tenant!r} " f"but bundle.tenant_id is {bundle_tenant!r}" ) for index, receipt in enumerate(receipts): label = f"receipt[{index}]" receipt_session, receipt_tenant, identity_error = receipt_identity(receipt, label) if identity_error: return False, identity_error if receipt_session is None: return False, ( f"{label} names no session at request.session_id or session_id; " f"an unbindable receipt cannot be admitted to a bound export" ) if receipt_session != session_id: return False, ( f"{label}.session={receipt_session!r} != bundle session " f"{session_id!r} (cross-session mix)" ) if receipt_tenant is None: return False, ( f"{label} names no tenant at request.tenant_id or tenant_id " f"(required to bind tenants)" ) if receipt_tenant != bundle_tenant: return False, ( f"mixed tenant_id across signed objects: {label} declares {receipt_tenant!r} " f"but bundle.tenant_id is {bundle_tenant!r}" ) # Name the layers that were compared, and nothing else. A fixed sentence # would keep claiming "every event" on an export that carried none. compared = ["bundle", "chain wrapper"] if events: compared.append(f"{len(events)} event(s)") if receipts: compared.append(f"{len(receipts)} receipt(s)") layer_list = compared[0] if len(compared) == 1 else ", ".join(compared[:-1]) + f" and {compared[-1]}" return True, ( f"{layer_list} all declare session {session_id} and tenant {bundle_tenant}" ) def check_anchor( chain: Dict[str, Any], events: List[Dict[str, Any]], keyring_mode: bool, ) -> Tuple[str, str]: """Recompute the signed chain-tip anchor. Returns (PASS|FAIL|SKIP, message). Without an anchor an export can be truncated to a shorter prefix of its own chain and every remaining layer still verifies, so keyring verification requires one. The legacy explicit-key exception is narrow, reported, and never applies once a keyring is pinned. """ declared = chain.get("anchor") if declared is None: if keyring_mode: return "FAIL", ( "keyring verification requires chain.anchor; without the signed chain tip an " "export can be truncated to a prefix of its own chain and still verify" ) return "SKIP", ( "absent; legacy explicit-key exception for pre-anchor bundles " "(keyring verification requires it)" ) if not isinstance(declared, dict): return "FAIL", f"chain.anchor must be an object; got {type(declared).__name__}" if set(declared) != CHAIN_ANCHOR_FIELDS: return "FAIL", ( f"chain.anchor fields must be exactly {sorted(CHAIN_ANCHOR_FIELDS)}; " f"got {sorted(declared)}" ) if not events: return "FAIL", "chain.anchor is declared but the chain carries no events to recompute it from" try: expected = compute_chain_anchor(events) except (CanonicalizationError, KeyError, TypeError) as exc: return "FAIL", f"chain.anchor could not be recomputed from the submitted chain: {exc}" if declared != expected: return "FAIL", ( f"declared anchor does not match the submitted chain tip -- " f"declared {json.dumps(declared, sort_keys=True)} " f"recomputed {json.dumps(expected, sort_keys=True)}" ) return "PASS", ( f"chain tip recomputed: {expected['event_count']} events, {expected['algorithm']}" ) # --- session export ------------------------------------------------------- def verify_session_export( export_path: Path, public_key_hex: Optional[str], keyring: Optional[Dict[str, Dict[str, Any]]], verbose: bool, expect_key_id: Optional[str] = None, ) -> Tuple[bool, List[Tuple[str, str, str]]]: """Verify envelope, chain, receipts, manifest, binding and chain anchor. Every component is reported with its own status; the overall result is PASS only when no component FAILs. SKIP is reserved for the one documented legacy exception (see check_anchor) and never hides a checkable property. """ raw = load_json_document(str(export_path), "--session-export") problem = session_export_root_problem(raw) if problem: raise InputError(f"--session-export {problem}") bundle, siblings = unwrap_session_export(raw) results: List[Tuple[str, str, str]] = [] def record(component: str, ok: bool, message: str) -> None: results.append((component, "PASS" if ok else "FAIL", message)) keyring_mode = keyring is not None # Fields sitting BESIDE veilengine_session_export inside the wrapper are # outside the signed envelope. They used to be dropped in silence, so a # wrapper could carry session_id "sess_other_source" and # audit_opinion "UNQUALIFIED PASS" alongside the real export and the run # reported PASS without mentioning them. An unsigned claim # travelling inside a signed evidence file is refused by name. if siblings: record( "wrapper", False, f"the veilengine_session_export wrapper carries unsigned sibling field(s) " f"[{siblings}]; they sit outside the signed envelope, nothing in this file " f"attests to them, and they must not travel with it" ) # The instant this export declares it was produced, and the instant the # pinned keyring's validity windows are enforced against. export_at, timestamp_error = artifact_timestamp( bundle, (("export_generated_utc",),), "bundle" ) if timestamp_error: record("validity", False, timestamp_error) return False, results if keyring_mode and export_at is None: record( "validity", False, "keyring verification requires bundle.export_generated_utc; without the export's own " "timestamp the keyring's key validity windows cannot be enforced, and a key that was " "not yet minted or had already expired would resolve anyway" ) return False, results signature = bundle.get("signature") if not signature: return False, results + [("envelope", "FAIL", "bundle has no top-level 'signature'")] envelope_signature_fault = signature_problem(signature, "envelope bundle.signature") if envelope_signature_fault: return False, results + [("envelope", "FAIL", envelope_signature_fault)] resolved, envelope_key_id, error = resolve_key( bundle, public_key_hex, keyring, "envelope", expect_key_id, export_at ) if error or resolved is None: return False, results + [("envelope", "FAIL", error or "key resolution failed")] body = {k: v for k, v in bundle.items() if k != "signature"} envelope_ok, envelope_message = verify_signed_object(body, signature, resolved) record("envelope", envelope_ok, envelope_message + key_id_note([envelope_key_id], keyring_mode)) chain = bundle.get("chain") if isinstance(bundle.get("chain"), dict) else {} events = chain.get("events") if isinstance(chain.get("events"), list) else [] chain_ok, diagnostics, chain_key_ids, chain_stamps = verify_chain( events, public_key_hex, keyring, expect_key_id ) if verbose: for line in diagnostics: print(f" {line}") record( "chain", chain_ok, (f"{len(events)} events; prev_event_hash links + per-event signatures" + key_id_note(chain_key_ids, keyring_mode)) if chain_ok else diagnostics[-1], ) receipts = bundle.get("receipts") if isinstance(bundle.get("receipts"), list) else [] receipt_stamps: List[Tuple[str, Optional[datetime]]] = [] if not receipts: record("receipts", False, "bundle carries no receipts") else: failures = [] receipt_key_ids: List[Optional[str]] = [] for index, receipt in enumerate(receipts): receipt_at, timestamp_error = artifact_timestamp( receipt, RECEIPT_TIME_PATHS, f"receipt[{index}]" ) if timestamp_error: ok, message, key_id = False, timestamp_error, None else: receipt_stamps.append((f"receipt[{index}]", receipt_at)) ok, message, key_id = verify_receipt_obj( receipt, public_key_hex, keyring, f"receipt[{index}]", expect_key_id, receipt_at, ) receipt_key_ids.append(key_id) if verbose: print(f" {message}") if not ok: failures.append(message) record( "receipts", not failures, (f"{len(receipts)} receipt(s) verified under their own canonicalization" + key_id_note(receipt_key_ids, keyring_mode)) if not failures else failures[0], ) manifest_name = bundle.get("manifest_file") if not isinstance(manifest_name, str) or not manifest_name: record("manifest", False, "bundle declares no 'manifest_file'") else: manifest_path = export_path.parent / manifest_name if not manifest_path.exists(): record("manifest", False, f"declared manifest not found: {manifest_name}") else: record("manifest", *verify_manifest_file( manifest_path, public_key_hex, keyring, verbose, expect_key_id, export_at)) record("binding", *check_binding(bundle, chain, events, receipts)) results.append(("anchor", *check_anchor(chain, events, keyring_mode))) # Named, not implied. Each signed object's key window was compared against # THAT OBJECT's own instant; objects that declare none are listed as not # compared rather than quietly counted as passing. Substituting the # export's date for an event's would refuse a pre-rotation event signed by # the retired key -- exactly the historical verification that key exists # for -- so the rule is per object, and the line says which. if keyring_mode: stamps: List[Tuple[str, Optional[datetime]]] = ( [("envelope", export_at)] + list(chain_stamps) + list(receipt_stamps) + [("manifest", None)] ) compared = [name for name, when in stamps if when is not None] uncompared = [name for name, when in stamps if when is None] message = ( f"key validity windows compared against each object's own timestamp for " f"{', '.join(compared)} ({len(compared)} of {len(stamps)} signed objects); " f"every one was inside its key's window" ) if uncompared: message += ( f". {', '.join(uncompared)} declare no timestamp, so no window was compared " f"for them" ) record("validity", True, message) return all(status != "FAIL" for _, status, _ in results), results # --- offline self-test ---------------------------------------------------- def self_test() -> int: """Sign synthetic payloads under both canonicalizations; confirm behavior.""" from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat def new_key(index: int) -> Tuple[Any, str]: """Derive throwaway key `index` from the published synthetic seed. Deterministic on purpose. `Ed25519PrivateKey.generate()` made three key fingerprints change on every run, which made the recorded transcript unreproducible and therefore unmanifestable. """ seed = hashlib.sha256(f"{SELF_TEST_KEY_SEED}/{index}".encode("utf-8")).digest() key = Ed25519PrivateKey.from_private_bytes(seed) return key, key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw).hex() private_key, public_hex = new_key(0) checks: List[Tuple[str, bool]] = [] def check(name: str, passed: bool, detail: str) -> None: checks.append((name, passed)) print(f" [{'PASS' if passed else 'FAIL'}] {name}: {detail}") # The payload is deliberately DIVERGENT: a float and an astral-plane key # canonicalize to different bytes under the two schemes (legacy keeps the # trailing ".0" and sorts keys by code point; JCS drops it and sorts by # UTF-16 code unit). A payload that canonicalized identically would make # every check below pass without exercising the version resolution at all. def divergent(declared: Optional[str]) -> Dict[str, Any]: payload: Dict[str, Any] = { "n": 1, "latency_ms": 1847.0, "nested": {"b": [1, 2], "a": True}, "￿": "bmp", "\U00010000": "astral", } if declared is not None: payload["canonicalization_version"] = declared return payload check("the two canonicalizations produce different bytes for this payload", canonical_json(divergent(None), LEGACY_CANON) != canonical_json(divergent(None), JCS_CANON), f"legacy={len(canonical_json(divergent(None), LEGACY_CANON))}B " f"jcs={len(canonical_json(divergent(None), JCS_CANON))}B") for label, declared in (("absent field", None), ("declared", LEGACY_CANON), ("declared", JCS_CANON)): payload = divergent(declared) version = resolve_canonicalization_version(payload) signature = "ed25519:" + private_key.sign(canonical_json(payload, version)).hex() ok, message = verify_signed_object(payload, signature, public_hex) check(f"valid signature accepted, {label} {version}", ok, message) ok, message = verify_signed_object(dict(payload, n=2), signature, public_hex) check(f"modified payload rejected, {label} {version}", not ok, message) cross = divergent(JCS_CANON) wrong_canon_sig = "ed25519:" + private_key.sign(canonical_json(cross, LEGACY_CANON)).hex() ok, message = verify_signed_object(cross, wrong_canon_sig, public_hex) check("signature made under the wrong canonicalization rejected", not ok, message) unknown = {"canonicalization_version": "someone-elses-canon-v9", "n": 1} ok, message = verify_signed_object(unknown, wrong_canon_sig, public_hex) check("unknown canonicalization_version rejected by name", not ok and "not supported" in message, message) good = {"canonicalization_version": JCS_CANON, "n": 1} signature = "ed25519:" + private_key.sign(canonical_json(good, JCS_CANON)).hex() _, other_hex = new_key(1) ok, message = verify_signed_object(good, signature, other_hex) check("signature checked against the wrong key rejected", not ok, message) # --- cross-layer binding (the cross-session layer) --------------------------- # Every case below is a validly SIGNED object; only its identity differs. # That is the whole point: signature checks cannot catch a cross-session mix. bundle = {"session_id": "sess_a", "tenant_id": "tenant_1"} chain = {"session_id": "sess_a", "tenant_id": "tenant_1"} events = [{"session_id": "sess_a", "tenant_id": "tenant_1"}] def receipt_of( session: Optional[str], tenant: Optional[str], nested: bool = True, top_session: Optional[str] = None, top_tenant: Optional[str] = None, ) -> Dict[str, Any]: """Build a receipt declaring identity nested, at the top level, or both. `top_session` / `top_tenant` add a SECOND declaration alongside the nested one, which is how the dual-path conflict cases are built. """ identity: Dict[str, Any] = {} if session is not None: identity["session_id"] = session if tenant is not None: identity["tenant_id"] = tenant receipt: Dict[str, Any] = ( {"receipt_id": "rcpt_self_test", "request": identity} if nested else dict({"receipt_id": "rcpt_self_test"}, **identity) ) if top_session is not None: receipt["session_id"] = top_session if top_tenant is not None: receipt["tenant_id"] = top_tenant return receipt ok, message = check_binding(bundle, chain, events, [receipt_of("sess_a", "tenant_1")]) check("binding accepts one session and one tenant across every signed object", ok, message) ok, message = check_binding(bundle, chain, events, [receipt_of("sess_a", "tenant_1", nested=False)]) check("binding reads top-level receipt identity too (gateway schema)", ok, message) ok, message = check_binding(bundle, chain, events, [receipt_of("sess_other", "tenant_1")]) check("cross-session receipt mix rejected", not ok and "cross-session mix" in message, message) ok, message = check_binding(bundle, chain, events, [receipt_of("sess_a", "tenant_2")]) check("mixed-tenant receipt rejected", not ok and "mixed tenant_id" in message, message) ok, message = check_binding(bundle, chain, events, [receipt_of(None, "tenant_1")]) check("receipt naming no session rejected, not skipped", not ok and "names no session" in message, message) ok, message = check_binding(bundle, chain, events, [receipt_of("sess_a", None)]) check("receipt naming no tenant rejected", not ok and "names no tenant" in message, message) ok, message = check_binding( bundle, {"session_id": "sess_other", "tenant_id": "tenant_1"}, events, [receipt_of("sess_a", "tenant_1")], ) check("cross-session chain wrapper rejected", not ok and "wrapper mix" in message, message) ok, message = check_binding( bundle, chain, [{"session_id": "sess_other", "tenant_id": "tenant_1"}], [receipt_of("sess_a", "tenant_1")], ) check("cross-session event mix rejected", not ok and "cross-session mix" in message, message) # --- the schema contract: every identity-bearing layer must carry it ---- # Each case below is a validly SIGNED, internally consistent export whose # only defect is an identity a layer did not declare, or declared twice and # differently. Both classes verified clean before 2026-09-02. ok, message = check_binding( {"session_id": "sess_a"}, chain, events, [receipt_of("sess_a", "tenant_1")]) check("bundle declaring no tenant rejected, not treated as optional", not ok and "bundle.tenant_id is required" in message, message) ok, message = check_binding( bundle, {"session_id": "sess_a"}, events, [receipt_of("sess_a", "tenant_1")]) check("chain wrapper declaring no tenant rejected", not ok and "chain.tenant_id is required" in message, message) ok, message = check_binding( bundle, {"session_id": "sess_a", "tenant_id": "tenant_2"}, events, [receipt_of("sess_a", "tenant_1")], ) check("chain wrapper declaring a different tenant rejected", not ok and "mixed tenant_id" in message and "chain wrapper" in message, message) ok, message = check_binding( bundle, chain, events, [receipt_of("sess_a", "tenant_1", top_session="sess_other")], ) check("receipt declaring conflicting nested and top-level session rejected", not ok and "conflicting session_id" in message and "receipt[0]" in message, message) ok, message = check_binding( bundle, chain, events, [receipt_of("sess_a", "tenant_1", top_tenant="tenant_2")], ) check("receipt declaring conflicting nested and top-level tenant rejected", not ok and "conflicting tenant_id" in message and "receipt[0]" in message, message) ok, message = check_binding( bundle, chain, events, [receipt_of("sess_a", "tenant_1", top_session="sess_a", top_tenant="tenant_1")], ) check("receipt declaring the SAME identity at both paths accepted", ok, message) ok, message = check_binding(bundle, chain, events, [receipt_of("sess_a", "tenant_1")]) check("the binding success line names only the layers it compared", ok and "bundle, chain wrapper, 1 event(s) and 1 receipt(s)" in message and "tenant tenant_1" in message, message) # --- identity is TYPED, and presence is not absence --------------------- # Each case declares the identity at a path that IS PRESENT, holding a # value that is not a non-empty string, while the other path carries the # bundle's own identity. Before 2026-09-02 the non-string path was dropped, # the remaining path compared against itself, and the all-object binding # assurance was printed over a comparison that never happened. TYPE_CLASSES: List[Tuple[str, Any, str]] = [ ("a number", 999, "int"), ("an array", ["sess_evil"], "list"), ("an object", {"id": "sess_evil"}, "dict"), ("a boolean", False, "bool"), ("null", None, "NoneType"), ("the empty string", "", "str"), ] for label, planted, type_name in TYPE_CLASSES: receipt = { "receipt_id": "rcpt_self_test", "request": {"session_id": planted, "tenant_id": "tenant_1"}, "session_id": "sess_a", } ok, message = check_binding(bundle, chain, events, [receipt]) check( f"receipt declaring session_id as {label} rejected by name, not treated as absent", not ok and "declares session_id at request.session_id" in message and f"as {type_name} " in message and "identity must be a non-empty string" in message and "receipt[0]" in message, message, ) receipt = { "receipt_id": "rcpt_self_test", "request": {"session_id": "sess_a", "tenant_id": planted}, "tenant_id": "tenant_1", } ok, message = check_binding(bundle, chain, events, [receipt]) check( f"receipt declaring tenant_id as {label} rejected by name, not treated as absent", not ok and "declares tenant_id at request.tenant_id" in message and f"as {type_name} " in message and "identity must be a non-empty string" in message, message, ) receipt = {"receipt_id": "rcpt_self_test", "session_id": planted, "tenant_id": "tenant_1"} ok, message = check_binding(bundle, chain, events, [receipt]) check( f"receipt declaring TOP-LEVEL session_id as {label} rejected by name", not ok and "declares session_id at session_id" in message and f"as {type_name} " in message, message, ) # The same typing rule on the layers that are read by direct key, so a # non-string there is refused by name rather than compared with != (which # produced a "cross-session mix" reason for a schema violation) or read # as a missing required field. ok, message = check_binding(dict(bundle, session_id=999), chain, events, [receipt_of("sess_a", "tenant_1")]) check("bundle declaring a non-string session_id rejected by name", not ok and "bundle declares session_id at session_id as int 999" in message, message) ok, message = check_binding(bundle, dict(chain, tenant_id=["tenant_1"]), events, [receipt_of("sess_a", "tenant_1")]) check("chain wrapper declaring a non-string tenant_id rejected by name", not ok and "chain wrapper declares tenant_id at tenant_id as list" in message, message) ok, message = check_binding(bundle, chain, [{"session_id": None, "tenant_id": "tenant_1"}], [receipt_of("sess_a", "tenant_1")]) check("event declaring a null session_id rejected by name, not as a cross-session mix", not ok and "event[0] declares session_id at session_id as NoneType None" in message and "cross-session mix" not in message, message) ok, message = check_binding(bundle, chain, [{"tenant_id": "tenant_1"}], [receipt_of("sess_a", "tenant_1")]) check("event omitting session_id entirely rejected as a missing required field", not ok and "event[0].session_id is required" in message, message) # --- chain anchor (truncation defence) -------------------------------- anchor_events = [ {"sequence": 0, "canonicalization_version": LEGACY_CANON, "payload": {"n": 1}, "event_signature": "a"}, {"sequence": 1, "canonicalization_version": LEGACY_CANON, "payload": {"n": 2}, "event_signature": "b"}, ] current_anchor = compute_chain_anchor(anchor_events) status, message = check_anchor({"anchor": current_anchor}, anchor_events, keyring_mode=True) check("matching chain anchor accepted under the keyring", status == "PASS", message) status, message = check_anchor({}, anchor_events, keyring_mode=True) check("missing chain anchor rejected under the keyring", status == "FAIL", message) status, message = check_anchor({}, anchor_events, keyring_mode=False) check("missing chain anchor is the reported legacy exception under an explicit key", status == "SKIP", message) status, message = check_anchor({"anchor": "not-an-object"}, anchor_events, keyring_mode=True) check("malformed chain anchor rejected", status == "FAIL", message) status, message = check_anchor( {"anchor": dict(current_anchor, extra=1)}, anchor_events, keyring_mode=True) check("chain anchor with unexpected fields rejected", status == "FAIL", message) status, message = check_anchor( {"anchor": compute_chain_anchor(anchor_events[:1])}, anchor_events, keyring_mode=True) check("stale chain anchor (an earlier tip of the same chain) rejected", status == "FAIL", message) # --- keyring descriptor validation ------------------------------------ _, retired_hex = new_key(2) def descriptor(key_hex: str, key_id: Optional[str] = None, **overrides: Any) -> Dict[str, Any]: entry = { "alg": "ed25519", "key_id": key_id if key_id is not None else f"vke_self_test_{key_hex[:8]}", "public_key_hex": key_hex, "status": "active", "valid_from": "2026-01-01T00:00:00Z", "valid_until": None, } entry.update(overrides) return entry def keyring_of(*entries: Dict[str, Any]) -> Dict[str, Any]: return {"schema_version": KEYRING_SCHEMA_VERSION, "keys": list(entries)} def keyring_rejects(name: str, obj: Dict[str, Any], needle: str) -> None: try: parse_public_keyring(obj) except ValueError as exc: check(name, needle in str(exc), str(exc)) else: check(name, False, "ACCEPTED — the keyring loader did not reject it") try: loaded = parse_public_keyring(keyring_of( descriptor(public_hex), descriptor(retired_hex, key_id=key_fingerprint(retired_hex), status="retired", valid_until="2026-05-01T00:00:00Z"), )) check("valid keyring accepted (label id + derived id, active + retired)", len(loaded) == 2, f"{len(loaded)} descriptors loaded") except ValueError as exc: check("valid keyring accepted (label id + derived id, active + retired)", False, str(exc)) keyring_rejects("keyring valid_from of the wrong type rejected", keyring_of(descriptor(public_hex, valid_from=123)), "valid_from must be a non-empty string") keyring_rejects("keyring empty valid_from rejected", keyring_of(descriptor(public_hex, valid_from="")), "valid_from must be a non-empty string") keyring_rejects("keyring valid_until of the wrong type rejected", keyring_of(descriptor(public_hex, valid_until=123)), "valid_until must be a string or null") keyring_rejects("keyring duplicate key id rejected", keyring_of(descriptor(public_hex, key_id="vke_dup"), descriptor(retired_hex, key_id="vke_dup")), "duplicate key_id") keyring_rejects("keyring derived-shaped key id that does not derive rejected", keyring_of(descriptor(public_hex, key_id="0" * 16)), "key_id mismatch") keyring_rejects("keyring private-key-shaped field rejected by name", keyring_of(dict(descriptor(public_hex), private_key_hex="00" * 32)), "private-key-shaped") keyring_rejects("keyring aliasing one public key under two ids rejected", keyring_of(descriptor(public_hex, key_id="vke_a"), descriptor(public_hex, key_id="vke_b")), "same public key") keyring_rejects("keyring unknown status rejected", keyring_of(descriptor(public_hex, status="probationary")), "must be active, retired, or revoked") revoked_ring = parse_public_keyring(keyring_of(descriptor(public_hex, key_id="vke_revoked", status="revoked"))) _, _, error = resolve_key({"signing_key_id": "vke_revoked"}, None, revoked_ring, "object") check("revoked key refuses to resolve, naming the key id", bool(error) and "vke_revoked" in error and "revoked" in error, error or "resolved anyway") retired_ring = parse_public_keyring(keyring_of(descriptor(retired_hex, key_id="vke_retired", status="retired"))) resolved_hex, _, error = resolve_key({"signing_key_id": "vke_retired"}, None, retired_ring, "object") check("retired key still resolves, so historical events stay verifiable", resolved_hex == retired_hex and error is None, error or f"resolved {str(resolved_hex)[:16]}...") # --- key identity is held to the identity contract ---------------------- # Every path a signed object can name its key at is read and they must # agree; explicit-key mode binds a DERIVED-shape declared id to the # supplied key, prints a LABEL id on the layer line, and honours the # --expect-key-id pin. Before 2026-09-02 explicit-key mode consulted none # of this, which made the tool looser than the reference on the mode the # README leads with. _, error = declared_key_id( {"signing_key_id": "vke_a", "verification": {"signer_key_id": "vke_b"}}, "receipt[0]") check("object declaring conflicting key ids at two paths rejected by name", bool(error) and "conflicting key ids" in error and "signing_key_id='vke_a'" in error and "verification.signer_key_id='vke_b'" in error, error or "accepted anyway") _, error = declared_key_id({"signing_key_id": 7}, "envelope") check("key id declared as a non-string rejected by name", bool(error) and "declares a key id at signing_key_id as int 7" in error, error or "accepted anyway") resolved_hex, key_id, error = resolve_key( {"signing_key_id": "vke_a", "verification": {"signer_key_id": "vke_a"}}, public_hex, None, "envelope") check("agreeing key ids at two paths resolve to one id", resolved_hex == public_hex and key_id == "vke_a" and error is None, error or f"resolved key id {key_id!r}") _, _, error = resolve_key( {"signing_key_id": key_fingerprint(retired_hex)}, public_hex, None, "envelope") check("explicit-key mode rejects a DERIVED-shape key id that is not the supplied key's", bool(error) and key_fingerprint(retired_hex) in error and key_fingerprint(public_hex) in error and "does not match the supplied public key" in error, error or "accepted anyway") resolved_hex, _, error = resolve_key( {"signing_key_id": key_fingerprint(public_hex)}, public_hex, None, "envelope") check("explicit-key mode accepts the DERIVED-shape key id that IS the supplied key's", resolved_hex == public_hex and error is None, error or "accepted") _, _, error = resolve_key( {"signing_key_id": "vke_label"}, public_hex, None, "envelope", expect_key_id="vke_expected") check("--expect-key-id pins a label id in explicit-key mode", bool(error) and "vke_label" in error and "vke_expected" in error, error or "accepted anyway") _, _, error = resolve_key({}, public_hex, None, "envelope", expect_key_id="vke_expected") check("--expect-key-id rejects a layer that declares no key id at all", bool(error) and "declares no key id" in error, error or "accepted anyway") _, _, error = resolve_key( {"signing_key_id": "vke_retired"}, None, retired_ring, "object", expect_key_id="vke_other") check("--expect-key-id is enforced under the keyring too", bool(error) and "vke_retired" in error and "vke_other" in error, error or "accepted anyway") # --- input shape: every mode asserts what it accepts ------------------- # Each case below used to be a traceback and exit 1, or a clean PASS # over a layer that never ran. export_shaped = {"chain": {"events": []}, "signature": "ed25519:00"} receipt_shaped = {"receipt_id": "rcpt_x"} check("--receipt refuses an object that declares no receipt_id", (receipt_root_problem(export_shaped) or "").startswith("a receipt must declare"), str(receipt_root_problem(export_shaped))) check("--receipt refuses a JSON array outright", "must be a JSON object" in (receipt_root_problem([1, 2, 3]) or ""), str(receipt_root_problem([1, 2, 3]))) check("--receipt accepts a real receipt", receipt_root_problem(receipt_shaped) is None, str(receipt_root_problem(receipt_shaped))) check("--session-export refuses an object carrying no chain", "must carry a 'chain'" in (session_export_root_problem(receipt_shaped) or ""), str(session_export_root_problem(receipt_shaped))) check("--session-export accepts the wrapped and unwrapped shapes", session_export_root_problem(export_shaped) is None and session_export_root_problem({"veilengine_session_export": export_shaped}) is None, "both accepted") def artifacts_of(*entries: Any) -> Dict[str, Any]: return {"artifacts": list(entries)} good_artifact = {"path": "receipt.json", "sha256": "a" * 64, "bytes": 10} check("--manifest accepts a well-formed descriptor", manifest_root_problem(artifacts_of(good_artifact)) is None, str(manifest_root_problem(artifacts_of(good_artifact)))) check("--manifest refuses an EMPTY artifact list", "commits to nothing" in (manifest_root_problem(artifacts_of()) or ""), str(manifest_root_problem(artifacts_of()))) check("--manifest refuses artifacts that are not a list", "must be a list" in (manifest_root_problem({"artifacts": {"a": 1}}) or ""), str(manifest_root_problem({"artifacts": {"a": 1}}))) check("--manifest refuses a descriptor with no path", "is missing ['path']" in (manifest_root_problem( artifacts_of({"sha256": "a" * 64, "bytes": 1})) or ""), str(manifest_root_problem(artifacts_of({"sha256": "a" * 64, "bytes": 1})))) for outside_path in ("../outside.txt", "a/../../outside.txt", "..\\outside.txt"): problem = manifest_root_problem(artifacts_of(dict(good_artifact, path=outside_path))) check(f"--manifest refuses the traversing path {outside_path!r}", "resolves outside the package directory" in (problem or ""), str(problem)) for absolute in ("/etc/passwd", "C:/Windows/system.ini", "\\\\server\\share"): problem = manifest_root_problem(artifacts_of(dict(good_artifact, path=absolute))) check(f"--manifest refuses the absolute path {absolute!r}", "is absolute" in (problem or ""), str(problem)) # THE MANIFEST-PATH GRAMMAR. Each rule is checked by a # string that a filesystem would resolve to a package artifact while the # descriptor reads as a second, separate member. for bad_path, needle, why in ( ("README.md.", "ending in '.'", "Win32 strips a trailing dot"), ("README.md ", "ending in a space", "Win32 strips a trailing space"), ("README.md:hidden", "alternate-stream syntax", "NTFS ':' names a stream"), ("NUL", "reserved Windows device", "the null device is not an artifact"), ("NUL.txt", "reserved Windows device", "an extension does not lift the reservation"), ("nul.txt", "reserved Windows device", "device names are case-insensitive"), ("com1.json", "reserved Windows device", "COM1 with an extension"), ("LPT9.md", "reserved Windows device", "LPT9 with an extension"), ("cafe\u0301.txt", "not NFC-normalized", "decomposed accent, one file two spellings"), ("docs\\readme.md", "contains a backslash", "'\\' is a filename character on POSIX"), (".hidden", "outside the manifest-path grammar", "a segment must start alphanumeric"), ("a b.txt", "outside the manifest-path grammar", "space inside the segment"), ): problem = manifest_root_problem(artifacts_of(dict(good_artifact, path=bad_path))) check(f"--manifest refuses the aliasing path {bad_path!r} ({why})", needle in (problem or ""), str(problem)) for good_path in ("README.md", "docs/framework_mapping.md", "verifier_cli.py", "CONSOLE.md", "COM0.txt", "a-b_c.1.json"): problem = manifest_root_problem(artifacts_of(dict(good_artifact, path=good_path))) check(f"--manifest accepts the in-grammar path {good_path!r}", problem is None, str(problem)) check("--manifest refuses a sha256 that is not 64 hex characters", "must be 64 hex" in (manifest_root_problem( artifacts_of(dict(good_artifact, sha256="deadbeef"))) or ""), str(manifest_root_problem(artifacts_of(dict(good_artifact, sha256="deadbeef"))))) check("--manifest refuses a non-integer byte count", "must be a non-negative integer" in (manifest_root_problem( artifacts_of(dict(good_artifact, bytes="10"))) or ""), str(manifest_root_problem(artifacts_of(dict(good_artifact, bytes="10"))))) # MEMBERSHIP. Ten descriptors naming nine files, and a # descriptor naming the directory rather than a file in it. twin = dict(good_artifact) check("--manifest refuses two descriptors naming one file", "already declared at" in (manifest_root_problem( {"artifacts": [dict(twin), dict(twin)]}) or ""), str(manifest_root_problem({"artifacts": [dict(twin), dict(twin)]}))) upper = dict(good_artifact, path=str(good_artifact["path"]).upper()) check("--manifest refuses two descriptors differing only by case", "only by case" in (manifest_root_problem( {"artifacts": [dict(twin), upper]}) or ""), str(manifest_root_problem({"artifacts": [dict(twin), upper]}))) for directoryish, needle in ((".", "names the package directory itself"), ("./x.json", "not in canonical form"), ("a//b.json", "not in canonical form"), ("docs/", "ends in a separator")): problem = manifest_root_problem(artifacts_of(dict(good_artifact, path=directoryish))) check(f"--manifest refuses the non-file path {directoryish!r}", needle in (problem or ""), str(problem)) # SIGNATURE TYPE AND ENCODING. A truthy non-string # passed every layer's truth test and reached .strip(). for shape in (123, ["ed25519:aa"], {"hex": "aa"}, True, 1.5): problem = signature_problem(shape, "test signature") check(f"a {type(shape).__name__} is refused as a signature, by type", "must be a hex string" in (problem or ""), str(problem)) check("an empty signature is refused as empty", "is empty" in (signature_problem(" ", "test signature") or ""), str(signature_problem(" ", "test signature"))) check("a hex string is accepted as a signature shape", signature_problem("ed25519:" + "ab" * 64, "test signature") is None, str(signature_problem("ed25519:" + "ab" * 64, "test signature"))) ok_typed, message_typed = ed25519_verify("ab" * 32, b"payload", ["not", "a", "string"]) check("ed25519_verify judges a wrong-typed signature instead of raising", not ok_typed and "must be a hex string" in message_typed, message_typed) try: json.loads('{"session_id":"sess_other_source","session_id":"sess_real"}', object_pairs_hook=_reject_duplicate_keys) except ValueError as exc: check("a JSON object naming one key twice is refused by key name", "duplicate JSON key 'session_id'" in str(exc), str(exc)) else: check("a JSON object naming one key twice is refused by key name", False, "ACCEPTED -- last-wins parsing verified bytes the reader never sees") try: parsed = json.loads('{"a":1,"b":{"a":2}}', object_pairs_hook=_reject_duplicate_keys) except ValueError as exc: # pragma: no cover - a nested reuse is not a duplicate check("the duplicate-key rule is per object, not per document", False, str(exc)) else: check("the duplicate-key rule is per object, not per document", parsed == {"a": 1, "b": {"a": 2}}, json.dumps(parsed)) check("a label key id is printed on the layer line in explicit-key mode", key_id_note(["vke_synthetic_2026-05"], False) == "; declares key id vke_synthetic_2026-05", repr(key_id_note(["vke_synthetic_2026-05"], False))) check("the keyring already binds the label, so the note is not repeated there", key_id_note(["vke_synthetic_2026-05"], True) == "", repr(key_id_note(["vke_synthetic_2026-05"], True))) # --- a declared event_hash is recomputed -------------------------------- # The event's own hash was already computed to link the next event and # never compared to what the event DECLARED, so an event could carry an # arbitrary lie about itself inside its correctly signed body. hash_events = [ {"sequence": 0, "canonicalization_version": LEGACY_CANON, "payload": {"n": 1}, "event_signature": "a", "metadata_hash": sha256_hex(canonical_json({"n": 1}, LEGACY_CANON)), "prev_event_hash": CHAIN_HEAD_SENTINEL}, ] honest = sha256_hex(canonical_json( {k: v for k, v in hash_events[0].items() if k not in ("event_hash", "event_signature")}, LEGACY_CANON)) truthful = [dict(hash_events[0], event_hash=honest)] lying = [dict(hash_events[0], event_hash="de" + "ad" * 31 + "be")] # Signature verification is not the property under test here, so these run # through verify_chain with a key that cannot verify; the event_hash # comparison happens before the signature is reached only when it fails, # so the honest case is asserted on the hash value directly. _, lying_diagnostics, _, _ = verify_chain(lying, "00" * 32, None) check("an event declaring a false event_hash is rejected by name", any("event_hash mismatch" in line and "dead" in line for line in lying_diagnostics), lying_diagnostics[-1] if lying_diagnostics else "no diagnostics") _, truthful_diagnostics, _, _ = verify_chain(truthful, "00" * 32, None) check("an event declaring a TRUE event_hash passes the hash check and fails only on signature", not any("event_hash mismatch" in line for line in truthful_diagnostics), truthful_diagnostics[-1] if truthful_diagnostics else "no diagnostics") check("an event that declares no event_hash is still accepted", not any("event_hash" in line for line in verify_chain(hash_events, "00" * 32, None)[1]), "optional, as the schema has it") # --- unsigned wrapper siblings ----------------------------------------- _, siblings = unwrap_session_export( {"veilengine_session_export": {"chain": {}}, "session_id": "sess_other_source", "audit_opinion": "UNQUALIFIED PASS"}) check("unsigned fields beside the signed export are detected and named", siblings == "audit_opinion, session_id", str(siblings)) _, siblings = unwrap_session_export({"veilengine_session_export": {"chain": {}}}) check("a clean wrapper reports no siblings", siblings is None, str(siblings)) # --- keyring validity windows are ENFORCED ----------------------------- # Shape-checked and then never compared to anything: an active key whose # window read 2099-01-01..2099-01-02 verified a 2026 export, and a # valid_from of "not-a-date" passed the shape check. keyring_rejects("keyring valid_from that is not an RFC 3339 timestamp rejected", keyring_of(descriptor(public_hex, valid_from="not-a-date")), "must be an RFC 3339 timestamp") keyring_rejects("keyring valid_until that is not an RFC 3339 timestamp rejected", keyring_of(descriptor(public_hex, valid_until="also-not-a-date")), "must be an RFC 3339 timestamp") keyring_rejects("keyring validity window running backwards rejected", keyring_of(descriptor(public_hex, valid_from="2026-05-01T00:00:00Z", valid_until="2026-01-01T00:00:00Z")), "validity window is reversed") window_ring = parse_public_keyring(keyring_of(descriptor( public_hex, key_id="vke_window", valid_from="2026-05-01T00:00:00Z", valid_until="2026-06-01T00:00:00Z"))) inside = parse_rfc3339("2026-05-18T14:35:00.000Z", "test") before = parse_rfc3339("2026-04-30T23:59:59Z", "test") after = parse_rfc3339("2026-06-01T00:00:01Z", "test") resolved_hex, _, error = resolve_key( {"signing_key_id": "vke_window"}, None, window_ring, "envelope", at=inside) check("a key resolves inside its validity window", resolved_hex == public_hex and error is None, error or "resolved") _, _, error = resolve_key( {"signing_key_id": "vke_window"}, None, window_ring, "envelope", at=before) check("an artifact PREDATING the key's window is rejected by name", bool(error) and "PREDATES" in error and "vke_window" in error, error or "accepted anyway") _, _, error = resolve_key( {"signing_key_id": "vke_window"}, None, window_ring, "envelope", at=after) check("an artifact POSTDATING the key's window is rejected by name", bool(error) and "POSTDATES" in error and "vke_window" in error, error or "accepted anyway") _, _, error = resolve_key( {"signing_key_id": "vke_window"}, None, window_ring, "envelope", at=None) check("with no timestamp the window is not compared, and the layer line says so", error is None and "NOT compared" in validity_note(None, True), validity_note(None, True)) check("an open-ended window (valid_until null) has no upper bound", resolve_key({"signing_key_id": "vke_open"}, None, parse_public_keyring(keyring_of(descriptor( public_hex, key_id="vke_open", valid_until=None))), "envelope", at=after)[2] is None, "accepted") for good in ("2026-05-18T14:35:00Z", "2026-05-18T14:35:00.000Z", "2026-05-18T14:35:00.123456789Z", "2026-05-18T09:35:00-05:00"): try: parse_rfc3339(good, "test") check(f"RFC 3339 timestamp {good!r} parsed", True, "parsed") except ValueError as exc: # pragma: no cover check(f"RFC 3339 timestamp {good!r} parsed", False, str(exc)) for bad in ("2026-05-18", "not-a-date", "2026-05-18T14:35:00", 20260518): try: parse_rfc3339(bad, "test") check(f"non-RFC-3339 value {bad!r} rejected", False, "ACCEPTED") except ValueError as exc: check(f"non-RFC-3339 value {bad!r} rejected", "RFC 3339" in str(exc), str(exc)) failed = [name for name, passed in checks if not passed] print() print(f"Self-test: {len(checks) - len(failed)}/{len(checks)} checks passed") for name in failed: print(f" FAILED: {name}") return 1 if failed else 0 # --- CLI ------------------------------------------------------------------ def make_output_encodable() -> None: """Let this tool NAME any path it rejects, on any console. The manifest-path grammar refuses non-NFC spellings, so the verifier has to be able to print a path carrying a combining mark. On a Windows console running a legacy code page, stdout is cp1252 and printing that diagnostic raised UnicodeEncodeError -- the verifier crashed while reporting the very defect it had correctly detected, and the documented contract is that this tool fails by JUDGING, never by raising. Output is UTF-8 here, and any byte an inherited encoding still cannot carry is backslash-escaped rather than fatal: a mangled character in a diagnostic is a cosmetic loss, an exception while reporting a rejection is a verdict lost. """ for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8", errors="backslashreplace") except (AttributeError, ValueError, OSError): # pragma: no cover - host-dependent continue def main() -> int: make_output_encodable() parser = argparse.ArgumentParser( description="Verify a VeilEngine receipt, manifest, event chain, or session export.", ) parser.add_argument("--receipt", help="Path to a receipt JSON file") parser.add_argument("--manifest", help="Path to a package manifest JSON file") parser.add_argument("--chain", help="Path to a chain or session-export JSON file") parser.add_argument("--session-export", dest="session_export", help="Path to a session export JSON file") parser.add_argument("--public-key-file", help="Path to a JSON file containing 'public_key_hex'") parser.add_argument("--public-key-hex", help="Public key as 64 hex characters") parser.add_argument("--public-keyring-file", dest="public_keyring_file", help="Path to an externally pinned public-only keyring JSON file") parser.add_argument("--expect-key-id", dest="expect_key_id", help="Pin the key id every signed layer must declare; any layer that " "declares a different id, or none, is rejected by name") parser.add_argument("--self-test", action="store_true", help="Run the offline self-test and exit") parser.add_argument("--verbose", "-v", action="store_true", help="Per-artifact detail") args = parser.parse_args() # --self-test used to silently SUPPRESS any other requested mode and return # 0: `--self-test --receipt receipt.json` ran the self-test, exited 0, and # never looked at the receipt. The two are mutually exclusive # now. This is argparse's own rejection path -- parser.error() prints usage # and exits 2 -- rather than add_mutually_exclusive_group, because a group # would also forbid combining the artifact modes with EACH OTHER, which # main() supports and which nothing here has any reason to withdraw. artifact_modes = [ flag for flag, value in ( ("--receipt", args.receipt), ("--manifest", args.manifest), ("--chain", args.chain), ("--session-export", args.session_export), ) if value ] if args.self_test and artifact_modes: parser.error( f"--self-test cannot be combined with {', '.join(artifact_modes)}: the self-test " f"checks the VERIFIER, not an artifact, and running both would report one verdict " f"for two different questions. Run them as separate commands." ) print("== VeilEngine Receipt Verifier ==") if args.self_test: print("Self-test: synthetic payloads, both canonicalizations, no network") return self_test() if not artifact_modes: sys.stderr.write( "ERROR: supply at least one of --receipt, --manifest, --chain, --session-export, --self-test\n" ) return 2 try: public_key_hex, keyring = load_key_material(args) except InputError as exc: sys.stderr.write(f"[ENV] {exc}\n") return 2 except Exception as exc: sys.stderr.write(f"ERROR: {exc}\n") return 2 if keyring is not None: print(f"Keyring: {len(keyring)} externally pinned public keys") if args.verbose: for key_id, descriptor in keyring.items(): print(f" {key_id}: {descriptor['status']}, fingerprint {descriptor['fingerprint']}") elif args.verbose and public_key_hex: print(f"Public key: {public_key_hex[:16]}... (fingerprint {key_fingerprint(public_key_hex)})") overall = True keyring_mode = keyring is not None expect_key_id = args.expect_key_id try: # Every mode asserts the SHAPE of its own input before verifying it, and # a shape fault is an environment fault (exit 2), not a verification # failure. See receipt_root_problem / manifest_root_problem / # session_export_root_problem for what each mode requires and why. if args.receipt: print(f"Loading receipt: {args.receipt}") receipt = load_json_document(args.receipt, "--receipt") problem = receipt_root_problem(receipt) if problem: raise InputError(f"--receipt {problem}") receipt_at, timestamp_error = artifact_timestamp( receipt, (("request", "timestamp_utc"), ("timestamp_utc",)), "receipt" ) if timestamp_error: raise InputError(f"--receipt {timestamp_error}") ok, message, key_id = verify_receipt_obj( receipt, public_key_hex, keyring, expect_key_id=expect_key_id, at=receipt_at ) # IDENTITY IS TYPED IN EVERY MODE. A single receipt has no export to # bind to, so nothing is compared here -- but an identity it declares # must still be a non-empty string at every path that carries it. A # receipt whose session_id is a list verified clean in this mode while # the same receipt was refused inside an export; the rule is about the # object, not about the mode that reads it. identity_message = None if ok: _session, _tenant, identity_error = receipt_identity(receipt, "receipt") if identity_error: ok = False identity_message = identity_error elif args.verbose and (_session or _tenant): identity_message = ( f"declares session {_session!r} and tenant {_tenant!r} " "(single-receipt mode: nothing to bind them to)" ) print(f"Receipt: {'VERIFIED' if ok else 'FAILED'}") suffix = (key_id_note([key_id], keyring_mode) + validity_note(receipt_at, keyring_mode)) if ok else "" print(f" {message}{suffix}") if identity_message: print(f" identity: {identity_message}") overall = overall and ok if args.manifest: print(f"Loading manifest: {args.manifest}") problem = manifest_root_problem(load_json_document(args.manifest, "--manifest")) if problem: raise InputError(f"--manifest {problem}") ok, message = verify_manifest_file( Path(args.manifest).resolve(), public_key_hex, keyring, args.verbose, expect_key_id ) print(f"Manifest: {'VERIFIED' if ok else 'FAILED'}") print(f" {message}{validity_note(None, keyring_mode) if ok else ''}") overall = overall and ok if args.chain: print(f"Loading chain: {args.chain}") events = extract_events(load_json_document(args.chain, "--chain")) if events is None: raise InputError( "--chain expects a JSON array of events, an object with an 'events' array, " "or a session export with chain.events" ) # SIGNATURE TYPE, AT THE ROOT. In --chain mode # the events ARE the input, so a wrong-typed event_signature is a # shape fault and exits 2; verify_chain still reports it as a FAIL # when the same events arrive inside an export. for index, event in enumerate(events): if isinstance(event, dict) and "event_signature" in event: problem = signature_problem(event["event_signature"], f"--chain event[{index}] event_signature") if problem: raise InputError(problem) ok, diagnostics, chain_key_ids, chain_stamps = verify_chain( events, public_key_hex, keyring, expect_key_id ) for line in diagnostics: print(f" {line}") compared = next((s for _, s in chain_stamps if s is not None), None) note = (key_id_note(chain_key_ids, keyring_mode) + validity_note(compared, keyring_mode)) if ok else "" print(f"Chain: {'VERIFIED' if ok else 'FAILED'} ({len(events)} events){note}") overall = overall and ok if args.session_export: print(f"Loading session export: {args.session_export}") ok, results = verify_session_export( Path(args.session_export).resolve(), public_key_hex, keyring, args.verbose, expect_key_id, ) for component, status, message in results: print(f" [{status}] {component:<9} {message}") print(f"Session export: {'VERIFIED' if ok else 'FAILED'}") overall = overall and ok except InputError as exc: sys.stderr.write(f"[ENV] {exc}\n") return 2 except DependencyError as exc: sys.stderr.write(f"[ENV] {exc}\n") return 2 except Exception as exc: # noqa: BLE001 - deliberate boundary, see below # THE DOCUMENTED CONTRACT, ENFORCED AT THE BOUNDARY. # README.md promises "0 PASS | 1 FAIL | 2 argument or # environment error", and exit 1 has one meaning: this artifact was # evaluated and did not verify. Anything that surfaces as an unexpected # exception did NOT evaluate the artifact, so reporting it as 1 -- or # as a Python traceback, which is what actually happened for a truthy # non-string signature and for a broken cryptography backend -- tells # the reader their evidence failed when the tool never looked at it. # Each such fault is named and leaves by the exit-2 door. Layers that # DID evaluate and reject still return their own verdicts above; this # is the last resort, not the first. sys.stderr.write(f"[INPUT] unexpected {type(exc).__name__} while verifying: {exc}\n") return 2 print() print(f"Overall: {'PASS' if overall else 'FAIL'}") return 0 if overall else 1 def guarded_main() -> int: """The process boundary. Nothing leaves this tool as a traceback. The handler inside main() covers the verification dispatch. This one covers everything else -- argument handling, key-material loading, the self-test -- so the contract in README.md holds for the PROCESS, not just for the part of it someone remembered to wrap. SystemExit and KeyboardInterrupt are BaseException and pass through untouched, so argparse's own --help and exit codes are unaffected. """ try: return main() except InputError as exc: sys.stderr.write(f"[ENV] {exc}\n") return 2 except DependencyError as exc: sys.stderr.write(f"[ENV] {exc}\n") return 2 except Exception as exc: # noqa: BLE001 - deliberate process boundary sys.stderr.write(f"[INPUT] unexpected {type(exc).__name__}: {exc}\n") return 2 if __name__ == "__main__": sys.exit(guarded_main())