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 等
This commit is contained in:
Your Name
+163
-51
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user