diff --git a/agentscore_commerce/payment/signer.py b/agentscore_commerce/payment/signer.py index 6806936..9e6f171 100644 --- a/agentscore_commerce/payment/signer.py +++ b/agentscore_commerce/payment/signer.py @@ -1,26 +1,42 @@ -"""Network-aware signer extraction from x402 (EVM EIP-3009) credentials. - -Returns `{address, network}` so vendors can pass the network into `capture_wallet(...)` -without inferring it themselves. For Tempo MPP and Solana signers, callers must extract -the signer themselves and pass `signer={address, network}` to `GateClient.check`/`acheck` -directly. +"""Network-aware signer extraction from x402 and MPP payment credentials. + +`extract_payment_signer(x402_payment_header, *, authorization_header=...)` returns +`{address, network}` so vendors can pass the network into `capture_wallet(...)` +without inferring it themselves. Reads from either: + +* the x402 EIP-3009 base64 payload (``payment-signature`` / ``x-payment`` header), + matching ``payload.authorization.from``; or +* the MPP ``Authorization: Payment `` header value, matching the + ``source`` (or ``challenge.source``) DID inside the credential + (``did:pkh:eip155::`` for EVM, ``did:pkh:solana::`` + for Solana). + +Decoded inline (no ``mpp._parsing`` dependency); falls through to ``None`` for +anything malformed. + +The MPP path requires the credential to carry a spec-compliant ``did:pkh`` +source (top-level or under ``challenge``). Credentials that omit the source +field and rely on the Solana TransferChecked-authority fallback (extracting +the signer from the signed-tx payload via ``@solana/kit``) are recovered by +the Node sibling, not by this Python helper — Python has no ``@solana/kit`` +equivalent. Production MPP clients emit the ``did:pkh`` source field, so +this is a non-issue for spec-compliant traffic. """ from __future__ import annotations import base64 import json -import re from dataclasses import dataclass from typing import TYPE_CHECKING, Literal +from agentscore_commerce.identity.address import is_solana_address, is_valid_evm_address, normalize_address from agentscore_commerce.identity.signer import extract_x402_signer if TYPE_CHECKING: from collections.abc import Mapping SignerNetwork = Literal["evm", "solana"] -_EVM_RE = re.compile(r"^0x[0-9a-fA-F]{40}$") @dataclass(frozen=True) @@ -34,14 +50,31 @@ class PaymentSigner: network: SignerNetwork -def extract_payment_signer(x402_payment_header: str | None) -> PaymentSigner | None: - """Decode an x402 header and return `{address, network}` or None. +def extract_payment_signer( + x402_payment_header: str | None = None, + /, + *, + authorization_header: str | None = None, +) -> PaymentSigner | None: + """Decode an x402 or MPP payment header and return ``{address, network}`` or ``None``. + + Tries the x402 base64 payload first (EIP-3009 ``payload.authorization.from``). + Falls through to the MPP ``Authorization: Payment `` header if supplied + (reads ``source`` or ``challenge.source`` DID). - Returns the EVM `from` address with `network='evm'` when the payload is EIP-3009 - shape. Returns None for any malformed/missing header. + Returns ``None`` for any missing, malformed, or unsupported payload. """ - if not x402_payment_header: - return None + if x402_payment_header: + result = _extract_from_x402(x402_payment_header) + if result is not None: + return result + if authorization_header: + return _extract_from_mpp_auth(authorization_header) + return None + + +def _extract_from_x402(x402_payment_header: str) -> PaymentSigner | None: + """Recover the EVM signer from an x402 EIP-3009 base64 payload.""" try: decoded = base64.b64decode(x402_payment_header, validate=False).decode("utf-8") parsed = json.loads(decoded) @@ -49,7 +82,6 @@ def extract_payment_signer(x402_payment_header: str | None) -> PaymentSigner | N return None if not isinstance(parsed, dict): return None - payload = parsed.get("payload") if not isinstance(payload, dict): return None @@ -57,8 +89,46 @@ def extract_payment_signer(x402_payment_header: str | None) -> PaymentSigner | N if not isinstance(authorization, dict): return None sender = authorization.get("from") - if isinstance(sender, str) and _EVM_RE.match(sender): - return PaymentSigner(address=sender.lower(), network="evm") + if isinstance(sender, str) and is_valid_evm_address(sender): + return PaymentSigner(address=normalize_address(sender), network="evm") + return None + + +def _extract_from_mpp_auth(authorization: str) -> PaymentSigner | None: + """Recover the signer from an MPP ``Authorization: Payment `` header value. + + Strips the ``Payment`` scheme prefix (case-insensitive per RFC 7235), base64-decodes + the remainder, parses as JSON, and reads ``source`` or ``challenge.source`` as a + ``did:pkh:eip155::`` / ``did:pkh:solana::`` DID. + """ + if not authorization.lower().startswith("payment "): + return None + token = authorization[len("payment ") :].strip() + if not token: + return None + try: + decoded = base64.b64decode(token, validate=False).decode("utf-8") + credential = json.loads(decoded) + except (ValueError, TypeError): + return None + if not isinstance(credential, dict): + return None + source = credential.get("source") + if not isinstance(source, str): + challenge = credential.get("challenge") + if isinstance(challenge, dict): + source = challenge.get("source") + if not isinstance(source, str): + return None + parts = source.split(":") + if len(parts) < 4 or parts[0] != "did" or parts[1] != "pkh": + return None + family = parts[2] + addr = parts[-1] + if family == "eip155" and is_valid_evm_address(addr): + return PaymentSigner(address=normalize_address(addr), network="evm") + if family == "solana" and is_solana_address(addr): + return PaymentSigner(address=normalize_address(addr), network="solana") return None diff --git a/tests/test_payment_signer.py b/tests/test_payment_signer.py index a9cbf70..84ffba8 100644 --- a/tests/test_payment_signer.py +++ b/tests/test_payment_signer.py @@ -3,6 +3,8 @@ import base64 import json +import pytest + from agentscore_commerce.payment.signer import ( PaymentSigner, extract_payment_signer, @@ -51,6 +53,126 @@ def test_returns_none_when_from_is_not_an_evm_address(self): ) assert extract_payment_signer(header) is None + def test_returns_none_when_decoded_json_is_not_an_object(self): + """x402 payload that decodes to a JSON array/scalar instead of an object.""" + list_header = base64.b64encode(json.dumps([1, 2, 3]).encode()).decode() + assert extract_payment_signer(list_header) is None + + def test_returns_none_when_payload_field_is_not_a_dict(self): + """x402 with `payload: null` or `payload: "string"` is malformed; no signer recoverable.""" + null_header = _encode_x402({"payload": None}) + assert extract_payment_signer(null_header) is None + string_header = _encode_x402({"payload": "oops"}) + assert extract_payment_signer(string_header) is None + + +# ── MPP `Authorization: Payment ` path ──────────────────────────────── +# +# The locked fixtures below are shared with the Node sibling at +# `node-commerce/tests/payment/signer.test.ts`. Both files reference identical +# Authorization header values + expected PaymentSigner outputs. A drift in +# either language (DID parsing, base64 handling, scheme prefix) fails that +# language's test against the locked value. + +# ASCII-safe Authorization header values (literal "Payment " prefix + base64 token). +_MPP_DID_EIP155_TOP_LEVEL = ( + "Payment eyJzb3VyY2UiOiAiZGlkOnBraDplaXAxNTU6NDIxNzoweEFCQ0RlZjEyMzQ1Njc4OTAxMjM0NTY3ODkwMTIzNDU2Nzg5MDEyMzQifQ==" +) +_MPP_DID_SOLANA_CHALLENGE = ( + "Payment " + "eyJjaGFsbGVuZ2UiOiB7InNvdXJjZSI6ICJkaWQ6cGtoOnNvbGFuYTo1ZXlrdDRVc0Z2OFA4TkpkVFJFcFkxdnpxS3FaS3ZkcFVrZkZw" + "OjduUUVneHFFVzFiRHFhVDNrWldhOEtxVWs0V2ZoNFZiY3cifX0=" +) +_MPP_NO_SOURCE = "Payment eyJmb28iOiAiYmFyIn0=" # {"foo": "bar"} — no source field anywhere +_MPP_NON_DICT_JSON = "Payment WzEsIDIsIDNd" # [1, 2, 3] — JSON list, not an object +_MPP_NON_DID_SOURCE = "Payment eyJzb3VyY2UiOiAiaHR0cHM6Ly9leGFtcGxlLmNvbSJ9" # {"source": "https://example.com"} +# {"source": "did:pkh:tezos:NetXdQprcVkpaWU:tz1abc..."} — valid did:pkh shape but unknown family +_MPP_UNKNOWN_FAMILY = "Payment eyJzb3VyY2UiOiAiZGlkOnBraDp0ZXpvczpOZXRYZFFwcmNWa3BhV1U6dHoxYWJjZGVmZ2hpamtsbW5vcCJ9" +# {"source": "did:pkh:eip155:4217:not-an-evm-address"} — valid did:pkh but malformed address +_MPP_MALFORMED_ADDR = "Payment eyJzb3VyY2UiOiAiZGlkOnBraDplaXAxNTU6NDIxNzpub3QtYW4tZXZtLWFkZHJlc3MifQ==" + +_MPP_FIXTURES: list[tuple[str, str, PaymentSigner | None]] = [ + ( + "did_pkh_eip155_top_level_source", + _MPP_DID_EIP155_TOP_LEVEL, + PaymentSigner(address="0xabcdef1234567890123456789012345678901234", network="evm"), + ), + ( + "did_pkh_solana_challenge_source", + _MPP_DID_SOLANA_CHALLENGE, + PaymentSigner(address="7nQEgxqEW1bDqaT3kZWa8KqUk4Wfh4Vbcw", network="solana"), + ), + ("credential_without_source", _MPP_NO_SOURCE, None), + ("credential_not_json_object", _MPP_NON_DICT_JSON, None), + ("source_not_did_pkh", _MPP_NON_DID_SOURCE, None), + ("did_pkh_unknown_family", _MPP_UNKNOWN_FAMILY, None), + ("did_pkh_malformed_address", _MPP_MALFORMED_ADDR, None), + ("bearer_not_payment_scheme", "Bearer abc.def.ghi", None), + ("payment_with_empty_token", "Payment ", None), + ("payment_with_non_base64_token", "Payment !!!not-base64!!!", None), + ("empty_string", "", None), +] + + +class TestExtractPaymentSignerMppPath: + """MPP ``Authorization: Payment `` extraction; locked cross-language fixtures.""" + + @pytest.mark.parametrize( + ("label", "auth_header", "expected"), + _MPP_FIXTURES, + ids=[label for label, _, _ in _MPP_FIXTURES], + ) + def test_locked_cross_language_fixture( + self, + label: str, + auth_header: str, + expected: PaymentSigner | None, + ) -> None: + del label # consumed by parametrize ids + assert extract_payment_signer(authorization_header=auth_header) == expected + + def test_case_insensitive_payment_scheme(self) -> None: + """``payment``, ``PAYMENT``, and ``Payment`` are equivalent per RFC 7235.""" + result_upper = extract_payment_signer( + authorization_header=_MPP_DID_EIP155_TOP_LEVEL.replace("Payment ", "PAYMENT "), + ) + result_lower = extract_payment_signer( + authorization_header=_MPP_DID_EIP155_TOP_LEVEL.replace("Payment ", "payment "), + ) + expected = PaymentSigner(address="0xabcdef1234567890123456789012345678901234", network="evm") + assert result_upper == expected + assert result_lower == expected + + def test_x402_header_takes_precedence_over_mpp(self) -> None: + """When both headers are supplied, the x402 path is tried first.""" + x402_evm = base64.b64encode( + json.dumps({"payload": {"authorization": {"from": EVM_MIXED}}}).encode(), + ).decode() + result = extract_payment_signer(x402_evm, authorization_header=_MPP_DID_SOLANA_CHALLENGE) + assert result == PaymentSigner(address=EVM_LOWER, network="evm") + + def test_mpp_only_when_x402_absent(self) -> None: + """No x402 supplied → MPP path runs.""" + result = extract_payment_signer(None, authorization_header=_MPP_DID_EIP155_TOP_LEVEL) + assert result == PaymentSigner(address="0xabcdef1234567890123456789012345678901234", network="evm") + + def test_no_headers_returns_none(self) -> None: + assert extract_payment_signer() is None + assert extract_payment_signer(None) is None + assert extract_payment_signer(None, authorization_header=None) is None + + def test_does_not_require_mpp_parsing_module(self) -> None: + """Regression: the helper must NOT import ``mpp._parsing`` (private upstream). + + This is a smoke test — we don't try to mock the import absence, just confirm + the helper works without pympp's private parser being involved. The function + body relies only on stdlib (base64 + json) for the MPP path. + """ + # Re-running a fixture confirms no import-time side effects on the MPP path. + result = extract_payment_signer(authorization_header=_MPP_DID_EIP155_TOP_LEVEL) + assert result is not None + assert result.network == "evm" + class TestReadX402PaymentHeader: def test_prefers_payment_signature(self):