Merge pull request #6 from ccronca/feat/stdio-transport-support

Add stdio transport support

Author: sidhpurwala-huzaifa

README.md

@@ -1,6 +1,6 @@
 MCP Security Scanner
 
-This is a Python-based penetration testing tool for Model Context Protocol (MCP) servers. It supports Streamable HTTP and SSE transports, runs a suite of checks mapped to `scanner_specs.schema` (auth, transport, tools, prompts, resources), and includes a deliberately insecure MCP-like server for testing.
+This is a Python-based penetration testing tool for Model Context Protocol (MCP) servers. It supports HTTP, stdio, and experimental SSE transports, runs a suite of checks mapped to `scanner_specs.schema` (auth, transport, tools, prompts, resources), and includes a deliberately insecure MCP-like server for testing.
 
 **Note: SSE transport is discontinued in the latest version of MCP. Support for SSE in this tool is purely experimental and may not work!!!**
 
@@ -54,7 +54,7 @@ insecure-mcp-server --host 127.0.0.1 --port 9001
 insecure-mcp-server --host 127.0.0.1 --port 9001 --test 0/1/2/3/4/5/6/7
 ```
 
-### Scan the server (HTTP or SSE)
+### Scan the server (HTTP, stdio, or SSE)
 ```bash
 # HTTP: Text report (no discovery; --url is the JSON-RPC endpoint)
 mcp-scan scan --url http://127.0.0.1:9001/mcp --format text
@@ -65,16 +65,21 @@ mcp-scan scan --url http://127.0.0.1:9001/mcp --format json --output report.json
 # HTTP: Verbose tracing (real-time)
 mcp-scan scan --url http://127.0.0.1:9001/mcp --verbose
 
+# stdio: Scan local MCP servers via stdin/stdout
+mcp-scan scan --transport stdio --command "npx -y @modelcontextprotocol/server-memory" --format json
+
 # SSE: connect to explicit SSE endpoint, then scan via emitted /messages?sessionId=...
 mcp-scan scan --url https://your-mcp.example.com --transport sse --sse-endpoint /sse --timeout 30 --verbose
 ```
 
 ### New: RPC passthrough (Inspector-like)
+**Note: RPC commands only support HTTP and SSE transports, not stdio.**
+
 ```bash
-# List tools
+# List tools (HTTP)
 mcp-scan rpc --url https://your-mcp.example.com/mcp --method tools/list --transport http
 
-# Call a tool
+# Call a tool (HTTP)
 mcp-scan rpc --url https://your-mcp.example.com/mcp \
   --method tools/call \
   --params '{"name":"weather","arguments":{"city":"Paris"}}' \
@@ -99,14 +104,17 @@ mcp-scan scan --url https://your-mcp.example.com/mcp --explain X-01
 
 ### Only health
 - `--only-health` prints server details and enumerations without running the full scan.
-- Works for HTTP and SSE (SSE uses the provided endpoint and the stream-emitted POST path).
+- Works for HTTP, stdio, and SSE transports.
 - Supports `--format text` and `--format json`.
 
 Examples:
 ```bash
 # HTTP
 mcp-scan scan --url https://your-mcp.example.com/mcp --only-health --format text
 
+# stdio
+mcp-scan scan --transport stdio --command "npx -y @modelcontextprotocol/server-memory" --only-health --format json
+
 # SSE
 mcp-scan scan --url https://your-mcp.example.com --transport sse --sse-endpoint /sse --only-health --format json
 ```
@@ -129,11 +137,35 @@ mcp-scan scan \
 ```
 
 ### Transport, timeouts, session
-- **--transport auto|http|sse**: Hint preferred transport; no dynamic discovery. Provide working URLs.
+- **--transport auto|http|stdio|sse**: Hint preferred transport; no dynamic discovery.
+  - `http`: Requires `--url` for JSON-RPC endpoint
+  - `stdio`: Requires `--command` for local MCP server process
+  - `sse`: Requires `--url` and `--sse-endpoint` (experimental)
 - **--timeout <seconds>**: Per-request read timeout (default 12s). Increase for slow streams.
 - **--session-id <SID>**: Pre-established session (`Mcp-Session-Id` header).
 
 
+## Testing
+
+### Running Tests
+```bash
+# Set up environment (if not already done)
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+pip install -e .
+
+# Run all tests
+python -m pytest tests/ -v
+
+# Run specific test module
+python -m pytest tests/test_stdio_scanner.py -v
+python -m pytest tests/test_security_checks.py -v
+
+# Run specific test class
+python -m pytest tests/test_stdio_scanner.py::TestStdioIntegration -v
+```
+
 ### Running with Container
 
 ```bash

src/mcp_scanner/cli.py

@@ -12,6 +12,7 @@
 from .models import Report
 from .spec import load_spec
 from .http_checks import run_full_http_checks, scan_http_base, get_server_health, rpc_call
+from .stdio_scanner import scan_stdio, get_stdio_health
 from .auth import build_auth_headers
 import httpx
 
@@ -21,16 +22,16 @@
 
 @click.group()
 def main() -> None:
-    """MCP Security Scanner CLI (HTTP-only)."""
+    """MCP Security Scanner CLI."""
 
 
 @main.command("scan")
-@click.option("--url", required=True, help="Target MCP server base URL (http:// or https://)")
+@click.option("--url", help="Target MCP server base URL (http:// or https://), not required for stdio transport")
 @click.option("--spec", type=click.Path(exists=True, dir_okay=False), help="Path to scanner_specs.schema")
 @click.option("--format", "fmt", type=click.Choice(["text", "json"]), default="text")
 @click.option("--verbose", is_flag=True, default=False, help="Print full request/response trace and leaked data")
 @click.option("--explain", "explain_id", help="Explain a specific finding by ID (e.g., X-01)")
-@click.option("--transport", type=click.Choice(["auto", "http", "sse"]), default="auto", show_default=True, help="Preferred transport hint; auto tries SSE when available")
+@click.option("--transport", type=click.Choice(["auto", "http", "sse", "stdio"]), default="auto", show_default=True, help="Preferred transport hint; auto tries SSE when available")
 @click.option("--only-health", is_flag=True, default=False, help="Dump endpoints, tools, prompts, resources and exit (no scan)")
 @click.option("--sse-endpoint", help="When --transport sse, append this path to --url for SSE (e.g., /sse)")
 @click.option("--auth-type", type=click.Choice(["bearer", "oauth2-client-credentials"]))
@@ -42,10 +43,24 @@ def main() -> None:
 @click.option("--output", type=click.Path(dir_okay=False), help="Write report to file")
 @click.option("--timeout", type=float, default=12.0, show_default=True, help="Per-request read timeout in seconds")
 @click.option("--session-id", help="Pre-supplied session id to include in Mcp-Session-Id header")
-def scan_cmd(url: str, spec: Optional[str], fmt: str, verbose: bool, explain_id: Optional[str], auth_type: Optional[str], auth_token: Optional[str], token_url: Optional[str], client_id: Optional[str], client_secret: Optional[str], scope: Optional[str], output: Optional[str], timeout: float, session_id: Optional[str], transport: str, only_health: bool, sse_endpoint: Optional[str]) -> None:
+@click.option("--command", help="Command to run MCP server (required for stdio transport)")
+def scan_cmd(url: str, spec: Optional[str], fmt: str, verbose: bool, explain_id: Optional[str], auth_type: Optional[str], auth_token: Optional[str], token_url: Optional[str], client_id: Optional[str], client_secret: Optional[str], scope: Optional[str], output: Optional[str], timeout: float, session_id: Optional[str], transport: str, only_health: bool, sse_endpoint: Optional[str], command: Optional[str]) -> None:
     if verbose and explain_id:
         console.print("--verbose and --explain are mutually exclusive; using --explain.")
         verbose = False
+
+    # Validate stdio transport requirements
+    if transport == "stdio":
+        if not command:
+            raise click.ClickException("--command is required when using --transport stdio")
+        if url and url != "stdio://command":
+            console.print("Note: --url is ignored when using stdio transport, using provided --command instead")
+    else:
+        if not url:
+            raise click.ClickException("--url is required when not using stdio transport")
+        if command:
+            raise click.ClickException("--command can only be used with --transport stdio")
+
     if transport == "sse":
         console.print("SSE is deprecated in MCP!!! SSE support in the scanner is experimental and may not work!!!")
     class RealtimeTrace:
@@ -104,14 +119,15 @@ def __iter__(self) -> Iterator[Dict[str, Any]]:
     if session_id:
         auth_headers = {**auth_headers, "Mcp-Session-Id": session_id}
 
-    # Preflight reachability check
-    if not (url.lower().startswith("http://") or url.lower().startswith("https://")):
-        raise click.ClickException("--url must start with http:// or https://")
-    try:
-        with httpx.Client(follow_redirects=True, timeout=httpx.Timeout(connect=3.0, read=timeout, write=timeout, pool=timeout)) as _c:
-            _c.get(url, timeout=httpx.Timeout(connect=3.0, read=timeout, write=timeout, pool=timeout))
-    except httpx.RequestError as e:  # noqa: PERF203
-        raise click.ClickException(f"Cannot reach MCP server at {url}: {type(e).__name__}: {e}")
+    # Preflight reachability check for HTTP/HTTPS transports only
+    if transport != "stdio":
+        if not (url.lower().startswith("http://") or url.lower().startswith("https://")):
+            raise click.ClickException("--url must start with http:// or https://")
+        try:
+            with httpx.Client(follow_redirects=True, timeout=httpx.Timeout(connect=3.0, read=timeout, write=timeout, pool=timeout)) as _c:
+                _c.get(url, timeout=httpx.Timeout(connect=3.0, read=timeout, write=timeout, pool=timeout))
+        except httpx.RequestError as e:  # noqa: PERF203
+            raise click.ClickException(f"Cannot reach MCP server at {url}: {type(e).__name__}: {e}")
 
     spec_file = Path(spec) if spec else None
     if spec_file is not None:
@@ -125,23 +141,36 @@ def __iter__(self) -> Iterator[Dict[str, Any]]:
     elif transport == "http" and "Accept" not in auth_headers:
         auth_headers = {**auth_headers, "Accept": "application/json, text/event-stream"}
     if only_health:
-        health = get_server_health(url, headers=auth_headers, trace=trace, verbose=verbose, timeout=timeout, transport=transport, sse_endpoint=sse_endpoint)
+        if transport == "stdio":
+            health = get_stdio_health(command)
+        else:
+            health = get_server_health(url, headers=auth_headers, trace=trace, verbose=verbose, timeout=timeout, transport=transport, sse_endpoint=sse_endpoint)
         if fmt == "json":
             console.rule("Health (JSON)")
             console.print_json(json.dumps(health))
             return
         # Text output
         console.rule("Health")
-        base = health.get("base_url")
-        msg_url = health.get("msg_url")
-        sse_url = health.get("sse_url")
+
+        # Handle stdio vs HTTP health data
+        if transport == "stdio":
+            target = health.get("target", "stdio")
+            console.print(f"Target: {target}")
+            console.print(f"Transport: {health.get('transport', 'stdio')}")
+            if "error" in health:
+                console.print(f"[red]Error: {health['error']}[/red]")
+        else:
+            base = health.get("base_url")
+            msg_url = health.get("msg_url")
+            sse_url = health.get("sse_url")
+            console.print(f"Base URL: {base}")
+            console.print(f"Message endpoint: {msg_url}")
+            console.print(f"SSE URL: {sse_url}")
+
         init_obj = health.get("initialize") or {}
         tools = health.get("tools") or []
         prompts = health.get("prompts") or []
         resources = health.get("resources") or []
-        console.print(f"Base URL: {base}")
-        console.print(f"Message endpoint: {msg_url}")
-        console.print(f"SSE URL: {sse_url}")
         if isinstance(init_obj, dict):
             keys = list((init_obj.get("result") or {}).keys()) if "result" in init_obj else list(init_obj.keys())
             console.print(f"Initialize keys: {keys}")
@@ -178,8 +207,13 @@ def __iter__(self) -> Iterator[Dict[str, Any]]:
             rtable.add_row("-", "No resources discovered", "")
         console.print(rtable)
         return
-    findings = run_full_http_checks(url, spec_index, headers=auth_headers, trace=trace, verbose=verbose, timeout=timeout, transport=transport, sse_endpoint=sse_endpoint)
-    report = Report.new(target=url, findings=findings)
+
+    # Run scanning based on transport type
+    if transport == "stdio":
+        report = scan_stdio(command, spec_index)
+    else:
+        findings = run_full_http_checks(url, spec_index, headers=auth_headers, trace=trace, verbose=verbose, timeout=timeout, transport=transport, sse_endpoint=sse_endpoint)
+        report = Report.new(target=url, findings=findings)
 
     if only_health:
         return