execute_command
Execute commands in a Kali Linux penetration testing environment via MCP server, enabling security testing, vulnerability scanning, and password cracking without interactive prompts.
Instructions
(无需交互式比如ping 127.0.0.1)在Kali Linux渗透测试环境中执行命令。支持所有Kali Linux内置的安全测试工具和常规Linux命令。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | 要在Kali Linux环境中执行的命令。可以是任何安全测试、漏洞扫描、密码破解等渗透测试命令。 |
Implementation Reference
- src/index.ts:76-89 (schema)Tool schema definition for 'execute_command', including input schema requiring 'command' parameter.
{ name: "execute_command", description: "(无需交互式比如ping 127.0.0.1)在Kali Linux渗透测试环境中执行命令。支持所有Kali Linux内置的安全测试工具和常规Linux命令。", inputSchema: { type: "object", properties: { command: { type: "string", description: "要在Kali Linux环境中执行的命令。可以是任何安全测试、漏洞扫描、密码破解等渗透测试命令。" } }, required: ["command"] } }, - src/index.ts:73-156 (registration)Registration of 'execute_command' tool in the ListToolsRequestSchema handler.
server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: [ { name: "execute_command", description: "(无需交互式比如ping 127.0.0.1)在Kali Linux渗透测试环境中执行命令。支持所有Kali Linux内置的安全测试工具和常规Linux命令。", inputSchema: { type: "object", properties: { command: { type: "string", description: "要在Kali Linux环境中执行的命令。可以是任何安全测试、漏洞扫描、密码破解等渗透测试命令。" } }, required: ["command"] } }, { name: "start_interactive_command", description: "(需要交互式比如mysql -u root -p)在Kali Linux环境中启动一个交互式命令，并返回会话ID。交互式命令可以接收用户输入,可以在不close_interactive_command的情况下同时执行execute_command。", inputSchema: { type: "object", properties: { command: { type: "string", description: "要在Kali Linux环境中执行的交互式命令。" } }, required: ["command"] } }, { name: "send_input_to_command", description: "(自行判断是AI输入还是用户手动输入)向正在运行的交互式命令发送用户输入。", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "交互式会话ID。" }, input: { type: "string", description: "发送给命令的输入文本。" }, end_line: { type: "boolean", description: "是否在输入后添加换行符。默认为true。" } }, required: ["session_id", "input"] } }, { name: "get_command_output", description: "获取交互式命令的最新输出。", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "交互式会话ID。" } }, required: ["session_id"] } }, { name: "close_interactive_command", description: "关闭交互式命令会话。", inputSchema: { type: "object", properties: { session_id: { type: "string", description: "交互式会话ID。" } }, required: ["session_id"] } } ] }; }); - src/index.ts:175-209 (handler)MCP CallToolRequest handler case for 'execute_command' that validates input, calls CommandExecutor.executeCommand with realtime, and formats response.
case "execute_command": { const command = String(request.params.arguments?.command); if (!command) { throw new McpError(ErrorCode.InvalidParams, "命令是必需的"); } const env = {}; const timeout = 30000000; try { log.info(`准备执行命令: ${command}`); // 执行命令，启用实时推送 const result = await commandExecutor.executeCommand(command, { timeout: timeout, env: env as Record<string, string>, enableRealtime: true }); log.info("命令执行成功"); return { content: [{ type: "text", text: `命令输出:\nstdout: ${result.stdout}\nstderr: ${result.stderr}` }] }; } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); log.error(`命令执行失败: ${errorMessage}`); throw new McpError( ErrorCode.InternalError, `无法执行Kali Linux命令: ${errorMessage}` ); } } - src/executor.ts:255-316 (handler)Core implementation of command execution via SSH in CommandExecutor.executeCommand, supporting realtime delegation, timeout, env vars, and output cleaning.
async executeCommand( command: string, options: { timeout?: number; // 命令执行超时时间(毫秒) cwd?: string; // 工作目录 env?: Record<string, string>; // 环境变量 enableRealtime?: boolean; // 是否启用实时推送 } = {} ): Promise<{ stdout: string; stderr: string }> { const { timeout = 30000000, cwd = '/', env = {}, enableRealtime = false } = options; if (!this.isConnected) { throw new Error('SSH未连接，请先调用connect方法'); } // 如果启用实时推送，使用实时执行方法 if (enableRealtime) { return await this.executeCommandWithRealtime(command, { timeout, cwd, env }); } try { log.info(`执行命令: ${command}`); log.debug(`命令超时: ${timeout}ms, 工作目录: ${cwd}`); // 如果有环境变量，构建环境变量设置命令 let execCommand = command; if (Object.keys(env).length > 0) { const envSetup = Object.entries(env) .map(([key, value]) => `export ${key}="${String(value).replace(/"/g, '\\"')}"`) .join(' && '); execCommand = `${envSetup} && ${command}`; } // 执行命令，带超时控制 const result = await Promise.race([ this.ssh.execCommand(execCommand, { cwd }), new Promise<never>((_, reject) => { setTimeout(() => { reject(new Error('命令执行超时')); }, timeout); }) ]) as { stdout: string; stderr: string }; log.debug(`命令执行完成，stdout长度: ${result.stdout.length}, stderr长度: ${result.stderr.length}`); // 清理输出中的ANSI转义序列 return { stdout: stripAnsiCodes(result.stdout), stderr: stripAnsiCodes(result.stderr) }; } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); log.error(`命令执行失败: ${errorMessage}`); if (errorMessage.includes('超时')) { // 超时情况返回已收集的输出 return { stdout: '命令执行时间过长，已被中断', stderr: '命令执行超时' }; } throw error; } } - src/executor.ts:13-36 (helper)Helper function to strip ANSI escape codes from command output for clean text response.
export function stripAnsiCodes(str: string): string { if (!str) return ''; // 先移除所有零值字符（\u0000）和其他控制字符 const strWithoutControlChars = str.replace(/[\x00-\x08\x0B-\x1F]/g, ''); // 分步处理不同类型的ANSI序列和控制码 let result = strWithoutControlChars // 移除标准ANSI转义序列 .replace(/\x1B(?:[@-Z\\-_]|\[[0-9?;]*[0-9A-Za-z])/g, '') // 移除括号B开头的控制序列 (B[0;7m(B[m 等 .replace(/\(B\[[0-9;]*m/g, '') // 移除单纯的(B序列 .replace(/\(B/g, '') // 移除[?数字h格式的序列 [?2004h 等 .replace(/\[\?[0-9]+[a-z]/g, '') // 移除单独出现的控制序列（如[0m, [4m等） .replace(/\[[0-9;]+m/g, ''); // 处理可能遗漏的其他特殊序列 result = result.replace(/\[[0-9;]*[A-Za-z]/g, ''); // 任何剩余的 [数字字母 序列 return result; }