Agents Pay

scripts/x402_policy.py

"""Deterministic x402 payment policy gate — the trusted decision point.

Every payment decision is made HERE, in code, from a policy file the operator
controls. Nothing in this module reads model output, chat history, or publisher
content as authorization. That is the whole point: an agent (or a prompt
injection inside paid content) cannot talk its way past these checks.

Why this file exists
--------------------
Instructions to a model are not an access control. This module makes the
payment rules executable, so the guarantee holds even when the model is wrong
or hostile.

Config file
-----------
The runtime resolves ~/.agents-pay/config.json from the operating-system account,
not HOME or a config-path environment variable. Administrative callers and tests
may pass a path explicitly. The file must be regular, owned by the current user,
mode 0600, in a directory that is not group/world-writable, with no symlinks (see
load_config). One file, two sections:

    {
      "resources": {                   # what to pay WITH
        "payment_manager_arn": "arn:aws:bedrock-agentcore:...:payment-manager/pm-1",
        "payment_instrument_id": "pi-...",
        "payment_session_id": "ps-...",   # a spending credential
        "user_id": "alice",
        "region": "us-west-2"
      },
      "policy": {                      # what may be paid
        "max_per_payment_usd": "0.05",   # PER-PAYMENT ceiling, required
        "allowed_networks": ["eip155:84532"],
        "allowed_assets": {             # network -> exact asset contract(s)
          "eip155:84532": ["0x036CbD53842c5426634e7929541eC2318f3dCF7e"]
        },
        "allowed_recipients": ["0x1111111111111111111111111111111111111111"],
        # Or, instead of allowed_recipients:
        # "allow_any_recipient": true,
        "allowed_origins": ["https://sandbox.node4all.com"],
        "allowed_schemes": ["exact"]
      }
    }

Absent keys deny rather than allow. There is no implicit wildcard.

Recipient validation
--------------------
By default, the payee (`payTo`) must appear in `allowed_recipients`, an
operator-approved allowlist in the config file. Unknown recipients are refused
before signing. An operator may instead set `allow_any_recipient` to the literal
boolean `true`. The two modes are mutually exclusive, and the runtime rejects a
policy that enables both.

Allowing any recipient means a publisher controls the beneficiary. Network,
asset, scheme, origin/resource, per-payment, and cumulative session limits still
apply, but recipient allowlisting no longer protects against a malicious payee.

Two ceilings, not one
---------------------
`max_per_payment_usd` is NOT a duplicate of the session budget:

  * the session budget is CUMULATIVE — total spend before a human must re-approve;
  * `max_per_payment_usd` is PER TRANSACTION.

With only the session budget, one hostile challenge for the full remaining balance
drains it in a single payment. A trusted positive maximum for each payment keeps
both bounds meaningful.

Scope / non-goals
-----------------
This module decides IF a payment may proceed and derives the idempotency key.
It never holds provider credentials, never signs, never performs network I/O.
Fetching and settlement live in x402_fetch.py.
"""

from __future__ import annotations

import hashlib
import ipaddress
import json
import os
import socket
import stat
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import urlparse

# USDC has 6 decimals on every chain AgentCore Payments supports. x402 quotes
# amounts as integer base units, so 500000 base units == 0.50 USDC.
USDC_DECIMALS = 6

# Fields we require on an accepts[] entry before it is eligible for payment.
_REQUIRED_ACCEPT_FIELDS = ("scheme", "network", "asset", "payTo")


class PolicyError(Exception):
    """A payment was refused, or the policy itself is unusable.

    Carries no challenge values, so the message is safe to surface to a model.
    """


def _fail(reason: str) -> "PolicyError":
    return PolicyError(reason)


def runtime_config_path() -> Path:
    """Return the config path bound to the current OS account.

    Path.home(), HOME, and config-path environment variables are caller-controlled
    in shell-capable harnesses. The runtime therefore asks the OS account database
    for the home directory and fails closed where that trusted lookup is unavailable.
    """
    try:
        import pwd
    except ImportError:
        raise _fail(
            "Cannot resolve the runtime payment config from the OS account on this "
            "platform. Payments are refused."
        ) from None

    try:
        home = Path(pwd.getpwuid(os.getuid()).pw_dir)
    except KeyError:
        raise _fail(
            "The current uid has no OS account home directory. Payments are refused."
        ) from None
    if not home.is_absolute():
        raise _fail("The OS account home directory is not absolute. Payments are refused.")
    return home / ".agents-pay" / "config.json"


def load_config(path: str | os.PathLike[str] | None = None) -> dict:
    """Load and validate the operator config: resource identifiers + payment policy.

    One file, two sections. `resources` holds the identifiers the runtime needs
    (manager ARN, instrument, session, user); `policy` holds the limits. They live
    together because they are written at the same moment by the same human, and
    splitting them made the operator hand-copy identifiers between steps.

    Merging them also buys a real control, not just convenience: the session ID is
    a spending credential, and the sanctioned runtime cannot select a replacement
    policy file through HOME or a config-path environment variable. Resource values
    in the file also win over environment fallbacks — see resolve_resource().

    The file is a security control, so one that anyone else can write is treated as
    no config at all. lstat (not stat) rejects symlinks rather than following them.
    These checks do not protect against arbitrary code already running as the file
    owner; that requires OS, container, or IAM isolation around the signer.
    """
    p = Path(path) if path is not None else runtime_config_path()

    try:
        st = p.lstat()
    except FileNotFoundError:
        raise _fail(
            f"No payment config at {p}. Payments are refused until an operator "
            "creates one (agents_pay_admin.py init-config). This is intentional: "
            "there is no permissive default."
        ) from None

    if stat.S_ISLNK(st.st_mode):
        raise _fail(f"Payment config {p} is a symlink; refusing to follow it.")
    if not stat.S_ISREG(st.st_mode):
        raise _fail(f"Payment config {p} is not a regular file.")
    if st.st_uid != os.getuid():
        raise _fail(f"Payment config {p} is not owned by the current user.")
    # Reject any group/other bit: a config others can rewrite is not a control.
    if st.st_mode & 0o077:
        raise _fail(
            f"Payment config {p} has mode {stat.filemode(st.st_mode)}; "
            "expected 0600. Run: chmod 600 " + str(p)
        )

    # The containing directory matters as much as the file. Write access to the
    # directory lets another principal rename a wider config into place, which
    # no amount of checking on the old inode would detect.
    try:
        dir_st = p.parent.lstat()
    except OSError:
        raise _fail(f"Cannot stat the directory containing {p}.") from None
    if dir_st.st_uid != os.getuid():
        raise _fail(f"Directory {p.parent} is not owned by the current user.")
    if dir_st.st_mode & 0o022:
        raise _fail(
            f"Directory {p.parent} is group/world-writable "
            f"({stat.filemode(dir_st.st_mode)}); another user could replace the "
            "config file. Run: chmod 700 " + str(p.parent)
        )

    try:
        raw = json.loads(p.read_text())
    except json.JSONDecodeError as e:
        raise _fail(f"Payment config {p} is not valid JSON: {e}") from None
    if not isinstance(raw, dict):
        raise _fail(f"Payment config {p} must be a JSON object.")

    # A flat file (policy keys at the top level) is still accepted, so an existing
    # policy.json keeps working rather than failing open or failing loudly.
    policy = raw.get("policy") if isinstance(raw.get("policy"), dict) else raw
    resources = raw.get("resources") if isinstance(raw.get("resources"), dict) else {}

    if "max_per_payment_usd" in policy:
        cap_key = "max_per_payment_usd"
    elif "max_amount_usd" in policy:      # earlier name, still honoured
        cap_key = "max_amount_usd"
    else:
        raise _fail(
            "Payment policy is missing 'max_per_payment_usd'. This is the PER-PAYMENT "
            "ceiling and is not the same thing as the session budget: the session "
            "budget is cumulative, so without a per-payment cap a single hostile "
            "challenge can drain the whole budget in one transaction."
        )
    try:
        cap = Decimal(str(policy[cap_key]))
    except (InvalidOperation, ValueError):
        raise _fail(f"Payment policy '{cap_key}' is not a decimal number.") from None
    policy["max_per_payment_usd"] = policy[cap_key]  # normalize for callers
    policy["_resources"] = resources                 # carried for resolve_resource()
    if cap <= 0:
        raise _fail(f"Payment policy '{cap_key}' must be greater than zero.")

    return policy


#: config `resources` key -> the environment variable that may substitute for it.
RESOURCE_ENV = {
    "payment_manager_arn": "PAYMENT_MANAGER_ARN",
    "payment_instrument_id": "PAYMENT_INSTRUMENT_ID",
    "payment_session_id": "PAYMENT_SESSION_ID",
    "user_id": "PAYMENT_USER_ID",
    "region": "AWS_REGION",
}


def resolve_resource(policy: dict, key: str) -> str | None:
    """Resolve one resource identifier: **config file first**, environment second.

    The precedence is the security-relevant part, and it is deliberately the
    opposite of the usual "env overrides file" convention.

    The session ID is a spending credential: it names the budget being drawn down.
    If the environment could override the file, an agent able to set a variable
    could point the runtime at some other session with a larger budget, and the
    0600 file would be decorative. So the file wins wherever it speaks.

    The environment remains the fallback for deployments with no writable home —
    containers, Lambda — where identifiers arrive by injection. That is a real
    need, but it is the weaker mode: anything that can set the environment can
    choose the session.
    """
    resources = policy.get("_resources") or {}
    from_file = resources.get(key)
    if from_file:
        return str(from_file)
    env_name = RESOURCE_ENV.get(key)
    return os.environ.get(env_name) if env_name else None

def per_payment_cap(policy: dict) -> str:
    """The per-payment USD ceiling, under either the current or the earlier key.

    load_config() normalizes this, but callers may hand a policy dict straight in
    (tests, embedded use), so resolve it here too rather than assuming. Missing is
    a refusal, never an unbounded payment.
    """
    for key in ("max_per_payment_usd", "max_amount_usd"):
        if key in policy:
            return str(policy[key])
    raise _fail(
        "Payment policy has no per-payment ceiling ('max_per_payment_usd'). "
        "Refusing to pay: the session budget is cumulative and does not bound a "
        "single transaction."
    )


def _policy_list(policy: dict, key: str) -> list[str]:
    """Read a list-valued policy key. Missing or malformed means deny (empty)."""
    value = policy.get(key)
    if value is None:
        return []
    if not isinstance(value, list):
        raise _fail(f"Payment policy '{key}' must be a list.")
    return [str(v) for v in value]


def _canonical_origin(url: str) -> str:
    """scheme://host[:port], lowercased, with the default HTTPS port dropped."""
    u = urlparse(url)
    host = (u.hostname or "").lower()
    if u.port and u.port != 443:
        return f"{u.scheme}://{host}:{u.port}"
    return f"{u.scheme}://{host}"


def assert_public_https_url(url: str) -> str:
    """Require HTTPS and a publicly routable destination; return the origin.

    Resolves every address the hostname maps to and rejects the request if ANY
    of them is internal. Checking only the first answer would let a host with
    one public and one private address through.

    This is a pre-flight check. It does not by itself defeat DNS rebinding,
    because the OS resolves again when the socket is opened — x402_fetch.py
    closes that gap by pinning the connection to a vetted address.
    """
    u = urlparse(url)
    if u.scheme != "https":
        raise _fail("Only https:// payment URLs are allowed.")
    if not u.hostname:
        raise _fail("Payment URL has no host.")
    if u.username or u.password:
        raise _fail("Payment URL must not contain embedded credentials.")

    try:
        infos = socket.getaddrinfo(u.hostname, u.port or 443, proto=socket.IPPROTO_TCP)
    except socket.gaierror:
        raise _fail("Payment URL hostname does not resolve.") from None

    for info in infos:
        assert_public_ip(info[4][0])

    return _canonical_origin(url)


def assert_public_ip(addr: str) -> None:
    """Reject loopback, private, link-local, and other non-public ranges.

    Deliberately broader than is_private/is_loopback/is_link_local, which alone
    still admit multicast and the CGNAT 100.64/10 shared-address space.
    """
    try:
        ip = ipaddress.ip_address(addr)
    except ValueError:
        raise _fail("Payment URL resolved to an unparseable address.") from None

    # Unwrap ::ffff:127.0.0.1 style addresses so v4 rules apply to them.
    mapped = getattr(ip, "ipv4_mapped", None)
    if mapped is not None:
        ip = mapped

    if (
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
        or ip.is_unspecified
    ):
        raise _fail("Payment URL resolves to a non-public address.")

    # 100.64.0.0/10 (CGNAT) is not flagged by any is_* property but is not
    # publicly routable, and 169.254.169.254 lives behind it in some networks.
    if ip.version == 4 and ip in ipaddress.ip_network("100.64.0.0/10"):
        raise _fail("Payment URL resolves to a non-public address.")


def base_units_to_usd(amount: str | int, decimals: int = USDC_DECIMALS) -> Decimal:
    """Convert an x402 integer base-unit amount to a USD Decimal.

    Accepts ONLY a canonical non-negative integer literal: digits, nothing else.
    Decimal() would otherwise happily parse "1E+7", "Infinity", "500000.0",
    " 500000 ", and "+500000". None of those breach the ceiling on their own
    (the comparison still holds), but a publisher should not get to express an
    amount in a form that a human reviewing logs would misread. An exotic
    encoding therefore fails closed rather than relying on downstream
    arithmetic to save us.
    """
    text = amount if isinstance(amount, str) else str(amount)
    if not isinstance(amount, int) and not text.isdigit():
        raise _fail("Payment challenge amount must be a plain integer in base units.")
    try:
        raw = Decimal(text)
    except InvalidOperation:
        raise _fail("Payment challenge amount is not a number.") from None
    if raw <= 0:
        raise _fail("Payment challenge amount must be positive.")
    return raw / (Decimal(10) ** decimals)


def validated_amount_units(entry: dict) -> str:
    """Resolve one canonical amount and reject conflicting aliases.

    x402 v1 commonly uses maxAmountRequired while v2 uses amount. Supporting both
    is safe only when they cannot describe different transactions.
    """
    amount = entry.get("amount")
    maximum = entry.get("maxAmountRequired")
    if amount is not None and maximum is not None and str(amount) != str(maximum):
        raise _fail("Payment challenge contains conflicting amount fields.")
    resolved = amount if amount is not None else maximum
    if resolved is None:
        raise _fail("Payment challenge has no amount.")
    text = str(resolved)
    base_units_to_usd(text)
    return text


def select_accept_entry(challenge: dict, policy: dict) -> dict:
    """Return the first accepts[] entry that satisfies policy, else raise.

    Each entry must earn selection by passing every configured check. The publisher
    cannot choose the network, asset, recipient, or amount unilaterally.
    """
    accepts = challenge.get("accepts")
    if not isinstance(accepts, list) or not accepts:
        raise _fail("Payment challenge has no accepts entries.")

    allowed_networks = _policy_list(policy, "allowed_networks")
    allowed_schemes = _policy_list(policy, "allowed_schemes")
    if "allowed_schemes" not in policy:
        allowed_schemes = ["exact"]
    allow_any_recipient = policy.get("allow_any_recipient", False)
    if not isinstance(allow_any_recipient, bool):
        raise _fail("Payment policy 'allow_any_recipient' must be a boolean.")
    if (
        "allow_any_recipient" in policy
        and "allowed_recipients" in policy
    ):
        raise _fail(
            "Payment policy 'allow_any_recipient' and 'allowed_recipients' "
            "are mutually exclusive."
        )
    allowed_recipients = [r.lower() for r in _policy_list(policy, "allowed_recipients")]
    allowed_assets = policy.get("allowed_assets") or {}
    if not isinstance(allowed_assets, dict):
        raise _fail("Payment policy 'allowed_assets' must be an object.")
    cap_usd = Decimal(str(per_payment_cap(policy)))

    if not allowed_networks:
        raise _fail("Payment policy allows no networks.")
    if not allowed_schemes:
        raise _fail("Payment policy allows no schemes.")
    if not allow_any_recipient and not allowed_recipients:
        raise _fail("Payment policy allows no recipients.")

    for entry in accepts:
        if not isinstance(entry, dict):
            continue
        if any(entry.get(f) in (None, "") for f in _REQUIRED_ACCEPT_FIELDS):
            continue
        if str(entry["scheme"]) not in allowed_schemes:
            continue
        network = str(entry["network"])
        if network not in allowed_networks:
            continue
        # Asset contracts are compared case-insensitively: EVM addresses are
        # hex and often differ only by EIP-55 checksum capitalization.
        permitted_assets = [a.lower() for a in allowed_assets.get(network, [])]
        if str(entry["asset"]).lower() not in permitted_assets:
            continue
        if (
            not allow_any_recipient
            and str(entry["payTo"]).lower() not in allowed_recipients
        ):
            continue
        try:
            amount_units = validated_amount_units(entry)
            if base_units_to_usd(amount_units) > cap_usd:
                continue
        except PolicyError:
            continue
        return entry

    # Uniform refusal: naming the failed field would let a publisher probe the
    # policy by iterating challenges until the message changed.
    raise _fail(
        "No payment option in this challenge satisfies the configured policy "
        "(scheme, network, asset, recipient, and amount ceiling are all enforced). "
        "Refusing to pay."
    )


def _validated_resource(challenge: dict, url: str) -> dict | None:
    """Validate and sanitize the challenge's resource object before signing.

    The x402 v2 spec requires the signed header to echo the resource for URL
    binding. However, the publisher controls the challenge, so we must verify
    that resource.url matches the URL we actually requested. This prevents a
    hostile publisher from binding the signature to a different resource.

    Comparison uses origin+path only: query parameters appended by the
    publisher (e.g. tracking params) do not change which server gets paid.

    Returns a sanitized dict with only the `url` field, or None if the
    challenge has no resource. Raises PolicyError on mismatch.
    """
    resource = challenge.get("resource")
    if resource is None:
        return None
    if not isinstance(resource, dict):
        return None
    resource_url = resource.get("url")
    if not isinstance(resource_url, str):
        return None
    # Bind: resource.url origin+path must match the requested origin+path.
    requested = urlparse(url)
    challenged = urlparse(resource_url)
    req_base = f"{requested.scheme}://{requested.netloc}{requested.path}".rstrip("/")
    ch_base = f"{challenged.scheme}://{challenged.netloc}{challenged.path}".rstrip("/")
    if req_base != ch_base:
        raise _fail("Challenge resource.url does not match the requested URL.")
    # Forward only the url field — no arbitrary publisher-controlled keys.
    return {"url": resource_url}


def authorize_payment(
    url: str,
    challenge: dict,
    policy: dict | None = None,
    purchase_id: str | None = None,
) -> dict:
    """Full gate: validate destination and challenge, return an approved decision.

    Returns the vetted accepts entry plus the derived amount, origin, and a
    stable idempotency key. Raises PolicyError on any refusal.

    `purchase_id` distinguishes deliberate repeat purchases of the same resource;
    see derive_client_token.
    """
    policy = policy if policy is not None else load_config()

    origin = assert_public_https_url(url)

    # Origin allowlisting is optional. HTTPS, address vetting, redirect refusal,
    # rebinding protection, timeouts, and byte limits are always enforced. An
    # operator with a known merchant set can pin it with allowed_origins.
    allowed_origins = _policy_list(policy, "allowed_origins")
    if allowed_origins and origin.lower() not in [o.lower().rstrip("/") for o in allowed_origins]:
        raise _fail(f"Origin {origin} is not in the configured allowed_origins.")

    if not isinstance(challenge, dict):
        raise _fail("Payment challenge is not a JSON object.")

    entry = select_accept_entry(challenge, policy)
    amount_units = validated_amount_units(entry)
    x402_version = int(challenge.get("x402Version") or challenge.get("version") or 1)

    # Sign exactly the amount the gate validated. v1 uses maxAmountRequired and
    # v2 uses amount; the non-canonical alias is removed so downstream code cannot
    # apply different precedence from the policy gate.
    vetted_entry = dict(entry)
    vetted_entry.pop("amount", None)
    vetted_entry.pop("maxAmountRequired", None)
    if x402_version == 1:
        vetted_entry["maxAmountRequired"] = amount_units
    else:
        vetted_entry["amount"] = amount_units

    return {
        "accept": vetted_entry,
        "origin": origin,
        "resource": _validated_resource(challenge, url),
        "amount_base_units": str(amount_units),
        "amount_usd": str(base_units_to_usd(amount_units)),
        "x402_version": x402_version,
        "client_token": derive_client_token(
            url,
            vetted_entry,
            challenge,
            purchase_id,
            session_id=resolve_resource(policy, "payment_session_id") or "",
        ),
    }


def derive_client_token(
    url: str,
    accept: dict,
    challenge: dict,
    purchase_id: str | None = None,
    *,
    session_id: str | None = None,
    policy: dict | None = None,
) -> str:
    """Derive a stable idempotency token for one logical purchase.

    Part of the same fix batch as the region-resolution changes in x402_fetch.py
    (see that module's docstring): the `policy=` parameter below closes a second
    instance of the "config.json has the value, code reads the environment
    instead" pattern found in this batch, this time for the session ID used as
    idempotency-token material rather than for region.

    Same (session, resource, network, asset, recipient, amount) always yields the
    same token, so a retry after a lost response replays the SAME authorization
    instead of creating a second irreversible payment. Derived rather than random
    precisely so it survives a process restart.

    The publisher's nonce is deliberately NOT part of the material. A retry
    re-fetches the 402 and many servers issue a fresh nonce each time, so mixing
    it in would produce a different token per attempt — turning the retry this
    function exists to protect into a second real payment, and handing a hostile
    publisher a way to force double charges by rotating nonces.

    The trade-off: two intentional purchases of the same resource, for the same
    amount, in the same session collapse to one token, so the second would be
    suppressed as a replay. Pass an explicit `purchase_id` (an order number, a
    turn counter — anything the caller controls) to distinguish deliberate
    repeat buys. Suppressing a duplicate charge is the safer default when the
    caller has not said otherwise.

    session_id resolution: an explicit `session_id` always wins. If neither
    `session_id` nor `policy` is given, this falls back to a raw environment
    read for backward compatibility with existing callers (tests, embedded use)
    that predate the `policy` parameter. A caller that DOES pass `policy` gets
    resolve_resource()'s documented config-file-first precedence instead — the
    correct behavior for any new caller resolving the session itself rather than
    passing an explicit session_id (the production call site in
    authorize_payment() already always passes session_id explicitly and is
    unaffected either way).
    """
    if session_id is not None:
        session = session_id
    elif policy is not None:
        session = resolve_resource(policy, "payment_session_id") or ""
    else:
        session = os.environ.get("PAYMENT_SESSION_ID", "")
    material = "\x1f".join(  # unit separator: cannot appear in these values
        [
            session,
            _canonical_origin(url),
            urlparse(url).path or "/",
            str(accept.get("network", "")),
            str(accept.get("asset", "")).lower(),
            str(accept.get("payTo", "")).lower(),
            validated_amount_units(accept),
            purchase_id or "",
        ]
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()