diff --git a/capabilities/web-security/capability.yaml b/capabilities/web-security/capability.yaml index dd9ea51..58e6049 100644 --- a/capabilities/web-security/capability.yaml +++ b/capabilities/web-security/capability.yaml @@ -86,6 +86,14 @@ mcp: url: "http://127.0.0.1:9876/" init_timeout: 15 timeout: 3600 + securitycontext: + command: "uv" + args: + - "run" + - "${CAPABILITY_ROOT}/mcp/securitycontext.py" + env: + SECURITYCONTEXT_API_KEY: "${SECURITYCONTEXT_API_KEY:-}" + init_timeout: 30 dependencies: packages: diff --git a/capabilities/web-security/mcp/securitycontext.py b/capabilities/web-security/mcp/securitycontext.py new file mode 100644 index 0000000..ead5f90 --- /dev/null +++ b/capabilities/web-security/mcp/securitycontext.py @@ -0,0 +1,200 @@ +#!/usr/bin/env -S uv run +# /// script +# requires-python = ">=3.12" +# dependencies = [ +# "fastmcp>=2.0", +# "httpx>=0.28", +# ] +# /// +"""SecurityContext MCP proxy. + +Stdio bridge to the remote SecurityContext Streamable HTTP server at +https://securitycontext.dev/mcp. Runs locally as a stdio MCP server +so the Dreadnode runtime can connect reliably, and proxies tool calls +to the remote endpoint. +""" + +from __future__ import annotations + +import asyncio +import json +import os + +import httpx +from fastmcp import FastMCP + +REMOTE_URL = os.environ.get( + "SECURITYCONTEXT_URL", "https://securitycontext.dev/mcp" +) +API_KEY = os.environ.get("SECURITYCONTEXT_API_KEY", "") +TIMEOUT = 120 + +mcp = FastMCP("securitycontext") + + +class _RemoteSession: + """Manages a single MCP session against the remote server.""" + + def __init__(self) -> None: + self._client = httpx.AsyncClient(timeout=TIMEOUT) + self._session_id: str | None = None + self._initialized = False + + async def _ensure_initialized(self) -> None: + if self._initialized: + return + + headers = self._base_headers() + # Step 1: initialize + init_req = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-03-26", + "capabilities": {}, + "clientInfo": {"name": "dreadnode-agent", "version": "1.0"}, + }, + } + r = await self._client.post(REMOTE_URL, headers=headers, json=init_req) + r.raise_for_status() + self._session_id = r.headers.get("mcp-session-id") + + # Step 2: send initialized notification + notif = { + "jsonrpc": "2.0", + "method": "notifications/initialized", + "params": {}, + } + headers = self._base_headers() + r2 = await self._client.post(REMOTE_URL, headers=headers, json=notif) + # 202 expected for notifications + self._initialized = True + + def _base_headers(self) -> dict[str, str]: + h: dict[str, str] = { + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + } + if self._session_id: + h["mcp-session-id"] = self._session_id + if API_KEY: + h["X-Api-Key"] = API_KEY + return h + + async def call_tool(self, name: str, arguments: dict) -> str: + await self._ensure_initialized() + req = { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": {"name": name, "arguments": arguments}, + } + r = await self._client.post( + REMOTE_URL, headers=self._base_headers(), json=req + ) + r.raise_for_status() + return self._parse_response(r.text) + + @staticmethod + def _parse_response(body: str) -> str: + """Extract text content from SSE or plain JSON response.""" + # Try SSE format first + for line in body.split("\n"): + if line.startswith("data: "): + try: + data = json.loads(line[6:]) + if "result" in data: + parts = [] + for c in data["result"].get("content", []): + if c.get("type") == "text": + parts.append(c["text"]) + if parts: + return "\n".join(parts) + if "error" in data: + return f"Error: {data['error']}" + except json.JSONDecodeError: + continue + + # Fall back to plain JSON + try: + data = json.loads(body) + if "result" in data: + parts = [] + for c in data["result"].get("content", []): + if c.get("type") == "text": + parts.append(c["text"]) + if parts: + return "\n".join(parts) + if "error" in data: + return f"Error: {data['error']}" + except json.JSONDecodeError: + pass + + return body + + +_session = _RemoteSession() + + +@mcp.tool +async def get_security_context( + repo: str, + wait: int = 30, +) -> str: + """Get ready-to-use security context for a GitHub repo: the project's + history of fixed vulnerabilities, disclosed CVEs, recurring weak spots, + a hunting brief, and a preview of the top variant leads. + + Call this before auditing, fuzzing, or reviewing a repo. If no context + exists yet, call create_security_context first. + + Args: + repo: GitHub repo in owner/name format (e.g. "vercel/next.js"). + wait: Seconds to wait if context is still building (max 60). + """ + return await _session.call_tool( + "get_security_context", {"repo": repo, "wait": wait} + ) + + +@mcp.tool +async def create_security_context( + repo: str, + wait: int = 60, +) -> str: + """Build (or refresh) the security context for a GitHub repo. Use this + when get_security_context reports that none exists yet. A build typically + takes ~30-60s; this tool waits for completion. + + Args: + repo: GitHub repo in owner/name format (e.g. "vercel/next.js"). + wait: Seconds to wait for completion (max 60). + """ + return await _session.call_tool( + "create_security_context", {"repo": repo, "wait": wait} + ) + + +@mcp.tool +async def get_vulnerability_leads( + repo: str, + severity: str = "", + limit: int = 25, +) -> str: + """List the ranked variant-lead backlog for a GitHub repo: concrete spots + in the CURRENT code that match a past fix's dangerous pattern. + + Args: + repo: GitHub repo in owner/name format. + severity: Filter by severity (critical, high, medium, low). Empty for all. + limit: Max leads to return. + """ + args: dict = {"repo": repo, "limit": limit} + if severity: + args["severity"] = severity + return await _session.call_tool("get_vulnerability_leads", args) + + +if __name__ == "__main__": + mcp.run()