"""x402 v1 "exact" scheme rules: pure functions, no I/O."""
from __future__ import annotations

import base64
import json
import re
from decimal import Decimal, InvalidOperation
from typing import Any

USDC = {
    "base": {"address": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", "name": "USD Coin", "version": "2"},
    "base-sepolia": {"address": "0x036CbD53842c5426634e7929541eC2318f3dCF7e", "name": "USDC", "version": "2"},
}
NETWORKS = tuple(USDC)
_ADDRESS = re.compile(r"^0x[0-9a-fA-F]{40}$")
_PRICE = re.compile(r"^\d+(\.\d{1,6})?$")


def to_atomic_usdc(price: str) -> str:
    """'0.01' USDC to atomic units (6 decimals): '10000'."""
    if not isinstance(price, str):
        raise TypeError('price must be a decimal string such as "0.01"')
    p = price.strip().lstrip("$")
    if not _PRICE.match(p):
        raise ValueError(f'price must be a positive decimal with at most 6 decimals, got "{price}"')
    try:
        atomic = int(Decimal(p) * 1_000_000)
    except InvalidOperation as e:  # pragma: no cover
        raise ValueError("bad price") from e
    if atomic <= 0:
        raise ValueError("price must be greater than 0")
    return str(atomic)


def is_address(a: Any) -> bool:
    return isinstance(a, str) and bool(_ADDRESS.match(a))


def build_requirements(*, price: str, pay_to: str, network: str, resource: str, description: str,
                       mime_type: str = "application/json", max_timeout_seconds: int = 60) -> dict:
    usdc = USDC[network]
    return {
        "scheme": "exact",
        "network": network,
        "maxAmountRequired": to_atomic_usdc(price),
        "resource": resource,
        "description": description,
        "mimeType": mime_type,
        "payTo": pay_to,
        "maxTimeoutSeconds": max_timeout_seconds,
        "asset": usdc["address"],
        "extra": {"name": usdc["name"], "version": usdc["version"]},
    }


def encode_header(value: Any) -> str:
    return base64.b64encode(json.dumps(value, separators=(",", ":")).encode()).decode()


def decode_header(header: str) -> Any:
    return json.loads(base64.b64decode(header.strip(), validate=True).decode())


def read_payment(header: str, req: dict) -> tuple[dict | None, str | None]:
    """Decodes X-PAYMENT and checks scheme and network. Returns (payment, None) or (None, reason)."""
    try:
        p = decode_header(header)
    except Exception:
        return None, "X-PAYMENT is not base64 encoded JSON"
    if not isinstance(p, dict) or not isinstance(p.get("payload"), dict):
        return None, "X-PAYMENT has no payload"
    if p.get("x402Version") != 1:
        return None, f"unsupported x402Version {p.get('x402Version')}"
    if p.get("scheme") != req["scheme"]:
        return None, f"unsupported scheme {p.get('scheme')}"
    if p.get("network") != req["network"]:
        return None, f"wrong network {p.get('network')}, expected {req['network']}"
    return p, None

agentgate-source: code mirror on Cloudflare Pages. All repos