#!/usr/bin/env python3 """Verify a ProvenClosed attestation. Protocol section 7. python verify.py certificate.html [--keys PATH_OR_URL] [--expect NAME] [--issuer URL] **This file imports nothing from ProvenClosed and talks to nothing of ProvenClosed's except the key set**, which can be pinned to a local file. That is the whole test section 9 describes: if verification needs the product, the certificate is a report with extra steps. **It trusts one issuer: https://provenclosed.com.** The key set comes from there, or from the file `--keys` names -- never from the issuer a document names about itself. A verifier that asked the document whom to trust would verify any forgery that arrived with its own key set: shown on 2026-09-26 with a document naming `https://pr0venclosed.com`, which the previous version of this file called VALID. A document naming any other issuer is refused, unless the reader says with `--issuer` that they mean to trust that one instead. **A production certificate is also in a public transparency log.** Its proof travels beside the signed document: the log's signed timestamp, an inclusion proof and the log's signed checkpoint, all checked here against log keys pinned in this file -- copied from Sigstore's published trusted_root.json, so a reader can compare them with Sigstore's own publication without trusting ours. Nothing about the certificate is in the log except a fingerprint of it. It is deliberately one file, standard library plus `cryptography`, and plain ASCII so that it reads the same in every editor. Every dependency here is something a reviewer must install and trust, and the format is small enough not to ask that twice. **Exit codes**: 0 verified, production; 3 the signature is valid and the issuer trusted, but the document is not production-verified -- an alpha certificate, a production one whose proof is missing, or a proof from Sigstore's staging log; 1 refused; 2 could not be read. A refusal is not an error -- it is the answer -- but a script that cannot tell them apart will treat a forged document as an outage. """ from __future__ import annotations import argparse import base64 import hashlib import json import re import sys import urllib.request from datetime import UTC, datetime from pathlib import Path from typing import Any, NamedTuple try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa except ImportError: # pragma: no cover - the one dependency, named plainly print("this verifier needs `cryptography`: pip install cryptography", file=sys.stderr) raise SystemExit(2) from None PAYLOAD_TYPE = "application/vnd.in-toto+json" STATEMENT_TYPE = "https://in-toto.io/Statement/v1" PREDICATE_TYPE = "urn:vexora:attestation:exposure-evidence:v0.1" #: A bundle: one signed document for every finding matching a stated rule. #: `docs/attestation-bundle-design.md`. BUNDLE_PREDICATE_TYPE = "urn:vexora:attestation:exposure-set:v0.1" KEY_SET_PATH = "/.well-known/vexora-attestation-keys.json" #: The issuer this verifier trusts. `--issuer` replaces it, and the output says so. TRUST_ANCHOR = "https://provenclosed.com" #: A signed document that commits to having been put in a transparency log. PROFILE_PRODUCTION = "v0.1-production" #: The proof beside a production certificate. Versioned, because the log that #: writes it will change: Rekor v1 today, v2 when Sigstore sends writes there. TRANSPARENCY_FORMAT = "provenclosed-transparency/v1" #: The transparency logs whose proofs this verifier can check, keyed by log ID -- #: the SHA-256 of the log's public key, which is recomputed before any use. #: Copied on 2026-09-26 from Sigstore's trusted_root.json: production from #: github.com/sigstore/root-signing, staging from sigstore/root-signing-staging. #: A proof from any other log is refused. TRUSTED_LOGS: dict[str, dict[str, Any]] = { "c0d23d6ad406973f9559f3ba2d1ca01f84147d8ffc5b8445c224f98b9591801d": { "name": "Sigstore Rekor", "url": "https://rekor.sigstore.dev", "production": True, "pem": ( "-----BEGIN PUBLIC KEY-----\n" "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE2G2Y+2tabdTV5BcGiBIx0a9fAFwr\n" "kBbmLSGtks4L3qX6yYY0zufBnhC8Ur/iy55GhWP/9A/bY2LhC30M9+RYtw==\n" "-----END PUBLIC KEY-----\n" ), }, "d32f30a3c32d639c2b762205a21c7bb07788e68283a4ae6f42118723a1bea496": { "name": "Sigstore Rekor STAGING", "url": "https://rekor.sigstage.dev", "production": False, "pem": ( "-----BEGIN PUBLIC KEY-----\n" "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEDODRU688UYGuy54mNUlaEBiQdTE9\n" "nYLr0lg6RXowI/QV/RE1azBn4Eg5/2uTOMbhB1/gfcHzijzFi9Tk+g1Prg==\n" "-----END PUBLIC KEY-----\n" ), }, } #: How much earlier than the log's own time a document may say it was issued. #: A certificate is signed and logged within seconds; one the log first saw this #: much later was signed long before anybody could check it -- which is what #: backdating looks like, and the one thing a log exists to expose. BACKDATE_TOLERANCE_SECONDS = 15 * 60 #: How much later than the log's time: two clocks never agree exactly. CLOCK_SKEW_SECONDS = 5 * 60 class Refused(Exception): """The document does not verify, and the message says why.""" # -- the format --------------------------------------------------------------- def pae(payload: bytes, payload_type: str) -> bytes: t = payload_type.encode("utf-8") return b" ".join([b"DSSEv1", str(len(t)).encode(), t, str(len(payload)).encode(), payload]) def _int(raw: str) -> int: padded = raw + "=" * (-len(raw) % 4) return int.from_bytes(base64.urlsafe_b64decode(padded), "big") def public_key_of(jwk: dict[str, Any]) -> rsa.RSAPublicKey: return rsa.RSAPublicNumbers(_int(jwk["e"]), _int(jwk["n"])).public_key() # -- whom to trust --------------------------------------------------------------- class Trust(NamedTuple): """Whom this run trusts. Decided by the reader -- never by the document. A NamedTuple, not a dataclass: this file is loaded by path, without being registered as a module, and a dataclass under postponed annotations looks itself up in `sys.modules` and fails there. """ anchor: str = TRUST_ANCHOR #: True when the reader replaced the anchor with `--issuer`. chosen: bool = False def _url(value: Any) -> str: return str(value or "").strip().rstrip("/").lower() def check_issuer(issuer: dict[str, Any], trust: Trust) -> tuple[str, str, list[str]]: """TRUSTED or PROVISIONAL -- or refused. Returns what keeps it from production. A provisional issuer was named before the issuer host was settled, so there is nothing to compare with the anchor; what is trusted there is the key, which had to be in the key set the reader chose. """ named = str(issuer.get("url") or "?") if issuer.get("provisional"): return ( "PROVISIONAL", f"{named} -- named before the issuer was settled; do not pin it", ["the issuer was provisional when this was signed"], ) if _url(named) != _url(trust.anchor): raise Refused( f"this document names {named} as its issuer, and this verifier trusts " f"{trust.anchor}. A document may name any issuer it likes, which is why the " "name is checked against the anchor rather than believed. To trust a " "different issuer on purpose, pass --issuer" ) return ("TRUSTED", f"{named} (named with --issuer)" if trust.chosen else named, []) # -- the transparency log -------------------------------------------------------- class Logged(NamedTuple): state: str detail: str #: Why this is not production-verified. Empty when it is. reasons: tuple[str, ...] = () #: When the log recorded it, when there is a proof. Trusted, unlike any date #: the document states about itself. at: datetime | None = None def _leaf_hash(data: bytes) -> bytes: return hashlib.sha256(b"\x00" + data).digest() def _node_hash(left: bytes, right: bytes) -> bytes: return hashlib.sha256(b"\x01" + left + right).digest() def root_from_inclusion_proof(index: int, size: int, leaf: bytes, proof: list[bytes]) -> bytes: """RFC 9162 section 2.1.3.2: the root an audit path leads to.""" if not 0 <= index < size: raise Refused("the inclusion proof names a position outside the tree") fn, sn, root = index, size - 1, leaf for sibling in proof: if sn == 0: raise Refused("the inclusion proof is longer than the tree allows") if fn & 1 or fn == sn: root = _node_hash(sibling, root) if not fn & 1: while True: fn >>= 1 sn >>= 1 if fn & 1 or fn == 0: break else: root = _node_hash(root, sibling) fn >>= 1 sn >>= 1 if sn != 0: raise Refused("the inclusion proof is shorter than the tree requires") return root def check_checkpoint(checkpoint: str, log_id: str, key: Any, size: int, root: bytes) -> None: """The log's signed note for the tree the proof leads to. Format: origin, tree size, base64 root hash, a blank line, then signature lines of the form ` `, the key hint being the first four bytes of the log ID. The signature covers every line above the blank one, each ending in a newline. """ head, blank, signatures = checkpoint.partition("\n\n") lines = head.split("\n") if not blank or len(lines) < 3: raise Refused("the log's checkpoint is not in the signed-note format") try: stated_size = int(lines[1]) stated_root = base64.b64decode(lines[2], validate=True) except ValueError: raise Refused("the log's checkpoint is not in the signed-note format") from None if stated_size != size or stated_root != root: raise Refused("the log's checkpoint is for a different tree than the inclusion proof") hint = bytes.fromhex(log_id)[:4] for signed in signatures.strip().split("\n"): parts = signed.split(" ") if len(parts) != 3: continue try: raw = base64.b64decode(parts[2], validate=True) except ValueError: continue if raw[:4] != hint: continue try: key.verify(raw[4:], (head + "\n").encode("utf-8"), ec.ECDSA(hashes.SHA256())) except InvalidSignature: raise Refused("the log's signature on its checkpoint does not verify") from None return raise Refused("the checkpoint carries no signature by the log that answered") def _when(moment: str | None) -> datetime | None: if not moment: return None try: then = datetime.fromisoformat(moment) except ValueError: return None return then if then.tzinfo else then.replace(tzinfo=UTC) def check_transparency( proof: dict[str, Any] | None, *, profile: str, signed: bytes, signature: bytes, jwk: dict[str, Any], issued_at: str | None, ) -> Logged: """Is this very document -- these bytes, this signature, this key -- in a log? `signed` is the PAE: the bytes the DSSE signature covers, whose SHA-256 is the only thing about the certificate that ever reaches the log. """ if not proof: why = ( "this document says it was logged, and no transparency proof came with it" if profile == PROFILE_PRODUCTION else f"a {profile} certificate carries no transparency proof" ) return Logged("NOT PRESENT", f"({profile})", (why,)) if proof.get("format") != TRANSPARENCY_FORMAT: raise Refused(f"unknown transparency proof format {proof.get('format')!r}") log_id = str((proof.get("log") or {}).get("id") or "") known = TRUSTED_LOGS.get(log_id) if known is None: raise Refused( f"the transparency proof is from a log this verifier does not know " f"({log_id[:16] or '?'}...)" ) log_key = serialization.load_pem_public_key(known["pem"].encode("ascii")) der = log_key.public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo ) if hashlib.sha256(der).hexdigest() != log_id: raise Refused( "a log key pinned in this file does not match its own ID -- the file was edited" ) entry = proof.get("entry") or {} inclusion = proof.get("inclusionProof") or {} try: body_b64 = str(entry["body"]) body = base64.b64decode(body_b64, validate=True) spec = json.loads(body)["spec"] digest = spec["data"]["hash"] entry_signature = base64.b64decode(spec["signature"]["content"], validate=True) entry_key = serialization.load_pem_public_key( base64.b64decode(spec["signature"]["publicKey"]["content"], validate=True) ) logged_at = datetime.fromtimestamp(int(entry["integratedTime"]), UTC) index, tree_size = int(inclusion["logIndex"]), int(inclusion["treeSize"]) path = [bytes.fromhex(h) for h in inclusion["hashes"]] stated_root = bytes.fromhex(inclusion["rootHash"]) except (KeyError, TypeError, ValueError): raise Refused("the transparency proof is not readable") from None # 1. The entry is about this document, this signature and this key. fingerprint = hashlib.sha256(signed).hexdigest() if digest.get("algorithm") != "sha256" or digest.get("value") != fingerprint: raise Refused("the log entry is about a different document") if entry_signature != signature: raise Refused("the log entry carries a different signature") if not isinstance(entry_key, rsa.RSAPublicKey) or ( entry_key.public_numbers() != public_key_of(jwk).public_numbers() ): raise Refused("the log entry was made with a different key") # 2. The log promised to include it, at this time. promise = json.dumps( { "body": body_b64, "integratedTime": entry["integratedTime"], "logID": log_id, "logIndex": entry["logIndex"], }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") try: log_key.verify( base64.b64decode(str(entry.get("signedEntryTimestamp", "")), validate=True), promise, ec.ECDSA(hashes.SHA256()), ) except (InvalidSignature, ValueError): raise Refused("the log's signed timestamp does not verify") from None # 3. And it did: the audit path leads to a root the log signed. root = root_from_inclusion_proof(index, tree_size, _leaf_hash(body), path) if root != stated_root: raise Refused("the inclusion proof does not lead to the root it states") check_checkpoint(str(inclusion.get("checkpoint") or ""), log_id, log_key, tree_size, root) # 4. Not backdated: the document's own date against the log's. issued = _when(issued_at) if issued is None: raise Refused("a logged document has to say when it was issued, and this one does not") late = (issued - logged_at).total_seconds() early = (logged_at - issued).total_seconds() if late > CLOCK_SKEW_SECONDS: raise Refused( f"the document says it was issued at {issued_at}, after the log recorded it " f"at {logged_at.isoformat()}" ) if early > BACKDATE_TOLERANCE_SECONDS: raise Refused( f"the document says it was issued at {issued_at}, and the log first saw it at " f"{logged_at.isoformat()} -- {int(early // 60)} minutes later. A certificate is " "logged as it is signed; one this much older than its log entry was backdated" ) reasons: list[str] = [] if not known["production"]: reasons.append(f"the proof is from {known['name']}, a test log") if profile != PROFILE_PRODUCTION: reasons.append(f"the document is {profile}, which does not commit to being logged") return Logged( "VALID", f"{known['name']} entry {entry['logIndex']}, logged {logged_at:%Y-%m-%d %H:%M} UTC", tuple(reasons), logged_at, ) def attestation_verdict( out: list[str], *, issuer: tuple[str, str, list[str]], logged: Logged, ) -> int: """The four lines the owner specified, in this order, and the exit code. Printed directly under the signature, before any detail, so that nothing below can be read as stronger than they are. """ state, detail, not_production = issuer out.append(line("Issuer", state, detail)) out.append(line("Transparency", logged.state, logged.detail)) reasons = [*not_production, *logged.reasons] if reasons: out.append(line("Attestation", "NOT PRODUCTION-VERIFIED")) out.extend(f" - {why}" for why in reasons) return 3 out.append(line("Attestation", "VERIFIED", "production")) return 0 # -- the steps, in section 7.1's order ----------------------------------------------- def load_key_set(where: str) -> dict[str, Any]: """From a file or a URL. A pinned local copy is the offline case.""" if where.startswith(("http://", "https://")): with urllib.request.urlopen(where, timeout=15) as response: # noqa: S310 return json.loads(response.read()) return json.loads(Path(where).read_text(encoding="utf-8")) def find_key(key_set: dict[str, Any], kid: str) -> dict[str, Any]: for jwk in key_set.get("keys", []): if jwk.get("kid") == kid: return jwk raise Refused( f"the key set has no key {kid!r}. Either this document was not signed by " "this issuer, or the issuer dropped a key -- which the attestation key set " "must never do" ) def check_key_window(jwk: dict[str, Any], issued_at: str | None) -> str: """CURRENT, or RETIRED and only if it signed while it was still valid. **The key set dates the retirement, not the document.** A stolen retired key could sign anything and claim any `issued_at`, so reading the date from the certificate would be asking the forger when the forgery happened. This check is what retention buys and it is *not* the whole answer -- only a transparency log proves a document existed when it says it did (section 5). """ retired = jwk.get("retired_at") if not retired: return "CURRENT" if not issued_at: raise Refused("signed by a retired key, and the document carries no issue date") if issued_at >= retired: raise Refused( f"signed by a key retired at {retired}, but the document claims to have " f"been issued at {issued_at} -- after the key stopped being valid" ) return f"RETIRED {retired[:10]}" def _dots(label: str) -> str: return (label + " ").ljust(18, ".")[:18] def line(label: str, verdict: str, detail: str = "") -> str: return f"{_dots(label)} {verdict:<11} {detail}".rstrip() def _short(digest: str | None) -> str: return f"{digest[:12]}..." if digest else "none" def _age(moment: str | None) -> str: if not moment: return "never" try: then = datetime.fromisoformat(moment) except ValueError: return moment if then.tzinfo is None: then = then.replace(tzinfo=UTC) days = (datetime.now(UTC) - then).days return f"{days} day(s) ago" if days else "today" def _rule(value: Any, unrestricted: str) -> str: """Render one clause of the selection for a person. The field is a list, and an empty one means no restriction -- so the words are made here rather than stored in the document, where a shape-shifting field would mislead everything that parses it. A plain string is still accepted: documents issued before 2026-09-22 carry the prose inline, and a verifier that stopped reading the certificates it has already handed out would be worse than the defect it fixed. """ if isinstance(value, str): return value if not value: return unrestricted return ", ".join(str(item) for item in value) def report_bundle( doc: dict[str, Any], payload: bytes, signature: bytes, kid: str, key_set: dict[str, Any], from_a_page: bool = False, trust: Trust | None = None, proof: dict[str, Any] | None = None, ) -> int: """One document, many findings -- and the rule that chose them, printed first. **The order here is the specification, not a layout preference.** `docs/attestation-bundle-design.md` section 6: a verifier that prints *"18 of 25 verified"* before the rule that chose the 25 has taught the reader the wrong thing in its first line, and nothing further down undoes it. So: signature, issuer, **selection**, then counts, then every finding, then what was left out. The arithmetic never appears before the rule. """ predicate = doc["predicate"] jwk = find_key(key_set, kid) key_status = check_key_window(jwk, (predicate.get("selection") or {}).get("as_of")) try: public_key_of(jwk).verify( signature, pae(payload, PAYLOAD_TYPE), padding.PKCS1v15(), hashes.SHA256() ) except Exception: raise Refused("the signature does not match this key -- the document was altered") from None trust = trust or Trust() issued_at = (predicate.get("selection") or {}).get("as_of") issuer = check_issuer(predicate.get("issuer") or {}, trust) logged = check_transparency( proof, profile=str(predicate.get("profile", "?")), signed=pae(payload, PAYLOAD_TYPE), signature=signature, jwk=jwk, issued_at=issued_at, ) if logged.at is not None: # The log's time, not the document's, decides a retired key's window. key_status = check_key_window(jwk, logged.at.isoformat()) out = ["", "ProvenClosed ATTESTATION -- SET OF FINDINGS", ""] if from_a_page: out += [PROSE_IS_NOT_SIGNED, ""] out.append(line("Signature", "VALID", f"DSSE, RS256, kid {kid[:12]}...")) code = attestation_verdict(out, issuer=issuer, logged=logged) out.append(line("Key status", key_status)) # **Before any number.** A reader who has seen the count first will read the # rule as a caveat on it; a reader who has seen the rule first reads the # count as a consequence of something they have already judged. selection = predicate.get("selection") or {} out += ["", "WHAT WAS SELECTED -- read this before the counts"] out.append(f" Scopes : {_rule(selection.get('scopes'), 'every scope')}") out.append(f" Severities : {_rule(selection.get('severities'), 'every severity')}") out.append(f" Status : {selection.get('status')}") out.append(f" As of : {selection.get('as_of')} ({_age(selection.get('as_of'))})") counts = predicate.get("counts") or {} out += ["", "RESULT"] # The summary already carries its own coverage caveat in one sentence -- # printing the numbers without it would undo the reason it is one sentence. out.append(predicate.get("summary", "?")) out.append( f" carried {counts.get('carried', '?')}" f" | observed absent {counts.get('observed_absent', '?')}" f" | still open {counts.get('still_open', '?')}" ) # **The split a reader acts on.** Eight VERIFIED and two # VERIFIED_WITH_LIMITATIONS is a different document from ten of the second, # and the line above cannot tell them apart. if "verified" in counts: out.append( f" VERIFIED {counts.get('verified', 0)}" f" | VERIFIED_WITH_LIMITATIONS {counts.get('verified_with_limitations', 0)}" f" | OBSERVED_ONLY {counts.get('observed_only', 0)}" ) # Whether every observation can be traced to the run that wrote it, at the # top rather than left for somebody to work out from the set. integrity = predicate.get("evidence_integrity") or {} if integrity: bound = integrity.get("fully_run_bound") out.append( line( " Run binding", "COMPLETE" if bound is True else "PARTIAL" if bound is False else "CANNOT SAY", f"{integrity.get('unbound_observations', '?')} of " f"{integrity.get('observations', '?')} observations are not bound " "to the run that produced them", ) ) coverage = predicate.get("coverage") or [] if coverage: out += ["", "WHAT THE SCANS BEHIND THIS COVERED"] for row in coverage: verdict = ( "COMPLETE" if row.get("complete") is True else "PARTIAL" if row.get("complete") is False else "CANNOT SAY" ) out.append(line(f" {row.get('scan_type', '?')}", verdict, row.get("note", ""))) findings = predicate.get("findings") or [] if findings: out += ["", "EVERY FINDING IN THIS SET"] for entry in findings: exposure = entry.get("exposure") or {} subject = entry.get("subject") or {} strict = entry.get("verification_state") or {} # The strict state where the document carries one; the kind where it # does not, so a bundle issued before this existed still prints. mark = strict.get("state") or ( "VERIFIED" if entry.get("kind") == "remediation_verified" else "OBSERVED_ONLY" ) out.append(f" [{mark:>26}] {subject.get('name', '?')} -- {exposure.get('title', '?')}") # **Underneath the line it weakens, not in a footnote.** A reason # kept for the end is one the reader meets after they have already # formed a view of the row it applies to. for reason in strict.get("reasons") or []: out.append(f" - {reason}") limits = predicate.get("limits") or [] if limits: out += ["", "WHAT THIS DOCUMENT COULD NOT ESTABLISH"] out += [f" - {text}" for text in limits] out += [ "", f"This proves what {_who(trust)} observed and that this document is unaltered.", "It does not prove the estate is secure, and it says nothing about any", "moment after the date above.", "", ] print("\n".join(out)) return code def _who(trust: Trust) -> str: return "the issuer named above" if trust.chosen else "ProvenClosed" def report( envelope: dict[str, Any], key_set: dict[str, Any], expect: str | None, trust: Trust | None = None, ) -> int: trust = trust or Trust() # Popped before anything else reads the envelope, so neither marker can # reach a signature check or be mistaken for part of the document. from_a_page = bool(envelope.pop(FROM_A_PAGE, False)) proof = envelope.pop(BESIDE, None) # 1. Parse, and refuse anything but the declared types. Fail closed: a # verifier that shrugs at an unknown predicate will one day verify a # document whose fields mean something else entirely. if envelope.get("payloadType") != PAYLOAD_TYPE: raise Refused(f"unexpected payloadType {envelope.get('payloadType')!r}") try: signature = base64.b64decode(envelope["signatures"][0]["sig"]) kid = envelope["signatures"][0]["keyid"] payload = base64.b64decode(envelope["payload"]) except (KeyError, IndexError, ValueError, TypeError) as bad: raise Refused(f"this is not a DSSE envelope ({bad})") from None # **Guarded, because this runs before the signature check.** # The payload has to be parsed to know which kind of document it is, so a # corrupted one reaches `json.loads` while still unverified. Unguarded, that # was a `JSONDecodeError` traceback and exit 1 -- a verifier handing a # stack trace to an auditor who asked a yes-or-no question. try: doc = json.loads(payload) except ValueError as bad: raise Refused( f"the signed payload is not readable ({bad}) -- this document was altered" ) from None if not isinstance(doc, dict): raise Refused("the signed payload is not an object -- this document was altered") if doc.get("_type") != STATEMENT_TYPE: raise Refused(f"not an in-toto statement ({doc.get('_type')!r})") if doc.get("predicateType") == BUNDLE_PREDICATE_TYPE: return report_bundle(doc, payload, signature, kid, key_set, from_a_page, trust, proof) if doc.get("predicateType") != PREDICATE_TYPE: raise Refused(f"unknown predicateType {doc.get('predicateType')!r}") predicate = doc["predicate"] # 2 + 3. The key, its window, and the signature over the PAE. jwk = find_key(key_set, kid) key_status = check_key_window(jwk, predicate.get("issued_at")) try: public_key_of(jwk).verify( signature, pae(payload, PAYLOAD_TYPE), padding.PKCS1v15(), hashes.SHA256() ) except Exception: raise Refused("the signature does not match this key -- the document was altered") from None issuer = check_issuer(predicate.get("issuer") or {}, trust) logged = check_transparency( proof, profile=str(predicate.get("profile", "?")), signed=pae(payload, PAYLOAD_TYPE), signature=signature, jwk=jwk, issued_at=predicate.get("issued_at"), ) if logged.at is not None: # The log's time, not the document's, decides a retired key's window. key_status = check_key_window(jwk, logged.at.isoformat()) out = ["", "ProvenClosed ATTESTATION", ""] if from_a_page: out += [PROSE_IS_NOT_SIGNED, ""] out.append(line("Signature", "VALID", f"DSSE, RS256, kid {kid[:12]}...")) code = attestation_verdict(out, issuer=issuer, logged=logged) out.append(line("Key status", key_status)) # **Pinned and reproducible are two claims, and only one of them is VALID.** # A digest names the containers as the issuing machine knew them; pulling # them needs the images to have been published. section 2.2(b) -- "the check is # re-runnable" -- is the second of the three things that reduce the need to # trust the issuer, so a verifier that prints VALID for an unpublished # toolchain is doing the issuer a favour at the reader's expense. chain = (predicate.get("observed") or {}).get("toolchain") or {} published = (predicate.get("reproducible") or {}).get("images_published") digest = _short(chain.get("manifest_digest")) if not chain.get("pinned"): out.append( line("Evidence", "QUALIFIED", f"no pinned toolchain -- {chain.get('recorded_as')!r}") ) elif published is True: out.append(line("Evidence", "VALID", f"toolchain {digest}, images published")) elif published is False: out.append(line("Evidence", "QUALIFIED", f"toolchain {digest} -- images never published")) else: out.append(line("Evidence", "CANNOT SAY", f"toolchain {digest} -- publication unknown")) # 4. The verifier asks; it does not assume. subject = (doc.get("subject") or [{}])[0] name = subject.get("name", "?") if expect is None: out.append(line("Resource binding", "NOT CHECKED", f"{name} -- pass --expect to confirm")) elif expect.strip().lower().rstrip(".") == str(name).strip().lower().rstrip("."): out.append(line("Resource binding", "VALID", name)) else: raise Refused(f"this certificate is about {name!r}, not {expect!r}") verification = predicate.get("verification") or {} absent_at = verification.get("absent_at") kind = predicate.get("kind") if kind == "remediation_verified" and absent_at: out.append(line("Verification", "VALID", f"outcome=absent at {absent_at}")) else: out.append(line("Verification", "NOT CLAIMED", "nothing has observed this to be absent")) # 4b. **What is known about the check, not about the host.** # # A reader weighing `Verification: VALID` above is entitled to ask how the # check that said `absent` is known to be able to say it. Three verdicts, # and the missing block is its own verdict rather than silence: documents # issued before this field existed must not read as documents whose check # was never demonstrated. # # `verifies` is **deliberately not read here.** It is a declaration written # by the same deployment that signed this, and a verifier that counted it # would be quoting the issuer back at the reader as though it were evidence. cls = predicate.get("class_validation") if cls is None: out.append(line("Check proven", "NOT STATED", "this document does not say")) else: shown = cls.get("verification_demonstrated") or {} where = [k for k in ("testbed", "production") if shown.get(k)] if where: out.append(line("Check proven", "YES", "verification seen in " + " and ".join(where))) else: out.append( line( "Check proven", "NO", "this check has never been seen to report a weakness as gone", ) ) # 5. The tri-state, never collapsed. coverage = predicate.get("coverage") or {} complete = coverage.get("complete") note = coverage.get("note") or "no coverage was stated" verdict = {True: "COMPLETE", False: "PARTIAL"}.get(complete, "CANNOT SAY") out.append(line("Coverage", verdict, note)) # 5b. **The other half of the exposure, and its three states.** # # Same discipline as coverage and `class_validation`: absent, "we were not # allowed to look", and an answer are three different documents and this # verifier never lets two of them print the same line. The basis is printed # verbatim rather than summarised, because the whole argument for this block # is that the reason is a call somebody granted and not a coincidence. same = predicate.get("correlation") if same is None: out.append(line("Same machine", "NOT STATED", "this document does not say")) elif not same.get("machines"): if same.get("asked_for"): out.append( line( "Same machine", "NOT OBSERVED", "the account was read and did not place this resource on an instance", ) ) else: out.append( line( "Same machine", "NOT ASKED FOR", "this account has not granted the call that would answer it", ) ) else: machine = same["machines"][0] more = len(same["machines"]) - 1 where = f"{machine.get('instance_id', '?')} in {machine.get('region', '?')}" if more: where += f" (+{more} more)" out.append(line("Same machine", "YES", where)) out.append(line(" basis", "", same.get("basis") or "?")) out.append(line(" as of", "", machine.get("observed_at") or "not stated")) if not same.get("asked_for"): out.append(line(" freshness", "STALE", "the account is no longer being read")) others = same.get("also_on_this_machine") or [] for other in others: out.append( line( " also here", other.get("status", "?").upper(), f"{other.get('check_id', '?')} on {other.get('resource', '?')}", ) ) if same.get("truncated"): out.append(line(" ", "", "...and more not listed")) # 5c. **The identity half, and its own three states.** # # A separate grant, so a separate answer. A document whose roles were never # asked for and whose machines were reads as a complete picture with nothing # in it, and the only thing that stops that is printing the difference. identity = (same or {}).get("roles") if isinstance(same, dict) else None if identity is None: if same is not None: out.append(line("Runs as", "NOT STATED", "this document does not say")) elif not identity.get("identities"): if identity.get("asked_for"): out.append( line( "Runs as", "NOT OBSERVED", "the roles were read and none was tied to this", ) ) else: out.append( line( "Runs as", "NOT ASKED FOR", "this account has not granted the calls that read its roles", ) ) else: out.append(line("Runs as", "YES", identity.get("basis") or "?")) for one in identity["identities"]: out.append(line(" role", "", one.get("role", "?"))) out.append(line(" how", "", one.get("reached_by", "?"))) if one.get("used_by"): out.append(line(" used by", "", ", ".join(one["used_by"]))) # **Unreadable is printed before the list, not instead of it.** # An empty `may_be_assumed_by` under a policy nobody could parse is # the safe-sounding lie this whole block is written against. if not one.get("trust_readable", True): out.append( line( " trust", "UNREADABLE", "who may assume this role is not stated", ) ) for who in one.get("may_be_assumed_by") or (): out.append( line( " assumed by", "CONDITIONAL" if who.get("conditional") else "", who.get("role", "?"), ) ) for who in one.get("may_assume") or (): out.append(line(" may assume", "", who)) out.append(line(" as of", "", one.get("observed_at") or "not stated")) if identity.get("truncated"): out.append(line(" ", "", "...and more roles not listed")) # 5d. **The control mapping, and the sentences that stop it reading as an # assessment.** # # This is the one block in the document where the danger is not a wrong # value but a right value read wrongly. A verifier that printed # # Controls ......... EVIDENCE FOR soc2 CC6.1, iso27001 A.5.17 # # and stopped there would produce the strongest possible version of the # thing the mapping is written not to be: a machine-checked line, directly # under a signature this tool has just called VALID, that a reader files as # *"CC6.1 verified"*. So the notes print with it, always, and are not behind # a flag, a width check or a verbosity level. # # Three states, like everything else here. A document with no block is an # older issuer; `evidence_for: null` is an issuer that does not have this # check; `[]` is a check mapped to nothing on purpose. controls = predicate.get("controls") if controls is None: out.append(line("Controls", "NOT STATED", "this document does not say")) else: named = controls.get("evidence_for") if named is None: out.append( line( "Controls", "NOT STATED", "the issuer does not recognise the check behind this finding", ) ) elif not named: out.append(line("Controls", "NONE MAPPED", "no control is mapped to this check")) else: out.append( line( "Controls", "EVIDENCE FOR", ", ".join( f"{one.get('framework', '?')} {one.get('identifier', '?')}" for one in named ), ) ) out += [f" {text}" for text in controls.get("notes") or ()] # 5e. **Which run produced the observation that closed this.** # # Four states and never fewer (v2.2 section 1.4). A row that predates the binding # and a run that has aged out of retention are different facts about the # evidence, and the one thing a verifier must not do is print them alike. # # `toolchain_agrees: false` is the loud one: the row's toolchain and its # run's toolchain disagree, which means the document is contradicting its # own database. That is worth more than a clean line, so it is printed as a # mismatch rather than folded into the id. history = predicate.get("history") or [] closing = next( (h for h in reversed(history) if h.get("outcome") == "absent"), history[-1] if history else None, ) if closing is not None: run = closing.get("run") if run is None: out.append(line("Scan run", "NOT BOUND", "this observation predates run binding")) elif run.get("resolved") is None: out.append(line("Scan run", "NOT CHECKED", f"{_short(run.get('id'))} not looked up")) elif run.get("resolved") is False: out.append( line( "Scan run", "AGED OUT", f"{_short(run.get('id'))} -- named, and removed under retention", ) ) else: out.append( line( "Scan run", "NAMED", f"{_short(run.get('id'))} {run.get('type', '?')}/{run.get('mode', '?')}", ) ) agrees = run.get("toolchain_agrees") if agrees is False: out.append( line( " toolchain", "MISMATCH", "this row's toolchain is not the run's -- the document " "disagrees with its own database", ) ) elif agrees is None: out.append(line(" toolchain", "NOT COMPARABLE", "one side records none")) # 5f. **Form and access, on separate lines, because they are separate # facts.** `registry_digest` says the toolchain digest is a registry # repo digest rather than a machine-local image id; `images_published` # says a stranger can fetch those images. A push to a private registry # makes the first true and the second false, and a verifier that read # one as the other would report a reproducibility this document does # not claim -- v2.2 section 0. able = predicate.get("reproducible") or {} if "registry_digest" in able: out.append( line( "Digest kind", "REGISTRY" if able.get("registry_digest") else "LOCAL", "the same string in any deployment that can reach it" if able.get("registry_digest") else "a machine-local image id; two builds of the same source differ", ) ) pullable = able.get("images_published") out.append( line( "Anyone can pull", {True: "YES", False: "NO"}.get(pullable, "NOT MEASURED"), "a stranger fetched these images" if pullable is True else "the registry refuses anonymous access" if pullable is False else "nobody has asked the registry", ) ) since = predicate.get("since") or {} out += ["", "RESULT"] strict = predicate.get("verification_state") or {} if strict: out.append(f"STATE: {strict.get('state')}") for reason in strict.get("reasons") or []: out.append(f" - {reason}") if kind == "remediation_verified" and absent_at: out.append(f"EXPOSURE VERIFIED ABSENT at {absent_at}") else: exposure = predicate.get("exposure") or {} out.append(f"EXPOSURE OBSERVED: {exposure.get('title', '?')}") # 6. Always the age, because a certificate nobody has re-checked for five # weeks should not read as current. regressions = since.get("regression_count", 0) watched = since.get("watched_until") out.append( f"Watched since; {'no regression' if not regressions else f'{regressions} regression(s)'}" f" as of {_age(watched)}." ) limits = predicate.get("limits") or [] if limits: out += ["", "WHAT THIS DOCUMENT COULD NOT ESTABLISH"] out += [f" - {text}" for text in limits] # 8. Part of the specification, not decoration. out += [ "", f"This proves what {_who(trust)} observed and that this document is unaltered.", "It does not prove the resource is secure.", "", ] print("\n".join(out)) return code #: The element a certificate page keeps the transparency proof in -- beside the #: signed envelope, never inside it, because the log answers after signing. PROOF_EMBEDDED = re.compile( r']*type="application/vnd\.provenclosed\.transparency\+json"[^>]*>(.*?)', re.DOTALL | re.IGNORECASE, ) #: Where `load_document` hands the proof to `report`, popped before anything #: else reads the envelope. BESIDE = "transparency beside the envelope" #: The element a ProvenClosed certificate page keeps its signed envelope in. EMBEDDED = re.compile( r']*type="application/vexora\+dsse"[^>]*>(.*?)', re.DOTALL | re.IGNORECASE, ) #: Set when the envelope was lifted out of a certificate page rather than read #: as JSON. **It changes what the reader must be told**, so it is printed, not #: merely known: see `PROSE_IS_NOT_SIGNED`. FROM_A_PAGE = "embedded in a page" PROSE_IS_NOT_SIGNED = ( "The signature covers the document this page carries, NOT the text printed " "around it. Everything below is re-rendered here from the signed bytes -- " "read it rather than the page, because the page's own wording could have " "been edited without breaking anything above." ) def load_document(text: str) -> dict[str, Any]: """A DSSE envelope, whether it arrives as JSON or as the certificate page. **Added 2026-09-22, the day the HTML certificate shipped, because without it the certificate was not verifiable.** The page carries the signed envelope inside it, which is the whole argument for one self-contained file -- and that argument is only true if the tool an auditor runs can read that file. It could not: this script called `json.loads` on whatever it was given and exited 2 on a page, so a customer who forwarded the readable artefact had forwarded something the verifier refused. The embedded text is ordinary JSON -- the renderer writes `<` as the JSON escape `<` so that no ` int: # **This script runs on somebody else's machine, and that is the point.** # # A Windows console defaults to cp1252, which cannot encode most of what a # careful document says -- the first bundle printed here died on a `U+21B3` # with a `UnicodeEncodeError` and no report at all. A verifier that crashes # on the reader's code page has failed at the one job it has. # # `errors="replace"` rather than a raise: a character that cannot be shown # should cost the reader that character, never the verdict. for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError): # pragma: no cover - older streams pass parser = argparse.ArgumentParser(description="Verify a ProvenClosed attestation.") parser.add_argument("attestation", help="the signed document") parser.add_argument( "--keys", help="the attestation key set: a local pinned file, or a URL. Defaults to " f"{TRUST_ANCHOR}{KEY_SET_PATH} -- the anchor's, never the document's", ) parser.add_argument( "--issuer", help=f"trust this issuer instead of {TRUST_ANCHOR}. Say it only on purpose: " "a document naming another issuer is refused without it", ) parser.add_argument("--expect", help="the resource this certificate should be about") args = parser.parse_args() try: envelope = load_document(Path(args.attestation).read_text(encoding="utf-8")) except OSError as bad: print(f"could not read {args.attestation}: {bad}", file=sys.stderr) return 2 except Refused as why: print(f"could not read {args.attestation}: {why}", file=sys.stderr) return 2 trust = Trust(anchor=args.issuer.rstrip("/"), chosen=True) if args.issuer else Trust() # **From the anchor, never from the document.** This used to read the # issuer out of the unverified payload and fetch that issuer's key set -- # so a forgery that named its own host arrived with its own keys. where = args.keys or trust.anchor + KEY_SET_PATH try: key_set = load_key_set(where) except Exception as bad: print(f"could not read the key set at {where}: {bad}", file=sys.stderr) return 2 try: return report(envelope, key_set, args.expect, trust) except Refused as why: print(f"\nProvenClosed ATTESTATION\n\nREFUSED {why}\n", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())