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
This commit is contained in:
Chiguyong 2026-07-06 13:12:34 +08:00
parent 9fe1810497
commit b6f7d82ff5
2 changed files with 167 additions and 5 deletions

View File

@ -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

View File

@ -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)