From 7b8c71fe1047ae5c362e29ff10c3c9574883c7a6 Mon Sep 17 00:00:00 2001 From: vvillait88 Date: Thu, 14 May 2026 08:30:03 -0700 Subject: [PATCH] feat: extend extract_payment_signer to MPP credentials Adds an `authorization_header` kwarg to the public signer helper. When supplied, decodes the `Authorization: Payment ` header inline (no `mpp._parsing` dependency) and reads `source` or `challenge.source` as a `did:pkh:eip155::` / `did:pkh:solana::` DID. Returns `{address, network}` matching the existing x402 path. The positional x402 arg keeps the existing signature; merchants who only care about x402 see no change. Merchants who previously imported `mpp._parsing.parse_authorization` to recover the MPP signer can now use the public helper. Tests lock 11 cross-language MPP fixtures (top-level source, nested challenge.source, defensive paths: non-dict JSON, non-did:pkh source, unknown family, malformed address, malformed base64, bearer scheme, empty token, empty string). Parametrized so multiple drifts surface independently. Adds x402 defensive coverage (JSON array, null payload, string payload). Co-Authored-By: Claude Opus 4.7 (1M context) --- agentscore_commerce/payment/signer.py | 104 ++++++++++++++++++---- tests/test_payment_signer.py | 122 ++++++++++++++++++++++++++ 2 files changed, 209 insertions(+), 17 deletions(-) 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):