#!/usr/bin/env python3
"""InferPrice LLM exposure check: is your own LLM or MCP server open to anyone on the internet?

Run it from a machine OUTSIDE your network (a laptop on a phone hotspot works) against a URL you own:

    python3 llm-exposure-check.py http://203.0.113.7:11434
    python3 llm-exposure-check.py https://llm.example.com --mcp-path /mcp

Only test servers you own or are authorized to test. Python 3.8+, standard library only, read-only requests
(it never sends a prompt). Exit code 1 if any check FAILs. https://inferprice.pages.dev/monitor
"""
import argparse
import json
import sys
import urllib.error
import urllib.request

EVIL = "https://exposure-check.invalid"


def request(url, method="GET", body=None, headers=None, timeout=8):
    """(status, headers, text) or (None, {}, error message). Never raises."""
    req = urllib.request.Request(url, data=body, method=method, headers={"User-Agent": "inferprice-exposure-check", **(headers or {})})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, {k.lower(): v for k, v in r.headers.items()}, r.read(200_000).decode("utf-8", "replace")
    except urllib.error.HTTPError as ex:
        return ex.code, {k.lower(): v for k, v in ex.headers.items()}, ex.read(20_000).decode("utf-8", "replace")
    except Exception as ex:
        return None, {}, str(ex)


def is_json_with(text, key):
    try:
        d = json.loads(text)
    except ValueError:
        return False
    return isinstance(d, dict) and key in d


def checks(base, mcp_path="/mcp"):
    """List of (name, verdict, detail); verdict is FAIL, WARN, PASS or SKIP."""
    base = base.rstrip("/")
    out = []
    s, h, t = request(base + "/api/tags")
    if s == 200 and is_json_with(t, "models"):
        out.append(("Ollama API without auth", "FAIL", "GET /api/tags lists models to anyone; anyone can run prompts on your GPU"))
    elif s in (401, 403):
        out.append(("Ollama API without auth", "PASS", f"/api/tags answered {s}"))
    else:
        out.append(("Ollama API without auth", "SKIP", "no Ollama API here"))
    s, h, t = request(base + "/v1/models", headers={"Origin": EVIL})
    if s == 200 and is_json_with(t, "data"):
        out.append(("OpenAI-compatible API without auth", "FAIL", "GET /v1/models works with no API key"))
    elif s in (401, 403):
        out.append(("OpenAI-compatible API without auth", "PASS", f"/v1/models answered {s} without a key"))
    else:
        out.append(("OpenAI-compatible API without auth", "SKIP", "no OpenAI-compatible API here"))
    acao = h.get("access-control-allow-origin", "")
    if s is not None and (acao == "*" or acao == EVIL):
        out.append(("CORS open to any website", "FAIL" if s == 200 else "WARN",
                    f"Access-Control-Allow-Origin: {acao}; any web page a user opens can call this server from their browser"))
    elif s is not None:
        out.append(("CORS open to any website", "PASS", "no wildcard or reflected origin"))
    s, h, t = request(base + "/metrics")
    if s == 200 and ("# HELP" in t or "# TYPE" in t):
        out.append(("Metrics readable by anyone", "WARN", "/metrics is public: model names, load and request counts leak"))
    else:
        out.append(("Metrics readable by anyone", "PASS" if s in (401, 403, 404) else "SKIP", f"/metrics answered {s}"))
    body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
        "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "exposure-check", "version": "1"}}}).encode()
    s, h, t = request(base + mcp_path, "POST", body, {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"})
    if s == 200 and ('"result"' in t and ('"serverInfo"' in t or '"capabilities"' in t)):
        out.append(("MCP server without auth", "FAIL", f"POST {mcp_path} initialize succeeded with no token"))
    elif s == 401:
        out.append(("MCP server without auth", "PASS", f"{mcp_path} asks for a token (401)"))
    else:
        out.append(("MCP server without auth", "SKIP", f"no MCP server at {mcp_path}"))
    return out


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("url", nargs="+", help="base URL of a server you own, e.g. http://203.0.113.7:8000")
    ap.add_argument("--mcp-path", default="/mcp")
    ap.add_argument("--json", action="store_true", help="print JSON instead of a table")
    a = ap.parse_args(argv)
    results, failed = {}, False
    for url in a.url:
        results[url] = [{"check": n, "verdict": v, "detail": d} for n, v, d in checks(url, a.mcp_path)]
        failed |= any(r["verdict"] == "FAIL" for r in results[url])
    if a.json:
        print(json.dumps(results, indent=2))
    else:
        for url, rs in results.items():
            print(f"\n{url}")
            for r in rs:
                print(f"  {r['verdict']:<4}  {r['check']}: {r['detail']}")
        print("\nDaily monitoring with alerts: https://inferprice.pages.dev/monitor")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())