#!/usr/bin/env python3 """ SSP (Sovereign Sign-off Protocol) offline verifier for a Sovereign Record. Verifies a decision.signed Sovereign Record (record.ssp.json) with NO server: 1. claim_canonical_hash == SHA-256(JCS(claim)) (canonical integrity) 2. User signature - WebAuthn assertion over claim_canonical_hash, checked against the COSE public key embedded in the record (ES256 or EdDSA). Non-custodial; air-gapped. 3. Trust signature - Ed25519 over (claim hash + user signature), i.e. the trust key wraps the user signature. 4. (optional) Artifact - sha256(artifact) == claim.presentation.artifact_hash 4b.(optional) Context - sha256(JCS(context)) == claim.presentation.context_hash 5. (optional) Tlog - sha256(record file) == tlog data hash (online) 6. (optional) Rekor - SET (signed inclusion promise) + RFC 6962 inclusion proof + checkpoint (signed tree head), all offline from the stored anchors.json 7. (optional) RFC 3161 - timestamp token messageImprint == claim hash, CMS signature valid, signer chains to the TSA CA The trust signature collapses per-user-key trust to one well-known key, so an offline verifier only needs the record + the Sovereign trust public key. The Rekor SET and RFC 3161 token are likewise verified with their issuers' public keys — no network call to the log or the TSA. The Sovereign trust public key is always supplied by the verifier (out-of-band). It is never read from the bundle — otherwise a forged bundle could ship a matching key and "self-certify". A .sspb therefore carries only evidence (record, anchors, context, timestamp, TSA CA), and the trust key is a separate argument. The other trust anchors — the TSA CA (--tsa-cert) and the transparency-log public key (--rekor-pubkey) — can likewise be supplied out-of-band and are then used in preference to the copies carried in the bundle/anchors. If you don't supply them, the bundled copies are used and the output says so (the anchor is checked, but against a key that travelled with the evidence rather than one you trust independently). Usage: # Verify a .sspb bundle directly (no unzipping) with your trusted copy of the key: python3 verify-ssp.py bundle.sspb trust-public.pem # ...anchoring time/inclusion to keys you trust, supplied as files: python3 verify-ssp.py bundle.sspb trust-public.pem --tsa-cert ca.pem --rekor-pubkey rekor.pem # ...or pulled live from the anchor URLs (logs URL + fingerprint, flags rotation/revocation): python3 verify-ssp.py bundle.sspb trust-public.pem --pull-keys # Verify a .sspa attestation bundle (zip: record + anchors + timestamp + TSA CA, # the same as a .sspb without the PII context) — record + tlog + tsa: python3 verify-ssp.py attestation.sspa trust-public.pem --pull-keys # Or point at individual files: python3 verify-ssp.py [--artifact FILE] [--context-file FILE] [--rp-id app.local] [--ledger-uuid UUID] [--tlog-server URL] (online tlog) [--anchors-file anchors.json] [--rekor-pubkey rekor.pem] (offline Rekor) [--tsa-token timestamp.tsr --tsa-cert ca.crt] (offline TSA) Dependencies: pip3 install cryptography asn1crypto (CBOR is decoded inline; asn1crypto is only needed for the RFC 3161 check) """ import sys import os import json import base64 import hashlib import argparse import tempfile import zipfile import urllib.request import urllib.error import ssl # ---- helpers --------------------------------------------------------------- def b64url_decode(s): return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) def b64url_nopad(b): return base64.urlsafe_b64encode(b).decode().rstrip("=") def jcs(obj): """RFC 8785-style canonical JSON: recursively key-sorted, compact, UTF-8. Sufficient for SSP claims (string/integer/object values).""" return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() def cbor_decode(data, i=0): """Minimal CBOR decoder — enough for COSE_Key maps (ints, byte strings).""" b = data[i]; major = b >> 5; minor = b & 0x1F; i += 1 def length(minor, i): if minor < 24: return minor, i if minor == 24: return data[i], i + 1 if minor == 25: return int.from_bytes(data[i:i + 2], "big"), i + 2 if minor == 26: return int.from_bytes(data[i:i + 4], "big"), i + 4 if minor == 27: return int.from_bytes(data[i:i + 8], "big"), i + 8 raise ValueError("unsupported CBOR length") if major == 0: v, i = length(minor, i); return v, i if major == 1: v, i = length(minor, i); return -1 - v, i if major == 2: n, i = length(minor, i); return data[i:i + n], i + n if major == 3: n, i = length(minor, i); return data[i:i + n].decode(), i + n if major == 4: n, i = length(minor, i); arr = [] for _ in range(n): v, i = cbor_decode(data, i); arr.append(v) return arr, i if major == 5: n, i = length(minor, i); d = {} for _ in range(n): k, i = cbor_decode(data, i); v, i = cbor_decode(data, i); d[k] = v return d, i raise ValueError("unsupported CBOR major type {}".format(major)) def verify_webauthn(user, claim_hash_hex, rp_id): """Verify the WebAuthn assertion offline. Returns (ok, detail).""" from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.hazmat.primitives import hashes from cryptography.exceptions import InvalidSignature auth_data = b64url_decode(user["authenticator_data"]) client_data = b64url_decode(user["client_data_json"]) sig = b64url_decode(user["signature"]) cose = cbor_decode(b64url_decode(user["public_key"]["cose"]))[0] cd = json.loads(client_data) if cd.get("type") != "webauthn.get": return False, "clientData type is not webauthn.get" expected_challenge = b64url_nopad(bytes.fromhex(claim_hash_hex)) if cd.get("challenge") != expected_challenge: return False, "challenge != claim_canonical_hash" # rpIdHash (first 32 bytes of authenticatorData) must be sha256(rp_id) if auth_data[:32] != hashlib.sha256(rp_id.encode()).digest(): return False, "rpIdHash does not match rp-id '{}'".format(rp_id) signed = auth_data + hashlib.sha256(client_data).digest() kty = cose.get(1) try: if kty == 2: # EC2 (ES256) x = int.from_bytes(cose[-2], "big") y = int.from_bytes(cose[-3], "big") pub = ec.EllipticCurvePublicNumbers(x, y, ec.SECP256R1()).public_key() pub.verify(sig, signed, ec.ECDSA(hashes.SHA256())) elif kty == 1: # OKP (EdDSA / Ed25519) pub = Ed25519PublicKey.from_public_bytes(cose[-2]) pub.verify(sig, signed) else: return False, "unsupported COSE key type {}".format(kty) except InvalidSignature: return False, "WebAuthn signature INVALID" return True, "kty={} ({})".format(kty, "EC2/ES256" if kty == 2 else "OKP/EdDSA") def rfc6962_root(index, size, leaf, proof_hex): """Recompute the Merkle root from an inclusion proof (RFC 6962 2.1.1).""" if index >= size: return None def node(l, r): return hashlib.sha256(b"\x01" + l + r).digest() fn, sn, r = index, size - 1, leaf for ph in proof_hex: p = bytes.fromhex(ph) if fn == sn or fn % 2 == 1: r = node(p, r) if fn % 2 == 0: while fn != 0 and fn % 2 == 0: fn >>= 1 sn >>= 1 else: r = node(r, p) fn >>= 1 sn >>= 1 return r if sn == 0 else None def verify_rekor(rekor, record_file_bytes, log_pubkey_pem=None): """Verify a stored Rekor proof offline: SET, inclusion path, checkpoint, and that the entry binds to this record. Returns a list of (label, ok) checks. log_pubkey_pem, when given, is the verifier's out-of-band copy of the log's public key and is used in preference to the key embedded in the anchors (so the proof is anchored to a trusted key, not one that travelled with it).""" from cryptography.hazmat.primitives.serialization import load_pem_public_key from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives import hashes from cryptography.exceptions import InvalidSignature out = [] try: pub = load_pem_public_key((log_pubkey_pem or rekor["public_key_pem"]).encode()) except Exception as e: # A malformed or wrong-type log key fails every key-dependent check cleanly. out.append(("SET (signed inclusion promise)", False)) ip = rekor.get("inclusion_proof", {}) leaf = hashlib.sha256(b"\x00" + base64.b64decode(rekor["body"])).digest() root = rfc6962_root(ip.get("logIndex", 0), ip.get("treeSize", 0), leaf, ip.get("hashes", [])) out.append(("Inclusion proof (Merkle path to root)", root is not None and root.hex() == ip.get("rootHash"))) out.append(("Checkpoint (signed tree head)", False)) return out # SET: ECDSA over JCS({body, integratedTime, logID, logIndex}) payload = { "body": rekor["body"], "integratedTime": rekor["integrated_time"], "logID": rekor["log_id"], "logIndex": rekor["log_index"], } try: pub.verify(base64.b64decode(rekor["set"]), jcs(payload), ec.ECDSA(hashes.SHA256())) out.append(("SET (signed inclusion promise)", True)) except Exception: # InvalidSignature, or a non-ECDSA key supplied (TypeError) — either way, fail. out.append(("SET (signed inclusion promise)", False)) # Inclusion proof: leaf = SHA256(0x00 || body) folded to rootHash ip = rekor["inclusion_proof"] leaf = hashlib.sha256(b"\x00" + base64.b64decode(rekor["body"])).digest() root = rfc6962_root(ip["logIndex"], ip["treeSize"], leaf, ip["hashes"]) out.append(("Inclusion proof (Merkle path to root)", root is not None and root.hex() == ip["rootHash"])) # Checkpoint: ECDSA over the note body, root binding to inclusion proof cp_ok = False cp = ip.get("checkpoint", "") if cp: lines = cp.split("\n") sig_idx = next((i for i, l in enumerate(lines) if l.startswith("— ")), -1) if sig_idx >= 4: body = "\n".join(lines[:sig_idx]).encode() try: raw = base64.b64decode(lines[sig_idx].split(" ", 2)[2]) root_match = base64.b64decode(lines[2]).hex() == ip["rootHash"] pub.verify(raw[4:], body, ec.ECDSA(hashes.SHA256())) cp_ok = root_match except Exception: cp_ok = False out.append(("Checkpoint (signed tree head)", cp_ok)) # Binding: the entry's data hash must equal sha256(this record file) bind_ok = False try: rekord = json.loads(base64.b64decode(rekor["body"] + "==")) entry_hash = rekord.get("spec", {}).get("data", {}).get("hash", {}).get("value", "") bind_ok = entry_hash == hashlib.sha256(record_file_bytes).hexdigest() except Exception: pass out.append(("Entry binds to this record", bind_ok)) return out def verify_rfc3161(token_der, ca_pem_bytes, claim_hash_hex): """Verify an RFC 3161 timestamp token offline: messageImprint matches the claim hash, the CMS signature is valid, and the signer chains to the TSA CA. Returns (ok, detail). Requires asn1crypto.""" try: from asn1crypto import tsp, cms, x509 as asn1x509 except ImportError: return None, "asn1crypto not installed (pip3 install asn1crypto)" from cryptography import x509 from cryptography.hazmat.primitives.serialization import Encoding from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa from cryptography.exceptions import InvalidSignature # The CMS digest algorithm is read from the token, not assumed — public TSAs # sign with sha256/sha384/sha512 (FreeTSA uses sha512/ECDSA, our dev TSA sha256/RSA). hash_by_name = {"sha1": hashes.SHA1, "sha256": hashes.SHA256, "sha384": hashes.SHA384, "sha512": hashes.SHA512} hashlib_by_name = {"sha1": hashlib.sha1, "sha256": hashlib.sha256, "sha384": hashlib.sha384, "sha512": hashlib.sha512} resp = tsp.TimeStampResp.load(token_der) signed_data = resp["time_stamp_token"]["content"] tst_info = signed_data["encap_content_info"]["content"].parsed imprint = tst_info["message_imprint"]["hashed_message"].native gen_time = tst_info["gen_time"].native.strftime("%Y-%m-%dT%H:%M:%SZ") if imprint.hex() != claim_hash_hex: return False, "messageImprint != claim hash" signer_info = signed_data["signer_infos"][0] signer_cert = None for cert in signed_data["certificates"]: c = cert.chosen if c.serial_number == signer_info["sid"].chosen["serial_number"].native: signer_cert = x509.load_der_x509_certificate(c.dump()) break if signer_cert is None: return False, "signer cert not in token" digest_name = signer_info["digest_algorithm"]["algorithm"].native hcls = hash_by_name.get(digest_name) hfun = hashlib_by_name.get(digest_name) if hcls is None: return False, "unsupported CMS digest %s" % digest_name # When signed attributes are present, the signature is over their DER (with the # IMPLICIT [0] tag rewritten to SET OF), and the messageDigest attribute must # equal the digest of the eContent (the TSTInfo) — bind the signature to it. signed_attrs = signer_info["signed_attrs"] if signed_attrs.native is None: return False, "timestamp token has no signed attributes" msg_digest = None for attr in signed_attrs: if attr["type"].native == "message_digest": msg_digest = attr["values"][0].native if msg_digest != hfun(tst_info.dump()).digest(): return False, "messageDigest attribute does not match the timestamp content" signed_bytes = b"\x31" + signed_attrs.dump()[1:] # IMPLICIT [0] -> SET OF sig = signer_info["signature"].native pub = signer_cert.public_key() try: if isinstance(pub, rsa.RSAPublicKey): pub.verify(sig, signed_bytes, padding.PKCS1v15(), hcls()) elif isinstance(pub, ec.EllipticCurvePublicKey): pub.verify(sig, signed_bytes, ec.ECDSA(hcls())) else: return False, "unsupported signer key" except InvalidSignature: return False, "CMS signature INVALID" # Chain the signer to the provided TSA CA. try: ca = x509.load_pem_x509_certificate(ca_pem_bytes) except Exception: return False, "TSA CA is not a valid certificate" try: ca_pub = ca.public_key() if isinstance(ca_pub, rsa.RSAPublicKey): ca_pub.verify(signer_cert.signature, signer_cert.tbs_certificate_bytes, padding.PKCS1v15(), signer_cert.signature_hash_algorithm) elif isinstance(ca_pub, ec.EllipticCurvePublicKey): ca_pub.verify(signer_cert.signature, signer_cert.tbs_certificate_bytes, ec.ECDSA(signer_cert.signature_hash_algorithm)) except InvalidSignature: return False, "signer cert does not chain to TSA CA" return True, "timestamp valid, genTime={}".format(gen_time) def http_get_bytes(url): """Fetch a URL (http/https; relaxed TLS for self-signed dev hosts).""" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE req = urllib.request.Request(url, headers={"Accept": "*/*"}) with urllib.request.urlopen(req, context=ctx, timeout=10) as resp: return resp.read() def pem_fingerprint(pem_data): """SHA-256 of the DER inside a PEM block (a key's SPKI or a certificate), so the same key/cert fingerprints identically however it is wrapped.""" if isinstance(pem_data, bytes): pem_data = pem_data.decode("ascii", "ignore") b64 = "".join(l for l in pem_data.splitlines() if l and not l.startswith("-----")) try: der = base64.b64decode(b64) except Exception: return None return hashlib.sha256(der).hexdigest() def fetch_tlog_entry(uuid, tlog_server): url = "{}/api/v1/log/entries/{}".format(tlog_server.rstrip("/"), uuid) ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE try: req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, context=ctx, timeout=10) as resp: return json.loads(resp.read()).get(uuid), None except Exception as e: return None, str(e) # ---- main ------------------------------------------------------------------ def run(record_path, trust_pubkey_path, artifact_path, context_path, rp_id, ledger_uuid, tlog_server, anchors_path, tsa_token_path, tsa_cert_path, rekor_pubkey_path=None, tsa_cert_trusted=True, pull_keys=False): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.hazmat.primitives.serialization import load_pem_public_key from cryptography.exceptions import InvalidSignature ok = True with open(record_path, "rb") as f: record_file_bytes = f.read() record = json.loads(record_file_bytes) trust_pub = None if trust_pubkey_path: with open(trust_pubkey_path) as f: trust_pub = load_pem_public_key(f.read().encode()) claim = record["claim"] declared_hash = record["claim_canonical_hash"]["value"] user = record["signatures"]["user"] trust = record["signatures"]["trust"] print("Record type : SSP {} - {}".format(record.get("ssp_version"), claim.get("type"))) print("Claim id : {}".format(claim.get("claim_id"))) # [1] canonical integrity computed_hash = hashlib.sha256(jcs(claim)).hexdigest() print("\n[1] Canonical claim hash") print(" Declared : {}".format(declared_hash)) print(" Computed : {}".format(computed_hash)) if computed_hash == declared_hash: print(" PASS - claim_canonical_hash matches SHA-256(JCS(claim))") else: print(" FAIL - canonical hash mismatch") ok = False # [2] user WebAuthn signature print("\n[2] User signature (WebAuthn, non-custodial)") try: u_ok, detail = verify_webauthn(user, declared_hash, rp_id) print(" {} - {}".format("PASS" if u_ok else "FAIL", detail)) ok = ok and u_ok except ImportError: print(" SKIP - 'cryptography' not installed (pip3 install cryptography)") ok = False # [3] trust signature wraps the user signature print("\n[3] Trust signature (Ed25519, wraps user signature)") if trust_pub is None: print(" SKIP - no trust public key supplied; the user signature above is") print(" self-contained, but cannot be tied to the Sovereign trust key") elif not isinstance(trust_pub, Ed25519PublicKey): print(" FAIL - provided trust key is not Ed25519") ok = False else: msg = (declared_hash + "." + user["signature"]).encode() try: trust_pub.verify(b64url_decode(trust["value"]), msg) print(" PASS - trust signature valid (covers: {})".format(trust.get("covers"))) except InvalidSignature: print(" FAIL - trust signature INVALID") ok = False # [4] artifact (optional) if artifact_path: print("\n[4] Artifact vs presentation digest") with open(artifact_path, "rb") as f: artifact_hash = hashlib.sha256(f.read()).hexdigest() subject = claim.get("presentation", {}).get("artifact_hash", "") print(" Presentation : sha256:{}".format(subject)) print(" Artifact : sha256:{}".format(artifact_hash)) if subject and subject == artifact_hash: print(" PASS - artifact matches what was presented") else: print(" FAIL - artifact does not match presentation digest") ok = False # [4b] domain context (evidence held outside the claim) vs context_hash if context_path: print("\n[4b] Evidence context vs context_hash") with open(context_path) as f: ctx = json.load(f) expected = claim.get("presentation", {}).get("context_hash", "") computed = "sha256:" + hashlib.sha256(jcs(ctx)).hexdigest() print(" Claim context_hash : {}".format(expected or "(none)")) print(" Computed : {}".format(computed)) if expected and expected == computed: print(" PASS - evidence context matches the signed context_hash") else: print(" FAIL - context does not match context_hash") ok = False # [5] tlog (optional) if ledger_uuid: print("\n[5] Transparency log (server: {})".format(tlog_server)) entry, err = fetch_tlog_entry(ledger_uuid, tlog_server) if err or not entry: print(" FAIL - could not fetch entry: {}".format(err or "not found")) ok = False else: tlog_hash = "" try: rekord = json.loads(base64.b64decode(entry.get("body", "") + "==")) tlog_hash = rekord.get("spec", {}).get("data", {}).get("hash", {}).get("value", "") except Exception: pass record_hash = hashlib.sha256(record_file_bytes).hexdigest() print(" Tlog hash : {}".format(tlog_hash or "(not found)")) print(" Record hash : {}".format(record_hash)) if tlog_hash and tlog_hash == record_hash: print(" PASS - record matches the transparency log entry") else: print(" FAIL - record does not match tlog entry") ok = False anchors = {} if anchors_path: with open(anchors_path) as f: anchors = json.load(f) # [6] Rekor inclusion anchor (offline, from stored anchors.json) if anchors_path: print("\n[6] Rekor transparency-log anchor (offline)") rekor = anchors.get("rekor", {}) if not rekor.get("set"): print(" SKIP - anchors file has no full Rekor proof") else: log_key_pem = None embedded_fp = pem_fingerprint(rekor.get("public_key_pem", "")) if rekor.get("public_key_pem") else None if rekor_pubkey_path: with open(rekor_pubkey_path) as f: log_key_pem = f.read() print(" Log key : supplied out-of-band ({}, sha256:{})".format( rekor_pubkey_path, pem_fingerprint(log_key_pem))) elif pull_keys and rekor.get("log"): url = rekor["log"].rstrip("/") + "/api/v1/log/publicKey" try: log_key_pem = http_get_bytes(url).decode() fp = pem_fingerprint(log_key_pem) print(" Log key : pulled from {} (sha256:{})".format(url, fp)) if embedded_fp and fp != embedded_fp: print(" ! current log key DIFFERS from the one anchored at sealing " "(sha256:{}) — rotated or revoked since; the SET below will not".format(embedded_fp)) print(" verify against it. Cross-check this fingerprint against the log's revocation status.") elif embedded_fp: print(" key matches the one anchored at sealing") except Exception as e: print(" Log key : FAIL - could not pull from {}: {}".format(url, e)) log_key_pem = None ok = False else: print(" Log key : from the anchors (supply --rekor-pubkey or --pull-keys to anchor to a trusted log key)") try: for label, check in verify_rekor(rekor, record_file_bytes, log_key_pem): print(" {} - {}".format("PASS" if check else "FAIL", label)) ok = ok and check except ImportError: print(" SKIP - 'cryptography' not installed") ok = False # [7] RFC 3161 timestamp (offline, token + TSA CA) if tsa_token_path: print("\n[7] RFC 3161 timestamp (offline)") with open(tsa_token_path, "rb") as f: token_der = f.read() ca_pem = None tsa_url = anchors.get("rfc3161", {}).get("tsa") if tsa_cert_path and tsa_cert_trusted: with open(tsa_cert_path, "rb") as f: ca_pem = f.read() print(" TSA CA : supplied out-of-band ({}, sha256:{})".format( tsa_cert_path, pem_fingerprint(ca_pem))) elif pull_keys and tsa_url: # /files/cacert.pem mirrors public TSAs (e.g. freetsa.org). url = tsa_url.rstrip("/") + "/files/cacert.pem" try: ca_pem = http_get_bytes(url) print(" TSA CA : pulled from {} (sha256:{})".format(url, pem_fingerprint(ca_pem))) except Exception as e: print(" TSA CA : FAIL - could not pull from {}: {}".format(url, e)) ok = False elif tsa_cert_path: with open(tsa_cert_path, "rb") as f: ca_pem = f.read() print(" TSA CA : from the bundle (sha256:{}; supply --tsa-cert or --pull-keys " "to anchor to a trusted CA)".format(pem_fingerprint(ca_pem))) else: print(" FAIL - TSA CA required (pass --tsa-cert, or --pull-keys with an anchor URL)") ok = False if ca_pem is not None: t_ok, detail = verify_rfc3161(token_der, ca_pem, declared_hash) if t_ok is None: print(" SKIP - {}".format(detail)) else: print(" {} - {}".format("PASS" if t_ok else "FAIL", detail)) ok = ok and t_ok # summary dec = claim.get("decision", {}) human = claim.get("human", {}) print("\n[*] Decision") print(" {} - {}".format(dec.get("class"), dec.get("outcome"))) print(" Approver : {} ({}, webauthn={})".format(human.get("id"), human.get("role"), human.get("webauthn"))) print(" Rationale : {}".format(claim.get("rationale", {}).get("text"))) print(" Signed at : {}".format(claim.get("issued_at"))) print("\n{}".format("PASS - All checks passed" if ok else "FAIL - One or more checks failed")) return ok def run_bundle(bundle_path, trust_pubkey_path, tmpdir, artifact_path, rp_id, ledger_uuid, tlog_server, tsa_cert_path=None, rekor_pubkey_path=None, pull_keys=False): """Verify directly from a .sspb bundle (a zip) — no manual unzipping. Members are extracted to a private temp dir and fed to run(). The bundle carries the evidence (record, anchors, context, timestamp, TSA CA) but NOT the Sovereign trust public key: that is the root of trust and must be supplied out-of-band, so a bundle cannot self-certify with a forged key. The TSA CA and the Rekor log key are likewise trust anchors. A verifier- supplied --tsa-cert / --rekor-pubkey is used in preference to the copies that travel in the bundle; falling back to the bundled ones is flagged in output. """ if not zipfile.is_zipfile(bundle_path): print("FAIL - {} is not a .sspb bundle (zip archive)".format(bundle_path)) return False with zipfile.ZipFile(bundle_path) as z: names = set(z.namelist()) # Only extract the members we know — never honour absolute/parent paths. for member in ("record.ssp.json", "anchors.json", "context.json", "timestamp.tsr", "tsa-ca.pem"): if member in names: with z.open(member) as src, open(os.path.join(tmpdir, member), "wb") as dst: dst.write(src.read()) def have(name): path = os.path.join(tmpdir, name) return path if os.path.exists(path) else None record = have("record.ssp.json") if not record: print("FAIL - bundle is missing record.ssp.json") return False # TSA CA precedence: explicit --tsa-cert (out-of-band, trusted) > pulled online # (--pull-keys) > the copy bundled in the .sspb. Only fall back to the bundled # CA when neither an out-of-band file nor --pull-keys was given. tsa_cert = tsa_cert_path tsa_cert_trusted = tsa_cert_path is not None if tsa_cert is None and not pull_keys: tsa_cert = have("tsa-ca.pem") print("Bundle : {} (verifying without unzipping)\n".format(bundle_path)) return run(record, trust_pubkey_path, artifact_path, have("context.json"), rp_id, ledger_uuid, tlog_server, have("anchors.json"), have("timestamp.tsr"), tsa_cert, rekor_pubkey_path=rekor_pubkey_path, tsa_cert_trusted=tsa_cert_trusted, pull_keys=pull_keys) if __name__ == "__main__": p = argparse.ArgumentParser(description="SSP Sovereign Record offline verifier", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__) p.add_argument("record", nargs="?", help="A .sspb or .sspa bundle (both zips, verified directly without unzip), or record.ssp.json") p.add_argument("trust_pubkey", nargs="?", help="The Sovereign trust public key PEM (supplied out-of-band — never read from the bundle). " "Required to verify the trust signature.") p.add_argument("--artifact", help="Original artifact file to check against the presentation digest") p.add_argument("--context-file", help="Domain context JSON to check against presentation.context_hash") p.add_argument("--rp-id", default="app.local", help="Expected WebAuthn rp.id (default: app.local)") p.add_argument("--ledger-uuid", help="Also check the transparency log entry (online)") p.add_argument("--tlog-server", default="https://tlog.local", help="Tlog base URL (default: https://tlog.local)") p.add_argument("--anchors-file", help="Stored anchors.json to verify the Rekor proof offline (SET + inclusion + checkpoint)") p.add_argument("--tsa-token", help="RFC 3161 timestamp token (timestamp.tsr) to verify offline") p.add_argument("--tsa-cert", help="TSA CA certificate PEM (supplied out-of-band — preferred over any " "tsa-ca.pem that travels in a .sspb bundle)") p.add_argument("--rekor-pubkey", help="Transparency-log public key PEM (supplied out-of-band — preferred " "over the log key embedded in the anchors, so the proof anchors to a " "trusted key)") p.add_argument("--pull-keys", action="store_true", help="Pull the TSA CA and the transparency-log public key from the anchor URLs recorded in " "anchors.json (rfc3161.tsa, rekor.log) at validation time. Logs each source URL and the " "SHA-256 fingerprint of the key pulled, and flags when it differs from the key anchored " "at sealing (rotated/revoked). Explicit --tsa-cert / --rekor-pubkey take precedence.") args = p.parse_args() if not args.record: p.error("provide a .sspb bundle or a .sspa attestation, plus the trust public key") # A .sspb bundle (zip) is verified directly; the trust key must be supplied # separately and is never taken from the bundle itself. The TSA CA and Rekor # log key may likewise be supplied out-of-band, preferred over bundled copies. if zipfile.is_zipfile(args.record): if not args.trust_pubkey: p.error("the Sovereign trust public key is required: verify-ssp.py ") with tempfile.TemporaryDirectory() as tmp: ok = run_bundle(args.record, args.trust_pubkey, tmp, args.artifact, args.rp_id, args.ledger_uuid, args.tlog_server, tsa_cert_path=args.tsa_cert, rekor_pubkey_path=args.rekor_pubkey, pull_keys=args.pull_keys) sys.exit(0 if ok else 1) sys.exit(0 if run(args.record, args.trust_pubkey, args.artifact, args.context_file, args.rp_id, args.ledger_uuid, args.tlog_server, args.anchors_file, args.tsa_token, args.tsa_cert, rekor_pubkey_path=args.rekor_pubkey, tsa_cert_trusted=True, pull_keys=args.pull_keys) else 1)