diff --git a/e2e/clients/httpx/main.py b/e2e/clients/httpx/main.py index 8313d9ecec..fdf976316a 100644 --- a/e2e/clients/httpx/main.py +++ b/e2e/clients/httpx/main.py @@ -31,7 +31,10 @@ exit(1) if not evm_private_key and not svm_private_key: - error_result = {"success": False, "error": "At least one of EVM_PRIVATE_KEY or SVM_PRIVATE_KEY must be set"} + error_result = { + "success": False, + "error": "At least one of EVM_PRIVATE_KEY or SVM_PRIVATE_KEY must be set", + } print(json.dumps(error_result)) exit(1) @@ -76,7 +79,9 @@ async def main(): } # Check for payment response header (V2: PAYMENT-RESPONSE, V1: X-PAYMENT-RESPONSE) - payment_header = response.headers.get("PAYMENT-RESPONSE") or response.headers.get("X-PAYMENT-RESPONSE") + payment_header = response.headers.get( + "PAYMENT-RESPONSE" + ) or response.headers.get("X-PAYMENT-RESPONSE") if payment_header: payment_response = decode_payment_response_header(payment_header) result["payment_response"] = payment_response.model_dump() diff --git a/e2e/clients/requests/main.py b/e2e/clients/requests/main.py index 873e071655..e1bcef0b6b 100644 --- a/e2e/clients/requests/main.py +++ b/e2e/clients/requests/main.py @@ -13,7 +13,6 @@ from x402.mechanisms.evm.exact import register_exact_evm_client from x402.mechanisms.svm import KeypairSigner from x402.mechanisms.svm.exact import register_exact_svm_client -import requests # Load environment variables load_dotenv() @@ -30,7 +29,10 @@ exit(1) if not evm_private_key and not svm_private_key: - error_result = {"success": False, "error": "At least one of EVM_PRIVATE_KEY or SVM_PRIVATE_KEY must be set"} + error_result = { + "success": False, + "error": "At least one of EVM_PRIVATE_KEY or SVM_PRIVATE_KEY must be set", + } print(json.dumps(error_result)) exit(1) @@ -70,7 +72,9 @@ def main(): } # Check for payment response header (V2: PAYMENT-RESPONSE, V1: X-PAYMENT-RESPONSE) - payment_header = response.headers.get("PAYMENT-RESPONSE") or response.headers.get("X-PAYMENT-RESPONSE") + payment_header = response.headers.get( + "PAYMENT-RESPONSE" + ) or response.headers.get("X-PAYMENT-RESPONSE") if payment_header: payment_response = decode_payment_response_header(payment_header) result["payment_response"] = payment_response.model_dump() diff --git a/e2e/facilitators/python/bazaar.py b/e2e/facilitators/python/bazaar.py index af8fae1111..cc5accf6ba 100644 --- a/e2e/facilitators/python/bazaar.py +++ b/e2e/facilitators/python/bazaar.py @@ -43,8 +43,7 @@ def to_dict(self) -> dict[str, Any]: class BazaarCatalog: - """Catalog for storing discovered x402 resources. - """ + """Catalog for storing discovered x402 resources.""" def __init__(self) -> None: self._resources: dict[str, DiscoveredResource] = {} @@ -79,9 +78,7 @@ def catalog_resource( metadata={}, ) - def get_resources( - self, limit: int = 100, offset: int = 0 - ) -> dict[str, Any]: + def get_resources(self, limit: int = 100, offset: int = 0) -> dict[str, Any]: """Get paginated list of discovered resources. Args: @@ -108,4 +105,3 @@ def get_resources( def get_count(self) -> int: """Get total count of discovered resources.""" return len(self._resources) - diff --git a/e2e/facilitators/python/main.py b/e2e/facilitators/python/main.py index 292583c247..9160860027 100644 --- a/e2e/facilitators/python/main.py +++ b/e2e/facilitators/python/main.py @@ -14,7 +14,6 @@ import os import sys -from datetime import datetime from typing import Any from dotenv import load_dotenv @@ -23,7 +22,7 @@ from solders.keypair import Keypair from x402 import x402Facilitator -from x402.extensions.bazaar import DiscoveredResource, extract_discovery_info +from x402.extensions.bazaar import extract_discovery_info from x402.mechanisms.evm import FacilitatorWeb3Signer from x402.mechanisms.evm.exact import register_exact_evm_facilitator from x402.mechanisms.svm import FacilitatorKeypairSigner diff --git a/e2e/legacy/servers/fastapi/main.py b/e2e/legacy/servers/fastapi/main.py index 230ee05b39..9921dd6ba4 100644 --- a/e2e/legacy/servers/fastapi/main.py +++ b/e2e/legacy/servers/fastapi/main.py @@ -2,11 +2,10 @@ import signal import sys import asyncio -from typing import Any, Dict, Optional +from typing import Any, Dict from dotenv import load_dotenv from fastapi import FastAPI, HTTPException -from fastapi.responses import JSONResponse from x402.fastapi.middleware import require_payment from x402.types import EIP712Domain, TokenAmount, TokenAsset from x402.chains import ( diff --git a/e2e/legacy/servers/flask/main.py b/e2e/legacy/servers/flask/main.py index f20e5ae79e..5d5cc624de 100644 --- a/e2e/legacy/servers/flask/main.py +++ b/e2e/legacy/servers/flask/main.py @@ -2,7 +2,7 @@ import signal import sys import logging -from flask import Flask, jsonify, request +from flask import Flask, jsonify from dotenv import load_dotenv from x402.flask.middleware import PaymentMiddleware diff --git a/e2e/servers/fastapi/main.py b/e2e/servers/fastapi/main.py index fb9da5aa0f..1d582a8823 100644 --- a/e2e/servers/fastapi/main.py +++ b/e2e/servers/fastapi/main.py @@ -8,14 +8,12 @@ from dotenv import load_dotenv from fastapi import FastAPI, HTTPException -from fastapi.responses import JSONResponse # Import from new x402 package from x402 import x402ResourceServer from x402.http import FacilitatorConfig, HTTPFacilitatorClient from x402.http.middleware.fastapi import payment_middleware from x402.mechanisms.evm.exact import ( - ExactEvmServerScheme, register_exact_evm_server, ) from x402.mechanisms.svm.exact import register_exact_svm_server diff --git a/examples/python/clients/payment-identifier/.env-local b/examples/python/clients/payment-identifier/.env-local new file mode 100644 index 0000000000..aab972c207 --- /dev/null +++ b/examples/python/clients/payment-identifier/.env-local @@ -0,0 +1,3 @@ +EVM_PRIVATE_KEY= +RESOURCE_SERVER_URL=http://localhost:4022 +ENDPOINT_PATH=/weather diff --git a/examples/python/clients/payment-identifier/README.md b/examples/python/clients/payment-identifier/README.md new file mode 100644 index 0000000000..f8f1e4ab9d --- /dev/null +++ b/examples/python/clients/payment-identifier/README.md @@ -0,0 +1,126 @@ +# Payment-Identifier Extension Client Example + +Example client demonstrating how to use the `payment-identifier` extension to enable **idempotency** when making payments. + +## How It Works + +1. Client generates a unique payment ID using `generate_payment_id()` +2. Client includes the payment ID in the `PaymentPayload` using `append_payment_identifier_to_extensions()` +3. Server caches responses keyed by payment ID +4. Retry requests with the same payment ID return cached responses without re-processing payment + +```python +from x402 import x402Client +from x402.extensions.payment_identifier import ( + append_payment_identifier_to_extensions, + generate_payment_id, +) +from x402.http.clients import x402HttpxClient + +client = x402Client() +# ... register schemes ... + +# Generate a unique payment ID for this logical request +payment_id = generate_payment_id() + +# Hook into payment flow to add the payment ID before payload creation +async def before_payment_creation(context): + extensions = context.payment_required.extensions + if extensions is not None: + append_payment_identifier_to_extensions(extensions, payment_id) + +client.on_before_payment_creation(before_payment_creation) + +async with x402HttpxClient(client) as http: + # First request - payment is processed + response1 = await http.get(url) + + # Retry with same payment ID - cached response returned (no payment) + response2 = await http.get(url) +``` + +## Prerequisites + +- Python 3.10+ +- uv (install via [docs.astral.sh/uv](https://docs.astral.sh/uv/getting-started/installation/)) +- A running payment-identifier server (see [payment-identifier server example](../../servers/payment-identifier)) +- Valid EVM private key for making payments (Base Sepolia with USDC) + +## Setup + +1. Install dependencies: + +```bash +uv sync +``` + +2. Copy `.env-local` to `.env` and add your private key: + +```bash +cp .env-local .env +``` + +Required environment variable: + +- `EVM_PRIVATE_KEY` - Ethereum private key for EVM payments + +3. Start the payment-identifier server (in another terminal): + +```bash +cd ../../servers/payment-identifier +uv run python main.py +``` + +4. Run the client: + +```bash +uv run python main.py +``` + +## Expected Output + +``` +Generated Payment ID: pay_7d5d747be160e280504c099d984bcfe0 + +==================================================== +First Request (with payment ID: pay_7d5d747be160e280504c099d984bcfe0) +==================================================== +Making request to: http://localhost:4022/weather + +Response (1523ms): {"report": {"weather": "sunny", "temperature": 70, "cached": false}} + +Payment settled on eip155:84532 + +==================================================== +Second Request (SAME payment ID: pay_7d5d747be160e280504c099d984bcfe0) +==================================================== +Making request to: http://localhost:4022/weather + +Expected: Server returns cached response without payment processing + +Response (45ms): {"report": {"weather": "sunny", "temperature": 70, "cached": true}} + +No payment processed - response served from cache! + +==================================================== +Summary +==================================================== + Payment ID: pay_7d5d747be160e280504c099d984bcfe0 + First request: 1523ms (payment processed) + Second request: 45ms (cached) + Cached response was 97% faster! +``` + +## Use Cases + +- **Network failures**: Safely retry failed requests without duplicate payments +- **Client crashes**: Resume requests after restart using persisted payment IDs +- **Load balancing**: Same request can hit different servers with shared cache +- **Testing**: Replay requests during development without spending funds + +## Best Practices + +1. **Generate payment IDs at the logical request level**, not per retry +2. **Persist payment IDs** for long-running operations so they survive restarts +3. **Use descriptive prefixes** (e.g., `order_`, `sub_`) to identify payment types +4. **Don't reuse payment IDs** across different logical requests diff --git a/examples/python/clients/payment-identifier/main.py b/examples/python/clients/payment-identifier/main.py new file mode 100644 index 0000000000..8de03d540a --- /dev/null +++ b/examples/python/clients/payment-identifier/main.py @@ -0,0 +1,131 @@ +"""Payment-Identifier Extension Client Example. + +Demonstrates how to use the payment-identifier extension to enable idempotency +when making payments. This allows safe retries without duplicate payments. + +This example: +1. Makes a request with a unique payment ID +2. Makes a second request with the SAME payment ID +3. The second request returns from cache without payment processing + +Required environment variables: +- EVM_PRIVATE_KEY: The private key of the EVM signer +""" + +import asyncio +import os +import sys +import time + +from dotenv import load_dotenv +from eth_account import Account + +from x402 import x402Client +from x402.extensions.payment_identifier import ( + append_payment_identifier_to_extensions, + generate_payment_id, +) +from x402.http import x402HTTPClient +from x402.http.clients import x402HttpxClient +from x402.mechanisms.evm import EthAccountSigner +from x402.mechanisms.evm.exact.register import register_exact_evm_client +from x402.schemas import PaymentCreationContext + +load_dotenv() + + +async def main() -> None: + """Main entry point demonstrating payment-identifier extension for idempotency.""" + # Validate environment + private_key = os.getenv("EVM_PRIVATE_KEY") + if not private_key: + print("Error: EVM_PRIVATE_KEY environment variable is required") + sys.exit(1) + + base_url = os.getenv("RESOURCE_SERVER_URL", "http://localhost:4022") + endpoint_path = os.getenv("ENDPOINT_PATH", "/weather") + url = f"{base_url}{endpoint_path}" + + # Create x402 client + account = Account.from_key(private_key) + client = x402Client() + register_exact_evm_client(client, EthAccountSigner(account)) + + # Generate a unique payment ID for this request + payment_id = generate_payment_id() + print(f"\nGenerated Payment ID: {payment_id}") + + # Hook into the payment flow to add payment identifier BEFORE payload creation + # We modify paymentRequired.extensions to include our payment ID + async def before_payment_creation(context: PaymentCreationContext) -> None: + extensions = context.payment_required.extensions + if extensions is not None: + # Append our payment ID to the extensions (only if server declared the extension) + append_payment_identifier_to_extensions(extensions, payment_id) + + client.on_before_payment_creation(before_payment_creation) + + # Create HTTP client helper for payment response extraction + http_client = x402HTTPClient(client) + + # First request - will process payment + print("\n" + "=" * 52) + print(f"First Request (with payment ID: {payment_id})") + print("=" * 52) + print(f"Making request to: {url}\n") + + async with x402HttpxClient(client) as http: + start_time1 = time.time() + response1 = await http.get(url) + await response1.aread() + duration1 = int((time.time() - start_time1) * 1000) + + print(f"Response ({duration1}ms): {response1.text}") + + if response1.is_success: + try: + settle_response = http_client.get_payment_settle_response( + lambda name: response1.headers.get(name) + ) + print(f"\nPayment settled on {settle_response.network}") + except ValueError: + pass + + # Second request - same payment ID, should return from cache + print("\n" + "=" * 52) + print(f"Second Request (SAME payment ID: {payment_id})") + print("=" * 52) + print(f"Making request to: {url}\n") + print("Expected: Server returns cached response without payment processing\n") + + start_time2 = time.time() + response2 = await http.get(url) + await response2.aread() + duration2 = int((time.time() - start_time2) * 1000) + + print(f"Response ({duration2}ms): {response2.text}") + + if response2.is_success: + try: + settle_response = http_client.get_payment_settle_response( + lambda name: response2.headers.get(name) + ) + print("\nPayment settled (unexpected - should have been cached)") + except ValueError: + print("\nNo payment processed - response served from cache!") + + # Summary + print("\n" + "=" * 52) + print("Summary") + print("=" * 52) + print(f" Payment ID: {payment_id}") + print(f" First request: {duration1}ms (payment processed)") + print(f" Second request: {duration2}ms (cached)") + if duration2 < duration1 and duration1 > 0: + speedup = round((1 - duration2 / duration1) * 100) + print(f" Cached response was {speedup}% faster!") + print() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/python/clients/payment-identifier/pyproject.toml b/examples/python/clients/payment-identifier/pyproject.toml new file mode 100644 index 0000000000..2061533c9f --- /dev/null +++ b/examples/python/clients/payment-identifier/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "x402-payment-identifier-client-example" +version = "0.1.0" +description = "Example of using x402 payment-identifier extension for idempotent payments" +requires-python = ">=3.10" +dependencies = [ + "python-dotenv>=1.0.0", + "x402[httpx,evm,extensions]", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["."] + +[tool.hatch.metadata] +allow-direct-references = true + +[tool.uv] +package = false + +[tool.uv.sources] +x402 = { path = "../../../../python/x402", editable = true } diff --git a/examples/python/facilitator/advanced/all_networks.py b/examples/python/facilitator/advanced/all_networks.py index e763833f9a..e78078663b 100644 --- a/examples/python/facilitator/advanced/all_networks.py +++ b/examples/python/facilitator/advanced/all_networks.py @@ -152,7 +152,7 @@ async def verify(request: VerifyRequest): } except Exception as e: print(f"Verify error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.post("/settle") @@ -194,7 +194,7 @@ async def settle(request: SettleRequest): "transaction": "", } - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/supported") @@ -222,7 +222,7 @@ async def supported(): } except Exception as e: print(f"Supported error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/health") diff --git a/examples/python/facilitator/advanced/bazaar.py b/examples/python/facilitator/advanced/bazaar.py index ced0e9ee55..3c18fadf30 100644 --- a/examples/python/facilitator/advanced/bazaar.py +++ b/examples/python/facilitator/advanced/bazaar.py @@ -212,7 +212,7 @@ async def verify(request: VerifyRequest): } except Exception as e: print(f"Verify error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.post("/settle") @@ -254,7 +254,7 @@ async def settle(request: SettleRequest): "payer": None, } - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/supported") @@ -282,7 +282,7 @@ async def supported(): } except Exception as e: print(f"Supported error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/discovery/resources") @@ -305,7 +305,7 @@ async def discovery_resources(): } except Exception as e: print(f"Discovery error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/health") diff --git a/examples/python/facilitator/basic/main.py b/examples/python/facilitator/basic/main.py index 3901a3f6ad..66197be635 100644 --- a/examples/python/facilitator/basic/main.py +++ b/examples/python/facilitator/basic/main.py @@ -152,7 +152,7 @@ async def verify(request: VerifyRequest): } except Exception as e: print(f"Verify error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.post("/settle") @@ -194,7 +194,7 @@ async def settle(request: SettleRequest): "transaction": "", } - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/supported") @@ -222,7 +222,7 @@ async def supported(): } except Exception as e: print(f"Supported error: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e @app.get("/health") diff --git a/examples/python/legacy/fullstack/fastapi/main.py b/examples/python/legacy/fullstack/fastapi/main.py index c924bd8075..e3914e0e14 100644 --- a/examples/python/legacy/fullstack/fastapi/main.py +++ b/examples/python/legacy/fullstack/fastapi/main.py @@ -1,5 +1,4 @@ import os -from typing import Any, Dict from dotenv import load_dotenv from fastapi import FastAPI diff --git a/examples/python/legacy/sync.py b/examples/python/legacy/sync.py index 142983229d..48a5c412cf 100755 --- a/examples/python/legacy/sync.py +++ b/examples/python/legacy/sync.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 -import os import subprocess from pathlib import Path diff --git a/examples/python/servers/payment-identifier/.env-local b/examples/python/servers/payment-identifier/.env-local new file mode 100644 index 0000000000..f35e44142c --- /dev/null +++ b/examples/python/servers/payment-identifier/.env-local @@ -0,0 +1 @@ +EVM_ADDRESS= diff --git a/examples/python/servers/payment-identifier/README.md b/examples/python/servers/payment-identifier/README.md new file mode 100644 index 0000000000..d736454b5c --- /dev/null +++ b/examples/python/servers/payment-identifier/README.md @@ -0,0 +1,135 @@ +# Payment-Identifier Extension Server Example + +Example server demonstrating how to implement the `payment-identifier` extension for **idempotent** payment processing. + +## How It Works + +1. Server advertises `payment-identifier` extension support in PaymentRequired responses +2. Server extracts payment ID from incoming PaymentPayload using `extract_payment_identifier()` +3. After settlement, server caches the response keyed by payment ID +4. Duplicate requests with the same payment ID return cached response without payment processing + +```python +from x402.extensions.payment_identifier import ( + PAYMENT_IDENTIFIER, + declare_payment_identifier_extension, + extract_payment_identifier, +) +from x402.server import x402ResourceServer + +server = x402ResourceServer(facilitator) + +# Advertise extension support in route config +routes = { + "GET /weather": RouteConfig( + accepts=[...], + extensions={ + PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=False), + }, + ), +} + +# Cache response after settlement +async def after_settle(ctx): + payment_id = extract_payment_identifier(ctx.payment_payload) + if payment_id: + idempotency_cache[payment_id] = cached_response + +server.on_after_settle(after_settle) +``` + +## Prerequisites + +- Python 3.10+ +- uv (install via [docs.astral.sh/uv](https://docs.astral.sh/uv/getting-started/installation/)) +- Valid EVM address for receiving payments (Base Sepolia) + +## Setup + +1. Install dependencies: + +```bash +uv sync +``` + +2. Copy `.env-local` to `.env` and add your EVM address: + +```bash +cp .env-local .env +``` + +Required environment variable: + +- `EVM_ADDRESS` - Ethereum address to receive payments + +3. Run the server: + +```bash +uv run python main.py +``` + +## Expected Output + +``` +Payment-Identifier Example Server + Listening at http://localhost:4022 + +Idempotency Configuration: + - Cache TTL: 1 hour + - Payment ID: optional (required: false) + +How it works: + 1. Client sends payment with a unique payment ID + 2. Server caches the response keyed by payment ID + 3. If same payment ID is seen within 1 hour, cached response is returned + 4. No duplicate payment processing occurs +``` + +When requests come in: + +``` +[Idempotency] Checking payment ID: pay_7d5d747be160e280504c099d984bcfe0 +[Idempotency] Cache MISS - proceeding with payment +[Idempotency] Caching response for payment ID: pay_7d5d747be160e280504c099d984bcfe0 + +[Idempotency] Checking payment ID: pay_7d5d747be160e280504c099d984bcfe0 +[Idempotency] Cache HIT - returning cached response (age: 2s) +``` + +## Testing with the Client + +Run the [payment-identifier client example](../../clients/payment-identifier) in another terminal: + +```bash +cd ../../clients/payment-identifier +uv run python main.py +``` + +## Extension Configuration + +### Optional Payment ID (Default) + +```python +extensions={ + PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=False), +} +``` + +Clients may optionally provide a payment ID. If provided, it enables idempotency. + +### Required Payment ID + +```python +extensions={ + PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=True), +} +``` + +Clients must provide a payment ID. Requests without one will be rejected. + +## Production Considerations + +1. **Use a distributed cache** (Redis, Memcached) instead of in-memory dict +2. **Configure appropriate TTL** based on your use case +3. **Consider cache key structure** to include route/method for multi-endpoint servers +4. **Monitor cache hit rates** to optimize performance diff --git a/examples/python/servers/payment-identifier/main.py b/examples/python/servers/payment-identifier/main.py new file mode 100644 index 0000000000..0830fb336b --- /dev/null +++ b/examples/python/servers/payment-identifier/main.py @@ -0,0 +1,221 @@ +"""Payment-Identifier Extension Server Example. + +Demonstrates how to implement a resource server that supports the payment-identifier +extension for idempotent payment processing. + +This server: +1. Advertises payment-identifier extension support in PaymentRequired responses +2. Caches responses keyed by payment ID after settlement +3. Returns cached responses for duplicate payment IDs without re-processing + +Required environment variables: +- EVM_ADDRESS: The EVM address to receive payments +""" + +import base64 +import json +import os +import time +from dataclasses import dataclass +from typing import Any + +from dotenv import load_dotenv +from fastapi import FastAPI, Request, Response +from pydantic import BaseModel + +from x402.extensions.payment_identifier import ( + PAYMENT_IDENTIFIER, + declare_payment_identifier_extension, + extract_payment_identifier, +) +from x402.http import FacilitatorConfig, HTTPFacilitatorClient, PaymentOption +from x402.http.middleware.fastapi import PaymentMiddlewareASGI +from x402.http.types import RouteConfig +from x402.mechanisms.evm.exact import ExactEvmServerScheme +from x402.schemas import Network, PaymentPayload, SettleContext +from x402.server import x402ResourceServer + +load_dotenv() + +# Config +EVM_ADDRESS = os.getenv("EVM_ADDRESS") +EVM_NETWORK: Network = "eip155:84532" # Base Sepolia +FACILITATOR_URL = os.getenv("FACILITATOR_URL", "https://x402.org/facilitator") + +if not EVM_ADDRESS: + raise ValueError("Missing required EVM_ADDRESS environment variable") + + +# Response schemas +class WeatherReport(BaseModel): + weather: str + temperature: int + cached: bool + + +class WeatherResponse(BaseModel): + report: WeatherReport + + +# Simple in-memory cache for idempotency +# In production, use Redis or another distributed cache +@dataclass +class CachedResponse: + timestamp: float + response: dict[str, Any] + + +idempotency_cache: dict[str, CachedResponse] = {} +CACHE_TTL_SECONDS = 60 * 60 # 1 hour + + +def cleanup_expired_entries() -> None: + """Clean up expired entries from the cache.""" + now = time.time() + expired_keys = [ + key + for key, value in idempotency_cache.items() + if now - value.timestamp > CACHE_TTL_SECONDS + ] + for key in expired_keys: + del idempotency_cache[key] + + +# App +app = FastAPI() + +# x402 Setup +facilitator = HTTPFacilitatorClient(FacilitatorConfig(url=FACILITATOR_URL)) +server = x402ResourceServer(facilitator) +server.register(EVM_NETWORK, ExactEvmServerScheme()) + + +# Hook after settlement to cache the response +async def after_settle(ctx: SettleContext) -> None: + """Cache the response after successful payment settlement.""" + payment_id = extract_payment_identifier(ctx.payment_payload) + if payment_id: + print(f"[Idempotency] Caching response for payment ID: {payment_id}") + idempotency_cache[payment_id] = CachedResponse( + timestamp=time.time(), + response={ + "report": { + "weather": "sunny", + "temperature": 70, + "cached": False, + } + }, + ) + + +server.on_after_settle(after_settle) + + +# Route configuration with payment-identifier extension advertised +routes = { + "GET /weather": RouteConfig( + accepts=[ + PaymentOption( + scheme="exact", + price="$0.001", + network=EVM_NETWORK, + pay_to=EVM_ADDRESS, + ), + ], + description="Weather data with idempotency support", + mime_type="application/json", + # Advertise payment-identifier extension support (required=False means optional) + extensions={ + PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=False), + }, + ), +} + + +# Custom middleware to check idempotency cache before payment processing +@app.middleware("http") +async def idempotency_middleware(request: Request, call_next: Any) -> Response: + """Check idempotency cache before payment processing.""" + # Clean up expired entries periodically + cleanup_expired_entries() + + # Only check for payment header on protected routes + payment_header = request.headers.get("X-Payment") + if payment_header and request.url.path == "/weather": + try: + # Decode payment header to extract payment ID + payment_data = json.loads(base64.b64decode(payment_header).decode("utf-8")) + payment_payload = PaymentPayload.model_validate(payment_data) + payment_id = extract_payment_identifier(payment_payload) + + if payment_id: + print(f"[Idempotency] Checking payment ID: {payment_id}") + cached = idempotency_cache.get(payment_id) + + if cached: + age = time.time() - cached.timestamp + if age < CACHE_TTL_SECONDS: + print( + f"[Idempotency] Cache HIT - returning cached response (age: {int(age)}s)" + ) + # Return cached response with cached flag set to true + cached_response = { + "report": { + **cached.response["report"], + "cached": True, + } + } + return Response( + content=json.dumps(cached_response), + media_type="application/json", + status_code=200, + ) + else: + print("[Idempotency] Cache EXPIRED - proceeding with payment") + del idempotency_cache[payment_id] + else: + print("[Idempotency] Cache MISS - proceeding with payment") + except Exception: + # Invalid payment header format, continue to normal flow + pass + + return await call_next(request) + + +# Add payment middleware after idempotency middleware +app.add_middleware(PaymentMiddlewareASGI, routes=routes, server=server) + + +# Routes +@app.get("/health") +async def health_check() -> dict[str, str]: + return {"status": "ok"} + + +@app.get("/weather") +async def get_weather() -> WeatherResponse: + """Return weather data. Response may be cached based on payment ID.""" + return WeatherResponse( + report=WeatherReport( + weather="sunny", + temperature=70, + cached=False, + ) + ) + + +if __name__ == "__main__": + import uvicorn + + print("\nPayment-Identifier Example Server") + print(" Listening at http://localhost:4022") + print("\nIdempotency Configuration:") + print(" - Cache TTL: 1 hour") + print(" - Payment ID: optional (required: false)") + print("\nHow it works:") + print(" 1. Client sends payment with a unique payment ID") + print(" 2. Server caches the response keyed by payment ID") + print(" 3. If same payment ID is seen within 1 hour, cached response is returned") + print(" 4. No duplicate payment processing occurs\n") + + uvicorn.run(app, host="0.0.0.0", port=4022) diff --git a/examples/python/servers/payment-identifier/pyproject.toml b/examples/python/servers/payment-identifier/pyproject.toml new file mode 100644 index 0000000000..96f6f400ec --- /dev/null +++ b/examples/python/servers/payment-identifier/pyproject.toml @@ -0,0 +1,26 @@ +[project] +name = "x402-payment-identifier-server-example" +version = "0.1.0" +description = "Example of using x402 payment-identifier extension for idempotent payments" +requires-python = ">=3.10" +dependencies = [ + "python-dotenv>=1.0.0", + "x402[fastapi,evm,extensions]", + "uvicorn[standard]>=0.40.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["."] + +[tool.hatch.metadata] +allow-direct-references = true + +[tool.uv] +package = false + +[tool.uv.sources] +x402 = { path = "../../../../python/x402", editable = true } diff --git a/python/x402/changelog.d/1111.bugfix.md b/python/x402/changelog.d/1111.bugfix.md new file mode 100644 index 0000000000..703a0818e1 --- /dev/null +++ b/python/x402/changelog.d/1111.bugfix.md @@ -0,0 +1 @@ +Added payment-identifier extension for tracking and validating payment identifiers diff --git a/python/x402/extensions/__init__.py b/python/x402/extensions/__init__.py index ae3f0f5803..ff74ea706b 100644 --- a/python/x402/extensions/__init__.py +++ b/python/x402/extensions/__init__.py @@ -22,7 +22,6 @@ QueryDiscoveryInfo, QueryInput, QueryParamMethods, - ValidationResult, bazaar_resource_server_extension, declare_discovery_extension, extract_discovery_info, @@ -30,6 +29,36 @@ validate_and_extract, validate_discovery_extension, ) +from .bazaar import ( + ValidationResult as BazaarValidationResult, +) + +# Create alias for backward compatibility +ValidationResult = BazaarValidationResult + +from .payment_identifier import ( # noqa: E402 + PAYMENT_ID_MAX_LENGTH, + PAYMENT_ID_MIN_LENGTH, + PAYMENT_ID_PATTERN, + PAYMENT_IDENTIFIER, + PaymentIdentifierExtension, + PaymentIdentifierInfo, + PaymentIdentifierSchema, + PaymentIdentifierValidationResult, + append_payment_identifier_to_extensions, + declare_payment_identifier_extension, + extract_and_validate_payment_identifier, + extract_payment_identifier, + generate_payment_id, + has_payment_identifier, + is_payment_identifier_extension, + is_payment_identifier_required, + is_valid_payment_id, + payment_identifier_resource_server_extension, + payment_identifier_schema, + validate_payment_identifier, + validate_payment_identifier_requirement, +) __all__ = [ # Constants @@ -55,7 +84,9 @@ "DeclareBodyDiscoveryConfig", "OutputConfig", # Result types - "ValidationResult", + "ValidationResult", # From bazaar (for backward compatibility) + "BazaarValidationResult", # From bazaar (alias for explicit usage) + "PaymentIdentifierValidationResult", # From payment_identifier "DiscoveredResource", # Server extension "bazaar_resource_server_extension", @@ -65,4 +96,33 @@ "extract_discovery_info", "extract_discovery_info_from_extension", "validate_and_extract", + # Payment Identifier constants + "PAYMENT_IDENTIFIER", + "PAYMENT_ID_MIN_LENGTH", + "PAYMENT_ID_MAX_LENGTH", + "PAYMENT_ID_PATTERN", + # Payment Identifier types + "PaymentIdentifierInfo", + "PaymentIdentifierExtension", + "PaymentIdentifierSchema", + # Payment Identifier schema + "payment_identifier_schema", + # Payment Identifier utils + "generate_payment_id", + "is_valid_payment_id", + # Payment Identifier client functions + "append_payment_identifier_to_extensions", + # Payment Identifier server functions + "declare_payment_identifier_extension", + "payment_identifier_resource_server_extension", + # Payment Identifier validation functions + "is_payment_identifier_extension", + "validate_payment_identifier", + "extract_payment_identifier", + "extract_and_validate_payment_identifier", + "has_payment_identifier", + "is_payment_identifier_required", + "validate_payment_identifier_requirement", + "PaymentIdentifierValidationResult", + "BazaarValidationResult", ] diff --git a/python/x402/extensions/bazaar/facilitator.py b/python/x402/extensions/bazaar/facilitator.py index 41672af670..abf31a9fbc 100644 --- a/python/x402/extensions/bazaar/facilitator.py +++ b/python/x402/extensions/bazaar/facilitator.py @@ -131,8 +131,8 @@ def _get_method_from_info(info: DiscoveryInfo | dict[str, Any]) -> str: input_data = info.get("input", {}) return input_data.get("method", "UNKNOWN") - if isinstance(info, (QueryDiscoveryInfo, BodyDiscoveryInfo)): - if isinstance(info.input, (QueryInput, BodyInput)): + if isinstance(info, QueryDiscoveryInfo | BodyDiscoveryInfo): + if isinstance(info.input, QueryInput | BodyInput): return info.input.method or "UNKNOWN" return "UNKNOWN" diff --git a/python/x402/extensions/payment_identifier/__init__.py b/python/x402/extensions/payment_identifier/__init__.py new file mode 100644 index 0000000000..084997ebb6 --- /dev/null +++ b/python/x402/extensions/payment_identifier/__init__.py @@ -0,0 +1,130 @@ +"""Payment-Identifier Extension for x402 v2. + +Enables clients to provide an idempotency key (`id`) that resource servers +can use for deduplication of payment requests. + +## Usage + +### For Resource Servers + +```python +from x402.extensions.payment_identifier import ( + declare_payment_identifier_extension, + PAYMENT_IDENTIFIER, +) + +# Advertise support in PaymentRequired response (optional identifier) +payment_required = { + "x402Version": 2, + "resource": {...}, + "accepts": [...], + "extensions": { + PAYMENT_IDENTIFIER: declare_payment_identifier_extension() + } +} + +# Require payment identifier +payment_required_strict = { + "x402Version": 2, + "resource": {...}, + "accepts": [...], + "extensions": { + PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=True) + } +} +``` + +### For Clients + +```python +from x402.extensions.payment_identifier import append_payment_identifier_to_extensions + +# Get extensions from server's PaymentRequired response +extensions = {...payment_required.extensions} + +# Append payment ID (only if server declared the extension) +append_payment_identifier_to_extensions(extensions) + +# Include in PaymentPayload +payment_payload = { + "x402Version": 2, + "resource": payment_required.resource, + "accepted": selected_payment_option, + "payload": {...}, + "extensions": extensions +} +``` + +### For Idempotency Implementation + +```python +from x402.extensions.payment_identifier import extract_payment_identifier + +# In your settle handler +id = extract_payment_identifier(payment_payload) +if id: + cached = await idempotency_store.get(id) + if cached: + return cached # Return cached response +``` +""" + +from .client import append_payment_identifier_to_extensions +from .schema import payment_identifier_schema +from .server import ( + PaymentIdentifierResourceServerExtension, + declare_payment_identifier_extension, + payment_identifier_resource_server_extension, +) +from .types import ( + PAYMENT_ID_MAX_LENGTH, + PAYMENT_ID_MIN_LENGTH, + PAYMENT_ID_PATTERN, + PAYMENT_IDENTIFIER, + PaymentIdentifierExtension, + PaymentIdentifierInfo, + PaymentIdentifierSchema, +) +from .utils import generate_payment_id, is_valid_payment_id +from .validation import ( + PaymentIdentifierValidationResult, + extract_and_validate_payment_identifier, + extract_payment_identifier, + has_payment_identifier, + is_payment_identifier_extension, + is_payment_identifier_required, + validate_payment_identifier, + validate_payment_identifier_requirement, +) + +__all__ = [ + # Constants + "PAYMENT_IDENTIFIER", + "PAYMENT_ID_MIN_LENGTH", + "PAYMENT_ID_MAX_LENGTH", + "PAYMENT_ID_PATTERN", + # Types + "PaymentIdentifierInfo", + "PaymentIdentifierExtension", + "PaymentIdentifierSchema", + # Schema + "payment_identifier_schema", + # Utils + "generate_payment_id", + "is_valid_payment_id", + # Client functions + "append_payment_identifier_to_extensions", + # Server functions + "declare_payment_identifier_extension", + "PaymentIdentifierResourceServerExtension", + "payment_identifier_resource_server_extension", + # Validation functions + "is_payment_identifier_extension", + "validate_payment_identifier", + "extract_payment_identifier", + "extract_and_validate_payment_identifier", + "has_payment_identifier", + "is_payment_identifier_required", + "validate_payment_identifier_requirement", + "PaymentIdentifierValidationResult", +] diff --git a/python/x402/extensions/payment_identifier/client.py b/python/x402/extensions/payment_identifier/client.py new file mode 100644 index 0000000000..efc370bd46 --- /dev/null +++ b/python/x402/extensions/payment_identifier/client.py @@ -0,0 +1,90 @@ +"""Client-side utilities for the Payment-Identifier Extension.""" + +from __future__ import annotations + +from typing import Any + +from .types import PAYMENT_IDENTIFIER, PaymentIdentifierExtension +from .utils import generate_payment_id, is_valid_payment_id +from .validation import is_payment_identifier_extension + + +def append_payment_identifier_to_extensions( + extensions: dict[str, Any], id: str | None = None +) -> dict[str, Any]: + """Append a payment identifier to the extensions object if the server declared support. + + This function reads the server's `payment-identifier` declaration from the extensions, + and appends the client's ID to it. If the extension is not present (server didn't declare it), + the extensions are returned unchanged. + + Args: + extensions: The extensions object from PaymentRequired (will be modified in place). + id: Optional custom payment ID. If not provided, a new ID will be generated. + + Returns: + The modified extensions object (same reference as input). + + Raises: + ValueError: If the provided ID is invalid. + + Example: + ```python + from x402.extensions.payment_identifier import append_payment_identifier_to_extensions + + # Get extensions from server's PaymentRequired response + extensions = {...payment_required.extensions} + + # Append a generated ID (only if server declared payment-identifier) + append_payment_identifier_to_extensions(extensions) + + # Or use a custom ID + append_payment_identifier_to_extensions(extensions, "pay_my_custom_id_12345") + + # Include in PaymentPayload + payment_payload = { + "x402Version": 2, + "resource": payment_required.resource, + "accepted": selected_payment_option, + "payload": {...}, + "extensions": extensions + } + ``` + """ + extension = extensions.get(PAYMENT_IDENTIFIER) + + # Only append if the server declared this extension with valid structure + if not is_payment_identifier_extension(extension): + return extensions + + payment_id = id if id is not None else generate_payment_id() + + if not is_valid_payment_id(payment_id): + raise ValueError( + f'Invalid payment ID: "{payment_id}". ' + "ID must be 16-128 characters and contain only alphanumeric characters, hyphens, and underscores." + ) + + # Convert extension to dict if it's a Pydantic model + if isinstance(extension, PaymentIdentifierExtension): + ext_dict = extension.model_dump(by_alias=True) + else: + ext_dict = dict(extension) + + # Get or create info section + info = ext_dict.get("info", {}) + if isinstance(info, dict): + info_dict = dict(info) + else: + # If it's a Pydantic model, convert to dict + if hasattr(info, "model_dump"): + info_dict = info.model_dump(by_alias=True) + else: + info_dict = {} + + # Append the ID to the existing extension info + info_dict["id"] = payment_id + ext_dict["info"] = info_dict + extensions[PAYMENT_IDENTIFIER] = ext_dict + + return extensions diff --git a/python/x402/extensions/payment_identifier/schema.py b/python/x402/extensions/payment_identifier/schema.py new file mode 100644 index 0000000000..b5ec105cc8 --- /dev/null +++ b/python/x402/extensions/payment_identifier/schema.py @@ -0,0 +1,24 @@ +"""JSON Schema definitions for the Payment-Identifier Extension.""" + +from typing import Any + +from .types import PAYMENT_ID_MAX_LENGTH, PAYMENT_ID_MIN_LENGTH + +# JSON Schema for validating payment identifier info. +# Compliant with JSON Schema Draft 2020-12. +payment_identifier_schema: dict[str, Any] = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "required": { + "type": "boolean", + }, + "id": { + "type": "string", + "minLength": PAYMENT_ID_MIN_LENGTH, + "maxLength": PAYMENT_ID_MAX_LENGTH, + "pattern": "^[a-zA-Z0-9_-]+$", + }, + }, + "required": ["required"], +} diff --git a/python/x402/extensions/payment_identifier/server.py b/python/x402/extensions/payment_identifier/server.py new file mode 100644 index 0000000000..e9473fdf2e --- /dev/null +++ b/python/x402/extensions/payment_identifier/server.py @@ -0,0 +1,102 @@ +"""Resource Server utilities for the Payment-Identifier Extension.""" + +from __future__ import annotations + +from typing import Any + +from .schema import payment_identifier_schema +from .types import PAYMENT_IDENTIFIER + + +def declare_payment_identifier_extension(required: bool = False) -> dict[str, Any]: + """Declare the payment-identifier extension for inclusion in PaymentRequired.extensions. + + Resource servers call this function to advertise support for payment identifiers. + The declaration indicates whether a payment identifier is required and includes + the schema that clients must follow. + + Args: + required: Whether clients must provide a payment identifier. Defaults to False. + + Returns: + A dictionary ready for PaymentRequired.extensions with the key PAYMENT_IDENTIFIER. + + Example: + ```python + from x402.extensions.payment_identifier import ( + declare_payment_identifier_extension, + PAYMENT_IDENTIFIER, + ) + + # Include in PaymentRequired response (optional identifier) + payment_required = { + "x402Version": 2, + "resource": {...}, + "accepts": [...], + "extensions": { + PAYMENT_IDENTIFIER: declare_payment_identifier_extension() + } + } + + # Require payment identifier + payment_required_strict = { + "x402Version": 2, + "resource": {...}, + "accepts": [...], + "extensions": { + PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=True) + } + } + ``` + """ + return { + "info": {"required": required}, + "schema": payment_identifier_schema, + } + + +class PaymentIdentifierResourceServerExtension: + """ResourceServerExtension implementation for payment-identifier. + + This extension doesn't require any enrichment hooks since the declaration + is static. It's provided for consistency with other extensions and for + potential future use with the extension registration system. + + Example: + ```python + from x402 import x402ResourceServer + from x402.extensions.payment_identifier import ( + payment_identifier_resource_server_extension, + ) + + server = x402ResourceServer(facilitator_client) + server.register_extension(payment_identifier_resource_server_extension) + ``` + """ + + @property + def key(self) -> str: + """Extension key.""" + return PAYMENT_IDENTIFIER + + def enrich_declaration( + self, + declaration: Any, + transport_context: Any, + ) -> Any: + """Enrich extension declaration with transport-specific data. + + For payment-identifier, no enrichment is needed since the declaration is static. + + Args: + declaration: The extension declaration to enrich. + transport_context: Framework-specific context (e.g., HTTP request). + + Returns: + Unchanged declaration (no enrichment needed). + """ + return declaration + + +# Singleton instance for convenience +payment_identifier_resource_server_extension = PaymentIdentifierResourceServerExtension() diff --git a/python/x402/extensions/payment_identifier/types.py b/python/x402/extensions/payment_identifier/types.py new file mode 100644 index 0000000000..c47652d971 --- /dev/null +++ b/python/x402/extensions/payment_identifier/types.py @@ -0,0 +1,62 @@ +"""Type definitions for the Payment-Identifier Extension. + +Enables clients to provide an idempotency key that resource servers +can use for deduplication of payment requests. +""" + +from __future__ import annotations + +import re +from typing import Any + +from pydantic import BaseModel, Field + +# Extension identifier constant for the payment-identifier extension +PAYMENT_IDENTIFIER = "payment-identifier" + +# Minimum length for payment identifier +PAYMENT_ID_MIN_LENGTH = 16 + +# Maximum length for payment identifier +PAYMENT_ID_MAX_LENGTH = 128 + +# Pattern for valid payment identifier characters (alphanumeric, hyphens, underscores) +PAYMENT_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$") + + +class PaymentIdentifierInfo(BaseModel): + """Payment identifier info containing the required flag and client-provided ID. + + Attributes: + required: Whether the server requires clients to include a payment identifier. + When true, clients must provide an `id` or receive a 400 Bad Request. + id: Client-provided unique identifier for idempotency. + Must be 16-128 characters, alphanumeric with hyphens and underscores allowed. + """ + + required: bool + id: str | None = None + + model_config = {"extra": "allow"} + + +class PaymentIdentifierExtension(BaseModel): + """Payment identifier extension with info and schema. + + Used both for server-side declarations (info without id) and + client-side payloads (info with id). + + Attributes: + info: The payment identifier info. + Server declarations have required only, clients add the id. + schema: JSON Schema validating the info structure. + """ + + info: PaymentIdentifierInfo + schema_: dict[str, Any] = Field(alias="schema") + + model_config = {"extra": "allow", "populate_by_name": True} + + +# JSON Schema type for the payment-identifier extension +PaymentIdentifierSchema = dict[str, Any] diff --git a/python/x402/extensions/payment_identifier/utils.py b/python/x402/extensions/payment_identifier/utils.py new file mode 100644 index 0000000000..5e3622accc --- /dev/null +++ b/python/x402/extensions/payment_identifier/utils.py @@ -0,0 +1,58 @@ +"""Utility functions for the Payment-Identifier Extension.""" + +from __future__ import annotations + +import uuid + +from .types import PAYMENT_ID_MAX_LENGTH, PAYMENT_ID_MIN_LENGTH, PAYMENT_ID_PATTERN + + +def generate_payment_id(prefix: str = "pay_") -> str: + """Generate a unique payment identifier. + + Args: + prefix: Optional prefix for the ID (e.g., "pay_"). Defaults to "pay_". + + Returns: + A unique payment identifier string. + + Example: + ```python + # With default prefix + id = generate_payment_id() # "pay_7d5d747be160e280504c099d984bcfe0" + + # With custom prefix + id = generate_payment_id("txn_") # "txn_7d5d747be160e280504c099d984bcfe0" + + # Without prefix + id = generate_payment_id("") # "7d5d747be160e280504c099d984bcfe0" + ``` + """ + # Generate UUID v4 without hyphens (32 hex chars) + uuid_str = uuid.uuid4().hex + return f"{prefix}{uuid_str}" + + +def is_valid_payment_id(id: str) -> bool: + """Validate that a payment ID meets the format requirements. + + Args: + id: The payment ID to validate. + + Returns: + True if the ID is valid, False otherwise. + + Example: + ```python + is_valid_payment_id("pay_7d5d747be160e280") # True (exactly 16 chars after prefix removal check) + is_valid_payment_id("abc") # False (too short) + is_valid_payment_id("pay_abc!@#") # False (invalid characters) + ``` + """ + if not isinstance(id, str): + return False + + if len(id) < PAYMENT_ID_MIN_LENGTH or len(id) > PAYMENT_ID_MAX_LENGTH: + return False + + return bool(PAYMENT_ID_PATTERN.match(id)) diff --git a/python/x402/extensions/payment_identifier/validation.py b/python/x402/extensions/payment_identifier/validation.py new file mode 100644 index 0000000000..00a7ffe08c --- /dev/null +++ b/python/x402/extensions/payment_identifier/validation.py @@ -0,0 +1,359 @@ +"""Validation and extraction utilities for the Payment-Identifier Extension.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from .types import PAYMENT_IDENTIFIER, PaymentIdentifierExtension, PaymentIdentifierInfo +from .utils import is_valid_payment_id + +if TYPE_CHECKING: + from x402.schemas.payments import PaymentPayload + + +@dataclass +class PaymentIdentifierValidationResult: + """Result of payment identifier validation. + + Attributes: + valid: Whether the payment identifier is valid. + errors: Error messages if validation failed. + """ + + valid: bool + errors: list[str] = field(default_factory=list) + + +def is_payment_identifier_extension(extension: Any) -> bool: + """Type guard to check if an object is a valid payment-identifier extension structure. + + This checks for the basic structure (info object with required boolean), + but does not validate the id format if present. + + Args: + extension: The object to check. + + Returns: + True if the object has the expected payment-identifier extension structure. + + Example: + ```python + if is_payment_identifier_extension(extensions["payment-identifier"]): + # TypeScript knows this is PaymentIdentifierExtension + print(extension.info.required) + ``` + """ + if not extension or not isinstance(extension, dict | PaymentIdentifierExtension): + return False + + if isinstance(extension, PaymentIdentifierExtension): + return True + + ext = extension + if not isinstance(ext, dict): + return False + + info = ext.get("info") + if not info or not isinstance(info, dict | PaymentIdentifierInfo): + return False + + if isinstance(info, PaymentIdentifierInfo): + return True + + info_dict = info + # Must have required boolean + if not isinstance(info_dict.get("required"), bool): + return False + + return True + + +def validate_payment_identifier(extension: Any) -> PaymentIdentifierValidationResult: + """Validate a payment-identifier extension object. + + Checks both the structure (using JSON Schema) and the ID format. + + Args: + extension: The extension object to validate. + + Returns: + Validation result with errors if invalid. + + Example: + ```python + result = validate_payment_identifier( + payment_payload.extensions.get("payment-identifier") + ) + if not result.valid: + print("Invalid payment identifier:", result.errors) + ``` + """ + if not extension or not isinstance(extension, dict | PaymentIdentifierExtension): + return PaymentIdentifierValidationResult( + valid=False, + errors=["Extension must be an object"], + ) + + # Convert to dict if it's a Pydantic model + if isinstance(extension, PaymentIdentifierExtension): + ext_dict = extension.model_dump(by_alias=True) + else: + ext_dict = extension + + # Check info exists + info = ext_dict.get("info") + if not info or not isinstance(info, dict | PaymentIdentifierInfo): + return PaymentIdentifierValidationResult( + valid=False, + errors=["Extension must have an 'info' property"], + ) + + # Convert info to dict if it's a Pydantic model + if isinstance(info, PaymentIdentifierInfo): + info_dict = info.model_dump(by_alias=True) + else: + info_dict = info + + # Check required field exists and is a boolean + if not isinstance(info_dict.get("required"), bool): + return PaymentIdentifierValidationResult( + valid=False, + errors=["Extension info must have a 'required' boolean property"], + ) + + # Check id exists and is a string (if provided) + id_value = info_dict.get("id") + if id_value is not None and not isinstance(id_value, str): + return PaymentIdentifierValidationResult( + valid=False, + errors=["Extension info 'id' must be a string if provided"], + ) + + # Validate ID format if provided + if id_value is not None and not is_valid_payment_id(id_value): + return PaymentIdentifierValidationResult( + valid=False, + errors=[ + "Invalid payment ID format. ID must be 16-128 characters and contain only alphanumeric characters, hyphens, and underscores." + ], + ) + + # If schema is provided, validate against it + schema = ext_dict.get("schema") + if schema: + # Lazy import jsonschema only when schema validation is needed + try: + import jsonschema + except ImportError: + return PaymentIdentifierValidationResult( + valid=False, + errors=[ + "Schema validation requires jsonschema. Install with: pip install x402[extensions]" + ], + ) + + try: + jsonschema.validate(instance=info_dict, schema=schema) + except jsonschema.ValidationError as e: + path = "/".join(str(p) for p in e.absolute_path) if e.absolute_path else "(root)" + return PaymentIdentifierValidationResult( + valid=False, + errors=[f"{path}: {e.message}"], + ) + except Exception as e: + return PaymentIdentifierValidationResult( + valid=False, + errors=[f"Schema validation failed: {e!s}"], + ) + + return PaymentIdentifierValidationResult(valid=True) + + +def extract_payment_identifier( + payment_payload: PaymentPayload, validate: bool = True +) -> str | None: + """Extract the payment identifier from a PaymentPayload. + + Args: + payment_payload: The payment payload to extract from. + validate: Whether to validate the ID before returning (default: True). + + Returns: + The payment ID string, or None if not present or invalid. + + Example: + ```python + id = extract_payment_identifier(payment_payload) + if id: + # Use for idempotency lookup + cached = await idempotency_store.get(id) + ``` + """ + if not payment_payload.extensions: + return None + + extension = payment_payload.extensions.get(PAYMENT_IDENTIFIER) + + if not extension: + return None + + # Convert to dict if it's a Pydantic model + if isinstance(extension, PaymentIdentifierExtension): + ext_dict = extension.model_dump(by_alias=True) + elif isinstance(extension, dict): + ext_dict = extension + else: + return None + + info = ext_dict.get("info") + if not info: + return None + + # Convert info to dict if it's a Pydantic model + if isinstance(info, PaymentIdentifierInfo): + info_dict = info.model_dump(by_alias=True) + elif isinstance(info, dict): + info_dict = info + else: + return None + + id_value = info_dict.get("id") + if not isinstance(id_value, str): + return None + + if validate and not is_valid_payment_id(id_value): + return None + + return id_value + + +def extract_and_validate_payment_identifier( + payment_payload: PaymentPayload, +) -> tuple[str | None, PaymentIdentifierValidationResult]: + """Extract and validate the payment identifier from a PaymentPayload. + + Args: + payment_payload: The payment payload to extract from. + + Returns: + Tuple of (id, validation_result) where id is the payment ID or None, + and validation_result indicates if validation passed. + + Example: + ```python + id, validation = extract_and_validate_payment_identifier(payment_payload) + if not validation.valid: + return res.status(400).json({"error": validation.errors}) + if id: + # Use for idempotency + pass + ``` + """ + if not payment_payload.extensions: + return None, PaymentIdentifierValidationResult(valid=True) + + extension = payment_payload.extensions.get(PAYMENT_IDENTIFIER) + + if not extension: + return None, PaymentIdentifierValidationResult(valid=True) + + validation = validate_payment_identifier(extension) + + if not validation.valid: + return None, validation + + # Extract the ID + id_value = extract_payment_identifier(payment_payload, validate=False) + return id_value, PaymentIdentifierValidationResult(valid=True) + + +def has_payment_identifier(payment_payload: PaymentPayload) -> bool: + """Check if a PaymentPayload contains a payment-identifier extension. + + Args: + payment_payload: The payment payload to check. + + Returns: + True if the extension is present. + """ + return bool(payment_payload.extensions and PAYMENT_IDENTIFIER in payment_payload.extensions) + + +def is_payment_identifier_required(extension: Any) -> bool: + """Check if the server requires a payment identifier based on the extension info. + + Args: + extension: The payment-identifier extension from PaymentRequired or PaymentPayload. + + Returns: + True if the server requires a payment identifier. + """ + if not extension or not isinstance(extension, dict | PaymentIdentifierExtension): + return False + + # Convert to dict if it's a Pydantic model + if isinstance(extension, PaymentIdentifierExtension): + ext_dict = extension.model_dump(by_alias=True) + else: + ext_dict = extension + + info = ext_dict.get("info") + if not info: + return False + + # Convert info to dict if it's a Pydantic model + if isinstance(info, PaymentIdentifierInfo): + info_dict = info.model_dump(by_alias=True) + elif isinstance(info, dict): + info_dict = info + else: + return False + + return info_dict.get("required") is True + + +def validate_payment_identifier_requirement( + payment_payload: PaymentPayload, server_required: bool +) -> PaymentIdentifierValidationResult: + """Validate that a payment identifier is provided when required. + + Use this to check if a client's PaymentPayload satisfies the server's requirement. + + Args: + payment_payload: The client's payment payload. + server_required: Whether the server requires a payment identifier (from PaymentRequired). + + Returns: + Validation result - invalid if required but not provided. + + Example: + ```python + server_extension = payment_required.extensions.get("payment-identifier") + server_required = is_payment_identifier_required(server_extension) + result = validate_payment_identifier_requirement(payment_payload, server_required) + if not result.valid: + return res.status(400).json({"error": result.errors}) + ``` + """ + if not server_required: + return PaymentIdentifierValidationResult(valid=True) + + id_value = extract_payment_identifier(payment_payload, validate=False) + + if not id_value: + return PaymentIdentifierValidationResult( + valid=False, + errors=["Server requires a payment identifier but none was provided"], + ) + + # Validate the ID format + if not is_valid_payment_id(id_value): + return PaymentIdentifierValidationResult( + valid=False, + errors=[ + "Invalid payment ID format. ID must be 16-128 characters and contain only alphanumeric characters, hyphens, and underscores." + ], + ) + + return PaymentIdentifierValidationResult(valid=True) diff --git a/python/x402/mechanisms/evm/utils.py b/python/x402/mechanisms/evm/utils.py index aab2fbd44d..34092fed08 100644 --- a/python/x402/mechanisms/evm/utils.py +++ b/python/x402/mechanisms/evm/utils.py @@ -281,7 +281,7 @@ def parse_money_to_decimal(money: str | float | int) -> float: Returns: Decimal amount as float. """ - if isinstance(money, (int, float)): + if isinstance(money, int | float): return float(money) # Clean string diff --git a/python/x402/mechanisms/svm/utils.py b/python/x402/mechanisms/svm/utils.py index ff2979932b..7a7df7ffc1 100644 --- a/python/x402/mechanisms/svm/utils.py +++ b/python/x402/mechanisms/svm/utils.py @@ -201,7 +201,7 @@ def parse_money_to_decimal(money: str | float | int) -> float: Returns: Decimal amount as float. """ - if isinstance(money, (int, float)): + if isinstance(money, int | float): return float(money) # Clean string diff --git a/python/x402/tests/mocks/cash.py b/python/x402/tests/mocks/cash.py index d1827dbcdb..398aaafc4e 100644 --- a/python/x402/tests/mocks/cash.py +++ b/python/x402/tests/mocks/cash.py @@ -233,7 +233,7 @@ def parse_price(self, price: Price, network: Network) -> AssetAmount: ) # Handle numeric prices - if isinstance(price, (int, float)): + if isinstance(price, int | float): return AssetAmount( amount=str(price), asset="USD", diff --git a/python/x402/tests/unit/extensions/payment_identifier/__init__.py b/python/x402/tests/unit/extensions/payment_identifier/__init__.py new file mode 100644 index 0000000000..242ccbd539 --- /dev/null +++ b/python/x402/tests/unit/extensions/payment_identifier/__init__.py @@ -0,0 +1 @@ +"""Tests for Payment-Identifier extension.""" diff --git a/python/x402/tests/unit/extensions/payment_identifier/test_client.py b/python/x402/tests/unit/extensions/payment_identifier/test_client.py new file mode 100644 index 0000000000..2a4c6ccff8 --- /dev/null +++ b/python/x402/tests/unit/extensions/payment_identifier/test_client.py @@ -0,0 +1,88 @@ +"""Tests for Payment-Identifier client utilities.""" + +import pytest + +from x402.extensions.payment_identifier import ( + PAYMENT_IDENTIFIER, + append_payment_identifier_to_extensions, + declare_payment_identifier_extension, +) + + +class TestAppendPaymentIdentifierToExtensions: + """Tests for append_payment_identifier_to_extensions function.""" + + def test_append_auto_generated_id(self) -> None: + """Test appending auto-generated ID when extension exists.""" + extensions = {PAYMENT_IDENTIFIER: declare_payment_identifier_extension()} + result = append_payment_identifier_to_extensions(extensions) + + assert result is extensions # Same reference + ext = extensions[PAYMENT_IDENTIFIER] + assert "info" in ext + assert "id" in ext["info"] + assert ext["info"]["id"].startswith("pay_") + assert len(ext["info"]["id"]) == 4 + 32 # "pay_" + 32 hex chars + assert ext["info"]["required"] is False + + def test_append_custom_id(self) -> None: + """Test appending custom ID when extension exists.""" + custom_id = "custom_id_1234567890" + extensions = {PAYMENT_IDENTIFIER: declare_payment_identifier_extension()} + append_payment_identifier_to_extensions(extensions, custom_id) + + ext = extensions[PAYMENT_IDENTIFIER] + assert ext["info"]["id"] == custom_id + + def test_preserve_required_flag(self) -> None: + """Test that required flag is preserved from server declaration.""" + extensions = {PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required=True)} + append_payment_identifier_to_extensions(extensions) + + ext = extensions[PAYMENT_IDENTIFIER] + assert ext["info"]["required"] is True + assert "id" in ext["info"] + + def test_no_modification_when_extension_not_present(self) -> None: + """Test that extensions are not modified when payment-identifier is not present.""" + extensions = {"other": {"foo": "bar"}} + result = append_payment_identifier_to_extensions(extensions) + + assert result is extensions + assert PAYMENT_IDENTIFIER not in extensions + assert extensions["other"] == {"foo": "bar"} + + def test_no_modification_when_extension_invalid(self) -> None: + """Test that extensions are not modified when extension structure is invalid.""" + extensions = {PAYMENT_IDENTIFIER: {"schema": {}}} + result = append_payment_identifier_to_extensions(extensions) + + assert result is extensions + ext = extensions[PAYMENT_IDENTIFIER] + assert "info" not in ext or "id" not in ext.get("info", {}) + + def test_raises_error_for_invalid_custom_id(self) -> None: + """Test that ValueError is raised for invalid custom ID.""" + extensions = {PAYMENT_IDENTIFIER: declare_payment_identifier_extension()} + + with pytest.raises(ValueError, match="Invalid payment ID"): + append_payment_identifier_to_extensions(extensions, "short") + + with pytest.raises(ValueError, match="Invalid payment ID"): + append_payment_identifier_to_extensions(extensions, "invalid!@#$%^&") + + def test_no_error_when_extension_not_present_and_custom_id_provided(self) -> None: + """Test that no error when extension doesn't exist and custom ID provided.""" + extensions = {"other": {}} + result = append_payment_identifier_to_extensions(extensions, "valid_id_12345678") + assert result is extensions + assert PAYMENT_IDENTIFIER not in extensions + + def test_overwrites_existing_id(self) -> None: + """Test that calling multiple times overwrites existing ID.""" + extensions = {PAYMENT_IDENTIFIER: declare_payment_identifier_extension()} + append_payment_identifier_to_extensions(extensions, "first_id_12345678") + assert extensions[PAYMENT_IDENTIFIER]["info"]["id"] == "first_id_12345678" + + append_payment_identifier_to_extensions(extensions, "second_id_12345678") + assert extensions[PAYMENT_IDENTIFIER]["info"]["id"] == "second_id_12345678" diff --git a/python/x402/tests/unit/extensions/payment_identifier/test_server.py b/python/x402/tests/unit/extensions/payment_identifier/test_server.py new file mode 100644 index 0000000000..d941f11da8 --- /dev/null +++ b/python/x402/tests/unit/extensions/payment_identifier/test_server.py @@ -0,0 +1,76 @@ +"""Tests for Payment-Identifier server extension.""" + +from x402.extensions.payment_identifier import ( + PAYMENT_IDENTIFIER, + declare_payment_identifier_extension, + payment_identifier_resource_server_extension, +) + + +class TestDeclarePaymentIdentifierExtension: + """Tests for declare_payment_identifier_extension function.""" + + def test_declare_with_default_required(self) -> None: + """Test declaring extension with default required=False.""" + declaration = declare_payment_identifier_extension() + assert declaration["info"]["required"] is False + assert "schema" in declaration + + def test_declare_with_required_true(self) -> None: + """Test declaring extension with required=True.""" + declaration = declare_payment_identifier_extension(required=True) + assert declaration["info"]["required"] is True + assert "schema" in declaration + + def test_declare_includes_schema(self) -> None: + """Test that declaration includes schema.""" + declaration = declare_payment_identifier_extension() + schema = declaration["schema"] + assert schema["$schema"] == "https://json-schema.org/draft/2020-12/schema" + assert schema["type"] == "object" + assert schema["properties"]["required"]["type"] == "boolean" + assert schema["properties"]["id"]["minLength"] == 16 + assert schema["properties"]["id"]["maxLength"] == 128 + + def test_declare_schema_required_field(self) -> None: + """Test that schema requires 'required' field.""" + declaration = declare_payment_identifier_extension() + schema = declaration["schema"] + assert "required" in schema["required"] + + +class TestPaymentIdentifierResourceServerExtension: + """Tests for PaymentIdentifierResourceServerExtension.""" + + def test_extension_key(self) -> None: + """Test extension key is correct.""" + assert payment_identifier_resource_server_extension.key == PAYMENT_IDENTIFIER + + def test_enrich_declaration_returns_unchanged(self) -> None: + """Test that enrich_declaration returns declaration unchanged.""" + declaration = declare_payment_identifier_extension() + context = {"method": "GET"} + + enriched = payment_identifier_resource_server_extension.enrich_declaration( + declaration, context + ) + + # Should return unchanged since payment-identifier doesn't need enrichment + assert enriched == declaration + + def test_enrich_with_none_context(self) -> None: + """Test enriching with None context.""" + declaration = declare_payment_identifier_extension() + enriched = payment_identifier_resource_server_extension.enrich_declaration( + declaration, None + ) + assert enriched == declaration + + def test_enrich_preserves_structure(self) -> None: + """Test that enrichment preserves declaration structure.""" + declaration = declare_payment_identifier_extension(required=True) + enriched = payment_identifier_resource_server_extension.enrich_declaration( + declaration, {"any": "context"} + ) + assert enriched["info"]["required"] is True + assert "schema" in enriched diff --git a/python/x402/tests/unit/extensions/payment_identifier/test_types.py b/python/x402/tests/unit/extensions/payment_identifier/test_types.py new file mode 100644 index 0000000000..7025780546 --- /dev/null +++ b/python/x402/tests/unit/extensions/payment_identifier/test_types.py @@ -0,0 +1,83 @@ +"""Tests for Payment-Identifier extension types.""" + +from x402.extensions.payment_identifier.types import ( + PAYMENT_ID_MAX_LENGTH, + PAYMENT_ID_MIN_LENGTH, + PAYMENT_ID_PATTERN, + PAYMENT_IDENTIFIER, + PaymentIdentifierExtension, + PaymentIdentifierInfo, +) + + +class TestConstants: + """Test extension constants.""" + + def test_payment_identifier_constant(self) -> None: + """Test PAYMENT_IDENTIFIER constant value.""" + assert PAYMENT_IDENTIFIER == "payment-identifier" + + def test_length_constants(self) -> None: + """Test length constants.""" + assert PAYMENT_ID_MIN_LENGTH == 16 + assert PAYMENT_ID_MAX_LENGTH == 128 + + def test_pattern_constant(self) -> None: + """Test PAYMENT_ID_PATTERN constant.""" + assert PAYMENT_ID_PATTERN is not None + assert PAYMENT_ID_PATTERN.match("valid_id_123") is not None + assert PAYMENT_ID_PATTERN.match("invalid!@#") is None + + +class TestPaymentIdentifierInfo: + """Test PaymentIdentifierInfo model.""" + + def test_info_with_required_only(self) -> None: + """Test PaymentIdentifierInfo with required flag only.""" + info = PaymentIdentifierInfo(required=False) + assert info.required is False + assert info.id is None + + def test_info_with_id(self) -> None: + """Test PaymentIdentifierInfo with id.""" + info = PaymentIdentifierInfo(required=False, id="pay_1234567890123456") + assert info.required is False + assert info.id == "pay_1234567890123456" + + def test_info_required_true(self) -> None: + """Test PaymentIdentifierInfo with required=True.""" + info = PaymentIdentifierInfo(required=True) + assert info.required is True + + def test_info_from_dict(self) -> None: + """Test PaymentIdentifierInfo from dictionary.""" + info = PaymentIdentifierInfo.model_validate({"required": False, "id": "test_id_12345678"}) + assert info.required is False + assert info.id == "test_id_12345678" + + +class TestPaymentIdentifierExtension: + """Test PaymentIdentifierExtension model.""" + + def test_extension_with_info_and_schema(self) -> None: + """Test PaymentIdentifierExtension with info and schema.""" + info = PaymentIdentifierInfo(required=False) + schema = {"type": "object", "properties": {}} + ext = PaymentIdentifierExtension(info=info, schema=schema) + assert ext.info.required is False + assert ext.schema_ == schema + + def test_extension_with_id(self) -> None: + """Test PaymentIdentifierExtension with id in info.""" + info = PaymentIdentifierInfo(required=False, id="pay_1234567890123456") + schema = {"type": "object"} + ext = PaymentIdentifierExtension(info=info, schema=schema) + assert ext.info.id == "pay_1234567890123456" + + def test_extension_schema_alias(self) -> None: + """Test PaymentIdentifierExtension with camelCase schema alias.""" + schema = {"type": "object"} + ext = PaymentIdentifierExtension.model_validate( + {"info": {"required": False}, "schema": schema} + ) + assert ext.schema_ == schema diff --git a/python/x402/tests/unit/extensions/payment_identifier/test_utils.py b/python/x402/tests/unit/extensions/payment_identifier/test_utils.py new file mode 100644 index 0000000000..ae70069dd7 --- /dev/null +++ b/python/x402/tests/unit/extensions/payment_identifier/test_utils.py @@ -0,0 +1,81 @@ +"""Tests for Payment-Identifier utility functions.""" + +from x402.extensions.payment_identifier.utils import generate_payment_id, is_valid_payment_id + + +class TestGeneratePaymentId: + """Tests for generate_payment_id function.""" + + def test_generate_with_default_prefix(self) -> None: + """Test generating ID with default prefix.""" + id = generate_payment_id() + assert id.startswith("pay_") + assert len(id) == 4 + 32 # "pay_" + 32 hex chars + assert is_valid_payment_id(id) + + def test_generate_with_custom_prefix(self) -> None: + """Test generating ID with custom prefix.""" + id = generate_payment_id("txn_") + assert id.startswith("txn_") + assert len(id) == 4 + 32 # "txn_" + 32 hex chars + assert is_valid_payment_id(id) + + def test_generate_without_prefix(self) -> None: + """Test generating ID without prefix.""" + id = generate_payment_id("") + assert not id.startswith("pay_") + assert len(id) == 32 # 32 hex chars + assert is_valid_payment_id(id) + + def test_generate_unique_ids(self) -> None: + """Test that generated IDs are unique.""" + ids = {generate_payment_id() for _ in range(100)} + assert len(ids) == 100 # All unique + + def test_generated_ids_pass_validation(self) -> None: + """Test that generated IDs pass validation.""" + for _ in range(10): + id = generate_payment_id() + assert is_valid_payment_id(id) + + +class TestIsValidPaymentId: + """Tests for is_valid_payment_id function.""" + + def test_valid_ids(self) -> None: + """Test valid payment IDs.""" + assert is_valid_payment_id("pay_7d5d747be160e280") is True + assert is_valid_payment_id("1234567890123456") is True # Exactly 16 chars + assert is_valid_payment_id("abcdefghijklmnop") is True + assert is_valid_payment_id("test_with-hyphens") is True + assert is_valid_payment_id("test_with_underscores") is True + + def test_too_short(self) -> None: + """Test IDs that are too short.""" + assert is_valid_payment_id("abc") is False + assert is_valid_payment_id("123456789012345") is False # 15 chars + + def test_too_long(self) -> None: + """Test IDs that are too long.""" + long_id = "a" * 129 + assert is_valid_payment_id(long_id) is False + + def test_boundary_lengths(self) -> None: + """Test IDs at boundary lengths.""" + min_id = "a" * 16 + max_id = "a" * 128 + assert is_valid_payment_id(min_id) is True + assert is_valid_payment_id(max_id) is True + + def test_invalid_characters(self) -> None: + """Test IDs with invalid characters.""" + assert is_valid_payment_id("pay_abc!@#$%^&*()") is False + assert is_valid_payment_id("pay_abc def ghij") is False # spaces + assert is_valid_payment_id("pay_abc.def.ghij") is False # dots + + def test_non_string_values(self) -> None: + """Test non-string values.""" + assert is_valid_payment_id(None) is False # type: ignore[arg-type] + assert is_valid_payment_id(123) is False # type: ignore[arg-type] + assert is_valid_payment_id([]) is False # type: ignore[arg-type] + assert is_valid_payment_id({}) is False # type: ignore[arg-type] diff --git a/python/x402/tests/unit/extensions/payment_identifier/test_validation.py b/python/x402/tests/unit/extensions/payment_identifier/test_validation.py new file mode 100644 index 0000000000..cbfbc91eac --- /dev/null +++ b/python/x402/tests/unit/extensions/payment_identifier/test_validation.py @@ -0,0 +1,474 @@ +"""Tests for Payment-Identifier validation functions.""" + +from x402.extensions.payment_identifier import ( + PAYMENT_IDENTIFIER, + declare_payment_identifier_extension, + extract_and_validate_payment_identifier, + extract_payment_identifier, + has_payment_identifier, + is_payment_identifier_extension, + is_payment_identifier_required, + validate_payment_identifier, + validate_payment_identifier_requirement, +) +from x402.extensions.payment_identifier.utils import generate_payment_id +from x402.extensions.payment_identifier.validation import PaymentIdentifierValidationResult +from x402.schemas.payments import PaymentPayload, PaymentRequirements + + +def create_extension_with_id(id: str | None = None, required: bool = False) -> dict: + """Helper to create an extension with ID appended.""" + extensions = {PAYMENT_IDENTIFIER: declare_payment_identifier_extension(required)} + if id: + from x402.extensions.payment_identifier.client import ( + append_payment_identifier_to_extensions, + ) + + try: + append_payment_identifier_to_extensions(extensions, id) + except ValueError: + # If ID is invalid, create extension directly without validation + # (for testing invalid ID scenarios) + ext = extensions[PAYMENT_IDENTIFIER] + if isinstance(ext, dict): + ext["info"]["id"] = id + else: + ext_dict = ( + ext.model_dump(by_alias=True) if hasattr(ext, "model_dump") else dict(ext) + ) + ext_dict["info"]["id"] = id + extensions[PAYMENT_IDENTIFIER] = ext_dict + else: + from x402.extensions.payment_identifier.client import ( + append_payment_identifier_to_extensions, + ) + + append_payment_identifier_to_extensions(extensions) + return extensions[PAYMENT_IDENTIFIER] + + +class TestIsPaymentIdentifierExtension: + """Tests for is_payment_identifier_extension function.""" + + def test_valid_extension(self) -> None: + """Test valid extension structure.""" + ext = declare_payment_identifier_extension() + assert is_payment_identifier_extension(ext) is True + + def test_invalid_extension(self) -> None: + """Test invalid extension structures.""" + assert is_payment_identifier_extension({}) is False + assert is_payment_identifier_extension(None) is False + assert is_payment_identifier_extension({"schema": {}}) is False + assert is_payment_identifier_extension({"info": {}}) is False + + def test_extension_without_required(self) -> None: + """Test extension without required field.""" + assert is_payment_identifier_extension({"info": {"id": "test"}}) is False + + +class TestValidatePaymentIdentifier: + """Tests for validate_payment_identifier function.""" + + def test_valid_extension(self) -> None: + """Test validating a correct extension.""" + extension = create_extension_with_id() + result = validate_payment_identifier(extension) + assert result.valid is True + assert len(result.errors) == 0 + + def test_reject_non_object(self) -> None: + """Test rejecting non-object extension.""" + assert validate_payment_identifier(None).valid is False + assert validate_payment_identifier("string").valid is False + assert validate_payment_identifier(123).valid is False + + def test_reject_extension_without_info(self) -> None: + """Test rejecting extension without info.""" + result = validate_payment_identifier({"schema": {}}) + assert result.valid is False + assert any("info" in error.lower() for error in result.errors) + + def test_reject_extension_without_required(self) -> None: + """Test rejecting extension without required in info.""" + result = validate_payment_identifier( + {"info": {"id": "pay_valid_id_12345678"}, "schema": {}} + ) + assert result.valid is False + assert any("required" in error.lower() for error in result.errors) + + def test_validate_extension_without_id(self) -> None: + """Test validating extension with required but no id.""" + result = validate_payment_identifier( + { + "info": {"required": False}, + "schema": declare_payment_identifier_extension()["schema"], + } + ) + assert result.valid is True + + def test_reject_invalid_id_format(self) -> None: + """Test rejecting extension with invalid id format.""" + result = validate_payment_identifier( + { + "info": {"required": False, "id": "short"}, + "schema": declare_payment_identifier_extension()["schema"], + } + ) + assert result.valid is False + + def test_reject_non_string_id(self) -> None: + """Test rejecting extension with non-string id.""" + result = validate_payment_identifier( + { + "info": {"required": False, "id": 123}, + "schema": declare_payment_identifier_extension()["schema"], + } + ) + assert result.valid is False + assert any("string" in error.lower() for error in result.errors) + + def test_validate_extension_with_valid_schema(self) -> None: + """Test validating extension with valid schema.""" + result = validate_payment_identifier( + { + "info": {"required": False, "id": "valid_id_12345678"}, + "schema": declare_payment_identifier_extension()["schema"], + } + ) + assert result.valid is True + + +class TestExtractPaymentIdentifier: + """Tests for extract_payment_identifier function.""" + + def test_extract_from_payload(self) -> None: + """Test extracting ID from PaymentPayload.""" + payment_id = generate_payment_id() + ext = create_extension_with_id(payment_id) + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + extracted_id = extract_payment_identifier(payload) + assert extracted_id == payment_id + + def test_extract_none_when_not_present(self) -> None: + """Test extracting when extension not present.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={}, + ) + + assert extract_payment_identifier(payload) is None + + def test_extract_none_when_no_extensions(self) -> None: + """Test extracting when extensions field is None.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions=None, + ) + + assert extract_payment_identifier(payload) is None + + def test_extract_with_validation_false(self) -> None: + """Test extracting with validate=False.""" + ext = create_extension_with_id("invalid_short") + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + # Should return None when validate=True (default) + assert extract_payment_identifier(payload, validate=True) is None + # Should return ID when validate=False + assert extract_payment_identifier(payload, validate=False) == "invalid_short" + + +class TestExtractAndValidatePaymentIdentifier: + """Tests for extract_and_validate_payment_identifier function.""" + + def test_extract_and_validate_valid(self) -> None: + """Test extracting and validating valid extension.""" + payment_id = generate_payment_id() + ext = create_extension_with_id(payment_id) + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + id_value, validation = extract_and_validate_payment_identifier(payload) + assert id_value == payment_id + assert validation.valid is True + + def test_extract_and_validate_invalid(self) -> None: + """Test extracting and validating invalid extension.""" + ext = {"info": {"required": False, "id": "short"}, "schema": {}} + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + id_value, validation = extract_and_validate_payment_identifier(payload) + assert id_value is None + assert validation.valid is False + + def test_extract_and_validate_no_extension(self) -> None: + """Test extracting and validating when no extension.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={}, + ) + + id_value, validation = extract_and_validate_payment_identifier(payload) + assert id_value is None + assert validation.valid is True + + +class TestHasPaymentIdentifier: + """Tests for has_payment_identifier function.""" + + def test_has_extension(self) -> None: + """Test has_payment_identifier returns True when present.""" + ext = declare_payment_identifier_extension() + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + assert has_payment_identifier(payload) is True + + def test_no_extension(self) -> None: + """Test has_payment_identifier returns False when not present.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={}, + ) + + assert has_payment_identifier(payload) is False + + def test_no_extensions_field(self) -> None: + """Test has_payment_identifier when extensions is None.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions=None, + ) + + assert has_payment_identifier(payload) is False + + +class TestIsPaymentIdentifierRequired: + """Tests for is_payment_identifier_required function.""" + + def test_required_false(self) -> None: + """Test is_payment_identifier_required with required=False.""" + ext = declare_payment_identifier_extension(required=False) + assert is_payment_identifier_required(ext) is False + + def test_required_true(self) -> None: + """Test is_payment_identifier_required with required=True.""" + ext = declare_payment_identifier_extension(required=True) + assert is_payment_identifier_required(ext) is True + + def test_invalid_extension(self) -> None: + """Test is_payment_identifier_required with invalid extension.""" + assert is_payment_identifier_required({}) is False + assert is_payment_identifier_required(None) is False + + +class TestValidatePaymentIdentifierRequirement: + """Tests for validate_payment_identifier_requirement function.""" + + def test_not_required(self) -> None: + """Test validation when server doesn't require identifier.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={}, + ) + + result = validate_payment_identifier_requirement(payload, server_required=False) + assert result.valid is True + + def test_required_and_provided(self) -> None: + """Test validation when required and provided.""" + payment_id = generate_payment_id() + ext = create_extension_with_id(payment_id) + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + result = validate_payment_identifier_requirement(payload, server_required=True) + assert result.valid is True + + def test_required_but_not_provided(self) -> None: + """Test validation when required but not provided.""" + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={}, + ) + + result = validate_payment_identifier_requirement(payload, server_required=True) + assert result.valid is False + assert len(result.errors) > 0 + # Check that error mentions requirement or identifier + assert any( + "required" in error.lower() or "identifier" in error.lower() for error in result.errors + ) + + def test_required_but_invalid_id(self) -> None: + """Test validation when required but ID is invalid.""" + ext = create_extension_with_id("short") + payload = PaymentPayload( + x402_version=2, + payload={}, + accepted=PaymentRequirements( + scheme="exact", + network="eip155:8453", + asset="0x123", + amount="1000", + pay_to="0x456", + max_timeout_seconds=300, + ), + extensions={PAYMENT_IDENTIFIER: ext}, + ) + + result = validate_payment_identifier_requirement(payload, server_required=True) + assert result.valid is False + assert any("format" in error.lower() for error in result.errors) + + +class TestPaymentIdentifierValidationResult: + """Tests for PaymentIdentifierValidationResult dataclass.""" + + def test_valid_result(self) -> None: + """Test valid validation result.""" + result = PaymentIdentifierValidationResult(valid=True) + assert result.valid is True + assert result.errors == [] + + def test_invalid_result_with_errors(self) -> None: + """Test invalid validation result with errors.""" + result = PaymentIdentifierValidationResult(valid=False, errors=["error1", "error2"]) + assert result.valid is False + assert len(result.errors) == 2 + assert "error1" in result.errors + assert "error2" in result.errors diff --git a/python/x402/tests/unit/extensions/test_exports.py b/python/x402/tests/unit/extensions/test_exports.py new file mode 100644 index 0000000000..842e85a4e0 --- /dev/null +++ b/python/x402/tests/unit/extensions/test_exports.py @@ -0,0 +1,67 @@ +"""Tests to verify that all __all__ exports from x402.extensions are importable. + +This test ensures that the public API contract is maintained - everything +listed in __all__ must be importable from the top-level x402.extensions module. +""" + +import pytest + +# Import the module to get access to __all__ +from x402 import extensions + + +class TestExtensionExports: + """Test that all __all__ exports are importable.""" + + def test_all_exports_are_importable(self) -> None: + """Test that every item in __all__ can be imported from x402.extensions.""" + # Get the list of exported names + exported_names = extensions.__all__ + + # Try to import each exported name + missing_exports = [] + for name in exported_names: + try: + # Try to get the attribute from the module + attr = getattr(extensions, name) + if attr is None: + missing_exports.append(f"{name} (is None)") + except AttributeError as e: + missing_exports.append(f"{name} ({e})") + + if missing_exports: + pytest.fail( + "The following exports from __all__ are not importable:\n" + + "\n".join(f" - {name}" for name in missing_exports) + ) + + def test_validation_result_imports(self) -> None: + """Test that ValidationResult and its aliases work correctly.""" + from x402.extensions import ( + BazaarValidationResult, + PaymentIdentifierValidationResult, + ValidationResult, + ) + + # ValidationResult should be an alias for BazaarValidationResult + assert ValidationResult is BazaarValidationResult + + # All three should be different classes + assert ValidationResult is not PaymentIdentifierValidationResult + assert BazaarValidationResult is not PaymentIdentifierValidationResult + + def test_import_all_star(self) -> None: + """Test that 'from x402.extensions import *' works correctly.""" + # Create a new namespace + namespace = {} + # Execute import * in that namespace + exec("from x402.extensions import *", namespace) + + # Verify all __all__ items are in the namespace + exported_names = extensions.__all__ + missing = [name for name in exported_names if name not in namespace] + if missing: + pytest.fail( + "The following exports are missing from 'import *':\n" + + "\n".join(f" - {name}" for name in missing) + )