scripts/x402_api/main.py

#!/usr/bin/env python3
"""secrets-audit x402 API — Secret/credential leak detection as a paid service (x402 v2).

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

Endpoints:
  GET  /          — Service info (free)
  GET  /health    — Health check (free)
  POST /audit     — Text secret-scan $0.01/call (x402)
  POST /audit/url — URL fetch + secret-scan $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,
v2-native, auto-lists on the discovery layer once the first payment settles.
Sibling of the skill-audit endpoint; same v2 stack, different scan engine.
"""
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 secrets-audit MCP server
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "mcp_servers", "secrets-audit"))
from server import scan, RULES  # 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="secrets-audit API",
    description="Detect leaked secrets/credentials in code, configs, and text. 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="Scan text for leaked secrets and credentials",
            extensions=_disc(
                {"content": "code or config text to scan"},
                {"properties": {"content": {"type": "string", "description": "Text to scan for secrets"}}, "required": ["content"]},
                {"risk_score": 0, "risk_level": "SAFE", "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 scan it for leaked secrets",
            extensions=_disc(
                {"url": "https://example.com/config.env"},
                {"properties": {"url": {"type": "string", "format": "uri", "description": "URL to fetch + scan"}}, "required": ["url"]},
                {"url": "https://example.com/config.env", "risk_score": 0, "risk_level": "SAFE", "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

def _shape(result: dict, **extra) -> dict:
    out = {
        "risk_score": result["risk_score"],
        "risk_level": result["risk_level"],
        "summary": result["summary"],
        "total_findings": len(result["findings"]),
        "findings": result["findings"],
    }
    out.update(extra)
    return out

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

@app.get("/")
async def root():
    return {
        "service": "secrets-audit API",
        "version": "2.0.0",
        "description": "Detect leaked secrets, API keys, and credentials in code, configs, and text.",
        "detects": sorted({r["name"] for r in RULES})[:24],
        "rule_count": len(RULES),
        "endpoints": {
            "GET /": "Service info (free)",
            "GET /health": "Health check (free)",
            "POST /audit": "Scan text content ($0.01 USDC)",
            "POST /audit/url": "Fetch URL + scan ($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)")
    return _shape(scan(content))

@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": "secrets-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]

    return _shape(scan(content), url=url, content_length=len(content))

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

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