From a964a23b8fd21cce37a3f7a9b63aa3b2bd088eaf Mon Sep 17 00:00:00 2001 From: Your Name Date: Sat, 25 Jul 2026 15:27:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=A4=9A=E8=A1=8C?= =?UTF-8?q?=E5=91=BD=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 多行命令通过 create_subprocess_shell 执行，逐行检查白名单： - 每行独立检查 binary 是否在白名单中 - 取所有行的最高权限模式（如有任何一行是 write，整体需审批） - 去除 _parse_command 中的多行拒绝逻辑 - _has_shell_operators 检测到 \n 也返回 True --- backend/agent_tools.py | 81 ++++++++++++++++++++++++++---------------- 1 file changed, 51 insertions(+), 30 deletions(-) diff --git a/backend/agent_tools.py b/backend/agent_tools.py index 9d81e43..7cf3257 100644 --- a/backend/agent_tools.py +++ b/backend/agent_tools.py @@ -48,10 +48,12 @@ class CommandDecision: def _has_shell_operators(command: str) -> bool: - """检测命令是否包含 shell 操作符（管道、重定向、分号、逻辑连接符等）。""" + """检测命令是否包含 shell 操作符（管道、重定向、分号、逻辑连接符、多行等）。""" stripped = command.strip() if not stripped: return False + if "\n" in stripped or "\r" in stripped: + return True # 多行命令需要 shell=True in_single = in_double = False for i, ch in enumerate(stripped): if ch == "'" and not in_double: @@ -112,13 +114,11 @@ def _command_has_dangerous_cmds(command: str) -> bool: def _parse_command(command: str) -> list[str]: """解析命令，检查合法性。 - 允许所有 shell 操作符（| > ; && || < 等），安全靠白名单检查。 + 允许所有 shell 操作符（| > ; && || < 等）以及多行命令，安全靠白名单检查。 武器级危险命令 sh/bash/dd/mkfs 等在 classify 阶段处理。 """ if not isinstance(command, str) or not command.strip(): raise ValueError("命令不能为空") - if "\n" in command or "\r" in command: - raise ValueError("不允许多行命令") if _command_has_dangerous_cmds(command): raise ValueError("命令包含不允许的危险命令（sh/bash/dd/mkfs 等），请手动执行") try: @@ -284,36 +284,57 @@ def classify_shell_command(command: str) -> CommandDecision: needs_shell = _has_shell_operators(command) if needs_shell: - # 含 shell 操作符：先检查武器级危险命令 + # 含 shell 操作符或为多行命令：先检查武器级危险命令 if _command_has_dangerous_cmds(command): raise ValueError("命令包含不允许的危险命令（sh/bash/dd/mkfs 等），请手动执行") - # 取第一个可执行段的 binary 做白名单检查 - segments = _split_into_commands(command) - first_seg = segments[0] if segments else command - try: - first_parts = shlex.split(first_seg) - except ValueError: - first_parts = [first_seg] - binary = first_parts[0] - if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS: - raise ValueError(f"不允许执行该命令: {binary}") - # 先检查写表 - if binary in SHELL_WRITE_COMMANDS: - allowed = SHELL_WRITE_COMMANDS[binary] - if allowed is None: - return CommandDecision("write", True, binary, [command], needs_shell=True) - verb = _first_non_flag(first_parts[1:]) - if verb is not None and verb in allowed: - return CommandDecision("write", True, binary, [command], needs_shell=True) - # 写表子命令不匹配 — 看是否在只读表中 - if binary in SHELL_READ_ONLY_COMMANDS: - ro_allowed = SHELL_READ_ONLY_COMMANDS[binary] - if ro_allowed is None or (verb is not None and verb in ro_allowed): - return CommandDecision("read", False, binary, [command], needs_shell=True) - raise ValueError(f"命令 {binary} 不支持该子命令: {verb}") + # 多行命令：按行分割后逐行检查，取最高权限模式 + lines = command.split("\n") if "\n" in command else [command] + overall_mode = "read" + overall_approval = False + first_binary = None - return CommandDecision("read", False, binary, [command], needs_shell=True) + for line in lines: + line = line.strip() + if not line or line.startswith("#"): + continue + try: + line_parts = shlex.split(line) + except ValueError: + continue + binary = line_parts[0] + if first_binary is None: + first_binary = binary + + if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS: + raise ValueError(f"不允许执行该命令: {binary}") + + if binary == "cat" and len(line_parts) > 1: + pass # cat 不限制路径 + + if binary in SHELL_WRITE_COMMANDS: + allowed = SHELL_WRITE_COMMANDS[binary] + if allowed is None: + overall_mode = "write" + overall_approval = True + else: + verb = _first_non_flag(line_parts[1:]) + if verb is not None and verb in allowed: + overall_mode = "write" + overall_approval = True + elif binary in SHELL_READ_ONLY_COMMANDS: + ro_allowed = SHELL_READ_ONLY_COMMANDS[binary] + if ro_allowed is None or (verb is not None and verb in ro_allowed): + pass # 该行是只读，不改变 overall + else: + raise ValueError(f"命令 {binary} 不支持该子命令: {verb}") + else: + raise ValueError(f"命令 {binary} 不支持该子命令: {verb}") + + if first_binary is None: + raise ValueError("命令不能为空") + + return CommandDecision(overall_mode, overall_approval, first_binary, [command], needs_shell=True) parts = parse_shell_command(command) binary = parts[0]