fix: 允许 AI Agent 使用管道(|)和重定向(>)进行精准排查

之前禁止所有 shell 操作符,导致 crictl ps -a|tail / df -h | grep vda
这类常用诊断管道无法使用。

改动:
1. FORBIDDEN_TOKENS 移除了 | > >> $,保留 ; || && < << `
2. 新增 _has_shell_operators() 检测命令是否需要 shell 执行
3. 新增 _pipe_to_dangerous_target() 阻止管道后执行 sh/bash/rm 等危险命令
4. classify_shell_command 对含管道命令: 只检查管道前 binary 白名单,
   子命令不做限制(管道后工具多样,精确分类无意义)
5. execute_command 对 needs_shell=True 的命令用 create_subprocess_shell 执行
6. AI Prompt 第1条安全规则改为允许管道和重定向
7. 分号仍然禁止;cat 不安全路径检查对管道场景同样生效
This commit is contained in:
Your Name
2026-07-25 14:40:12 +08:00
parent b19975c369
commit 8852e6d0ca
3 changed files with 178 additions and 22 deletions
+19 -10
View File
@@ -23,9 +23,14 @@ class KubectlCommandPolicyTest(unittest.TestCase):
self.assertTrue(decision.requires_approval)
def test_rejects_shell_operators_and_non_kubectl_commands(self):
# 分号 ; 仍被拒绝
with self.assertRaises(ValueError):
parse_kubectl_command("kubectl get pods; rm -rf /")
# 管道后的 sh/bash 等危险命令仍被拒绝
with self.assertRaises(ValueError):
parse_kubectl_command("kubectl get pods | sh")
# 非 kubectl 命令仍被拒绝
for command in (
"kubectl get pods; rm -rf /",
"kubectl get pods | sh",
"bash -c kubectl get pods",
"sudo kubectl get pods",
):
@@ -83,14 +88,18 @@ class ShellCommandPolicyTest(unittest.TestCase):
classify_shell_command("rm -rf /")
def test_rejects_shell_operators_in_shell_command(self):
for command in (
"systemctl status kubelet; rm -rf /",
"df -h | grep vda",
"journalctl -u kubelet > /tmp/log",
):
with self.subTest(command=command):
with self.assertRaises(ValueError):
parse_shell_command(command)
# 分号 ; 仍被拒绝
with self.assertRaises(ValueError):
parse_shell_command("systemctl status kubelet; rm -rf /")
# 管道 | 和重定向 > 现在允许通过分类器(server 端会用 shell=True 执行)
decision = classify_shell_command("df -h | grep vda")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
self.assertTrue(decision.needs_shell)
decision = classify_shell_command("journalctl -u kubelet > /tmp/log")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
self.assertTrue(decision.needs_shell)
def test_cat_only_allows_safe_paths(self):
decision = classify_shell_command("cat /etc/os-release")