#!/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)