"""结构化审批卡数据(feat-chv-muse-p12,叶子模块,纯标准库)。 接 Muse 官方要求(报告 §2.6):确定性 UI——结构化审批卡 (明确接受/拒绝动作);审批是严格能力,绑定特定连接器/ 目的地与用例,支持一次性、会话级、任务级、时限级、永久级 授权,Sentinel 决定向用户呈现哪几种授权类型;审批 UI 不在 对话内,Sentinel 直接向客户端发审批请求,用户答复直接回 Sentinel(防止对话内容伪造审批)。 本模块只做卡的**数据**:把 P3 ``ApprovalQueue`` 的 pending 记录 dict 转成客户端可呈现的结构化卡。不 import P3/P5 模块 ——它们不在本基线里,集成时由 round-11 合并;只按 dict 契约接线。 卡数据(纯数据 dict,可直接 JSON 序列化):: {"pending_id": str, "action": str, "target": {...}, "reasons": [str, ...], "risk": "low|medium|high", "actions": [{"id": "accept", "label": "接受"}, {"id": "deny", "label": "拒绝"}], "grant_options": [{"tier": str, "label": str, "requires_explicit_opt_in": bool}, ...], "expires_at": float | None} - ``risk`` 分级(确定性规则,不冒充模型分类器):写方法 (POST/PUT/PATCH/DELETE)+ 敏感路径(/admin、/settings、 /login、token 相关)或私有主机 → high;写方法但目标不 敏感 → medium;其余只读 → low。 - ``grant_options`` 的档位列表由调用方传入(Sentinel 决定 呈现哪几种);不传默认五档全列,永久档永远标注 ``requires_explicit_opt_in: true``(P5 口径:永久默认拒绝, 须显式 opt-in)。 - ``card_matches_channel()``:校验卡数据与 P5 对话外通道载荷 (``ApprovalRequest.payload()``)一致——pending_id 相同且 目标字段(连接器/方法/scheme/主机/端口/路径)全等;不一 致即客户端不该呈现该卡。 """ import ipaddress # 五档授权类型(P3/P5 口径):一次/会话/任务/时限/永久。 TIERS = ("one-shot", "session", "task", "timed", "permanent") TIER_LABELS = { "one-shot": "仅这一次", "session": "本次会话", "task": "本任务期间", "timed": "限时授权", "permanent": "永久授权", } # 目标描述字段:卡与通道载荷比对只认这些键。 TARGET_KEYS = ("connector", "method", "scheme", "host", "port", "path") _WRITE_METHODS = ("POST", "PUT", "PATCH", "DELETE") _SENSITIVE_PATHS = ("/admin", "/settings", "/login", "/token", "/oauth", "/credential") # 通道审批默认有效期(秒):15 分钟,与 P5 通道一致。 DEFAULT_TTL_SECONDS = 900 def _is_private_host(host: str) -> bool: """主机是否私有/本地:IP 段判断,localhost 视为私有。""" if not host: return False lowered = host.lower() if lowered in ("localhost", "localhost.localdomain"): return True try: return ipaddress.ip_address(host).is_private except ValueError: return False def _is_sensitive_path(path: str) -> bool: """路径是否敏感:命中敏感前缀即敏感。""" lowered = (path or "").lower() return any(lowered.startswith(p) for p in _SENSITIVE_PATHS) def risk_level(queue_record: dict) -> str: """审批风险分级:high/medium/low(确定性规则)。 写方法 +(敏感路径或私有主机)→ high;写方法 → medium; 其余只读 → low。 """ method = str(queue_record.get("method", "")).upper() if method not in _WRITE_METHODS: return "low" if _is_sensitive_path(str(queue_record.get("path", ""))): return "high" if _is_private_host(str(queue_record.get("host", ""))): return "high" return "medium" def _target_from_record(queue_record: dict) -> dict: """队列记录里只取目标描述字段,别的什么都不带出卡。""" return {k: queue_record[k] for k in TARGET_KEYS if k in queue_record} def _grant_options(tiers: tuple | list | None) -> list: """档位选项:未知档位丢弃,永久档永远标注须显式 opt-in。""" wanted = list(tiers) if tiers is not None else list(TIERS) options = [] for tier in wanted: if tier not in TIER_LABELS: continue options.append({ "tier": tier, "label": TIER_LABELS[tier], "requires_explicit_opt_in": tier == "permanent", }) return options def build_approval_card(queue_record: dict, tiers: tuple | list | None = None, ttl_seconds: int = DEFAULT_TTL_SECONDS ) -> dict: """P3 队列记录转结构化审批卡数据。 queue_record 是 ``ApprovalQueue.record()`` 记入的 dict: 目标描述(connector/method/scheme/host/port/path)+ pending_id + reasons + ts。缺 pending_id 抛 ValueError。 expires_at 由记录 ts + 有效期算出;无 ts 则为 None。 """ pending_id = queue_record.get("pending_id") if not pending_id: raise ValueError("队列记录缺 pending_id") target = _target_from_record(queue_record) connector = str(target.get("connector", "")) method = str(target.get("method", "")).upper() action = "%s %s" % (connector, method) if connector else method ts = queue_record.get("ts") expires_at = None if isinstance(ts, (int, float)): expires_at = float(ts) + ttl_seconds return { "pending_id": str(pending_id), "action": action, "target": target, "reasons": list(queue_record.get("reasons") or []), "risk": risk_level(queue_record), "actions": [ {"id": "accept", "label": "接受"}, {"id": "deny", "label": "拒绝"}, ], "grant_options": _grant_options(tiers), "expires_at": expires_at, } def card_matches_channel(card: dict, channel_payload: dict) -> bool: """卡数据与 P5 通道载荷是否一致(pending_id + 目标全等)。 任一目标字段不一致、或 pending_id 不同,即返回 False。 客户端只应呈现校验通过的卡。 """ if card.get("pending_id") != channel_payload.get("pending_id"): return False card_target = card.get("target") or {} channel_target = channel_payload.get("target") or {} for key in TARGET_KEYS: if key not in card_target and key not in channel_target: continue if card_target.get(key) != channel_target.get(key): return False return True