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

@@ -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 not in_single and not in_double:
-            # 包含重定向符号时，前面的部分也是可执行段
+            if ch in (';', '\n'):
                 if buf:
                     segments.append(''.join(buf).strip())
                     buf = []
-            if ch == '&':
+                i += 1
-                continue  # & 符号自身不产生段
                 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 == '&':
+                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,7 +131,20 @@ 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,
+    "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 中的专用块处理 (两段式命令)
@@ -288,44 +329,72 @@ 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
+            segments = _split_into_commands(line)
+            for segment in segments:
+                # 检测 kubectl 命令
+                if segment.startswith("kubectl ") or segment == "kubectl":
                     try:
-                line_parts = shlex.split(line)
+                        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
-            binary = line_parts[0]
+                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 == "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:])
+                        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
+                                pass  # 该段是只读，不改变 overall
                             else:
                                 raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
                         else:
@@ -338,30 +407,44 @@ def classify_shell_command(command: str) -> CommandDecision:
 
     parts = parse_shell_command(command)
     binary = parts[0]
+    # 支持 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