feat: 支持多行命令

多行命令通过 create_subprocess_shell 执行,逐行检查白名单:
- 每行独立检查 binary 是否在白名单中
- 取所有行的最高权限模式(如有任何一行是 write,整体需审批)
- 去除 _parse_command 中的多行拒绝逻辑
- _has_shell_operators 检测到 \n 也返回 True
This commit is contained in:
Your Name
2026-07-25 15:27:43 +08:00
parent defb09ef97
commit a964a23b8f
+40 -19
View File
@@ -48,10 +48,12 @@ class CommandDecision:
def _has_shell_operators(command: str) -> bool: def _has_shell_operators(command: str) -> bool:
"""检测命令是否包含 shell 操作符(管道、重定向、分号、逻辑连接符等)。""" """检测命令是否包含 shell 操作符(管道、重定向、分号、逻辑连接符、多行等)。"""
stripped = command.strip() stripped = command.strip()
if not stripped: if not stripped:
return False return False
if "\n" in stripped or "\r" in stripped:
return True # 多行命令需要 shell=True
in_single = in_double = False in_single = in_double = False
for i, ch in enumerate(stripped): for i, ch in enumerate(stripped):
if ch == "'" and not in_double: 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]: def _parse_command(command: str) -> list[str]:
"""解析命令,检查合法性。 """解析命令,检查合法性。
允许所有 shell 操作符(| > ; && || < 等),安全靠白名单检查。 允许所有 shell 操作符(| > ; && || < 等)以及多行命令,安全靠白名单检查。
武器级危险命令 sh/bash/dd/mkfs 等在 classify 阶段处理。 武器级危险命令 sh/bash/dd/mkfs 等在 classify 阶段处理。
""" """
if not isinstance(command, str) or not command.strip(): if not isinstance(command, str) or not command.strip():
raise ValueError("命令不能为空") raise ValueError("命令不能为空")
if "\n" in command or "\r" in command:
raise ValueError("不允许多行命令")
if _command_has_dangerous_cmds(command): if _command_has_dangerous_cmds(command):
raise ValueError("命令包含不允许的危险命令(sh/bash/dd/mkfs 等),请手动执行") raise ValueError("命令包含不允许的危险命令(sh/bash/dd/mkfs 等),请手动执行")
try: try:
@@ -284,36 +284,57 @@ def classify_shell_command(command: str) -> CommandDecision:
needs_shell = _has_shell_operators(command) needs_shell = _has_shell_operators(command)
if needs_shell: if needs_shell:
# 含 shell 操作符:先检查武器级危险命令 # 含 shell 操作符或为多行命令:先检查武器级危险命令
if _command_has_dangerous_cmds(command): if _command_has_dangerous_cmds(command):
raise ValueError("命令包含不允许的危险命令(sh/bash/dd/mkfs 等),请手动执行") raise ValueError("命令包含不允许的危险命令(sh/bash/dd/mkfs 等),请手动执行")
# 取第一个可执行段的 binary 做白名单检查
segments = _split_into_commands(command) # 多行命令:按行分割后逐行检查,取最高权限模式
first_seg = segments[0] if segments else command lines = command.split("\n") if "\n" in command else [command]
overall_mode = "read"
overall_approval = False
first_binary = None
for line in lines:
line = line.strip()
if not line or line.startswith("#"):
continue
try: try:
first_parts = shlex.split(first_seg) line_parts = shlex.split(line)
except ValueError: except ValueError:
first_parts = [first_seg] continue
binary = first_parts[0] 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: if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS:
raise ValueError(f"不允许执行该命令: {binary}") raise ValueError(f"不允许执行该命令: {binary}")
# 先检查写表
if binary == "cat" and len(line_parts) > 1:
pass # cat 不限制路径
if binary in SHELL_WRITE_COMMANDS: if binary in SHELL_WRITE_COMMANDS:
allowed = SHELL_WRITE_COMMANDS[binary] allowed = SHELL_WRITE_COMMANDS[binary]
if allowed is None: if allowed is None:
return CommandDecision("write", True, binary, [command], needs_shell=True) overall_mode = "write"
verb = _first_non_flag(first_parts[1:]) overall_approval = True
else:
verb = _first_non_flag(line_parts[1:])
if verb is not None and verb in allowed: if verb is not None and verb in allowed:
return CommandDecision("write", True, binary, [command], needs_shell=True) overall_mode = "write"
# 写表子命令不匹配 — 看是否在只读表中 overall_approval = True
if binary in SHELL_READ_ONLY_COMMANDS: elif binary in SHELL_READ_ONLY_COMMANDS:
ro_allowed = SHELL_READ_ONLY_COMMANDS[binary] ro_allowed = SHELL_READ_ONLY_COMMANDS[binary]
if ro_allowed is None or (verb is not None and verb in ro_allowed): if ro_allowed is None or (verb is not None and verb in ro_allowed):
return CommandDecision("read", False, binary, [command], needs_shell=True) pass # 该行是只读,不改变 overall
else:
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
else:
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}") raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
return CommandDecision("read", False, binary, [command], needs_shell=True) if first_binary is None:
raise ValueError("命令不能为空")
return CommandDecision(overall_mode, overall_approval, first_binary, [command], needs_shell=True)
parts = parse_shell_command(command) parts = parse_shell_command(command)
binary = parts[0] binary = parts[0]