feat: 全面放开 shell 命令连接符(; && || > >> < 等)

现在 AI Agent 支持在单条命令中使用:
- 管道 (|): crictl ps -a | grep nginx | head -5
- 重定向 (> / >>): journalctl -u kubelet > /tmp/log
- 分号 (;): systemctl status kubelet; df -h
- 逻辑连接符 (&& / ||): df -h && free -m

安全策略:
- FORBIDDEN_TOKENS 清空,所有 shell 操作符均放行
- 新增 _split_into_commands() 按操作符分割命令段
- 新增 _command_has_dangerous_cmds() 扫描所有段中的武器级命令
- 永远拒绝: sh/bash/dash/zsh/dd/format/.../parted 等
- 检查第一条命令的白名单,cat 不安全路径检查
- 只读白名单新增 echo printf cd pwd export test [
This commit is contained in:
Your Name
2026-07-25 14:48:32 +08:00
parent 3c3b7abda5
commit 00d95c4c5e
3 changed files with 72 additions and 91 deletions
+13 -8
View File
@@ -23,10 +23,7 @@ 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 等危险命令仍被拒绝
# 武器级危险命令(sh/bash仍被拒绝
with self.assertRaises(ValueError):
parse_kubectl_command("kubectl get pods | sh")
# 非 kubectl 命令仍被拒绝
@@ -93,18 +90,26 @@ class ShellCommandPolicyTest(unittest.TestCase):
classify_shell_command("hacktool -x")
def test_rejects_shell_operators_in_shell_command(self):
# 分号 ; 仍被拒绝
with self.assertRaises(ValueError):
parse_shell_command("systemctl status kubelet; rm -rf /")
# 管道 | 和重定向 > 现在允许通过分类器(server 端会用 shell=True 执行)
# 所有 shell 操作符(; | > && ||)现在都允许通过分类器
# 分号 ;
decision = classify_shell_command("systemctl status kubelet; df -h")
self.assertEqual(decision.mode, "read")
self.assertTrue(decision.needs_shell)
# 管道 |
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)
# 武器级 sh/bash 仍拒绝
with self.assertRaises(ValueError):
classify_shell_command("crictl ps -a | bash")
with self.assertRaises(ValueError):
classify_shell_command("crictl ps -a && sh -c 'echo x'")
def test_cat_only_allows_safe_paths(self):
decision = classify_shell_command("cat /etc/os-release")