diff --git a/.env.example b/.env.example index 49cb001..36e627a 100644 --- a/.env.example +++ b/.env.example @@ -57,3 +57,9 @@ TORPROXY_SOCKS_PORT=9050 # TORPROXY_REQUEST_TIMEOUT=3 # TORPROXY_MAX_WORKERS=32 # TORPROXY_MAX_BODY_BYTES=16384 + +# Model Context Protocol (MCP) server settings (ISSUE-122). +# Built-in zero-dependency MCP server on port 2501 for AI agents (Claude Desktop, Cursor, Antigravity). +# TORPROXY_MCP_ENABLED=1 +# TORPROXY_MCP_PORT=2501 +# TORPROXY_MCP_LISTEN_ADDRESS=0.0.0.0 diff --git a/.github/actions/live-verification/action.yml b/.github/actions/live-verification/action.yml index bfb9004..92bcf2a 100644 --- a/.github/actions/live-verification/action.yml +++ b/.github/actions/live-verification/action.yml @@ -65,6 +65,7 @@ runs: -p 8118:8118 \ -p 9050:9050 \ -p 8080:8080 \ + -p 2501:2501 \ "${{ inputs.image_ref }}") echo "container_id=$CONTAINER_ID" >> "$GITHUB_OUTPUT" @@ -213,6 +214,52 @@ runs: echo "✅ SOCKS5 Proxy OK: Exit-IP is $EXIT_IP (IsTor=true)" fi + # ------------------------------------------------------------- + # 5b. MCP Server & torproxy_fetch live verification + # ------------------------------------------------------------- + - name: Verify MCP Server & torproxy_fetch Live Canary + shell: bash + run: | + echo "Testing MCP Server tools and live fetch over Tor..." + RUNNER_IP="${{ steps.host-info.outputs.runner_ip }}" + + # 1. MCP Stdio test: ping and tool listing + PING_RESP=$(echo '{"jsonrpc":"2.0","id":1,"method":"ping"}' | docker exec -i torproxy-canary python3 -m torproxy mcp 2>/dev/null) + echo "MCP Ping: $PING_RESP" + echo "$PING_RESP" | jq -e '.result == {}' >/dev/null + + # 2. Live fetch test via torproxy_fetch tool over Tor + FETCH_REQ='{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"torproxy_fetch","arguments":{"url":"https://check.torproject.org/api/ip"}}}' + FETCH_RESP=$(echo "$FETCH_REQ" | docker exec -i torproxy-canary python3 -m torproxy mcp 2>/dev/null) + echo "MCP Fetch Response: $FETCH_RESP" + echo "$FETCH_RESP" | jq -e '.result.isError == false' >/dev/null + + PAYLOAD=$(echo "$FETCH_RESP" | jq -r '.result.content[0].text') + STATUS_CODE=$(echo "$PAYLOAD" | jq -r '.status') + BODY=$(echo "$PAYLOAD" | jq -r '.body') + + if [ "$STATUS_CODE" != "200" ]; then + echo "❌ MCP torproxy_fetch returned status $STATUS_CODE (expected 200)" + exit 1 + fi + + FETCH_IP=$(echo "$BODY" | jq -r '.IP // empty') + IS_TOR=$(echo "$BODY" | jq -r '.IsTor // empty') + + echo "MCP fetched exit IP: $FETCH_IP (IsTor: $IS_TOR)" + + if [ "$FETCH_IP" = "$RUNNER_IP" ]; then + echo "❌ CRITICAL: MCP torproxy_fetch leaked host IP ($FETCH_IP == $RUNNER_IP)!" + exit 1 + fi + + if [ "$IS_TOR" = "false" ]; then + echo "❌ CRITICAL: check.torproject.org reported MCP fetch was NOT through Tor!" + exit 1 + fi + + echo "✅ MCP Server live verification passed: fetched through Tor, leak-free." + # ------------------------------------------------------------- # 6. Circuit rotation (/reconnect -> SIGNAL NEWNYM) # ------------------------------------------------------------- diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ccd81d..66b54bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.6.0] — 2026-11-10 + +### Added +- **mcp**: Integrated Model Context Protocol (MCP) server on port 2501 and Stdio transport (ISSUE-122, #123) +- **mcp**: Zero-config `torproxy_fetch` tool for leak-free HTTP/HTTPS and `.onion` fetches +- **canary**: Add live MCP server and `torproxy_fetch` verification to canary and container tests + ## [2.1.1] — 2026-11-07 ### Fixed @@ -184,7 +191,8 @@ Sprint 1 security & contract correctness release. --- -[Unreleased]: https://github.com/hapheus/torproxy/compare/v2.1.1...HEAD +[Unreleased]: https://github.com/hapheus/torproxy/compare/v2.6.0...HEAD +[2.6.0]: https://github.com/hapheus/torproxy/compare/v2.5.0...v2.6.0 [2.1.1]: https://github.com/hapheus/torproxy/compare/v2.1.0...v2.1.1 [2.1.0]: https://github.com/hapheus/torproxy/compare/v2.0.2...v2.1.0 [2.0.2]: https://github.com/hapheus/torproxy/compare/v2.0.1...v2.0.2 diff --git a/README.md b/README.md index bbea110..6210057 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ [![OpenSSF Scorecard](https://api.scorecard.dev/projects/github.com/hapheus/torproxy/badge)](https://scorecard.dev/viewer/?uri=github.com/hapheus/torproxy) [![OpenSSF Best Practices](https://img.shields.io/cii/percentage/15828?style=flat-square&logo=linuxfoundation&label=OpenSSF%20Best%20Practices)](https://www.bestpractices.dev/projects/15828) -[Quick Start](#quick-start) · [Transparent Gateway](#transparent-tor-gateway) · [Architecture](#architecture) · [REST API](docs/api.md) · [Configuration](docs/configuration.md) · [Examples](examples/) · [Contributing](CONTRIBUTING.md) +[Quick Start](#quick-start) · [Model Context Protocol (MCP)](#-model-context-protocol-mcp) · [Transparent Gateway](#transparent-tor-gateway) · [Architecture](#architecture) · [REST API](docs/api.md) · [Configuration](docs/configuration.md) · [Examples](examples/) · [Contributing](CONTRIBUTING.md) @@ -113,6 +113,46 @@ services: --- +## 🤖 Model Context Protocol (MCP) + +`torproxy` includes a built-in, zero-dependency **Model Context Protocol (MCP)** server on port `2501` (SSE) and via Stdio (`torproxy mcp`), allowing AI agents (Claude Desktop, Cursor, Antigravity) to manage Tor circuits and execute anonymized web requests directly. + +### Claude Desktop Integration (`claude_desktop_config.json`) +Connect Claude Desktop to your running `torproxy` container via Stdio without exposing host ports: +```json +{ + "mcpServers": { + "torproxy": { + "command": "docker", + "args": ["exec", "-i", "torproxy", "torproxy", "mcp"] + } + } +} +``` + +### Cursor & Antigravity Integration (`.cursor/mcp.json`) +Connect via HTTP/SSE: +```json +{ + "mcpServers": { + "torproxy": { + "url": "http://127.0.0.1:2501/sse" + } + } +} +``` + +### Key Tools Available to Agents +- `torproxy_fetch`: **Zero-config fetch** — execute HTTP/HTTPS or `.onion` requests strictly routed through Tor (no local curl or host proxy setup needed). +- `torproxy_status`: Retrieve live bootstrap progress, egress IP, and GeoIP metadata. +- `torproxy_get_ip`: Query the current verified Tor exit IP. +- `torproxy_reconnect`: Request a new Tor identity (`SIGNAL NEWNYM`), optionally changing exit country (`country="DE"`). +- `torproxy_health` & `torproxy_diagnostics`: Automated operational health and circuit diagnostics. + +See [docs/mcp.md](docs/mcp.md) for full configuration details, authentication options, and tool schemas. + +--- + ## Available Interfaces and Ports | Interface | Default Port | Internal Service | Description | @@ -120,6 +160,7 @@ services: | **HTTP Proxy** | `8118` | Privoxy | HTTP/HTTPS forward proxy routing traffic to Tor. Fails closed (500/503) if Tor is disconnected. | | **SOCKS5 Proxy** | `9050` | Tor | Direct Tor SOCKS5 listener. Supports remote DNS resolution via `socks5h://`. | | **REST API** | `8080` | Python REST API | JSON management and health endpoint. Circuit rotation and connection management. | +| **MCP Server (SSE)** | `2501` | Python MCP Server | Model Context Protocol server (SSE) for AI agents (Cursor, Antigravity). | | **TransPort** *(Gateway Mode)* | `9040` | Tor | Transparent TCP proxy redirect listener when `TORPROXY_GATEWAY_MODE=true`. | | **DNSPort** *(Gateway Mode)* | `9053` | Tor | Transparent DNS resolver listener when `TORPROXY_GATEWAY_MODE=true`. | diff --git a/docker-compose.yml b/docker-compose.yml index dec6b47..f70ed51 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,8 +15,11 @@ services: TORPROXY_STRICT_NODES: ${TORPROXY_STRICT_NODES:-} TORPROXY_API_TOKEN: ${TORPROXY_API_TOKEN:-} TORPROXY_API_ALLOW_ANONYMOUS_HEALTH: ${TORPROXY_API_ALLOW_ANONYMOUS_HEALTH:-} + TORPROXY_MCP_ENABLED: ${TORPROXY_MCP_ENABLED:-true} + TORPROXY_MCP_PORT: ${TORPROXY_MCP_PORT:-2501} ports: - "127.0.0.1:${TORPROXY_HOST_PORT:-8118}:${TORPROXY_LISTEN_PORT:-8118}" - "127.0.0.1:${TORPROXY_API_PORT:-8080}:${TORPROXY_API_PORT:-8080}" - "127.0.0.1:${TORPROXY_SOCKS_PORT:-9050}:${TORPROXY_SOCKS_PORT:-9050}" + - "127.0.0.1:${TORPROXY_MCP_PORT:-2501}:${TORPROXY_MCP_PORT:-2501}" restart: unless-stopped diff --git a/docker/Dockerfile b/docker/Dockerfile index 2009cf5..d6a084f 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -7,7 +7,7 @@ ARG PYTHON_VERSION=3.14 LABEL org.opencontainers.image.title="torproxy" \ org.opencontainers.image.description="A minimal HTTP proxy routed through Tor" \ - org.opencontainers.image.version="2.5.0" \ + org.opencontainers.image.version="2.6.0" \ org.opencontainers.image.licenses="MIT" \ org.opencontainers.image.source="https://github.com/hapheus/torproxy" \ org.opencontainers.image.url="https://github.com/hapheus/torproxy" @@ -36,7 +36,9 @@ ENV PYTHON_VERSION=${PYTHON_VERSION} \ TORPROXY_LISTEN_ADDRESS=0.0.0.0 \ TORPROXY_LISTEN_PORT=8118 \ TORPROXY_API_PORT=8080 \ - TORPROXY_SOCKS_PORT=9050 + TORPROXY_SOCKS_PORT=9050 \ + TORPROXY_MCP_PORT=2501 \ + TORPROXY_MCP_ENABLED=true # Stage sources, then install into the sysconfig-derived site-packages so the # layout follows the interpreter actually present in the image (ISSUE-15). @@ -61,7 +63,7 @@ COPY --chmod=644 docker/torrc /etc/tor/torrc # torproxy) so the gateway netfilter rules can scope the WAN bypass to # the dedicated tor UID. No daemon ever runs as root. -EXPOSE 8118 8080 9050 +EXPOSE 8118 8080 9050 2501 HEALTHCHECK --interval=30s --timeout=15s --start-period=180s --retries=3 \ CMD ["/usr/local/bin/healthcheck.sh"] diff --git a/docs/compatibility.md b/docs/compatibility.md index e7ba02f..402a283 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -19,6 +19,7 @@ This document defines the strict, non-negotiable backward compatibility contract | **HTTP Proxy** | `8118` | `8118` (`127.0.0.1`) | HTTP (Privoxy) | Unauthenticated | Standard HTTP/HTTPS forward proxy routed via Tor SOCKS5. | | **SOCKS5 Proxy** | `9050` | `9050` (`127.0.0.1`) | SOCKS5 (Tor) | Unauthenticated | Tor SOCKS5 proxy with remote DNS resolution (`socks5h`). | | **Control REST API** | `8080` | `8080` (`127.0.0.1`) | HTTP/REST | Unauthenticated | JSON management and monitoring interface. | +| **MCP Server (SSE)** | `2501` | `2501` (`127.0.0.1`) | HTTP/SSE (JSON-RPC) | Inherits API Token | Built-in Model Context Protocol server for AI agents. | | **Tor ControlPort** | `9051` | Not Published | Tor Control | Cookie Auth | Internal loopback only (`127.0.0.1:9051`), never published to host. | ### Network Guarantees vs. Implementation Details @@ -49,6 +50,9 @@ This document defines the strict, non-negotiable backward compatibility contract | `TORPROXY_API_TOKEN` | *(unset)* | String | Optional: Bearer token / API key securing REST API (8080). Default: unset (unauthenticated). | | `TORPROXY_API_ALLOW_ANONYMOUS_HEALTH` | `0` | Boolean (`0`, `1`, `true`, `false`) | Optional: Permits unauthenticated `/health` probe access when token auth is enabled. | | `TORPROXY_SOCKS_ALLOWED_CIDRS` | *(unset)* | Comma-separated CIDRs | Optional: Restricts SOCKS port access by source CIDR (Tor `SocksPolicy`). Default unset = accept all (legacy behavior). | +| `TORPROXY_MCP_ENABLED` | `1` | Boolean (`0`, `1`, `true`, `false`) | Optional additive: Enables built-in MCP server. Default `1` (enabled). | +| `TORPROXY_MCP_PORT` | `2501` | Integer `1`–`65535` | Listening port for the MCP HTTP/SSE server. | +| `TORPROXY_MCP_LISTEN_ADDRESS` | `0.0.0.0` | Valid IP address (IPv4 or IPv6) | Bind address for the MCP server inside the container. | --- diff --git a/docs/mcp.md b/docs/mcp.md new file mode 100644 index 0000000..abdb3fa --- /dev/null +++ b/docs/mcp.md @@ -0,0 +1,178 @@ +# Model Context Protocol (MCP) Integration + +`torproxy` features a built-in, zero-dependency **Model Context Protocol (MCP)** server enabling AI coding assistants (such as Claude Desktop, Cursor, Antigravity, and custom LLM agent pipelines) to inspect Tor status, trigger circuit renewals, and perform anonymized HTTP/HTTPS or `.onion` network requests without manual script execution or external bridge daemons. + +--- + +## 1. Architecture & Dual-Transport Support + +The MCP server adheres strictly to the `torproxy` zero-dependency standard (using pure Python standard library: `http.server`, `json`, `threading`, `queue`). + +It supports two distinct transport protocols: + +1. **Stdio Transport (`torproxy mcp`):** + - Designed for local processes and container execution: `docker exec -i torproxy torproxy mcp`. + - Communicates using newline-delimited JSON-RPC 2.0 frames over `stdin` and `stdout`. + - All internal diagnostics and operational warnings are directed to `stderr` to maintain clean protocol isolation. + - Recommended for **Claude Desktop**. +2. **HTTP/SSE Transport (Port 2501):** + - Runs as a non-blocking daemon thread on port `2501` (`TORPROXY_MCP_PORT=2501`). + - Exposes `GET /sse` for Server-Sent Events stream initialization and `POST /messages?sessionId=...` for JSON-RPC dispatch. + - Recommended for **Cursor**, **Antigravity**, and remote container orchestrators. + +--- + +## 2. Quickstart Configuration Examples + +### Claude Desktop (`claude_desktop_config.json`) + +Connect Claude Desktop to your running `torproxy` container over Stdio without exposing any network ports: + +```json +{ + "mcpServers": { + "torproxy": { + "command": "docker", + "args": ["exec", "-i", "torproxy", "torproxy", "mcp"] + } + } +} +``` + +*When token authentication (`TORPROXY_API_TOKEN`) is enabled:* +```json +{ + "mcpServers": { + "torproxy": { + "command": "docker", + "args": ["exec", "-i", "-e", "TORPROXY_API_TOKEN=your-token", "torproxy", "torproxy", "mcp"] + } + } +} +``` + +--- + +### Cursor (`.cursor/mcp.json`) + +#### Option A: HTTP / SSE (Port 2501) +```json +{ + "mcpServers": { + "torproxy": { + "url": "http://127.0.0.1:2501/sse" + } + } +} +``` + +*With authentication (`TORPROXY_API_TOKEN`):* +```json +{ + "mcpServers": { + "torproxy": { + "url": "http://127.0.0.1:2501/sse?token=your-token", + "headers": { + "Authorization": "Bearer your-token" + } + } + } +} +``` + +#### Option B: Stdio Transport via Docker +```json +{ + "mcpServers": { + "torproxy": { + "command": "docker", + "args": ["exec", "-i", "torproxy", "torproxy", "mcp"] + } + } +} +``` + +--- + +### Antigravity & Gemini CLI (`settings.json`) + +```json +{ + "mcpServers": { + "torproxy": { + "command": "docker", + "args": ["exec", "-i", "torproxy", "torproxy", "mcp"] + } + } +} +``` + +--- + +### Docker Compose (`docker-compose.yml`) + +Publish port `2501` for SSE network clients: + +```yaml +services: + torproxy: + image: hapheus/torproxy:latest + container_name: torproxy + restart: unless-stopped + ports: + - "127.0.0.1:8118:8118" # HTTP/HTTPS Proxy (Privoxy) + - "127.0.0.1:9050:9050" # SOCKS5 Proxy (Tor) + - "127.0.0.1:8080:8080" # REST API + - "127.0.0.1:2501:2501" # MCP Server (SSE) + environment: + - TORPROXY_MCP_ENABLED=true + - TORPROXY_MCP_PORT=2501 + # Optional: + # - TORPROXY_API_TOKEN=your-token +``` + +--- + +## 3. Available Tools + +| Tool | Parameters | Description | +|---|---|---| +| `torproxy_fetch` | `url` *(required)*, `method`, `headers`, `body`, `timeout`, `max_bytes` | **Zero-config fetch:** Execute an HTTP/HTTPS or `.onion` request routed strictly through Tor without shell or curl. | +| `torproxy_status` | *(none)* | Live status JSON: bootstrap progress, egress IP, and GeoIP metadata. | +| `torproxy_get_ip` | *(none)* | Current egress IP address verified through Tor. | +| `torproxy_reconnect` | `country` *(optional)*, `strict` *(optional)* | Signal Tor for a new identity/circuit (`SIGNAL NEWNYM`), optionally changing exit nodes. | +| `torproxy_connect` | `country` *(optional)*, `strict` *(optional)* | Enable the Tor network (`SETCONF DisableNetwork=0`). | +| `torproxy_disconnect` | *(none)* | Disable the Tor network (`SETCONF DisableNetwork=1`). | +| `torproxy_health` | *(none)* | Probe status of Tor, Privoxy HTTP, and SOCKS readiness. | +| `torproxy_diagnostics` | *(none)* | Comprehensive telemetry (circuits, cache age, reconnection stats, gateway mode). | + +### `torproxy_fetch` Details +AI agents frequently need to fetch web pages or inspect `.onion` hidden services without having terminal or shell access on the host. `torproxy_fetch` executes requests from within the container routed through the local Privoxy proxy, guaranteeing zero clearnet leaks and remote Tor DNS resolution. + +**Arguments:** +- `url` (string, required): Destination URL (`http://...`, `https://...`, or `http://...onion`). +- `method` (string, optional, default: `"GET"`): HTTP method (`GET`, `POST`, `PUT`, `DELETE`, `HEAD`). +- `headers` (object, optional): Key-value string dictionary of HTTP headers. +- `body` (string, optional): String payload for POST or PUT requests. +- `timeout` (number, optional, default: `30`): Timeout in seconds. +- `max_bytes` (integer, optional, default: `1048576`): Caps response body length to protect LLM context windows. + +--- + +## 4. Available Resources + +| URI | MIME Type | Description | +|---|---|---| +| `torproxy://status` | `application/json` | Real-time JSON status snapshot of the proxy instance. | +| `torproxy://metrics` | `text/plain` | Prometheus telemetry metrics export. | + +--- + +## 5. Security & Authentication + +When `TORPROXY_API_TOKEN` is configured: +- **Stdio:** Pass the token via environment variable (`-e TORPROXY_API_TOKEN=...`). +- **SSE:** Authenticate using one of: + - `Authorization: Bearer ` + - `X-API-Key: ` + - URL query parameter: `/sse?token=` (useful for browser/EventSource clients where headers cannot be modified). diff --git a/docs/releases.md b/docs/releases.md index 2d1536f..d2ecab1 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -91,6 +91,12 @@ If an upgraded image exhibits unexpected behavior in your environment: ## 6. Release Notes +### 2.6.0 (Model Context Protocol server & torproxy_fetch) + +- **Integrated MCP Server (ISSUE-122, PR #123):** Built-in zero-dependency Model Context Protocol (MCP) server running on port `2501` (HTTP/SSE) and via Stdio (`torproxy mcp`), enabling autonomous AI agents (Claude Desktop, Cursor, Antigravity) to orchestrate Tor circuits and inspect status. +- **Native `torproxy_fetch`:** Zero-config HTTP/HTTPS and `.onion` fetching routed strictly through Tor without host proxy configuration or clearnet leaks. +- **Canary verification:** Daily live canary suite now verifies `torproxy_fetch` against `check.torproject.org` with runner host IP leak detection. + ### 2.5.0 (Feature roadmap follow-ups: TLS example, IPv6, SOCKS policy) - **GeoIP User-Agent fix (ISSUE-20 regression, #113, PR #114):** the GeoIP downloader no longer sends the custom `torproxy-geoip/1` User-Agent — GitLab answers raw files only to known UA patterns (`curl/*`, `Python-urllib/*`) with HTTP 403 for custom UAs. The daily canary failure is fixed; fallback to the bundled database was already working. diff --git a/pyproject.toml b/pyproject.toml index 21f0382..98b0f36 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "torproxy" -version = "2.5.0" +version = "2.6.0" description = "A minimal HTTP and SOCKS5 proxy routed through the Tor network with local REST control API" readme = "README.md" requires-python = ">=3.10" diff --git a/src/torproxy/bootstrap.py b/src/torproxy/bootstrap.py index e9b5fa9..542ab51 100644 --- a/src/torproxy/bootstrap.py +++ b/src/torproxy/bootstrap.py @@ -5,7 +5,10 @@ from torproxy.adapters.probes import SocksReadinessProbe, TorEgressProbe from torproxy.adapters.tor_control import TorControlClient from torproxy.configuration import Settings +from torproxy.mcp.protocol import McpHandler +from torproxy.mcp.sse import McpSseServer from torproxy.ports import Clock, ControlPort, EgressProbe, GeoIpStatus, SocksProbe +from torproxy.queries import ProxyQueries from torproxy.service import ProxyService from torproxy.state import RuntimeState from torproxy.web.router import Router @@ -37,3 +40,18 @@ def create_server(settings: Settings, address: tuple[str, int] | None = None) -> Router(create_services(settings)), settings, ) + + +def create_mcp_handler(settings: Settings, service: ProxyService | None = None) -> McpHandler: + svc = service or create_services(settings) + return McpHandler(svc, ProxyQueries(svc)) + + +def create_mcp_server( + settings: Settings, + handler: McpHandler | None = None, + address: tuple[str, int] | None = None, +) -> McpSseServer: + h = handler or create_mcp_handler(settings) + addr = address or (settings.mcp_listen_address, settings.mcp_port) + return McpSseServer(addr, h, settings) diff --git a/src/torproxy/configuration.py b/src/torproxy/configuration.py index 907c477..85a6a0a 100644 --- a/src/torproxy/configuration.py +++ b/src/torproxy/configuration.py @@ -86,6 +86,10 @@ class Settings: # isolation, never a credential check — so source-IP policy # is the strongest access control Tor itself offers. socks_allowed_cidrs: str = "" + # ISSUE-122: integrated Model Context Protocol (MCP) server + mcp_enabled: bool = True + mcp_port: int = 2501 + mcp_listen_address: str = "0.0.0.0" def __post_init__(self) -> None: # ISSUE-18b (#107): IPv4 and IPv6 listen addresses are @@ -98,12 +102,18 @@ def __post_init__(self) -> None: ipaddress.ip_address(self.api_listen_address) except ValueError: raise ValueError("TORPROXY_API_LISTEN_ADDRESS must be a valid IP address") from None + try: + ipaddress.ip_address(self.mcp_listen_address) + except ValueError: + raise ValueError("TORPROXY_MCP_LISTEN_ADDRESS must be a valid IP address") from None ports_to_check = [ ("TORPROXY_LISTEN_PORT", self.proxy_port), ("TORPROXY_API_PORT", self.api_port), ("TORPROXY_SOCKS_PORT", self.socks_port), ("TOR_CONTROL_PORT", self.control_port), ] + if self.mcp_enabled: + ports_to_check.append(("TORPROXY_MCP_PORT", self.mcp_port)) if self.gateway_mode: ports_to_check.extend( [ @@ -178,6 +188,10 @@ def port(name: str, default: int) -> int: raw_gw = env.get("TORPROXY_GATEWAY_MODE", "0").strip().lower() gateway_mode = raw_gw in ("1", "true", "yes") + raw_mcp = env.get("TORPROXY_MCP_ENABLED", "1").strip().lower() + mcp_enabled = raw_mcp in ("1", "true", "yes") + mcp_listen_address = env.get("TORPROXY_MCP_LISTEN_ADDRESS", "0.0.0.0") + geoip_auto, error = validate_strict_nodes_input( env.get("TORPROXY_GEOIP_AUTO_UPDATE") or None ) @@ -239,6 +253,9 @@ def _int_setting(name: str, default: int) -> int: int(env.get("TORPROXY_GEOIP_MAX_BYTES", str(32 * 1024 * 1024))), ), socks_allowed_cidrs=env.get("TORPROXY_SOCKS_ALLOWED_CIDRS", "").strip(), + mcp_enabled=mcp_enabled, + mcp_port=port("TORPROXY_MCP_PORT", 2501), + mcp_listen_address=mcp_listen_address, ) diff --git a/src/torproxy/mcp/__init__.py b/src/torproxy/mcp/__init__.py new file mode 100644 index 0000000..607cbc4 --- /dev/null +++ b/src/torproxy/mcp/__init__.py @@ -0,0 +1,5 @@ +"""Model Context Protocol (MCP) server support for torproxy.""" + +from torproxy.mcp.protocol import McpHandler + +__all__ = ["McpHandler"] diff --git a/src/torproxy/mcp/protocol.py b/src/torproxy/mcp/protocol.py new file mode 100644 index 0000000..e49f735 --- /dev/null +++ b/src/torproxy/mcp/protocol.py @@ -0,0 +1,495 @@ +"""Pure Python zero-dependency Model Context Protocol (MCP) handler.""" + +import json +import urllib.error +import urllib.parse +import urllib.request +from typing import Any + +from torproxy import __version__ +from torproxy.queries import ProxyQueries +from torproxy.service import ControlFailure, InvalidInput, ProxyService + +MCP_PROTOCOL_VERSION = "2024-11-05" + +TOOLS_SCHEMA: list[dict[str, Any]] = [ + { + "name": "torproxy_status", + "description": ( + "Retrieve current Tor status, bootstrap progress, egress IP, and GeoIP metadata." + ), + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_get_ip", + "description": "Query the current egress IP address routed through Tor.", + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_reconnect", + "description": ( + "Trigger a new Tor circuit/identity (SIGNAL NEWNYM), optionally switching " + "exit country or strict nodes enforcement." + ), + "inputSchema": { + "type": "object", + "properties": { + "country": { + "type": "string", + "description": ( + "Optional 2-letter ISO country code or comma-separated list of country " + "codes (e.g. 'de' or 'de,at'). Pass 'any' or empty string to reset." + ), + }, + "strict": { + "type": "boolean", + "description": ( + "Optional boolean to strictly enforce exit nodes without fallback." + ), + }, + }, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_connect", + "description": ( + "Enable the Tor network (SETCONF DisableNetwork=0), optionally updating " + "country or strict nodes." + ), + "inputSchema": { + "type": "object", + "properties": { + "country": { + "type": "string", + "description": ( + "Optional 2-letter ISO country code or comma-separated list of country " + "codes (e.g. 'de'). Pass 'any' or empty string to reset." + ), + }, + "strict": { + "type": "boolean", + "description": ( + "Optional boolean to strictly enforce exit nodes without fallback." + ), + }, + }, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_disconnect", + "description": "Disable the Tor network (SETCONF DisableNetwork=1).", + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_health", + "description": "Check health status of Tor proxy, Privoxy HTTP, and SOCKS readiness.", + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_diagnostics", + "description": ( + "Detailed diagnostics including bootstrap state, circuit config, cache TTL age, " + "reconnection metrics, and transparent gateway status." + ), + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "torproxy_fetch", + "description": ( + "Execute an HTTP/HTTPS or .onion request routed strictly through the Tor proxy." + ), + "inputSchema": { + "type": "object", + "properties": { + "url": { + "type": "string", + "description": "Target URL to fetch (HTTP, HTTPS, or .onion).", + }, + "method": { + "type": "string", + "description": "HTTP method (GET, POST, PUT, DELETE, HEAD). Default: GET.", + }, + "headers": { + "type": "object", + "description": "Optional HTTP headers mapping string keys to string values.", + }, + "body": { + "type": "string", + "description": "Optional string request body for POST/PUT requests.", + }, + "timeout": { + "type": "number", + "description": "Request timeout in seconds (default: 30).", + }, + "max_bytes": { + "type": "integer", + "description": "Max response body bytes to read (default: 1048576).", + }, + }, + "required": ["url"], + "additionalProperties": False, + }, + }, +] + +RESOURCES_SCHEMA: list[dict[str, Any]] = [ + { + "uri": "torproxy://status", + "name": "Torproxy Status", + "description": "Live JSON status of the proxy", + "mimeType": "application/json", + }, + { + "uri": "torproxy://metrics", + "name": "Torproxy Metrics", + "description": "Prometheus telemetry metrics text", + "mimeType": "text/plain", + }, +] + + +class McpHandler: + def __init__(self, service: ProxyService, queries: ProxyQueries) -> None: + self.service = service + self.queries = queries + + def handle_jsonrpc(self, req: dict[str, Any]) -> dict[str, Any] | None: + if not isinstance(req, dict) or req.get("jsonrpc") != "2.0": + return { + "jsonrpc": "2.0", + "id": req.get("id") if isinstance(req, dict) else None, + "error": {"code": -32600, "message": "Invalid Request: expected jsonrpc 2.0"}, + } + + req_id = req.get("id") + method = req.get("method") + params = req.get("params") or {} + if not isinstance(params, dict): + params = {} + + if "id" not in req: + return None + + if method == "initialize": + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": { + "tools": {}, + "resources": {}, + }, + "serverInfo": { + "name": "torproxy", + "version": __version__, + }, + }, + } + + if method == "ping": + return {"jsonrpc": "2.0", "id": req_id, "result": {}} + + if method == "tools/list": + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"tools": TOOLS_SCHEMA}, + } + + if method == "tools/call": + tool_name = params.get("name") + args = params.get("arguments") or {} + if not isinstance(args, dict): + args = {} + return self._call_tool(req_id, str(tool_name), args) + + if method == "resources/list": + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"resources": RESOURCES_SCHEMA}, + } + + if method == "resources/read": + uri = params.get("uri") + return self._read_resource(req_id, str(uri)) + + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"}, + } + + def _call_tool(self, req_id: Any, name: str, args: dict[str, Any]) -> dict[str, Any]: + if name == "torproxy_status": + resp = self.queries.status() + text = json.dumps(resp.body) if isinstance(resp.body, dict) else str(resp.body) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": text}], + "isError": resp.status >= 400, + }, + } + + if name == "torproxy_get_ip": + resp = self.queries.ip() + text = json.dumps(resp.body) if isinstance(resp.body, dict) else str(resp.body) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": text}], + "isError": resp.status >= 400, + }, + } + + if name in ("torproxy_reconnect", "torproxy_connect"): + action = "reconnect" if name == "torproxy_reconnect" else "connect" + country = args.get("country") + strict = args.get("strict") + try: + res = self.service.mutate(action, country=country, strict=strict) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": json.dumps(res)}], + "isError": False, + }, + } + except (InvalidInput, ControlFailure) as e: + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"Error: {e}"}], + "isError": True, + }, + } + + if name == "torproxy_disconnect": + try: + res = self.service.mutate("disconnect") + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": json.dumps(res)}], + "isError": False, + }, + } + except (InvalidInput, ControlFailure) as e: + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"Error: {e}"}], + "isError": True, + }, + } + + if name == "torproxy_health": + resp = self.queries.health() + text = json.dumps(resp.body) if isinstance(resp.body, dict) else str(resp.body) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": text}], + "isError": resp.status >= 400, + }, + } + + if name == "torproxy_diagnostics": + resp = self.queries.diagnostics() + text = json.dumps(resp.body) if isinstance(resp.body, dict) else str(resp.body) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": text}], + "isError": resp.status >= 400, + }, + } + + if name == "torproxy_fetch": + return self._call_fetch(req_id, args) + + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"Unknown tool: {name}"}], + "isError": True, + }, + } + + def _call_fetch(self, req_id: Any, args: dict[str, Any]) -> dict[str, Any]: + raw_url = args.get("url") + if not raw_url or not isinstance(raw_url, str): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [ + {"type": "text", "text": "Error: Missing required 'url' parameter"} + ], + "isError": True, + }, + } + + parsed = urllib.parse.urlparse(raw_url) + if parsed.scheme not in ("http", "https"): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [ + { + "type": "text", + "text": ( + f"Error: Unsupported URL scheme '{parsed.scheme}'. " + "Only http and https are allowed." + ), + } + ], + "isError": True, + }, + } + + method = str(args.get("method") or "GET").upper() + headers = args.get("headers") or {} + if not isinstance(headers, dict): + headers = {} + body = args.get("body") + data = body.encode("utf-8") if isinstance(body, str) else None + + try: + timeout = float(args.get("timeout", 30)) + except (ValueError, TypeError): + timeout = 30.0 + + try: + max_bytes = int(args.get("max_bytes", 1048576)) + except (ValueError, TypeError): + max_bytes = 1048576 + + proxy_url = f"http://127.0.0.1:{self.service.settings.proxy_port}" + proxy_handler = urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url}) + opener = urllib.request.build_opener(proxy_handler) + + req = urllib.request.Request(raw_url, data=data, headers=headers, method=method) + + try: + with opener.open(req, timeout=timeout) as response: + status = response.status + resp_headers = dict(response.headers) + raw_content = response.read(max_bytes) + text = raw_content.decode("utf-8", errors="replace") + result_payload = { + "status": status, + "headers": resp_headers, + "body": text, + "url": response.geturl(), + } + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": json.dumps(result_payload)}], + "isError": status >= 400, + }, + } + except urllib.error.HTTPError as err: + err_body = err.read(max_bytes).decode("utf-8", errors="replace") if err.fp else "" + result_payload = { + "status": err.code, + "headers": dict(err.headers), + "body": err_body, + "url": raw_url, + } + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": json.dumps(result_payload)}], + "isError": True, + }, + } + except Exception as err: + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"Fetch error: {err}"}], + "isError": True, + }, + } + + def _read_resource(self, req_id: Any, uri: str) -> dict[str, Any]: + if uri == "torproxy://status": + resp = self.queries.status() + text = ( + json.dumps(resp.body, indent=2) if isinstance(resp.body, dict) else str(resp.body) + ) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "contents": [ + { + "uri": uri, + "mimeType": "application/json", + "text": text, + } + ] + }, + } + + if uri == "torproxy://metrics": + resp = self.queries.metrics() + text = str(resp.body) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "contents": [ + { + "uri": uri, + "mimeType": "text/plain", + "text": text, + } + ] + }, + } + + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown resource URI: {uri}"}, + } diff --git a/src/torproxy/mcp/sse.py b/src/torproxy/mcp/sse.py new file mode 100644 index 0000000..00cc783 --- /dev/null +++ b/src/torproxy/mcp/sse.py @@ -0,0 +1,223 @@ +"""HTTP/SSE transport for Model Context Protocol (MCP).""" + +import json +import queue +import threading +import urllib.parse +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +from torproxy.configuration import Settings +from torproxy.mcp.protocol import McpHandler + + +class McpSseServer(ThreadingHTTPServer): + daemon_threads = True + + def __init__( + self, + server_address: tuple[str, int], + handler: McpHandler, + settings: Settings, + ) -> None: + self.handler = handler + self.settings = settings + self.sessions: dict[str, queue.Queue[str | None]] = {} + self.sessions_lock = threading.Lock() + self._is_shut_down = threading.Event() + super().__init__(server_address, McpSseHandler) + + def create_session(self) -> tuple[str, queue.Queue[str | None]]: + session_id = uuid.uuid4().hex + q: queue.Queue[str | None] = queue.Queue() + with self.sessions_lock: + self.sessions[session_id] = q + return session_id, q + + def get_session(self, session_id: str) -> queue.Queue[str | None] | None: + with self.sessions_lock: + return self.sessions.get(session_id) + + def remove_session(self, session_id: str) -> None: + with self.sessions_lock: + self.sessions.pop(session_id, None) + + def server_close(self) -> None: + self._is_shut_down.set() + with self.sessions_lock: + for q in self.sessions.values(): + q.put(None) + self.sessions.clear() + super().server_close() + + def handle_error(self, request: Any, client_address: Any) -> None: + import sys + + _, exc_value, _ = sys.exc_info() + if isinstance(exc_value, (BrokenPipeError, ConnectionResetError)): + return + super().handle_error(request, client_address) + + +class McpSseHandler(BaseHTTPRequestHandler): + server: McpSseServer + + def _check_auth(self, query_params: dict[str, list[str]]) -> bool: + expected = self.server.settings.api_token + if not expected: + return True + + # 1. Bearer header + auth_header = self.headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + token = auth_header[7:].strip() + if token == expected: + return True + + # 2. X-API-Key header + if self.headers.get("X-API-Key", "").strip() == expected: + return True + + # 3. Query param ?token=... + query_token = query_params.get("token", [""])[0].strip() + if query_token == expected: + return True + + return False + + def _send_cors_headers(self) -> None: + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.send_header( + "Access-Control-Allow-Headers", + "Content-Type, Authorization, X-API-Key", + ) + + def do_OPTIONS(self) -> None: + self.send_response(204) + self._send_cors_headers() + self.send_header("Content-Length", "0") + self.end_headers() + + def do_GET(self) -> None: + parsed = urllib.parse.urlparse(self.path) + qs = urllib.parse.parse_qs(parsed.query) + + if parsed.path != "/sse": + self.send_response(404) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Not Found"}\n') + return + + if not self._check_auth(qs): + self.send_response(401) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Unauthorized"}\n') + return + + session_id, q = self.server.create_session() + self.send_response(200) + self._send_cors_headers() + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "keep-alive") + self.end_headers() + + endpoint_url = f"/messages?sessionId={session_id}" + self.wfile.write(f"event: endpoint\r\ndata: {endpoint_url}\r\n\r\n".encode("utf-8")) + self.wfile.flush() + + try: + while not self.server._is_shut_down.is_set(): + try: + msg = q.get(timeout=15.0) + if msg is None: + break + self.wfile.write(f"event: message\r\ndata: {msg}\r\n\r\n".encode("utf-8")) + self.wfile.flush() + except queue.Empty: + # Keepalive comment + self.wfile.write(b": ping\r\n\r\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError): + # Client disconnected or socket closed; terminate SSE stream gracefully + pass + finally: + self.server.remove_session(session_id) + + def do_POST(self) -> None: + parsed = urllib.parse.urlparse(self.path) + qs = urllib.parse.parse_qs(parsed.query) + + if parsed.path != "/messages": + self.send_response(404) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Not Found"}\n') + return + + if not self._check_auth(qs): + self.send_response(401) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Unauthorized"}\n') + return + + session_id = qs.get("sessionId", [None])[0] + if not session_id: + self.send_response(400) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Missing sessionId query parameter"}\n') + return + + q = self.server.get_session(session_id) + if q is None: + self.send_response(404) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Session not found"}\n') + return + + content_length = int(self.headers.get("Content-Length", 0)) + if content_length > self.server.settings.max_body_bytes: + self.send_response(413) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Payload Too Large"}\n') + return + + body_bytes = self.rfile.read(content_length) + try: + req = json.loads(body_bytes.decode("utf-8")) + except json.JSONDecodeError: + self.send_response(400) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error": "Invalid JSON body"}\n') + return + + resp = self.server.handler.handle_jsonrpc(req) + if resp is not None: + q.put(json.dumps(resp)) + + self.send_response(202) + self._send_cors_headers() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"status": "accepted"}\n') + + def log_message(self, format: str, *args: Any) -> None: + # Suppress noisy standard HTTP access logs + pass diff --git a/src/torproxy/mcp/stdio.py b/src/torproxy/mcp/stdio.py new file mode 100644 index 0000000..0c35488 --- /dev/null +++ b/src/torproxy/mcp/stdio.py @@ -0,0 +1,42 @@ +"""Stdio transport for Model Context Protocol (MCP).""" + +import json +import sys +from typing import TextIO + +from torproxy.mcp.protocol import McpHandler + + +def run_stdio_server( + handler: McpHandler, + in_stream: TextIO | None = None, + out_stream: TextIO | None = None, + err_stream: TextIO | None = None, +) -> None: + stdin = in_stream or sys.stdin + stdout = out_stream or sys.stdout + stderr = err_stream or sys.stderr + + for line in stdin: + stripped = line.strip() + if not stripped: + continue + try: + req = json.loads(stripped) + except json.JSONDecodeError as err: + if stderr: + stderr.write(f"[mcp-stdio] JSON parse error: {err}\n") + stderr.flush() + err_resp = { + "jsonrpc": "2.0", + "id": None, + "error": {"code": -32700, "message": "Parse error"}, + } + stdout.write(json.dumps(err_resp) + "\n") + stdout.flush() + continue + + resp = handler.handle_jsonrpc(req) + if resp is not None: + stdout.write(json.dumps(resp) + "\n") + stdout.flush() diff --git a/src/torproxy/queries.py b/src/torproxy/queries.py index 9cc1ae8..a26779d 100644 --- a/src/torproxy/queries.py +++ b/src/torproxy/queries.py @@ -3,6 +3,7 @@ from datetime import datetime, timezone from typing import Any +from torproxy.metrics import render_metrics from torproxy.models import Response from torproxy.service import ProxyService @@ -184,3 +185,23 @@ def diagnostics(self) -> Response: "egress": {"verified": bool(ip), "ip": ip, "error": ip_error if not ip else None}, }, ) + + def metrics(self) -> Response: + bootstrap, _ = self.service.get_tor_bootstrap_status() + s = self.service.state + with s.lock: + return Response( + 200, + render_metrics( + self.service.state.snapshot(), + bootstrap, + self.service.check_socks5_ready(), + self.service.settings.gateway_mode, + requests_total=s.requests_total, + requests_rejected_total=s.requests_rejected_total, + http_4xx_total=s.http_4xx_total, + http_5xx_total=s.http_5xx_total, + latency_total_seconds=s.request_latency_seconds_total, + ), + "text/plain; version=0.0.4; charset=utf-8", + ) diff --git a/src/torproxy/runtime/cli.py b/src/torproxy/runtime/cli.py index 2fa2785..c5c48ed 100644 --- a/src/torproxy/runtime/cli.py +++ b/src/torproxy/runtime/cli.py @@ -3,10 +3,12 @@ import argparse import os import sys +import threading from torproxy import __version__ -from torproxy.bootstrap import create_server +from torproxy.bootstrap import create_mcp_handler, create_mcp_server, create_server from torproxy.configuration import Settings +from torproxy.mcp.stdio import run_stdio_server def build_parser() -> argparse.ArgumentParser: @@ -20,14 +22,33 @@ def build_parser() -> argparse.ArgumentParser: action="version", version=f"%(prog)s {__version__}", ) + subparsers = parser.add_subparsers(dest="command") + subparsers.add_parser( + "mcp", + help="Run Model Context Protocol (MCP) server over stdio", + ) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() - parser.parse_args(argv) + args = parser.parse_args(argv) + settings = Settings.from_mapping(os.environ) + + if args.command == "mcp": + handler = create_mcp_handler(settings) + run_stdio_server(handler) + return 0 + + server = create_server(settings) + mcp_server = None + if settings.mcp_enabled: + service = getattr(getattr(server, "router", None), "service", None) + handler = create_mcp_handler(settings, service=service) + mcp_server = create_mcp_server(settings, handler=handler) + threading.Thread(target=mcp_server.serve_forever, daemon=True).start() + print(f"Starting MCP SSE server on port {mcp_server.server_address[1]}...", flush=True) - server = create_server(Settings.from_mapping(os.environ)) try: print(f"Starting API server on port {server.server_port}...", flush=True) server.serve_forever() @@ -35,6 +56,9 @@ def main(argv: list[str] | None = None) -> int: # Graceful shutdown initiated by user interrupt (Ctrl+C) or termination signal pass finally: + if mcp_server is not None: + mcp_server.shutdown() + mcp_server.server_close() server.server_close() return 0 diff --git a/src/torproxy/web/router.py b/src/torproxy/web/router.py index 93f101a..9ae1582 100644 --- a/src/torproxy/web/router.py +++ b/src/torproxy/web/router.py @@ -1,6 +1,5 @@ """Translate HTTP paths to application operations.""" -from torproxy.metrics import render_metrics from torproxy.models import Response from torproxy.queries import ProxyQueries from torproxy.service import ControlFailure, InvalidInput, ProxyService @@ -9,15 +8,15 @@ class Router: def __init__(self, service: ProxyService) -> None: self.service = service - queries = ProxyQueries(service) + self.queries = ProxyQueries(service) self.read_routes = { - "": queries.status, - "status": queries.status, - "ip": queries.ip, - "health": queries.health, - "v1/health": queries.health, - "diagnostics": queries.diagnostics, - "v1/diagnostics": queries.diagnostics, + "": self.queries.status, + "status": self.queries.status, + "ip": self.queries.ip, + "health": self.queries.health, + "v1/health": self.queries.health, + "diagnostics": self.queries.diagnostics, + "v1/diagnostics": self.queries.diagnostics, "metrics": self.metrics, "v1/metrics": self.metrics, } @@ -50,21 +49,4 @@ def dispatch(self, path: str, country: object = None, strict: object = None) -> ) def metrics(self) -> Response: - bootstrap, _ = self.service.get_tor_bootstrap_status() - s = self.service.state - with s.lock: - return Response( - 200, - render_metrics( - self.service.state.snapshot(), - bootstrap, - self.service.check_socks5_ready(), - self.service.settings.gateway_mode, - requests_total=s.requests_total, - requests_rejected_total=s.requests_rejected_total, - http_4xx_total=s.http_4xx_total, - http_5xx_total=s.http_5xx_total, - latency_total_seconds=s.request_latency_seconds_total, - ), - "text/plain; version=0.0.4; charset=utf-8", - ) + return self.queries.metrics() diff --git a/tests/integration/verify_container.sh b/tests/integration/verify_container.sh index 2f3c9ea..d157105 100755 --- a/tests/integration/verify_container.sh +++ b/tests/integration/verify_container.sh @@ -71,6 +71,21 @@ echo "legacy GET /reconnect response: $LEGACY_RESP" echo "$LEGACY_RESP" | jq -e '.action == "reconnect" and .status == "success"' >/dev/null echo "✅ REST API contracts verified." +echo "=== 3b. MCP Server Verification ===" +# Test MCP Stdio ping & tools/list +MCP_STDIO_RESP=$(echo '{"jsonrpc":"2.0","id":1,"method":"ping"}' | docker exec -i "$CONTAINER_NAME" python3 -m torproxy mcp 2>/dev/null) +echo "MCP stdio ping response: $MCP_STDIO_RESP" +echo "$MCP_STDIO_RESP" | jq -e '.result == {}' >/dev/null + +MCP_TOOLS_RESP=$(echo '{"jsonrpc":"2.0","id":2,"method":"tools/list"}' | docker exec -i "$CONTAINER_NAME" python3 -m torproxy mcp 2>/dev/null) +echo "$MCP_TOOLS_RESP" | jq -e '.result.tools | map(.name) | contains(["torproxy_fetch", "torproxy_status", "torproxy_reconnect"])' >/dev/null + +# Test MCP SSE endpoint on internal port 2501 +SSE_ENDPOINT_RESP=$(docker exec "$CONTAINER_NAME" curl -fsS --max-time 3 http://127.0.0.1:2501/sse 2>/dev/null | head -n 2 || true) +echo "MCP SSE endpoint response: $SSE_ENDPOINT_RESP" +echo "$SSE_ENDPOINT_RESP" | grep -q "event: endpoint" +echo "✅ MCP Server (stdio & SSE port 2501) verified." + echo "=== 4. SOCKS5 Handshake Verification ===" # Send RFC1928 greeting: \x05\x01\x00 (Version 5, 1 auth method, NO AUTH) # Response must be \x05\x00 diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 918042e..a06fd6f 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -47,3 +47,53 @@ def test_cli_starts_server_with_default_args(self, capsys): mock_server.server_close.assert_called_once() captured = capsys.readouterr() assert "Starting API server on port 8080..." in captured.out + + def test_cli_mcp_subcommand(self): + mock_handler = MagicMock() + with ( + patch( + "torproxy.runtime.cli.create_mcp_handler", return_value=mock_handler + ) as mock_create_handler, + patch("torproxy.runtime.cli.run_stdio_server") as mock_run_stdio, + ): + exit_code = main(["mcp"]) + + assert exit_code == 0 + mock_create_handler.assert_called_once() + mock_run_stdio.assert_called_once_with(mock_handler) + + def test_cli_starts_mcp_sse_server_when_enabled(self, capsys): + mock_server = MagicMock() + mock_server.server_port = 8080 + mock_mcp = MagicMock() + mock_mcp.server_address = ("0.0.0.0", 2501) + + with ( + patch("torproxy.runtime.cli.create_server", return_value=mock_server), + patch( + "torproxy.runtime.cli.create_mcp_server", return_value=mock_mcp + ) as mock_create_mcp, + ): + exit_code = main([]) + + assert exit_code == 0 + mock_create_mcp.assert_called_once() + mock_mcp.server_close.assert_called_once() + captured = capsys.readouterr() + assert "Starting MCP SSE server on port 2501..." in captured.out + + def test_cli_does_not_start_mcp_when_disabled(self, monkeypatch, capsys): + monkeypatch.setenv("TORPROXY_MCP_ENABLED", "false") + mock_server = MagicMock() + mock_server.server_port = 8080 + + with ( + patch("torproxy.runtime.cli.create_server", return_value=mock_server), + patch("torproxy.runtime.cli.create_mcp_server") as mock_create_mcp, + ): + exit_code = main([]) + + assert exit_code == 0 + mock_create_mcp.assert_not_called() + captured = capsys.readouterr() + assert "Starting MCP SSE server" not in captured.out diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index acfe1e1..c9b497f 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -373,3 +373,47 @@ def test_render_torrc_without_gateway_mode(self): rendered = render_torrc(settings, template) assert "TransPort" not in rendered assert "DNSPort" not in rendered + + +@pytest.mark.unit +class TestMcpSettings: + def test_default_mcp_settings(self): + from torproxy.configuration import Settings + + s = Settings.from_mapping({}) + assert s.mcp_enabled is True + assert s.mcp_port == 2501 + assert s.mcp_listen_address == "0.0.0.0" + + def test_custom_mcp_settings(self): + from torproxy.configuration import Settings + + s = Settings.from_mapping( + { + "TORPROXY_MCP_ENABLED": "false", + "TORPROXY_MCP_PORT": "2502", + "TORPROXY_MCP_LISTEN_ADDRESS": "127.0.0.1", + } + ) + assert s.mcp_enabled is False + assert s.mcp_port == 2502 + assert s.mcp_listen_address == "127.0.0.1" + + @pytest.mark.parametrize("invalid_port", ["0", "65536", "-1", "xyz"]) + def test_invalid_mcp_port(self, invalid_port): + from torproxy.configuration import Settings + + with pytest.raises(ValueError): + Settings.from_mapping({"TORPROXY_MCP_PORT": invalid_port}) + + def test_invalid_mcp_listen_address(self): + from torproxy.configuration import Settings + + with pytest.raises(ValueError, match="TORPROXY_MCP_LISTEN_ADDRESS"): + Settings.from_mapping({"TORPROXY_MCP_LISTEN_ADDRESS": "not-an-ip"}) + + def test_mcp_port_collision_with_api_port(self): + from torproxy.configuration import Settings + + with pytest.raises(ValueError, match="distinct ports"): + Settings.from_mapping({"TORPROXY_MCP_PORT": "8080"}) diff --git a/tests/unit/test_mcp_protocol.py b/tests/unit/test_mcp_protocol.py new file mode 100644 index 0000000..1218df5 --- /dev/null +++ b/tests/unit/test_mcp_protocol.py @@ -0,0 +1,395 @@ +import json + +import pytest + +from torproxy import __version__ +from torproxy.bootstrap import create_services +from torproxy.configuration import Settings +from torproxy.queries import ProxyQueries + + +class FakeControl: + def __init__(self): + self.commands = [] + self.fail_on = None + + def send(self, command): + self.commands.append(command) + if command == self.fail_on: + return None, "command failed" + if command == "GETINFO status/bootstrap-phase": + return '250-status/bootstrap-phase=PROGRESS=100 TAG=done SUMMARY="Done"\r\n250 OK', None + if command == "GETCONF DisableNetwork": + return "DisableNetwork=0", None + return "250 OK", None + + +class FakeEgress: + def resolve(self): + return "198.51.100.42", None + + +class FakeSocks: + def ready(self): + return True + + +@pytest.fixture +def mcp_setup(): + from torproxy.mcp.protocol import McpHandler + + control = FakeControl() + egress = FakeEgress() + socks = FakeSocks() + settings = Settings() + service = create_services(settings, control=control, egress=egress, socks=socks) + queries = ProxyQueries(service) + handler = McpHandler(service, queries) + return handler, service, control + + +@pytest.mark.unit +def test_mcp_initialize(mcp_setup): + handler, _, _ = mcp_setup + req = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test-client", "version": "1.0"}, + }, + } + resp = handler.handle_jsonrpc(req) + assert resp is not None + assert resp["jsonrpc"] == "2.0" + assert resp["id"] == 1 + assert "result" in resp + result = resp["result"] + assert result["protocolVersion"] == "2024-11-05" + assert "tools" in result["capabilities"] + assert "resources" in result["capabilities"] + assert result["serverInfo"]["name"] == "torproxy" + assert result["serverInfo"]["version"] == __version__ + + +@pytest.mark.unit +def test_mcp_ping(mcp_setup): + handler, _, _ = mcp_setup + req = {"jsonrpc": "2.0", "id": "p-1", "method": "ping"} + resp = handler.handle_jsonrpc(req) + assert resp == {"jsonrpc": "2.0", "id": "p-1", "result": {}} + + +@pytest.mark.unit +def test_mcp_notification_returns_none(mcp_setup): + handler, _, _ = mcp_setup + req = {"jsonrpc": "2.0", "method": "notifications/initialized"} + resp = handler.handle_jsonrpc(req) + assert resp is None + + +@pytest.mark.unit +def test_mcp_tools_list(mcp_setup): + handler, _, _ = mcp_setup + req = {"jsonrpc": "2.0", "id": 2, "method": "tools/list"} + resp = handler.handle_jsonrpc(req) + assert resp is not None + tools = resp["result"]["tools"] + tool_names = {t["name"] for t in tools} + expected = { + "torproxy_status", + "torproxy_get_ip", + "torproxy_reconnect", + "torproxy_connect", + "torproxy_disconnect", + "torproxy_health", + "torproxy_diagnostics", + } + assert expected.issubset(tool_names) + for tool in tools: + assert "description" in tool + assert "inputSchema" in tool + assert tool["inputSchema"]["type"] == "object" + + +@pytest.mark.unit +def test_mcp_tools_call_status(mcp_setup): + handler, _, _ = mcp_setup + req = { + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": {"name": "torproxy_status", "arguments": {}}, + } + resp = handler.handle_jsonrpc(req) + assert resp is not None + assert resp["id"] == 3 + result = resp["result"] + assert result["isError"] is False + assert len(result["content"]) == 1 + content = json.loads(result["content"][0]["text"]) + assert content["status"] == "connected" + assert content["ip"] == "198.51.100.42" + + +@pytest.mark.unit +def test_mcp_tools_call_get_ip(mcp_setup): + handler, _, _ = mcp_setup + req = { + "jsonrpc": "2.0", + "id": 4, + "method": "tools/call", + "params": {"name": "torproxy_get_ip", "arguments": {}}, + } + resp = handler.handle_jsonrpc(req) + result = resp["result"] + assert result["isError"] is False + content = json.loads(result["content"][0]["text"]) + assert content["ip"] == "198.51.100.42" + + +@pytest.mark.unit +def test_mcp_tools_call_reconnect(mcp_setup): + handler, service, control = mcp_setup + req = { + "jsonrpc": "2.0", + "id": 5, + "method": "tools/call", + "params": { + "name": "torproxy_reconnect", + "arguments": {"country": "de", "strict": True}, + }, + } + resp = handler.handle_jsonrpc(req) + result = resp["result"] + assert result["isError"] is False + content = json.loads(result["content"][0]["text"]) + assert content["action"] == "reconnect" + assert content["status"] == "success" + assert content["exit_country"] == "DE" + assert content["strict_nodes"] is True + assert "SIGNAL NEWNYM" in control.commands + + +@pytest.mark.unit +def test_mcp_tools_call_reconnect_invalid_country(mcp_setup): + handler, _, _ = mcp_setup + req = { + "jsonrpc": "2.0", + "id": 6, + "method": "tools/call", + "params": { + "name": "torproxy_reconnect", + "arguments": {"country": "INVALID_CODE"}, + }, + } + resp = handler.handle_jsonrpc(req) + result = resp["result"] + assert result["isError"] is True + assert "Invalid ISO country code" in result["content"][0]["text"] + + +@pytest.mark.unit +def test_mcp_tools_call_connect_and_disconnect(mcp_setup): + handler, _, control = mcp_setup + req_disc = { + "jsonrpc": "2.0", + "id": 7, + "method": "tools/call", + "params": {"name": "torproxy_disconnect", "arguments": {}}, + } + resp_disc = handler.handle_jsonrpc(req_disc) + assert resp_disc["result"]["isError"] is False + assert "SETCONF DisableNetwork=1" in control.commands + + req_conn = { + "jsonrpc": "2.0", + "id": 8, + "method": "tools/call", + "params": {"name": "torproxy_connect", "arguments": {}}, + } + resp_conn = handler.handle_jsonrpc(req_conn) + assert resp_conn["result"]["isError"] is False + assert "SETCONF DisableNetwork=0" in control.commands + + +@pytest.mark.unit +def test_mcp_tools_call_health_and_diagnostics(mcp_setup): + handler, _, _ = mcp_setup + for tool_name in ("torproxy_health", "torproxy_diagnostics"): + req = { + "jsonrpc": "2.0", + "id": 9, + "method": "tools/call", + "params": {"name": tool_name, "arguments": {}}, + } + resp = handler.handle_jsonrpc(req) + assert resp["result"]["isError"] is False + assert len(resp["result"]["content"]) == 1 + + +@pytest.mark.unit +def test_mcp_tools_call_unknown_tool(mcp_setup): + handler, _, _ = mcp_setup + req = { + "jsonrpc": "2.0", + "id": 10, + "method": "tools/call", + "params": {"name": "non_existent_tool", "arguments": {}}, + } + resp = handler.handle_jsonrpc(req) + assert resp["result"]["isError"] is True + assert "Unknown tool" in resp["result"]["content"][0]["text"] + + +@pytest.mark.unit +def test_mcp_resources_list_and_read(mcp_setup): + handler, _, _ = mcp_setup + list_req = {"jsonrpc": "2.0", "id": 11, "method": "resources/list"} + list_resp = handler.handle_jsonrpc(list_req) + resources = list_resp["result"]["resources"] + uris = {r["uri"] for r in resources} + assert "torproxy://status" in uris + assert "torproxy://metrics" in uris + + # Read torproxy://status + read_req = { + "jsonrpc": "2.0", + "id": 12, + "method": "resources/read", + "params": {"uri": "torproxy://status"}, + } + read_resp = handler.handle_jsonrpc(read_req) + contents = read_resp["result"]["contents"] + assert contents[0]["uri"] == "torproxy://status" + assert contents[0]["mimeType"] == "application/json" + status_data = json.loads(contents[0]["text"]) + assert "status" in status_data + + # Read torproxy://metrics + read_metrics = { + "jsonrpc": "2.0", + "id": 13, + "method": "resources/read", + "params": {"uri": "torproxy://metrics"}, + } + metrics_resp = handler.handle_jsonrpc(read_metrics) + assert metrics_resp["result"]["contents"][0]["mimeType"] == "text/plain" + + # Read unknown resource + unknown_req = { + "jsonrpc": "2.0", + "id": 14, + "method": "resources/read", + "params": {"uri": "torproxy://invalid"}, + } + unknown_resp = handler.handle_jsonrpc(unknown_req) + assert "error" in unknown_resp + assert unknown_resp["error"]["code"] == -32602 + + +@pytest.mark.unit +def test_mcp_unknown_method_error(mcp_setup): + handler, _, _ = mcp_setup + req = {"jsonrpc": "2.0", "id": 99, "method": "unknown_rpc_method"} + resp = handler.handle_jsonrpc(req) + assert "error" in resp + assert resp["error"]["code"] == -32601 + + +@pytest.mark.unit +def test_mcp_tools_call_fetch_missing_or_invalid_url(mcp_setup): + handler, _, _ = mcp_setup + # Missing url + req1 = { + "jsonrpc": "2.0", + "id": 20, + "method": "tools/call", + "params": {"name": "torproxy_fetch", "arguments": {}}, + } + resp1 = handler.handle_jsonrpc(req1) + assert resp1["result"]["isError"] is True + assert "Missing required 'url'" in resp1["result"]["content"][0]["text"] + + # Invalid scheme + req2 = { + "jsonrpc": "2.0", + "id": 21, + "method": "tools/call", + "params": {"name": "torproxy_fetch", "arguments": {"url": "file:///etc/passwd"}}, + } + resp2 = handler.handle_jsonrpc(req2) + assert resp2["result"]["isError"] is True + assert "Unsupported URL scheme" in resp2["result"]["content"][0]["text"] + + +@pytest.mark.unit +def test_mcp_tools_call_fetch_success(mcp_setup, monkeypatch): + from unittest.mock import MagicMock + + handler, _, _ = mcp_setup + + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.read.return_value = b'{"origin": "198.51.100.1"}' + mock_resp.geturl.return_value = "https://example.com/api" + mock_resp.__enter__.return_value = mock_resp + mock_resp.__exit__.return_value = None + + mock_opener = MagicMock() + mock_opener.open.return_value = mock_resp + + monkeypatch.setattr("urllib.request.build_opener", lambda *args: mock_opener) + + req = { + "jsonrpc": "2.0", + "id": 22, + "method": "tools/call", + "params": { + "name": "torproxy_fetch", + "arguments": { + "url": "https://example.com/api", + "method": "GET", + "headers": {"User-Agent": "agent/1.0"}, + }, + }, + } + resp = handler.handle_jsonrpc(req) + assert resp["result"]["isError"] is False + content = json.loads(resp["result"]["content"][0]["text"]) + assert content["status"] == 200 + assert "198.51.100.1" in content["body"] + assert content["url"] == "https://example.com/api" + + +@pytest.mark.unit +def test_mcp_tools_call_fetch_http_error(mcp_setup, monkeypatch): + import io + import urllib.error + from unittest.mock import MagicMock + + handler, _, _ = mcp_setup + + fp = io.BytesIO(b"Not Found") + http_err = urllib.error.HTTPError( + "https://example.com/404", 404, "Not Found", {"Content-Type": "text/plain"}, fp + ) + + mock_opener = MagicMock() + mock_opener.open.side_effect = http_err + monkeypatch.setattr("urllib.request.build_opener", lambda *args: mock_opener) + + req = { + "jsonrpc": "2.0", + "id": 23, + "method": "tools/call", + "params": {"name": "torproxy_fetch", "arguments": {"url": "https://example.com/404"}}, + } + resp = handler.handle_jsonrpc(req) + assert resp["result"]["isError"] is True + content = json.loads(resp["result"]["content"][0]["text"]) + assert content["status"] == 404 + assert content["body"] == "Not Found" diff --git a/tests/unit/test_mcp_sse.py b/tests/unit/test_mcp_sse.py new file mode 100644 index 0000000..f0cc5e4 --- /dev/null +++ b/tests/unit/test_mcp_sse.py @@ -0,0 +1,232 @@ +import json +import threading +import urllib.error +import urllib.request + +import pytest + +from torproxy.bootstrap import create_services +from torproxy.configuration import Settings +from torproxy.mcp.protocol import McpHandler +from torproxy.mcp.sse import McpSseServer +from torproxy.queries import ProxyQueries + + +class DummyControl: + def send(self, command): + if command == "GETINFO status/bootstrap-phase": + return '250-status/bootstrap-phase=PROGRESS=100 TAG=done SUMMARY="Done"\r\n250 OK', None + if command == "GETCONF DisableNetwork": + return "DisableNetwork=0", None + return "250 OK", None + + +class DummyEgress: + def resolve(self): + return "198.51.100.42", None + + +class DummySocks: + def ready(self): + return True + + +@pytest.fixture +def running_sse_server(): + def _create(settings_override=None): + settings_dict = {"TORPROXY_MCP_PORT": "2501", "TORPROXY_MCP_LISTEN_ADDRESS": "127.0.0.1"} + if settings_override: + settings_dict.update(settings_override) + settings = Settings.from_mapping(settings_dict) + service = create_services( + settings, control=DummyControl(), egress=DummyEgress(), socks=DummySocks() + ) + queries = ProxyQueries(service) + handler = McpHandler(service, queries) + server = McpSseServer(("127.0.0.1", 0), handler, settings) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + base_url = f"http://127.0.0.1:{port}" + return server, base_url + + created_servers = [] + + def factory(settings_override=None): + srv, url = _create(settings_override) + created_servers.append(srv) + return srv, url + + yield factory + + for srv in created_servers: + srv.shutdown() + srv.server_close() + + +@pytest.mark.unit +def test_sse_options_cors(running_sse_server): + _, base_url = running_sse_server() + req = urllib.request.Request(f"{base_url}/sse", method="OPTIONS") + with urllib.request.urlopen(req) as resp: + assert resp.status in (200, 204) + assert resp.headers.get("Access-Control-Allow-Origin") == "*" + assert "POST" in resp.headers.get("Access-Control-Allow-Methods", "") + + +@pytest.mark.unit +def test_sse_endpoint_and_message_flow(running_sse_server): + _, base_url = running_sse_server() + + # 1. Connect to /sse stream + req = urllib.request.Request(f"{base_url}/sse") + resp = urllib.request.urlopen(req) + assert resp.headers.get("Content-Type") == "text/event-stream" + + # Read until endpoint event is received + line1 = resp.readline().decode().strip() + line2 = resp.readline().decode().strip() + resp.readline() # empty separator line + + assert line1 == "event: endpoint" + assert line2.startswith("data: /messages?sessionId=") + endpoint_path = line2.split("data: ")[1].strip() + + # 2. Send JSON-RPC ping to /messages?sessionId=... + post_url = f"{base_url}{endpoint_path}" + body = json.dumps({"jsonrpc": "2.0", "id": "msg-1", "method": "ping"}).encode() + post_req = urllib.request.Request( + post_url, data=body, headers={"Content-Type": "application/json"}, method="POST" + ) + with urllib.request.urlopen(post_req) as post_resp: + assert post_resp.status in (200, 202) + + # 3. Read message response on SSE stream + ev_line = resp.readline().decode().strip() + data_line = resp.readline().decode().strip() + resp.readline() # separator line + + assert ev_line == "event: message" + msg_data = json.loads(data_line.split("data: ")[1]) + assert msg_data["id"] == "msg-1" + assert msg_data["result"] == {} + + resp.close() + + +@pytest.mark.unit +def test_sse_authentication(running_sse_server): + _, base_url = running_sse_server({"TORPROXY_API_TOKEN": "topsecret"}) + + # Missing token -> 401 + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(f"{base_url}/sse") + assert exc_info.value.code == 401 + + # Valid token via Bearer header -> 200 + req = urllib.request.Request(f"{base_url}/sse", headers={"Authorization": "Bearer topsecret"}) + with urllib.request.urlopen(req) as resp: + assert resp.status == 200 + resp.close() + + # Valid token via Query Parameter ?token=topsecret -> 200 + req_query = urllib.request.Request(f"{base_url}/sse?token=topsecret") + with urllib.request.urlopen(req_query) as resp: + assert resp.status == 200 + resp.close() + + +@pytest.mark.unit +def test_sse_unknown_session_returns_404(running_sse_server): + _, base_url = running_sse_server() + body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "ping"}).encode() + post_req = urllib.request.Request( + f"{base_url}/messages?sessionId=non-existent-session", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(post_req) + assert exc_info.value.code == 404 + + +@pytest.mark.unit +def test_sse_invalid_paths_return_404(running_sse_server): + _, base_url = running_sse_server() + + # GET invalid path + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(f"{base_url}/unknown") + assert exc_info.value.code == 404 + + # POST invalid path + post_req = urllib.request.Request( + f"{base_url}/unknown", + data=b"{}", + headers={"Content-Type": "application/json"}, + method="POST", + ) + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(post_req) + assert exc_info.value.code == 404 + + +@pytest.mark.unit +def test_sse_post_missing_session_id(running_sse_server): + _, base_url = running_sse_server() + post_req = urllib.request.Request( + f"{base_url}/messages", + data=b"{}", + headers={"Content-Type": "application/json"}, + method="POST", + ) + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(post_req) + assert exc_info.value.code == 400 + + +@pytest.mark.unit +def test_sse_post_invalid_json(running_sse_server): + _, base_url = running_sse_server() + + # Get a valid session + stream_resp = urllib.request.urlopen(f"{base_url}/sse") + stream_resp.readline() + line2 = stream_resp.readline().decode().strip() + stream_resp.readline() + endpoint = line2.split("data: ")[1].strip() + + post_req = urllib.request.Request( + f"{base_url}{endpoint}", + data=b"{not valid json", + headers={"Content-Type": "application/json"}, + method="POST", + ) + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(post_req) + assert exc_info.value.code == 400 + stream_resp.close() + + +@pytest.mark.unit +def test_sse_post_payload_too_large(running_sse_server): + _, base_url = running_sse_server({"TORPROXY_MAX_BODY_BYTES": "100"}) + + stream_resp = urllib.request.urlopen(f"{base_url}/sse") + stream_resp.readline() + line2 = stream_resp.readline().decode().strip() + stream_resp.readline() + endpoint = line2.split("data: ")[1].strip() + + large_body = json.dumps({"jsonrpc": "2.0", "data": "x" * 200}).encode() + post_req = urllib.request.Request( + f"{base_url}{endpoint}", + data=large_body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with pytest.raises(urllib.error.HTTPError) as exc_info: + urllib.request.urlopen(post_req) + assert exc_info.value.code == 413 + stream_resp.close() diff --git a/tests/unit/test_mcp_stdio.py b/tests/unit/test_mcp_stdio.py new file mode 100644 index 0000000..5f1174a --- /dev/null +++ b/tests/unit/test_mcp_stdio.py @@ -0,0 +1,80 @@ +import io +import json + +import pytest + +from torproxy.bootstrap import create_services +from torproxy.configuration import Settings +from torproxy.mcp.protocol import McpHandler +from torproxy.mcp.stdio import run_stdio_server +from torproxy.queries import ProxyQueries + + +class DummyControl: + def send(self, command): + if command == "GETINFO status/bootstrap-phase": + return '250-status/bootstrap-phase=PROGRESS=100 TAG=done SUMMARY="Done"\r\n250 OK', None + if command == "GETCONF DisableNetwork": + return "DisableNetwork=0", None + return "250 OK", None + + +class DummyEgress: + def resolve(self): + return "198.51.100.99", None + + +class DummySocks: + def ready(self): + return True + + +@pytest.fixture +def stdio_handler(): + settings = Settings() + service = create_services( + settings, control=DummyControl(), egress=DummyEgress(), socks=DummySocks() + ) + queries = ProxyQueries(service) + return McpHandler(service, queries) + + +@pytest.mark.unit +def test_stdio_server_single_request(stdio_handler): + in_buf = io.StringIO(json.dumps({"jsonrpc": "2.0", "id": 1, "method": "ping"}) + "\n") + out_buf = io.StringIO() + err_buf = io.StringIO() + + run_stdio_server(stdio_handler, in_stream=in_buf, out_stream=out_buf, err_stream=err_buf) + + lines = [line.strip() for line in out_buf.getvalue().splitlines() if line.strip()] + assert len(lines) == 1 + resp = json.loads(lines[0]) + assert resp == {"jsonrpc": "2.0", "id": 1, "result": {}} + + +@pytest.mark.unit +def test_stdio_server_ignores_notification_output(stdio_handler): + in_buf = io.StringIO( + json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) + "\n" + ) + out_buf = io.StringIO() + err_buf = io.StringIO() + + run_stdio_server(stdio_handler, in_stream=in_buf, out_stream=out_buf, err_stream=err_buf) + + assert out_buf.getvalue().strip() == "" + + +@pytest.mark.unit +def test_stdio_server_malformed_json(stdio_handler): + in_buf = io.StringIO("{not valid json\n") + out_buf = io.StringIO() + err_buf = io.StringIO() + + run_stdio_server(stdio_handler, in_stream=in_buf, out_stream=out_buf, err_stream=err_buf) + + lines = [line.strip() for line in out_buf.getvalue().splitlines() if line.strip()] + assert len(lines) == 1 + resp = json.loads(lines[0]) + assert resp["error"]["code"] == -32700 # Parse error diff --git a/uv.lock b/uv.lock index 6900319..8775884 100644 --- a/uv.lock +++ b/uv.lock @@ -842,7 +842,7 @@ wheels = [ [[package]] name = "torproxy" -version = "2.5.0" +version = "2.6.0" source = { editable = "." } [package.optional-dependencies]