diff --git a/backend/agent_tools.py b/backend/agent_tools.py index bb0fe41..917e782 100644 --- a/backend/agent_tools.py +++ b/backend/agent_tools.py @@ -33,9 +33,10 @@ GLOBAL_FLAGS_WITH_VALUE = { } FORBIDDEN_TOKENS = {";", "||", "&&", "<", "<<", "`"} # 管道 |、重定向 >、>>、$ 变量引用 允许在诊断中使用 -# 但含 | 的命令中若管道后出现 sh/bash/rm/mv/chmod/chown/dd/format/mkfs 则拒绝 -_PIPE_DANGEROUS_TARGETS = {"sh", "bash", "dash", "zsh", "rm", "mv", "chmod", "chown", - "dd", "format", "mkfs", "fdisk", "parted", "wget", "curl"} +# 但含 | 的命令中若管道后出现 sh/bash/dd/format/mkfs/fdisk/parted 则直接拒绝(武器级危险) +# rm/mv/chmod/chown — 现已在 SHELL_WRITE_COMMANDS 中,管道后会触发审批而非直接拒绝 +_PIPE_DANGEROUS_TARGETS = {"sh", "bash", "dash", "zsh", + "dd", "format", "mkfs", "fdisk", "parted"} @dataclass(frozen=True) @@ -258,6 +259,15 @@ _ETCDCTL_WRITE_SUBVERBS = { # 系统修改命令白名单:需要人工审批 SHELL_WRITE_COMMANDS: dict[str, set[str] | None] = { "systemctl": {"restart", "start", "stop", "reload", "enable", "disable", "daemon-reload"}, + # 文件操作命令:任意参数都需审批 + "rm": None, + "cp": None, + "mv": None, + "chmod": None, + "chown": None, + "mkdir": None, + "touch": None, + "ln": None, } # ip 命令的写操作动词 (出现即拒绝自动执行) @@ -338,6 +348,12 @@ def classify_shell_command(command: str) -> CommandDecision: verb = _first_non_flag(before_parts[1:]) if verb is not None and verb in allowed: return CommandDecision("write", True, binary, [command], needs_shell=True) + # binary 在写表但子命令不匹配写表 — 走只读白名单检查 + 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) diff --git a/tests/test_agent_tools.py b/tests/test_agent_tools.py index ef254ea..f85e4f2 100644 --- a/tests/test_agent_tools.py +++ b/tests/test_agent_tools.py @@ -84,8 +84,13 @@ class ShellCommandPolicyTest(unittest.TestCase): self.assertTrue(decision.requires_approval) def test_rejects_unknown_binary(self): + # rm 现在在写白名单中(需审批),不会直接拒绝 + decision = classify_shell_command("rm -rf /tmp/cache") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + # 完全不在任何白名单中的命令才拒绝 with self.assertRaises(ValueError): - classify_shell_command("rm -rf /") + classify_shell_command("hacktool -x") def test_rejects_shell_operators_in_shell_command(self): # 分号 ; 仍被拒绝