diff --git a/tests/test_permissions.py b/tests/test_permissions.py new file mode 100644 index 0000000..131e0f7 --- /dev/null +++ b/tests/test_permissions.py @@ -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 diff --git a/tests/test_plan_mode.py b/tests/test_plan_mode.py index 57ebeab..90dd5b9 100644 --- a/tests/test_plan_mode.py +++ b/tests/test_plan_mode.py @@ -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 注入 ── diff --git a/xcli_core/safety.py b/xcli_core/safety.py index 38c2240..a73af90 100644 --- a/xcli_core/safety.py +++ b/xcli_core/safety.py @@ -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, ""