Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .cspell/custom-words.txt
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,7 @@ ropeproject
RPCURL
Rulebook
screenreaders
SECP
setlocal
sharedpref
Shopcider
Expand Down
33 changes: 28 additions & 5 deletions code/sdk/python/ap2/sdk/checkout_mandate_chain.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from ap2.sdk.generated.open_checkout_mandate import OpenCheckoutMandate
from ap2.sdk.generated.types.checkout import Checkout
from ap2.sdk.mandate import _log_event
from ap2.sdk.utils import b64url_decode
from ap2.sdk.utils import b64url_decode, compute_sha256_b64url
from pydantic import ValidationError


Expand Down Expand Up @@ -48,11 +48,20 @@ def verify(
) -> list[str]:
"""Verifies the constraints of the checkout mandate chain.

The closed Checkout Mandate is bound to one specific Checkout JWT
through its ``checkout_hash`` claim. This method derives the hash from
the ``checkout_jwt`` it is asked to verify and compares it with that
claim, so the binding is always checked and cannot be skipped by
omitting an argument.

Args:
expected_checkout_hash: An optional hash to check against the closed
mandate's checkout_hash.
checkout_jwt: The JWT containing the checkout details, required for
verifying open mandate constraints.
expected_checkout_hash: An optional hash the caller already expects
the closed mandate to be bound to. It is an additional consistency
assertion only; it never substitutes for the hash derived from
``checkout_jwt``.
checkout_jwt: The serialized JWT containing the checkout details.
Required, both to bind the closed mandate and to verify open
mandate constraints.

Returns:
A list of strings describing any violations found.
Expand Down Expand Up @@ -82,13 +91,27 @@ def verify(
violations.append(str(e))
return violations

# The closed mandate is bound to the exact serialized Checkout JWT
# whose hash is recorded in checkout_hash. Derive that hash from the
# checkout actually presented for verification rather than trusting a
# value supplied by the caller.
presented_checkout_hash = compute_sha256_b64url(checkout_jwt)
if presented_checkout_hash != self.closed_mandate.checkout_hash:
violations.append(
'Checkout checkout_hash mismatch: computed'
f' {presented_checkout_hash}, got'
f' {self.closed_mandate.checkout_hash}'
)

violations.extend(
check_checkout_constraints(
self.open_mandate,
checkout,
)
)

# Secondary, caller-driven consistency assertion. Retained for
# backwards compatibility; the binding above does not depend on it.
if (
expected_checkout_hash is not None
and expected_checkout_hash != self.closed_mandate.checkout_hash
Expand Down
200 changes: 196 additions & 4 deletions code/sdk/python/ap2/tests/checkout_mandate_chain_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
Item as MandateItem,
)
from ap2.sdk.generated.types.merchant import Merchant
from ap2.sdk.utils import compute_sha256_b64url
from ap2.tests.conftest import make_checkout_jwt, make_cnf, make_line_item
from cryptography.hazmat.primitives.asymmetric import ec

Expand Down Expand Up @@ -43,12 +44,13 @@ def test_checkout_chain_verify_constraint_violation():
),
CheckoutMandate(
checkout_jwt=checkout_jwt,
checkout_hash='hash',
checkout_hash=compute_sha256_b64url(checkout_jwt),
),
]
chain = CheckoutMandateChain.parse(payloads)
violations = chain.verify(checkout_jwt=checkout_jwt)
assert any('not in allowed list' in v for v in violations)
assert not any('checkout_hash mismatch' in v for v in violations)


def test_checkout_chain_hash_mismatch():
Expand All @@ -64,7 +66,7 @@ def test_checkout_chain_hash_mismatch():
),
CheckoutMandate(
checkout_jwt=checkout_jwt,
checkout_hash='actual_hash',
checkout_hash=compute_sha256_b64url(checkout_jwt),
),
]
chain = CheckoutMandateChain.parse(payloads)
Expand All @@ -73,6 +75,7 @@ def test_checkout_chain_hash_mismatch():
)
assert len(violations) == 1
assert 'Checkout checkout_hash mismatch' in violations[0]
assert 'expected expected_hash' in violations[0]


def test_full_checkout_end_to_end(
Expand Down Expand Up @@ -101,7 +104,10 @@ def test_full_checkout_end_to_end(
holder_key=agent_key,
mandate_token=open_tok,
payloads=[
CheckoutMandate(checkout_jwt=checkout_jwt, checkout_hash='hash')
CheckoutMandate(
checkout_jwt=checkout_jwt,
checkout_hash=compute_sha256_b64url(checkout_jwt),
)
],
aud='merchant',
nonce='merchant-nonce',
Expand Down Expand Up @@ -141,7 +147,10 @@ def test_checkout_line_items_constraint():
],
cnf=make_cnf(_DUMMY_KEY.public_key()),
),
CheckoutMandate(checkout_jwt=checkout_jwt, checkout_hash='hash'),
CheckoutMandate(
checkout_jwt=checkout_jwt,
checkout_hash=compute_sha256_b64url(checkout_jwt),
),
]
chain = CheckoutMandateChain.parse(payloads)
violations = chain.verify(checkout_jwt=checkout_jwt)
Expand Down Expand Up @@ -172,3 +181,186 @@ def test_checkout_fields_parsed():
assert len(checkout.line_items) == 1
assert checkout.line_items[0].item.id == 'item-1'
assert checkout.line_items[0].quantity == 2


def _bound_chain(checkout_jwt: str) -> CheckoutMandateChain:
"""Build a chain whose closed mandate is bound to ``checkout_jwt``."""
return CheckoutMandateChain(
open_mandate=OpenCheckoutMandate(
constraints=[], cnf=make_cnf(_DUMMY_KEY.public_key())
),
closed_mandate=CheckoutMandate(
checkout_jwt=checkout_jwt,
checkout_hash=compute_sha256_b64url(checkout_jwt),
),
)


def test_checkout_binding_matching_checkout_passes():
"""The Checkout JWT the closed mandate is bound to verifies cleanly."""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
chain = _bound_chain(checkout_a_jwt)
assert chain.verify(checkout_jwt=checkout_a_jwt) == []


def test_checkout_binding_rejects_other_checkout_without_expected_hash():
"""A different Checkout JWT is rejected with no caller-supplied hash.

This is the regression anchor: the closed mandate is bound to checkout A,
checkout B is presented for verification, and no
``expected_checkout_hash`` is supplied. Before the binding fix verify()
never derived the hash itself, so this returned no violations.
"""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
checkout_b_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_b',
)
assert checkout_a_jwt != checkout_b_jwt

chain = _bound_chain(checkout_a_jwt)

violations = chain.verify(checkout_jwt=checkout_b_jwt)

assert any('checkout_hash mismatch' in v for v in violations)


def test_checkout_chain_rejects_internally_inconsistent_closed_mandate():
"""A closed mandate whose own jwt and hash disagree is rejected.

Here the presented checkout is the closed mandate's own ``checkout_jwt``,
so the two execution contexts agree. The mandate is nonetheless invalid
because its ``checkout_hash`` records a different checkout, meaning it was
never bound to the checkout it carries.
"""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
checkout_b_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_b',
)
chain = CheckoutMandateChain(
open_mandate=OpenCheckoutMandate(
constraints=[], cnf=make_cnf(_DUMMY_KEY.public_key())
),
closed_mandate=CheckoutMandate(
checkout_jwt=checkout_a_jwt,
checkout_hash=compute_sha256_b64url(checkout_b_jwt),
),
)
# The presented checkout matches the mandate's embedded checkout exactly.
assert chain.closed_mandate.checkout_jwt == checkout_a_jwt

violations = chain.verify(checkout_jwt=checkout_a_jwt)

assert any('checkout_hash mismatch' in v for v in violations)


def test_checkout_binding_not_skipped_when_expected_hash_agrees():
"""A caller-supplied hash cannot mask a mismatched presented checkout.

The caller passes the closed mandate's own checkout_hash, so the optional
consistency assertion is satisfied. The derived binding check must still
reject the mismatched checkout.
"""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
checkout_b_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_b',
)
chain = _bound_chain(checkout_a_jwt)

violations = chain.verify(
expected_checkout_hash=chain.closed_mandate.checkout_hash,
checkout_jwt=checkout_b_jwt,
)

assert any('checkout_hash mismatch' in v for v in violations)


def test_checkout_binding_reported_with_constraints_satisfied():
"""Binding failure is reported even when open constraints all pass."""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Good Store'),
checkout_id='chk_a',
)
checkout_b_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Good Store'),
checkout_id='chk_b',
)
chain = CheckoutMandateChain(
open_mandate=OpenCheckoutMandate(
constraints=[
AllowedMerchants(
allowed=[Merchant(id='m-1', name='Good Store')],
),
],
cnf=make_cnf(_DUMMY_KEY.public_key()),
),
closed_mandate=CheckoutMandate(
checkout_jwt=checkout_a_jwt,
checkout_hash=compute_sha256_b64url(checkout_a_jwt),
),
)

violations = chain.verify(checkout_jwt=checkout_b_jwt)

assert not any('not in allowed list' in v for v in violations)
assert any('checkout_hash mismatch' in v for v in violations)


def test_checkout_binding_does_not_leak_checkout_jwt():
"""The mismatch violation does not embed the Checkout JWT contents."""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
checkout_b_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_b',
)
chain = _bound_chain(checkout_a_jwt)

violations = chain.verify(checkout_jwt=checkout_b_jwt)

assert violations
for violation in violations:
assert checkout_b_jwt not in violation
assert checkout_a_jwt not in violation


def test_checkout_binding_missing_checkout_jwt_still_fails():
"""A missing Checkout JWT remains fail-closed."""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
chain = _bound_chain(checkout_a_jwt)

violations = chain.verify(checkout_jwt=None)

assert any('checkout_jwt is required' in v for v in violations)


def test_checkout_binding_malformed_checkout_jwt_still_fails():
"""A malformed Checkout JWT remains fail-closed."""
checkout_a_jwt = make_checkout_jwt(
merchant=Merchant(id='m-1', name='Store'),
checkout_id='chk_a',
)
chain = _bound_chain(checkout_a_jwt)

violations = chain.verify(checkout_jwt='not-a-jwt')

assert any('Malformed checkout_jwt' in v for v in violations)
Loading