1"""Trusted x402 fetch: settle a 402 and return metadata without leaking secrets.
2
3Register `x402_fetch` as a tool in any Python framework. Everything
4security-relevant happens in this process, never in the model's context:
5
6  * The policy gate (x402_policy) decides whether to pay, before any signing.
7  * The signed payment proof is attached to the outbound request and then
8    discarded. It is never returned, logged, or shown to the model.
9  * Provider credentials are never parameters — this module never touches them.
10    Signing happens inside AgentCore Payments; only resource IDs live here.
11  * Paid content is not returned into the model context. The tool returns only
12    bounded metadata and a SHA-256 hash, so fetched instructions cannot become
13    payment-capable instructions.
14

x402_fetch.py on GitHub

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:

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."}

)