Safety Guide¶

def is_safe_test_environment():
    """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

    # Don't run on systems with systemd
    if os.path.exists("/run/systemd/system"):
        return False

    # Don't run on systems with init
    if os.path.exists("/etc/inittab"):
        return False

    return True

class SystemBlocker:
    def __init__(self):
        self.blocked_commands = {
            # File system operations
            "rm", "del", "format", "fdisk", "mkfs",

            # System control
            "shutdown", "reboot", "halt", "poweroff",

            # Process management
            "kill", "killall", "pkill", "xkill",

            # Network operations
            "iptables", "firewall-cmd", "ufw",

            # Package management
            "apt", "yum", "dnf", "pacman", "brew"
        }

        self.blocked_patterns = [
            # Dangerous file operations
            r"rm\s+-rf",
            r"del\s+/[sq]",
            r"format\s+[a-z]:",

            # System shutdown
            r"shutdown\s+",
            r"reboot\s+",
            r"halt\s+",

            # Dangerous process operations
            r"kill\s+-9",
            r"killall\s+-9"
        ]

def create_safe_path(self):
    """Create a safe PATH environment."""
    safe_path = "/tmp/mcp_fuzzer_safe_path"
    os.makedirs(safe_path, exist_ok=True)

    # Create fake executables for blocked commands
    for cmd in self.blocked_commands:
        fake_cmd_path = os.path.join(safe_path, cmd)
        with open(fake_cmd_path, "w") as f:
            f.write(f"""#!/bin/bash
echo "Command '{cmd}' is blocked by MCP Fuzzer safety system"
echo "This command could be dangerous and has been prevented from executing"
exit 1
""")
        os.chmod(fake_cmd_path, 0o755)

    return safe_path

class FilesystemSandbox:
    def __init__(self, root_path: str):
        self.root_path = os.path.abspath(root_path)
        self.ensure_safe_directory()

    def ensure_safe_directory(self):
        """Ensure the sandbox directory exists and is safe."""
        if not os.path.exists(self.root_path):
            os.makedirs(self.root_path, mode=0o700)

        # Ensure directory is not in dangerous locations
        dangerous_paths = ["/", "/home", "/etc", "/var", "/usr", "/bin", "/sbin"]
        for dangerous in dangerous_paths:
            if self.root_path.startswith(dangerous) and self.root_path != dangerous:
                raise ValueError(f"Sandbox path {self.root_path} is in dangerous location")

    def is_path_safe(self, path: str) -> bool:
        """Check if a path is within the safe sandbox."""
        try:
            abs_path = os.path.abspath(path)
            return abs_path.startswith(self.root_path)
        except (OSError, ValueError):
            return False

    def sanitize_path(self, path: str) -> str:
        """Sanitize a path to ensure it's within the sandbox."""
        if not self.is_path_safe(path):
            return os.path.join(self.root_path, "safe_default")
        return path

class SafeSubprocess:
    def __init__(self, timeout: float = 30.0):
        self.timeout = timeout

    async def run_safe(self, command: str, **kwargs) -> Tuple[bytes, bytes]:
        """Run a command safely with timeout and isolation."""
        # Set up safe environment
        safe_env = self.create_safe_environment()

        # Create process with safety measures
        process = await asyncio.create_subprocess_exec(
            *command.split(),
            env=safe_env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            **kwargs
        )

        try:
            # Wait for completion with timeout
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=self.timeout
            )

            # Check return code
            if process.returncode != 0:
                raise subprocess.CalledProcessError(
                    process.returncode, command, stdout, stderr
                )

            return stdout, stderr

        except asyncio.TimeoutError:
            # Kill process if it times out
            process.kill()
            await process.wait()
            raise MCPTimeoutError(f"Command timed out after {self.timeout} seconds")

    def create_safe_environment(self) -> Dict[str, str]:
        """Create a safe environment for subprocess execution."""
        env = os.environ.copy()

        # Restrict PATH to safe locations
        env["PATH"] = "/usr/bin:/bin"

        # Remove dangerous environment variables
        dangerous_vars = ["LD_PRELOAD", "LD_LIBRARY_PATH", "PYTHONPATH"]
        for var in dangerous_vars:
            env.pop(var, None)

        return env

class InputSanitizer:
    def __init__(self):
        self.dangerous_patterns = [
            # SQL injection patterns
            r"(\b(union|select|insert|update|delete|drop|create|alter)\b)",
            r"(--|\b(and|or)\b\s+\d+\s*[=<>]\s*\d+)",

            # XSS patterns
            r"(<script[^>]*>.*?</script>)",
            r"(javascript:.*?)",
            r"(<img[^>]*on\w+\s*=)",

            # Path traversal patterns
            r"(\.\./|\.\.\\)",
            r"(/etc/passwd|/etc/shadow)",
            r"(c:\\windows\\system32)",

            # Command injection patterns
            r"(\b(rm|del|format|shutdown|reboot)\b)",
            r"(\||&|;|`|\$\(|\\n)"
        ]

        self.compiled_patterns = [re.compile(pattern, re.IGNORECASE)
                                 for pattern in self.dangerous_patterns]

    def sanitize_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Sanitize tool arguments to remove dangerous content."""
        sanitized = {}

        for key, value in arguments.items():
            if isinstance(value, str):
                sanitized[key] = self.sanitize_string(value)
            elif isinstance(value, dict):
                sanitized[key] = self.sanitize_arguments(value)
            elif isinstance(value, list):
                sanitized[key] = [self.sanitize_item(item) for item in value]
            else:
                sanitized[key] = value

        return sanitized

    def sanitize_string(self, value: str) -> str:
        """Sanitize a string value."""
        for pattern in self.compiled_patterns:
            if pattern.search(value):
                # Replace dangerous content with safe alternatives
                value = pattern.sub("[REDACTED_DANGEROUS_CONTENT]", value)

        return value

class CustomSafetyProvider:
    """Custom safety provider for specific requirements."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.blocked_patterns = config.get("blocked_patterns", [])

    def get_safety(self) -> "SafetySystem":
        """Return a configured safety system."""
        return CustomSafetySystem(self.config)

    def is_command_allowed(self, command: str) -> bool:
        """Check if a command is allowed."""
        for pattern in self.blocked_patterns:
            if re.search(pattern, command, re.IGNORECASE):
                return False
        return True

class CustomSafetySystem(SafetySystem):
    """Custom safety system with extended functionality."""

    def __init__(self, config: Dict[str, Any]):
        super().__init__()
        self.custom_rules = config.get("custom_rules", {})

    def filter_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Apply custom filtering rules."""
        # Apply base filtering
        filtered = super().filter_arguments(arguments)

        # Apply custom rules
        for rule_name, rule_func in self.custom_rules.items():
            filtered = rule_func(filtered)

        return filtered

class SafetyLogger:
    def __init__(self, log_file: Optional[str] = None):
        self.log_file = log_file or "mcp_fuzzer_safety.log"

    def log_blocked_command(self, command: str, reason: str):
        """Log a blocked command attempt."""
        timestamp = datetime.now().isoformat()
        log_entry = {
            "timestamp": timestamp,
            "event": "command_blocked",
            "command": command,
            "reason": reason,
            "severity": "high"
        }

        self._write_log(log_entry)

    def log_safety_violation(self, violation_type: str, details: str):
        """Log a safety violation."""
        timestamp = datetime.now().isoformat()
        log_entry = {
            "timestamp": timestamp,
            "event": "safety_violation",
            "type": violation_type,
            "details": details,
            "severity": "critical"
        }

        self._write_log(log_entry)

    def _write_log(self, entry: Dict[str, Any]):
        """Write log entry to file."""
        with open(self.log_file, "a") as f:
            json.dump(entry, f)
            f.write("\n")

def test_safety_system():
    """Test safety system functionality."""
    safety = SafetySystem(fs_root="/tmp/test_safe")

    # Test environment detection
    assert safety.is_safe_environment() == True

    # Test command blocking
    blocker = SystemBlocker()
    assert blocker.is_blocked("rm -rf /") == True
    assert blocker.is_blocked("ls -la") == False

    # Test filesystem sandboxing
    sandbox = FilesystemSandbox("/tmp/test_safe")
    assert sandbox.is_path_safe("/tmp/test_safe/file.txt") == True
    assert sandbox.is_path_safe("/etc/passwd") == False