From e26a3e559fc68a95fa49bfb49df23152e51b0164 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sat, 25 Jul 2026 16:25:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=20shell=20=E5=8F=98?= =?UTF-8?q?=E9=87=8F=E5=AE=9A=E4=B9=89=EF=BC=88VAR=3Dvalue=20command?= =?UTF-8?q?=EF=BC=89=E3=80=81export=E3=80=81=E5=A4=9A=E8=A1=8C=E5=91=BD?= =?UTF-8?q?=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 完整 shell 能力支持： - VAR=value command 语法（自动跳过赋值检测实际 binary） - export VAR=value 支持（export 加入只读白名单） - 多行命令（\n 分割后逐段检查） - && || ; 等连接符的武器级命令检测 - _split_into_commands 重写：使用 while 索引遍历，正确处理 && / || / | - _command_has_dangerous_cmds 跳过 VAR=value 和 export 前缀 - classify_agent_command 检测 VAR=value kubectl 语法路由到 kubectl 分类器 - execute_command 支持变量赋值后的 binary 定位 - 新增 shell 内置白名单：echo printf cd pwd export source . sleep timeout nohup unset shift return local readonly trap type command set shopt eval exec alias unalias declare typeset read mapfile - 白名单只读命令新增：timeout、nohup、sleep 等 --- backend/agent_tools.py | 214 +++++++++++++++++++++++++++++++---------- 1 file changed, 163 insertions(+), 51 deletions(-) diff --git a/backend/agent_tools.py b/backend/agent_tools.py index 7cf3257..f86054f 100644 --- a/backend/agent_tools.py +++ b/backend/agent_tools.py @@ -73,27 +73,52 @@ def _has_shell_operators(command: str) -> bool: def _split_into_commands(command: str) -> list[str]: - """按 shell 连接符（; && || | > >> < << 等）分割命令块。 + """按 shell 连接符（; && || | \n 等）分割命令块。 - 返回每个可执行段，忽略引号内的操作符。用于白名单检查。 + 返回每个可执行段，忽略引号内的操作符。重定向符号 > < 不分割。 """ segments = [] buf = [] in_single = in_double = False - for ch in command: + i = 0 + while i < len(command): + ch = command[i] if ch == "'" and not in_double: in_single = not in_single elif ch == '"' and not in_single: in_double = not in_double - if not in_single and not in_double and ch in ('|', ';', '&', '>', '<'): - # 包含重定向符号时，前面的部分也是可执行段 - if buf: - segments.append(''.join(buf).strip()) - buf = [] + if not in_single and not in_double: + if ch in (';', '\n'): + if buf: + segments.append(''.join(buf).strip()) + buf = [] + i += 1 + continue + if ch == '|': + if i + 1 < len(command) and command[i + 1] == '|': + # || — 分割命令 + if buf: + segments.append(''.join(buf).strip()) + buf = [] + i += 2 + continue + # 单 | — 分割命令 + if buf: + segments.append(''.join(buf).strip()) + buf = [] + i += 1 + continue if ch == '&': - continue # & 符号自身不产生段 - continue + if i + 1 < len(command) and command[i + 1] == '&': + # && — 分割命令 + if buf: + segments.append(''.join(buf).strip()) + buf = [] + i += 2 + continue + # 单独 & — 保留 buf.append(ch) + i += 1 if buf: segments.append(''.join(buf).strip()) return [s for s in segments if s] @@ -106,8 +131,21 @@ def _command_has_dangerous_cmds(command: str) -> bool: seg_args = shlex.split(segment) except ValueError: continue - if seg_args and seg_args[0] in _ALWAYS_DENY_CMDS: + if not seg_args: + continue + # 跳过 VAR=value 赋值 + idx = 0 + while idx < len(seg_args) and "=" in seg_args[idx] and not seg_args[idx].startswith("-"): + idx += 1 + if idx < len(seg_args) and seg_args[idx] in _ALWAYS_DENY_CMDS: return True + # 也检查 export VAR=value 后的命令 + if idx < len(seg_args) and seg_args[idx] == "export": + idx2 = idx + 1 + while idx2 < len(seg_args) and "=" in seg_args[idx2]: + idx2 += 1 + if idx2 < len(seg_args) and seg_args[idx2] in _ALWAYS_DENY_CMDS: + return True return False @@ -209,8 +247,11 @@ SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = { "nc": None, "nmap": None, "telnet": None, "tcpdump": None, "ethtool": None, "lspci": None, "lsusb": None, "dmidecode": None, "sysctl": None, "udevadm": None, "echo": None, "printf": None, "cd": None, "pwd": None, "export": None, - "test": None, "[": None, "sleep": None, - "timeout": None, "nohup": None, + "test": None, "[": None, "sleep": None, "source": None, ".": None, + "timeout": None, "nohup": None, "unset": None, "shift": None, "return": None, + "local": None, "readonly": None, "trap": None, "type": None, "command": None, + "set": None, "shopt": None, "eval": None, "exec": None, "alias": None, + "unalias": None, "declare": None, "typeset": None, "read": None, "mapfile": None, } # etcdctl 由 classify_shell_command 中的专用块处理 (两段式命令) @@ -288,48 +329,76 @@ def classify_shell_command(command: str) -> CommandDecision: if _command_has_dangerous_cmds(command): raise ValueError("命令包含不允许的危险命令（sh/bash/dd/mkfs 等），请手动执行") - # 多行命令：按行分割后逐行检查，取最高权限模式 + # 多行或含操作符的命令：按行分割后，对每行逐段检查，取最高权限模式 lines = command.split("\n") if "\n" in command else [command] overall_mode = "read" overall_approval = False first_binary = None + first_segment = "" 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: + segments = _split_into_commands(line) + for segment in segments: + # 检测 kubectl 命令 + if segment.startswith("kubectl ") or segment == "kubectl": + try: + kdec = classify_kubectl_command(segment) + except (ValueError, IndexError) as exc: + raise ValueError(str(exc)) + if first_binary is None: + first_binary = "kubectl" + first_segment = segment + if kdec.mode == "write": 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 + continue + + try: + seg_parts = shlex.split(segment) + except ValueError: + continue + if not seg_parts: + continue + # 跳过 VAR=value 赋值前缀（含 export） + idx = 0 + while idx < len(seg_parts): + if "=" in seg_parts[idx] and not seg_parts[idx].startswith("-"): + idx += 1 + elif seg_parts[idx] == "export": + idx += 1 + else: + break + if idx >= len(seg_parts): + continue # 纯赋值段，跳过 + binary = seg_parts[idx] + check_parts = seg_parts[idx:] + 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 in SHELL_WRITE_COMMANDS: + allowed = SHELL_WRITE_COMMANDS[binary] + if allowed is None: + overall_mode = "write" + overall_approval = True + else: + verb = _first_non_flag(check_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}") - else: - raise ValueError(f"命令 {binary} 不支持该子命令: {verb}") if first_binary is None: raise ValueError("命令不能为空") @@ -338,30 +407,44 @@ def classify_shell_command(command: str) -> CommandDecision: parts = parse_shell_command(command) binary = parts[0] - verb = _first_non_flag(parts[1:]) + # 支持 VAR=value command 语法 + start_idx = 0 + while start_idx < len(parts) and "=" in parts[start_idx] and not parts[start_idx].startswith("-"): + start_idx += 1 + if start_idx > 0: + if start_idx < len(parts): + binary = parts[start_idx] + effective_parts = parts[start_idx:] + else: + # 纯赋值行如 LOG_DIR=/tmp/log — 放行 + return CommandDecision("read", False, parts[0], parts[1:]) + verb = _first_non_flag(effective_parts[1:]) + else: + effective_parts = parts + verb = _first_non_flag(parts[1:]) if binary in SHELL_WRITE_COMMANDS: allowed = SHELL_WRITE_COMMANDS[binary] if allowed is None or (verb is not None and verb in allowed): - return CommandDecision("write", True, binary, parts[1:]) + return CommandDecision("write", True, binary, effective_parts[1:]) # etcdctl: 单段式或两段式命令，verb/subverb 均跳过全局 flag (--endpoints 等) # member/alarm/auth 等同时有只读和写子命令，需按 (verb, subverb) 组合精确判断 if binary == "etcdctl": - non_flags = [a for a in parts[1:] if not a.startswith("-")] + non_flags = [a for a in effective_parts[1:] if not a.startswith("-")] ev = non_flags[0].lower() if non_flags else None esub = non_flags[1].lower() if len(non_flags) > 1 else None # 单段式 verb if ev in _ETCDCTL_READ_VERBS: - return CommandDecision("read", False, ev, parts[1:]) + return CommandDecision("read", False, ev, effective_parts[1:]) if ev in _ETCDCTL_WRITE_VERBS: - return CommandDecision("write", True, ev, parts[1:]) + return CommandDecision("write", True, ev, effective_parts[1:]) # 两段式: 先查写子命令 (精确匹配 subverb)，再查只读子命令 if ev in _ETCDCTL_WRITE_SUBVERBS and esub in _ETCDCTL_WRITE_SUBVERBS[ev]: - return CommandDecision("write", True, ev, parts[1:]) + return CommandDecision("write", True, ev, effective_parts[1:]) if ev in _ETCDCTL_READ_SUBVERBS: if esub is None or esub in _ETCDCTL_READ_SUBVERBS[ev]: - return CommandDecision("read", False, ev, parts[1:]) + return CommandDecision("read", False, ev, effective_parts[1:]) raise ValueError(f"etcdctl {ev} 不支持子命令: {esub}") raise ValueError(f"etcdctl 不支持该子命令: {ev}") @@ -369,9 +452,9 @@ def classify_shell_command(command: str) -> CommandDecision: allowed = SHELL_READ_ONLY_COMMANDS[binary] if allowed is not None and (verb is None or verb not in allowed): raise ValueError(f"命令 {binary} 不支持该子命令: {verb}") - if binary == "ip" and any(arg in _IP_WRITE_ACTIONS for arg in parts[1:]): + if binary == "ip" and any(arg in _IP_WRITE_ACTIONS for arg in effective_parts[1:]): raise ValueError("ip 命令包含写操作，请使用只读形式如 'ip addr show'") - return CommandDecision("read", False, binary, parts[1:]) + return CommandDecision("read", False, binary, effective_parts[1:]) raise ValueError(f"不允许执行该命令: {binary}") @@ -379,6 +462,29 @@ def classify_shell_command(command: str) -> CommandDecision: def classify_agent_command(command: str) -> CommandDecision: """统一分类 Agent 命令：kubectl 或白名单系统诊断命令。""" stripped = (command or "").strip() + # 检测 VAR=value kubectl 语法（有变量赋值时） + rest = stripped + while True: + try: + first = rest.split(None, 1)[0] + except IndexError: + break + if "=" in first and not first.startswith("-"): + idx = rest.index("=") + # 必须是合法变量赋值 VAR=value + parts_split = rest.split(None, 1) + if len(parts_split) < 2: + break + rest = parts_split[1] + elif first == "export": + parts_split = rest.split(None, 1) + if len(parts_split) < 2: + break + rest = parts_split[1] + else: + break + if rest.startswith("kubectl ") or rest == "kubectl": + return classify_kubectl_command(rest) if stripped.startswith("kubectl ") or stripped == "kubectl": return classify_kubectl_command(stripped) return classify_shell_command(stripped) @@ -393,6 +499,12 @@ async def execute_command(command: str, timeout: int = 30) -> dict: """ decision = classify_agent_command(command) parts = _parse_command(command) + # 支持 VAR=value command 语法，跳过赋值找实际 binary + cmd_idx = 0 + while cmd_idx < len(parts) and "=" in parts[cmd_idx] and not parts[cmd_idx].startswith("-"): + cmd_idx += 1 + if cmd_idx > 0 and cmd_idx < len(parts): + parts = parts[cmd_idx:] if parts[0] == "kubectl": executable = KUBECTL args = decision.args