#!/usr/bin/env python3 # pylint: disable=missing-module-docstring """qstrike-verify -- independent verifier for QStrike public bundles. Single-file Python CLI that lets any external expert independently re-derive three categories of trust claims published in a QStrike master bundle: 1. signature ECDSA-P256-SHA256 over RFC 8785 canonical JSON of the bundle (with attestation_chain.bundle_signature stripped before canonicalization). Verified against the published verification-key.json (ECDSA-P256 public key, SubjectPublicKeyInfo). 2. oid Each entry in bundle.crypto[].cryptoProperties.algorithmProperties.oid is looked up in a built-in OID registry sourced from RFC 5758, RFC 5480, RFC 8017, FIPS 203/204/205, and IANA. The canonical algorithm name is compared to the entry's `algorithm` / `parameterSetIdentifier` claim. 3. stats bundle.findings[] are re-checked against the Bonferroni predicate documented inside each finding (alpha_per_test = alpha_FWE / N_family; HIGH iff p < alpha_per_test AND confidence_basis == "deterministic"; MEDIUM iff p < alpha_FWE; else LOW). bundle.cross_platform_consistency.tvd[] is re-checked against the definitional identity agreement + tvd == 1. Subcommands also support `all` (run all three; exit-code is bitwise OR of component failures) and `info` (print version, fingerprint, exit-code legend). Exit codes (powers of two, OR-able for the `all` subcommand): 0 all verifications PASS 2 signature FAIL 4 oid FAIL 8 stats FAIL 32 input error (missing / malformed bundle / verification key) This script depends only on Python 3.10+ stdlib and the `cryptography` PyPI package (already installed on the QStrike production pod). No third-party RFC 8785 library, no scipy. Designed for unsupervised execution by external expert reviewers behind any corporate proxy. """ from __future__ import annotations import argparse import base64 import hashlib import json import os import sys import urllib.error import urllib.request from typing import Any, Optional, Tuple # --------------------------------------------------------------------------- # Version + exit codes # --------------------------------------------------------------------------- VERSION = "1.0.0-mvp" EXIT_OK = 0 EXIT_FAIL_SIGNATURE = 2 EXIT_FAIL_OID = 4 EXIT_FAIL_STATS = 8 EXIT_FAIL_QASM = 16 EXIT_INPUT_ERROR = 32 EXIT_FAIL_OCSF = 64 EXIT_FAIL_LEDGER = 128 DEFAULT_KEY_URL = "https://qstrike.qtonicquantum.com/verification-key.json" # --------------------------------------------------------------------------- # OID registry (RFC 5758, RFC 5480, RFC 8017, FIPS 203/204/205, IANA, NIST CSOR) # --------------------------------------------------------------------------- # Each entry: oid -> (canonical_name, source, aliases) # `aliases` is a list of acceptable strings the bundle is allowed to claim for # `algorithm` / `parameterSetIdentifier`. Comparison is case-insensitive. OID_REGISTRY: dict[str, tuple[str, str, list[str]]] = { # RSA family -- RFC 8017 (PKCS#1), RFC 4055 "1.2.840.113549.1.1.1": ( "rsaEncryption", "RFC 8017", ["rsa", "rsa-2048", "rsa-3072", "rsa-4096", "rsaEncryption"], ), "1.2.840.113549.1.1.11": ( "sha256WithRSAEncryption", "RFC 8017", ["rsa-sha256", "sha256WithRSAEncryption", "RSA-SHA256"], ), "1.2.840.113549.1.1.12": ( "sha384WithRSAEncryption", "RFC 8017", ["rsa-sha384", "sha384WithRSAEncryption"], ), "1.2.840.113549.1.1.13": ( "sha512WithRSAEncryption", "RFC 8017", ["rsa-sha512", "sha512WithRSAEncryption"], ), "1.2.840.113549.1.1.10": ( "id-RSASSA-PSS", "RFC 8017", ["rsa-pss", "rsassa-pss", "id-RSASSA-PSS"], ), # ECDSA -- RFC 5758, RFC 5480 "1.2.840.10045.2.1": ( "id-ecPublicKey", "RFC 5480", ["ecdsa", "ec", "id-ecPublicKey"], ), "1.2.840.10045.4.3.2": ( "ecdsa-with-SHA256", "RFC 5758", ["ecdsa-p256", "ecdsa-with-sha256", "ECDSA-P256", "ECDSA-P256-SHA256", "secp256r1"], ), "1.2.840.10045.4.3.3": ( "ecdsa-with-SHA384", "RFC 5758", ["ecdsa-p384", "ecdsa-with-sha384", "ECDSA-P384", "secp384r1"], ), "1.2.840.10045.4.3.4": ( "ecdsa-with-SHA512", "RFC 5758", ["ecdsa-p521", "ecdsa-with-sha512", "ECDSA-P521", "secp521r1"], ), # ECDSA named curves (RFC 5480 / SECG) "1.2.840.10045.3.1.7": ( "secp256r1", "RFC 5480", ["secp256r1", "prime256v1", "P-256", "ecdsa-p256"], ), "1.3.132.0.34": ( "secp384r1", "RFC 5480", ["secp384r1", "P-384", "ecdsa-p384"], ), "1.3.132.0.35": ( "secp521r1", "RFC 5480", ["secp521r1", "P-521", "ecdsa-p521"], ), "1.3.132.0.10": ( "secp256k1", "SECG SEC2", ["secp256k1", "P-256K"], ), # EdDSA -- RFC 8410 "1.3.101.110": ("id-X25519", "RFC 8410", ["x25519", "id-X25519"]), "1.3.101.111": ("id-X448", "RFC 8410", ["x448", "id-X448"]), "1.3.101.112": ("id-Ed25519", "RFC 8410", ["ed25519", "id-Ed25519", "EdDSA-Ed25519"]), "1.3.101.113": ("id-Ed448", "RFC 8410", ["ed448", "id-Ed448", "EdDSA-Ed448"]), # Diffie-Hellman -- RFC 3279 "1.2.840.113549.1.3.1": ( "dhKeyAgreement", "RFC 3279", ["dh", "diffie-hellman", "dhKeyAgreement"], ), # AES -- NIST FIPS 197, RFC 3565 "2.16.840.1.101.3.4.1.2": ("aes-128-cbc", "NIST FIPS 197", ["aes-128-cbc", "aes128-cbc"]), "2.16.840.1.101.3.4.1.22": ("aes-192-cbc", "NIST FIPS 197", ["aes-192-cbc", "aes192-cbc"]), "2.16.840.1.101.3.4.1.42": ( "aes-256-cbc", "NIST FIPS 197", ["aes-256-cbc", "aes256-cbc", "AES-256", "AES-256-CBC"], ), "2.16.840.1.101.3.4.1.6": ("aes-128-gcm", "NIST FIPS 197", ["aes-128-gcm", "aes128-gcm"]), "2.16.840.1.101.3.4.1.26": ("aes-192-gcm", "NIST FIPS 197", ["aes-192-gcm", "aes192-gcm"]), "2.16.840.1.101.3.4.1.46": ( "aes-256-gcm", "NIST FIPS 197", ["aes-256-gcm", "aes256-gcm", "AES-256-GCM"], ), # SHA-2 / SHA-3 -- NIST FIPS 180-4 / FIPS 202 "2.16.840.1.101.3.4.2.1": ("sha256", "NIST FIPS 180-4", ["sha-256", "sha256", "SHA2-256"]), "2.16.840.1.101.3.4.2.2": ("sha384", "NIST FIPS 180-4", ["sha-384", "sha384", "SHA2-384"]), "2.16.840.1.101.3.4.2.3": ("sha512", "NIST FIPS 180-4", ["sha-512", "sha512", "SHA2-512"]), "2.16.840.1.101.3.4.2.8": ("sha3-256", "NIST FIPS 202", ["sha3-256", "SHA3-256"]), "2.16.840.1.101.3.4.2.9": ("sha3-384", "NIST FIPS 202", ["sha3-384", "SHA3-384"]), "2.16.840.1.101.3.4.2.10": ("sha3-512", "NIST FIPS 202", ["sha3-512", "SHA3-512"]), # ML-KEM -- NIST FIPS 203 "2.16.840.1.101.3.4.4.1": ( "ml-kem-512", "NIST FIPS 203", ["ml-kem-512", "kyber-512", "ML-KEM-512"], ), "2.16.840.1.101.3.4.4.2": ( "ml-kem-768", "NIST FIPS 203", ["ml-kem-768", "kyber-768", "ML-KEM-768"], ), "2.16.840.1.101.3.4.4.3": ( "ml-kem-1024", "NIST FIPS 203", ["ml-kem-1024", "kyber-1024", "ML-KEM-1024"], ), # ML-DSA -- NIST FIPS 204 "2.16.840.1.101.3.4.3.17": ( "ml-dsa-44", "NIST FIPS 204", ["ml-dsa-44", "dilithium2", "ML-DSA-44"], ), "2.16.840.1.101.3.4.3.18": ( "ml-dsa-65", "NIST FIPS 204", ["ml-dsa-65", "dilithium3", "ML-DSA-65"], ), "2.16.840.1.101.3.4.3.19": ( "ml-dsa-87", "NIST FIPS 204", ["ml-dsa-87", "dilithium5", "ML-DSA-87"], ), # SLH-DSA -- NIST FIPS 205 (selected representatives) "2.16.840.1.101.3.4.3.20": ( "slh-dsa-sha2-128s", "NIST FIPS 205", ["slh-dsa-sha2-128s", "sphincs-sha2-128s", "SLH-DSA-SHA2-128s", "SLH-DSA-128s", "SPHINCS+-128s", "SPHINCS+-SHA256-128s"], ), "2.16.840.1.101.3.4.3.21": ( "slh-dsa-sha2-128f", "NIST FIPS 205", ["slh-dsa-sha2-128f", "sphincs-sha2-128f", "SLH-DSA-SHA2-128f", "SLH-DSA-128f", "SPHINCS+-128f", "SPHINCS+-SHA256-128f"], ), "2.16.840.1.101.3.4.3.22": ( "slh-dsa-sha2-192s", "NIST FIPS 205", ["slh-dsa-sha2-192s", "sphincs-sha2-192s", "SLH-DSA-SHA2-192s", "SLH-DSA-192s", "SPHINCS+-192s", "SPHINCS+-SHA256-192s"], ), "2.16.840.1.101.3.4.3.23": ( "slh-dsa-sha2-192f", "NIST FIPS 205", ["slh-dsa-sha2-192f", "sphincs-sha2-192f", "SLH-DSA-SHA2-192f", "SLH-DSA-192f", "SPHINCS+-192f", "SPHINCS+-SHA256-192f"], ), "2.16.840.1.101.3.4.3.24": ( "slh-dsa-sha2-256s", "NIST FIPS 205", ["slh-dsa-sha2-256s", "sphincs-sha2-256s", "SLH-DSA-SHA2-256s", "SLH-DSA-256s", "SPHINCS+-256s", "SPHINCS+-SHA256-256s"], ), "2.16.840.1.101.3.4.3.25": ( "slh-dsa-sha2-256f", "NIST FIPS 205", ["slh-dsa-sha2-256f", "sphincs-sha2-256f", "SLH-DSA-SHA2-256f", "SLH-DSA-256f", "SPHINCS+-256f", "SPHINCS+-SHA256-256f"], ), # TLS / X.509 helpers (RFC 5280) "1.2.840.113549.1.7.1": ("id-data", "RFC 5652", ["id-data"]), "2.5.29.15": ("id-ce-keyUsage", "RFC 5280", ["keyUsage"]), "2.5.29.37": ("id-ce-extKeyUsage", "RFC 5280", ["extKeyUsage"]), # Job 3 Phase 1.0b: expand registry to cover 7 unknown OIDs seen in Phase 0.5c bundle # + 20 additional common OIDs (X.509 extensions, hash functions, PKCS#1 / DH) # RSA -- additional PKCS#1 variants "1.2.840.113549.1.1.7": ( "id-RSAES-OAEP", "RFC 8017 §A.2.1", ["rsa-oaep", "rsaes-oaep", "RSA-OAEP", "RSA-3072", "RSA-OAEP-MGF1"], ), "1.2.840.113549.1.1.8": ( "id-mgf1", "RFC 8017 §A.2.1", ["mgf1", "id-mgf1"], ), "1.2.840.113549.1.1.9": ( "id-pSpecified", "RFC 8017 §A.2.1", ["pspecified", "id-pSpecified"], ), # HMAC -- RFC 4231 + PKCS#5 v2.1 "1.2.840.113549.2.7": ( "id-hmacWithSHA1", "RFC 4231", ["hmac-sha1", "HMAC-SHA1"], ), "1.2.840.113549.2.8": ( "id-hmacWithSHA224", "RFC 4231", ["hmac-sha224", "HMAC-SHA224"], ), "1.2.840.113549.2.9": ( "id-hmacWithSHA256", "RFC 4231", ["hmac-sha256", "HMAC-SHA256", "hmacWithSHA256"], ), "1.2.840.113549.2.10": ( "id-hmacWithSHA384", "RFC 4231", ["hmac-sha384", "HMAC-SHA384"], ), "1.2.840.113549.2.11": ( "id-hmacWithSHA512", "RFC 4231", ["hmac-sha512", "HMAC-SHA512"], ), # AEAD -- ChaCha20-Poly1305 (RFC 8103) "1.2.840.113549.1.9.16.3.18": ( "id-alg-AEADChaCha20Poly1305", "RFC 8103", ["chacha20-poly1305", "chacha20", "ChaCha20-Poly1305", "id-alg-AEADChaCha20Poly1305"], ), # Additional ECDSA curves -- SECG SEC2 "1.3.132.0.33": ( "secp224r1", "RFC 5480", ["secp224r1", "P-224"], ), "1.3.132.1.12": ( "id-ecDH", "SECG SEC1 §C.1", ["dh-group19", "DH-group19", "dh-group20", "DH-group20", "dh-group21", "DH-group21", "ecdh", "id-ecDH", "ECDH"], ), # X9.42 DH -- RFC 3279 "1.2.840.10046.2.1": ( "dhpublicnumber", "ANSI X9.42 / RFC 3279", ["dh-group14", "DH-group14", "dh", "dhpublicnumber", "x942-dh"], ), # PQClean experimental OIDs (no formal NIST assignment yet) # Falcon round-3/4 (FN-DSA) -- PQClean OID family 1.3.9999.6.7.x "1.3.9999.6.7.4": ( "falcon-512-pqclean", "PQClean / Open Quantum Safe", ["falcon-512", "falcon-1024", "fn-dsa-512", "fn-dsa-1024", "FN-DSA-512", "FN-DSA-1024", "FN-DSA-512-pilot"], ), # HQC -- PQClean experimental "1.3.6.1.4.1.2.267.12.4.4": ( "hqc-128-pqclean", "PQClean", ["hqc", "hqc-128", "HQC", "HQC-128", "HQC-128-pilot"], ), # X.509 extension OIDs (RFC 5280 §4.2) "2.5.29.14": ("id-ce-subjectKeyIdentifier", "RFC 5280", ["subjectKeyIdentifier", "id-ce-subjectKeyIdentifier"]), "2.5.29.17": ("id-ce-subjectAltName", "RFC 5280", ["subjectAltName", "SAN", "id-ce-subjectAltName"]), "2.5.29.19": ("id-ce-basicConstraints", "RFC 5280", ["basicConstraints", "id-ce-basicConstraints"]), "2.5.29.31": ("id-ce-cRLDistributionPoints", "RFC 5280", ["crlDistributionPoints", "id-ce-cRLDistributionPoints"]), "2.5.29.32": ("id-ce-certificatePolicies", "RFC 5280", ["certificatePolicies", "id-ce-certificatePolicies"]), "2.5.29.35": ("id-ce-authorityKeyIdentifier", "RFC 5280", ["authorityKeyIdentifier", "AKI", "id-ce-authorityKeyIdentifier"]), # PKCS#7/PKIX additional helpers (RFC 5652 / RFC 5280) "1.2.840.113549.1.7.2": ("id-signedData", "RFC 5652", ["id-signedData", "signedData"]), "1.2.840.113549.1.7.3": ("id-envelopedData", "RFC 5652", ["id-envelopedData", "envelopedData"]), "1.3.6.1.5.5.7.3.1": ("id-kp-serverAuth", "RFC 5280", ["serverAuth", "id-kp-serverAuth"]), "1.3.6.1.5.5.7.3.2": ("id-kp-clientAuth", "RFC 5280", ["clientAuth", "id-kp-clientAuth"]), "1.3.6.1.5.5.7.3.3": ("id-kp-codeSigning", "RFC 5280", ["codeSigning", "id-kp-codeSigning"]), "1.3.6.1.5.5.7.3.4": ("id-kp-emailProtection", "RFC 5280", ["emailProtection", "id-kp-emailProtection"]), # Hash function additional (NIST FIPS 180-4 / 202) "2.16.840.1.101.3.4.2.4": ("sha224", "NIST FIPS 180-4", ["sha-224", "sha224", "SHA2-224"]), "2.16.840.1.101.3.4.2.7": ("sha3-224", "NIST FIPS 202", ["sha3-224", "SHA3-224"]), "2.16.840.1.101.3.4.2.11": ("shake128", "NIST FIPS 202", ["shake128", "SHAKE-128"]), "2.16.840.1.101.3.4.2.12": ("shake256", "NIST FIPS 202", ["shake256", "SHAKE-256"]), # AES additional variants (NIST FIPS 197) "2.16.840.1.101.3.4.1.5": ("aes-128-ofb", "NIST FIPS 197", ["aes-128-ofb"]), "2.16.840.1.101.3.4.1.25": ("aes-192-ofb", "NIST FIPS 197", ["aes-192-ofb"]), "2.16.840.1.101.3.4.1.45": ("aes-256-ofb", "NIST FIPS 197", ["aes-256-ofb"]), # TripleDES -- RFC 8018 (PKCS#5 v2.1) "1.2.840.113549.3.7": ( "des-EDE3-CBC", "RFC 8018", ["3des", "des-ede3-cbc", "3DES", "DES3"], ), } # --------------------------------------------------------------------------- # I/O helpers # --------------------------------------------------------------------------- def fetch(path_or_url: str) -> bytes: """Load bytes from local file or http(s) URL. Raises FileNotFoundError / urllib.error.URLError on failure.""" if path_or_url.startswith(("http://", "https://")): req = urllib.request.Request( path_or_url, headers={"User-Agent": f"qstrike-verify/{VERSION}"}, ) with urllib.request.urlopen(req, timeout=30) as resp: return resp.read() with open(path_or_url, "rb") as fh: return fh.read() def load_json(path_or_url: str) -> Any: raw = fetch(path_or_url) return json.loads(raw.decode("utf-8")) # --------------------------------------------------------------------------- # RFC 8785 JSON Canonicalization Scheme (JCS) # --------------------------------------------------------------------------- # Recursive sorted-keys + no whitespace + UTF-8 + ECMAScript number formatting. # Mirrors qstrike-engine/signer.py canonicalize_json() so that the signature # bytes round-trip byte-for-byte. def _ecma_number(n: Any) -> str: """Format `n` to match JS JSON.stringify(n) per RFC 8785 sec 3.2.2 (ECMAScript ToString).""" if isinstance(n, bool): return "true" if n else "false" if isinstance(n, int): return str(n) if n != n: # NaN return "null" if n in (float("inf"), float("-inf")): return "null" if n == 0: return "0" s = repr(n) if s.endswith(".0"): return s[:-2] if "e" not in s and "E" not in s: return s ax = abs(n) if 1e-6 <= ax < 1e21: if ax >= 1: for prec in range(1, 18): d = format(n, f".{prec}f") if float(d) == n: break else: for prec in range(1, 22): d = format(n, f".{prec}f") if float(d) == n: break if "." in d: d = d.rstrip("0").rstrip(".") if d in ("", "-"): d = "0" return d m, e = s.lower().split("e") ei = int(e) if "." in m: m = m.rstrip("0").rstrip(".") return m + ("e+" if ei >= 0 else "e") + str(ei) def canonicalize_json(obj: Any) -> bytes: """RFC 8785 JCS: recursive sort + no whitespace + UTF-8 + ECMAScript number format.""" if obj is None: return b"null" if isinstance(obj, bool): return b"true" if obj else b"false" if isinstance(obj, (int, float)): return _ecma_number(obj).encode("utf-8") if isinstance(obj, str): return json.dumps(obj, ensure_ascii=False).encode("utf-8") if isinstance(obj, (list, tuple)): return b"[" + b",".join(canonicalize_json(v) for v in obj) + b"]" if isinstance(obj, dict): parts = [] for k in sorted(obj.keys()): ke = json.dumps(k, ensure_ascii=False).encode("utf-8") parts.append(ke + b":" + canonicalize_json(obj[k])) return b"{" + b",".join(parts) + b"}" raise TypeError(f"unsupported JSON type: {type(obj).__name__}") # --------------------------------------------------------------------------- # Signature verification # --------------------------------------------------------------------------- def _maybe_decode_signature(sig_str: str) -> bytes: """Accept either hex or base64. Bundle attestation_chain.bundle_signature is base64.""" s = sig_str.strip() if len(s) % 2 == 0 and all(c in "0123456789abcdefABCDEF" for c in s): try: return bytes.fromhex(s) except ValueError: pass return base64.b64decode(s) def _load_pub_key(key_doc: dict) -> Tuple[Any, str]: """Return (cryptography EllipticCurvePublicKey, sha256(DER) fingerprint).""" from cryptography.hazmat.primitives import serialization pem = key_doc.get("public_key_pem") if not pem or "BEGIN PUBLIC KEY" not in pem: raise ValueError("verification-key.json missing or malformed public_key_pem") pub = serialization.load_pem_public_key(pem.encode("utf-8")) der = pub.public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo, ) fp = hashlib.sha256(der).hexdigest() return pub, fp def verify_signature( bundle: dict, key_doc: dict, verbose: bool = False, tolerate_expired_keys: bool = False, ) -> tuple[int, dict]: """Verify bundle.attestation_chain.bundle_signature. Returns (exit_code, report).""" from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec report: dict[str, Any] = {"subcommand": "signature"} ac = bundle.get("attestation_chain") if not isinstance(ac, dict): report["status"] = "FAIL" report["error"] = "bundle missing attestation_chain" return EXIT_FAIL_SIGNATURE, report sig_str = ac.get("bundle_signature") if not isinstance(sig_str, str) or not sig_str: report["status"] = "FAIL" report["error"] = "bundle missing attestation_chain.bundle_signature" return EXIT_FAIL_SIGNATURE, report claimed_fp = ac.get("signed_by_fingerprint_sha256", "") report["claimed"] = { "signature_algorithm": ac.get("signature_algorithm", ""), "signature_envelope": ac.get("signature_envelope", ""), "signed_by_fingerprint_sha256": claimed_fp, "signed_at": ac.get("signed_at"), } try: pub, key_fp = _load_pub_key(key_doc) except Exception as exc: # pylint: disable=broad-except report["status"] = "FAIL" report["error"] = f"failed to load verification key: {exc}" return EXIT_FAIL_SIGNATURE, report report["verification_key_fingerprint_sha256"] = key_fp report["fingerprint_match"] = (key_fp == claimed_fp) # V5-001 (Phase 1.1 hardening): key-validity window check. # If verification-key.json declares valid_from/valid_until and the bundle # carries attestation_chain.signed_at, the signed_at MUST fall within the # window. Skip with --tolerate-expired-keys. vf = key_doc.get("valid_from") vu = key_doc.get("valid_until") signed_at = ac.get("signed_at") report["key_validity_window"] = { "valid_from": vf, "valid_until": vu, "signed_at": signed_at, "checked": bool(vf and vu and signed_at), "tolerate_expired": tolerate_expired_keys, } if vf and vu and signed_at and not tolerate_expired_keys: if signed_at < vf: report["status"] = "FAIL" report["error"] = ( f"bundle signed_at {signed_at} is before key valid_from {vf}; " f"set --tolerate-expired-keys to override" ) return EXIT_FAIL_SIGNATURE, report if signed_at > vu: report["status"] = "FAIL" report["error"] = ( f"bundle signed_at {signed_at} is after key valid_until {vu}; " f"set --tolerate-expired-keys to override" ) return EXIT_FAIL_SIGNATURE, report # Reproduce sign-time canonical bytes target = json.loads(json.dumps(bundle)) target["attestation_chain"].pop("bundle_signature", None) msg = canonicalize_json(target) report["canonical_bytes_length"] = len(msg) report["canonical_sha256"] = hashlib.sha256(msg).hexdigest() try: sig_bytes = _maybe_decode_signature(sig_str) except Exception as exc: # pylint: disable=broad-except report["status"] = "FAIL" report["error"] = f"failed to decode signature: {exc}" return EXIT_FAIL_SIGNATURE, report report["signature_bytes_len"] = len(sig_bytes) report["signature_bytes_hex"] = sig_bytes.hex() try: pub.verify(sig_bytes, msg, ec.ECDSA(hashes.SHA256())) report["status"] = "PASS" return EXIT_OK, report except InvalidSignature: report["status"] = "FAIL" if not report["fingerprint_match"]: report["error"] = ( "signature verification failed AND fingerprint mismatch: the " "bundle was signed with a different key than the verification-key " "loaded. Confirm verification-key.json is the current one." ) else: report["error"] = ( "signature verification failed despite fingerprint match: bundle " "may have been tampered with after signing, or signer used a " "different canonical-JSON envelope than RFC 8785." ) return EXIT_FAIL_SIGNATURE, report except Exception as exc: # pylint: disable=broad-except report["status"] = "FAIL" report["error"] = f"unexpected verification error: {exc}" return EXIT_FAIL_SIGNATURE, report # --------------------------------------------------------------------------- # OID verification # --------------------------------------------------------------------------- def _normalize_alg(s: str) -> str: """Normalize an algorithm string for fuzzy comparison: lowercase, drop separators, drop common interchangeable substrings. Examples: 'Kyber-512' -> 'kyber512' 'Dilithium-2' -> 'dilithium2' 'SPHINCS+-SHA256-128s' -> 'sphincssha2128s' (SHA256 -> sha2 collapsed) 'SLH-DSA-128s' -> 'slhdsa128s' 'slh-dsa-sha2-128s' -> 'slhdsasha2128s' """ n = (s or "").lower().strip() # Drop separators for ch in ("-", "_", " ", ".", "+"): n = n.replace(ch, "") # SHA256/sha2 are interchangeable for SLH-DSA / SPHINCS+ n = n.replace("sha256", "sha2") n = n.replace("sphincs", "slhdsa") # SPHINCS+ rebranded as SLH-DSA return n def _algorithm_matches(claimed: str, aliases: list[str]) -> bool: """Case-insensitive match. Accepts: (a) exact alias match (b) parameter-set extensions: 'ML-KEM-768-pilot' matches 'ml-kem-768' (c) normalized form match (drops separators, sha256<->sha2, sphincs<->slhdsa) (d) normalized substring containment in either direction """ c = (claimed or "").strip().lower() if not c: return False cn = _normalize_alg(claimed) for a in aliases: al = a.strip().lower() if c == al: return True if c.startswith(al + "-") or c.startswith(al + "."): return True an = _normalize_alg(a) if cn == an: return True # Containment either direction with reasonable length guard to avoid trivial matches if len(an) >= 5 and (an in cn or cn in an): return True return False # RFC 8017 / NIST FIPS 186: the RSA family (rsaEncryption, sha*WithRSAEncryption, # id-RSASSA-PSS) is identified by the same key, so bundle entries that simply # claim "RSA-2048" / "RSA-3072" / "RSA-4096" against ANY RSA OID are operationally # consistent. We treat the algorithm-family prefix as a match. _RSA_OID_PREFIXES = ( "1.2.840.113549.1.1.", # PKCS#1 family root ) _ECDSA_OID_PREFIXES = ( "1.2.840.10045.", # X9.62 / RFC 5480 / RFC 5758 (id-ecPublicKey + ecdsa-with-SHA*) ) def _family_matches(oid: str, claimed: str) -> bool: """Family-level OID -> algorithm match. Treats RSA-* claims as compatible with any RSA-family OID; ECDSA-P* claims as compatible with any ECDSA-family OID.""" c = (claimed or "").strip().upper() if not c: return False if any(oid.startswith(p) for p in _RSA_OID_PREFIXES) and c.startswith("RSA-"): return True if any(oid.startswith(p) for p in _ECDSA_OID_PREFIXES) and c.startswith("ECDSA-"): return True return False def verify_oids(bundle: dict, verbose: bool = False) -> tuple[int, dict]: """Iterate bundle.crypto[] and check OID -> canonical-name lookup.""" report: dict[str, Any] = {"subcommand": "oid"} crypto = bundle.get("crypto") if not isinstance(crypto, list): report["status"] = "FAIL" report["error"] = "bundle missing crypto[]" return EXIT_FAIL_OID, report rows = [] counts = {"match": 0, "mismatch": 0, "unknown": 0, "missing_oid": 0} for idx, entry in enumerate(crypto): if not isinstance(entry, dict): continue props = entry.get("cryptoProperties", {}) if not isinstance(props, dict): props = {} alg_props = props.get("algorithmProperties", {}) if not isinstance(alg_props, dict): alg_props = {} oid = alg_props.get("oid") claimed_alg = entry.get("algorithm") or alg_props.get("parameterSetIdentifier") or "" entry_id = entry.get("id") or entry.get("asset_id") or f"" if not oid: counts["missing_oid"] += 1 rows.append({ "entry": entry_id, "oid": None, "canonical_name": None, "claimed_algorithm": claimed_alg, "status": "missing_oid", "source": None, }) continue reg = OID_REGISTRY.get(oid) if reg is None: counts["unknown"] += 1 rows.append({ "entry": entry_id, "oid": oid, "canonical_name": None, "claimed_algorithm": claimed_alg, "status": "unknown_oid", "source": None, }) continue canonical, source, aliases = reg if _algorithm_matches(claimed_alg, aliases) or _family_matches(oid, claimed_alg): status = "match" else: status = "mismatch" counts[status if status == "match" else "mismatch"] += 1 rows.append({ "entry": entry_id, "oid": oid, "canonical_name": canonical, "claimed_algorithm": claimed_alg, "status": status, "source": source, }) report["total_entries"] = len(crypto) report["counts"] = counts report["rows"] = rows if counts["mismatch"] == 0: report["status"] = "PASS" return EXIT_OK, report report["status"] = "FAIL" report["error"] = f"{counts['mismatch']} OID/algorithm mismatch(es)" return EXIT_FAIL_OID, report # --------------------------------------------------------------------------- # Statistics verification # --------------------------------------------------------------------------- def verify_stats(bundle: dict, verbose: bool = False) -> tuple[int, dict]: """Verify Bonferroni predicate per finding + TVD identity per pair.""" report: dict[str, Any] = {"subcommand": "stats"} findings = bundle.get("findings") if not isinstance(findings, list): report["status"] = "FAIL" report["error"] = "bundle missing findings[]" return EXIT_FAIL_STATS, report bonf_rows = [] bonf_violations = 0 bonf_checked = 0 for idx, f in enumerate(findings): if not isinstance(f, dict): continue p = f.get("p_value") if p is None: continue alpha_fwe = f.get("alpha_fwe_applied") alpha_per = f.get("alpha_per_test_applied") basis = f.get("confidence_basis", "") claimed_conf = f.get("confidence", "") finding_id = f.get("id", f"") if alpha_fwe is None or alpha_per is None: bonf_rows.append({ "finding": finding_id, "p_value": p, "alpha_per_test": alpha_per, "alpha_fwe": alpha_fwe, "basis": basis, "claimed_confidence": claimed_conf, "predicate_says": None, "status": "incomplete", }) continue if p < alpha_per and (basis or "").lower() == "deterministic": predicted = "HIGH" elif p < alpha_fwe: predicted = "MEDIUM" else: predicted = "LOW" status = "match" if predicted == (claimed_conf or "").upper() else "mismatch" if status == "mismatch": bonf_violations += 1 bonf_checked += 1 bonf_rows.append({ "finding": finding_id, "p_value": p, "alpha_per_test": alpha_per, "alpha_fwe": alpha_fwe, "basis": basis, "claimed_confidence": claimed_conf, "predicate_says": predicted, "status": status, }) cpc = bundle.get("cross_platform_consistency", {}) or {} tvd_list = cpc.get("tvd") if isinstance(cpc, dict) else None tvd_rows = [] tvd_violations = 0 # Bundle stores agreement at 3 decimal places and TVD at 4 decimal places; # the residual rounding error has magnitude up to 5e-4. Use 1e-3 tolerance. tvd_tol = 1e-3 if isinstance(tvd_list, list): for idx, pair in enumerate(tvd_list): if not isinstance(pair, dict): continue agreement = pair.get("agreement") tvd = pair.get("tvd") pair_name = pair.get("pair") or pair.get("modality_pair") or f"" if agreement is None or tvd is None: tvd_rows.append({ "pair": pair_name, "agreement": agreement, "tvd": tvd, "sum": None, "status": "incomplete", }) continue try: a = float(agreement) t = float(tvd) except (TypeError, ValueError): tvd_rows.append({ "pair": pair_name, "agreement": agreement, "tvd": tvd, "sum": None, "status": "non_numeric", }) tvd_violations += 1 continue # P2 hardening: TVD domain check before identity check. # agreement and tvd are probabilities/distances; both must be in [0, 1]. # Without this check, agreement=1.5 and tvd=-0.5 sums to 1.0 and passes # the identity check, but represents impossible values. if not (0.0 <= a <= 1.0 and 0.0 <= t <= 1.0): tvd_violations += 1 tvd_rows.append({ "pair": pair_name if isinstance(pair_name, str) else json.dumps(pair_name), "agreement": a, "tvd": t, "sum": a + t, "status": "domain_violation", }) continue s = a + t status = "match" if abs(s - 1.0) <= tvd_tol else "mismatch" if status == "mismatch": tvd_violations += 1 tvd_rows.append({ "pair": pair_name if isinstance(pair_name, str) else json.dumps(pair_name), "agreement": a, "tvd": t, "sum": s, "status": status, }) report["bonferroni"] = { "checked": bonf_checked, "violations": bonf_violations, "rows": bonf_rows, } report["tvd"] = { "checked": len(tvd_rows), "violations": tvd_violations, "tolerance": tvd_tol, "rows": tvd_rows, } if bonf_violations == 0 and tvd_violations == 0: report["status"] = "PASS" return EXIT_OK, report report["status"] = "FAIL" report["error"] = ( f"{bonf_violations} bonferroni violation(s), {tvd_violations} tvd identity violation(s)" ) return EXIT_FAIL_STATS, report # --------------------------------------------------------------------------- # QASM verification (Phase 1.0b) # --------------------------------------------------------------------------- # Per-backend canonical basis-gate sets (extensible). Source: each vendor's # documented native gate set as of 2026-Q1. QASM_BASIS_GATES: dict[str, list[str]] = { "ibm": ["ecr", "rz", "sx", "x", "measure", "barrier", "reset", "delay"], "ibm_quantum": ["ecr", "rz", "sx", "x", "measure", "barrier", "reset", "delay"], "quantinuum": ["rzz", "rx", "ry", "rz", "measure", "barrier", "reset"], "ionq": ["rxx", "rx", "ry", "rz", "measure", "barrier", "reset"], "rigetti": ["cz", "rx", "rz", "measure", "barrier", "reset"], "google": ["cz", "sx", "rz", "x", "measure", "barrier", "reset"], "google-quantum": ["cz", "sx", "rz", "x", "measure", "barrier", "reset"], "google_quantum": ["cz", "sx", "rz", "x", "measure", "barrier", "reset"], "willow": ["cz", "sx", "rz", "x", "measure", "barrier", "reset"], } # Providers that do NOT use gate-model circuits (annealers, photonic, neutral-atom). QASM_NON_GATE_MODEL = {"d-wave", "dwave", "quera", "xanadu"} def _parse_qasm_gate_breakdown(qasm: str) -> dict: """Tokenize OpenQASM-3 source and count gate occurrences by name. Robust to comments (//, /* */), whitespace, parameter expressions, and multi-qubit operands. Does NOT depend on qiskit -- pure regex parser so the verifier remains stdlib-only by default. """ import re as _re qasm = _re.sub(r"//[^\n]*", "", qasm) qasm = _re.sub(r"/\*.*?\*/", "", qasm, flags=_re.DOTALL) counts: dict[str, int] = {} skip = {"OPENQASM", "include", "qubit", "bit", "gate", "if", "else", "for", "while", "return", "def"} depth = 0 for line in qasm.splitlines(): s = line.strip() if not s: continue if s.startswith("gate "): depth = 1 if "{" in s and "}" not in s else 0 continue if depth: if "}" in s: depth = 0 continue # Assignment forms: "name[i] = gate q[j];" or "name = gate q[j];" # The gate is on the RHS, not the LHS. Detect '=' then re-match. if "=" in s: rhs = s.split("=", 1)[1].strip() m = _re.match(r"([A-Za-z_][A-Za-z0-9_]*)\b", rhs) if not m: continue tok = m.group(1) if tok in skip: continue counts[tok] = counts.get(tok, 0) + 1 continue # Phase 1.0c Q5 — detect malformed register declarations. # A "looks-like a register decl" head-token (bit/qubit/creg/qreg/complex/...) # MUST be followed by an optional [] then whitespace + identifier + ';'. # If the head-token is one of these AND the well-formed pattern fails, # the line is malformed QASM -- raise so verify_qasm reports parse_error. looks_like_decl = _re.match( r"(?:bit|qubit|creg|qreg|complex|float|int|uint|angle|duration|stretch|bool)\b", s, ) wellformed_decl = _re.match( r"(?:bit|qubit|creg|qreg|complex|float|int|uint|angle|duration|stretch|bool)" r"(?:\s*\[\s*\d+\s*\])?\s+[A-Za-z_][A-Za-z0-9_]*\s*;\s*$", s, ) if wellformed_decl: continue if looks_like_decl: raise ValueError(f"malformed register-declaration line: {s!r}") m = _re.match(r"([A-Za-z_][A-Za-z0-9_]*)\b", s) if not m: continue tok = m.group(1) if tok in skip: continue counts[tok] = counts.get(tok, 0) + 1 return counts def verify_qasm(bundle: dict, verbose: bool = False) -> tuple[int, dict]: """Verify each provider_jobs[].circuit_qasm uses only the backend's basis gates and that the parsed gate counts match circuit_qasm_transpile_metadata within +/-10%. Non-gate-model providers (D-Wave, QuEra, Xanadu) are skipped with a note. """ pjs = (bundle.get("attestation_chain") or {}).get("provider_jobs") or [] rows: list[dict] = [] fail = False skipped = 0 verified = 0 for j in pjs: prov_raw = (j.get("provider") or "").lower() qasm = j.get("circuit_qasm") claimed_basis = list(j.get("circuit_qasm_basis_gates") or []) tmd = j.get("circuit_qasm_transpile_metadata") or {} if prov_raw in QASM_NON_GATE_MODEL or not qasm: rows.append({ "job_id": j.get("job_id"), "provider": prov_raw, "status": "skip", "reason": "non-gate-model" if prov_raw in QASM_NON_GATE_MODEL else "no_circuit_qasm", }) skipped += 1 continue try: parsed = _parse_qasm_gate_breakdown(qasm) except Exception as exc: rows.append({ "job_id": j.get("job_id"), "provider": prov_raw, "status": "fail", "reason": f"parse_error: {exc}", }) fail = True continue canonical_basis = QASM_BASIS_GATES.get(prov_raw) or claimed_basis or [] basis_set = set(canonical_basis) | set(claimed_basis) basis_set |= {"measure", "barrier", "reset", "delay"} unknown_gates = [g for g in parsed.keys() if g not in basis_set] claimed_breakdown = (tmd or {}).get("gate_breakdown") or {} count_mismatches: list[str] = [] if claimed_breakdown: for gate, claimed_n in claimed_breakdown.items(): parsed_n = parsed.get(gate, 0) if claimed_n == 0: if parsed_n != 0: count_mismatches.append(f"{gate}: claimed=0 parsed={parsed_n}") else: lo = claimed_n * 0.9 hi = claimed_n * 1.1 if not (lo <= parsed_n <= hi): count_mismatches.append(f"{gate}: claimed={claimed_n} parsed={parsed_n} (>10% drift)") row_status = "pass" reasons: list[str] = [] if unknown_gates: row_status = "fail" reasons.append(f"unknown_gates={unknown_gates}") fail = True if count_mismatches: if len(count_mismatches) >= 2: row_status = "fail" fail = True reasons.append("count_drift=" + "; ".join(count_mismatches)) rows.append({ "job_id": j.get("job_id"), "provider": prov_raw, "status": row_status, "basis_gates": claimed_basis, "parsed_gates": sorted(parsed.keys()), "parsed_counts": parsed, "claimed_breakdown": claimed_breakdown, "reasons": reasons, }) if row_status == "pass": verified += 1 status = "PASS" if not fail else "FAIL" counts = {"verified": verified, "skipped": skipped, "failed": sum(1 for r in rows if r.get("status") == "fail")} report = { "subcommand": "qasm", "status": status, "total_jobs": len(pjs), "counts": counts, "rows": rows, } if fail: return EXIT_FAIL_QASM, report return EXIT_OK, report # --------------------------------------------------------------------------- # OCSF verification (Phase 1.0b) # --------------------------------------------------------------------------- import re as _re_ocsf OCSF_REQUIRED_FIELDS = ["class_uid", "activity_id", "severity_id", "metadata"] OCSF_VALID_SEVERITY_IDS = {0, 1, 2, 3, 4, 5, 6} # Phase 1.0c — metadata.version validation. # Format: M.m[.p]; spec set tracks the OCSF release versions currently shipping. OCSF_VERSION_RE = _re_ocsf.compile(r"^[0-9]+\.[0-9]+(\.[0-9]+)?$") OCSF_KNOWN_SPEC_VERSIONS = {"1.0.0", "1.1.0", "1.2.0", "1.3.0", "1.4.0"} def verify_ocsf(bundle: dict, verbose: bool = False) -> tuple[int, dict]: """Check bundle.securityEvents[] for OCSF 1.4 minimum-required fields. A bundle.securityEvents[] entry is treated as an OCSF claim only if it carries the OCSF discriminator field `class_uid`. If the bundle uses a different schema (QStrike-internal), the verifier records an informational note but does not fail -- the bundle isn't claiming OCSF conformance. """ se = bundle.get("securityEvents") or [] if not se: return EXIT_OK, { "subcommand": "ocsf", "status": "PASS", "total_entries": 0, "rows": [], "note": "no securityEvents to verify", } ocsf_claimed = any(isinstance(e, dict) and "class_uid" in e for e in se) if not ocsf_claimed: return EXIT_OK, { "subcommand": "ocsf", "status": "PASS", "total_entries": len(se), "rows": [], "note": ( f"securityEvents[] uses schema-internal (QStrike) format -- " f"no OCSF discriminator (class_uid) present in {len(se)} entries; " "no OCSF claim to validate (informational)." ), } rows: list[dict] = [] fail = False valid = 0 invalid = 0 for i, e in enumerate(se): if not isinstance(e, dict): rows.append({"index": i, "status": "fail", "reason": "not a dict"}) invalid += 1 fail = True continue missing = [f for f in OCSF_REQUIRED_FIELDS if f not in e] reasons = [] if missing: reasons.append(f"missing_fields={missing}") sev = e.get("severity_id") if sev is not None and sev not in OCSF_VALID_SEVERITY_IDS: reasons.append(f"invalid_severity_id={sev} (must be 0..6)") # Phase 1.0c O3 — class_uid range check (positive integer required). cu = e.get("class_uid") if cu is not None and not (isinstance(cu, int) and not isinstance(cu, bool) and cu > 0): reasons.append(f"class_uid out of range: {cu} (must be positive int)") md = e.get("metadata") if md is not None and not isinstance(md, dict): reasons.append("metadata not a dict") elif isinstance(md, dict) and "version" not in md: reasons.append("metadata.version missing") elif isinstance(md, dict): # Phase 1.0c O4 — validate metadata.version format AND against known OCSF spec set. ver = md.get("version") if not (isinstance(ver, str) and OCSF_VERSION_RE.match(ver)): reasons.append(f"metadata.version malformed: {ver!r}") elif ver not in OCSF_KNOWN_SPEC_VERSIONS: reasons.append( f"metadata.version {ver!r} not in OCSF spec set " f"{sorted(OCSF_KNOWN_SPEC_VERSIONS)}" ) time_field = e.get("time") or e.get("time_dt") if time_field is None: reasons.append("time/time_dt missing") if reasons: rows.append({"index": i, "id": e.get("id"), "status": "fail", "reasons": reasons}) invalid += 1 fail = True else: valid += 1 status = "PASS" if not fail else "FAIL" rc = EXIT_FAIL_OCSF if fail else EXIT_OK return rc, { "subcommand": "ocsf", "status": status, "total_entries": len(se), "valid": valid, "invalid": invalid, "rows": rows, } # --------------------------------------------------------------------------- # Phase 1.3 Fix 2: OCSF/ECS strict enforcement on bundle.findings[] # --------------------------------------------------------------------------- ECS_KNOWN_VERSIONS = {"8.14.0", "8.15.0", "8.16.0", "8.17.0"} _OCSF_TIME_DT_RE = _re_ocsf.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}") def verify_findings_ocsf(bundle: dict, verbose: bool = False) -> tuple[int, dict]: """Validate per-finding ocsf_record (OCSF 1.4) + ecs_record (ECS 8.17). Phase 1.3 emits these records on every finding. The check is strict-mode iff ANY finding has ocsf_record present: * every finding must have ocsf_record * ocsf_record.class_uid must be positive int * ocsf_record.activity_id must be int * ocsf_record.time_dt must look like ISO-8601 * ocsf_record.severity_id must be in 1..6 * ocsf_record.metadata.version must be in OCSF_KNOWN_SPEC_VERSIONS * ecs_record.ecs.version must be in ECS_KNOWN_VERSIONS Returns EXIT_FAIL_OCSF if any record malformed; EXIT_OK if none claim OCSF. """ findings = bundle.get("findings") or [] if not isinstance(findings, list) or not findings: return EXIT_OK, { "subcommand": "ocsf_findings", "status": "PASS", "total_findings": 0, "rows": [], "note": "no findings to verify", } has_any_ocsf = any( isinstance(f, dict) and "ocsf_record" in f for f in findings ) if not has_any_ocsf: return EXIT_OK, { "subcommand": "ocsf_findings", "status": "PASS", "total_findings": len(findings), "rows": [], "note": ( "findings[] carry no ocsf_record discriminator; " "no OCSF 1.4 claim to validate (informational)." ), } rows: list[dict] = [] fail = False valid = 0 invalid = 0 for i, f in enumerate(findings): if not isinstance(f, dict): rows.append({"index": i, "status": "fail", "reason": "not a dict"}) invalid += 1 fail = True continue reasons: list[str] = [] ocsf = f.get("ocsf_record") if not isinstance(ocsf, dict): reasons.append("ocsf_record missing on finding (strict-findings mode)") else: cu = ocsf.get("class_uid") if not (isinstance(cu, int) and not isinstance(cu, bool) and cu > 0): reasons.append(f"class_uid invalid: {cu!r} (must be positive int)") act = ocsf.get("activity_id") if not (isinstance(act, int) and not isinstance(act, bool)): reasons.append(f"activity_id invalid: {act!r}") sev = ocsf.get("severity_id") if sev is None or sev not in {1, 2, 3, 4, 5, 6}: reasons.append(f"severity_id invalid: {sev!r} (must be 1..6)") tdt = ocsf.get("time_dt") if not (isinstance(tdt, str) and _OCSF_TIME_DT_RE.match(tdt)): reasons.append(f"time_dt malformed: {tdt!r}") md = ocsf.get("metadata") if not isinstance(md, dict): reasons.append("metadata missing/not a dict") else: ver = md.get("version") if ver not in OCSF_KNOWN_SPEC_VERSIONS: reasons.append( f"metadata.version {ver!r} not in OCSF spec set " f"{sorted(OCSF_KNOWN_SPEC_VERSIONS)}" ) ecs = f.get("ecs_record") if not isinstance(ecs, dict): reasons.append("ecs_record missing on finding") else: ecs_block = ecs.get("ecs") if not isinstance(ecs_block, dict): reasons.append("ecs_record.ecs missing/not a dict") else: ev = ecs_block.get("version") if ev not in ECS_KNOWN_VERSIONS: reasons.append( f"ecs_record.ecs.version {ev!r} not in ECS spec set " f"{sorted(ECS_KNOWN_VERSIONS)}" ) if reasons: rows.append( { "index": i, "id": f.get("id"), "status": "fail", "reasons": reasons, } ) invalid += 1 fail = True else: valid += 1 status = "PASS" if not fail else "FAIL" rc = EXIT_FAIL_OCSF if fail else EXIT_OK return rc, { "subcommand": "ocsf_findings", "status": status, "total_findings": len(findings), "valid": valid, "invalid": invalid, "rows": rows, } # --------------------------------------------------------------------------- # Ledger verification (Phase 1.0b) # --------------------------------------------------------------------------- import re as _re_ledger _HEX64 = _re_ledger.compile(r"^[0-9a-f]{64}\Z") def verify_ledger(bundle: dict, verbose: bool = False) -> tuple[int, dict]: """Verify per-entry sha256 well-formedness + optional chain consistency. Per task brief, this is a self-consistency check, not a Merkle proof. Strict per-entry self-hash recomputation is NOT applied because QStrike bundles use sha256 as a content-lineage anchor (sha256(source||id)), not an entry self-hash; recomputing would mismatch every entry and produce zero signal. The check tests sha256 well-formedness (64 hex chars), chain consistency if prev_hash is claimed, and sort-sanity (warning only). """ el = bundle.get("evidence_ledger") or [] if not el: return EXIT_OK, { "subcommand": "ledger", "status": "PASS", "total_entries": 0, "rows": [], "note": "no evidence_ledger to verify", } rows: list[dict] = [] fail = False valid = 0 malformed = 0 chain_errors = 0 has_prev_hash = any(isinstance(e, dict) and "prev_hash" in e for e in el) # Phase 1.0c L5 — track entry IDs and fail on duplicates. seen_ids: set[str] = set() duplicate_ids = 0 prev_sha = None for i, e in enumerate(el): if not isinstance(e, dict): rows.append({"index": i, "status": "fail", "reason": "not a dict"}) malformed += 1 fail = True continue sha = e.get("sha256") if sha is None: rows.append({"index": i, "id": e.get("id"), "status": "skip", "reason": "no sha256 claim"}) continue if not isinstance(sha, str) or not _HEX64.match(sha): rows.append({ "index": i, "id": e.get("id"), "status": "fail", "reason": "sha256_malformed", "sha256": str(sha)[:80], }) malformed += 1 fail = True continue # Phase 1.0c L5 — reject duplicate ledger entry IDs. eid = e.get("id") if isinstance(eid, str) and eid: if eid in seen_ids: rows.append({ "index": i, "id": eid, "status": "fail", "reason": "duplicate_id", }) duplicate_ids += 1 malformed += 1 fail = True continue seen_ids.add(eid) if has_prev_hash: claimed_prev = e.get("prev_hash") if i == 0: if claimed_prev is not None and claimed_prev != "0" * 64: rows.append({ "index": i, "id": e.get("id"), "status": "fail", "reason": "genesis_prev_hash_nonzero", }) chain_errors += 1 fail = True else: if claimed_prev != prev_sha: rows.append({ "index": i, "id": e.get("id"), "status": "fail", "reason": f"prev_hash_chain_mismatch (claimed={claimed_prev}, expected={prev_sha})", }) chain_errors += 1 fail = True valid += 1 prev_sha = sha sort_warning = None times = [e.get("observed_at") or e.get("timestamp") for e in el if isinstance(e, dict)] if times and all(t is not None for t in times): sorted_times = sorted(times) if times != sorted_times: sort_warning = ( "evidence_ledger is not sorted by observed_at/timestamp " "(non-monotonic; informational only)." ) status = "PASS" if not fail else "FAIL" rc = EXIT_FAIL_LEDGER if fail else EXIT_OK report = { "subcommand": "ledger", "status": status, "total_entries": len(el), "valid": valid, "malformed": malformed, "duplicate_ids": duplicate_ids, "chain_errors": chain_errors, "has_prev_hash": has_prev_hash, "sort_warning": sort_warning, "rows": rows[:50] if not verbose else rows, } return rc, report # --------------------------------------------------------------------------- # Printing # --------------------------------------------------------------------------- def _print_signature_report(rep: dict, verbose: bool) -> None: print(f"[signature] {rep['status']}") if verbose or rep["status"] != "PASS": c = rep.get("claimed", {}) print(f" signature_algorithm: {c.get('signature_algorithm')}") print(f" signature_envelope: {c.get('signature_envelope')}") print(f" signed_at: {c.get('signed_at')}") print(f" claimed_fingerprint_sha256: {c.get('signed_by_fingerprint_sha256')}") print(f" verification_key_fingerprint: {rep.get('verification_key_fingerprint_sha256')}") print(f" fingerprint_match: {rep.get('fingerprint_match')}") print(f" canonical_bytes_length: {rep.get('canonical_bytes_length')}") print(f" canonical_sha256: {rep.get('canonical_sha256')}") sig = rep.get("signature_bytes_hex", "") if sig: print(f" signature_bytes_hex ({rep.get('signature_bytes_len')} B): {sig[:64]}...") if rep.get("error"): print(f" error: {rep['error']}") def _print_oid_report(rep: dict, verbose: bool) -> None: print(f"[oid] {rep['status']}") counts = rep.get("counts", {}) print(f" entries: {rep.get('total_entries', 0)}") print(f" match: {counts.get('match', 0)}") print(f" mismatch: {counts.get('mismatch', 0)}") print(f" unknown_oid: {counts.get('unknown', 0)}") print(f" missing_oid: {counts.get('missing_oid', 0)}") if rep.get("error"): print(f" error: {rep['error']}") if verbose: print(f" {'entry':<30} {'oid':<32} {'canonical':<28} {'claimed':<24} {'status'}") for r in rep.get("rows", []): print( f" {str(r.get('entry'))[:28]:<30} " f"{str(r.get('oid') or '-')[:30]:<32} " f"{str(r.get('canonical_name') or '-')[:26]:<28} " f"{str(r.get('claimed_algorithm') or '-')[:22]:<24} " f"{r.get('status')}" ) else: bad = [r for r in rep.get("rows", []) if r.get("status") == "mismatch"] for r in bad[:10]: print( f" MISMATCH entry={r.get('entry')} oid={r.get('oid')} " f"canonical={r.get('canonical_name')} claimed={r.get('claimed_algorithm')}" ) if len(bad) > 10: print(f" ... {len(bad) - 10} more mismatch(es) suppressed") def _print_stats_report(rep: dict, verbose: bool) -> None: print(f"[stats] {rep['status']}") bonf = rep.get("bonferroni", {}) tvd = rep.get("tvd", {}) print( f" bonferroni: checked={bonf.get('checked', 0)} " f"violations={bonf.get('violations', 0)}" ) print( f" tvd: checked={tvd.get('checked', 0)} " f"violations={tvd.get('violations', 0)} " f"tol={tvd.get('tolerance')}" ) if rep.get("error"): print(f" error: {rep['error']}") if verbose: print(" --- Bonferroni rows ---") for r in bonf.get("rows", []): print( f" {r.get('status'):<10} {str(r.get('finding'))[:30]:<32} " f"p={r.get('p_value')} alpha_per={r.get('alpha_per_test')} " f"alpha_fwe={r.get('alpha_fwe')} basis={r.get('basis')} " f"claimed={r.get('claimed_confidence')} predict={r.get('predicate_says')}" ) print(" --- TVD rows ---") for r in tvd.get("rows", []): print( f" {r.get('status'):<10} {str(r.get('pair'))[:40]:<42} " f"agreement={r.get('agreement')} tvd={r.get('tvd')} sum={r.get('sum')}" ) else: for r in bonf.get("rows", []): if r.get("status") == "mismatch": print( f" BONF_MISMATCH {r.get('finding')} p={r.get('p_value')} " f"claimed={r.get('claimed_confidence')} predict={r.get('predicate_says')}" ) for r in tvd.get("rows", []): if r.get("status") == "mismatch": print( f" TVD_MISMATCH {r.get('pair')} " f"agreement={r.get('agreement')} tvd={r.get('tvd')} sum={r.get('sum')}" ) # --------------------------------------------------------------------------- # Subcommand dispatchers # --------------------------------------------------------------------------- def _print_qasm_report(rep: dict, verbose: bool) -> None: print(f"[qasm] {rep['status']}") counts = rep.get("counts", {}) print(f" total_jobs: {rep.get('total_jobs', 0)}") print(f" verified: {counts.get('verified', 0)}") print(f" skipped: {counts.get('skipped', 0)} (non-gate-model / no QASM)") print(f" failed: {counts.get('failed', 0)}") rows = rep.get("rows", []) failures = [r for r in rows if r.get("status") == "fail"] if verbose or failures: for r in rows: tag = r.get("status", "?").upper().rjust(5) prov = r.get("provider", "?") reasons = r.get("reasons") or [r.get("reason")] if r.get("reason") else r.get("reasons") or [] if verbose: print(f" {tag} {prov:<18} {r.get('job_id','')} {reasons}") elif r.get("status") == "fail": print(f" FAIL {prov:<18} {r.get('job_id','')} {reasons}") def _print_ocsf_report(rep: dict, verbose: bool) -> None: print(f"[ocsf] {rep['status']}") print(f" total_entries: {rep.get('total_entries', 0)}") if rep.get("note"): print(f" note: {rep['note']}") if "valid" in rep: print(f" valid: {rep.get('valid', 0)}") print(f" invalid: {rep.get('invalid', 0)}") rows = rep.get("rows", []) failures = [r for r in rows if r.get("status") == "fail"] if (verbose or failures) and failures: for r in failures[:25]: print(f" FAIL index={r.get('index')} id={r.get('id')} reasons={r.get('reasons')}") if len(failures) > 25: print(f" ... +{len(failures) - 25} more") def _print_ocsf_findings_report(rep: dict, verbose: bool) -> None: print(f"[ocsf_findings] {rep['status']}") print(f" total_findings: {rep.get('total_findings', 0)}") if rep.get("note"): print(f" note: {rep['note']}") if "valid" in rep: print(f" valid: {rep.get('valid', 0)}") print(f" invalid: {rep.get('invalid', 0)}") rows = rep.get("rows", []) failures = [r for r in rows if r.get("status") == "fail"] if (verbose or failures) and failures: for r in failures[:25]: print(f" FAIL index={r.get('index')} id={r.get('id')} reasons={r.get('reasons')}") if len(failures) > 25: print(f" ... +{len(failures) - 25} more") def _print_ledger_report(rep: dict, verbose: bool) -> None: print(f"[ledger] {rep['status']}") print(f" total_entries: {rep.get('total_entries', 0)}") if "valid" in rep: print(f" valid: {rep.get('valid', 0)}") print(f" malformed: {rep.get('malformed', 0)}") print(f" chain_errors: {rep.get('chain_errors', 0)}") print(f" has_prev_hash: {rep.get('has_prev_hash')}") if rep.get("sort_warning"): print(f" warning: {rep['sort_warning']}") if rep.get("note"): print(f" note: {rep['note']}") rows = rep.get("rows", []) failures = [r for r in rows if r.get("status") == "fail"] if (verbose or failures) and failures: for r in failures[:25]: print(f" FAIL index={r.get('index')} id={r.get('id')} reason={r.get('reason')}") if len(failures) > 25: print(f" ... +{len(failures) - 25} more") def _resolve_key(args) -> dict: """Load verification-key.json document (from --key, env, or default).""" key_src = ( getattr(args, "key", None) or os.environ.get("QSTRIKE_VERIFY_KEY") or DEFAULT_KEY_URL ) try: return load_json(key_src) except (urllib.error.URLError, FileNotFoundError, OSError) as exc: raise SystemExit( f"qstrike-verify: cannot load verification key from {key_src}: {exc}\n" f" (override with --key or QSTRIKE_VERIFY_KEY env var)" ) from exc except UnicodeDecodeError as exc: raise SystemExit( f"qstrike-verify: verification key is not UTF-8 (binary or wrong file?): {exc}" ) from exc def _resolve_bundle(args) -> dict: try: return load_json(args.bundle) except (urllib.error.URLError, FileNotFoundError, OSError) as exc: raise SystemExit(f"qstrike-verify: cannot load bundle from {args.bundle}: {exc}") from exc except UnicodeDecodeError as exc: raise SystemExit( f"qstrike-verify: bundle is not UTF-8 (binary or wrong file?): {exc}" ) from exc except json.JSONDecodeError as exc: raise SystemExit(f"qstrike-verify: bundle is not valid JSON: {exc}") from exc def cmd_signature(args) -> int: try: bundle = _resolve_bundle(args) key_doc = _resolve_key(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR tolerate = bool(getattr(args, "tolerate_expired_keys", False)) rc, rep = verify_signature(bundle, key_doc, args.verbose, tolerate_expired_keys=tolerate) _print_signature_report(rep, args.verbose) if args.json: print(json.dumps(rep, indent=2, sort_keys=True)) return rc def cmd_oid(args) -> int: try: bundle = _resolve_bundle(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR rc, rep = verify_oids(bundle, args.verbose) _print_oid_report(rep, args.verbose) if args.json: print(json.dumps(rep, indent=2, sort_keys=True)) return rc def cmd_stats(args) -> int: try: bundle = _resolve_bundle(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR rc, rep = verify_stats(bundle, args.verbose) _print_stats_report(rep, args.verbose) if args.json: print(json.dumps(rep, indent=2, sort_keys=True)) return rc def cmd_qasm(args) -> int: try: bundle = _resolve_bundle(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR rc, rep = verify_qasm(bundle, args.verbose) _print_qasm_report(rep, args.verbose) if args.json: print(json.dumps(rep, indent=2, sort_keys=True)) return rc def cmd_ocsf(args) -> int: try: bundle = _resolve_bundle(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR rc_se, rep_se = verify_ocsf(bundle, args.verbose) _print_ocsf_report(rep_se, args.verbose) # Phase 1.3 Fix 2: strict OCSF/ECS check on bundle.findings[]. rc_fi, rep_fi = verify_findings_ocsf(bundle, args.verbose) _print_ocsf_findings_report(rep_fi, args.verbose) combined = rc_se | rc_fi if args.json: print(json.dumps({"securityEvents": rep_se, "findings": rep_fi}, indent=2, sort_keys=True)) return combined def cmd_ledger(args) -> int: try: bundle = _resolve_bundle(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR rc, rep = verify_ledger(bundle, args.verbose) _print_ledger_report(rep, args.verbose) if args.json: print(json.dumps(rep, indent=2, sort_keys=True)) return rc def cmd_all(args) -> int: try: bundle = _resolve_bundle(args) key_doc = _resolve_key(args) except SystemExit as exc: print(str(exc), file=sys.stderr) return EXIT_INPUT_ERROR rc_sig, rep_sig = verify_signature(bundle, key_doc, args.verbose, tolerate_expired_keys=bool(getattr(args, 'tolerate_expired_keys', False))) _print_signature_report(rep_sig, args.verbose) print() rc_oid, rep_oid = verify_oids(bundle, args.verbose) _print_oid_report(rep_oid, args.verbose) print() rc_stats, rep_stats = verify_stats(bundle, args.verbose) _print_stats_report(rep_stats, args.verbose) print() rc_qasm, rep_qasm = verify_qasm(bundle, args.verbose) _print_qasm_report(rep_qasm, args.verbose) print() rc_ocsf, rep_ocsf = verify_ocsf(bundle, args.verbose) _print_ocsf_report(rep_ocsf, args.verbose) print() # Phase 1.3 Fix 2: strict OCSF/ECS check on bundle.findings[]. rc_ocsf_fi, rep_ocsf_fi = verify_findings_ocsf(bundle, args.verbose) _print_ocsf_findings_report(rep_ocsf_fi, args.verbose) rc_ocsf = rc_ocsf | rc_ocsf_fi print() rc_ledger, rep_ledger = verify_ledger(bundle, args.verbose) _print_ledger_report(rep_ledger, args.verbose) combined = rc_sig | rc_oid | rc_stats | rc_qasm | rc_ocsf | rc_ledger print() print( f"[all] composite_exit_code={combined} " f"(signature={rc_sig}, oid={rc_oid}, stats={rc_stats}, " f"qasm={rc_qasm}, ocsf={rc_ocsf}, ledger={rc_ledger})" ) if args.json: print(json.dumps({ "signature": rep_sig, "oid": rep_oid, "stats": rep_stats, "qasm": rep_qasm, "ocsf": rep_ocsf, "ledger": rep_ledger, "exit_code": combined, }, indent=2, sort_keys=True)) return combined def cmd_info(args) -> int: print(f"qstrike-verify {VERSION}") print() print("Subcommands:") print(" signature verify ECDSA-P256-SHA256 over RFC 8785 canonical bundle JSON") print(" oid verify each crypto[].cryptoProperties.algorithmProperties.oid") print(" against built-in IANA / RFC / FIPS registry") print(" stats re-derive Bonferroni predicate per finding;") print(" re-derive agreement + tvd == 1 per cross-platform pair") print(" qasm parse provider_jobs[].circuit_qasm, check basis-gate subset") print(" + counts vs transpile_metadata (within +/-10%)") print(" ocsf OCSF 1.4 minimum-required fields on securityEvents[]") print(" (info-skip if bundle doesn't claim OCSF schema)") print(" ledger per-entry sha256 well-formedness + optional chain check") print(" on evidence_ledger[]") print(" all run all six; exit-code is bitwise OR of components") print(" info this screen") print() print("Exit-code legend (powers of two; OR-able for `all`):") print(f" {EXIT_OK:>4} all verifications PASS") print(f" {EXIT_FAIL_SIGNATURE:>4} signature FAIL") print(f" {EXIT_FAIL_OID:>4} oid FAIL") print(f" {EXIT_FAIL_STATS:>4} stats FAIL") print(f" {EXIT_FAIL_QASM:>4} qasm FAIL") print(f" {EXIT_INPUT_ERROR:>4} input error (missing bundle / malformed verification key)") print(f" {EXIT_FAIL_OCSF:>4} ocsf FAIL") print(f" {EXIT_FAIL_LEDGER:>4} ledger FAIL") print() print(f"Default verification-key URL: {DEFAULT_KEY_URL}") print(f"OID registry size: {len(OID_REGISTRY)} entries") print() try: key_doc = load_json(DEFAULT_KEY_URL) fp = key_doc.get("fingerprint_sha256", "(not present in key doc)") print(f"Current verification-key fingerprint_sha256: {fp}") except Exception as exc: # pylint: disable=broad-except print(f"(could not fetch verification-key.json: {exc})") return EXIT_OK def build_parser() -> argparse.ArgumentParser: parent = argparse.ArgumentParser( prog="qstrike-verify", description=( "Independent verifier for QStrike public bundles " "(ECDSA-P256-SHA256 + OID registry + Bonferroni/TVD re-derivation)." ), formatter_class=argparse.RawDescriptionHelpFormatter, epilog=( "Exit codes are powers of two (OR-able):\n" f" {EXIT_OK} PASS | {EXIT_FAIL_SIGNATURE} sig FAIL | " f"{EXIT_FAIL_OID} oid FAIL | {EXIT_FAIL_STATS} stats FAIL | " f"{EXIT_FAIL_QASM} qasm FAIL | {EXIT_FAIL_OCSF} ocsf FAIL | " f"{EXIT_FAIL_LEDGER} ledger FAIL | {EXIT_INPUT_ERROR} input error" ), ) parent.add_argument("--version", action="version", version=f"qstrike-verify {VERSION}") sp = parent.add_subparsers(dest="cmd", required=True) def _bundle_args(p: argparse.ArgumentParser, with_key: bool = False) -> None: p.add_argument("--bundle", required=True, help="path or https:// URL to bundle JSON") if with_key: p.add_argument( "--key", help=( "path or https:// URL to verification-key.json " f"(default {DEFAULT_KEY_URL})" ), ) p.add_argument( "--tolerate-expired-keys", action="store_true", help=( "accept signatures whose attestation_chain.signed_at falls outside " "key.valid_from/valid_until (V5-001 opt-in escape hatch)" ), ) p.add_argument("--verbose", "-v", action="store_true", help="emit per-row detail") p.add_argument("--json", action="store_true", help="emit JSON report after table") p_sig = sp.add_parser("signature", help="verify bundle.attestation_chain.bundle_signature") _bundle_args(p_sig, with_key=True) p_sig.set_defaults(func=cmd_signature) p_oid = sp.add_parser("oid", help="verify crypto[] OID -> claimed algorithm") _bundle_args(p_oid) p_oid.set_defaults(func=cmd_oid) p_stats = sp.add_parser("stats", help="verify Bonferroni predicate + TVD identity") _bundle_args(p_stats) p_stats.set_defaults(func=cmd_stats) p_qasm = sp.add_parser("qasm", help="verify provider_jobs[].circuit_qasm basis-gate + counts") _bundle_args(p_qasm) p_qasm.set_defaults(func=cmd_qasm) p_ocsf = sp.add_parser("ocsf", help="verify OCSF 1.4 fields on securityEvents[]") _bundle_args(p_ocsf) p_ocsf.set_defaults(func=cmd_ocsf) p_ledger = sp.add_parser("ledger", help="verify sha256 well-formedness + chain on evidence_ledger[]") _bundle_args(p_ledger) p_ledger.set_defaults(func=cmd_ledger) p_all = sp.add_parser("all", help="run signature + oid + stats + qasm + ocsf + ledger") _bundle_args(p_all, with_key=True) p_all.set_defaults(func=cmd_all) p_info = sp.add_parser("info", help="print version, fingerprint, exit-code legend") p_info.set_defaults(func=cmd_info) return parent def main(argv: Optional[list[str]] = None) -> int: parser = build_parser() args = parser.parse_args(argv) if not hasattr(args, "func"): parser.print_help() return EXIT_INPUT_ERROR return args.func(args) if __name__ == "__main__": sys.exit(main())