scripts/x402_api/main.py

#!/usr/bin/env python3
"""skill-audit x402 API — Security audit as a paid service (x402 protocol v2).

Deploy: uvicorn scripts.x402_api.main:app --host 0.0.0.0 --port $PORT
Local:  uvicorn scripts.x402_api.main:app --port 8402

Endpoints:
  GET  /          — Service info (free)
  GET  /health    — Health check (free)
  POST /audit     — Text audit $0.01/call (x402)
  POST /audit/url — URL fetch + audit $0.03/call (x402)

Payment: USDC on Base mainnet (eip155:8453) via the official x402 v2 SDK.
Facilitator default = Dexter (https://x402.dexter.cash): zero-gate, 0% seller fee,
gas-sponsored for buyers, v2-native, and auto-lists this endpoint on the discovery
layer (Bazaar) once the first payment settles.

v2 migration note: the legacy fastapi-x402 (v0.1.x) only spoke x402 v1, which the new
discovery ecosystem (x402scan / CDP Bazaar v2) rejects ("migrate to v2 spec"). This
file uses the official `x402` SDK so the 402 challenge is valid v2 and discoverable.
The old USDC-name hack is gone: the SDK uses the correct on-chain EIP-712 domain.
"""
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

# Import scan engine from skill-audit MCP server
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "mcp_servers", "skill-audit"))
from server import scan, PATTERNS  # noqa: E402

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Config
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

WALLET = os.environ.get("BASE_WALLET_ADDRESS", "0x2B60E27BE6BF979DE4Ed769838A8ddbB8AFe7392")
BASE_MAINNET = "eip155:8453"  # CAIP-2 network id for Base mainnet
FACILITATOR_URL = os.environ.get("FACILITATOR_URL", "https://x402.dexter.cash")

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# App
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

app = FastAPI(
    title="skill-audit API",
    description="Detect malicious patterns in AI agent skills/plugins. x402 v2 micropayments on Base.",
    version="2.0.0",
)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# x402 v2 payment middleware (official SDK)
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

_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):
        # Bazaar discovery extension for a POST/JSON endpoint. declare_discovery_extension
        # leaves info.input.method for runtime enrichment, but the schema marks it required;
        # inject "POST" so the extension validates at registration and CDP Bazaar can catalog it.
        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 /audit": RouteConfig(
            accepts=[PaymentOption(scheme="exact", pay_to=WALLET, price="$0.01", network=BASE_MAINNET)],
            mime_type="application/json",
            description="Audit text for malicious AI-skill patterns",
            extensions=_disc(
                {"content": "skill or plugin text to scan"},
                {"properties": {"content": {"type": "string", "description": "Text to audit"}}, "required": ["content"]},
                {"risk_score": 0, "risk_level": "clean", "total_findings": 0, "findings": []},
            ),
        ),
        "POST /audit/url": RouteConfig(
            accepts=[PaymentOption(scheme="exact", pay_to=WALLET, price="$0.03", network=BASE_MAINNET)],
            mime_type="application/json",
            description="Fetch a URL and audit its content",
            extensions=_disc(
                {"url": "https://example.com/skill.md"},
                {"properties": {"url": {"type": "string", "format": "uri", "description": "URL to fetch + audit"}}, "required": ["url"]},
                {"url": "https://example.com/skill.md", "risk_score": 0, "risk_level": "clean", "total_findings": 0, "findings": []},
            ),
        ),
    }
    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}")

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Models
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

class AuditRequest(BaseModel):
    content: str

class AuditUrlRequest(BaseModel):
    url: str
    max_size: Optional[int] = 500_000  # 500KB default limit

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Endpoints
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

@app.get("/")
async def root():
    pattern_count = sum(len(pg) for pgs in PATTERNS.values() for pg in pgs)
    return {
        "service": "skill-audit API",
        "version": "2.0.0",
        "description": "Detect malicious patterns in AI agent skills, plugins, and prompts.",
        "detects": [
            "download-and-execute", "credential exfiltration", "key generation",
            "prompt injection", "privilege escalation", "code execution",
            "identity impersonation", "seed phrase harvesting",
        ],
        "pattern_count": pattern_count,
        "endpoints": {
            "GET /": "Service info (free)",
            "GET /health": "Health check (free)",
            "POST /audit": "Audit text content ($0.01 USDC)",
            "POST /audit/url": "Fetch URL + audit ($0.03 USDC)",
        },
        "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("/audit")
async def audit_text(req: AuditRequest):
    content = req.content
    if not content or not content.strip():
        raise HTTPException(400, "content is required and must not be empty")
    if len(content) > 1_000_000:
        raise HTTPException(413, "content too large (max 1MB)")
    result = scan(content)
    return {
        "risk_score": result["risk_score"],
        "risk_level": result["risk_level"],
        "summary": result["summary"],
        "total_findings": result["total_findings"],
        "findings": result["findings"],
    }

@app.post("/audit/url")
async def audit_url(req: AuditUrlRequest):
    url = req.url
    if not url or not url.startswith(("http://", "https://")):
        raise HTTPException(400, "valid http/https URL required")

    import httpx
    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=15.0) as client:
            resp = await client.get(url, headers={"User-Agent": "skill-audit/1.0"})
            resp.raise_for_status()
    except httpx.HTTPStatusError as e:
        raise HTTPException(502, f"upstream returned {e.response.status_code}")
    except Exception as e:
        raise HTTPException(502, f"fetch failed: {type(e).__name__}: {e}")

    content = resp.text
    if len(content) > req.max_size:
        content = content[:req.max_size]

    result = scan(content)
    return {
        "url": url,
        "content_length": len(content),
        "risk_score": result["risk_score"],
        "risk_level": result["risk_level"],
        "summary": result["summary"],
        "total_findings": result["total_findings"],
        "findings": result["findings"],
    }

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Entry
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 8402))
    print(f"\n  skill-audit API (x402 v2) starting on :{port}")
    print(f"  x402: {'ENABLED' if _x402_available else 'DISABLED (pip install x402[fastapi,evm,extensions])'}")
    print(f"  Facilitator: {FACILITATOR_URL}")
    print(f"  Wallet: {WALLET}\n")
    uvicorn.run(app, host="0.0.0.0", port=port)