"""Microsoft Teams 工具策略 (Tools Policy)。 Team/Channel/Sender 三级 tools allow/deny 策略解析。 支持全局配置 + teams 嵌套覆盖 + toolsBySender 按发送者策略。 """ from __future__ import annotations from typing import Any class ToolPolicy: """MSTeams 三级工具策略解析器。 优先级:Sender > Channel > Team > Global deny > Global allow """ def __init__(self, config: dict[str, Any] | None = None): config = config or {} self._global_allow: set[str] = set(config.get("tools_allow", [])) self._global_deny: set[str] = set(config.get("tools_deny", [])) self._tools_by_sender: dict[str, dict[str, set[str]]] = {} sender_tools = config.get("tools_by_sender", []) or [] if isinstance(sender_tools, dict): for sender_id, tools in sender_tools.items(): if isinstance(tools, list): self._tools_by_sender[sender_id] = {"allow": set(tools), "deny": set()} elif isinstance(tools, dict): self._tools_by_sender[sender_id] = { "allow": set(tools.get("allow", [])), "deny": set(tools.get("deny", [])), } elif isinstance(sender_tools, list): for entry in sender_tools: if isinstance(entry, dict): sid = entry.get("sender_id", "") if sid: self._tools_by_sender[sid] = { "allow": set(entry.get("allow", [])), "deny": set(entry.get("deny", [])), } self._teams_config: dict[str, dict[str, Any]] = config.get("teams", {}) def is_tool_allowed( self, tool_name: str, sender_id: str = "", team_id: str = "", channel_id: str = "", ) -> bool: sender_rules = self._tools_by_sender.get(sender_id) if sender_rules is not None: if tool_name in sender_rules["deny"]: return False if sender_rules["allow"]: return tool_name in sender_rules["allow"] channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id) if tool_name in channel_deny: return False if channel_allow and tool_name not in channel_allow: return False team_allow, team_deny = self._resolve_team_tools(team_id) if tool_name in team_deny: return False if team_allow and tool_name not in team_allow: return False if tool_name in self._global_deny: return False if self._global_allow: return tool_name in self._global_allow return True def _resolve_team_tools(self, team_id: str) -> tuple[set[str], set[str]]: team_config = self._teams_config.get(team_id) or self._teams_config.get("*") or {} team_tools = team_config.get("tools", {}) allow = set(team_tools.get("allow", [])) deny = set(team_tools.get("deny", [])) return allow, deny def _resolve_channel_tools(self, team_id: str, channel_id: str) -> tuple[set[str], set[str]]: if not team_id or not channel_id: return set(), set() team_config = self._teams_config.get(team_id) or self._teams_config.get("*") or {} channels_config = team_config.get("channels", {}) ch_config = channels_config.get(channel_id) or channels_config.get("*") or {} ch_tools = ch_config.get("tools", {}) allow = set(ch_tools.get("allow", [])) deny = set(ch_tools.get("deny", [])) return allow, deny def get_allowed_tools( self, sender_id: str = "", team_id: str = "", channel_id: str = "", ) -> list[str] | None: sender_rules = self._tools_by_sender.get(sender_id) if sender_rules is not None: if sender_rules["allow"]: return sorted(sender_rules["allow"] - sender_rules["deny"]) channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id) team_allow, team_deny = self._resolve_team_tools(team_id) if channel_allow: return sorted(channel_allow - channel_deny) if team_allow: return sorted(team_allow - team_deny) if self._global_allow: return sorted(self._global_allow - self._global_deny) return None def get_denied_tools( self, sender_id: str = "", team_id: str = "", channel_id: str = "", ) -> set[str]: sender_rules = self._tools_by_sender.get(sender_id) sender_deny = sender_rules["deny"] if sender_rules else set() channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id) team_allow, team_deny = self._resolve_team_tools(team_id) return sender_deny | channel_deny | team_deny | self._global_deny def resolve_tool_policy( config: dict[str, Any], tool_name: str, sender_id: str = "", team_id: str = "", channel_id: str = "", ) -> bool: """快捷函数:三级工具策略解析。 优先级:Sender > Channel > Team > Global """ policy = ToolPolicy(config) return policy.is_tool_allowed(tool_name, sender_id, team_id, channel_id)