Reference¶

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional

class TransportProtocol(ABC):
    """Abstract base class for transport protocols."""

    @abstractmethod
    async def send_request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Any:
        """Send a JSON-RPC request to the server."""
        pass

    @abstractmethod
    async def send_raw(self, payload: Any) -> Any:
        """Send raw payload to the server."""
        pass

    @abstractmethod
    async def send_notification(self, method: str, params: Optional[Dict[str, Any]] = None) -> None:
        """Send a JSON-RPC notification to the server."""
        pass

    async def get_tools(self) -> List[Dict[str, Any]]:
        """Get list of available tools from the server."""
        # Implementation details...

    async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
        """Call a specific tool with arguments."""
        # Implementation details...

class UnifiedMCPFuzzerClient:
    """Unified client for MCP fuzzing operations."""

    def __init__(self, transport: TransportProtocol, safety_system: Optional[SafetySystem] = None):
        self.transport = transport
        self.safety_system = safety_system or SafetySystem()

    async def fuzz_tools(self, runs: int = 10, phase: str = "aggressive") -> Dict[str, Any]:
        """Fuzz tools with specified number of runs and phase."""
        # Implementation details...

    async def fuzz_protocol(self, runs_per_type: int = 5, protocol_type: Optional[str] = None, phase: str = "aggressive") -> Dict[str, Any]:
        """Fuzz protocol types with specified parameters."""
        # Implementation details...

class SafetySystem:
    """Core safety system for protecting against dangerous operations."""

    def __init__(self, fs_root: Optional[str] = None, enable_system_blocking: bool = True):
        self.fs_root = fs_root or os.path.expanduser("~/.mcp_fuzzer")
        self.enable_system_blocking = enable_system_blocking
        self.system_blocker = SystemBlocker() if enable_system_blocking else None

    def is_safe_environment(self) -> bool:
        """Check if current environment is safe for dangerous operations."""
        # Implementation details...

    def filter_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Filter potentially dangerous arguments."""
        # Implementation details...

class RealisticToolStrategy:
    """Generates realistic, valid data for tool testing."""

    def generate_string(self) -> str:
        """Generate realistic string values."""
        return draw(st.text(min_size=1, max_size=100))

    def generate_number(self) -> Union[int, float]:
        """Generate realistic numeric values."""
        return draw(st.one_of(st.integers(), st.floats()))

    def generate_boolean(self) -> bool:
        """Generate boolean values."""
        return draw(st.booleans())

class AggressiveToolStrategy:
    """Generates malicious/malformed data for security testing."""

    def generate_sql_injection(self) -> str:
        """Generate SQL injection attempts."""
        return draw(st.sampled_from([
            "' OR 1=1; --",
            "'; DROP TABLE users; --",
            "' UNION SELECT * FROM users --"
        ]))

    def generate_xss(self) -> str:
        """Generate XSS attack attempts."""
        return draw(st.sampled_from([
            "<script>alert('xss')</script>",
            "javascript:alert('xss')",
            "<img src=x onerror=alert('xss')>"
        ]))

{
    "tools": [
        {
            "name": "tool_name",
            "success_rate": 85.0,
            "total_runs": 10,
            "successful_runs": 8,
            "exception_count": 2,
            "exceptions": [
                "Invalid argument type",
                "Missing required parameter"
            ],
            "average_response_time": 0.15
        }
    ],
    "overall": {
        "total_tools": 3,
        "total_runs": 30,
        "overall_success_rate": 90.0,
        "total_exceptions": 3
    }
}

{
    "protocol_types": [
        {
            "name": "InitializeRequest",
            "total_runs": 5,
            "successful_runs": 5,
            "exception_count": 0,
            "success_rate": 100.0,
            "exceptions": [],
            "average_response_time": 0.12
        }
    ],
    "overall": {
        "total_protocol_types": 15,
        "total_runs": 75,
        "overall_success_rate": 93.3,
        "total_exceptions": 5
    }
}

def is_safe_test_environment() -> bool:
    """Check if we're in a safe environment for dangerous tests."""
    # Don't run dangerous tests on production systems
    if (os.getenv("CI") or
        os.getenv("PRODUCTION") or
        os.getenv("DANGEROUS_TESTS_DISABLED")):
        return False

    # Don't run on systems with critical processes
    try:
        with open("/proc/1/comm", "r") as f:
            init_process = f.read().strip()
            if init_process in ["systemd", "init"]:
                return False
    except (OSError, IOError):
        pass

    return True

class SystemBlocker:
    """Blocks dangerous system commands during fuzzing."""

    def __init__(self):
        self.blocked_commands = {
            "rm", "del", "format", "shutdown", "reboot",
            "kill", "killall", "pkill", "xkill"
        }
        self.blocked_patterns = [
            r"rm\s+-rf",
            r"del\s+/[sq]",
            r"format\s+[a-z]:",
            r"shutdown\s+",
            r"reboot\s+"
        ]

    def is_blocked(self, command: str) -> bool:
        """Check if command should be blocked."""
        # Implementation details...