diff --git a/agentscore_commerce/payment/__init__.py b/agentscore_commerce/payment/__init__.py index 32f82d0..d374e28 100644 --- a/agentscore_commerce/payment/__init__.py +++ b/agentscore_commerce/payment/__init__.py @@ -9,7 +9,7 @@ build_payment_request_blob, payment_directive, ) -from agentscore_commerce.payment.dispatch import dispatch_settlement_by_network +from agentscore_commerce.payment.dispatch import detect_rail_from_headers, dispatch_settlement_by_network from agentscore_commerce.payment.headers import ( BuildPaymentHeadersInput, PaymentHeadersRail, @@ -124,6 +124,7 @@ "coerce_resource_config", "create_mppx_server", "create_x402_server", + "detect_rail_from_headers", "dispatch_settlement_by_network", "extract_payment_signer", "extract_x402_signer", diff --git a/agentscore_commerce/payment/dispatch.py b/agentscore_commerce/payment/dispatch.py index 312d78e..fef8225 100644 --- a/agentscore_commerce/payment/dispatch.py +++ b/agentscore_commerce/payment/dispatch.py @@ -1,13 +1,44 @@ -"""Settlement dispatch by CAIP-2 network family (eip155→evm, solana→svm).""" +"""Payment dispatch helpers. + +* :func:`detect_rail_from_headers` — detect which payment-protocol family + (x402 vs MPP) the inbound request carries, based on header presence. +* :func:`dispatch_settlement_by_network` — route a settlement payload to + evm vs svm handler based on the CAIP-2 network family in + ``payload.accepted.network``. +""" import inspect -from collections.abc import Awaitable, Callable -from typing import Any, TypeVar, cast +from collections.abc import Awaitable, Callable, Mapping +from typing import Any, Literal, TypeVar, cast T = TypeVar("T") Handler = Callable[[Any], T | Awaitable[T]] +def detect_rail_from_headers(headers: Mapping[str, str]) -> Literal["x402", "mpp"] | None: + """Detect which payment-protocol family the inbound request carries. + + Returns ``"mpp"`` when an ``Authorization`` header starts with the ``Payment`` + scheme (case-insensitive per RFC 7235). Returns ``"x402"`` when a non-empty + ``payment-signature`` or ``x-payment`` header is present. Returns ``None`` + otherwise. + + In practice a client constructs a request with exactly one protocol's headers; + both arriving together is a client bug or misconfigured proxy. The helper + checks MPP first so the rare degenerate case resolves to MPP. Empty header + values are treated as absent. Header-name lookups are case-insensitive + (RFC 7230 §3.2). The narrower rail naming (``"tempo"`` vs ``"solana"`` inside + MPP) is merchant-side, derived from the credential body, not this helper. + """ + lower = {k.lower(): v for k, v in headers.items()} + auth = lower.get("authorization") or "" + if auth.lower().startswith("payment "): + return "mpp" + if lower.get("payment-signature") or lower.get("x-payment"): + return "x402" + return None + + async def dispatch_settlement_by_network( payload: Any, *, diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py new file mode 100644 index 0000000..24024a4 --- /dev/null +++ b/tests/test_dispatch.py @@ -0,0 +1,64 @@ +"""Tests for ``agentscore_commerce.payment.dispatch.detect_rail_from_headers``. + +The fixture corpus below is locked as the cross-language contract with the +Node sibling at ``node-commerce/tests/payment/detect_rail_from_headers.test.ts``. +Both files reference identical header maps + expected results. A drift in either +language (case-handling, empty-value treatment, scheme-prefix matching) fails +that language's test against the locked value. +""" + +from __future__ import annotations + +import pytest + +from agentscore_commerce.payment import detect_rail_from_headers + +# Cross-language fixtures: (label, headers_dict, expected_rail). +_FIXTURES: list[tuple[str, dict[str, str], str | None]] = [ + ("empty", {}, None), + ("payment_signature_only", {"payment-signature": "abc"}, "x402"), + ("x_payment_only", {"x-payment": "abc"}, "x402"), + ("authorization_payment", {"authorization": "Payment abc"}, "mpp"), + ("authorization_bearer", {"authorization": "Bearer xyz"}, None), + ("authorization_lowercase_scheme", {"authorization": "payment abc"}, "mpp"), + ("authorization_uppercase_name", {"Authorization": "Payment abc"}, "mpp"), + ("x_payment_uppercase_name", {"X-Payment": "abc"}, "x402"), + ("empty_values_dont_count", {"payment-signature": "", "x-payment": ""}, None), + ( + "mpp_wins_when_both_present", + {"x-payment": "abc", "authorization": "Payment xyz"}, + "mpp", + ), + ("payment_without_space_is_not_mpp", {"authorization": "PaymentNoSpace"}, None), + ("payment_with_only_space_is_mpp", {"authorization": "Payment "}, "mpp"), + ("mixed_case_authorization_name", {"AUTHORIZATION": "Payment abc"}, "mpp"), + ("authorization_uppercase_scheme", {"authorization": "PAYMENT abc"}, "mpp"), +] + + +@pytest.mark.parametrize( + ("label", "headers", "expected"), + _FIXTURES, + ids=[label for label, _, _ in _FIXTURES], +) +def test_locked_cross_language_fixture( + label: str, + headers: dict[str, str], + expected: str | None, +) -> None: + del label # `label` is consumed by parametrize ids; bind locally so linters don't flag it. + """Each fixture header set maps to the locked cross-language rail value.""" + assert detect_rail_from_headers(headers) == expected + + +def test_returns_x402_for_non_string_truthy_value() -> None: + """Any non-empty header value is treated as present (no validation of contents).""" + assert detect_rail_from_headers({"x-payment": "0"}) == "x402" + + +def test_does_not_mutate_input_headers() -> None: + headers = {"X-Payment": "abc", "Authorization": "Payment xyz"} + detect_rail_from_headers(headers) + # Keys preserved verbatim; helper only reads via a lowercase projection. + assert "X-Payment" in headers + assert "Authorization" in headers