Agents Pay

scripts/x402_fetch.py

"""Trusted x402 fetch: settle a 402 and return metadata without leaking secrets.

Register `x402_fetch` as a tool in any Python framework. Everything
security-relevant happens in this process, never in the model's context:

  * The policy gate (x402_policy) decides whether to pay, before any signing.
  * The signed payment proof is attached to the outbound request and then
    discarded. It is never returned, logged, or shown to the model.
  * Provider credentials are never parameters — this module never touches them.
    Signing happens inside AgentCore Payments; only resource IDs live here.
  * Paid content is not returned into the model context. The tool returns only
    bounded metadata and a SHA-256 hash, so fetched instructions cannot become
    payment-capable instructions.

Configuration
-------------
Resource identifiers and the payment policy come from one operator-owned file,
~/.agents-pay/config.json. The runtime binds this path to the operating-system
account and ignores HOME, AGENTS_PAY_CONFIG, and X402_POLICY_FILE. Administrative
commands may use --path or AGENTS_PAY_CONFIG explicitly. The file holds no provider
credentials, only ARNs and IDs plus the limits. See x402_policy for its shape.

The config file takes precedence over the environment for every identifier. That
ordering is deliberate: the session ID names the budget being drawn down, so if a
variable could override the 0600 file, anything able to set that variable could
redirect spending to a larger session. The environment remains a fallback for
container and Lambda deployments with no writable home — a weaker mode, because
whatever sets the environment there chooses the session.

Region/resource-resolution fixes (see agents_pay_admin.py for the admin-side
half of this batch: resolve_region() and resolve_manager_arn() there fix the
same pattern for the admin CLI's config.json lookups)
-------------------------------------------------------------------------------
Three call sites in this file used to build PaymentManager with
`region_name=pol.resolve_resource(policy, "region") or "us-west-2"`. That
hardcoded fallback only fires when nothing configures a region anywhere — a
normal state for a deployment that relies on its AWS profile/IMDS region rather
than setting one explicitly — and PaymentManager already falls back to
boto3.Session().region_name internally before its own "us-west-2" default. So
forcing "us-west-2" here could send a real payment against the wrong AWS region
and fail with a confusing manager-not-found error. Fixed by passing
pol.resolve_resource(policy, "region") (or None) and letting boto3/PaymentManager
resolve it themselves. See payment_session_status(), prepare_browser_payment(),
and x402_fetch() below.

A related fix in x402_policy.derive_client_token() closes a second instance of
the same pattern: it read PAYMENT_SESSION_ID from the environment directly
whenever a caller omitted session_id, bypassing resolve_resource()'s documented
config-file-first precedence for a spending credential. See its docstring.

Tunables (behaviour only, never identifiers):

  X402_MAX_BODY_BYTES    response cap (default 262144, clamped 1 KiB - 64 MiB)
  X402_TIMEOUT_SECONDS   per-request timeout (default 20, clamped 1 - 120)

This module deliberately does NOT provide session creation or infrastructure
setup. Those are separate trusted paths with separate credentials, so a
compromised agent cannot mint budget or create payment resources.
"""

from __future__ import annotations

import hashlib
import json
import os
import secrets
import socket
import ssl
import time
from typing import Any
from urllib.parse import urlparse

import httpx

import x402_policy as pol

AGENT_NAME = "openclaw-aws-agents-pay"


def _bounded_env(name: str, default: float, minimum: float, maximum: float) -> float:
    """Read a numeric tunable, ignoring values that are invalid or out of range.

    Parsed defensively at import: a malformed value raising here would escape
    every handler and break the contract that x402_fetch never raises into the
    agent loop. A zero or negative limit would also silently disable the control,
    so values are clamped rather than trusted.
    """
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return default
    if value < minimum or value > maximum:
        return default
    return value


# Cap between 1 KiB and 64 MiB; a 0/negative/garbage value falls back to default.
MAX_BODY_BYTES = int(_bounded_env("X402_MAX_BODY_BYTES", 256 * 1024, 1024, 64 * 1024 * 1024))
TIMEOUT_SECONDS = _bounded_env("X402_TIMEOUT_SECONDS", 20.0, 1.0, 120.0)
# Transient on-chain settlement can leave a paid retry at 402; replay the SAME
# authorization up to this many times. Bounded so a bad value cannot loop forever.
MAX_PAYMENT_ATTEMPTS = int(_bounded_env("X402_MAX_PAYMENT_ATTEMPTS", 5, 1, 10))

class PaymentBlocked(Exception):
    """Raised when trusted code refuses to pay or to fetch."""


def _require_resource(policy: dict, key: str) -> str:
    """Resolve a required resource identifier, config file first then environment.

    Config wins over the environment on purpose: the session ID names the budget
    being spent, so if a variable could override the 0600 file, anything able to
    set that variable could redirect spending to a larger session.
    """
    value = pol.resolve_resource(policy, key)
    if not value:
        env = pol.RESOURCE_ENV.get(key, key.upper())
        raise PaymentBlocked(
            f"Missing payment resource '{key}'. Add it to the config "
            f"(agents_pay_admin.py init-config) or set {env}. Provision resources "
            "with the trusted setup path first."
        )
    return value


def _ssl_context() -> ssl.SSLContext:
    """Strict TLS context: hostname checking and chain verification always on.

    Prefers certifi's CA bundle when present, because some hosts (including
    mise-managed Pythons) have no system CA file and would otherwise fail
    verification. Verification is never disabled — if no bundle is available the
    request fails closed rather than proceeding unverified.
    """
    try:
        import certifi

        ctx = ssl.create_default_context(cafile=certifi.where())
    except ImportError:
        ctx = ssl.create_default_context()
    ctx.check_hostname = True
    ctx.verify_mode = ssl.CERT_REQUIRED
    return ctx


class _PinnedResolverTransport(httpx.HTTPTransport):
    """HTTP transport that only ever connects to a pre-vetted IP address.

    Closes the DNS-rebinding window. Validating a hostname and then letting the
    HTTP client resolve it again independently leaves a TOCTOU gap: the second
    lookup can return an internal address. Here resolution happens once, every
    answer is vetted, and the socket is opened directly to a vetted address.

    Implemented by overriding the connection pool's socket creation rather than
    by patching `socket.getaddrinfo`. Patching that global is not thread-safe —
    two concurrent fetches to different hosts can restore or observe each
    other's state, and a request can end up resolving *unpinned*, which silently
    reopens the very window this class exists to close.

    The URL keeps its real hostname, so SNI and certificate verification are
    unaffected; only the dialed address is fixed. (Rewriting the URL to a bare
    IP would break certificate validation.)
    """

    def __init__(self, pinned_ip: str, **kwargs):
        super().__init__(**kwargs)
        self._pinned_ip = pinned_ip
        pool = self._pool  # httpcore connection pool
        original_factory = pool._network_backend

        class _PinnedBackend:
            """Delegates to the real backend, forcing the destination address."""

            def __init__(self, backend, pinned: str):
                self._backend = backend
                self._pinned = pinned

            def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
                # Re-vet at connect time; the pin was vetted at resolve time too.
                pol.assert_public_ip(self._pinned)
                return self._backend.connect_tcp(
                    self._pinned, port, timeout=timeout,
                    local_address=local_address, socket_options=socket_options,
                )

            def __getattr__(self, name):
                return getattr(self._backend, name)

        pool._network_backend = _PinnedBackend(original_factory, pinned_ip)


def _resolve_and_vet(hostname: str, port: int) -> str:
    """Resolve once, reject if ANY answer is internal, return the address to pin."""
    try:
        infos = socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP)
    except socket.gaierror:
        raise PaymentBlocked("Payment URL hostname does not resolve.") from None
    if not infos:
        raise PaymentBlocked("Payment URL hostname does not resolve.")
    for info in infos:
        pol.assert_public_ip(info[4][0])
    return infos[0][4][0]


#: Only side-effect-free verbs. A body-bearing request (POST/PUT/PATCH) would let the
#: agent push agent-chosen data to an arbitrary origin, which widens the very
#: exfiltration surface, and the policy gate validates the URL, not a request body.
#: Paid *retrieval* is what this skill is for.
_ALLOWED_METHODS = ("GET", "HEAD")


def _get(url: str, headers: dict[str, str] | None = None, method: str = "GET") -> httpx.Response:
    """One hardened HTTPS request: no redirects, pinned address, bounded body."""
    method = method.upper()
    if method not in _ALLOWED_METHODS:
        raise PaymentBlocked(
            f"Method {method} is not allowed; this skill fetches paid content with "
            f"{' or '.join(_ALLOWED_METHODS)}. A request body would let the agent send "
            "data to an arbitrary origin, which the policy gate does not validate."
        )
    parsed = urlparse(url)
    host, port = parsed.hostname or "", parsed.port or 443
    pinned_ip = _resolve_and_vet(host, port)

    transport = _PinnedResolverTransport(pinned_ip, verify=_ssl_context(), retries=0)

    request_headers = {
        "Accept": "application/json, text/plain",
        # Refuse compressed responses. httpx advertises gzip/deflate by default
        # and iter_bytes() yields DECOMPRESSED bytes, so a small compressed body
        # can expand by ~1000x and blow past the size cap before it is checked
        # (a decompression bomb against a payment-capable process).
        "Accept-Encoding": "identity",
    }
    if headers:
        request_headers.update(headers)

    with httpx.Client(
        transport=transport,
        timeout=httpx.Timeout(TIMEOUT_SECONDS, connect=min(5.0, TIMEOUT_SECONDS)),
        follow_redirects=False,  # a 30x is surfaced, never auto-followed
        max_redirects=0,
    ) as client:
        with client.stream(method, url, headers=request_headers) as response:
            # Reject an oversized body before reading any of it, when declared.
            declared = response.headers.get("content-length")
            if declared and declared.isdigit() and int(declared) > MAX_BODY_BYTES:
                raise PaymentBlocked(
                    f"Response declares {declared} bytes, over the {MAX_BODY_BYTES} limit."
                )

            body = bytearray()
            # Check BEFORE extending, so the buffer can never exceed the cap even
            # if a single chunk arrives larger than expected (e.g. a decompressing
            # transport). Small chunks also keep peak memory close to the cap.
            for chunk in response.iter_bytes(chunk_size=8192):
                if len(body) + len(chunk) > MAX_BODY_BYTES:
                    raise PaymentBlocked(
                        f"Response exceeded {MAX_BODY_BYTES} bytes; refusing to buffer more."
                    )
                body.extend(chunk)
            response.read_body = bytes(body)  # type: ignore[attr-defined]
            return response


def _body_text(response: httpx.Response) -> str:
    raw: bytes = getattr(response, "read_body", b"")
    return raw.decode(response.encoding or "utf-8", errors="replace")


def _safe_content(response: httpx.Response, policy: dict | None = None) -> dict[str, Any]:
    """Return metadata about paid content; optionally return the body.

    By default, paid publisher content is withheld from model context as a
    security control (it may contain prompt injection). Operators can opt in
    to body return by setting `return_body: true` in the policy section of
    the config file (~/.agents-pay/config.json). The config file is bound to
    the OS account (0600) and cannot be set by the model at runtime.
    """
    content_type = (response.headers.get("content-type") or "").split(";")[0].strip().lower()
    raw: bytes = getattr(response, "read_body", b"")
    result: dict[str, Any] = {
        "content_type": content_type or "unknown",
        "body_sha256": hashlib.sha256(raw).hexdigest(),
        "bytes": len(raw),
    }
    return_body = (policy or {}).get("return_body") is True
    if return_body:
        result["body_returned"] = True
        result["body"] = raw.decode("utf-8", errors="replace")[:10240]
        result["truncated"] = len(raw) > 10240
        result["untrusted"] = True
    else:
        result["body_returned"] = False
        result["note"] = (
            "Body withheld: paid publisher content is untrusted and must not enter "
            "the payment-capable model context. Use a separate no-payment/no-network "
            "analysis context if summarisation is required. To opt in, set "
            "'return_body: true' in the policy section of the config file."
        )
    return result


def payment_session_status() -> str:
    """Report whether the configured payment session can currently be spent.

    Read-only: it cannot create, extend, or fund anything. Exposing this to the
    model is safe and useful — it lets an agent say "the budget is exhausted, ask
    your operator" instead of discovering it as a failed payment mid-task.

    Returns JSON with a `usable` boolean. When false, `next_step` says what a
    HUMAN must do; the agent has no way to remedy it itself.
    """
    try:
        try:
            policy = pol.load_config()
        except pol.PolicyError:
            policy = {"_resources": {}}   # no config: fall back to the environment
        session_id = pol.resolve_resource(policy, "payment_session_id")
        manager_arn = pol.resolve_resource(policy, "payment_manager_arn")
        user_id = pol.resolve_resource(policy, "user_id")
        if not session_id or not manager_arn:
            return json.dumps(
                {
                    "usable": False,
                    "reason": "No payment session is configured.",
                    "next_step": (
                        "An operator must run: agents_pay_admin.py new-session "
                        "(requires a human at a terminal)."
                    ),
                }
            )

        try:
            from bedrock_agentcore.payments import PaymentManager
        except ImportError:
            return json.dumps(
                {
                    "usable": False,
                    "reason": "bedrock-agentcore payments support is not installed.",
                    "next_step": "Install bedrock-agentcore>=1.19.0.",
                }
            )

        manager = PaymentManager(
            payment_manager_arn=manager_arn,
            # Region resolution here must match agents_pay_admin.py's resolve_region():
            # config.json's resources.region, else AWS_REGION, else let boto3/PaymentManager
            # resolve it (profile, IMDS, etc.) themselves. A hardcoded "us-west-2" fallback
            # would silently override a correctly-resolved region whenever the operator's
            # deployment lives elsewhere and neither config nor env sets one explicitly —
            # PaymentManager itself already falls back to boto3.Session().region_name before
            # its own "us-west-2" default, so passing None here is safe and correct.
            region_name=pol.resolve_resource(policy, "region"),
            agent_name=AGENT_NAME,
        )
        session = manager.get_payment_session(payment_session_id=session_id, user_id=user_id)

        # Field names vary across SDK revisions; read defensively and report only
        # what we can positively confirm rather than guessing a usable=True.
        status = str(session.get("status") or session.get("sessionStatus") or "").upper()
        limits = session.get("limits") or {}
        spend_cap = (limits.get("maxSpendAmount") or {}).get("value")
        spent = (session.get("spentAmount") or {}).get("value")

        usable = status in ("ACTIVE", "READY")
        result: dict[str, Any] = {"usable": usable, "status": status or "unknown"}
        if spend_cap is not None:
            result["budget_usd"] = str(spend_cap)
        if spent is not None:
            result["spent_usd"] = str(spent)
        if not usable:
            result["next_step"] = (
                "An operator must run: agents_pay_admin.py new-session "
                "(requires a human at a terminal). The agent cannot mint budget."
            )
        return json.dumps(result)

    except Exception as e:  # noqa: BLE001 - never raise into the agent loop
        return json.dumps(
            {
                "usable": False,
                "reason": f"{type(e).__name__} while reading session status.",
                "next_step": "An operator must check PAYMENT_SESSION_ID and AWS credentials.",
            }
        )


# --- Browser path: opaque handles instead of raw proofs -----------------------
#
# Some paid resources must render in a real browser, so the payment proof has to
# be attached to a navigation the agent controls. Handing the proof to the model
# must not expose the proof to the model, so it never leaves this process. It is held
# here and referenced by a single-use handle bound to one origin and resource.
#
# The handle is useless to an attacker who reads the transcript — it is not a
# credential, cannot be replayed elsewhere, and expires.

_PROOF_VAULT: dict[str, dict[str, Any]] = {}
_HANDLE_TTL_SECONDS = 90.0


def _purge_expired(now: float) -> None:
    for handle in [h for h, e in _PROOF_VAULT.items() if e["expires_at"] <= now]:
        entry = _PROOF_VAULT.pop(handle, None)
        if entry:
            entry["header"].clear()


def prepare_browser_payment(url: str, purchase_id: str | None = None) -> str:
    """Pay for a URL and return an OPAQUE HANDLE for a browser to replay.

    Same trusted pipeline as `x402_fetch` — policy gate, SSRF checks, strict
    challenge parsing, derived idempotency — but instead of fetching the content
    it retains the signed proof in-process and returns a handle.

    The model receives the handle and a redacted receipt. It never sees the proof.
    Pass the handle to `attach_browser_payment()` at navigation time.
    """
    try:
        policy = pol.load_config()
        origin = pol.assert_public_https_url(url)
        # Origin pinning is optional, so an empty allowed_origins list permits public
        # HTTPS origins. The mandatory SSRF controls above and below still apply.
        allowed = [o.lower().rstrip("/") for o in (policy.get("allowed_origins") or [])]
        if allowed and origin.lower() not in allowed:
            raise PaymentBlocked(f"Origin {origin} is not in the configured allowed_origins.")

        response = _get(url)
        if response.status_code != 402:
            return json.dumps(
                {
                    "paid": False,
                    "refused": True,
                    "reason": f"URL returned {response.status_code}, not 402. No payment needed.",
                }
            )

        challenge = _extract_challenge(response)
        decision = pol.authorize_payment(url, challenge, policy, purchase_id)

        manager_arn = _require_resource(policy, "payment_manager_arn")
        instrument_id = _require_resource(policy, "payment_instrument_id")
        session_id = _require_resource(policy, "payment_session_id")
        user_id = _require_resource(policy, "user_id")

        try:
            from bedrock_agentcore.payments import PaymentManager
        except ImportError:
            raise PaymentBlocked(
                "bedrock-agentcore with payments support is not installed."
            ) from None

        manager = PaymentManager(
            payment_manager_arn=manager_arn,
            # See the resolve_region()-equivalent rationale in payment_session_status():
            # config.json/env first, else let boto3 resolve region itself rather than
            # forcing "us-west-2" and risking a manager-not-found against an ARN that
            # actually lives in the operator's real deployment region.
            region_name=pol.resolve_resource(policy, "region"),
            agent_name=AGENT_NAME,
        )

        # Only the vetted entry reaches the signer — resource is included only
        # after URL-binding validation in the policy gate (see _validated_resource).
        vetted = {"x402Version": decision["x402_version"], "accepts": [decision["accept"]]}
        if decision.get("resource"):
            vetted["resource"] = decision["resource"]
        vetted_challenge = json.dumps(vetted)
        payment_header = manager.generate_payment_header(
            payment_instrument_id=instrument_id,
            payment_session_id=session_id,
            user_id=user_id,
            client_token=decision["client_token"],
            payment_required_request={
                "statusCode": 402,
                "headers": {"content-type": "application/json"},
                "body": vetted_challenge,
            },
        )
        if not isinstance(payment_header, dict):
            raise PaymentBlocked("Payment header generation returned an unexpected shape.")

        now = time.monotonic()
        _purge_expired(now)
        handle = "x402h_" + secrets.token_urlsafe(24)
        _PROOF_VAULT[handle] = {
            "header": payment_header,     # stays here; never returned
            "origin": decision["origin"],
            "path": urlparse(url).path or "/",
            "expires_at": now + _HANDLE_TTL_SECONDS,
        }

        return json.dumps(
            {
                "paid": True,
                "handle": handle,
                "expires_in_seconds": int(_HANDLE_TTL_SECONDS),
                "receipt": {
                    "amount_usd": decision["amount_usd"],
                    "network": decision["accept"]["network"],
                    "resource": f"{decision['origin']}{urlparse(url).path}",
                },
                "next_step": (
                    "Call attach_browser_payment(handle, url) to get the header to set on "
                    "the browser, then navigate. The handle is single-use and expires."
                ),
            }
        )

    except (pol.PolicyError, PaymentBlocked) as e:
        return json.dumps({"paid": False, "refused": True, "reason": str(e)})
    except Exception as e:  # noqa: BLE001
        return json.dumps(
            {"paid": False, "refused": True, "reason": f"{type(e).__name__} during payment flow."}
        )


def attach_browser_payment(handle: str, url: str) -> dict[str, str]:
    """Redeem a handle for the header to set on a browser, for ONE navigation.

    Returns the actual `{header: value}` mapping, so this is the one point where
    the proof becomes visible to the caller. Give it to browser-driving code, not
    to the model: a framework should register `prepare_browser_payment` as the
    model-facing tool and call this from trusted glue at navigation time.

    Single-use and origin+path bound: a handle stolen from a transcript cannot be
    redeemed for a different resource, and cannot be redeemed twice.
    """
    now = time.monotonic()
    _purge_expired(now)

    entry = _PROOF_VAULT.get(handle)
    if entry is None:
        raise PaymentBlocked("Unknown, already-used, or expired payment handle.")

    # Bind redemption to the exact resource the payment was authorized for.
    if pol._canonical_origin(url).lower() != entry["origin"].lower():
        raise PaymentBlocked("Handle does not match this origin.")
    if (urlparse(url).path or "/") != entry["path"]:
        raise PaymentBlocked("Handle does not match this resource path.")

    _PROOF_VAULT.pop(handle, None)  # single use: consumed on redemption
    header = entry["header"]
    return dict(header)


def _extract_challenge(response: httpx.Response) -> dict:
    """Parse an x402 challenge from the `payment-required` header or the body.

    Strict: a challenge must be a JSON object carrying `accepts`, with no default
    version fallback or tolerance for merely plausible shapes.
    """
    import base64

    header = response.headers.get("payment-required") or response.headers.get("x-payment-required")
    if header:
        for decode in (lambda h: base64.b64decode(h, validate=True).decode("utf-8"), lambda h: h):
            try:
                parsed = json.loads(decode(header))
                if isinstance(parsed, dict) and parsed.get("accepts"):
                    return parsed
            except Exception:  # noqa: BLE001 - try the next decoding strategy
                continue

    try:
        parsed = json.loads(_body_text(response))
    except json.JSONDecodeError:
        raise PaymentBlocked("402 response contains no parseable x402 challenge.") from None

    if isinstance(parsed, dict):
        if parsed.get("accepts"):
            return parsed
        inner = parsed.get("challenge")
        if isinstance(inner, dict) and inner.get("accepts"):
            return inner
    raise PaymentBlocked("402 challenge is missing the required 'accepts' array.")


def x402_fetch(url: str, purchase_id: str | None = None, method: str = "GET") -> str:
    """Fetch an x402-protected URL, paying only if trusted policy allows it.

    Returns a JSON string. On success it contains status, response metadata,
    and a redacted payment receipt (amount, network, resource) — never the
    signed proof, credential, or transaction signature. The paid body is
    returned only when `return_body: true` is set in the operator's config
    file; otherwise only content type, byte count, and SHA-256 hash are
    included.
    """
    try:
        # Pre-flight: load policy and clear the origin BEFORE any network I/O, so
        # an unapproved host is never contacted at all. Checking only after the
        # probe would still leave the probe itself as an SSRF primitive.
        policy = pol.load_config()
        origin = pol.assert_public_https_url(url)
        # Origin pinning is optional, so an empty allowed_origins list permits public
        # HTTPS origins. The mandatory SSRF controls above and below still apply.
        allowed = [o.lower().rstrip("/") for o in (policy.get("allowed_origins") or [])]
        if allowed and origin.lower() not in allowed:
            raise PaymentBlocked(f"Origin {origin} is not in the configured allowed_origins.")

        response = _get(url, method=method)

        if response.status_code != 402:
            return json.dumps(
                {
                    "status_code": response.status_code,
                    "paid": False,
                    "untrusted": True,
                    **_safe_content(response, policy),
                }
            )

        # --- 402 path: every decision below is made in trusted code. ---
        challenge = _extract_challenge(response)
        # Re-runs the destination and origin checks, then validates the challenge
        # (scheme, network, asset, recipient, amount) against the same policy.
        decision = pol.authorize_payment(url, challenge, policy, purchase_id)  # raises to refuse

        manager_arn = _require_resource(policy, "payment_manager_arn")
        instrument_id = _require_resource(policy, "payment_instrument_id")
        session_id = _require_resource(policy, "payment_session_id")
        user_id = _require_resource(policy, "user_id")

        try:
            from bedrock_agentcore.payments import PaymentManager
        except ImportError:
            raise PaymentBlocked(
                "bedrock-agentcore with payments support is not installed "
                "(needs a version providing bedrock_agentcore.payments)."
            ) from None

        manager = PaymentManager(
            payment_manager_arn=manager_arn,
            # Same fix as the two call sites above and agents_pay_admin.py's
            # resolve_region(): never force "us-west-2" over a region that config.json,
            # the environment, or boto3's own session/profile resolution already has
            # right — doing so on this signing path risked a real payment attempt
            # failing with a confusing manager-not-found instead of succeeding.
            region_name=pol.resolve_resource(policy, "region"),
            agent_name=AGENT_NAME,
        )

        # CRITICAL: hand the signer ONLY the entry the policy approved.
        #
        # Forwarding the publisher's raw 402 (its headers and body) would mean
        # validating one document and signing another: the challenge can carry
        # several accepts entries, or a compliant header alongside a hostile
        # body, so the SDK could settle terms the gate never saw. The vetted
        # entry is reserialized into a single-entry challenge; the resource
        # object is included only after URL-binding validation in the policy
        # gate (see _validated_resource).
        vetted = {"x402Version": decision["x402_version"], "accepts": [decision["accept"]]}
        if decision.get("resource"):
            vetted["resource"] = decision["resource"]
        vetted_challenge = json.dumps(vetted)

        # Settle and replay, retrying a TRANSIENT post-payment 402.
        #
        # On Base Sepolia the proof is often valid while on-chain settlement lags, so
        # the paid retry still returns 402. Without a retry that surfaces as a failed
        # fetch for a payment the user already made. The SDK only builds the header —
        # it does not make the merchant call — so the retry has to live here.
        #
        # Safe because the SAME derived client_token is reused for every attempt:
        # ProcessPayment is idempotent on it, so each attempt replays one
        # authorization/nonce. A retry either settles the not-yet-settled payment or,
        # if it had already settled, reverts on-chain. It cannot charge twice.
        paid_response = None
        for attempt in range(1, MAX_PAYMENT_ATTEMPTS + 1):
            # The proof lives only in this local variable, for the length of one
            # request. It is never returned to the caller and never logged. The
            # `finally` guarantees it is dropped even if the paid request raises —
            # a bare `del` after the call would be skipped on the exception path.
            payment_header = None
            try:
                payment_header = manager.generate_payment_header(
                    payment_instrument_id=instrument_id,
                    payment_session_id=session_id,
                    user_id=user_id,
                    client_token=decision["client_token"],  # stable => retry-safe
                    payment_required_request={
                        "statusCode": 402,
                        "headers": {"content-type": "application/json"},
                        "body": vetted_challenge,
                    },
                )
                if not isinstance(payment_header, dict):
                    # Defensive: the SDK contract is a {header: value} mapping. Bail
                    # out rather than passing an unknown shape to the HTTP layer.
                    raise PaymentBlocked("Payment header generation returned an unexpected shape.")
                paid_response = _get(url, headers=payment_header, method=method)
            finally:
                if isinstance(payment_header, dict):
                    payment_header.clear()  # overwrite the mapping, then drop it
                del payment_header

            if paid_response.status_code != 402:
                break  # 2xx, or a non-transient error worth surfacing as-is

        if paid_response is not None and paid_response.status_code == 402:
            return json.dumps(
                {
                    "paid": False,
                    "refused": False,
                    "status_code": 402,
                    "attempts": MAX_PAYMENT_ATTEMPTS,
                    "reason": (
                        f"Paid and replayed {MAX_PAYMENT_ATTEMPTS} times but the merchant "
                        "still returns 402 — usually transient on-chain settlement. The same "
                        "authorization was replayed each time, so there is no double charge. "
                        "Retry shortly, or raise X402_MAX_PAYMENT_ATTEMPTS."
                    ),
                }
            )

        return json.dumps(
            {
                "status_code": paid_response.status_code,
                "paid": 200 <= paid_response.status_code < 300,
                "untrusted": True,
                "receipt": {  # redacted by construction
                    "amount_usd": decision["amount_usd"],
                    "network": decision["accept"]["network"],
                    "resource": f"{decision['origin']}{urlparse(url).path}",
                },
                **_safe_content(paid_response, policy),
            }
        )

    except (pol.PolicyError, PaymentBlocked) as e:
        # Refusals are safe to surface: they carry no challenge values or secrets.
        return json.dumps({"paid": False, "refused": True, "reason": str(e)})
    except Exception as e:  # noqa: BLE001
        # Never let a raw exception escape — SDK errors can embed request detail.
        return json.dumps(
            {"paid": False, "refused": True, "reason": f"{type(e).__name__} during payment flow."}
        )