feat: extend extract_payment_signer to MPP credentials

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 <base64>`` header value, matching the
+  ``source`` (or ``challenge.source``) DID inside the credential
+  (``did:pkh:eip155:<chain>:<addr>`` for EVM, ``did:pkh:solana:<genesis>:<addr>``
+  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,31 +50,85 @@ 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 <base64>`` 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)
     except (ValueError, TypeError):
         return None
     if not isinstance(parsed, dict):
         return None
-
     payload = parsed.get("payload")
     if not isinstance(payload, dict):
         return None
     authorization = payload.get("authorization")
     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 <base64>`` 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:<chain>:<addr>`` / ``did:pkh:solana:<genesis>:<addr>`` 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
 
 

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 <base64>` 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 <base64>`` 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):