diff --git a/backend/agent_tools.py b/backend/agent_tools.py index 04a0a79..514c1de 100644 --- a/backend/agent_tools.py +++ b/backend/agent_tools.py @@ -127,6 +127,34 @@ SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = { "timedatectl": None, "date": None, "stat": None, "file": None, "blkid": None, } +# etcdctl / etcdutl 由 classify_shell_command 中的专用块处理 (两段式命令) +# 从通用白名单中移除，避免被单段式逻辑误判 + +# etcdctl 单段式只读 verb +_ETCDCTL_READ_VERBS = {"get", "watch", "version", "lock", "lease", "move-leader", "leader"} +# etcdctl 单段式写 verb (需人工审批) +_ETCDCTL_WRITE_VERBS = {"put", "del", "delete", "compact", "txn", "snapshot"} +# etcdctl 两段式 (verb, subverb) 只读组合 +_ETCDCTL_READ_SUBVERBS = { + "endpoint": {"status", "health", "hashkv"}, + "member": {"list", "list-urls"}, + "alarm": {"list"}, + "auth": {"status"}, + "user": {"list", "get"}, + "role": {"list", "get"}, + "check": {"perf", "datascale"}, +} +# etcdctl 两段式 (verb, subverb) 写组合 (需人工审批) +_ETCDCTL_WRITE_SUBVERBS = { + "member": {"add", "remove", "update"}, + "alarm": {"disarm"}, + "auth": {"enable", "disable"}, + "user": {"add", "delete", "grant-role", "revoke-role", "passwd"}, + "role": {"add", "delete", "grant-permission", "revoke-permission"}, +} +# etcdutl snapshot 只读子命令 +_ETCDUTL_SNAPSHOT_READ = {"status", "hash"} + # 系统修改命令白名单：需要人工审批 SHELL_WRITE_COMMANDS: dict[str, set[str] | None] = { "systemctl": {"restart", "start", "stop", "reload", "enable", "disable", "daemon-reload"}, @@ -178,6 +206,38 @@ def classify_shell_command(command: str) -> CommandDecision: if allowed is None or (verb is not None and verb in allowed): return CommandDecision("write", True, binary, parts[1:]) + # etcdctl: 单段式或两段式命令，verb/subverb 均跳过全局 flag (--endpoints 等) + # member/alarm/auth 等同时有只读和写子命令，需按 (verb, subverb) 组合精确判断 + if binary == "etcdctl": + non_flags = [a for a in parts[1:] if not a.startswith("-")] + ev = non_flags[0].lower() if non_flags else None + esub = non_flags[1].lower() if len(non_flags) > 1 else None + # 单段式 verb + if ev in _ETCDCTL_READ_VERBS: + return CommandDecision("read", False, ev, parts[1:]) + if ev in _ETCDCTL_WRITE_VERBS: + return CommandDecision("write", True, ev, parts[1:]) + # 两段式: 先查写子命令 (精确匹配 subverb)，再查只读子命令 + if ev in _ETCDCTL_WRITE_SUBVERBS and esub in _ETCDCTL_WRITE_SUBVERBS[ev]: + return CommandDecision("write", True, ev, parts[1:]) + if ev in _ETCDCTL_READ_SUBVERBS: + if esub is None or esub in _ETCDCTL_READ_SUBVERBS[ev]: + return CommandDecision("read", False, ev, parts[1:]) + raise ValueError(f"etcdctl {ev} 不支持子命令: {esub}") + raise ValueError(f"etcdctl 不支持该子命令: {ev}") + + if binary == "etcdutl": + non_flags = [a for a in parts[1:] if not a.startswith("-")] + ev = non_flags[0].lower() if non_flags else None + esub = non_flags[1].lower() if len(non_flags) > 1 else None + if ev == "snapshot": + if esub in _ETCDUTL_SNAPSHOT_READ: + return CommandDecision("read", False, ev, parts[1:]) + return CommandDecision("write", True, ev, parts[1:]) + if ev == "version": + return CommandDecision("read", False, ev, parts[1:]) + raise ValueError(f"etcdutl 不支持该子命令: {ev}") + if binary in SHELL_READ_ONLY_COMMANDS: allowed = SHELL_READ_ONLY_COMMANDS[binary] if allowed is not None and (verb is None or verb not in allowed): diff --git a/backend/ai_agent.py b/backend/ai_agent.py index 3404ce3..421902b 100644 --- a/backend/ai_agent.py +++ b/backend/ai_agent.py @@ -30,6 +30,7 @@ AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职 可用工具： 1. kubectl 命令 - 查询集群内资源状态。例如: kubectl get pods -A, kubectl describe node , kubectl logs -n 2. 系统诊断 shell 命令 - 查询节点宿主机状态。例如: systemctl status kubelet, journalctl -u kubelet -n 100 --no-pager, crictl ps, df -h, free -m, ss -tlnp, ip addr +3. etcd 诊断命令 - 检查 etcd 集群健康与数据。例如: etcdctl endpoint status, etcdctl member list, etcdctl alarm list, etcdctl get /kubernetes --prefix, etcdutl snapshot status 行为要求： 1. 如果证据不足，必须返回 command 动作并逐条调查；至少检查与故障资源直接相关的状态、describe、events 或 logs。 diff --git a/tests/test_agent_tools.py b/tests/test_agent_tools.py index 66ae3d7..7ad4b76 100644 --- a/tests/test_agent_tools.py +++ b/tests/test_agent_tools.py @@ -102,6 +102,71 @@ class ShellCommandPolicyTest(unittest.TestCase): classify_shell_command("cat /etc/shadow") +class EtcdCommandPolicyTest(unittest.TestCase): + def test_etcdctl_endpoint_status_is_read_only(self): + decision = classify_shell_command("etcdctl endpoint status") + self.assertEqual(decision.mode, "read") + self.assertFalse(decision.requires_approval) + + def test_etcdctl_with_global_flags_still_classifies(self): + decision = classify_shell_command( + "etcdctl --endpoints=https://127.0.0.1:2379 --cacert=/etc/ssl/ca.pem endpoint health" + ) + self.assertEqual(decision.mode, "read") + self.assertFalse(decision.requires_approval) + + def test_etcdctl_member_list_is_read_only(self): + decision = classify_shell_command("etcdctl member list") + self.assertEqual(decision.mode, "read") + self.assertFalse(decision.requires_approval) + + def test_etcdctl_get_and_version_are_read_only(self): + for command in ("etcdctl get /kubernetes", "etcdctl version", "etcdctl alarm list"): + with self.subTest(command=command): + decision = classify_shell_command(command) + self.assertEqual(decision.mode, "read") + self.assertFalse(decision.requires_approval) + + def test_etcdutl_snapshot_status_is_read_only(self): + decision = classify_shell_command("etcdutl snapshot status /tmp/etcd.db") + self.assertEqual(decision.mode, "read") + self.assertFalse(decision.requires_approval) + + def test_etcdctl_put_requires_approval(self): + decision = classify_shell_command("etcdctl put key value") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + + def test_etcdctl_member_remove_requires_approval(self): + decision = classify_shell_command("etcdctl member remove abc123") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + + def test_etcdctl_snapshot_save_requires_approval(self): + decision = classify_shell_command("etcdctl snapshot save /tmp/etcd.db") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + + def test_etcdctl_compact_requires_approval(self): + decision = classify_shell_command("etcdctl compact 12345") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + + def test_etcdctl_auth_enable_requires_approval(self): + decision = classify_shell_command("etcdctl auth enable") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + + def test_etcdutl_snapshot_restore_requires_approval(self): + decision = classify_shell_command("etcdutl snapshot restore /tmp/etcd.db") + self.assertEqual(decision.mode, "write") + self.assertTrue(decision.requires_approval) + + def test_etcdctl_rejects_unknown_subcommand(self): + with self.assertRaises(ValueError): + classify_shell_command("etcdctl bogus command") + + class AgentCommandDispatchTest(unittest.TestCase): def test_dispatches_kubectl_command_to_kubectl_classifier(self): decision = classify_agent_command("kubectl get pods -A")