From 4a7decbaa5cd7a7709d310b38c61e5496368f1b3 Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Tue, 21 Jul 2026 12:45:51 -0400 Subject: [PATCH 1/2] fix(schema): accept valid SD-JWT serializations in checkout_mandate pattern The checkout_mandate pattern could not match any serialization that is valid for the format the field is documented to carry: - RFC 9901 compact SD-JWT ends with '~' when no KB-JWT is present; the pattern's (~[A-Za-z0-9_-]+)* groups forbid a trailing tilde. - An SD-JWT with Key Binding ends in a KB-JWT, which contains dots; the tilde-segment character class forbids dots, so every SD-JWT+KB was rejected. - Delegated presentations (as emitted by the AP2 reference implementation) join tokens with '~~', which the pattern also rejects. The only strings the old pattern accepted (a bare compact JWT, or dot-free tilde segments without the trailing tilde) are not valid RFC 9901 serializations. CI never caught this because the complete-request example validates against the base checkout schema, which does not exercise the extension's pattern. The new pattern admits: compact SD-JWT with or without KB-JWT, and '~~' delegate chains of such tokens; it is a strict superset of the old pattern, so no previously-valid payload breaks. Also documents the accepted wire forms in ap2-mandates.md. Fixes #599 (option 2). --- docs/specification/ap2-mandates.md | 9 +++++++++ source/schemas/shopping/ap2_mandate.json | 4 ++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/specification/ap2-mandates.md b/docs/specification/ap2-mandates.md index 3d305d232..e9dd07e65 100644 --- a/docs/specification/ap2-mandates.md +++ b/docs/specification/ap2-mandates.md @@ -236,6 +236,15 @@ in UCP requests and responses. The mandate credential structure (claims, selective disclosure, key binding) is defined by the [AP2 Protocol Specification](https://ap2-protocol.org/specification). +**Wire format:** `ap2.checkout_mandate` carries a compact-serialized SD-JWT +credential: `~~...~[]` per +[RFC 9901](https://datatracker.ietf.org/doc/html/rfc9901). Presentations +produced through delegation (for example by the +[AP2 reference implementation](https://github.com/google-agentic-commerce/AP2)) +serialize a *chain* of such tokens joined by `~~`. The schema's `pattern` +admits both forms; it checks syntactic form only — signature, key-binding, +and disclosure verification are defined by the AP2 Protocol Specification. + ### Canonicalization All JSON payloads **MUST** be canonicalized using **JSON Canonicalization diff --git a/source/schemas/shopping/ap2_mandate.json b/source/schemas/shopping/ap2_mandate.json index 4d8b7d071..3b256122d 100644 --- a/source/schemas/shopping/ap2_mandate.json +++ b/source/schemas/shopping/ap2_mandate.json @@ -13,9 +13,9 @@ }, "checkout_mandate": { "title": "Checkout Mandate", - "description": "SD-JWT+kb credential in `ap2.checkout_mandate`. Proving user authorization for the checkout. Contains the full checkout including `ap2.merchant_authorization`.", + "description": "SD-JWT credential in `ap2.checkout_mandate`, proving user authorization for the checkout. Contains the full checkout including `ap2.merchant_authorization`. Accepted serializations (syntactic form only; cryptographic verification is separate): a compact SD-JWT with or without a trailing Key Binding JWT (RFC 9901 `~~...~[]`), or a delegated SD-JWT chain of such tokens joined by `~~` as emitted by the AP2 reference implementation.", "type": "string", - "pattern": "^[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(~[A-Za-z0-9_-]+)*$" + "pattern": "^([A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(~[A-Za-z0-9_-]+)*(~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)?~~)*[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(~[A-Za-z0-9_-]+)*(~|~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)?$" }, "ap2_with_merchant_authorization": { "type": "object", From 191e7e732f9247d1fb417bd9efb11561fcd071b1 Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Tue, 21 Jul 2026 13:18:56 -0400 Subject: [PATCH 2/2] refine: tighten chain hops to reject bare-JWT segments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per adversarial review: the previous pattern let a delegate-chain hop be a bare JWT (e.g. `J~~J`), a shape the AP2 reference verifier rejects (every hop must be a well-formed SD-JWT — at least one disclosure, or a KB-JWT). The back-compat leniency for the old pattern's bare/no-terminator forms now applies only to a single (non-chained) token, not inside chains. Still a strict superset of the original pattern (verified: 30k-string fuzz over the old grammar, zero regressions), and all reference-emitted serializations (1-3 hop chains, KB and no-KB finals, real reference-minted wires) still match. --- _review/ap2___init__.py | 20 ++ _review/ap2_chain.py | 299 ++++++++++++++++ _review/ap2_common.py | 311 +++++++++++++++++ _review/ap2_kb_sd_jwt.py | 205 +++++++++++ _review/ap2_mandate_sdk.py | 417 +++++++++++++++++++++++ _review/ap2_sd_jwt.py | 88 +++++ _review/redos.mjs | 29 ++ _review/redos.py | 29 ++ _review/run_node.mjs | 22 ++ _review/run_py.py | 41 +++ _review/rustcheck/Cargo.lock | 144 ++++++++ _review/rustcheck/Cargo.toml | 8 + _review/rustcheck/src/main.rs | 40 +++ _review/vectors.json | 47 +++ docs/specification/ap2-mandates.md | 6 +- source/schemas/shopping/ap2_mandate.json | 2 +- 16 files changed, 1704 insertions(+), 4 deletions(-) create mode 100644 _review/ap2___init__.py create mode 100644 _review/ap2_chain.py create mode 100644 _review/ap2_common.py create mode 100644 _review/ap2_kb_sd_jwt.py create mode 100644 _review/ap2_mandate_sdk.py create mode 100644 _review/ap2_sd_jwt.py create mode 100644 _review/redos.mjs create mode 100644 _review/redos.py create mode 100644 _review/run_node.mjs create mode 100644 _review/run_py.py create mode 100644 _review/rustcheck/Cargo.lock create mode 100644 _review/rustcheck/Cargo.toml create mode 100644 _review/rustcheck/src/main.rs create mode 100644 _review/vectors.json diff --git a/_review/ap2___init__.py b/_review/ap2___init__.py new file mode 100644 index 000000000..44595e036 --- /dev/null +++ b/_review/ap2___init__.py @@ -0,0 +1,20 @@ +"""SD-JWT primitive subsystem used by the AP2 mandate facade.""" + +from ap2.sdk.sdjwt import chain, kb_sd_jwt, sd_jwt +from ap2.sdk.sdjwt.common import ( + ParsedToken, + compute_issuer_jwt_hash, + compute_sd_hash, + parse_token, +) + + +__all__ = [ + 'ParsedToken', + 'chain', + 'compute_issuer_jwt_hash', + 'compute_sd_hash', + 'kb_sd_jwt', + 'parse_token', + 'sd_jwt', +] diff --git a/_review/ap2_chain.py b/_review/ap2_chain.py new file mode 100644 index 000000000..e75875fd3 --- /dev/null +++ b/_review/ap2_chain.py @@ -0,0 +1,299 @@ +"""dSD-JWT chain orchestration (draft-gco-oauth-delegate-sd-jwt-00 §6). + +This module owns the *chain-level* concerns: + +- Compact serialization: tokens are joined by ``~~`` in AP2's dSD-JWT chain. +- Per-hop dispatch: ``verify_chain`` calls + :func:`ap2.sdk.sdjwt.sd_jwt.verify` for the first hop and + :func:`ap2.sdk.sdjwt.kb_sd_jwt.verify` for KB-SD-JWT hops. +- ``cnf`` walking: each hop's signing key is the previous hop's + ``cnf.jwk``. +- Time checks and x5c root-of-trust verification for the first hop. + +The per-token crypto + ``typ`` + ``sd_hash``/``issuer_jwt_hash`` checks live +in the per-primitive modules, not here. +""" + +from __future__ import annotations + +import binascii +import json +import logging +import time + +from collections.abc import Callable +from typing import Any, Protocol + +from ap2.sdk.sdjwt import common, kb_sd_jwt, sd_jwt +from ap2.sdk.sdjwt.common import ParsedToken +from ap2.sdk.utils import b64url_decode +from cryptography import x509 +from cryptography.hazmat.primitives.asymmetric import ec +from jwcrypto.jwk import JWK + + +class PublicKeyProvider(Protocol): + """Resolve the root token's verification key.""" + + def __call__(self, token: ParsedToken) -> JWK: ... + + +class X5cOrKidPublicKeyProvider: + """Resolve a root key from ``x5c`` or fall back to a ``kid`` lookup.""" + + def __init__( + self, + kid_lookup: Callable[[str], JWK], + trusted_roots: list[x509.Certificate] | None = None, + ) -> None: + self._kid_lookup = kid_lookup + self._trusted_roots = trusted_roots + + def __call__(self, token: ParsedToken) -> JWK: + header = token.header + if 'x5c' in header: + return self._resolve_x5c_key(header) + + key_id = header.get('kid') + if not key_id: + raise ValueError( + "Missing or invalid 'kid' or 'x5c' in the first token header" + ) + key = self._kid_lookup(key_id) + if key is None: + raise ValueError(f'Provider returned no key for key_id: {key_id}') + if not isinstance(key, JWK): + raise TypeError( + 'PublicKeyProvider must return a jwcrypto.jwk.JWK; got ' + f'{type(key).__name__}. Wrap cryptography EC keys with ' + 'JWK.from_pyca(...).' + ) + return key + + def _resolve_x5c_key(self, header: dict[str, Any]) -> JWK: + """Resolve the verification :class:`JWK` from an ``x5c`` header.""" + x5c = header['x5c'] + if not isinstance(x5c, list) or not x5c: + raise ValueError('x5c header must be a non-empty list') + certs = [ + x509.load_der_x509_certificate(b64url_decode(cert_b64)) + for cert_b64 in x5c + ] + for i in range(len(certs) - 1): + issuer_cert = certs[i + 1] + subject_cert = certs[i] + issuer_cert.public_key().verify( + subject_cert.signature, + subject_cert.tbs_certificate_bytes, + ec.ECDSA(subject_cert.signature_hash_algorithm), + ) + if self._trusted_roots: + verified = False + last_cert = certs[-1] + for root in self._trusted_roots: + try: + root.public_key().verify( + last_cert.signature, + last_cert.tbs_certificate_bytes, + ec.ECDSA(last_cert.signature_hash_algorithm), + ) + verified = True + break + except Exception: + continue + if not verified: + raise ValueError( + 'Certificate chain does not chain to a trusted root' + ) + return JWK.from_pyca(certs[0].public_key()) + + +# RFC 9901 §4.2: a disclosure is the base64url of a JSON array of either +# 2 elements (array-element disclosure: [salt, value]) or 3 elements +# (object-property disclosure: [salt, name, value]). +_SD_JWT_DISCLOSURE_ARRAY_LEN = 2 +_SD_JWT_DISCLOSURE_PROPERTY_LEN = 3 + + +# ── Chain verification ─────────────────────────────────────────────────── + + +def verify_chain( # noqa: PLR0913 (public API; each kwarg is an independent verifier input) + tokens: list[ParsedToken], + public_key_provider: PublicKeyProvider, + clock_skew_seconds: int = 300, + expected_aud: str | None = None, + expected_nonce: str | None = None, + current_time: int | None = None, +) -> list[dict[str, Any]]: + """Verify a dSD-JWT delegation chain and return per-hop effective payloads. + + Walks the chain from root: + + - index 0 (root SD-JWT): verified with ``public_key_provider(token)``. + - index i (KB-SD-JWT): verified with the previous hop's ``cnf.jwk`` via + :func:`ap2.sdk.sdjwt.kb_sd_jwt.verify`, enforcing ``expected_aud`` / + ``expected_nonce`` on the terminal hop when provided. + + Returns a list of effective payload dicts (one per token, resolved from + ``delegate_payload[0]`` when present). + """ + if not tokens: + raise ValueError('Tokens list cannot be empty') + + payloads: list[dict[str, Any]] = [] + now = current_time if current_time is not None else int(time.time()) + parsed_tokens = tokens + + root_token = parsed_tokens[0] + current_key = public_key_provider(root_token) + root_payload = sd_jwt.verify(root_token.canonical, current_key) + root_items = _effective_payloads(root_payload, root_token, 0, True) + # KB hops verify with the previous hop's cnf.jwk. That key may only be + # available after SD-JWT verification resolves delegate_payload disclosures, + # so store the verified payload on the parsed token before walking onward. + parsed_tokens[0] = root_token.with_verified_payload( + root_payload, root_items + ) + _check_time_claims([root_payload], 0, now, clock_skew_seconds) + _check_time_claims(root_items, 0, now, clock_skew_seconds) + payloads.extend(root_items if root_items else [root_payload]) + + for i, current_token in enumerate(parsed_tokens[1:], start=1): + is_last = i == len(parsed_tokens) - 1 + payload = kb_sd_jwt.verify( + current_token, + parsed_tokens[i - 1], + expected_aud=expected_aud if is_last else None, + expected_nonce=expected_nonce if is_last else None, + ) + delegate_items = _effective_payloads( + payload, current_token, i, require_single=not is_last + ) + _check_time_claims([payload], i, now, clock_skew_seconds) + _check_time_claims(delegate_items, i, now, clock_skew_seconds) + payloads.extend(delegate_items if delegate_items else [payload]) + parsed_tokens[i] = current_token.with_verified_payload( + payload, delegate_items + ) + + return payloads + + +# ── Helpers ────────────────────────────────────────────────────────────── + + +def _effective_payloads( + payload: dict[str, Any], + token: ParsedToken, + token_index: int, + require_single: bool, +) -> list[dict[str, Any]]: + delegate_items = _resolve_delegate_items( + payload.get('delegate_payload'), token, token_index + ) + if require_single and len(delegate_items) > 1: + raise ValueError( + f'Token {token_index}: delegate_payload has ' + f'{len(delegate_items)} disclosed items, expected exactly 1' + ) + return delegate_items + + +def _resolve_delegate_items( + delegate_payload: list[Any] | None, + token: ParsedToken, + token_index: int, +) -> list[dict[str, Any]]: + """Decode and resolve items inside a ``delegate_payload`` list. + + Per draft-gco-oauth-delegate-sd-jwt-00 §5.1.4 the ``delegate_payload`` + claim, when present on a verified JWT, is a JSON array whose elements are + either selectively-disclosable dicts (already resolved by + ``SDJWTVerifier``) or base64url disclosure strings that encode dicts. We + tolerate a non-list/``None`` value defensively and return ``[]``. + """ + if not isinstance(delegate_payload, list): + return [] + items: list[dict[str, Any]] = [] + for item in delegate_payload: + if isinstance(item, dict): + _inline_sd_claims(item, token) + items.append(item) + elif isinstance(item, str): + decoded = _decode_disclosure_dict(item, token_index) + if decoded is not None: + items.append(decoded) + return items + + +def _inline_sd_claims( + item: dict[str, Any], + token: ParsedToken, +) -> None: + """Resolve ``_sd`` digests in ``item`` in place from token disclosures.""" + sd_digests = item.get('_sd', []) + if not sd_digests or not token.disclosures: + return + for digest in sd_digests: + for d in token.disclosures: + if common.compute_disclosure_digest(d, token.sd_alg) == digest: + decoded = json.loads(b64url_decode(d).decode('utf-8')) + if len(decoded) == _SD_JWT_DISCLOSURE_PROPERTY_LEN: + item[decoded[1]] = decoded[2] + break + + +def _decode_disclosure_dict( + disclosure: str, + token_index: int, +) -> dict[str, Any] | None: + """Decode a base64url disclosure into its dict value, if present.""" + try: + arr = json.loads(b64url_decode(disclosure).decode('utf-8')) + if not isinstance(arr, list): + return None + if len(arr) == _SD_JWT_DISCLOSURE_PROPERTY_LEN: + val = arr[2] + elif len(arr) == _SD_JWT_DISCLOSURE_ARRAY_LEN: + val = arr[1] + else: + val = None + return val if isinstance(val, dict) else None + except (binascii.Error, UnicodeDecodeError, json.JSONDecodeError) as e: + logging.warning( + 'Token %d: Failed to decode disclosure in delegate_payload: %s', + token_index, + e, + ) + return None + + +def _check_time_claims( + payloads: list[dict[str, Any]], + token_index: int, + now: int, + clock_skew: int, +) -> None: + """Validate exp/iat on a list of effective payloads.""" + for p in payloads: + exp = p.get('exp') + if exp is not None: + if not isinstance(exp, (int, float)): + raise ValueError( + f"Token {token_index} has invalid 'exp' claim type: " + f'{type(exp)}' + ) + if now > exp + clock_skew: + raise ValueError(f'Token {token_index} expired at {exp}') + iat = p.get('iat') + if iat is not None: + if not isinstance(iat, (int, float)): + raise ValueError( + f"Token {token_index} has invalid 'iat' claim type: " + f'{type(iat)}' + ) + if iat > now + clock_skew: + raise ValueError( + f'Token {token_index} iat is in the future: {iat}' + ) diff --git a/_review/ap2_common.py b/_review/ap2_common.py new file mode 100644 index 000000000..cdea78aed --- /dev/null +++ b/_review/ap2_common.py @@ -0,0 +1,311 @@ +"""Common SD-JWT hashing, issuance, and claim-validation helpers.""" + +from __future__ import annotations + +import binascii +import hashlib +import json + +from collections.abc import Callable +from dataclasses import asdict, dataclass, replace +from typing import Any, Literal + +from ap2.sdk.disclosure_metadata import DisclosureMetadata +from ap2.sdk.generated.types.jwk import JsonWebKey +from ap2.sdk.utils import b64url_decode, b64url_encode +from jwcrypto.jwk import JWK +from pydantic import BaseModel +from sd_jwt.issuer import SDJWTIssuer + + +HashMode = Literal['sd_hash', 'issuer_jwt_hash'] +_COMPACT_JWT_PARTS = 3 + +_HASH_BY_SD_ALG: dict[str, Callable[[bytes], Any]] = { + 'sha-256': hashlib.sha256, + 'sha-384': hashlib.sha384, + 'sha-512': hashlib.sha512, +} + + +@dataclass(frozen=True) +class ParsedToken: + """Parsed SD-JWT token with header and payload available once.""" + + issuer_jwt: str + disclosures: list[str] + kb_jwt: str | None + header: dict[str, Any] + payload: dict[str, Any] + verified_payload: dict[str, Any] | None = None + delegate_items: list[dict[str, Any]] | None = None + + @property + def typ(self) -> str | None: + typ = self.header.get('typ') + return typ if isinstance(typ, str) else None + + @property + def sd_alg(self) -> str | None: + alg = self.payload.get('_sd_alg') + return alg if isinstance(alg, str) else None + + @property + def sd_jwt(self) -> str: + if self.disclosures: + return self.issuer_jwt + '~' + '~'.join(self.disclosures) + '~' + return self.issuer_jwt + '~' + + @property + def canonical(self) -> str: + if self.kb_jwt: + return self.sd_jwt + self.kb_jwt + return self.sd_jwt + + def with_verified_payload( + self, + payload: dict[str, Any], + delegate_items: list[dict[str, Any]], + ) -> ParsedToken: + """Return a copy with verified, disclosure-resolved payload state.""" + return replace( + self, + verified_payload=payload, + delegate_items=delegate_items, + ) + + def cnf_jwk(self) -> JWK | None: + """Return this verified token's resolved ``cnf.jwk``, if present.""" + if self.verified_payload is None: + raise ValueError( + 'Token has not been verified; cnf.jwk is unavailable' + ) + cnf = self._find_cnf() + if isinstance(cnf, dict) and 'jwk' in cnf: + jwk_model = JsonWebKey.model_validate(cnf['jwk']) + return JWK(**jwk_model.model_dump(exclude_none=True)) + return None + + def _find_cnf(self) -> dict[str, Any] | None: + delegate_items = self.delegate_items or [] + for item in delegate_items: + cnf = item.get('cnf') + if isinstance(cnf, dict) and 'jwk' in cnf: + return cnf + if self.verified_payload is None: + return None + delegate_payload = self.verified_payload.get('delegate_payload') + if isinstance(delegate_payload, list): + for item in delegate_payload: + if not isinstance(item, dict): + continue + cnf = item.get('cnf') + if isinstance(cnf, dict) and 'jwk' in cnf: + return cnf + cnf = self.verified_payload.get('cnf') + if isinstance(cnf, dict) and 'jwk' in cnf: + return cnf + return None + + +def parse_token(token: str) -> ParsedToken: + """Parse and canonicalize a compact SD-JWT token.""" + if token.startswith('~'): + raise ValueError('Malformed SD-JWT: empty issuer JWT') + if '~' not in token: + raise ValueError('Malformed SD-JWT: missing disclosure separator') + + parts = token.split('~') + issuer_jwt = parts[0] + disclosure_parts = parts[1:-1] + if any(not disclosure for disclosure in disclosure_parts): + raise ValueError('Malformed SD-JWT: empty disclosure segment') + if token.endswith('~'): + disclosures = disclosure_parts + kb_jwt = None + else: + kb_jwt = parts[-1] + if len(kb_jwt.split('.')) != _COMPACT_JWT_PARTS: + raise ValueError( + 'Malformed KB-JWT: expected header.payload.signature' + ) + disclosures = disclosure_parts + + jwt_parts = issuer_jwt.split('.') + if len(jwt_parts) != _COMPACT_JWT_PARTS: + raise ValueError( + 'Malformed SD-JWT: issuer JWT must have header.payload.signature' + ) + header_segment, payload_segment, _signature_segment = jwt_parts + header = decode_jwt_segment(header_segment, 'header') + payload = decode_jwt_segment(payload_segment, 'payload') + return ParsedToken(issuer_jwt, disclosures, kb_jwt, header, payload) + + +def decode_jwt_segment(segment: str, part_name: str) -> dict[str, Any]: + """Decode a compact JWT header or payload segment into a JSON object.""" + try: + decoded = json.loads(b64url_decode(segment)) + except (binascii.Error, json.JSONDecodeError) as exc: + raise ValueError(f'Cannot parse JWT {part_name}: {exc}') from exc + if not isinstance(decoded, dict): + raise ValueError(f'JWT {part_name} must decode to a JSON object') + return decoded + + +def _hash_for_alg(sd_alg: str | None) -> Callable[[bytes], Any]: + if sd_alg is None: + return hashlib.sha256 + try: + return _HASH_BY_SD_ALG[sd_alg] + except KeyError as exc: + raise ValueError(f'Unsupported _sd_alg: {sd_alg!r}') from exc + + +def _hash_ascii(value: str, sd_alg: str | None) -> str: + digest = _hash_for_alg(sd_alg)(value.encode('ascii')).digest() + return b64url_encode(digest) + + +def compute_sd_hash(token: ParsedToken) -> str: + """Hash an SD-JWT including disclosures, excluding a trailing KB-JWT.""" + return _hash_ascii(token.sd_jwt, token.sd_alg) + + +def compute_issuer_jwt_hash(token: ParsedToken) -> str: + """Hash only the issuer-signed JWT portion of an SD-JWT.""" + return _hash_ascii(token.issuer_jwt, token.sd_alg) + + +def compute_disclosure_digest(disclosure: str, sd_alg: str | None) -> str: + """Hash a disclosure string using the issuer token's SD algorithm.""" + return _hash_ascii(disclosure, sd_alg) + + +def compute_binding( + prev_token: ParsedToken, hash_mode: HashMode +) -> tuple[str, str]: + """Return the (claim_name, value) pair for the binding hash.""" + if hash_mode == 'sd_hash': + return 'sd_hash', compute_sd_hash(prev_token) + if hash_mode == 'issuer_jwt_hash': + return 'issuer_jwt_hash', compute_issuer_jwt_hash(prev_token) + raise ValueError( + f"hash_mode must be 'sd_hash' or 'issuer_jwt_hash', got {hash_mode!r}" + ) + + +def verify_binding(payload: dict[str, Any], prev_token: ParsedToken) -> None: + """Enforce that exactly one binding claim is present and matches.""" + has_sd = 'sd_hash' in payload + has_iss = 'issuer_jwt_hash' in payload + if has_sd == has_iss: + raise ValueError( + "KB-SD-JWT payload must contain exactly one of 'sd_hash' or " + f"'issuer_jwt_hash' (got sd_hash={has_sd}, " + f'issuer_jwt_hash={has_iss})' + ) + if has_sd: + expected = compute_sd_hash(prev_token) + actual = payload['sd_hash'] + if actual != expected: + raise ValueError( + f"sd_hash mismatch: expected '{expected}', got '{actual}'" + ) + else: + expected = compute_issuer_jwt_hash(prev_token) + actual = payload['issuer_jwt_hash'] + if actual != expected: + raise ValueError( + f"issuer_jwt_hash mismatch: expected '{expected}', " + f"got '{actual}'" + ) + + +def delegate_claims_from_model(payload: BaseModel) -> dict[str, Any]: + """Serialize a Pydantic payload into AP2 delegate claims.""" + if not isinstance(payload, BaseModel): + raise TypeError('payload must be an instance of pydantic.BaseModel') + return payload.model_dump(by_alias=True, exclude_none=True) + + +def selectively_disclosable_claims( + delegate_claims: dict[str, Any], + sd: DisclosureMetadata | None, + extra_claims: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Build SDObj-wrapped claims with ``delegate_payload`` as the SD root.""" + claims: dict[str, Any] = { + 'delegate_payload': [delegate_claims], + '_sd': { + 'children': { + 'delegate_payload': {'disclose_all': True}, + } + }, + } + if extra_claims: + claims.update(extra_claims) + if sd is not None: + claims['_sd']['children']['delegate_payload']['all_array_children'] = ( + asdict(sd) + ) + + user_claims = claims.copy() + metadata = DisclosureMetadata.from_dict(user_claims.pop('_sd')) + return metadata.apply(user_claims) + + +def header_parameters( + signing_key: JWK, typ: str | None = None +) -> dict[str, Any]: + """Return SD-JWT header parameters, including ``kid`` when present.""" + jwk_dict = json.loads(signing_key.export()) + kid = jwk_dict.get('kid') + params: dict[str, Any] = {} + if typ is not None: + params['typ'] = typ + if kid: + params['kid'] = kid + return params + + +def issue_sd_jwt( + *, + claims: dict[str, Any], + issuer_key: JWK, + header_params: dict[str, Any], + add_decoy_claims: bool, + serialization_format: str, +) -> SDJWTIssuer: + """Create an ``SDJWTIssuer`` with AP2's common issuer options.""" + return SDJWTIssuer( + user_claims=claims, + issuer_key=issuer_key, + holder_key=None, + sign_alg=None, + add_decoy_claims=add_decoy_claims, + serialization_format=serialization_format, + extra_header_parameters=header_params, + ) + + +def verify_expected_claims( + payload: dict[str, Any], + *, + expected_aud: str | None, + expected_nonce: str | None, + token_label: str, +) -> None: + """Validate common KB-SD-JWT claims after signature verification.""" + if 'iat' not in payload: + raise ValueError(f"{token_label} missing required 'iat' claim") + if expected_aud is not None and payload.get('aud') != expected_aud: + raise ValueError( + f"{token_label} aud mismatch: expected '{expected_aud}'," + f" got '{payload.get('aud')}'" + ) + if expected_nonce is not None and payload.get('nonce') != expected_nonce: + raise ValueError( + f"{token_label} nonce mismatch: expected '{expected_nonce}'," + f" got '{payload.get('nonce')}'" + ) diff --git a/_review/ap2_kb_sd_jwt.py b/_review/ap2_kb_sd_jwt.py new file mode 100644 index 000000000..35c709b39 --- /dev/null +++ b/_review/ap2_kb_sd_jwt.py @@ -0,0 +1,205 @@ +"""KB-SD-JWT primitive (draft-gco-oauth-delegate-sd-jwt-00 §5.1.4). + +This module handles both AP2 delegation hop variants: + +- intermediate hop: ``typ="kb+sd-jwt+kb"``, delegate payload MUST contain + ``cnf`` for the next delegate. +- terminal hop: ``typ="kb+sd-jwt"``, delegate payload MUST NOT contain + ``cnf``. + +Both variants carry ``iat``/``aud``/``nonce`` and one of ``sd_hash`` / +``issuer_jwt_hash``. No separate trailing KB-JWT is used in the AP2 flow. +""" + +from __future__ import annotations + +import json +import time + +from typing import Any + +from ap2.sdk.disclosure_metadata import DisclosureMetadata +from ap2.sdk.sdjwt import common +from ap2.sdk.sdjwt.common import HashMode, ParsedToken +from ap2.sdk.sdjwt.sd_jwt import verify as sd_jwt_verify +from ap2.sdk.utils import b64url_decode +from jwcrypto.jwk import JWK +from pydantic import BaseModel +from sd_jwt.issuer import SDJWTIssuer + + +TYP_TERMINAL = ['kb+sd-jwt', 'kb-sd-jwt'] +TYP_INTERMEDIATE = ['kb+sd-jwt+kb', 'kb-sd-jwt+kb'] + + +def create( # noqa: PLR0913 (public create() surface: each kwarg is orthogonal) + prev_token: ParsedToken, + holder_key: JWK, + payload: BaseModel, + aud: str, + nonce: str, + sd: DisclosureMetadata | None = None, + hash_mode: HashMode = 'sd_hash', + add_decoy_claims: bool = False, + serialization_format: str = 'compact', +) -> SDJWTIssuer: + """Build and sign a KB-SD-JWT delegation hop. + + Args: + prev_token: The preceding SD-JWT or KB-SD-JWT being delegated. + holder_key: The signing key of this hop. + payload: Delegate payload model. + aud: Audience for this hop. + nonce: Nonce from the verifier/next delegate. + sd: Optional selective-disclosure metadata; ``None`` auto-derives. + hash_mode: Binding mode against ``prev_token``. + add_decoy_claims: Add decoy ``_sd`` digests. + serialization_format: SD-JWT serialization format. + + Returns: + An ``SDJWTIssuer`` whose ``.sd_jwt_issuance`` is the signed token. + """ + if not aud or not nonce: + raise ValueError('aud and nonce are required for KB-SD-JWT hops') + + delegate_claims = common.delegate_claims_from_model(payload) + has_cnf = 'cnf' in delegate_claims + + if sd is None: + sd = DisclosureMetadata.from_model(payload) + + binding_claim, binding_value = common.compute_binding(prev_token, hash_mode) + + extra_claims: dict[str, Any] = { + 'iat': int(time.time()), + 'aud': aud, + 'nonce': nonce, + binding_claim: binding_value, + } + terminal = not has_cnf + typ = TYP_TERMINAL[0] if terminal else TYP_INTERMEDIATE[0] + + sd_claims = common.selectively_disclosable_claims( + delegate_claims, sd, extra_claims + ) + return common.issue_sd_jwt( + claims=sd_claims, + issuer_key=holder_key, + header_params=common.header_parameters(holder_key, typ), + add_decoy_claims=add_decoy_claims, + serialization_format=serialization_format, + ) + + +def verify( + token: ParsedToken, + prev_token: ParsedToken, + expected_aud: str | None = None, + expected_nonce: str | None = None, +) -> dict[str, Any]: + """Verify a KB-SD-JWT hop. + + Checks: + - Header ``typ`` is a known AP2 KB-SD-JWT type. + - Signature verifies under the preceding hop's ``cnf.jwk``. + - Exactly one of ``sd_hash`` / ``issuer_jwt_hash`` is present and + matches the hash of ``prev_token``. + - ``iat`` is present. + - If ``expected_aud`` / ``expected_nonce`` are provided, they match. + - Terminal hops do not contain ``cnf``; intermediate hops do. + """ + typ = token.typ + if typ not in TYP_TERMINAL + TYP_INTERMEDIATE: + raise ValueError( + f"Unexpected JWT typ: expected one of {TYP_TERMINAL + TYP_INTERMEDIATE}, " + f"got '{token.typ}'" + ) + + prev_key = prev_token.cnf_jwk() + if prev_key is None: + raise ValueError('Previous token missing cnf.jwk') + payload = sd_jwt_verify(token.canonical, prev_key) + # Resolve SD-JWT digests in delegate_payload against token disclosures. + # CMWallet places mandate commitment digests directly in delegate_payload + # rather than via a standard top-level _sd array; this step normalises + # them into inline dicts so the cnf check below works correctly. + _resolve_delegate_payload(payload, token) + common.verify_binding(payload, prev_token) + if typ in TYP_TERMINAL: + common.verify_expected_claims( + payload, + expected_aud=expected_aud, + expected_nonce=expected_nonce, + token_label='KB-SD-JWT', + ) + has_cnf = _delegate_payload_has_cnf(payload) + if typ in TYP_TERMINAL and has_cnf: + raise ValueError("Terminal KB-SD-JWT MUST NOT carry a 'cnf' claim") + if typ in TYP_INTERMEDIATE and not has_cnf: + raise ValueError(f"Intermediate {typ} requires a 'cnf' claim") + return payload + + +def _delegate_payload_has_cnf(payload: dict[str, Any]) -> bool: + delegate_payload = payload.get('delegate_payload') + if not isinstance(delegate_payload, list): + return False + return any( + isinstance(item, dict) and isinstance(item.get('cnf'), dict) + for item in delegate_payload + ) + + +def _try_resolve_digest( + digest: str, + disclosures: list[str], + sd_alg: str | None, +) -> dict[str, Any] | None: + """Return the dict value of the first disclosure whose hash equals ``digest``. + + Handles the CMWallet format where ``delegate_payload`` items are SD-JWT + ``_sd``-style digest strings referencing mandate disclosures that are + appended to the token, rather than inline dict objects. + """ + for disc in disclosures: + if common.compute_disclosure_digest(disc, sd_alg) != digest: + continue + try: + arr = json.loads(b64url_decode(disc).decode('utf-8')) + if not isinstance(arr, list): + continue + # [salt, value] for array-element disclosures; [salt, name, value] + # for object-property disclosures. + val = arr[1] if len(arr) == 2 else arr[2] if len(arr) == 3 else None + if isinstance(val, dict): + return val + except Exception: + continue + return None + + +def _resolve_delegate_payload( + payload: dict[str, Any], + token: ParsedToken, +) -> None: + """Resolve SD-JWT digests inside ``delegate_payload`` against token disclosures. + + Mutates ``payload`` in place. Items that are already dicts are left + unchanged; string items are checked against ``token.disclosures`` using + the token's ``_sd_alg``. If a matching disclosure is found its dict + value replaces the digest string. + """ + dp = payload.get('delegate_payload') + if not isinstance(dp, list) or not token.disclosures: + return + sd_alg = token.sd_alg + resolved = [] + for item in dp: + if isinstance(item, dict): + resolved.append(item) + elif isinstance(item, str): + decoded = _try_resolve_digest(item, token.disclosures, sd_alg) + resolved.append(decoded if decoded is not None else item) + else: + resolved.append(item) + payload['delegate_payload'] = resolved diff --git a/_review/ap2_mandate_sdk.py b/_review/ap2_mandate_sdk.py new file mode 100644 index 000000000..30d9d7a00 --- /dev/null +++ b/_review/ap2_mandate_sdk.py @@ -0,0 +1,417 @@ +"""High-level mandate facade. + +``MandateClient`` is the outer API that AP2 roles use: + +- ``create(payloads, issuer_key, sd=None)`` — mint a root SD-JWT. +- ``present(holder_key, mandate_token, payloads, ...)`` — append one + delegation hop via :mod:`ap2.sdk.sdjwt.kb_sd_jwt`. +- ``verify(token, key_or_provider, ...)`` — verify a single token or any + ``~~``-joined delegation chain via + :func:`ap2.sdk.sdjwt.chain.verify_chain`. + +This module also owns the typed :class:`SdJwtMandate` wrapper used for +single-token convenience verification. +""" + +from __future__ import annotations + +import datetime +import json +import logging +import pathlib + +from abc import ABC, abstractmethod +from typing import Any, Generic, TypeVar + +from ap2.sdk.disclosure_metadata import ( + DisclosureMetadata, + sd_claims_to_disclose, +) +from ap2.sdk.sdjwt import chain as _chain +from ap2.sdk.sdjwt import common, kb_sd_jwt, sd_jwt +from ap2.sdk.sdjwt.chain import PublicKeyProvider +from ap2.sdk.utils import b64url_decode +from jwcrypto.jwk import JWK +from sd_jwt.holder import SDJWTHolder + + +T = TypeVar('T', bound=Any) + +_SDK_ROOT = pathlib.Path(__file__).resolve().parents[3] +LOG_FILE_PATH = str(_SDK_ROOT / '.logs' / 'mandate_operations.log') + +# RFC 9901 §4.2: a disclosure is the base64url of a JSON array of either +# 2 elements (array element: [salt, value]) or 3 elements +# (object property: [salt, name, value]). +_SD_JWT_DISCLOSURE_ARRAY_LEN = 2 +_SD_JWT_DISCLOSURE_PROPERTY_LEN = 3 +_COMPACT_JWT_PARTS = 3 + + +def _log_event(event_type: str, stage: str, data: dict[str, Any]) -> None: + """Append a structured event to the mandate operations log file.""" + log_entry = { + 'timestamp': datetime.datetime.now().isoformat(), + 'event': event_type, + 'stage': stage, + 'data': data, + } + try: + log_path = pathlib.Path(LOG_FILE_PATH) + log_path.parent.mkdir(parents=True, exist_ok=True) + with log_path.open('a') as f: + f.write(json.dumps(log_entry) + '\n') + except OSError as e: + logging.warning('Failed to write to mandate log file: %s', e) + + +def _canonical_chain_segment(segment: str, index: int, total: int) -> str: + """Restore the trailing ``~`` stripped when joining dSD-JWT segments.""" + if index == total - 1 or segment.endswith('~'): + return segment + last_segment = segment.rsplit('~', maxsplit=1)[-1] + if len(last_segment.split('.')) == _COMPACT_JWT_PARTS: + return segment + return segment + '~' + + +# ── Typed mandate wrappers ─────────────────────────────────────────────── + + +class Mandate(ABC, Generic[T]): # noqa: UP046 (PEP 695 requires py312; repo supports py311+) + """Base interface for all mandates.""" + + @property + @abstractmethod + def serialized(self) -> str: + """Returns the serialized representation of the mandate.""" + + @property + @abstractmethod + def mandate_payload(self) -> T: + """Returns the underlying mandate payload object.""" + + def is_valid(self) -> bool: + """Checks if the mandate is currently valid.""" + return True + + +class SdJwtMandate(Mandate[T]): + """Mandate backed by a single SD-JWT compact serialization.""" + + def __init__(self, sd_jwt_issuance: str, mandate_payload: T): + self._serialized = sd_jwt_issuance + self._mandate_payload = mandate_payload + + @property + def serialized(self) -> str: + return self._serialized + + @property + def mandate_payload(self) -> T: + return self._mandate_payload + + @classmethod + def from_sd_jwt( + cls, + compact_serialization: str, + issuer_public_key: JWK, + payload_type: type[T], + expected_aud: str | None = None, + expected_nonce: str | None = None, + ) -> SdJwtMandate[T]: + """Verify a single SD-JWT and wrap it as a Mandate.""" + verified_payload = sd_jwt.verify( + compact_serialization, + issuer_public_key, + expected_aud=expected_aud, + expected_nonce=expected_nonce, + ) + delegate_payload = verified_payload.get('delegate_payload') + if isinstance(delegate_payload, list): + disclosed = [ + item for item in delegate_payload if isinstance(item, dict) + ] + if len(disclosed) != 1: + raise ValueError( + f'delegate_payload has {len(disclosed)} disclosed items,' + ' expected exactly 1' + ) + effective = disclosed[0] + else: + effective = verified_payload + payload = payload_type.model_validate(effective) + return cls(compact_serialization, payload) + + +# ── MandateClient facade ───────────────────────────────────────────────── + + +_HashMode = kb_sd_jwt.HashMode + + +class MandateClient: + """Stateless client for creating, presenting, and verifying AP2 mandates.""" + + def create( + self, + payloads: list[Any], + issuer_key: JWK, + sd: DisclosureMetadata | None = None, + ) -> str: + """Sign a root SD-JWT for ``payloads[0]``. + + See :func:`ap2.sdk.sdjwt.sd_jwt.create` for details. ``sd=None`` + auto-derives selective-disclosure metadata from the payload model's + annotations. + """ + issuer = sd_jwt.create( + payload=payloads[0], + issuer_key=issuer_key, + sd=sd, + ) + return issuer.sd_jwt_issuance + + def verify( # noqa: PLR0913 (public API: each kwarg is independent input) + self, + token: str, + key_or_provider: JWK | PublicKeyProvider, + payload_type: type[T] | None = None, + expected_aud: str | None = None, + expected_nonce: str | None = None, + clock_skew_seconds: int = 300, + current_time: int | None = None, + ) -> Mandate[T] | list[dict[str, Any]]: + """Unified verifier for single tokens and ``~~``-joined chains. + + ``key_or_provider`` MUST be one of: + + - a :class:`jwcrypto.jwk.JWK` — used as-is for a single-token verify, or + - a callable ``(ParsedToken) -> JWK`` — used for the root hop of a + chain. + + Wrap a ``cryptography`` EC public key with ``JWK.from_pyca(pub)`` before + passing it in. Returns an :class:`SdJwtMandate[T]` for single tokens + when ``payload_type`` is given, otherwise a list of per-hop effective + payload dicts. + """ + tokens = token.split('~~') + is_single = len(tokens) == 1 + + if is_single: + if not isinstance(key_or_provider, JWK): + raise TypeError( + 'Single mandate verification requires a jwcrypto.jwk.JWK; ' + f'got {type(key_or_provider).__name__}. Wrap ' + 'cryptography EC keys ' + 'with JWK.from_pyca(...).' + ) + if payload_type is None: + raise ValueError( + 'Single mandate verification requires payload_type.' + ) + single_key = key_or_provider + + def public_key_provider(_token: common.ParsedToken) -> JWK: + return single_key + else: + if isinstance(key_or_provider, JWK): + raise ValueError( + 'Chain verification requires a public key provider ' + 'function (ParsedToken -> JWK), not a single JWK.' + ) + public_key_provider = key_or_provider + + final_token = tokens[-1] + if '~' in final_token: + has_kb_jwt = bool(final_token.split('~')[-1].strip()) + if has_kb_jwt and not (expected_aud and expected_nonce): + raise ValueError( + 'The provided presentation token contains a Key Binding ' + 'JWT, but expected_aud and expected_nonce were not ' + 'provided. Both must be supplied to securely verify ' + 'this presentation.' + ) + elif is_single: + raise NotImplementedError( + 'Only SD-JWT formats are currently supported for verification.' + ) + parsed_tokens = [ + common.parse_token(_canonical_chain_segment(t, i, len(tokens))) + for i, t in enumerate(tokens) + ] + + _log_event( + 'verify', + 'before', + { + 'mode': 'single' if is_single else 'chain', + 'num_tokens': len(tokens), + 'payload_type': payload_type.__name__ if payload_type else None, + 'has_aud': bool(expected_aud), + 'has_nonce': bool(expected_nonce), + }, + ) + + payloads = _chain.verify_chain( + tokens=parsed_tokens, + public_key_provider=public_key_provider, + clock_skew_seconds=clock_skew_seconds, + expected_aud=expected_aud, + expected_nonce=expected_nonce, + current_time=current_time, + ) + + _log_event( + 'verify', + 'after', + {'success': True, 'num_payloads': len(payloads)}, + ) + + if is_single: + return SdJwtMandate( + parsed_tokens[0].canonical, + payload_type.model_validate(payloads[0]), + ) + return payloads + + def present( # noqa: PLR0913 (public API: one branch per arg combination) + self, + holder_key: JWK, + mandate_token: str, + payloads: list[Any], + sd: DisclosureMetadata | None = None, + claims_to_disclose: dict[str, Any] | None = None, + nonce: str | None = None, + aud: str | None = None, + hash_mode: _HashMode = 'sd_hash', + ) -> str: + """Append one delegation hop on top of ``mandate_token``. + + Dispatches to :func:`ap2.sdk.sdjwt.kb_sd_jwt.create`; closed mandates + are terminal hops, while open mandates with ``cnf`` delegate further. + + ``hash_mode`` selects how this new hop binds to ``mandate_token``: + - ``"sd_hash"`` (default): commits to the preceding hop's exact + disclosures. Next delegate cannot further redact them. + - ``"issuer_jwt_hash"``: commits only to the preceding issuer-signed + JWT, allowing the next delegate to drop disclosures from it + (draft-gco-oauth-delegate-sd-jwt-00 §5.1.4). + """ + _log_event( + 'create_presentation', + 'before', + { + 'has_claims_to_disclose': claims_to_disclose is not None, + 'has_nonce': bool(nonce), + 'has_aud': bool(aud), + 'hash_mode': hash_mode, + }, + ) + + if (nonce or aud) and not holder_key: + raise ValueError( + 'nonce and aud require the holder_key parameter to be provided.' + ) + + if not payloads: + raise ValueError('payloads list cannot be empty.') + + if claims_to_disclose is not None: + holder_open = SDJWTHolder(mandate_token) + selected_disclosures = [] + # ``_input_disclosures`` is the canonical list of disclosures + # SDJWTHolder parsed from the input token. The sd_jwt library does + # not expose a public equivalent, so we reach in here. See + # SDJWTHolder.__init__ in py-sd-jwt for the contract. + for d in holder_open._input_disclosures: # noqa: SLF001 + decoded = json.loads(b64url_decode(d).decode('utf-8')) + if len(decoded) == _SD_JWT_DISCLOSURE_ARRAY_LEN: + val = decoded[1] + if isinstance(val, dict) and 'vct' in val: + selected_disclosures.append(d) + elif len(decoded) == _SD_JWT_DISCLOSURE_PROPERTY_LEN: + key = decoded[1] + if key == 'cnf' or key in claims_to_disclose: + selected_disclosures.append(d) + jwt_part = mandate_token.split('~', maxsplit=1)[0] + redacted_open_tok = ( + jwt_part + '~' + '~'.join(selected_disclosures) + '~' + ) + wrapped = {'delegate_payload': [claims_to_disclose]} + else: + redacted_open_tok = None + wrapped = {'delegate_payload': [sd_claims_to_disclose(payloads[0])]} + + # Bind against the exact on-wire form of the preceding token. When the + # caller filters disclosures with ``claims_to_disclose``, the + # redacted-and-sent form is what downstream verifiers will re-hash, so + # we must compute ``sd_hash`` over that form — not over the original + # full-disclosure token. + prev_token_for_binding = ( + redacted_open_tok + if redacted_open_tok is not None + else mandate_token + ) + prev_token_parsed = common.parse_token(prev_token_for_binding) + + if not (aud and nonce): + raise ValueError('aud and nonce are required for KB-SD-JWT hops.') + issuer = kb_sd_jwt.create( + prev_token=prev_token_parsed, + holder_key=holder_key, + payload=payloads[0], + aud=aud, + nonce=nonce, + sd=sd, + hash_mode=hash_mode, + ) + + holder = SDJWTHolder(issuer.sd_jwt_issuance) + + holder.create_presentation(claims_to_disclose=wrapped) + pres_jwt = holder.sd_jwt_presentation + + _log_event( + 'create_presentation', + 'after', + { + 'success': True, + 'pres_jwt': pres_jwt, + 'aud': aud, + 'holder_pub': ( + json.loads(holder_key.export_public()) + if (nonce or aud) + else None + ), + }, + ) + + if redacted_open_tok is not None: + open_tok_to_join = ( + redacted_open_tok[:-1] + if redacted_open_tok.endswith('~') + else redacted_open_tok + ) + return f'{open_tok_to_join}~~{pres_jwt}' + mandate_tok_to_join = ( + mandate_token[:-1] if mandate_token.endswith('~') else mandate_token + ) + return f'{mandate_tok_to_join}~~{pres_jwt}' + + def get_closed_mandate_jwt(self, presentation_token: str) -> str: + """Return the closed-mandate JWT (leaf) of a dSD-JWT chain. + + Examples: + - ``""`` -> ``""`` + - ``"~"`` -> ``""`` + - ``"~~~"`` -> ``""`` + - ``"~~~d1~d2~~~"`` -> ``""`` + + Use the SHA-256 of this string as the stable receipt reference, so a + receipt stays bound to the same closed mandate regardless of how many + delegation hops precede it or which open-mandate disclosures were + revealed. + """ + last_segment = presentation_token.rsplit('~~', 1)[-1] + return last_segment.split('~', 1)[0] diff --git a/_review/ap2_sd_jwt.py b/_review/ap2_sd_jwt.py new file mode 100644 index 000000000..ac55ee0e0 --- /dev/null +++ b/_review/ap2_sd_jwt.py @@ -0,0 +1,88 @@ +"""Root SD-JWT primitive (RFC 9901). + +This module owns the root, issuer-signed SD-JWT: the very first token in a +delegation chain. It exposes: + +- ``create(payload, issuer_key, sd=None)`` — sign an SD-JWT. +- ``verify(token, issuer_public_key, ...)`` — verify signature and resolve + disclosures into a plain ``dict``. +""" + +from __future__ import annotations + +from typing import Any + +from ap2.sdk.disclosure_metadata import DisclosureMetadata +from ap2.sdk.sdjwt import common +from jwcrypto.jwk import JWK +from pydantic import BaseModel +from sd_jwt.issuer import SDJWTIssuer +from sd_jwt.verifier import SDJWTVerifier + + +def create( + payload: BaseModel, + issuer_key: JWK, + sd: DisclosureMetadata | None = None, + add_decoy_claims: bool = False, + serialization_format: str = 'compact', +) -> SDJWTIssuer: + """Sign a root SD-JWT for ``payload``. + + The claim is wrapped under ``delegate_payload`` so the same resolver logic + works for both root tokens and KB-SD-JWT[+KB] hops. No ``sd_hash``, + ``iat``, ``aud``, or ``nonce`` is injected — those belong on KB-SD-JWTs. + + Args: + payload: Pydantic model whose fields become the delegate payload. + issuer_key: JWK used to sign the resulting JWT. + sd: Selective-disclosure metadata. ``None`` auto-derives from model + annotations. + add_decoy_claims: Add decoy ``_sd`` digests (RFC 9901 §4.2.5). + serialization_format: SD-JWT serialization format. + + Returns: + An ``SDJWTIssuer`` whose ``.sd_jwt_issuance`` is the compact-serialized + SD-JWT string. + """ + delegate_claims = common.delegate_claims_from_model(payload) + if sd is None: + sd = DisclosureMetadata.from_model(payload) + + sd_claims = common.selectively_disclosable_claims(delegate_claims, sd) + return common.issue_sd_jwt( + claims=sd_claims, + issuer_key=issuer_key, + header_params=common.header_parameters(issuer_key), + add_decoy_claims=add_decoy_claims, + serialization_format=serialization_format, + ) + + +def verify( + token: str, + issuer_public_key: JWK, + expected_aud: str | None = None, + expected_nonce: str | None = None, +) -> dict[str, Any]: + """Verify an SD-JWT signature and return the fully-resolved payload. + + ``issuer_public_key`` MUST be a :class:`jwcrypto.jwk.JWK`. To verify with + a ``cryptography`` EC public key (e.g. extracted from an ``x5c`` cert), + wrap it via ``JWK.from_pyca(pub_key)`` first. + """ + + def cb_get_issuer_key( + _issuer: str, + _header_parameters: dict[str, Any], + ) -> JWK: + return issuer_public_key + + verifier = SDJWTVerifier( + token, + cb_get_issuer_key, + expected_aud=expected_aud, + expected_nonce=expected_nonce, + serialization_format='compact', + ) + return verifier.get_verified_payload() diff --git a/_review/redos.mjs b/_review/redos.mjs new file mode 100644 index 000000000..e186cc4ac --- /dev/null +++ b/_review/redos.mjs @@ -0,0 +1,29 @@ +import { readFileSync } from 'fs'; +const dir = new URL('.', import.meta.url).pathname; +const schema = JSON.parse(readFileSync(dir + '../source/schemas/shopping/ap2_mandate.json', 'utf8')); +const re = new RegExp(schema['$defs'].checkout_mandate.pattern); + +const families = { + 'A: ("a.b.c~~"*n)+"!"': n => 'a.b.c~~'.repeat(n) + '!', + 'B: ("a.b.c~"*n)': n => 'a.b.c~'.repeat(n), + 'C: "a.b.c"+("~a"*n)+"~!"': n => 'a.b.c' + '~a'.repeat(n) + '~!', + 'D: ("a.a.a~a.a.a~~"*n)+"a"': n => 'a.a.a~a.a.a~~'.repeat(n) + 'a', + 'E: ("a.b.c~aa~~"*n)+"~"': n => 'a.b.c~aa~~'.repeat(n) + '~', + 'F: ("a.b.c~a~"*n)+"!"': n => 'a.b.c~a~'.repeat(n) + '!', + 'G: "a.b.c"+("~aaaa"*n)+"."': n => 'a.b.c' + '~aaaa'.repeat(n) + '.', + 'H: ("a..a~~"*n)+"a..a~a"+"!"': n => 'a..a~~'.repeat(n) + 'a..a~a' + '!', + 'I: ("a.b.c~~a.b.c~a"*n)': n => 'a.b.c~~a.b.c~a'.repeat(n), + 'J: ("aaaa.aaaa.aaaa~"*n)+"+"': n => 'aaaa.aaaa.aaaa~'.repeat(n) + '+', +}; +for (const [name, gen] of Object.entries(families)) { + const times = []; + for (const n of [1000, 2000, 4000, 8000, 16000]) { + const s = gen(n); + const t0 = process.hrtime.bigint(); + re.test(s); + const ms = Number(process.hrtime.bigint() - t0) / 1e6; + times.push(`n=${n}(len ${s.length}): ${ms.toFixed(1)}ms`); + if (ms > 3000) { times.push('BLOWUP - stopping family'); break; } + } + console.log(name + '\n ' + times.join(' ')); +} diff --git a/_review/redos.py b/_review/redos.py new file mode 100644 index 000000000..656b19861 --- /dev/null +++ b/_review/redos.py @@ -0,0 +1,29 @@ +import json, re, os, time +d = os.path.dirname(os.path.abspath(__file__)) +schema = json.load(open(os.path.join(d, '../source/schemas/shopping/ap2_mandate.json'))) +r = re.compile(schema['$defs']['checkout_mandate']['pattern']) + +families = { + 'A: ("a.b.c~~"*n)+"!"': lambda n: 'a.b.c~~' * n + '!', + 'B: ("a.b.c~"*n)': lambda n: 'a.b.c~' * n, + 'C: "a.b.c"+("~a"*n)+"~!"': lambda n: 'a.b.c' + '~a' * n + '~!', + 'D: ("a.a.a~a.a.a~~"*n)+"a"': lambda n: 'a.a.a~a.a.a~~' * n + 'a', + 'E: ("a.b.c~aa~~"*n)+"~"': lambda n: 'a.b.c~aa~~' * n + '~', + 'F: ("a.b.c~a~"*n)+"!"': lambda n: 'a.b.c~a~' * n + '!', + 'G: "a.b.c"+("~aaaa"*n)+"."': lambda n: 'a.b.c' + '~aaaa' * n + '.', + 'H: ("a..a~~"*n)+"a..a~a!"': lambda n: 'a..a~~' * n + 'a..a~a!', + 'I: ("a.b.c~~a.b.c~a"*n)': lambda n: 'a.b.c~~a.b.c~a' * n, + 'J: ("aaaa.aaaa.aaaa~"*n)+"+"':lambda n: 'aaaa.aaaa.aaaa~' * n + '+', +} +for name, gen in families.items(): + out = [] + for n in (500, 1000, 2000, 4000, 8000): + s = gen(n) + t0 = time.perf_counter() + r.search(s) + ms = (time.perf_counter() - t0) * 1000 + out.append(f'n={n}(len {len(s)}): {ms:.1f}ms') + if ms > 3000: + out.append('BLOWUP - stopping family') + break + print(name + '\n ' + ' '.join(out)) diff --git a/_review/run_node.mjs b/_review/run_node.mjs new file mode 100644 index 000000000..04e684f9c --- /dev/null +++ b/_review/run_node.mjs @@ -0,0 +1,22 @@ +import { readFileSync } from 'fs'; +const dir = new URL('.', import.meta.url).pathname; +const schema = JSON.parse(readFileSync(dir + '../source/schemas/shopping/ap2_mandate.json', 'utf8')); +const pat = schema['$defs'].checkout_mandate.pattern; +// JSON Schema semantics: unanchored search, ECMA-262 'u'-less regex +const re = new RegExp(pat); +const vectors = JSON.parse(readFileSync(dir + 'vectors.json', 'utf8')); +let fails = 0; +for (const [group, expected] of [['expect_accept', true], ['expect_reject', false]]) { + for (const [name, s] of vectors[group]) { + const got = re.test(s); + if (got !== expected) { fails++; console.log(`FAIL [${group}] ${name}: ${JSON.stringify(s)} -> ${got}`); } + } +} +console.log(fails === 0 ? 'accept/reject tables: ALL PASS (node)' : `${fails} FAILURES (node)`); +for (const [name, s] of vectors.questionable) { + console.log(`Q: ${name}: ${JSON.stringify(s)} -> ${re.test(s) ? 'ACCEPT' : 'reject'}`); +} +// engine semantics: trailing newline +for (const s of ['a.b.c~\n', 'a.b.c\n', 'a.b.c~d~\n']) { + console.log(`NEWLINE node: ${JSON.stringify(s)} -> ${re.test(s)}`); +} diff --git a/_review/run_py.py b/_review/run_py.py new file mode 100644 index 000000000..3d99cd88a --- /dev/null +++ b/_review/run_py.py @@ -0,0 +1,41 @@ +import json, re, os, sys +d = os.path.dirname(os.path.abspath(__file__)) +schema = json.load(open(os.path.join(d, '../source/schemas/shopping/ap2_mandate.json'))) +pat = schema['$defs']['checkout_mandate']['pattern'] +r = re.compile(pat) +vectors = json.load(open(os.path.join(d, 'vectors.json'))) +fails = 0 +for group, expected in (('expect_accept', True), ('expect_reject', False)): + for name, s in vectors[group]: + got = bool(r.search(s)) # jsonschema lib uses re.search + if got != expected: + fails += 1 + print(f'FAIL [{group}] {name}: {s!r} -> {got}') +print('accept/reject tables: ALL PASS (python re.search)' if fails == 0 else f'{fails} FAILURES (python)') +for name, s in vectors['questionable']: + print(f'Q: {name}: {s!r} -> {"ACCEPT" if r.search(s) else "reject"}') +# engine semantics: trailing newline ($ before final \n in python) +for s in ['a.b.c~\n', 'a.b.c\n', 'a.b.c~d~\n', 'a.b.c\n\n', '\na.b.c']: + print(f'NEWLINE py search: {s!r} -> {bool(r.search(s))} fullmatch: {bool(r.fullmatch(s))}') +# jsonschema library behavior if installed +try: + import jsonschema + v = jsonschema.Draft202012Validator({'type': 'string', 'pattern': pat}) + for s in ['a.b.c~\n', 'a.b.c~']: + print(f'jsonschema {jsonschema.__version__}: {s!r} valid={v.is_valid(s)}') +except ImportError: + print('jsonschema lib not installed') +try: + import pydantic + from pydantic import BaseModel, StringConstraints + from typing import Annotated + class M(BaseModel): + x: Annotated[str, StringConstraints(pattern=pat)] + for s in ['a.b.c~\n', 'a.b.c~']: + try: + M(x=s); ok = True + except Exception: + ok = False + print(f'pydantic {pydantic.VERSION}: {s!r} valid={ok}') +except ImportError: + print('pydantic not installed') diff --git a/_review/rustcheck/Cargo.lock b/_review/rustcheck/Cargo.lock new file mode 100644 index 000000000..52dbb4509 --- /dev/null +++ b/_review/rustcheck/Cargo.lock @@ -0,0 +1,144 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "regex" +version = "1.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843bc0191f75f3e22651ae5f1e72939ab2f72a4bc30fa80a066bd66edefc24d4" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "rustcheck" +version = "0.1.0" +dependencies = [ + "regex", + "serde_json", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/_review/rustcheck/Cargo.toml b/_review/rustcheck/Cargo.toml new file mode 100644 index 000000000..007e0eb18 --- /dev/null +++ b/_review/rustcheck/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name="rustcheck" +version="0.1.0" +edition="2021" + +[dependencies] +regex="1.12.2" +serde_json="1" diff --git a/_review/rustcheck/src/main.rs b/_review/rustcheck/src/main.rs new file mode 100644 index 000000000..6d0bac9f1 --- /dev/null +++ b/_review/rustcheck/src/main.rs @@ -0,0 +1,40 @@ +use regex::Regex; + +fn main() { + let schema: serde_json::Value = serde_json::from_str( + &std::fs::read_to_string("../../source/schemas/shopping/ap2_mandate.json").unwrap(), + ) + .unwrap(); + let pat = schema["$defs"]["checkout_mandate"]["pattern"].as_str().unwrap(); + let re = Regex::new(pat).expect("pattern must compile in rust regex crate"); + let vectors: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string("../vectors.json").unwrap()).unwrap(); + let mut fails = 0; + for (group, expected) in [("expect_accept", true), ("expect_reject", false)] { + for v in vectors[group].as_array().unwrap() { + let name = v[0].as_str().unwrap(); + let s = v[1].as_str().unwrap(); + let got = re.is_match(s); + if got != expected { + fails += 1; + println!("FAIL [{}] {}: {:?} -> {}", group, name, s, got); + } + } + } + if fails == 0 { + println!("accept/reject tables: ALL PASS (rust regex 1.12.2)"); + } else { + println!("{} FAILURES (rust)", fails); + } + for v in vectors["questionable"].as_array().unwrap() { + println!( + "Q: {}: {:?} -> {}", + v[0].as_str().unwrap(), + v[1].as_str().unwrap(), + if re.is_match(v[1].as_str().unwrap()) { "ACCEPT" } else { "reject" } + ); + } + for s in ["a.b.c~\n", "a.b.c\n", "a.b.c~d~\n"] { + println!("NEWLINE rust: {:?} -> {}", s, re.is_match(s)); + } +} diff --git a/_review/vectors.json b/_review/vectors.json new file mode 100644 index 000000000..5af2f8068 --- /dev/null +++ b/_review/vectors.json @@ -0,0 +1,47 @@ +{ + "expect_accept": [ + ["bare JWT (back-compat a)", "a.b.c"], + ["realistic bare JWT", "eyJhbGciOiJFUzI1NiJ9.eyJzdWIiOiIxIn0.sig_-123"], + ["RFC9901 no-KB zero disclosures", "a.b.c~"], + ["RFC9901 no-KB one disclosure", "a.b.c~d~"], + ["RFC9901 no-KB two disclosures", "a.b.c~d1~d2~"], + ["KB zero disclosures", "a.b.c~x.y.z"], + ["KB with disclosures", "a.b.c~d~x.y.z"], + ["chain: stripped no-KB seg + final tilde", "a.b.c~d~~e.f.g~"], + ["chain: bare-JWT seg + final tilde", "a.b.c~~e.f.g~"], + ["chain: seg w/ inline KB + final w/ KB", "a.b.c~d~x.y.z~~e.f.g~h~q.w.e"], + ["3-segment chain", "a.b.c~d~~e.f.g~h~~i.j.k~"], + ["chain: KB seg zero-disclosures", "a.b.c~x.y.z~~e.f.g~"] + ], + "expect_reject": [ + ["empty string", ""], + ["lone tilde", "~"], + ["leading tilde", "~a.b.c~"], + ["whitespace inside", "a.b.c ~d~"], + ["trailing space", "a.b.c~ "], + ["standard b64 plus", "a.b.c~d+e~"], + ["standard b64 pad", "a.b.c~d=~"], + ["one-dot JWT", "a.b~d~"], + ["four-part JWT", "a.b.c.d~"], + ["empty disclosure J~~D~", "a.b.c~~~"], + ["trailing tilde after KB", "a.b.c~d~x.y.z~"], + ["trailing tilde after KB (0 discl)", "a.b.c~x.y.z~"], + ["chain sep then nothing", "a.b.c~~"], + ["unstripped seg (triple tilde)", "a.b.c~d~~~e.f.g~"], + ["chain trailing sep", "a.b.c~d~~e.f.g~~"], + ["disclosure after KB", "a.b.c~x.y.z~d~"], + ["empty header", ".b.c~"], + ["empty signature", "a.b.~"], + ["newline embedded", "a.b.c~d~\nx"] + ], + "questionable": [ + ["empty payload bare (old pattern also accepted)", "a..b"], + ["empty payload SD-JWT", "a..b~d~"], + ["KB with empty payload", "a.b.c~d~x..z"], + ["disclosures w/o terminator (old pattern accepted)", "a.b.c~d"], + ["multi disclosures w/o terminator", "a.b.c~d1~d2"], + ["chain final = bare JWT", "a.b.c~~e.f.g"], + ["chain final = J~D no terminator", "a.b.c~~e.f.g~d"], + ["bare JWT then chain of bare JWTs", "a.b.c~~e.f.g~~i.j.k"] + ] +} diff --git a/docs/specification/ap2-mandates.md b/docs/specification/ap2-mandates.md index e9dd07e65..05df25aa3 100644 --- a/docs/specification/ap2-mandates.md +++ b/docs/specification/ap2-mandates.md @@ -241,9 +241,9 @@ credential: `~~...~[]` per [RFC 9901](https://datatracker.ietf.org/doc/html/rfc9901). Presentations produced through delegation (for example by the [AP2 reference implementation](https://github.com/google-agentic-commerce/AP2)) -serialize a *chain* of such tokens joined by `~~`. The schema's `pattern` -admits both forms; it checks syntactic form only — signature, key-binding, -and disclosure verification are defined by the AP2 Protocol Specification. +serialize a *chain* of such tokens joined by `~~`. The schema's `pattern` admits both forms — inside a chain each hop is itself a +well-formed SD-JWT — and checks syntactic form only; signature, key-binding, and +disclosure verification are defined by the AP2 Protocol Specification. ### Canonicalization diff --git a/source/schemas/shopping/ap2_mandate.json b/source/schemas/shopping/ap2_mandate.json index 3b256122d..a1d6c8da9 100644 --- a/source/schemas/shopping/ap2_mandate.json +++ b/source/schemas/shopping/ap2_mandate.json @@ -15,7 +15,7 @@ "title": "Checkout Mandate", "description": "SD-JWT credential in `ap2.checkout_mandate`, proving user authorization for the checkout. Contains the full checkout including `ap2.merchant_authorization`. Accepted serializations (syntactic form only; cryptographic verification is separate): a compact SD-JWT with or without a trailing Key Binding JWT (RFC 9901 `~~...~[]`), or a delegated SD-JWT chain of such tokens joined by `~~` as emitted by the AP2 reference implementation.", "type": "string", - "pattern": "^([A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(~[A-Za-z0-9_-]+)*(~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)?~~)*[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(~[A-Za-z0-9_-]+)*(~|~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)?$" + "pattern": "^(?:[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(?:(?:~[A-Za-z0-9_-]+)+|(?:~[A-Za-z0-9_-]+)*~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)~~)+[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(?:~[A-Za-z0-9_-]+)*(?:~|~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)$|^[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(?:~[A-Za-z0-9_-]+)*(?:~|~[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+)$|^[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]+(?:~[A-Za-z0-9_-]+)*$" }, "ap2_with_merchant_authorization": { "type": "object",