feat: 支持 shell 变量定义（VAR=value command）、export、多行命令

完整 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

@@ -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:
@@ -71,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 not in_single and not in_double:
-            # 包含重定向符号时，前面的部分也是可执行段
+            if ch in (';', '\n'):
-            if buf:
+                if buf:
-                segments.append(''.join(buf).strip())
+                    segments.append(''.join(buf).strip())
-                buf = []
+                    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  # & 符号自身不产生段
+                if i + 1 < len(command) and command[i + 1] == '&':
-            continue
+                    # && — 分割命令
+                    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]
@@ -104,21 +131,32 @@ 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
 
 
 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:
@@ -190,7 +228,7 @@ SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = {
     },
     "journalctl": None,  # 只读，任意参数 (-u, -n, --no-pager, --since 等)
     "crictl": {"ps", "inspect", "logs", "info", "version", "images", "stats"},
-    "ctr": {"containers", "images", "namespaces", "tasks", "snapshots", "info", "version", "plugins", "content", "leases"},
+    "ctr": None,
     "nerdctl": {"ps", "logs", "inspect", "info", "version", "images", "stats", "top", "events", "system", "namespace", "network", "volume"},
     "docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
     "df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
@@ -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,
+    "test": None, "[": None, "sleep": None, "source": None, ".": None,
-    "timeout": None, "nohup": 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 中的专用块处理 (两段式命令)
@@ -284,63 +325,126 @@ 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}")
+        lines = command.split("\n") if "\n" in command else [command]
-        # 先检查写表
+        overall_mode = "read"
-        if binary in SHELL_WRITE_COMMANDS:
+        overall_approval = False
-            allowed = SHELL_WRITE_COMMANDS[binary]
+        first_binary = None
-            if allowed is None:
+        first_segment = ""
-                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}")
 
-        return CommandDecision("read", False, binary, [command], needs_shell=True)
+        for line in lines:
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            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
+                    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}")
+
+        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]
-    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}")
 
@@ -348,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}")
 
@@ -358,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)
@@ -372,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