Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a964a23b8f | |||
| defb09ef97 |
+41
-20
@@ -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:
|
||||||
@@ -190,7 +190,7 @@ SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = {
|
|||||||
},
|
},
|
||||||
"journalctl": None, # 只读,任意参数 (-u, -n, --no-pager, --since 等)
|
"journalctl": None, # 只读,任意参数 (-u, -n, --no-pager, --since 等)
|
||||||
"crictl": {"ps", "inspect", "logs", "info", "version", "images", "stats"},
|
"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"},
|
"nerdctl": {"ps", "logs", "inspect", "info", "version", "images", "stats", "top", "events", "system", "namespace", "network", "volume"},
|
||||||
"docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
|
"docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
|
||||||
"df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
|
"df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
|
||||||
@@ -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]
|
||||||
|
|||||||
Reference in New Issue
Block a user