#!/usr/bin/env python3 """Offline verifier for AttestLayer Buyer Review Pack ZIPs. The receipt does not carry or trust its own public key. Supply an independently obtained issuer JWKS snapshot published by the AttestLayer Registry with ``--jwks``. The verifier performs ZIP safety checks, validates every canonical manifest entry, validates request lineage, and verifies the Ed25519 receipt signature. Dependency: ``cryptography`` (Python 3.10+). """ from __future__ import annotations import argparse import base64 import hashlib import json import re import sys import unicodedata import zipfile from datetime import datetime, timedelta, timezone from pathlib import Path, PurePosixPath from typing import Any from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey try: from services.fulfillment.generated.brp_supported_input_contract import ( BRP_INPUT_CORPUS_VERSION, BRP_SUPPORTED_INPUT_CONTRACT, BRP_SUPPORTED_INPUT_CONTRACT_SHA256, BRP_SUPPORTED_INPUT_CONTRACT_VERSION, ) BRP_SUPPORTED_INPUT_CONTRACT_EFFECTIVE_DATE = str( BRP_SUPPORTED_INPUT_CONTRACT.get("effective_date") or "2026-08-18" ) except ImportError: # Standalone verifier copied outside the repository. BRP_SUPPORTED_INPUT_CONTRACT_VERSION = "2026-09-27.1" BRP_SUPPORTED_INPUT_CONTRACT_SHA256 = "a1a010b02a16feee21476cdd6b9f5b23abaa68b22a946fa22f73ecfd7bf1a25d" BRP_INPUT_CORPUS_VERSION = "V5-CORPUS-2026-08-18.1" BRP_SUPPORTED_INPUT_CONTRACT_EFFECTIVE_DATE = "2026-08-18" MAX_ZIP_BYTES = 120 * 1024 * 1024 MAX_ENTRIES = 200 MAX_ENTRY_BYTES = 110 * 1024 * 1024 MAX_TOTAL_BYTES = 250 * 1024 * 1024 MAX_COMPRESSION_RATIO = 100 SHA256 = re.compile(r"^[0-9a-f]{64}$") CHECKSUM_LINE = re.compile(r"^([0-9a-f]{64}) (.+)$") SOURCE_AUTHORIZATION_VERSION = "BRP-SOURCE-AUTH-1.0" SOURCE_AUTHORIZATION_TEXT = ( "I represent that my organization is authorized to use and share these source " "records for this buyer review, that the request reference identifies this buyer " "request, and that I have not uploaded prohibited information. I understand that " "AttestLayer records this representation but does not independently verify record " "ownership or sharing authority." ) CUSTOMER_REVIEW_PAIR = ( "BRP-CUSTOMER-REVIEW-MANIFEST-1.0", "BRP-CUSTOMER-REVIEW-RECEIPT-1.0", ) BUYER_FORWARDABLE_PAIR = ( "BRP-BUYER-FORWARDABLE-MANIFEST-1.0", "BRP-BUYER-FORWARDABLE-RECEIPT-1.0", ) FINALIZATION_TEXT = ( "I reviewed the selected responses and evidence. I authorize AttestLayer to " "create the buyer-forwardable package from the approved items. My organization " "remains responsible for deciding what to send to the buyer." ) # BEGIN GENERATED BRP CATALOG BINDING BRP_CATALOG_VERSION = '2026-08-12' BRP_CATALOG_SHA256 = '644223d04d29a485a988c76187fee571ce79bef081fbeb674dd63da86570272a' # END GENERATED BRP CATALOG BINDING REQUEST_LINEAGE_BASE_FIELDS = { "parser_version", "request_filename", "request_sha256", "format", "requirement_count", "requirements", } REQUEST_LINEAGE_V5_FIELDS = REQUEST_LINEAGE_BASE_FIELDS | { "input_contract_version", "input_contract_sha256", "input_corpus_version", } REQUEST_LINEAGE_REQUIREMENT_FIELDS = { "req_id", "text", "source_locator", "text_sha256", "source_row_sha256", "category", "requested_evidence", } # Copied from packages.common.brp_shared_contract so a buyer can run this file # with only cryptography installed. tests keep the two copies the same. _REQUIREMENT_ID_PUNCTUATION = "._/-*()" def _valid_requirement_id(value: Any) -> bool: """Use the same identifier rule as Fit Check and paid submission.""" if not isinstance(value, str): return False if unicodedata.normalize("NFC", value).strip() != value or not value or len(value) > 64: return False if unicodedata.category(value[0])[:1] not in {"L", "N"}: return False return all( unicodedata.category(character)[:1] in {"L", "M", "N"} or character in _REQUIREMENT_ID_PUNCTUATION for character in value ) SUPPORTED_REQUEST_PARSERS = { "BRP-REQUEST-PARSER-2.0", "BRP-REQUEST-PARSER-2.1", "BRP-REQUEST-PARSER-2.2", "BRP-REQUEST-PARSER-2.3", "BRP-REQUEST-PARSER-2.4", "BRP-REQUEST-PARSER-2.5", "BRP-REQUEST-PARSER-2.6", "BRP-REQUEST-PARSER-2.7", } SUPPORTED_REQUEST_FORMATS = {"csv", "tsv", "xlsx", "docx", "pdf", "txt", "md"} COMMERCE_BINDING_BASE_FIELDS = { "order_id_sha256", "order_number", "entitlement_id_sha256", "fit_id", "request_sha256", "offer_sku", "profile", "catalog_version", "catalog_sha256", "offer_snapshot_sha256", "policy_manifest_sha256", "purchase_date", } COMMERCE_BINDING_V5_FIELDS = COMMERCE_BINDING_BASE_FIELDS | { "input_contract_version", "input_contract_sha256", "input_corpus_version", } class VerificationError(ValueError): def __init__(self, code: str, detail: str = ""): super().__init__(f"{code}: {detail}" if detail else code) self.code = code self.detail = detail def _jcs(value: Any) -> str: """Canonical JSON for the BRP schema's string/int/list/object domain.""" if value is None: return "null" if value is True: return "true" if value is False: return "false" if isinstance(value, str): return json.dumps(value, ensure_ascii=False, separators=(",", ":")) if isinstance(value, int) and not isinstance(value, bool): return str(value) if isinstance(value, float): if not value == value or value in (float("inf"), float("-inf")): raise VerificationError("NON_FINITE_JSON_NUMBER") return str(int(value)) if value.is_integer() else repr(value) if isinstance(value, list): return "[" + ",".join(_jcs(item) for item in value) + "]" if isinstance(value, dict): return "{" + ",".join( json.dumps(str(key), ensure_ascii=False) + ":" + _jcs(value[key]) for key in sorted(value) ) + "}" raise VerificationError("UNSUPPORTED_JSON_TYPE", type(value).__name__) def _jcs_bytes(value: Any) -> bytes: return _jcs(value).encode("utf-8") def _sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _b64url(value: str) -> bytes: try: return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) except Exception as exc: raise VerificationError("INVALID_BASE64URL") from exc def _safe_name(name: str) -> bool: if not name or "\x00" in name or "\\" in name or name.startswith("/"): return False path = PurePosixPath(name) return not path.is_absolute() and all(part not in {"", ".", ".."} for part in path.parts) def _assert_zip_envelope(zip_path: Path) -> None: data = zip_path.read_bytes() start = max(0, len(data) - (0xFFFF + 22)) signature = b"PK\x05\x06" for offset in range(len(data) - 22, start - 1, -1): if data[offset:offset + 4] != signature: continue comment_length = int.from_bytes(data[offset + 20:offset + 22], "little") if offset + 22 + comment_length == len(data): return raise VerificationError("ZIP_TRAILING_DATA_OR_EOCD_INVALID") def _load_json(data: bytes, code: str) -> dict[str, Any]: try: parsed = json.loads(data.decode("utf-8")) except Exception as exc: raise VerificationError(code) from exc if not isinstance(parsed, dict): raise VerificationError(code, "expected an object") return parsed def _safe_members(zf: zipfile.ZipFile) -> dict[str, zipfile.ZipInfo]: infos = zf.infolist() if len(infos) > MAX_ENTRIES: raise VerificationError("ZIP_TOO_MANY_ENTRIES", str(len(infos))) out: dict[str, zipfile.ZipInfo] = {} folded: set[str] = set() total = 0 for info in infos: name = info.filename if info.is_dir(): continue if not _safe_name(name): raise VerificationError("ZIP_UNSAFE_PATH", name) canonical = unicodedata.normalize("NFC", name).casefold() if canonical in folded: raise VerificationError("ZIP_DUPLICATE_ENTRY", name) folded.add(canonical) if info.flag_bits & 0x1: raise VerificationError("ZIP_ENCRYPTED_ENTRY", name) if info.file_size > MAX_ENTRY_BYTES: raise VerificationError("ZIP_ENTRY_TOO_LARGE", name) total += info.file_size if total > MAX_TOTAL_BYTES: raise VerificationError("ZIP_EXPANDED_TOTAL_TOO_LARGE") if info.file_size > 1024 * 1024 and info.compress_size > 0: if info.file_size / info.compress_size > MAX_COMPRESSION_RATIO: raise VerificationError("ZIP_COMPRESSION_RATIO_TOO_HIGH", name) out[name] = info return out def _read(zf: zipfile.ZipFile, members: dict[str, zipfile.ZipInfo], name: str) -> bytes: info = members.get(name) if info is None: raise VerificationError("ZIP_MEMBER_MISSING", name) data = zf.read(info) if len(data) != info.file_size: raise VerificationError("ZIP_MEMBER_SIZE_MISMATCH", name) return data def _trusted_key(jwks: dict[str, Any], key_id: str) -> Ed25519PublicKey: keys = jwks.get("keys") if isinstance(jwks, dict) else None if not isinstance(keys, list) or not keys: raise VerificationError("JWKS_SHAPE_INVALID") seen_kids: set[str] = set() seen_public_keys: set[str] = set() active_count = 0 for candidate in keys: if not isinstance(candidate, dict): raise VerificationError("JWKS_KEY_NOT_ED25519") candidate_kid = candidate.get("kid") if ( not isinstance(candidate_kid, str) or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{2,127}", candidate_kid) is None or candidate_kid in seen_kids ): raise VerificationError("JWKS_KEY_NOT_FOUND_OR_AMBIGUOUS", str(candidate_kid or "")) if ( candidate.get("kty") != "OKP" or candidate.get("crv") != "Ed25519" or candidate.get("use") != "sig" or candidate.get("alg") != "EdDSA" or candidate.get("key_ops") != ["verify"] or candidate.get("status") not in {"active", "rotated", "revoked"} or "d" in candidate ): raise VerificationError("JWKS_KEY_NOT_ED25519", candidate_kid) raw = _b64url(str(candidate.get("x") or "")) if len(raw) != 32: raise VerificationError("JWKS_KEY_LENGTH_INVALID", candidate_kid) canonical_x = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") if canonical_x != candidate.get("x"): raise VerificationError("JWKS_KEY_ENCODING_INVALID", candidate_kid) if canonical_x in seen_public_keys: raise VerificationError("JWKS_KEY_NOT_FOUND_OR_AMBIGUOUS", candidate_kid) seen_kids.add(candidate_kid) seen_public_keys.add(canonical_x) if candidate.get("status") == "active": active_count += 1 if active_count != 1: raise VerificationError("JWKS_ACTIVE_KEY_COUNT_INVALID", str(active_count)) matches = [key for key in keys if key.get("kid") == key_id] if len(matches) != 1: raise VerificationError("JWKS_KEY_NOT_FOUND_OR_AMBIGUOUS", key_id) key = matches[0] if key.get("status") == "revoked": raise VerificationError("JWKS_KEY_REVOKED", key_id) raw = _b64url(str(key.get("x") or "")) return Ed25519PublicKey.from_public_bytes(raw) def _validate_request_lineage(lineage: Any) -> bool: """Validate one of the two explicit signed lineage generations. Legacy core packages used the six-field lineage. V5 adds exactly three immutable supported-input fields. Validate that closed union here so core packages remain inspectable; separated customer packages require V5 at the caller. Partial V5, misspelled, and injected fields always fail. Return True only for the V5 form. """ if not isinstance(lineage, dict): raise VerificationError("REQUEST_LINEAGE_INVALID") fields = set(lineage) is_v5 = fields == REQUEST_LINEAGE_V5_FIELDS if fields != REQUEST_LINEAGE_BASE_FIELDS and not is_v5: raise VerificationError("REQUEST_LINEAGE_INVALID", "field_set") if ( lineage.get("parser_version") not in SUPPORTED_REQUEST_PARSERS or not isinstance(lineage.get("request_filename"), str) or not lineage["request_filename"] or any(unicodedata.category(char) in {"Cc", "Cf"} for char in lineage["request_filename"]) or lineage.get("format") not in SUPPORTED_REQUEST_FORMATS or not SHA256.fullmatch(str(lineage.get("request_sha256") or "")) or not isinstance(lineage.get("requirement_count"), int) or isinstance(lineage.get("requirement_count"), bool) or not (1 <= lineage["requirement_count"] <= 300) ): raise VerificationError("REQUEST_LINEAGE_INVALID", "metadata") if is_v5 and ( lineage.get("input_contract_version") != BRP_SUPPORTED_INPUT_CONTRACT_VERSION or lineage.get("input_contract_sha256") != BRP_SUPPORTED_INPUT_CONTRACT_SHA256 or lineage.get("input_corpus_version") != BRP_INPUT_CORPUS_VERSION ): raise VerificationError("REQUEST_LINEAGE_INVALID", "v5_contract_binding") requirements = lineage.get("requirements") if ( not isinstance(requirements, list) or len(requirements) != lineage["requirement_count"] ): raise VerificationError("REQUEST_LINEAGE_COUNT_INVALID") ids: set[str] = set() for index, requirement in enumerate(requirements): if not isinstance(requirement, dict) or set(requirement) != REQUEST_LINEAGE_REQUIREMENT_FIELDS: raise VerificationError("REQUEST_LINEAGE_ENTRY_INVALID", str(index)) req_id = requirement.get("req_id") text = unicodedata.normalize("NFC", str(requirement.get("text") or "")).strip() folded_id = req_id.casefold() if isinstance(req_id, str) else "" source_row_hash = str(requirement.get("source_row_sha256") or "") locator = str(requirement.get("source_locator") or "") identity = f"{folded_id}\n{locator}" if ( not _valid_requirement_id(req_id) or identity in ids or not text or text != requirement.get("text") or not isinstance(requirement.get("source_locator"), str) or not requirement["source_locator"] or not isinstance(requirement.get("category"), str) or not isinstance(requirement.get("requested_evidence"), str) or _sha256(text.encode("utf-8")) != requirement.get("text_sha256") or (source_row_hash != "" and not SHA256.fullmatch(source_row_hash)) ): raise VerificationError( "REQUEST_LINEAGE_ENTRY_INVALID", req_id if isinstance(req_id, str) and req_id else str(index), ) ids.add(identity) return is_v5 def _version_effective_at(value: str, code: str) -> datetime: match = re.fullmatch(r"(\d{4}-\d{2}-\d{2})(?:[.-][A-Za-z0-9._-]+)?", value) if not match: raise VerificationError(code) try: return datetime.fromisoformat(match.group(1)).replace(tzinfo=timezone.utc) except ValueError: raise VerificationError(code) from None def _malware_signature_timestamp(value: str) -> datetime | None: synthetic = re.search(r"synthetic-sample-(\d{4}-\d{2}-\d{2})", value) if synthetic: return datetime.fromisoformat(synthetic.group(1)).replace(tzinfo=timezone.utc) if "/" in value: try: return datetime.strptime( value.split("/", 1)[1].strip(), "%a %b %d %H:%M:%S %Y", ).replace(tzinfo=timezone.utc) except ValueError: pass return None def _validate_separated_source_authorization( manifest: dict[str, Any], receipt: dict[str, Any], lineage: dict[str, Any], authorized_package_identifier: str, *, buyer_forwardable: bool, ) -> None: authorization = manifest.get("source_authorization") expected_fields = { "version", "text", "accepted", "accepted_at", "order_id_sha256", "request_identity", "request_reference_sha256", "source_set_sha256", "supplier_name_sha256", "package_id_sha256", } rip_profile = manifest.get("profile_identifier") == "RIP-1.0" if rip_profile: expected_fields.add("completed_response_sha256") if not isinstance(authorization, dict) or set(authorization) != expected_fields: raise VerificationError("SOURCE_AUTHORIZATION_INVALID") if ( authorization.get("version") != SOURCE_AUTHORIZATION_VERSION or authorization.get("text") != SOURCE_AUTHORIZATION_TEXT or authorization.get("accepted") is not True or not isinstance(authorization.get("accepted_at"), str) or not authorization.get("accepted_at") ): raise VerificationError("SOURCE_AUTHORIZATION_INVALID") try: accepted_at = datetime.fromisoformat( str(authorization["accepted_at"]).replace("Z", "+00:00") ).astimezone(timezone.utc) generated_at = datetime.fromisoformat( str(manifest.get("generated_at") or "").replace("Z", "+00:00") ).astimezone(timezone.utc) except (TypeError, ValueError): raise VerificationError("SOURCE_AUTHORIZATION_TIMESTAMP_INVALID") from None if accepted_at > generated_at + timedelta(minutes=1) or ( not buyer_forwardable and accepted_at < generated_at - timedelta(minutes=30) ): raise VerificationError("SOURCE_AUTHORIZATION_TIMESTAMP_INVALID") for field in ( "order_id_sha256", "request_identity", "request_reference_sha256", "source_set_sha256", "supplier_name_sha256", "package_id_sha256", ): if not SHA256.fullmatch(str(authorization.get(field) or "")): raise VerificationError("SOURCE_AUTHORIZATION_INVALID", field) completed_hash = str(authorization.get("completed_response_sha256") or "") if (rip_profile and not SHA256.fullmatch(completed_hash)) or ( not rip_profile and completed_hash ): raise VerificationError("SOURCE_AUTHORIZATION_INVALID", "completed_response_sha256") identity_material = "|".join([ re.sub(r"\s+", " ", unicodedata.normalize("NFC", str(manifest.get("buyer_name") or "")).strip()), re.sub(r"\s+", " ", unicodedata.normalize("NFC", str(manifest.get("request_type") or "")).strip()), str(lineage.get("request_sha256") or ""), ]) if _sha256(identity_material.encode("utf-8")) != authorization.get("request_identity"): raise VerificationError("SOURCE_AUTHORIZATION_REQUEST_MISMATCH") supplier = re.sub( r"\s+", " ", unicodedata.normalize("NFC", str(manifest.get("supplier_name") or "")).strip() ) if _sha256(supplier.encode("utf-8")) != authorization.get("supplier_name_sha256"): raise VerificationError("SOURCE_AUTHORIZATION_SUPPLIER_MISMATCH") if _sha256(authorized_package_identifier.encode("utf-8")) != authorization.get("package_id_sha256"): raise VerificationError("SOURCE_AUTHORIZATION_PACKAGE_MISMATCH") if receipt.get("source_authorization_sha256") != _sha256(_jcs_bytes(authorization)): raise VerificationError("SOURCE_AUTHORIZATION_DIGEST_MISMATCH") def verify(zip_path: Path, jwks_path: Path, expected_jwks_sha256: str = "") -> dict[str, Any]: if zip_path.stat().st_size > MAX_ZIP_BYTES: raise VerificationError("ZIP_TOO_LARGE", str(zip_path.stat().st_size)) _assert_zip_envelope(zip_path) jwks_bytes = jwks_path.read_bytes() if expected_jwks_sha256: expected = expected_jwks_sha256.casefold() if not SHA256.fullmatch(expected) or _sha256(jwks_bytes) != expected: raise VerificationError("JWKS_SNAPSHOT_HASH_MISMATCH") jwks = _load_json(jwks_bytes, "JWKS_JSON_INVALID") with zipfile.ZipFile(zip_path, "r") as zf: members = _safe_members(zf) if "08-canonical-manifest.json" in members or "09-signed-receipt.json" in members: manifest_name = "08-canonical-manifest.json" receipt_name = "09-signed-receipt.json" elif "06-canonical-manifest.json" in members: manifest_name = "06-canonical-manifest.json" receipt_name = "07-signed-receipt.json" else: manifest_name = "06-manifest.json" receipt_name = "07-signed-receipt.json" manifest_bytes = _read(zf, members, manifest_name) receipt_bytes = _read(zf, members, receipt_name) manifest = _load_json(manifest_bytes, "MANIFEST_JSON_INVALID") receipt = _load_json(receipt_bytes, "RECEIPT_JSON_INVALID") schema_pair = ( manifest.get("schema_version"), receipt.get("receipt_version"), ) if schema_pair not in { ("BRP-MANIFEST-1.1", "BRP-RECEIPT-1.0"), ("BRP-MANIFEST-1.2", "BRP-RECEIPT-1.1"), ("BRP-MANIFEST-1.3", "BRP-RECEIPT-1.2"), CUSTOMER_REVIEW_PAIR, BUYER_FORWARDABLE_PAIR, }: raise VerificationError("MANIFEST_SCHEMA_UNSUPPORTED", str(manifest.get("schema_version"))) is_authorization_schema = schema_pair in { ("BRP-MANIFEST-1.2", "BRP-RECEIPT-1.1"), ("BRP-MANIFEST-1.3", "BRP-RECEIPT-1.2"), } is_rip_schema = schema_pair == ("BRP-MANIFEST-1.3", "BRP-RECEIPT-1.2") is_customer_review_schema = schema_pair == CUSTOMER_REVIEW_PAIR is_buyer_forwardable_schema = schema_pair == BUYER_FORWARDABLE_PAIR is_separated_schema = is_customer_review_schema or is_buyer_forwardable_schema manifest_fields = { "schema_version", "package_identifier", "profile_identifier", "supplier_name", "buyer_name", "request_type", "buyer_due_date", "generated_at", "files", "request_lineage", "evidence", } receipt_fields = { "receipt_version", "key_id", "package_identifier", "profile_identifier", "manifest_sha256", "generated_at", "verification_url", "request_sha256", "requirement_count", "signature", } if is_separated_schema: manifest_fields = { "schema_version", "package_identifier", "package_stage", "profile_identifier", "issuer_key_id", "supplier_name", "buyer_name", "request_type", "buyer_due_date", "generated_at", "files", "request_lineage", "source_authorization", "commerce_binding", "statement_origins", "evidence", "predecessor_manifest_sha256", "customer_finalized_at", "engine_commit", "engine_build_time", "ai_models", "inference_geo", "prompt_hashes", "rules_version", } if is_buyer_forwardable_schema: manifest_fields |= { "customer_decisions", "finalization_authorization_sha256", "source_review_package_identifier", } receipt_fields = { "receipt_version", "key_id", "package_identifier", "package_stage", "profile_identifier", "manifest_sha256", "generated_at", "verification_url", "request_sha256", "requirement_count", "source_authorization_sha256", "commerce_binding_sha256", "predecessor_manifest_sha256", "customer_finalized_at", "signature", } elif is_authorization_schema: manifest_fields.add("source_authorization") manifest_fields.add("supplier_identity_basis") receipt_fields.add("source_authorization_sha256") if is_rip_schema: manifest_fields.add("response_integrity") receipt_fields.add("completed_response_sha256") # For the separated schema, validate the signed request lineage before # the outer manifest field set so chronology defects are not masked by # a secondary envelope mismatch. lineage = manifest.get("request_lineage") lineage_is_v5 = _validate_request_lineage(lineage) if is_separated_schema and not lineage_is_v5: raise VerificationError("REQUEST_LINEAGE_INVALID", "v5_contract_required") if is_separated_schema and lineage_is_v5: try: generated_at = datetime.fromisoformat( str(manifest.get("generated_at") or "").replace("Z", "+00:00") ).astimezone(timezone.utc) except ValueError: raise VerificationError("PACKAGE_GENERATED_AT_INVALID") from None if _version_effective_at( BRP_SUPPORTED_INPUT_CONTRACT_EFFECTIVE_DATE, "REQUEST_LINEAGE_CONTRACT_VERSION_INVALID", ) > generated_at: raise VerificationError("REQUEST_LINEAGE_CHRONOLOGY_INVALID") if set(manifest) != manifest_fields: raise VerificationError("MANIFEST_FIELD_SET_INVALID") if set(receipt) != receipt_fields: raise VerificationError("RECEIPT_FIELD_SET_INVALID") if manifest_bytes != _jcs_bytes(manifest): raise VerificationError("MANIFEST_NOT_CANONICAL_JCS") files = manifest.get("files") if not isinstance(files, list) or not files: raise VerificationError("MANIFEST_FILES_INVALID") declared: set[str] = set() for entry in files: if not isinstance(entry, dict): raise VerificationError("MANIFEST_FILE_INVALID") name = str(entry.get("path") or "") expected_hash = str(entry.get("sha256") or "").casefold() expected_size = entry.get("size_bytes") if not _safe_name(name) or name in declared: raise VerificationError("MANIFEST_FILE_PATH_INVALID", name) if name in { manifest_name, receipt_name, "11-checksums.txt" if is_customer_review_schema else "09-checksums.txt" if is_buyer_forwardable_schema else "checksums.sha256", }: raise VerificationError("MANIFEST_SELF_REFERENCE_INVALID", name) if is_separated_schema and set(entry) != {"path", "sha256", "size_bytes", "role"}: raise VerificationError("MANIFEST_FILE_METADATA_INVALID", name) if is_separated_schema and entry.get("role") != "package_payload": raise VerificationError("MANIFEST_FILE_METADATA_INVALID", name) if not SHA256.fullmatch(expected_hash) or not isinstance(expected_size, int) or expected_size < 0: raise VerificationError("MANIFEST_FILE_METADATA_INVALID", name) data = _read(zf, members, name) if len(data) != expected_size or _sha256(data) != expected_hash: raise VerificationError("MANIFEST_FILE_HASH_OR_SIZE_MISMATCH", name) declared.add(name) checksum_name = ( "11-checksums.txt" if is_customer_review_schema else "09-checksums.txt" if is_buyer_forwardable_schema else "checksums.sha256" ) required_derived = {manifest_name, receipt_name, checksum_name} actual_names = set(members) expected_names = declared | required_derived extras = sorted(actual_names - expected_names) if extras: raise VerificationError("ZIP_UNDECLARED_ENTRY", extras[0]) missing = sorted(expected_names - actual_names) if missing: raise VerificationError("ZIP_MEMBER_MISSING", missing[0]) checksum_bytes = _read(zf, members, checksum_name) try: checksum_text = checksum_bytes.decode("utf-8") except UnicodeDecodeError as exc: raise VerificationError("CHECKSUM_INDEX_INVALID_UTF8") from exc checksum_entries: dict[str, str] = {} checksum_folded: set[str] = set() for line in checksum_text.splitlines(): match = CHECKSUM_LINE.fullmatch(line) if not match: raise VerificationError("CHECKSUM_INDEX_LINE_INVALID") expected_hash, name = match.groups() folded = unicodedata.normalize("NFC", name).casefold() if ( not _safe_name(name) or name == checksum_name or folded in checksum_folded ): raise VerificationError("CHECKSUM_INDEX_PATH_INVALID", name) checksum_folded.add(folded) checksum_entries[name] = expected_hash expected_checksum_names = actual_names - {checksum_name} if set(checksum_entries) != expected_checksum_names: raise VerificationError("CHECKSUM_INDEX_MEMBERSHIP_MISMATCH") for name, expected_hash in checksum_entries.items(): if _sha256(_read(zf, members, name)) != expected_hash: raise VerificationError("CHECKSUM_INDEX_HASH_MISMATCH", name) requirements = lineage["requirements"] if is_separated_schema: expected_stage = "customer_review" if is_customer_review_schema else "buyer_forwardable" if ( manifest.get("package_stage") != expected_stage or receipt.get("package_stage") != expected_stage or receipt.get("profile_identifier") != manifest.get("profile_identifier") or receipt.get("generated_at") != manifest.get("generated_at") or receipt.get("predecessor_manifest_sha256") != manifest.get("predecessor_manifest_sha256") or receipt.get("customer_finalized_at") != manifest.get("customer_finalized_at") or receipt.get("key_id") != manifest.get("issuer_key_id") ): raise VerificationError("PACKAGE_STAGE_LINEAGE_MISMATCH") commerce = manifest.get("commerce_binding") commerce_fields = set(commerce) if isinstance(commerce, dict) else set() commerce_is_v5 = commerce_fields == COMMERCE_BINDING_V5_FIELDS try: purchase_at = datetime.fromisoformat( str((commerce or {}).get("purchase_date") or "") + "T00:00:00+00:00" ) except (AttributeError, ValueError): purchase_at = None if ( not isinstance(commerce, dict) or not commerce_is_v5 or any( not SHA256.fullmatch(str(commerce.get(field) or "")) for field in ( "order_id_sha256", "entitlement_id_sha256", "request_sha256", "catalog_sha256", "offer_snapshot_sha256", "policy_manifest_sha256", ) ) or commerce.get("order_id_sha256") != (manifest.get("source_authorization") or {}).get("order_id_sha256") or commerce.get("request_sha256") != lineage.get("request_sha256") or commerce.get("profile") != manifest.get("profile_identifier") or commerce.get("catalog_version") != BRP_CATALOG_VERSION or commerce.get("catalog_sha256") not in { BRP_CATALOG_SHA256, "8bd747c9880378c6ceb48ff986fb43802e54b664e9df330c2a7ad2b49d1d4fce", "8f0f736486291d401ec6770cb647a49a58b7cca6a27cf48f97ff7f57fd269c6f", } or not re.fullmatch(r"BRP-[A-F0-9]{8}", str(commerce.get("order_number") or "")) or not re.fullmatch(r"[a-f0-9-]{36}", str(commerce.get("fit_id") or "")) or not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", str(commerce.get("purchase_date") or "")) or receipt.get("commerce_binding_sha256") != _sha256(_jcs_bytes(commerce)) or commerce.get("input_contract_version") != BRP_SUPPORTED_INPUT_CONTRACT_VERSION or commerce.get("input_contract_sha256") != BRP_SUPPORTED_INPUT_CONTRACT_SHA256 or commerce.get("input_corpus_version") != BRP_INPUT_CORPUS_VERSION or not lineage_is_v5 or commerce.get("input_contract_version") != lineage.get("input_contract_version") or commerce.get("input_contract_sha256") != lineage.get("input_contract_sha256") or commerce.get("input_corpus_version") != lineage.get("input_corpus_version") or purchase_at is None or _version_effective_at( BRP_CATALOG_VERSION, "COMMERCE_CATALOG_VERSION_INVALID", ) > purchase_at or purchase_at > generated_at ): raise VerificationError("COMMERCE_BINDING_INVALID") statement_origins = manifest.get("statement_origins") if not isinstance(statement_origins, list) or not statement_origins: raise VerificationError("STATEMENT_ORIGINS_INVALID") seen_statement_ids: set[str] = set() for statement in statement_origins: if ( not isinstance(statement, dict) or set(statement) != {"requirement_id", "origin", "status"} or statement.get("requirement_id") not in { str(item.get("req_id") or "") for item in requirements if isinstance(item, dict) } or statement.get("requirement_id") in seen_statement_ids or statement.get("origin") not in { "attestlayer_generated", "customer_confirmed", "customer_supplied", } or statement.get("status") not in { "Supported", "Customer confirmation required", "Not supported by submitted records", "Customer-confirmed", "Customer-supplied", "Confirmed", "Contradicted", "Unsupported", "Missing answer", "Not checked", } ): raise VerificationError("STATEMENT_ORIGINS_INVALID") seen_statement_ids.add(statement["requirement_id"]) evidence = manifest.get("evidence") if not isinstance(evidence, list): raise VerificationError("MANIFEST_EVIDENCE_INVALID") evidence_names: set[str] = set() for item in evidence: if ( not isinstance(item, dict) or set(item) != { "original_name", "safe_package_name", "sha256", "media_type", "size_bytes", "access_mode", "source_record_id", "malware_scan_status", "malware_scan", "customer_authorized_at", "included_in_buyer_pack", } or not isinstance(item.get("safe_package_name"), str) or not _safe_name(item["safe_package_name"]) or "/" in item["safe_package_name"] or item["safe_package_name"].casefold() in evidence_names or not SHA256.fullmatch(str(item.get("sha256") or "")) or not isinstance(item.get("size_bytes"), int) or item["size_bytes"] < 0 or item.get("access_mode") not in {"buyer_forwardable", "internal_only"} or not isinstance(item.get("included_in_buyer_pack"), bool) or item.get("malware_scan_status") != "clean" ): raise VerificationError("MANIFEST_EVIDENCE_INVALID") malware_scan = item.get("malware_scan") if ( not isinstance(malware_scan, dict) or set(malware_scan) != { "status", "scanner", "engine_version", "signature_version", "scanned_at", "sha256", } or malware_scan.get("status") != "clean" or malware_scan.get("scanner") != "clamav" or not isinstance(malware_scan.get("engine_version"), str) or not (1 <= len(malware_scan["engine_version"]) <= 64) or not isinstance(malware_scan.get("signature_version"), str) or not (1 <= len(malware_scan["signature_version"]) <= 128) or not re.fullmatch( r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z", str(malware_scan.get("scanned_at") or ""), ) or not SHA256.fullmatch(str(malware_scan.get("sha256") or "")) or malware_scan.get("sha256") != item.get("sha256") ): raise VerificationError("MANIFEST_EVIDENCE_MALWARE_SCAN_INVALID") try: scanned_at = datetime.fromisoformat( str(malware_scan["scanned_at"]).replace("Z", "+00:00") ).astimezone(timezone.utc) generated_at = datetime.fromisoformat( str(manifest.get("generated_at") or "").replace("Z", "+00:00") ).astimezone(timezone.utc) except (TypeError, ValueError): raise VerificationError( "MANIFEST_EVIDENCE_MALWARE_SCAN_INVALID" ) from None signature_at = _malware_signature_timestamp( str(malware_scan.get("signature_version") or "") ) if scanned_at > generated_at or ( signature_at is not None and signature_at > scanned_at ): raise VerificationError("MANIFEST_EVIDENCE_SCAN_CHRONOLOGY_INVALID") evidence_names.add(item["safe_package_name"].casefold()) evidence_path = f"buyer_forwardable_evidence/{item['safe_package_name']}" included = item.get("included_in_buyer_pack") is True if included != (evidence_path in declared): raise VerificationError("MANIFEST_EVIDENCE_BINDING_MISMATCH", evidence_path) if is_buyer_forwardable_schema and ( item.get("access_mode") != "buyer_forwardable" or not included ): raise VerificationError("BUYER_FORWARDABLE_EVIDENCE_INVALID", evidence_path) package_identifier = str(manifest.get("package_identifier") or "") if is_customer_review_schema: if ( manifest.get("predecessor_manifest_sha256") != "" or manifest.get("customer_finalized_at") != "" or set(seen_statement_ids) != { str(item.get("req_id") or "") for item in requirements if isinstance(item, dict) } ): raise VerificationError("CUSTOMER_REVIEW_LINEAGE_INVALID") authorized_package_identifier = package_identifier else: source_review_id = str(manifest.get("source_review_package_identifier") or "") decisions = manifest.get("customer_decisions") if ( package_identifier != f"{source_review_id}-F1" or not SHA256.fullmatch(str(manifest.get("predecessor_manifest_sha256") or "")) or not isinstance(manifest.get("customer_finalized_at"), str) or manifest.get("customer_finalized_at") != manifest.get("generated_at") or manifest.get("finalization_authorization_sha256") != _sha256(FINALIZATION_TEXT.encode()) or not isinstance(decisions, list) or not decisions ): raise VerificationError("BUYER_FORWARDABLE_LINEAGE_INVALID") decision_ids: set[str] = set() for decision in decisions: if ( not isinstance(decision, dict) or set(decision) != {"requirement_id", "action", "customer_text_sha256"} or decision.get("requirement_id") in decision_ids or decision.get("requirement_id") not in { str(item.get("req_id") or "") for item in requirements if isinstance(item, dict) } or decision.get("action") not in {"approve", "exclude", "confirm", "replace"} or ( decision.get("customer_text_sha256") != "" and not SHA256.fullmatch(str(decision.get("customer_text_sha256") or "")) ) ): raise VerificationError("CUSTOMER_DECISIONS_INVALID") decision_ids.add(decision["requirement_id"]) if decision_ids != { str(item.get("req_id") or "") for item in requirements if isinstance(item, dict) }: raise VerificationError("CUSTOMER_DECISIONS_INCOMPLETE") authorized_package_identifier = source_review_id _validate_separated_source_authorization( manifest, receipt, lineage, authorized_package_identifier, buyer_forwardable=is_buyer_forwardable_schema, ) if is_authorization_schema: if manifest.get("supplier_identity_basis") != "customer_declared_paid_order": raise VerificationError("SUPPLIER_IDENTITY_BASIS_INVALID") authorization = manifest.get("source_authorization") authorization_fields = { "version", "text", "accepted", "accepted_at", "order_id_sha256", "request_identity", "request_reference_sha256", "source_set_sha256", "supplier_name_sha256", "package_id_sha256", } rip_authorization_fields = authorization_fields | {"completed_response_sha256"} if ( not isinstance(authorization, dict) or set(authorization) not in (authorization_fields, rip_authorization_fields) or (is_rip_schema and set(authorization) != rip_authorization_fields) ): raise VerificationError("SOURCE_AUTHORIZATION_INVALID") if ( authorization.get("version") != SOURCE_AUTHORIZATION_VERSION or authorization.get("text") != SOURCE_AUTHORIZATION_TEXT or authorization.get("accepted") is not True or not isinstance(authorization.get("accepted_at"), str) or not authorization.get("accepted_at") ): raise VerificationError("SOURCE_AUTHORIZATION_INVALID") for field in ( "order_id_sha256", "request_identity", "request_reference_sha256", "source_set_sha256", "supplier_name_sha256", "package_id_sha256", ): if not SHA256.fullmatch(str(authorization.get(field) or "")): raise VerificationError("SOURCE_AUTHORIZATION_INVALID", field) try: accepted_at = datetime.fromisoformat( str(authorization["accepted_at"]).replace("Z", "+00:00") ) generated_at = datetime.fromisoformat( str(manifest.get("generated_at") or "").replace("Z", "+00:00") ) if accepted_at.tzinfo is None or generated_at.tzinfo is None: raise ValueError accepted_at = accepted_at.astimezone(timezone.utc) generated_at = generated_at.astimezone(timezone.utc) if not ( generated_at - timedelta(minutes=30) <= accepted_at <= generated_at + timedelta(minutes=1) ): raise ValueError except (TypeError, ValueError): raise VerificationError("SOURCE_AUTHORIZATION_TIMESTAMP_INVALID") from None identity_material = "|".join([ re.sub(r"\s+", " ", unicodedata.normalize("NFC", str(manifest.get("buyer_name") or "")).strip()), re.sub(r"\s+", " ", unicodedata.normalize("NFC", str(manifest.get("request_type") or "")).strip()), str(lineage.get("request_sha256") or ""), ]) if hashlib.sha256(identity_material.encode("utf-8")).hexdigest() != authorization.get("request_identity"): raise VerificationError("SOURCE_AUTHORIZATION_REQUEST_MISMATCH") normalized_supplier = re.sub( r"\s+", " ", unicodedata.normalize( "NFC", str(manifest.get("supplier_name") or "") ).strip() ) if hashlib.sha256(normalized_supplier.encode("utf-8")).hexdigest() != authorization.get("supplier_name_sha256"): raise VerificationError("SOURCE_AUTHORIZATION_SUPPLIER_MISMATCH") if hashlib.sha256( str(manifest.get("package_identifier") or "").encode("utf-8") ).hexdigest() != authorization.get("package_id_sha256"): raise VerificationError("SOURCE_AUTHORIZATION_PACKAGE_MISMATCH") authorization_digest = _sha256(_jcs_bytes(authorization)) if receipt.get("source_authorization_sha256") != authorization_digest: raise VerificationError("SOURCE_AUTHORIZATION_DIGEST_MISMATCH") completed_response_sha256 = str(authorization.get("completed_response_sha256") or "") if completed_response_sha256 and not SHA256.fullmatch(completed_response_sha256): raise VerificationError("SOURCE_AUTHORIZATION_INVALID", "completed_response_sha256") if is_rip_schema: analysis = manifest.get("response_integrity") expected_analysis_fields = { "schema_version", "completed_response_sha256", "parser_version", "original_requirement_count", "completed_response_entry_count", "linked_entry_count", "unlinked_entry_count", "all_claims_linked", "answers_rewritten", "truth_verified", "findings", } if ( manifest.get("profile_identifier") != "RIP-1.0" or not isinstance(analysis, dict) or set(analysis) != expected_analysis_fields or analysis.get("schema_version") != "RIP-INTEGRITY-1.0" or analysis.get("parser_version") not in { "BRP-REQUEST-PARSER-2.0", "BRP-REQUEST-PARSER-2.1", "BRP-REQUEST-PARSER-2.2", "BRP-REQUEST-PARSER-2.3", "BRP-REQUEST-PARSER-2.4", "BRP-REQUEST-PARSER-2.5", "BRP-REQUEST-PARSER-2.6", "BRP-REQUEST-PARSER-2.7", } or analysis.get("answers_rewritten") is not False or analysis.get("truth_verified") is not False or analysis.get("completed_response_sha256") != completed_response_sha256 or receipt.get("completed_response_sha256") != completed_response_sha256 or analysis.get("original_requirement_count") != lineage.get("requirement_count") ): raise VerificationError("RESPONSE_INTEGRITY_INVALID") findings = analysis.get("findings") if not isinstance(findings, list) or not findings: raise VerificationError("RESPONSE_INTEGRITY_INVALID") linked = 0 for finding in findings: if not isinstance(finding, dict) or set(finding) != { "requirement_id", "status", "response_text_sha256", "source_file_sha256", "claim_count", "unlinked_claim_sha256s", "reason", }: raise VerificationError("RESPONSE_INTEGRITY_FINDING_INVALID") status = finding.get("status") if status not in { "LINKED", "UNLINKED", "CONTRADICTED", "MISSING_RESPONSE", "UNLINKED_EXTRA_RESPONSE", }: raise VerificationError("RESPONSE_INTEGRITY_FINDING_INVALID") response_hash = str(finding.get("response_text_sha256") or "") source_hash = str(finding.get("source_file_sha256") or "") unlinked_hashes = finding.get("unlinked_claim_sha256s") if ( not str(finding.get("requirement_id") or "") or (response_hash and not SHA256.fullmatch(response_hash)) or (source_hash and not SHA256.fullmatch(source_hash)) or not isinstance(finding.get("claim_count"), int) or finding["claim_count"] < 0 or not isinstance(unlinked_hashes, list) or any(not SHA256.fullmatch(str(value or "")) for value in unlinked_hashes) or not str(finding.get("reason") or "") ): raise VerificationError("RESPONSE_INTEGRITY_FINDING_INVALID") if status == "LINKED": linked += 1 if not response_hash or not source_hash or unlinked_hashes: raise VerificationError("RESPONSE_INTEGRITY_FINDING_INVALID") if ( analysis.get("linked_entry_count") != linked or analysis.get("unlinked_entry_count") != len(findings) - linked or analysis.get("all_claims_linked") is not (linked == len(findings)) ): raise VerificationError("RESPONSE_INTEGRITY_COUNT_MISMATCH") manifest_hash = _sha256(manifest_bytes) if receipt.get("manifest_sha256") != manifest_hash: raise VerificationError("RECEIPT_MANIFEST_HASH_MISMATCH") if receipt.get("package_identifier") != manifest.get("package_identifier"): raise VerificationError("PACKAGE_IDENTIFIER_MISMATCH") if receipt.get("request_sha256") != lineage.get("request_sha256"): raise VerificationError("RECEIPT_REQUEST_HASH_MISMATCH") if receipt.get("requirement_count") != lineage.get("requirement_count"): raise VerificationError("RECEIPT_REQUIREMENT_COUNT_MISMATCH") signature = receipt.get("signature") if ( not isinstance(signature, dict) or set(signature) != {"algorithm", "signature_payload", "value"} or signature.get("algorithm") != "Ed25519" ): raise VerificationError("RECEIPT_SIGNATURE_METADATA_INVALID") if signature.get("signature_payload") != "JCS(receipt without signature)": raise VerificationError("RECEIPT_SIGNATURE_PAYLOAD_INVALID") unsigned = {key: value for key, value in receipt.items() if key != "signature"} public_key = _trusted_key(jwks, str(receipt.get("key_id") or "")) try: signature_bytes = _b64url(str(signature.get("value") or "")) if len(signature_bytes) != 64: raise VerificationError("RECEIPT_SIGNATURE_LENGTH_INVALID") public_key.verify(signature_bytes, _jcs_bytes(unsigned)) except Exception as exc: raise VerificationError("RECEIPT_SIGNATURE_INVALID") from exc matched = next( (item for item in (jwks.get("keys") or []) if isinstance(item, dict) and item.get("kid") == receipt.get("key_id")), {}, ) test_issuer = ( matched.get("environment") == "test" or str(matched.get("kid") or "").startswith("attestlayer-brp-testmode-") or str(matched.get("kid") or "") in { "attestlayer-brp-issuer-2026-09", "attestlayer-brp-testmode-2026-07", } ) result = { "verdict": "PASS", "package_identifier": manifest.get("package_identifier"), "manifest_sha256": manifest_hash, "request_sha256": lineage.get("request_sha256"), "requirement_count": lineage.get("requirement_count"), "verified_files": len(files), "key_id": receipt.get("key_id"), "package_stage": manifest.get("package_stage", "legacy_completed"), "supplier_identity_basis": manifest.get( "supplier_identity_basis", "customer_declared_paid_order" if is_separated_schema else "legacy_undisclosed", ), "supplier_identity_notice": ( "Customer-declared from the paid order; not KYC or legal-entity verification." if is_authorization_schema or is_separated_schema else "Legacy package: supplier identity basis was not signed in this schema." ), } if test_issuer: result["issuer_environment"] = "test" result["message"] = ( "PACKAGE INTEGRITY VERIFIED — TEST ISSUER. " "The files are unchanged and were signed by AttestLayer's test environment. " "This is not a production-issued package." ) return result def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Verify an AttestLayer Buyer Review Pack offline") parser.add_argument("package", type=Path, help="buyer-review-pack.zip") parser.add_argument( "--jwks", type=Path, required=True, help="trusted issuer JWKS snapshot published by the AttestLayer Registry", ) parser.add_argument("--expected-jwks-sha256", default="", help="optional pinned SHA-256 of the JWKS bytes") args = parser.parse_args(argv) try: result = verify(args.package, args.jwks, args.expected_jwks_sha256) except (OSError, zipfile.BadZipFile, VerificationError) as exc: code = exc.code if isinstance(exc, VerificationError) else type(exc).__name__.upper() print(json.dumps({"verdict": "FAIL", "reason_code": code, "detail": str(exc)}, indent=2)) return 1 print(json.dumps(result, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())