feat: 声明式权限(opencode 式 allow/ask/deny 配置化)
- safety.check 在 plan 模式之后、安全模式之前插入声明式权限判定: deny(阻断) > ask(强制确认) > allow(自动放行) > 默认(按安全模式) - 规则语法支持 "*"全匹配 / "tool"精确 / "tool:*"任意操作 / "tool:op"操作前缀 - 读取 config.permissions;提供 set_permissions 显式注入(测试/启动用) - config.json 顶层加 permissions 段(allow/ask/deny 示例,该文件被 gitignore 不入库) - 修复:test_normal_mode_does_not_plan_block 隔离真实 config 权限,避免污染 - 新增 tests/test_permissions.py(14 项:规则匹配+deny/ask/allow+优先级) Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
This commit is contained in:
1 parent
9dd418ec89
commit
160d4e924e
3 files changed
+196
No files matched your search
@@ -0,0 +1,116 @@
|
||||
"""声明式权限(allow/ask/deny)回归测试。不触真实引擎/网络。"""
|
||||
import pytest
|
||||
from xcli_core.safety import SafetyLayer, MODE_UNRESTRICTED, MODE_NORMAL, MODE_MANUAL
|
||||
|
||||
|
||||
def _new_safety(allow=None, ask=None, deny=None, mode=MODE_NORMAL):
|
||||
s = SafetyLayer()
|
||||
s.set_permissions(allow=allow, ask=ask, deny=deny)
|
||||
s.mode = mode
|
||||
return s
|
||||
|
||||
|
||||
# ── 纯函数:规则匹配 ──
|
||||
|
||||
def test_match_wildcard():
|
||||
s = SafetyLayer()
|
||||
assert s._permission_match("*", "anything", "rm -rf /") is True
|
||||
|
||||
|
||||
def test_match_exact_tool():
|
||||
s = SafetyLayer()
|
||||
assert s._permission_match("git_tools", "git_tools", "status") is True
|
||||
assert s._permission_match("git_tools", "cmd_executor", "status") is False
|
||||
|
||||
|
||||
def test_match_tool_star():
|
||||
s = SafetyLayer()
|
||||
assert s._permission_match("git_tools:*", "git_tools", "push origin") is True
|
||||
|
||||
|
||||
def test_match_tool_op_prefix():
|
||||
s = SafetyLayer()
|
||||
assert s._permission_match("cmd_executor:rm", "cmd_executor", "rm -rf /tmp") is True
|
||||
assert s._permission_match("cmd_executor:rm", "cmd_executor", "ls -la") is False
|
||||
# 操作词精确相等也应命中
|
||||
assert s._permission_match("git_tools:push", "git_tools", "push") is True
|
||||
|
||||
|
||||
def test_match_empty_rule_ignored():
|
||||
s = SafetyLayer()
|
||||
assert s._permission_match("", "git_tools", "status") is False
|
||||
|
||||
|
||||
# ── deny 最高优先级 ──
|
||||
|
||||
def test_deny_blocks():
|
||||
s = _new_safety(deny=["git_tools:reset"], mode=MODE_UNRESTRICTED)
|
||||
ok, msg = s.check("git_tools", "reset --hard HEAD")
|
||||
assert ok is False
|
||||
assert "deny" in msg
|
||||
|
||||
|
||||
def test_deny_blocks_even_unrestricted():
|
||||
s = _new_safety(deny=["cmd_executor:rm"], mode=MODE_UNRESTRICTED)
|
||||
ok, _ = s.check("cmd_executor", "rm -rf /")
|
||||
assert ok is False
|
||||
|
||||
|
||||
def test_deny_overrides_allow():
|
||||
# 同一工具同时出现在 deny 与 allow,deny 胜出
|
||||
s = _new_safety(deny=["git_tools:reset"], allow=["git_tools:reset"], mode=MODE_NORMAL)
|
||||
ok, _ = s.check("git_tools", "reset --hard")
|
||||
assert ok is False
|
||||
|
||||
|
||||
# ── ask 强制确认 ──
|
||||
|
||||
def test_ask_forces_confirm():
|
||||
s = _new_safety(ask=["cmd_executor"], mode=MODE_NORMAL)
|
||||
confirmed = []
|
||||
s._ask_user_confirm = lambda *a, **k: (confirmed.append(True) or (True, "ok"))
|
||||
ok, _ = s.check("cmd_executor", "ls")
|
||||
assert ok is True
|
||||
assert confirmed == [True]
|
||||
|
||||
|
||||
def test_ask_overrides_allow():
|
||||
# 同时 allow 与 ask,ask(确认)胜出,不会自动放行
|
||||
s = _new_safety(allow=["cmd_executor"], ask=["cmd_executor"], mode=MODE_NORMAL)
|
||||
called = []
|
||||
s._ask_user_confirm = lambda *a, **k: (called.append(True) or (True, "ok"))
|
||||
s.check("cmd_executor", "ls")
|
||||
assert called == [True]
|
||||
|
||||
|
||||
def test_ask_works_in_unrestricted():
|
||||
s = _new_safety(ask=["git_tools:push"], mode=MODE_UNRESTRICTED)
|
||||
called = []
|
||||
s._ask_user_confirm = lambda *a, **k: (called.append(True) or (False, "no"))
|
||||
ok, _ = s.check("git_tools", "push origin master")
|
||||
assert ok is False
|
||||
assert called == [True]
|
||||
|
||||
|
||||
# ── allow 自动放行 ──
|
||||
|
||||
def test_allow_auto_passes():
|
||||
s = _new_safety(allow=["git_tools:status"], mode=MODE_NORMAL)
|
||||
# 不应触发 AI 分析 / 确认
|
||||
s._ask_user_confirm = lambda *a, **k: (False, "SHOULD_NOT_BE_CALLED")
|
||||
ok, msg = s.check("git_tools", "status")
|
||||
assert ok is True
|
||||
assert msg == ""
|
||||
|
||||
|
||||
def test_allow_no_match_falls_through():
|
||||
s = _new_safety(allow=["git_tools:status"], mode=MODE_UNRESTRICTED)
|
||||
ok, _ = s.check("cmd_executor", "ls")
|
||||
# 无规则命中 → 走默认(无限制模式直接放行)
|
||||
assert ok is True
|
||||
|
||||
|
||||
def test_default_without_rules_unrestricted_passes():
|
||||
s = _new_safety(mode=MODE_UNRESTRICTED)
|
||||
ok, _ = s.check("file_manager", "read foo.txt")
|
||||
assert ok is True
|
||||
@@ -118,9 +118,11 @@ def test_normal_mode_does_not_plan_block():
|
||||
safety = get_safety()
|
||||
saved_cli = safety.cli
|
||||
saved_mode = safety.mode
|
||||
saved_perm = safety._permissions_override
|
||||
fake_cli = types.SimpleNamespace(plan_mode=False)
|
||||
safety.set_cli(fake_cli)
|
||||
safety.set_mode(0) # MODE_UNRESTRICTED:直接放行,不消耗 AI
|
||||
safety.set_permissions(allow=[], ask=[], deny=[]) # 隔离真实 config 的权限规则
|
||||
try:
|
||||
allowed, _ = safety.check("code_editor", "write a.py x")
|
||||
assert allowed is True
|
||||
@@ -129,6 +131,7 @@ def test_normal_mode_does_not_plan_block():
|
||||
finally:
|
||||
safety.set_cli(saved_cli)
|
||||
safety.set_mode(saved_mode)
|
||||
safety._permissions_override = saved_perm
|
||||
|
||||
|
||||
# ── system prompt 注入 ──
|
||||
|
||||
@@ -85,6 +85,7 @@ class SafetyLayer:
|
||||
self.cli = None
|
||||
self._analysis_cache = {} # 缓存分析结果
|
||||
self._lock = threading.Lock()
|
||||
self._permissions_override = None # set_permissions 注入的显式权限
|
||||
|
||||
def set_cli(self, cli):
|
||||
self.cli = cli
|
||||
@@ -103,6 +104,62 @@ class SafetyLayer:
|
||||
self.mode = (self.mode + 1) % 3
|
||||
return self.get_mode_name()
|
||||
|
||||
# ── 声明式权限(opencode 式 allow/ask/deny)──
|
||||
# 规则语法:
|
||||
# "*" → 匹配任意工具/操作
|
||||
# "tool" → 精确匹配工具名
|
||||
# "tool:*" → 该工具任意操作
|
||||
# "tool:op" → 工具名 + 参数首个词命中 op 前缀
|
||||
# 优先级:deny(阻断)> ask(强制确认)> allow(自动放行)> 默认(按模式)
|
||||
|
||||
def set_permissions(self, allow=None, ask=None, deny=None):
|
||||
"""显式注入权限规则(测试或启动时调用),优先级最高。"""
|
||||
self._permissions_override = {
|
||||
"allow": list(allow or []),
|
||||
"ask": list(ask or []),
|
||||
"deny": list(deny or []),
|
||||
}
|
||||
|
||||
def _get_permissions(self) -> dict:
|
||||
if self._permissions_override is not None:
|
||||
return self._permissions_override
|
||||
return self._load_permissions()
|
||||
|
||||
def _load_permissions(self) -> dict:
|
||||
try:
|
||||
from .config import load_config
|
||||
except ImportError:
|
||||
try:
|
||||
from config import load_config
|
||||
except ImportError:
|
||||
return {"allow": [], "ask": [], "deny": []}
|
||||
cfg = load_config() or {}
|
||||
perm = cfg.get("permissions") or {}
|
||||
return {
|
||||
"allow": perm.get("allow", []) or [],
|
||||
"ask": perm.get("ask", []) or [],
|
||||
"deny": perm.get("deny", []) or [],
|
||||
}
|
||||
|
||||
def _permission_match(self, rule: str, tool_name: str, tool_args) -> bool:
|
||||
rule = (rule or "").strip()
|
||||
if not rule:
|
||||
return False
|
||||
if rule == "*":
|
||||
return True
|
||||
if ":" in rule:
|
||||
rtool, raction = rule.split(":", 1)
|
||||
rtool = rtool.strip()
|
||||
raction = raction.strip()
|
||||
if rtool != tool_name:
|
||||
return False
|
||||
if raction in ("*", ""):
|
||||
return True
|
||||
op = (str(tool_args).strip().split()[0].lower()
|
||||
if str(tool_args).strip() else "")
|
||||
return op == raction or op.startswith(raction)
|
||||
return rule == tool_name
|
||||
|
||||
# ── 核心:拦截检查 ──
|
||||
|
||||
def check(self, tool_name: str, tool_args: str) -> Tuple[bool, str]:
|
||||
@@ -118,6 +175,26 @@ class SafetyLayer:
|
||||
return False, "PLAN 模式:写操作已暂挂(只读调研中)。请在 /build 批准后执行。"
|
||||
return True, ""
|
||||
|
||||
# ── 声明式权限(config.permissions: deny > ask > allow)──
|
||||
perm = self._get_permissions()
|
||||
# deny 最高优先级:直接阻断(即便无限制模式也生效)
|
||||
for rule in perm.get("deny", []):
|
||||
if self._permission_match(rule, tool_name, tool_args):
|
||||
return False, f"权限拒绝(deny 规则 '{rule}' 命中): {tool_name}"
|
||||
# ask:强制人工确认(覆盖自动放行,即便无限制模式也确认)
|
||||
for rule in perm.get("ask", []):
|
||||
if self._permission_match(rule, tool_name, tool_args):
|
||||
return self._ask_user_confirm(
|
||||
tool_name, tool_args,
|
||||
risk_level="中",
|
||||
impact="命中 ask 规则,需你确认后执行",
|
||||
reason=f"权限规则 '{rule}'"
|
||||
)
|
||||
# allow:自动放行(跳过 AI 风险分析)
|
||||
for rule in perm.get("allow", []):
|
||||
if self._permission_match(rule, tool_name, tool_args):
|
||||
return True, ""
|
||||
|
||||
# 无限制模式:直接放行
|
||||
if self.mode == MODE_UNRESTRICTED:
|
||||
return True, ""
|
||||
|
||||
Reference in new issue
Block a user