116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""司命决策器：按工具策略 + 风险等级 + 是否已确认，输出 allow / ask / deny。
|
||
|
||
风险分级与权限策略取值见契约字典（risk_level L0–L4、permission_policy
|
||
allow/ask/always_ask/forbidden）；决策枚举对齐 AuditEvent.decision（allow/deny/ask）。
|
||
|
||
策略来源：``SafetyGovernor()`` 用代码内置默认；``from_policy_dict(data)`` 从已解析字典
|
||
加载（文件 IO 由 ``services/safety.load_safety_governor`` 承担，本模块保持纯领域、零 IO）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from enum import StrEnum
|
||
|
||
|
||
class PermissionPolicy(StrEnum):
|
||
allow = "allow"
|
||
ask = "ask"
|
||
always_ask = "always_ask"
|
||
forbidden = "forbidden"
|
||
|
||
|
||
class SafetyDecision(StrEnum):
|
||
allow = "allow"
|
||
ask = "ask"
|
||
deny = "deny"
|
||
|
||
|
||
# 风险等级排序，便于阈值判断
|
||
RISK_RANK: dict[str, int] = {"L0": 0, "L1": 1, "L2": 2, "L3": 3, "L4": 4}
|
||
|
||
|
||
def default_policy(risk_level: str) -> PermissionPolicy:
|
||
"""无显式工具策略时，按风险等级推导默认策略。"""
|
||
rank = RISK_RANK.get(risk_level, 2)
|
||
if rank >= 4:
|
||
return PermissionPolicy.forbidden
|
||
if rank >= 3:
|
||
return PermissionPolicy.always_ask
|
||
if rank >= 2:
|
||
return PermissionPolicy.ask
|
||
return PermissionPolicy.allow
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ToolPolicy:
|
||
"""工具的固定策略（由工具定义或配置提供）。"""
|
||
|
||
risk_level: str # L0–L4
|
||
permission_policy: PermissionPolicy
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SafetyResult:
|
||
decision: SafetyDecision
|
||
reason: str
|
||
risk_level: str
|
||
policy: PermissionPolicy
|
||
|
||
|
||
class SafetyGovernor:
|
||
"""司命决策器。无状态、同步——便于测试与上层组合确认流。"""
|
||
|
||
def __init__(self, policies: dict[str, ToolPolicy] | None = None) -> None:
|
||
self._policies: dict[str, ToolPolicy] = dict(policies or {})
|
||
|
||
@classmethod
|
||
def from_policy_dict(cls, data: dict) -> SafetyGovernor:
|
||
"""从已解析的策略字典构造 Governor（纯逻辑，不读文件）。
|
||
|
||
文件读取由 ``services/safety.load_safety_governor`` 承担，调用方传入已
|
||
``yaml.safe_load`` 的字典；空字典或缺 ``tools`` 返回默认（空策略）Governor。
|
||
"""
|
||
policies: dict[str, ToolPolicy] = {}
|
||
tools = data.get("tools") if isinstance(data, dict) else None
|
||
if isinstance(tools, dict):
|
||
for name, cfg in tools.items():
|
||
if not isinstance(cfg, dict):
|
||
continue
|
||
policies[name] = ToolPolicy(
|
||
risk_level=str(cfg.get("risk_level", "L2")),
|
||
permission_policy=PermissionPolicy(str(cfg.get("permission_policy", "ask"))),
|
||
)
|
||
return cls(policies)
|
||
|
||
def evaluate(
|
||
self,
|
||
*,
|
||
tool: str,
|
||
risk_level: str | None = None,
|
||
confirmed: bool = False,
|
||
) -> SafetyResult:
|
||
policy = self._policies.get(tool)
|
||
if policy is None:
|
||
resolved_risk = risk_level or "L2"
|
||
perm = default_policy(resolved_risk)
|
||
else:
|
||
resolved_risk = policy.risk_level
|
||
perm = policy.permission_policy
|
||
|
||
# L4 一律禁止，即便策略写 allow
|
||
if RISK_RANK.get(resolved_risk, 2) >= 4:
|
||
return SafetyResult(
|
||
SafetyDecision.deny, "L4 风险禁止执行", resolved_risk, PermissionPolicy.forbidden
|
||
)
|
||
if perm is PermissionPolicy.forbidden:
|
||
return SafetyResult(SafetyDecision.deny, "工具策略为 forbidden", resolved_risk, perm)
|
||
if perm is PermissionPolicy.allow:
|
||
return SafetyResult(SafetyDecision.allow, "策略放行", resolved_risk, perm)
|
||
# ask / always_ask
|
||
if confirmed:
|
||
return SafetyResult(SafetyDecision.allow, "用户已确认", resolved_risk, perm)
|
||
return SafetyResult(
|
||
SafetyDecision.ask, f"策略 {perm.value} 需用户确认", resolved_risk, perm
|
||
)
|