mcp_security_scanner.py

#!/usr/bin/env python3
"""
MCP Server Security Scanner — Detect common vulnerabilities in MCP server configurations and code.

Checks for:
- Missing authentication (CVE-2026-32211 pattern)
- Network binding to 0.0.0.0 (exposed to local network)
- Hardcoded API keys / secrets
- Unsafe subprocess usage (command injection risk)
- Path traversal vulnerabilities
- Missing input validation on tool parameters

Usage:
    py mcp_security_scanner.py /path/to/mcp-server/
    py mcp_security_scanner.py /path/to/mcp-server/ --verbose
    py mcp_security_scanner.py /path/to/mcp-server/ --json
    py mcp_security_scanner.py --check-config mcp_config.json
"""

import argparse
import json
import os
import re
import sys
from pathlib import Path
from typing import Any


# --- Vulnerability Patterns ---

# Hardcoded secrets patterns
SECRET_PATTERNS = [
    (r'(?i)(api[_-]?key|apikey)\s*[=:]\s*["\'][a-zA-Z0-9_\-]{20,}["\']', "Hardcoded API key"),
    (r'(?i)(secret|token|password|passwd|pwd)\s*[=:]\s*["\'][^\s"\']{8,}["\']', "Hardcoded secret/token"),
    (r'(?i)(sk-[a-zA-Z0-9]{20,})', "OpenAI API key pattern"),
    (r'(?i)(anthropic[_-]?api[_-]?key)\s*[=:]\s*["\']', "Anthropic API key reference"),
    (r'(?i)(ghp_[a-zA-Z0-9]{36})', "GitHub personal access token"),
    (r'(?i)(Bearer\s+[a-zA-Z0-9_\-\.]{20,})', "Hardcoded Bearer token"),
]

# Unsafe subprocess patterns
SUBPROCESS_PATTERNS = [
    (r'subprocess\.(call|run|Popen)\s*\([^)]*shell\s*=\s*True', "subprocess with shell=True (command injection risk)"),
    (r'os\.system\s*\(', "os.system() call (command injection risk)"),
    (r'os\.popen\s*\(', "os.popen() call (command injection risk)"),
    (r'exec\s*\(', "exec() call (code injection risk)"),
    (r'eval\s*\(', "eval() call (code injection risk)"),
    (r'child_process\.exec\s*\(', "Node child_process.exec (command injection risk)"),
    (r'child_process\.execSync\s*\(', "Node child_process.execSync (command injection risk)"),
    (r'execSync\s*\(', "execSync without sanitization"),
]

# Network binding patterns
NETWORK_PATTERNS = [
    (r'(?:host|bind|address)\s*[=:]\s*["\']?0\.0\.0\.0', "Bound to 0.0.0.0 (all interfaces — network exposed)"),
    (r'\.listen\s*\(\s*\d+\s*,\s*["\']0\.0\.0\.0["\']', "Server listening on 0.0.0.0"),
    (r'INADDR_ANY', "Bound to INADDR_ANY (all interfaces)"),
]

# Path traversal patterns
PATH_TRAVERSAL_PATTERNS = [
    (r'\.\./', "Potential path traversal pattern (../)"),
    (r'path\.join\s*\([^)]*(?:req|params|args|input|query)', "Path.join with user input (path traversal risk)"),
    (r'readFile(?:Sync)?\s*\([^)]*(?:req|params|args|input|query)', "File read with user input"),
    (r'open\s*\([^)]*(?:request|params|args|input|query)', "File open with user input"),
]

# Auth check patterns (absence = vulnerability)
AUTH_PATTERNS = [
    r'(?i)(auth|authentication|authorize|jwt|bearer|oauth|session|login)',
    r'(?i)(middleware.*auth|auth.*middleware)',
    r'(?i)(verify.*token|token.*verify|validate.*token)',
    r'(?i)(api[_-]?key.*check|check.*api[_-]?key)',
]

# Input validation patterns (absence = risk)
VALIDATION_PATTERNS = [
    r'(?i)(zod|joi|yup|ajv|validate|sanitize|escape)',
    r'(?i)(input.*valid|valid.*input|schema.*valid)',
    r'(?i)(type.*check|check.*type|assert.*type)',
]

# Severity levels
CRITICAL = "CRITICAL"
HIGH = "HIGH"
MEDIUM = "MEDIUM"
LOW = "LOW"
INFO = "INFO"


def scan_file(filepath: Path) -> list[dict]:
    """Scan a single file for security issues."""
    issues = []
    try:
        content = filepath.read_text(encoding="utf-8", errors="ignore")
    except Exception:
        return issues

    lines = content.splitlines()

    # Check for hardcoded secrets
    for pattern, description in SECRET_PATTERNS:
        for i, line in enumerate(lines, 1):
            if re.search(pattern, line):
                # Skip if it's in a comment or example
                stripped = line.strip()
                if stripped.startswith("#") or stripped.startswith("//") or "example" in stripped.lower():
                    continue
                issues.append({
                    "severity": CRITICAL,
                    "type": "hardcoded_secret",
                    "file": str(filepath),
                    "line": i,
                    "detail": description,
                    "text": stripped[:100],
                })

    # Check for unsafe subprocess
    for pattern, description in SUBPROCESS_PATTERNS:
        for i, line in enumerate(lines, 1):
            if re.search(pattern, line):
                issues.append({
                    "severity": HIGH,
                    "type": "unsafe_subprocess",
                    "file": str(filepath),
                    "line": i,
                    "detail": description,
                    "text": line.strip()[:100],
                })

    # Check for network exposure
    for pattern, description in NETWORK_PATTERNS:
        for i, line in enumerate(lines, 1):
            if re.search(pattern, line):
                issues.append({
                    "severity": HIGH,
                    "type": "network_exposure",
                    "file": str(filepath),
                    "line": i,
                    "detail": description,
                    "text": line.strip()[:100],
                })

    # Check for path traversal
    for pattern, description in PATH_TRAVERSAL_PATTERNS:
        for i, line in enumerate(lines, 1):
            if re.search(pattern, line):
                stripped = line.strip()
                if stripped.startswith("#") or stripped.startswith("//"):
                    continue
                issues.append({
                    "severity": MEDIUM,
                    "type": "path_traversal",
                    "file": str(filepath),
                    "line": i,
                    "detail": description,
                    "text": stripped[:100],
                })

    return issues


def check_auth_presence(project_dir: Path) -> dict:
    """Check if the project implements any authentication."""
    source_files = list(project_dir.rglob("*.py")) + list(project_dir.rglob("*.ts")) + \
                   list(project_dir.rglob("*.js")) + list(project_dir.rglob("*.go"))

    auth_found = False
    auth_files = []

    for f in source_files:
        try:
            content = f.read_text(encoding="utf-8", errors="ignore")
            for pattern in AUTH_PATTERNS:
                if re.search(pattern, content):
                    auth_found = True
                    auth_files.append(str(f))
                    break
        except Exception:
            continue

    return {
        "has_auth": auth_found,
        "auth_files": list(set(auth_files)),
        "severity": LOW if auth_found else CRITICAL,
        "detail": "Authentication mechanisms found" if auth_found else
                  "NO AUTHENTICATION DETECTED (CVE-2026-32211 pattern — CVSS 9.1)",
    }


def check_input_validation(project_dir: Path) -> dict:
    """Check if the project validates tool inputs."""
    source_files = list(project_dir.rglob("*.py")) + list(project_dir.rglob("*.ts")) + \
                   list(project_dir.rglob("*.js"))

    validation_found = False
    for f in source_files:
        try:
            content = f.read_text(encoding="utf-8", errors="ignore")
            for pattern in VALIDATION_PATTERNS:
                if re.search(pattern, content):
                    validation_found = True
                    break
        except Exception:
            continue
        if validation_found:
            break

    return {
        "has_validation": validation_found,
        "severity": INFO if validation_found else MEDIUM,
        "detail": "Input validation library/patterns found" if validation_found else
                  "No input validation detected -- tool parameters may not be sanitized",
    }


def check_mcp_config(config_path: str) -> list[dict]:
    """Check an MCP client config for security issues."""
    issues = []
    with open(config_path, "r", encoding="utf-8") as f:
        config = json.load(f)

    servers = config.get("mcpServers", config.get("servers", {}))
    for name, server in servers.items():
        # Check for env vars with secrets in plain text
        env = server.get("env", {})
        for key, value in env.items():
            if any(s in key.upper() for s in ["KEY", "TOKEN", "SECRET", "PASSWORD"]):
                if value and not value.startswith("${") and not value.startswith("env:"):
                    issues.append({
                        "severity": CRITICAL,
                        "type": "config_secret",
                        "file": config_path,
                        "line": 0,
                        "detail": f"Server '{name}': plaintext secret in env var '{key}'",
                        "text": f"{key}={value[:8]}...",
                    })

        # Check for --allow-dangerous or --no-verify flags
        args = server.get("args", [])
        for arg in args:
            if isinstance(arg, str) and ("dangerous" in arg.lower() or "no-verify" in arg.lower() or "insecure" in arg.lower()):
                issues.append({
                    "severity": HIGH,
                    "type": "unsafe_flag",
                    "file": config_path,
                    "line": 0,
                    "detail": f"Server '{name}': unsafe flag '{arg}'",
                })

    return issues


def scan_project(project_dir: Path) -> dict:
    """Full security scan of an MCP server project."""
    scannable_extensions = {".py", ".ts", ".js", ".go", ".rs", ".json", ".yaml", ".yml", ".toml", ".env"}
    source_files = [f for f in project_dir.rglob("*") if f.suffix in scannable_extensions
                    and "node_modules" not in str(f) and ".git" not in str(f)
                    and "__pycache__" not in str(f) and "dist" not in str(f)]

    all_issues = []
    for f in source_files:
        all_issues.extend(scan_file(f))

    # Check auth and validation at project level
    auth_check = check_auth_presence(project_dir)
    validation_check = check_input_validation(project_dir)

    # Count by severity
    severity_counts = {}
    for issue in all_issues:
        sev = issue["severity"]
        severity_counts[sev] = severity_counts.get(sev, 0) + 1

    # Security score (0-100, higher = more secure)
    score = 100
    score -= severity_counts.get(CRITICAL, 0) * 25
    score -= severity_counts.get(HIGH, 0) * 15
    score -= severity_counts.get(MEDIUM, 0) * 5
    score -= severity_counts.get(LOW, 0) * 2
    if not auth_check["has_auth"]:
        score -= 30
    if not validation_check["has_validation"]:
        score -= 10
    score = max(0, score)

    return {
        "project": str(project_dir),
        "files_scanned": len(source_files),
        "security_score": score,
        "auth": auth_check,
        "validation": validation_check,
        "issues": all_issues,
        "severity_counts": severity_counts,
    }


def format_report(result: dict, verbose: bool = False) -> str:
    """Format scan results as human-readable report."""
    lines = []
    lines.append("=" * 60)
    lines.append("  MCP SERVER SECURITY SCAN")
    lines.append("=" * 60)
    lines.append("")
    lines.append(f"  Project:        {result['project']}")
    lines.append(f"  Files scanned:  {result['files_scanned']}")
    lines.append(f"  Security score: {result['security_score']}/100")
    lines.append("")

    # Auth status
    auth = result["auth"]
    icon = "[ok]" if auth["has_auth"] else "[!!!]"
    lines.append(f"  {icon} Auth: {auth['detail']}")

    # Validation status
    val = result["validation"]
    icon = "[ok]" if val["has_validation"] else "[!]"
    lines.append(f"  {icon} Validation: {val['detail']}")
    lines.append("")

    # Severity summary
    sc = result["severity_counts"]
    if sc:
        lines.append("-" * 60)
        lines.append(f"  ISSUES: {sc.get(CRITICAL,0)} critical, {sc.get(HIGH,0)} high, "
                     f"{sc.get(MEDIUM,0)} medium, {sc.get(LOW,0)} low")
        lines.append("-" * 60)

        for issue in sorted(result["issues"], key=lambda x: {CRITICAL:0,HIGH:1,MEDIUM:2,LOW:3,INFO:4}.get(x["severity"],5)):
            sev = issue["severity"]
            lines.append(f"  [{sev:>8}] {issue['type']}")
            lines.append(f"            {issue['detail']}")
            lines.append(f"            {issue['file']}:{issue.get('line','')}")
            if verbose and "text" in issue:
                lines.append(f"            > {issue['text']}")
            lines.append("")
    else:
        lines.append("  No issues found.")

    lines.append("=" * 60)

    # Recommendations
    if result["security_score"] < 50:
        lines.append("")
        lines.append("  RECOMMENDATIONS:")
        if not auth["has_auth"]:
            lines.append("  1. Add authentication (OAuth 2.1, API key validation, or JWT)")
            lines.append("     See: https://modelcontextprotocol.io/specification")
        if sc.get(CRITICAL, 0) > 0:
            lines.append("  2. Remove hardcoded secrets immediately")
            lines.append("     Use environment variables or a secrets manager")
        if sc.get(HIGH, 0) > 0:
            lines.append("  3. Replace shell=True with explicit command arrays")
            lines.append("     Bind servers to 127.0.0.1, not 0.0.0.0")
        lines.append("")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="MCP Server Security Scanner -- detect common vulnerabilities"
    )
    parser.add_argument("path", nargs="?", help="Path to MCP server project directory")
    parser.add_argument("--check-config", help="Check an MCP client config file for secrets")
    parser.add_argument("--verbose", "-v", action="store_true", help="Show issue context")
    parser.add_argument("--json", action="store_true", help="Output as JSON")

    args = parser.parse_args()

    if args.check_config:
        issues = check_mcp_config(args.check_config)
        if args.json:
            print(json.dumps(issues, indent=2))
        else:
            print(f"\nMCP Config Security Check: {args.check_config}")
            print(f"Issues found: {len(issues)}")
            for issue in issues:
                print(f"  [{issue['severity']}] {issue['detail']}")
        return

    if not args.path:
        print("Usage: py mcp_security_scanner.py /path/to/mcp-server/")
        print("       py mcp_security_scanner.py --check-config mcp_config.json")
        sys.exit(1)

    project_dir = Path(args.path)
    if not project_dir.exists():
        print(f"Error: path not found: {project_dir}")
        sys.exit(1)

    result = scan_project(project_dir)

    if args.json:
        print(json.dumps(result, indent=2, default=str))
    else:
        print(format_report(result, verbose=args.verbose))


if __name__ == "__main__":
    main()