Payments security hardening

src/bedrock_agentcore/payments/_validation.py

@@ -1,7 +1,9 @@
 """Shared validation for payments SDK inputs."""
 
+import functools
+import inspect
 import re
-from typing import Optional
+from typing import Any, Optional
 
 _PERMIT2_ALLOWANCE_LIMIT_PATTERN = re.compile(r"[0-9]{1,78}\Z")
 
@@ -16,3 +18,60 @@ def validate_permit2_allowance_limit(value: Optional[str]) -> None:
         )
     if not _PERMIT2_ALLOWANCE_LIMIT_PATTERN.fullmatch(value) or int(value) <= 0:
         raise ValueError(f"permit2_allowance_limit must be a positive ASCII integer of 1-78 digits, got {value!r}")
+
+
+def is_async_callback(callback: Any) -> bool:
+    """True if the callback ultimately resolves to an async coroutine function.
+
+    Covers the wrappings a plain ``inspect.iscoroutinefunction`` can miss: callable
+    objects with an ``async def __call__``, ``functools.partial`` (whose async-ness is
+    not reliably visible through the partial across Python versions), and partials of
+    either. ``callable()`` cannot distinguish a sync callback from an async one, so it is
+    never a safe gate on a money path: an unawaited coroutine object is truthy.
+    """
+    target = callback
+    while isinstance(target, functools.partial):
+        target = target.func
+    if inspect.iscoroutinefunction(target):
+        return True
+    call = getattr(target, "__call__", None)  # noqa: B004
+    return inspect.iscoroutinefunction(call)
+
+
+class PreSignPolicyRejected(Exception):
+    """Raised when ``pre_sign_policy`` blocks a payment before signing.
+
+    Carries no sensitive detail; callers translate it into the integration's own
+    payment error type so the signing path fails closed.
+    """
+
+
+def apply_pre_sign_policy(policy: Optional[Any], payment_required_request: Any) -> None:
+    """Enforce ``pre_sign_policy`` at a signing chokepoint. Fail-closed.
+
+    Shared by the Strands plugin and the LangGraph middleware so both paths enforce the
+    gate with identical semantics. The policy must return ``True`` exactly to allow
+    signing; any other value blocks (so a policy that returns a reason dict, a status
+    string, or forgets to return at all blocks rather than silently approving). Raising
+    from the policy also blocks.
+
+    Args:
+        policy: The configured ``pre_sign_policy`` callback, or ``None`` (no-op).
+        payment_required_request: The 402 request handed to the signer (carries ``payTo``
+            and amount). Must not be ``None`` — the caller is responsible for never
+            invoking signing with an unresolved request.
+
+    Raises:
+        PreSignPolicyRejected: If the policy returns anything other than ``True``, raises,
+            or the request is ``None``.
+    """
+    if policy is None:
+        return
+    if payment_required_request is None:
+        raise PreSignPolicyRejected("pre_sign_policy cannot evaluate an unresolved (None) payment request.")
+    try:
+        allowed = policy(payment_required_request)
+    except Exception as exc:  # fail-closed: a policy error blocks signing
+        raise PreSignPolicyRejected(f"pre_sign_policy raised {type(exc).__name__}.") from exc
+    if allowed is not True:
+        raise PreSignPolicyRejected("pre_sign_policy did not return True; blocking before signing.")

src/bedrock_agentcore/payments/integrations/config.py

@@ -1,11 +1,47 @@
 """Configuration for AgentCore Payments integrations (Strands and LangGraph)."""
 
+import logging
 from dataclasses import dataclass, field
 from typing import Any, Callable, Dict, List, Optional
 
-from .._validation import validate_permit2_allowance_limit
+from .._validation import is_async_callback, validate_permit2_allowance_limit
 from .handlers import PaymentResponseHandler
 
+logger = logging.getLogger(__name__)
+
+
+def resolve_authoritative_user_id(configured_user_id: Optional[str], requested_user_id: Optional[str]) -> Optional[str]:
+    """Resolve the effective user_id for a payment tool call, config-authoritative.
+
+    Defense-in-depth: the Strands plugin and LangGraph tools expose
+    ``user_id`` as an LLM-controllable argument. A prompt-injected agent could supply a
+    different ``user_id`` to attempt an identity override. The managed service already
+    scopes every read to the server-resolved identity, so this never crossed a tenant
+    boundary -- but the SDK should not let the model attempt it in the first place.
+
+    This helper makes the configured identity authoritative: a tool-supplied ``user_id``
+    that differs from the configured one is ignored (and logged). When no identity is
+    configured (bearer/JWT auth, where the server resolves the user from the token), the
+    requested value is passed through unchanged, since there is nothing to override.
+
+    Args:
+        configured_user_id: The identity from plugin/middleware config (authoritative).
+        requested_user_id: The user_id supplied on the tool call (LLM-controllable).
+
+    Returns:
+        The configured identity when set; otherwise the (stripped) requested value.
+    """
+    requested = requested_user_id.strip() if requested_user_id else None
+    if configured_user_id:
+        if requested and requested != configured_user_id:
+            logger.warning(
+                "Ignoring tool-supplied user_id %r that differs from the configured identity; "
+                "using the configured identity (defense-in-depth).",
+                requested,
+            )
+        return configured_user_id
+    return requested
+
 
 @dataclass
 class AgentCorePaymentsPluginConfig:
@@ -61,6 +97,31 @@ class AgentCorePaymentsPluginConfig:
             sponsors fees via methodDetails.feePayer; when the seller does not, the service
             signs only if this is True and otherwise fails validation. None (default) leaves
             the protocol default in place and the field unsent. Ignored for x402.
+        allow_unmarked_402_detection: Whether to infer an x402 payment challenge from raw
+            tool output that does NOT carry the trusted ``PAYMENT_REQUIRED:`` marker
+            (LangGraph only). Default True (current behavior). When False, a payment is
+            derived only from the integration's own payment-bearing tools via the marker,
+            so arbitrary tool content can never mint a signed authorization to an
+            actor-chosen recipient. CWE-863: set False to harden, or keep
+            True and enforce ``pre_sign_policy``.
+        pre_sign_policy: Optional synchronous callback invoked at the signing chokepoint of
+            BOTH integrations (Strands plugin and LangGraph middleware), on every detection
+            path (marked, unmarked, and error-retry), before any payment is signed. Receives
+            the payment-required request and must return ``True`` exactly to allow signing;
+            any other return value blocks (so a reason dict, a status string, or a missing
+            return all fail closed). Raising from the callback also blocks. Must be
+            synchronous: an async policy is rejected at construction, because it could not be
+            awaited here and an unawaited coroutine is truthy (fail-open). Use it to enforce a
+            recipient/amount allowlist regardless of the ``allow_unmarked_402_detection``
+            default. None (default) applies no extra gate.
+
+            The request is a dict of the shape ``{"statusCode", "headers", "body"}``. Where
+            ``payTo`` and amount live depends on the protocol/version, so a robust policy must
+            handle all four: x402 v1 dict body -> ``body["accepts"][i]["payTo"]``; x402 v1
+            string body -> inside the JSON string in ``body``; x402 v2 -> base64 in
+            ``headers["Payment-Required"]``; MPP -> base64url in the ``WWW-Authenticate``
+            challenge. A policy that only reads ``body["accepts"]`` fail-closes every v2 and
+            MPP payment.
     """
 
     payment_manager_arn: str
@@ -86,6 +147,8 @@ class AgentCorePaymentsPluginConfig:
     max_error_retries: int = 3
     buyer_pays_gas_fees: Optional[bool] = None
     permit2_allowance_limit: Optional[str] = None
+    allow_unmarked_402_detection: bool = True
+    pre_sign_policy: Optional[Callable[[Any], bool]] = None
 
     def __post_init__(self) -> None:
         """Validate configuration after initialization."""
@@ -152,6 +215,20 @@ def __post_init__(self) -> None:
                 f"buyer_pays_gas_fees must be a boolean or None, got {type(self.buyer_pays_gas_fees).__name__}"
             )
 
+        if not isinstance(self.allow_unmarked_402_detection, bool):
+            raise ValueError(
+                "allow_unmarked_402_detection must be a boolean, got "
+                f"{type(self.allow_unmarked_402_detection).__name__}"
+            )
+        if self.pre_sign_policy is not None and not callable(self.pre_sign_policy):
+            raise ValueError(f"pre_sign_policy must be callable, got {type(self.pre_sign_policy).__name__}")
+        if self.pre_sign_policy is not None and is_async_callback(self.pre_sign_policy):
+            raise TypeError(
+                "pre_sign_policy must be a synchronous callable. An async policy cannot be awaited at "
+                "the signing chokepoint, so it would fail open (the unawaited coroutine is truthy and the "
+                "payment would sign). Wrap any async work in a synchronous policy, or run it ahead of time."
+            )
+
         validate_permit2_allowance_limit(self.permit2_allowance_limit)
 
     def update_payment_session_id(self, payment_session_id: str) -> None:

src/bedrock_agentcore/payments/integrations/langgraph/middleware.py

@@ -1,8 +1,6 @@
 """AgentCorePaymentsMiddleware for LangGraph agents."""
 
 import asyncio
-import functools
-import inspect
 import json
 import logging
 import time
@@ -15,6 +13,11 @@
 from langgraph.prebuilt.tool_node import ToolCallRequest
 from langgraph.types import Command
 
+from bedrock_agentcore.payments._validation import (
+    PreSignPolicyRejected,
+    apply_pre_sign_policy,
+    is_async_callback,
+)
 from bedrock_agentcore.payments.integrations.error_messages import get_payment_error_message
 from bedrock_agentcore.payments.integrations.handlers import (
     GenericPaymentHandler,
@@ -240,7 +243,7 @@ def _detect_402(self, request: ToolCallRequest, result: ToolMessage) -> Optional
             status_code = detection_handler.extract_status_code(prepared)
 
         # Lenient fallback if marker detection didn't find 402
-        if status_code != 402 and not has_custom_handler:
+        if status_code != 402 and not has_custom_handler and self.config.allow_unmarked_402_detection:
             fallback = self._fallback_detect_402(result.content)
             if fallback is not None:
                 status_code = 402
@@ -336,6 +339,11 @@ def _check_retry_rejection(
 
             _rh = GenericPaymentHandler()
             retry_status = _rh.extract_status_code(retry_prepared)
+            # Not gated by allow_unmarked_402_detection: this runs AFTER the payment is
+            # signed and sent, on the merchant's own response, so it closes no attack
+            # surface. Gating it would let an unmarked post-payment rejection be returned to
+            # the model as a successful tool result -- money moved, merchant refused, agent
+            # told it worked. The detection flag belongs only on the pre-payment sites.
             if retry_status != 402:
                 fallback = self._fallback_detect_402(retry_result.content)
                 if fallback is not None:
@@ -383,6 +391,23 @@ def _generate_payment_header(self, payment_required_request: Dict[str, Any]) ->
         The payment protocol (x402 or MPP) is detected by PaymentManager from the 402
         response, so this method is protocol-agnostic.
         """
+        # An unresolved request must never reach the signer. If _extract_payment_request
+        # raised (e.g. a custom handler threw), the caller previously passed ``{}`` here,
+        # against which an allowlist like ``all(a["payTo"] == TRUSTED for a in accepts)``
+        # resolves ``accepts`` to ``[]`` and ``all([])`` is True -- the pre-sign gate would
+        # pass on empty input. Fail closed instead.
+        if not payment_required_request:
+            raise PaymentError("Cannot sign a payment from an unresolved payment-required request.")
+
+        # Defense-in-depth pre-sign gate. Runs on every signing path (marked, unmarked, and
+        # error-retry) before any value moves. Lets integrators enforce a payTo/amount
+        # allowlist regardless of the detection default. Fail-closed: the policy must return
+        # True exactly, or signing is blocked (a raise or any non-True value blocks).
+        try:
+            apply_pre_sign_policy(self.config.pre_sign_policy, payment_required_request)
+        except PreSignPolicyRejected as exc:
+            raise PaymentError(f"pre_sign_policy blocked the payment before signing ({exc}).") from exc
+
         if self.config.payment_instrument_id is None:
             raise PaymentInstrumentConfigurationRequired("payment_instrument_id is required for payments.")
         if self.config.payment_session_id is None:
@@ -422,18 +447,10 @@ def _create_auto_session(self) -> None:
     def _is_async_callback(callback: Any) -> bool:
         """True if the callback ultimately resolves to an async coroutine function.
 
-        Covers the wrappings a plain ``inspect.iscoroutinefunction`` can miss: callable
-        objects with an ``async def __call__``, ``functools.partial`` (whose async-ness is
-        not reliably visible through the partial across Python versions), and partials of
-        either. (``callable()`` — ruff B004's suggestion — cannot distinguish sync from async.)
+        Delegates to the shared :func:`is_async_callback` so async-detection has one
+        source of truth across the config validation and the integrations.
         """
-        target = callback
-        while isinstance(target, functools.partial):
-            target = target.func
-        if inspect.iscoroutinefunction(target):
-            return True
-        call = getattr(target, "__call__", None)  # noqa: B004
-        return inspect.iscoroutinefunction(call)
+        return is_async_callback(callback)
 
     @staticmethod
     def _error_tool_message(request: ToolCallRequest, exception: Exception) -> ToolMessage:
@@ -590,7 +607,7 @@ def _invoke_error_handler(
 
             try:
                 if payment_header is None:
-                    payment_header = self._generate_payment_header(payment_required_request or {})
+                    payment_header = self._generate_payment_header(payment_required_request)
 
                 inject_error = self._inject_for_error_retry(request, tool_name, tool_args, payment_header)
                 if inject_error is not None:
@@ -727,9 +744,7 @@ async def _ainvoke_error_handler(
 
             try:
                 if payment_header is None:
-                    payment_header = await asyncio.to_thread(
-                        self._generate_payment_header, payment_required_request or {}
-                    )
+                    payment_header = await asyncio.to_thread(self._generate_payment_header, payment_required_request)
 
                 inject_error = self._inject_for_error_retry(request, tool_name, tool_args, payment_header)
                 if inject_error is not None:

src/bedrock_agentcore/payments/integrations/langgraph/tools.py

@@ -7,6 +7,8 @@
 import httpx
 from langchain.tools import tool
 
+from ..config import resolve_authoritative_user_id
+
 logger = logging.getLogger(__name__)
 
 
@@ -82,7 +84,7 @@ def get_payment_instrument(
 
         Args:
             payment_instrument_id: Instrument ID (falls back to middleware config).
-            user_id: User ID (falls back to middleware config).
+            user_id: User ID. The configured identity is authoritative; a differing value is ignored.
             payment_connector_id: Connector ID (optional).
 
         Returns:
@@ -91,7 +93,7 @@ def get_payment_instrument(
         resolved_id = (
             payment_instrument_id.strip() if payment_instrument_id else None
         ) or middleware.config.payment_instrument_id
-        resolved_user = (user_id.strip() if user_id else None) or middleware.config.user_id
+        resolved_user = resolve_authoritative_user_id(middleware.config.user_id, user_id)
         return middleware.payment_manager.get_payment_instrument(
             user_id=resolved_user,
             payment_instrument_id=resolved_id,
@@ -114,15 +116,15 @@ def list_payment_instruments(
         """List all payment instruments for a user.
 
         Args:
-            user_id: User ID (falls back to middleware config).
+            user_id: User ID. The configured identity is authoritative; a differing value is ignored.
             payment_connector_id: Filter by connector (optional).
             max_results: Maximum results to return (default 100).
             next_token: Pagination token (optional).
 
         Returns:
             Dictionary with paymentInstruments list and optional nextToken.
         """
-        resolved_user = (user_id.strip() if user_id else None) or middleware.config.user_id
+        resolved_user = resolve_authoritative_user_id(middleware.config.user_id, user_id)
         return middleware.payment_manager.list_payment_instruments(
             user_id=resolved_user,
             payment_connector_id=payment_connector_id,
@@ -151,12 +153,12 @@ def get_payment_instrument_balance(
             chain: Blockchain chain (e.g., BASE_SEPOLIA, SOLANA_DEVNET).
             token: Token to query (e.g., USDC).
             payment_connector_id: Connector ID (falls back to config).
-            user_id: User ID (falls back to config).
+            user_id: User ID. The configured identity is authoritative; a differing value is ignored.
 
         Returns:
             Dictionary with balance information.
         """
-        resolved_user = (user_id.strip() if user_id else None) or middleware.config.user_id
+        resolved_user = resolve_authoritative_user_id(middleware.config.user_id, user_id)
         resolved_connector = payment_connector_id or middleware.config.payment_connector_id
         return middleware.payment_manager.get_payment_instrument_balance(
             payment_connector_id=resolved_connector,
@@ -181,15 +183,15 @@ def get_payment_session(
 
         Args:
             payment_session_id: Session ID (falls back to middleware config).
-            user_id: User ID (falls back to middleware config).
+            user_id: User ID. The configured identity is authoritative; a differing value is ignored.
 
         Returns:
             Payment session details dictionary.
         """
         resolved_id = (
             payment_session_id.strip() if payment_session_id else None
         ) or middleware.config.payment_session_id
-        resolved_user = (user_id.strip() if user_id else None) or middleware.config.user_id
+        resolved_user = resolve_authoritative_user_id(middleware.config.user_id, user_id)
         return middleware.payment_manager.get_payment_session(
             user_id=resolved_user,
             payment_session_id=resolved_id,