From b6f7d82ff573e5c9b78b19fd13526b7da8665ebc Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 13:12:34 +0800 Subject: [PATCH] feat(config): U1 add DangerousToolsConfig for PLAN_EXEC autonomy (KTD2) - New DangerousToolsConfig dataclass with regex-based tool_patterns - Default whitelist: rm/rmdir/deploy/kubectl/helm/git_push_force/alembic/migrate/payment_* - is_dangerous(tool_name) method for autonomy mode gate - ServerConfig.from_dict parses dangerous_tools section - 10 unit tests covering defaults, matching, disabled, custom patterns --- src/agentkit/server/config.py | 52 +++++++++++++- tests/unit/test_server_config.py | 120 +++++++++++++++++++++++++++++-- 2 files changed, 167 insertions(+), 5 deletions(-) diff --git a/src/agentkit/server/config.py b/src/agentkit/server/config.py index e634908..ee34e06 100644 --- a/src/agentkit/server/config.py +++ b/src/agentkit/server/config.py @@ -4,7 +4,7 @@ import asyncio import logging import os import re -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable @@ -57,6 +57,48 @@ class MCPServerConfig: ) +@dataclass +class DangerousToolsConfig: + """Configuration for dangerous tool detection in PLAN_EXEC autonomy mode. + + When enabled, tools matching any pattern in ``tool_patterns`` trigger a + confirmation request before execution. KTD2: conservative whitelist — + ponytail ceiling = missed dangerous ops not in the list; upgrade path = + LLM-assisted classification. + """ + + enabled: bool = True + tool_patterns: list[str] = field( + default_factory=lambda: [ + r"^rm$", + r"^rmdir$", + r"^deploy$", + r"^kubectl", + r"^helm", + r"^git_push_force$", + r"^alembic", + r"^migrate", + r"^payment_", + ] + ) + + def is_dangerous(self, tool_name: str) -> bool: + """Check if a tool name matches any dangerous pattern.""" + if not self.enabled: + return False + return any(re.match(p, tool_name) for p in self.tool_patterns) + + @classmethod + def from_dict(cls, data: dict | None) -> "DangerousToolsConfig": + """Create from dict (parsed from YAML). Empty/None returns defaults.""" + if not data: + return cls() + return cls( + enabled=data.get("enabled", True), + tool_patterns=data.get("tool_patterns") or cls().tool_patterns, + ) + + def _resolve_env_vars(value: Any) -> Any: """Resolve ${VAR:-default} patterns in string values from environment variables.""" if not isinstance(value, str): @@ -124,6 +166,9 @@ class ServerConfig: # G6/U2: PLAN_EXEC phase policy config (opt-in — None = disabled). # Parsed via PhasePolicy.policy_from_config() at chat.py wiring time. plan_exec: dict[str, Any] | None = None, + # IQ-Boost/U1: dangerous_tools config for PLAN_EXEC autonomy mode (KTD2). + # Parsed via DangerousToolsConfig.from_dict(); defaults applied when None. + dangerous_tools: dict[str, Any] | None = None, on_change: Callable[["ServerConfig"], None] | None = None, ): self.host = host @@ -168,6 +213,8 @@ class ServerConfig: # Resolved to PhasePolicy via agentkit.core.phase.policy_from_config() # at chat.py WebSocket wiring time (U4). self.plan_exec = plan_exec or {} + # IQ-Boost/U1: dangerous_tools parsed config (KTD2 conservative whitelist). + self.dangerous_tools = DangerousToolsConfig.from_dict(dangerous_tools) self.on_change = on_change # Config watching state @@ -261,6 +308,8 @@ class ServerConfig: fallback_chain_data = data.get("fallback_chain", {}) # G6/U2: plan_exec phase policy 配置 (从 YAML 读取, opt-in) plan_exec_data = data.get("plan_exec", {}) + # IQ-Boost/U1: dangerous_tools 配置 (从 YAML 读取, KTD2 whitelist) + dangerous_tools_data = data.get("dangerous_tools", {}) return cls( host=server.get("host", "0.0.0.0"), @@ -295,6 +344,7 @@ class ServerConfig: rollback=rollback_data, fallback_chain=fallback_chain_data, plan_exec=plan_exec_data, + dangerous_tools=dangerous_tools_data, ) @staticmethod diff --git a/tests/unit/test_server_config.py b/tests/unit/test_server_config.py index e8d1b12..3d4dacf 100644 --- a/tests/unit/test_server_config.py +++ b/tests/unit/test_server_config.py @@ -6,7 +6,13 @@ from pathlib import Path import pytest -from agentkit.server.config import ServerConfig, find_config_path, _resolve_env_vars, _deep_resolve +from agentkit.server.config import ( + ServerConfig, + DangerousToolsConfig, + find_config_path, + _resolve_env_vars, + _deep_resolve, +) class TestEnvVarResolution: @@ -178,7 +184,9 @@ prompt: # Update yaml_content with absolute path yaml_content_updated = yaml_content.replace("./skills", str(skills_dir)) - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False, dir=tmpdir) as f: + with tempfile.NamedTemporaryFile( + mode="w", suffix=".yaml", delete=False, dir=tmpdir + ) as f: f.write(yaml_content_updated) f.flush() config = ServerConfig.from_yaml(f.name) @@ -209,7 +217,9 @@ skills: paths: - "{skill_yaml}" """ - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False, dir=tmpdir) as f: + with tempfile.NamedTemporaryFile( + mode="w", suffix=".yaml", delete=False, dir=tmpdir + ) as f: f.write(yaml_content) f.flush() config = ServerConfig.from_yaml(f.name) @@ -246,7 +256,9 @@ skills: paths: - "{skills_dir}" """ - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False, dir=tmpdir) as f: + with tempfile.NamedTemporaryFile( + mode="w", suffix=".yaml", delete=False, dir=tmpdir + ) as f: f.write(yaml_content) f.flush() config = ServerConfig.from_yaml(f.name) @@ -444,3 +456,103 @@ class TestConfigHotReload: assert config.port == 9000 os.unlink(config_path) + + +class TestDangerousToolsConfig: + """Test DangerousToolsConfig — U1: dangerous tool whitelist for PLAN_EXEC autonomy (KTD2).""" + + def test_defaults_when_no_data(self): + cfg = DangerousToolsConfig.from_dict(None) + assert cfg.enabled is True + assert len(cfg.tool_patterns) > 0 + # Default whitelist includes rm, deploy, kubectl, etc. + assert cfg.is_dangerous("rm") + assert cfg.is_dangerous("deploy") + assert cfg.is_dangerous("kubectl_apply") + + def test_defaults_when_empty_dict(self): + cfg = DangerousToolsConfig.from_dict({}) + assert cfg.enabled is True + assert cfg.is_dangerous("rmdir") + + def test_dangerous_tool_match(self): + cfg = DangerousToolsConfig() + assert cfg.is_dangerous("rm") is True + assert cfg.is_dangerous("rmdir") is True + assert cfg.is_dangerous("deploy") is True + assert cfg.is_dangerous("kubectl_rollout") is True + assert cfg.is_dangerous("helm_upgrade") is True + assert cfg.is_dangerous("git_push_force") is True + assert cfg.is_dangerous("alembic_upgrade") is True + assert cfg.is_dangerous("migrate_db") is True + assert cfg.is_dangerous("payment_charge") is True + + def test_non_dangerous_tool(self): + cfg = DangerousToolsConfig() + assert cfg.is_dangerous("read_file") is False + assert cfg.is_dangerous("write_file") is False + assert cfg.is_dangerous("search") is False + assert cfg.is_dangerous("list_dir") is False + + def test_payment_glob_pattern(self): + cfg = DangerousToolsConfig() + # payment_* pattern matches any tool starting with payment_ + assert cfg.is_dangerous("payment_refund") is True + assert cfg.is_dangerous("payment_transfer") is True + + def test_disabled_returns_false(self): + cfg = DangerousToolsConfig(enabled=False) + assert cfg.is_dangerous("rm") is False + assert cfg.is_dangerous("deploy") is False + + def test_custom_patterns_from_dict(self): + data = {"enabled": True, "tool_patterns": [r"^custom_dangerous"]} + cfg = DangerousToolsConfig.from_dict(data) + assert cfg.is_dangerous("custom_dangerous_op") is True + assert cfg.is_dangerous("rm") is False # default patterns overridden + + def test_enabled_false_from_dict(self): + data = {"enabled": False} + cfg = DangerousToolsConfig.from_dict(data) + assert cfg.enabled is False + assert cfg.is_dangerous("rm") is False + + def test_server_config_loads_dangerous_tools(self): + yaml_content = """ +server: + host: "0.0.0.0" + port: 8001 + +dangerous_tools: + enabled: true + tool_patterns: + - "^rm$" + - "^kubectl" +""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + f.write(yaml_content) + f.flush() + config = ServerConfig.from_yaml(f.name) + + assert config.dangerous_tools.enabled is True + assert config.dangerous_tools.is_dangerous("rm") is True + assert config.dangerous_tools.is_dangerous("kubectl_apply") is True + # Custom patterns override defaults + assert config.dangerous_tools.is_dangerous("deploy") is False + os.unlink(f.name) + + def test_server_config_defaults_dangerous_tools_when_missing(self): + yaml_content = """ +server: + host: "0.0.0.0" + port: 8001 +""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + f.write(yaml_content) + f.flush() + config = ServerConfig.from_yaml(f.name) + + # Missing config section → defaults applied + assert config.dangerous_tools.enabled is True + assert config.dangerous_tools.is_dangerous("rm") is True + os.unlink(f.name)