OpenWRT SSH MCP Server

Instructions

Implementation Reference

• openwrt_ssh_mcp/tools.py:17-47 (handler)

  Main handler function OpenWRTTools.execute_command that validates the command, ensures SSH connection, executes via ssh_client, and returns the execution result with success status, output, error, exit code, and execution time.
  @staticmethod async def execute_command(command: str) -> dict[str, Any]: """ Execute a validated command on the OpenWRT router. Args: command: Shell command to execute Returns: dict: Execution result """ # Validate command is_valid, error_msg = SecurityValidator.validate_command(command) if not is_valid: return { "success": False, "error": error_msg, "output": "", } # Execute await ssh_client.ensure_connected() result = await ssh_client.execute(command) return { "success": result["success"], "output": result["stdout"], "error": result["stderr"], "exit_code": result["exit_code"], "execution_time": result["execution_time"], }
• openwrt_ssh_mcp/server.py:44-61 (schema)

  Tool schema definition for openwrt_execute_command registering it with the MCP server, including the command parameter (string, required) and description indicating command validation against a security whitelist.
  Tool( name="openwrt_execute_command", description=( "Execute a validated shell command on the OpenWRT router. " "Commands are validated against a security whitelist. " "Use this for commands not covered by other specialized tools." ), inputSchema={ "type": "object", "properties": { "command": { "type": "string", "description": "Shell command to execute (must be in whitelist)", }, }, "required": ["command"], }, ),
• openwrt_ssh_mcp/server.py:298-302 (registration)

  Tool registration routing in the call_tool handler that extracts the command argument and calls OpenWRTTools.execute_command with validation for missing arguments.
  elif name == "openwrt_execute_command": command = arguments.get("command") if not command: raise ValueError("Missing required argument: command") result = await OpenWRTTools.execute_command(command)
• openwrt_ssh_mcp/security.py:128-156 (helper)

  Security validation function SecurityValidator.validate_command that checks commands against dangerous blocked patterns and an allowed whitelist, returning validation status and error message.
  @classmethod def validate_command(cls, command: str) -> tuple[bool, Optional[str]]: """ Validate if a command is safe to execute. Returns: tuple[bool, Optional[str]]: (is_valid, error_message) """ if not settings.enable_command_validation: logger.warning("Command validation is DISABLED - executing without checks") return True, None # Check for blocked patterns first for pattern in cls.BLOCKED_PATTERNS: if re.search(pattern, command, re.IGNORECASE): error = f"Command blocked by security policy: matches dangerous pattern '{pattern}'" logger.warning(f"SECURITY: Blocked command: {command}") return False, error # Check against whitelist for pattern in cls.ALLOWED_PATTERNS: if re.match(pattern, command.strip()): logger.debug(f"Command validated: {command}") return True, None # Command not in whitelist error = f"Command not in whitelist: {command}" logger.warning(f"SECURITY: Command rejected (not whitelisted): {command}") return False, error
• openwrt_ssh_mcp/ssh_client.py:89-185 (helper)

  SSH execution function SSHClient.execute that runs commands on the OpenWRT router via asyncssh, handles timeouts, logs execution, and returns structured results with stdout, stderr, exit code, and execution time.
  async def execute(self, command: str, timeout: Optional[int] = None) -> dict: """ Execute a command on the OpenWRT router. Args: command: Command to execute timeout: Execution timeout in seconds (defaults to SSH_TIMEOUT) Returns: dict: Execution result with keys: - success: bool - stdout: str - stderr: str - exit_code: int - execution_time: float """ if not self.is_connected or not self.connection: raise ConnectionError("SSH connection not established. Call connect() first.") if timeout is None: timeout = settings.ssh_timeout start_time = datetime.now() try: logger.debug(f"Executing command: {command}") # Execute command result = await asyncio.wait_for( self.connection.run(command, check=False), timeout=timeout ) execution_time = (datetime.now() - start_time).total_seconds() # Parse result response = { "success": result.exit_status == 0, "stdout": result.stdout.strip() if result.stdout else "", "stderr": result.stderr.strip() if result.stderr else "", "exit_code": result.exit_status, "execution_time": execution_time, } # Log execution audit_logger.log_command( command=command, success=response["success"], output=response["stdout"], error=response["stderr"] if not response["success"] else None, execution_time=execution_time, ) if response["success"]: logger.debug(f"Command succeeded in {execution_time:.2f}s") else: logger.warning( f"Command failed with exit code {response['exit_code']}: {response['stderr']}" ) return response except asyncio.TimeoutError: execution_time = (datetime.now() - start_time).total_seconds() error = f"Command execution timed out after {timeout}s" logger.error(error) audit_logger.log_command( command=command, success=False, error=error, execution_time=execution_time, ) return { "success": False, "stdout": "", "stderr": error, "exit_code": -1, "execution_time": execution_time, } except Exception as e: execution_time = (datetime.now() - start_time).total_seconds() error = f"Command execution error: {str(e)}" logger.error(error) audit_logger.log_command( command=command, success=False, error=error, execution_time=execution_time, ) return { "success": False, "stdout": "", "stderr": error, "exit_code": -1, "execution_time": execution_time, }