x402 Security Best Practices: Protecting Your Agent Payment Endpoints

You've added x402 payments to your API. Now how do you keep it secure? Here's a practical security checklist for production x402 deployments.

The Threat Model

Agents are autonomous. They don't have human oversight. Your security must handle:

• Automated attack attempts
• Receipt replay attacks
• Challenge replay attacks
• Nonce exhaustion
• Receipt forgery

1. Receipt Replay Protection

import hashlib
import time
from collections import OrderedDict

class ReceiptCache:
    LRU cache for seen receipts. Prevents replay attacks.
    def __init__(self, max_size=10000, ttl_seconds=300):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl_seconds

    def is_replay(self, receipt_str: str) -> bool:
        receipt_hash = hashlib.sha256(receipt_str.encode()).hexdigest()
        now = time.time()

        # Clean expired
        expired = [k for k, v in self.cache.items() if now - v > self.ttl]
        for k in expired:
            del self.cache[k]

        if receipt_hash in self.cache:
            return True  # REPLAY ATTACK

        self.cache[receipt_hash] = now

        # Evict oldest if over capacity
        while len(self.cache) > self.max_size:
            self.cache.popitem(last=False)

        return False

# Global instance
_receipt_cache = ReceiptCache(max_size=10000, ttl_seconds=300)

def check_replay(receipt_header: str) -> bool:
    return _receipt_cache.is_replay(receipt_header)

2. Challenge Replay Protection

class ChallengeStore:
    Store active challenges with expiry and single-use enforcement.

    def __init__(self, ttl_seconds=600):  # 10 min default
        self.challenges = {}
        self.ttl = ttl_seconds

    def create_challenge(self, path: str, price_usd: float) -> dict:
        nonce = secrets.token_urlsafe(16)
        expires_at = time.time() + self.ttl
        challenge = {
            "nonce": nonce,
            "path": path,
            "price_usd": price_usd,
            "created_at": time.time(),
            "expires_at": expires_at,
            "used": False
        }
        self.challenges[nonce] = challenge
        return challenge

    def consume_challenge(self, nonce: str) -> dict:
        challenge = self.challenges.get(nonce)
        if not challenge:
            raise ValueError("Invalid or expired challenge")

        if challenge["used"]:
            raise ValueError("Challenge already used")

        if time.time() > challenge["expires_at"]:
            raise ValueError("Challenge expired")

        challenge["used"] = True
        return challenge

    def cleanup_expired(self):
        now = time.time()
        expired = [n for n, c in self.challenges.items() if now > c["expires_at"]]
        for n in expired:
            del self.challenges[n]

3. Receipt Verification Hardening

def verify_receipt_hardened(receipt_header: str, challenge_store: ChallengeStore) -> bool:
    Production-grade receipt verification with multiple checks.

    # 1. Basic format validation
    if not receipt_header or len(receipt_header) < 100:
        return False

    # 2. Parse receipt
    try:
        receipt = parse_receipt(receipt_header)
    except Exception:
        return False

    # 3. Check replay
    if check_replay(receipt_header):
        log_security_event("REPLAY_ATTEMPT", receipt_header[:50])
        return False

    # 4. Validate challenge exists and unused
    nonce = receipt.get("nonce")
    if not nonce:
        return False

    try:
        challenge = challenge_store.consume_challenge(nonce)
    except ValueError as e:
        log_security_event("CHALLENGE_ERROR", str(e))
        return False

    # 4. Verify payment amount matches challenge
    if receipt.get("amount") != challenge["price_usd"]:
        return False

    # 5. Verify wallet matches challenge
    if receipt.get("wallet") != TREASURY_WALLET:
        return False

    # 6. Cryptographic verification (use SDK in production)
    if not verify_cryptographic_receipt(receipt):
        return False

    return True

4. Rate Limiting & Abuse Prevention

from collections import defaultdict
from time import time

class RateLimiter:
    def __init__(self):
        self.requests = defaultdict(list)
        self.limits = {
            "challenge": (10, 60),      # 10 challenges/min
            "verify": (100, 60),        # 100 verifications/min
            "free_call": (5, 86400),    # 5 free calls/day
        }

    def check_limit(self, client_ip: str, action: str) -> bool:
        now = time.time()
        window = self.limits.get(action, (100, 60))[1]
        limit = self.limits.get(action, (100, 60))[0]

        # Clean old entries
        cutoff = now - window
        self.requests[client_ip] = [t for t in self.requests[client_ip] if t > cutoff]

        if len(self.requests[client_ip]) >= limit:
            return False

        self.requests[client_ip].append(now)
        return True

5. Monitoring & Alerting

# Key metrics to alert on
SECURITY_ALERTS = {
    "receipt_replay_rate": {"threshold": 0.01, "action": "investigate"},
    "challenge_replay_rate": {"threshold": 0.05, "action": "investigate"},
    "verification_failure_rate": {"threshold": 0.05, "action": "alert"},
    "challenge_expiry_rate": {"threshold": 0.2, "action": "tune_ttl"},
    "free_tier_exhaustion_rate": {"threshold": 0.3, "action": "review_limits"},
    "receipt_verification_failures": {"threshold": 10, "window": 300, "action": "alert"},
}

def check_security_alerts(metrics: dict):
    alerts = []
    for metric, config in SECURITY_ALERTS.items():
        if metrics.get(metric, 0) > config["threshold"]:
            alerts.append({
                "metric": metric,
                "value": metrics[metric],
                "threshold": config["threshold"],
                "action": config["action"]
            })
    return alerts

6. Incident Response Playbook

Incident	Detection	Immediate Action	Investigation
Receipt replay spike	Alert on receipt_replay_rate	Block offending IPs	Check for compromised keys
Challenge replay	Alert on challenge_replay_rate	Increase challenge TTL	Check for automation
Verification failures	Alert on verification_failure_rate	Check SDK version	Check verifier service
Free tier abuse	High 429 rate	Reduce free tier limit	Check for bot nets

Security Checklist (Pre-Launch)

• [ ] Receipt replay cache deployed (5-min TTL)
• [ ] Challenge store with single-use enforcement
• [ ] Receipt verification uses SDK (not fallback)
• [ ] Rate limiting on all endpoints
• [ ] Security alerts configured
• [ ] Incident response runbook documented
• [ ] Penetration test completed
• [ ] Dependency scanning enabled
• [ ] Secrets rotation policy defined
• [ ] Audit logging for all payment events

---

SDK: pip install nano-empire-tollbooth
Live proof: https://api.nanoempireai.com/proof/summary