#!/usr/bin/env python3 """Verify a Lenz warranty certificate. Offline. Without Lenz. python verify_certificate.py certificate.json --keys lenz-certificate-keys.json \ [--tsa-roots qtsp-roots.pem] A Lenz certificate says: this exact statement got this exact verdict, this analysis stood behind it, and it existed at this time. This script checks all three without contacting Lenz, because a warranty you can only verify by asking the warrantor is not much of a warranty. WHAT IT CHECKS 1. **The leaf.** Recomputes SHA-256 over `record_version || 0x00 || RFC8785(payload)` and compares it to the leaf on the document. This is what catches an edited claim, a changed verdict, a removed warning, a swapped source. RFC 8785 (JSON Canonicalization Scheme) is what makes it reproducible in any language; every string is NFC-normalized first, so the same sentence typed two ways gives the same leaf. 2. **The signature.** ECDSA P-256 over SHA-256 of the leaf, against the public key published in `lenz-certificate-keys.json`. The 32-byte digest of the leaf's ASCII hex is what gets signed, as is: hash the leaf once, never twice. Fetch that file once from lenz.io and keep it: it is append-only, so an old copy can only ever be missing a newer key. 3. **The qualified timestamp.** An RFC 3161 token from a QTSP on the EU Trusted List, over `SHA-256(leaf || 0x00 || signature)`. This is the anchor the warranty runs on: the contract requires the timestamp to PRECEDE the publication. **Pass `--tsa-roots` with roots you trust.** The chain stored in the document keeps the token verifiable after those certificates expire, but it is evidence, not a trust anchor — whoever hands you the document also wrote the chain. Without `--tsa-roots` this reports UNVERIFIED. 4. **OpenTimestamps**, structurally. Its Bitcoin attestation needs a Bitcoin node to check properly; this script confirms the receipt is well-formed and covers the same digest, then tells you to run `ots verify` for the rest. WHAT IT DOES NOT CHECK Whether the verdict is correct. That is what the warranty is for. Whether the verdict was later WITHDRAWN. `withdrawn_at` sits outside the signed payload by design — withdrawal must not invalidate a record someone may already be entitled to claim on — which also means it can be deleted from a downloaded file with no trace. Check the live certificate page for the current status. EXIT CODES 0 OK every required check passed 1 FAILED the document does not match what Lenz signed 2 usage error 3 INCONCLUSIVE a required check could not run (e.g. no --keys). NOT a pass. DEPENDENCIES pip install cryptography rfc8785 rfc3161-client opentimestamps-client `cryptography` and `rfc8785` are enough for checks 1 and 2, which are the ones that catch tampering. `rfc3161-client` adds check 3, `opentimestamps-client` adds check 4. Anything missing is reported as SKIPPED, never as a pass. """ from __future__ import annotations import argparse import base64 import hashlib import json import sys import unicodedata EXIT_OK = 0 EXIT_FAILED = 1 EXIT_USAGE = 2 # A required check did not run. Distinct from both OK and FAILED: nothing is # wrong with the document, and nothing has been established about it either. EXIT_INCONCLUSIVE = 3 _VERSION_SEPARATOR = b'\x00' # --- Canonicalization -------------------------------------------------------- # The v1 payload's key set is CLOSED. An unknown key is not a harmless # addition — every consumer reads this document by name, so a key nobody # validates is a place to hide a second value. PAYLOAD_FIELDS_V1 = frozenset( { 'record_version', 'atomic_claim', 'verdict', 'warranted_pole', 'lenz_score', 'confidence', 'executive_summary', 'warnings', 'sources', 'analysis_completed_at', 'issued_at', 'as_of', 'terms_version', 'currency', 'cap', 'aggregate', 'gate_version', } ) SOURCE_FIELDS_V1 = frozenset({'url', 'title', 'snippet'}) # Which currency each terms version is written in. Frozen and append-only, and # it must MIRROR lenz/coverage/leaf.py's table exactly — this file's whole job # is to reproduce that module's decisions without trusting Lenz, so a currency # it would reject must be rejected here too. TERMS_CURRENCY = { 'v1': 'EUR', } class PayloadShapeError(ValueError): """Not a v1 payload. Do not hash it, do not report on it.""" def _normalize(value): """NFC-normalize every string VALUE, recursively. Keys are left alone. Keys are deliberately NOT normalized, and this must match the issuer exactly. NFC plus strip() is not injective, so normalizing keys would let `" atomic_claim "` and `"atomic_claim"` collapse into one — last one wins — and two documents saying opposite things would share a leaf, a signature and a timestamp. """ if isinstance(value, str): return unicodedata.normalize('NFC', value).strip() if isinstance(value, dict): return {k: _normalize(v) for k, v in value.items()} if isinstance(value, list): return [_normalize(v) for v in value] return value def require_v1_shape(payload: dict) -> None: """Reject a payload whose keys are not exactly the v1 set.""" if not isinstance(payload, dict) or set(payload) != PAYLOAD_FIELDS_V1: keys = set(payload) if isinstance(payload, dict) else set() raise PayloadShapeError( f'not a v1 payload: unexpected={sorted(keys - PAYLOAD_FIELDS_V1)} ' f'missing={sorted(PAYLOAD_FIELDS_V1 - keys)}' ) # The stated currency must match the terms version the document cites. # Without this a document could say `{"currency": "USD", # "terms_version": "v1"}` over euro-denominated terms and still verify — # the signature and the timestamp are over the bytes, and they cannot see # that the document contradicts the contract it names. expected = TERMS_CURRENCY.get(payload.get('terms_version')) if expected is None: raise PayloadShapeError(f'unknown terms_version: {payload.get("terms_version")!r}') if payload.get('currency') != expected: raise PayloadShapeError( f'terms {payload.get("terms_version")} is written in {expected}, not {payload.get("currency")!r}' ) for field in ('cap', 'aggregate'): value = payload.get(field) # bool is an int in Python, and True would canonicalize as `true`. if not isinstance(value, int) or isinstance(value, bool) or value < 0: raise PayloadShapeError(f'{field} must be a non-negative integer in major units, got {value!r}') sources = payload.get('sources') if not isinstance(sources, list): raise PayloadShapeError('sources must be a list') for index, source in enumerate(sources): if not isinstance(source, dict) or set(source) != SOURCE_FIELDS_V1: raise PayloadShapeError(f'source {index} is not exactly {sorted(SOURCE_FIELDS_V1)}') def compute_leaf(payload: dict, record_version: str) -> str: import rfc8785 require_v1_shape(payload) digest = hashlib.sha256() digest.update(record_version.encode('utf-8')) digest.update(_VERSION_SEPARATOR) digest.update(rfc8785.dumps(_normalize(payload))) return digest.hexdigest() def anchor_digest(leaf: str, signature: str) -> bytes: digest = hashlib.sha256() digest.update(leaf.encode('ascii')) digest.update(_VERSION_SEPARATOR) digest.update(signature.encode('ascii')) return digest.digest() # --- Checks ------------------------------------------------------------------ # The checks that must PASS before this script says anything reassuring. The # leaf alone proves only that the document is self-consistent — anyone can # write a self-consistent JSON file. The signature is what ties it to Lenz. _REQUIRED_CHECKS = ('leaf', 'signature') class Result: def __init__(self): self.lines: list[tuple[str, str]] = [] self.failed = False self.passed: set[str] = set() def ok(self, label: str, detail: str = '') -> None: self.passed.add(label) self.lines.append(('PASS', f'{label}{f" — {detail}" if detail else ""}')) def bad(self, label: str, detail: str = '') -> None: self.failed = True self.lines.append(('FAIL', f'{label}{f" — {detail}" if detail else ""}')) def skip(self, label: str, detail: str) -> None: self.lines.append(('SKIP', f'{label} — {detail}')) def note(self, text: str) -> None: self.lines.append(('', text)) @property def inconclusive(self) -> list[str]: """Required checks that did not run. Not a pass, and not a failure.""" return [check for check in _REQUIRED_CHECKS if check not in self.passed] @property def verdict(self) -> str: if self.failed: return 'FAILED' return 'INCONCLUSIVE' if self.inconclusive else 'OK' def check_leaf(document: dict, result: Result) -> None: try: recomputed = compute_leaf(document['payload'], document['record_version']) except ImportError: result.skip('leaf', 'pip install rfc8785 to check this — it is the important one') return except PayloadShapeError as exc: result.bad('leaf', str(exc)) return except (KeyError, TypeError) as exc: result.bad('leaf', f'malformed document: {exc}') return if recomputed == document.get('leaf'): result.ok('leaf', 'the claim, verdict, warnings and sources are as issued') else: result.bad('leaf', f'expected {document.get("leaf")}, recomputed {recomputed}') def check_signature(document: dict, keys: dict | None, result: Result) -> None: signature = document.get('signature') if not signature: result.skip('signature', 'this certificate has not been signed yet') return if keys is None: result.skip('signature', 'pass --keys lenz-certificate-keys.json to check this') return try: from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec, utils except ImportError: result.skip('signature', 'pip install cryptography to check this') return entry = next((k for k in keys.get('keys', []) if k.get('key_id') == document.get('key_id')), None) if entry is None: result.bad('signature', f'no published key with id {document.get("key_id")!r}') return try: public_key = serialization.load_pem_public_key(entry['public_key_pem'].encode('ascii')) # Lenz signs SHA-256(leaf) as a digest: Cloud KMS signs the digest it is # sent and does not hash it again. `Prehashed` says exactly that — passing # the digest to plain ECDSA(SHA256()) would hash it a second time. public_key.verify( base64.b64decode(signature), hashlib.sha256(document['leaf'].encode('ascii')).digest(), ec.ECDSA(utils.Prehashed(hashes.SHA256())), ) except Exception as exc: # noqa: BLE001 — any failure is a failed signature result.bad('signature', str(exc) or 'does not verify against the published key') return result.ok('signature', f'signed by {entry["key_id"]}') def check_timestamp(document: dict, result: Result, roots_pem: bytes | None) -> None: """The qualified timestamp — and only against roots YOU chose to trust. The certificate carries the TSA's own chain so the token stays verifiable after those certificates expire. That chain is evidence, not a trust anchor: whoever hands you the document also wrote the chain, so validating the token against a root taken from position -1 of that same field proves nothing. Anyone can mint a CA, issue themselves a timestamping cert, and sign a token with any `genTime` they like. So: pass `--tsa-roots` with the QTSP roots you trust (Lenz publishes which authority signs, and its roots are on the EU Trusted List). Without it this reports UNVERIFIED rather than PASS. The leaf certificate is selected by extended key usage, not by position: the CMS `certificates` field is a SET OF with no chain ordering, so "certificates[0] is the leaf" is an assumption the DER encoding does not make. """ rfc3161 = (document.get('anchors') or {}).get('rfc3161') or {} token = rfc3161.get('token') if not token: result.skip('qualified timestamp', 'not anchored yet — do not publish behind this certificate') return try: from rfc3161_client import VerifierBuilder, decode_timestamp_response except ImportError: result.skip('qualified timestamp', 'pip install rfc3161-client to check this') return chain = rfc3161.get('certificate_chain') if not chain: result.bad('qualified timestamp', 'no certificate chain stored; cannot be verified') return if not roots_pem: result.skip( 'qualified timestamp', 'no --tsa-roots given. The chain inside the document is not a trust anchor, so the timestamp is UNVERIFIED', ) return try: from cryptography import x509 from cryptography.x509.oid import ExtendedKeyUsageOID embedded = x509.load_pem_x509_certificates(chain.encode('ascii')) roots = x509.load_pem_x509_certificates(roots_pem) leaf = _timestamping_leaf(embedded, ExtendedKeyUsageOID.TIME_STAMPING) if leaf is None: result.bad('qualified timestamp', 'no timestamping certificate in the stored chain') return builder = VerifierBuilder().tsa_certificate(leaf) for intermediate in embedded: if intermediate is not leaf: builder = builder.add_intermediate_certificate(intermediate) for root in roots: builder = builder.add_root_certificate(root) response = decode_timestamp_response(base64.b64decode(token)) digest = anchor_digest(document['leaf'], document['signature']) builder.build().verify(response, digest) gen_time = response.tst_info.gen_time except Exception as exc: # noqa: BLE001 — any failure is a failed timestamp result.bad('qualified timestamp', str(exc) or 'does not verify') return # Report the time the TOKEN asserts, never the `timestamped_at` field — that # one is unsigned envelope data the document's author controls. result.ok('qualified timestamp', f'{gen_time.isoformat()} (from the token)') def _timestamping_leaf(certificates, timestamping_oid): """The cert with the timeStamping EKU. Position in the chain means nothing.""" from cryptography import x509 for certificate in certificates: try: usages = certificate.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value except x509.ExtensionNotFound: continue if timestamping_oid in usages: return certificate return None def check_opentimestamps(document: dict, result: Result) -> None: """Does the receipt actually attest to THIS document? The Bitcoin attestation is the one anchor that needs no authority to still exist in ten years, and it is also the one a reader cannot eyeball. So the binding gets checked here: a receipt is only evidence about this certificate if the digest inside it equals ``anchor_digest(leaf, signature)``. Without that comparison a proof for an entirely different document — or 3 KB of well-formed random base64 — reads as "present and well-formed", which is what this check said for its first version. What is still NOT checked, and says so: whether the attestation is in the Bitcoin chain. That needs a node. Run ``ots verify`` for it. """ ots = (document.get('anchors') or {}).get('opentimestamps') or {} proof = ots.get('proof') if not proof: result.skip('opentimestamps', 'no receipt on this certificate') return try: raw = base64.b64decode(proof, validate=True) except Exception: # noqa: BLE001 result.bad('opentimestamps', 'receipt is not readable') return try: from opentimestamps.core.serialize import BytesDeserializationContext from opentimestamps.core.timestamp import DetachedTimestampFile except ImportError: result.skip('opentimestamps', 'pip install opentimestamps-client to check the receipt binds to this document') return try: detached = DetachedTimestampFile.deserialize(BytesDeserializationContext(raw)) stamped = detached.timestamp.msg except Exception as exc: # noqa: BLE001 result.bad('opentimestamps', f'receipt does not parse: {exc}') return expected = anchor_digest(document['leaf'], document['signature']) if stamped != expected: result.bad( 'opentimestamps', 'the receipt timestamps a different digest — it is not evidence about this certificate', ) return result.skip( 'opentimestamps', 'receipt covers this record; run `ots verify` with a Bitcoin node for the chain attestation', ) def check_withdrawal(document: dict, result: Result) -> None: withdrawn = document.get('withdrawn_at') if withdrawn: result.note( f'NOTE: Lenz withdrew this verdict on {withdrawn}. Publications made BEFORE that ' 'notice stay covered; later ones do not.' ) # --- CLI --------------------------------------------------------------------- def verify(document: dict, keys: dict | None, roots_pem: bytes | None = None) -> Result: result = Result() check_leaf(document, result) check_signature(document, keys, result) check_timestamp(document, result, roots_pem) check_opentimestamps(document, result) check_withdrawal(document, result) return result def _load(path: str) -> dict: with open(path, encoding='utf-8') as handle: return json.load(handle) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('certificate', help='the certificate JSON, downloaded from its page') parser.add_argument( '--keys', help='lenz-certificate-keys.json, fetched once from lenz.io. Without it the signature ' 'cannot be checked and the result is INCONCLUSIVE.', ) parser.add_argument( '--tsa-roots', help='PEM file of trusted timestamping-authority ROOT certificates. Without it the ' 'qualified timestamp is reported as unverified: the chain inside the document proves ' 'nothing on its own, because whoever wrote the document also wrote the chain.', ) args = parser.parse_args(argv) try: document = _load(args.certificate) keys = _load(args.keys) if args.keys else None roots_pem = open(args.tsa_roots, 'rb').read() if args.tsa_roots else None except (OSError, ValueError) as exc: print(f'could not read input: {exc}', file=sys.stderr) return EXIT_USAGE result = verify(document, keys, roots_pem) payload = document.get('payload') or {} print(f'certificate {document.get("certificate_id", "?")}') print(f' claim: {payload.get("atomic_claim", "?")}') print(f' verdict: {payload.get("verdict", "?")} (warranted as of {payload.get("as_of", "?")})') print( f' cover: {payload.get("currency", "?")} {payload.get("cap", "?")} per certificate, ' f'{payload.get("currency", "?")} {payload.get("aggregate", "?")} aggregate, ' f'terms {payload.get("terms_version", "?")}' ) print() for status, line in result.lines: print(f' {status:4} {line}' if status else f' {line}') print() if result.failed: print('FAILED — this document does not match what Lenz signed.') return EXIT_FAILED if result.inconclusive: # The old behaviour printed OK and exited 0 here, which meant running # the documented command without --keys reported a fabricated # certificate as fine. A check that did not run is not a check that # passed. missing = ', '.join(result.inconclusive) print(f'INCONCLUSIVE — {missing} could not be checked. This document has NOT been verified.') return EXIT_INCONCLUSIVE print('OK — the record matches what Lenz signed.') return EXIT_OK if __name__ == '__main__': raise SystemExit(main())