scripts/x402_api/main.py

#!/usr/bin/env python3
"""contract-guard x402 API — pre-interaction risk check for an EVM contract/token.

The on-chain-data category is the busiest on the x402 discovery layer, but it's all
raw RPC proxies. contract-guard adds a security verdict on top: upgradeable-proxy
detection (EIP-1967 + legacy), EIP-7702 delegated-EOA detection, ERC20 metadata,
and a risk score — what an agent actually needs before approving/swapping a token.

Endpoints:
  GET  /        — Service info (free)
  GET  /health  — Health check (free)
  POST /check   — Risk check for an address ($0.005 USDC, x402) — body {address, chain}

Payment: USDC on Base mainnet (eip155:8453), official x402 v2 SDK, Dexter facilitator.
"""
import os
import sys
from datetime import datetime
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "engine"))
from server import analyze, RPCS  # noqa: E402

WALLET = os.environ.get("BASE_WALLET_ADDRESS", "0x2B60E27BE6BF979DE4Ed769838A8ddbB8AFe7392")
BASE_MAINNET = "eip155:8453"
FACILITATOR_URL = os.environ.get("FACILITATOR_URL", "https://x402.dexter.cash")

app = FastAPI(
    title="contract-guard API",
    description="Pre-interaction risk signals for an EVM contract/token (proxy, EIP-7702, ERC20 metadata). x402 v2 on Base.",
    version="1.0.0",
)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

_x402_available = False
try:
    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.server import x402ResourceServer
    from x402.extensions.bazaar import declare_discovery_extension, OutputConfig

    facilitator = HTTPFacilitatorClient(FacilitatorConfig(url=FACILITATOR_URL))
    server = x402ResourceServer(facilitator)
    server.register(BASE_MAINNET, ExactEvmServerScheme())

    def _disc(input_example, input_schema, output_example):
        ext = declare_discovery_extension(
            input=input_example, input_schema=input_schema,
            body_type="json", output=OutputConfig(example=output_example),
        )
        ext["bazaar"]["info"]["input"]["method"] = "POST"
        return ext

    routes = {
        "POST /check": RouteConfig(
            accepts=[PaymentOption(scheme="exact", pay_to=WALLET, price="$0.005", network=BASE_MAINNET)],
            mime_type="application/json",
            description="Risk check for an EVM contract/token address",
            extensions=_disc(
                {"address": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", "chain": "base"},
                {"properties": {
                    "address": {"type": "string", "description": "EVM contract/token address (0x...)"},
                    "chain": {"type": "string", "enum": ["base", "ethereum"], "description": "Chain (default base)"},
                }, "required": ["address"]},
                {"address": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", "chain": "base",
                 "is_contract": True, "is_proxy": True, "risk_level": "HIGH", "risk_score": 40,
                 "token": {"name": "USD Coin", "symbol": "USDC", "decimals": 6}, "flags": ["..."]},
            ),
        ),
    }
    app.add_middleware(PaymentMiddlewareASGI, routes=routes, server=server)
    _x402_available = True
except Exception as e:  # pragma: no cover
    print(f"  x402 v2 init warning: {type(e).__name__}: {e}")


class CheckRequest(BaseModel):
    address: str
    chain: Optional[str] = "base"


@app.get("/")
async def root():
    return {
        "service": "contract-guard API",
        "version": "1.0.0",
        "description": "Pre-interaction risk signals for an EVM contract/token.",
        "checks": [
            "is-contract / EOA / self-destructed",
            "EIP-7702 delegated-EOA detection",
            "upgradeable proxy (EIP-1967 + legacy zeppelinos)",
            "ERC20 metadata (name/symbol/decimals/totalSupply)",
            "risk score + actionable flags",
        ],
        "chains": list(RPCS.keys()),
        "endpoints": {
            "GET /": "Service info (free)",
            "GET /health": "Health check (free)",
            "POST /check": "Risk check ($0.005 USDC) — body {address, chain}",
        },
        "payment": {
            "method": "x402", "x402_version": 2, "currency": "USDC",
            "network": "Base (eip155:8453)", "facilitator": FACILITATOR_URL, "wallet": WALLET,
            "x402_enabled": _x402_available,
        },
    }


@app.get("/health")
async def health():
    return {"status": "ok", "timestamp": datetime.utcnow().isoformat() + "Z", "x402_enabled": _x402_available}


@app.post("/check")
async def check(req: CheckRequest):
    addr = (req.address or "").strip()
    if not addr:
        raise HTTPException(400, "address is required")
    try:
        result = analyze(addr, req.chain or "base")
    except Exception as e:
        raise HTTPException(502, f"rpc error: {type(e).__name__}: {e}")
    if "error" in result:
        raise HTTPException(400, result["error"])
    return result


if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 8405))
    print(f"\n  contract-guard API (x402 v2) on :{port}  facilitator={FACILITATOR_URL}  wallet={WALLET}\n")
    uvicorn.run(app, host="0.0.0.0", port=port)