linux-mcp-server.py

#!/usr/bin/env python3

import asyncio
import os
import logging
import re
from typing import Any, Sequence, List, Tuple
import paramiko
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Resource,
    Tool,
    TextContent,
    ImageContent,
    EmbeddedResource,
    LoggingLevel
)
from pydantic import AnyUrl
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("linux-mcp-server")

class CommandSafetyChecker:
    """Class to check for dangerous commands and provide warnings"""
    
    # Define dangerous commands and their risk levels
    DANGEROUS_COMMANDS = {
        # Critical - Can cause system destruction or data loss
        'critical': {
            'rm': 'Remove files/directories - can cause permanent data loss',
            'rmdir': 'Remove directories - can cause data loss',
            'dd': 'Disk dump - can overwrite entire disks and cause data loss',
            'format': 'Format storage devices - will erase all data',
            'fdisk': 'Partition disks - can destroy partition tables',
            'mkfs': 'Create filesystems - will erase existing data',
            'shred': 'Securely delete files - causes permanent data loss',
            'wipefs': 'Wipe filesystem signatures - can make data inaccessible',
        },
        # High - Can affect system state or security
        'high': {
            'shutdown': 'Shutdown the system - will terminate all processes',
            'reboot': 'Restart the system - will terminate all processes',
            'halt': 'Stop the system - will terminate all processes',
            'poweroff': 'Power off the system - will terminate all processes',
            'init': 'Change system runlevel - can shutdown or restart system',
            'systemctl': 'Control systemd services - can stop critical services',
            'service': 'Control system services - can stop critical services',
            'kill': 'Terminate processes - can stop important system processes',
            'killall': 'Kill processes by name - can stop important system processes',
            'pkill': 'Kill processes by pattern - can stop important system processes',
        },
        # Medium - Can modify system configuration
        'medium': {
            'chmod': 'Change file permissions - can affect system security',
            'chown': 'Change file ownership - can affect system security',
            'passwd': 'Change user passwords - affects system security',
            'usermod': 'Modify user accounts - affects system security',
            'userdel': 'Delete user accounts - can cause access issues',
            'groupdel': 'Delete groups - can cause access issues',
            'mount': 'Mount filesystems - can affect system state',
            'umount': 'Unmount filesystems - can cause data loss if not synced',
            'iptables': 'Configure firewall rules - can block network access',
            'ufw': 'Configure firewall - can block network access',
            'crontab': 'Modify scheduled tasks - can affect system automation',
            'docker': 'Docker operations - can affect container state',
        }
    }
    
    # Patterns that make commands more dangerous
    DANGEROUS_PATTERNS = [
        (r'rm\s+.*-r.*/', 'Recursive delete with absolute path'),
        (r'rm\s+.*-rf\s*/', 'Force recursive delete from root'),
        (r'chmod\s+777', 'Setting world-writable permissions'),
        (r'chown\s+.*:.*\s*/', 'Changing ownership of system directories'),
        (r'dd\s+.*of=/dev/', 'Writing directly to device files'),
        (r'>\s*/dev/', 'Redirecting output to device files'),
        (r'mkfs.*\s+/dev/', 'Creating filesystem on device'),
        (r'sudo\s+', 'Running with elevated privileges'),
    ]
    
    def __init__(self):
        self.warning_enabled = True
    
    def analyze_command(self, command: str) -> Tuple[str, str, List[str]]:
        """
        Analyze a command for potential dangers
        
        Returns:
            Tuple of (risk_level, base_command, warnings)
        """
        if not command.strip():
            return 'none', '', []
        
        # Clean and normalize the command
        clean_command = command.strip().lower()
        
        # Remove common prefixes
        for prefix in ['sudo ', 'su - ', 'su -c ']:
            if clean_command.startswith(prefix):
                clean_command = clean_command[len(prefix):].strip()
        
        # Extract the base command (first word)
        base_command = clean_command.split()[0] if clean_command.split() else ''
        
        # Check for dangerous commands
        risk_level = 'none'
        warnings = []
        
        # Check each risk category
        for level in ['critical', 'high', 'medium']:
            if base_command in self.DANGEROUS_COMMANDS[level]:
                risk_level = level
                warning_msg = self.DANGEROUS_COMMANDS[level][base_command]
                warnings.append(f"⚠️  {warning_msg}")
                break
        
        # Check for dangerous patterns
        original_command = command.lower()
        for pattern, description in self.DANGEROUS_PATTERNS:
            if re.search(pattern, original_command):
                warnings.append(f"🚨 Pattern Alert: {description}")
                if risk_level == 'none':
                    risk_level = 'medium'
                elif risk_level == 'medium' and 'rm.*-rf' in pattern:
                    risk_level = 'critical'
        
        # Special checks for specific dangerous combinations
        if 'rm' in base_command:
            if '-rf' in original_command or '-fr' in original_command:
                warnings.append("🔥 EXTREME CAUTION: Force recursive delete detected!")
                risk_level = 'critical'
            elif '-r' in original_command and ('/' in original_command or '*' in original_command):
                warnings.append("⚠️  Recursive delete with wildcards or root paths")
                if risk_level not in ['critical']:
                    risk_level = 'high'
        
        return risk_level, base_command, warnings
    
    def format_warning_message(self, command: str, risk_level: str, warnings: List[str]) -> str:
        """Format a comprehensive warning message"""
        if risk_level == 'none' or not warnings:
            return ""
        
        # Risk level indicators
        risk_indicators = {
            'critical': '🚨 CRITICAL RISK',
            'high': '⚠️  HIGH RISK',
            'medium': '⚡ MEDIUM RISK'
        }
        
        # Risk level colors (for terminal output)
        risk_colors = {
            'critical': '\033[91m',  # Red
            'high': '\033[93m',      # Yellow
            'medium': '\033[94m',    # Blue
        }
        reset_color = '\033[0m'
        
        header = f"\n{'='*60}"
        risk_header = f"{risk_colors.get(risk_level, '')}{risk_indicators.get(risk_level, '')} COMMAND WARNING{reset_color}"
        command_line = f"Command: {command}"
        separator = "-" * 60
        
        warning_lines = []
        for warning in warnings:
            warning_lines.append(f"  {warning}")
        
        footer = "="*60
        
        warning_message = "\n".join([
            header,
            risk_header,
            command_line,
            separator,
            "\n".join(warning_lines),
            "",
            "⚡ This command may have significant impact on the system.",
            "   Please review carefully before proceeding.",
            footer,
            ""
        ])
        
        return warning_message

    def requires_confirmation(self, risk_level: str) -> bool:
        """Check if a command requires user confirmation based on risk level"""
        return risk_level in ['critical', 'high']

    def get_confirmation_prompt(self, command: str, risk_level: str, warnings: List[str]) -> str:
        """Generate a user-friendly confirmation prompt"""
        warning_msg = self.format_warning_message(command, risk_level, warnings)
        
        prompt = (
            f"{warning_msg}\n"
            f"🛡️  SAFETY CONFIRMATION REQUIRED\n\n"
            f"The command '{command}' has been identified as {risk_level.upper()} RISK.\n\n"
            f"⚠️  POTENTIAL IMPACTS:\n"
        )
        
        for warning in warnings:
            prompt += f"   • {warning.replace('⚠️  ', '').replace('🚨 Pattern Alert: ', '')}\n"
        
        prompt += (
            f"\n"
            f"❓ Do you want to proceed with executing this command?\n\n"
            f"Please confirm by responding:\n"
            f"   • 'YES' or 'yes' to proceed\n"
            f"   • 'NO' or 'no' to cancel\n"
            f"   • Any other response will cancel the operation\n\n"
            f"⚠️  Make sure you understand the risks before confirming!\n"
        )
        
        return prompt

class LinuxMCPServer:
    def __init__(self):
        self.server = Server("linux-mcp-server")
        self.ssh_client = None
        self.safety_checker = CommandSafetyChecker()
        
        # Track pending dangerous commands awaiting confirmation
        self.pending_dangerous_commands = {}
        
        # Load configuration from environment
        self.host = os.getenv('LINUX_HOST')
        self.port = int(os.getenv('LINUX_PORT', '22'))
        self.username = os.getenv('LINUX_USERNAME')
        self.password = os.getenv('LINUX_PASSWORD')
        self.key_path = os.getenv('LINUX_SSH_KEY_PATH')
        self.timeout = int(os.getenv('LINUX_TIMEOUT', '30'))
        
        # Safety configuration
        self.enable_warnings = os.getenv('LINUX_ENABLE_WARNINGS', 'true').lower() == 'true'
        self.require_confirmation = os.getenv('LINUX_REQUIRE_CONFIRMATION', 'true').lower() == 'true'
        self.block_critical = os.getenv('LINUX_BLOCK_CRITICAL', 'false').lower() == 'true'
        
        if not self.host or not self.username:
            raise ValueError("LINUX_HOST and LINUX_USERNAME must be set in environment")
        
        self.setup_handlers()
    
    def setup_handlers(self):
        """Setup MCP server handlers"""
        
        @self.server.list_tools()
        async def handle_list_tools() -> list[Tool]:
            """List available tools"""
            return [
                Tool(
                    name="execute_command",
                    description="Execute a Linux command on the remote host",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "command": {
                                "type": "string",
                                "description": "The Linux command to execute"
                            },
                            "timeout": {
                                "type": "integer",
                                "description": "Command timeout in seconds (default: 30)",
                                "default": 30
                            },
                            "confirm_dangerous": {
                                "type": "boolean",
                                "description": "Prompt the user to confirm execution of dangerous commands after seeing the warning. This must be explicitly confirmed by the user for high-risk or critical commands.",
                                "default": False
                            }
                        },
                        "required": ["command"]
                    }
                ),
                Tool(
                    name="confirm_command",
                    description="Prompt the user to confirm the execution of a previously flagged dangerous command",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "confirmation": {
                                "type": "string",
                                "description": "User confirmation: 'YES' to proceed, 'NO' to cancel"
                            },
                            "command_id": {
                                "type": "string",
                                "description": "ID of the command to confirm (provided in the confirmation prompt)"
                            }
                        },
                        "required": ["confirmation"]
                    }
                ),
                Tool(
                    name="get_system_info",
                    description="Get basic system information from the remote host",
                    inputSchema={
                        "type": "object",
                        "properties": {}
                    }
                ),
                Tool(
                    name="list_directory",
                    description="List contents of a directory",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {
                                "type": "string",
                                "description": "Directory path to list (default: current directory)",
                                "default": "."
                            },
                            "detailed": {
                                "type": "boolean",
                                "description": "Show detailed listing (ls -la)",
                                "default": False
                            }
                        }
                    }
                ),
                Tool(
                    name="file_operations",
                    description="Perform file operations (read, write, append)",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "operation": {
                                "type": "string",
                                "enum": ["read", "write", "append"],
                                "description": "File operation to perform"
                            },
                            "file_path": {
                                "type": "string",
                                "description": "Path to the file"
                            },
                            "content": {
                                "type": "string",
                                "description": "Content to write/append (required for write/append operations)"
                            }
                        },
                        "required": ["operation", "file_path"]
                    }
                ),
                Tool(
                    name="trafficshaper",
                    description="Manage traffic shaping rules on PE routers using the trafficshaper CLI tool. Supports setting delays, showing rules, deleting rules, and other traffic impairment operations.",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "action": {
                                "type": "string",
                                "enum": ["show", "set", "del", "reset", "help", "version", "license"],
                                "description": "Action to perform: show (display rules), set (apply delay), del (delete rules), reset (reset all), help (show help), version (show version), license (show license)"
                            },
                            "router": {
                                "type": "string",
                                "enum": ["pe1", "pe2"],
                                "description": "Target router (pe1 or pe2). Required for show, set, del actions unless using 'all' target."
                            },
                            "interface": {
                                "type": "string",
                                "enum": ["ge002", "ge003", "all"],
                                "description": "Target interface: ge002 (ge-0/0/2), ge003 (ge-0/0/3), or all interfaces. Required for show, set, del actions."
                            },
                            "delay": {
                                "type": "integer",
                                "description": "Delay in milliseconds for set action. Required when action is 'set'.",
                                "minimum": 0
                            }
                        },
                        "required": ["action"],
                        "allOf": [
                            {
                                "if": {
                                    "properties": {"action": {"const": "set"}}
                                },
                                "then": {
                                    "required": ["router", "interface", "delay"]
                                }
                            },
                            {
                                "if": {
                                    "properties": {"action": {"enum": ["show", "del"]}}
                                },
                                "then": {
                                    "anyOf": [
                                        {
                                            "required": ["router", "interface"]
                                        },
                                        {
                                            "properties": {"interface": {"const": "all"}},
                                            "not": {"required": ["router"]}
                                        }
                                    ]
                                }
                            }
                        ]
                    }
                )
            ]
        
        @self.server.call_tool()
        async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[TextContent | ImageContent | EmbeddedResource]:
            """Handle tool calls"""
            
            if not arguments:
                arguments = {}
            
            try:
                await self._ensure_connection()
                
                if name == "execute_command":
                    return await self._execute_command(
                        arguments.get("command", ""),
                        arguments.get("timeout", self.timeout),
                        arguments.get("confirm_dangerous", False)
                    )
                
                elif name == "confirm_command":
                    return await self._confirm_command(
                        arguments.get("confirmation", ""),
                        arguments.get("command_id", "")
                    )
                
                elif name == "get_system_info":
                    return await self._get_system_info()
                
                elif name == "list_directory":
                    return await self._list_directory(
                        arguments.get("path", "."),
                        arguments.get("detailed", False)
                    )
                
                elif name == "file_operations":
                    return await self._file_operations(
                        arguments.get("operation"),
                        arguments.get("file_path"),
                        arguments.get("content")
                    )
                
                elif name == "trafficshaper":
                    return await self._trafficshaper(arguments)
                
                else:
                    raise ValueError(f"Unknown tool: {name}")
                    
            except Exception as e:
                logger.error(f"Error executing tool {name}: {str(e)}")
                return [TextContent(type="text", text=f"Error: {str(e)}")]
    
    async def _confirm_command(self, confirmation: str, command_id: str = "") -> list[TextContent]:
        """Handle confirmation of dangerous commands"""
        # For now, we'll use a simple approach - just look for the most recent dangerous command
        # In a more sophisticated implementation, you'd use command_id to track specific commands
        
        confirmation_lower = confirmation.lower().strip()
        
        if confirmation_lower in ['yes', 'y']:
            return [TextContent(type="text", text=
                "✅ CONFIRMATION RECEIVED\n\n"
                "Your confirmation has been noted. Please re-run the original command with "
                "'confirm_dangerous: true' parameter to execute it.\n\n"
                "Example: execute_command with parameters:\n"
                "- command: [your original command]\n"
                "- confirm_dangerous: true\n\n"
                "⚠️  Remember: You are taking responsibility for the execution of this potentially dangerous command."
            )]
        
        elif confirmation_lower in ['no', 'n']:
            return [TextContent(type="text", text=
                "🛡️  COMMAND CANCELLED\n\n"
                "The dangerous command has been cancelled for your safety.\n"
                "No changes have been made to the system.\n\n"
                "If you need to perform this operation safely, consider:\n"
                "• Breaking it down into smaller, safer commands\n"
                "• Using safer alternatives\n"
                "• Double-checking the command syntax\n"
                "• Making backups if necessary"
            )]
        
        else:
            return [TextContent(type="text", text=
                "❌ INVALID CONFIRMATION\n\n"
                f"'{confirmation}' is not a valid confirmation response.\n\n"
                "Please respond with:\n"
                "• 'YES' or 'yes' to proceed\n"
                "• 'NO' or 'no' to cancel\n\n"
                "The command remains cancelled for your safety."
            )]

    async def _trafficshaper(self, arguments: dict[str, Any]) -> list[TextContent]:
        """Handle trafficshaper operations"""
        action = arguments.get("action")
        router = arguments.get("router")
        interface = arguments.get("interface")
        delay = arguments.get("delay")
        
        if not action:
            return [TextContent(type="text", text="Error: action is required")]
        
        # Build the trafficshaper command
        cmd_parts = ["trafficshaper"]
        
        if action in ["help", "version", "license", "reset"]:
            # Simple commands that don't need additional parameters
            if action == "version":
                cmd_parts.append("cli")
                cmd_parts.append("version")
            else:
                cmd_parts.append(action)
        
        elif action == "show":
            cmd_parts.append("show")
            if interface == "all":
                cmd_parts.append("all")
            elif router and interface:
                cmd_parts.append(router)
                cmd_parts.append(interface)
            else:
                return [TextContent(type="text", text="Error: For 'show' action, either use interface='all' or provide both router and interface")]
        
        elif action == "set":
            if not all([router, interface, delay is not None]):
                return [TextContent(type="text", text="Error: For 'set' action, router, interface, and delay are required")]
            
            if interface == "all":
                return [TextContent(type="text", text="Error: Cannot use 'all' interface with 'set' action. Specify ge002 or ge003")]
            
            cmd_parts.extend(["set", router, interface, "--delay", str(delay)])
        
        elif action == "del":
            cmd_parts.append("del")
            if not router:
                return [TextContent(type="text", text="Error: For 'del' action, router is required")]
            
            cmd_parts.append(router)
            cmd_parts.append(interface or "all")
        
        else:
            return [TextContent(type="text", text=f"Error: Unknown action '{action}'")]
        
        # Execute the command
        command = " ".join(cmd_parts)
        logger.info(f"Executing trafficshaper command: {command}")
        
        try:
            result = await self._execute_command(command)
            
            # Add context to the result for better understanding
            context_info = f"Executed: {command}\n" + "="*50 + "\n"
            
            if result and result[0].text:
                original_text = result[0].text
                # Clean up the output if it contains our standard command output format
                if "STDOUT:" in original_text:
                    stdout_part = original_text.split("STDOUT:\n")[1].split("\n\nSTDERR:")[0]
                    return [TextContent(type="text", text=context_info + stdout_part)]
                else:
                    return [TextContent(type="text", text=context_info + original_text)]
            else:
                return [TextContent(type="text", text=context_info + "Command executed but no output returned")]
                
        except Exception as e:
            logger.error(f"Trafficshaper command failed: {str(e)}")
            return [TextContent(type="text", text=f"Error executing trafficshaper command '{command}': {str(e)}")]
    
    async def _ensure_connection(self):
        """Ensure SSH connection is established"""
        if self.ssh_client is None or not self.ssh_client.get_transport() or not self.ssh_client.get_transport().is_active():
            await self._connect()
    
    async def _connect(self):
        """Establish SSH connection"""
        try:
            self.ssh_client = paramiko.SSHClient()
            self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            # Connection parameters
            connect_kwargs = {
                'hostname': self.host,
                'port': self.port,
                'username': self.username,
                'timeout': self.timeout
            }
            
            # Use SSH key if provided, otherwise use password
            if self.key_path and os.path.exists(self.key_path):
                connect_kwargs['key_filename'] = self.key_path
                logger.info(f"Connecting to {self.host} using SSH key")
            elif self.password:
                connect_kwargs['password'] = self.password
                logger.info(f"Connecting to {self.host} using password")
            else:
                raise ValueError("Either LINUX_PASSWORD or LINUX_SSH_KEY_PATH must be provided")
            
            # Run connection in thread pool to avoid blocking
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, lambda: self.ssh_client.connect(**connect_kwargs))
            
            logger.info(f"Successfully connected to {self.host}")
            
        except Exception as e:
            logger.error(f"Failed to connect to {self.host}: {str(e)}")
            if self.ssh_client:
                self.ssh_client.close()
                self.ssh_client = None
            raise
    
    async def _execute_command(self, command: str, timeout: int = 30, confirm_dangerous: bool = False) -> list[TextContent]:
        """Execute a command on the remote host with configurable safety checks"""
        if not command.strip():
            return [TextContent(type="text", text="Error: Command cannot be empty")]
        
        # ALWAYS perform safety analysis
        risk_level, base_command, warnings = self.safety_checker.analyze_command(command)
        
        # CRITICAL COMMANDS: Check if should be blocked entirely
        if risk_level == 'critical' and self.block_critical:
            warning_msg = self.safety_checker.format_warning_message(command, risk_level, warnings)
            blocking_msg = (
                f"{warning_msg}\n"
                f"🚫 CRITICAL COMMAND BLOCKED\n\n"
                f"This CRITICAL risk command has been blocked by system policy.\n"
                f"The command '{command}' could cause severe system damage or data loss.\n\n"
                f"To enable critical commands, set LINUX_BLOCK_CRITICAL=false in your environment.\n\n"
                f"🔒 COMMAND NOT EXECUTED - System protected from critical operations\n"
                f"Risk Level: {risk_level.upper()}\n"
                f"Command: {command}\n"
            )
            return [TextContent(type="text", text=blocking_msg)]
        
        # HIGH/CRITICAL RISK: Check if confirmation is required
        if self.safety_checker.requires_confirmation(risk_level) and self.require_confirmation:
            if not confirm_dangerous:
                # Generate user-friendly confirmation prompt
                confirmation_prompt = self.safety_checker.get_confirmation_prompt(command, risk_level, warnings)
                
                blocking_msg = (
                    f"{confirmation_prompt}\n"
                    f"🛡️ COMMAND EXECUTION PAUSED FOR SAFETY\n\n"
                    f"Please respond to this message with your confirmation choice.\n"
                    f"The command will only execute after you confirm and re-run it with 'confirm_dangerous: true'.\n\n"
                    f"📝 COMMAND AWAITING CONFIRMATION:\n"
                    f"   Command: {command}\n"
                    f"   Risk Level: {risk_level.upper()}\n\n"
                    f"💡 To disable confirmation prompts, set LINUX_REQUIRE_CONFIRMATION=false in your environment.\n"
                )
                
                return [TextContent(type="text", text=blocking_msg)]
            
            else:
                # Command is dangerous but explicitly confirmed - show warning and proceed
                warning_msg = self.safety_checker.format_warning_message(command, risk_level, warnings) if self.enable_warnings else ""
                confirmation_note = f"✅ CONFIRMED: Executing {risk_level.upper()} risk command as explicitly requested.\n\n"
                
                # Execute the command and prepend the warning
                execution_result = await self._execute_actual_command(command, timeout)
                if execution_result and execution_result[0].text:
                    combined_output = warning_msg + confirmation_note + execution_result[0].text
                    return [TextContent(type="text", text=combined_output)]
                else:
                    return [TextContent(type="text", text=warning_msg + confirmation_note)]
        
        # DANGEROUS BUT CONFIRMATION DISABLED: Show warning if enabled and execute
        elif self.safety_checker.requires_confirmation(risk_level) and not self.require_confirmation:
            if self.enable_warnings:
                warning_msg = self.safety_checker.format_warning_message(command, risk_level, warnings)
                warning_note = f"⚠️  CONFIRMATION DISABLED: Executing {risk_level.upper()} risk command without confirmation.\n\n"
                execution_result = await self._execute_actual_command(command, timeout)
                if execution_result and execution_result[0].text:
                    combined_output = warning_msg + warning_note + execution_result[0].text
                    return [TextContent(type="text", text=combined_output)]
                else:
                    return [TextContent(type="text", text=warning_msg + warning_note)]
            else:
                # No warnings, just execute
                return await self._execute_actual_command(command, timeout)
        
        # MEDIUM RISK OR SAFE COMMANDS: Execute with warnings if enabled
        elif warnings and self.enable_warnings:
            warning_msg = self.safety_checker.format_warning_message(command, risk_level, warnings)
            execution_result = await self._execute_actual_command(command, timeout)
            if execution_result and execution_result[0].text:
                return [TextContent(type="text", text=warning_msg + "\n" + execution_result[0].text)]
            else:
                return [TextContent(type="text", text=warning_msg)]
        
        # Execute safe command normally
        return await self._execute_actual_command(command, timeout)
    
    async def _execute_actual_command(self, command: str, timeout: int = 30) -> list[TextContent]:
        """Actually execute the command without safety checks"""
        try:
            logger.info(f"Executing command: {command}")
            
            # Execute command in thread pool
            loop = asyncio.get_event_loop()
            stdin, stdout, stderr = await loop.run_in_executor(
                None, 
                lambda: self.ssh_client.exec_command(command, timeout=timeout)
            )
            
            # Read output
            stdout_data = await loop.run_in_executor(None, stdout.read)
            stderr_data = await loop.run_in_executor(None, stderr.read)
            exit_code = stdout.channel.recv_exit_status()
            
            # Format response
            result = []
            if stdout_data:
                result.append(f"STDOUT:\n{stdout_data.decode('utf-8')}")
            if stderr_data:
                result.append(f"STDERR:\n{stderr_data.decode('utf-8')}")
            result.append(f"Exit Code: {exit_code}")
            
            command_output = "\n\n".join(result)
            return [TextContent(type="text", text=command_output)]
            
        except Exception as e:
            logger.error(f"Command execution failed: {str(e)}")
            return [TextContent(type="text", text=f"Error executing command: {str(e)}")]
    
    async def _get_system_info(self) -> list[TextContent]:
        """Get system information"""
        commands = [
            ("Hostname", "hostname"),
            ("OS Info", "cat /etc/os-release | head -5"),
            ("Uptime", "uptime"),
            ("Memory", "free -h"),
            ("Disk Usage", "df -h | head -10"),
            ("CPU Info", "lscpu | head -10")
        ]
        
        info_parts = []
        for label, cmd in commands:
            try:
                result = await self._execute_actual_command(cmd)
                if result and result[0].text:
                    # Extract just the stdout part
                    output = result[0].text
                    if "STDOUT:" in output:
                        output = output.split("STDOUT:\n")[1].split("\n\nSTDERR:")[0]
                    info_parts.append(f"=== {label} ===\n{output}")
            except Exception as e:
                info_parts.append(f"=== {label} ===\nError: {str(e)}")
        
        return [TextContent(type="text", text="\n\n".join(info_parts))]
    
    async def _list_directory(self, path: str, detailed: bool = False) -> list[TextContent]:
        """List directory contents"""
        cmd = f"ls -la '{path}'" if detailed else f"ls '{path}'"
        return await self._execute_actual_command(cmd)
    
    async def _file_operations(self, operation: str, file_path: str, content: str = None) -> list[TextContent]:
        """Perform file operations"""
        if not file_path:
            return [TextContent(type="text", text="Error: file_path is required")]
        
        if operation == "read":
            return await self._execute_actual_command(f"cat '{file_path}'")
        
        elif operation == "write":
            if content is None:
                return [TextContent(type="text", text="Error: content is required for write operation")]
            # Use echo with proper escaping
            escaped_content = content.replace("'", "'\"'\"'")
            return await self._execute_actual_command(f"echo '{escaped_content}' > '{file_path}'")
        
        elif operation == "append":
            if content is None:
                return [TextContent(type="text", text="Error: content is required for append operation")]
            escaped_content = content.replace("'", "'\"'\"'")
            return await self._execute_actual_command(f"echo '{escaped_content}' >> '{file_path}'")
        
        else:
            return [TextContent(type="text", text=f"Error: Unknown operation '{operation}'")]
    
    def cleanup(self):
        """Cleanup resources"""
        if self.ssh_client:
            self.ssh_client.close()
            logger.info("SSH connection closed")

async def main():
    """Main entry point"""
    linux_server = LinuxMCPServer()
    
    try:
        async with stdio_server() as (read_stream, write_stream):
            await linux_server.server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name="linux-mcp-server",
                    server_version="1.0.0",
                    capabilities=linux_server.server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={},
                    ),
                ),
            )
    except KeyboardInterrupt:
        logger.info("Server interrupted by user")
    except Exception as e:
        logger.error(f"Server error: {str(e)}")
    finally:
        linux_server.cleanup()

if __name__ == "__main__":
    asyncio.run(main())