kangshifu-board/integrations/wecom-identity/identity_bridge.py

"""工具桥 v3：把「当前这次工具调用」绑定到可信身份（隔离副本，未上机）

v3 修订（对应 2026-11-02 二轮复核 1/2/4/5）：
1. **不再回落到 os.environ**：`get_session_env()` 在 ContextVar 为 `_UNSET` 时会去读环境变量
   （源码注释：「never set here」→ falls back to os.environ），这正是「合成 env 就能骗过」的口子。
   v3 直接读 `gateway.session_context._VAR_MAP` 里的 ContextVar：`_UNSET` → 空串（本回合未绑定），
   **绝不查环境变量**；`_VAR_MAP` 不存在 → `SESSION_READER_UNAVAILABLE`（明确失败）。
2. **有效会话键**：单聊原始帧无 chatid 时落盘 `chatId=null`，而 `SessionSource.chat_id=sender_id`，
   故核对用 `aibot_identity.effective_chat_id()`（= chatId or senderId），不再拿原始 chatId 硬比。
3. **botId 唯一来源 = 受控实例登记**（适配器自己写 `instance.json`），删掉顶层优先/extra 优先/正则兜底那套猜测。
4. **只服务 aibot**：`wecom_callback` 明确不支持（回调模式的来源/字段是另一套，不混接）。
"""
from __future__ import annotations

import json
import logging
import os
import pathlib
from typing import Any, Callable, Dict, Optional

try:  # 运行时（平台插件包内）
    from plugins.platforms.wecom.aibot_identity import (  # type: ignore
        DEFAULT_TTL_SECONDS, IdentityRejected, effective_chat_id, identity_summary,
        normalize_session_chat_type, read_context, read_instance_registry,
    )
except ImportError:  # 离线夹具/独立加载
    import sys
    sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
    from aibot_identity import (  # type: ignore
        DEFAULT_TTL_SECONDS, IdentityRejected, effective_chat_id, identity_summary,
        normalize_session_chat_type, read_context, read_instance_registry,
    )

logger = logging.getLogger(__name__)

TOOL_NAME = "wecom_identity_current"
TOOLSET = "wecom"
SUPPORTED_PLATFORMS = ("wecom",)   # 只 aibot；wecom_callback 不接

_UNSET_SENTINEL = object()


class SessionReaderUnavailable(Exception):
    """生产会话读取器不可用（版本不符/缺变量）—— 不许退回环境变量猜。"""

    def __init__(self, code: str):
        super().__init__(code)
        self.code = code


def strict_session_reader() -> Callable[[str], str]:
    """**无环境变量回落**的当前回合读取器：只认 ContextVar。

    生产 reader 缺失（`_VAR_MAP` 不在）或某个变量缺失 → 抛 SessionReaderUnavailable。
    `_UNSET`（本回合未绑定）→ 返回 ""（空），由上层判失败关闭；**不查 os.environ**。
    """
    try:
        import gateway.session_context as session_context  # type: ignore
    except Exception as exc:  # noqa: BLE001
        raise SessionReaderUnavailable("SESSION_READER_UNAVAILABLE") from exc

    var_map = getattr(session_context, "_VAR_MAP", None)
    unset = getattr(session_context, "_UNSET", None)
    if not isinstance(var_map, dict) or "HERMES_SESSION_MESSAGE_ID" not in var_map:
        raise SessionReaderUnavailable("SESSION_READER_UNAVAILABLE")
    if unset is None:
        # 没有 `_UNSET` 哨兵就无法区分「本回合未绑定」与「真值」——
        # 若退回自造哨兵，未绑定的 ContextVar 会把内部对象 repr 当身份值输出。
        # 一律固定失败码（不输出原始异常）。
        raise SessionReaderUnavailable("SESSION_READER_UNAVAILABLE")

    def _read(name: str) -> str:
        var = var_map.get(name)
        if var is None:
            raise SessionReaderUnavailable(f"SESSION_VAR_MISSING:{name}")
        try:
            value = var.get()
        except Exception:  # noqa: BLE001 — 固定码，不泄原始异常
            raise SessionReaderUnavailable("SESSION_VAR_UNAVAILABLE") from None
        if value is unset:            # 本回合未绑定 —— 明确为空，**不回落 env**
            return ""
        if isinstance(value, (str, bytes)):
            return value.decode("utf-8", "replace") if isinstance(value, bytes) else value
        # 非字符串值不该出现在会话变量里：视为不可信，固定失败码
        raise SessionReaderUnavailable("SESSION_VAR_UNAVAILABLE")

    return _read


def current_identity(args: Any = None, *, reader: Optional[Callable[[str], str]] = None,
                     directory: Optional[str] = None,
                     ttl_seconds: int = DEFAULT_TTL_SECONDS,
                     now: Optional[float] = None,
                     instance_dir: Optional[str] = None) -> Dict[str, Any]:
    """返回**已核验**的本次调用身份；任何不确定一律 ok=False（失败关闭）。"""
    try:
        read = reader or strict_session_reader()
    except SessionReaderUnavailable as exc:
        return {"ok": False, "error": exc.code}

    try:
        platform = str(read("HERMES_SESSION_PLATFORM") or "").strip().lower()
        message_id = str(read("HERMES_SESSION_MESSAGE_ID") or "").strip()
        sender_id = str(read("HERMES_SESSION_USER_ID") or "").strip()
        raw_type = str(read("HERMES_SESSION_CHAT_TYPE") or "").strip()
        session_chat_id = str(read("HERMES_SESSION_CHAT_ID") or "").strip() or None
    except SessionReaderUnavailable as exc:
        return {"ok": False, "error": exc.code}

    if not platform:
        # 本回合没有绑定平台 = 没有会话上下文（不是「别的平台」）
        return {"ok": False, "error": "SESSION_IDENTITY_INCOMPLETE",
                "have": {"platform": False}}
    if platform not in SUPPORTED_PLATFORMS:
        # 含 wecom_callback —— 明确不接，不猜
        return {"ok": False, "error": "UNSUPPORTED_PLATFORM", "platform": platform}

    registry = read_instance_registry(instance_dir or directory)
    if registry is None:
        return {"ok": False, "error": "INSTANCE_REGISTRY_MISSING"}   # 不猜配置、不读环境变量
    trusted_bot = str(registry["botId"]).strip()

    if not (message_id and sender_id):
        return {"ok": False, "error": "SESSION_IDENTITY_INCOMPLETE",
                "have": {"message_id": bool(message_id), "sender_id": bool(sender_id)}}

    if isinstance(args, dict):
        for key, trusted in (("message_id", message_id), ("sender_id", sender_id), ("bot_id", trusted_bot)):
            supplied = str(args.get(key) or "").strip()
            if supplied and supplied != trusted:
                return {"ok": False, "error": "CALLER_VALUE_CONFLICT", "field": key}

    chat_type = normalize_session_chat_type(raw_type)
    if chat_type is None:
        return {"ok": False, "error": "UNKNOWN_CHAT_TYPE", "chat_type": raw_type}
    if chat_type == "group" and not session_chat_id:
        return {"ok": False, "error": "MISSING_GROUP_CHAT_ID"}
    # 有效会话键：单聊 = session chat_id or sender_id（与 `chat_id = chatid or sender_id` 同一条规则）；
    # 群聊必须显式有 chat_id。这样「单聊帧无 chatid」不会因为落盘 chatId=null 而被误判。
    expected_chat = session_chat_id if chat_type == "group" else (session_chat_id or sender_id)

    try:
        ctx = read_context(trusted_bot, message_id,
                           expect={"senderId": sender_id, "chatType": chat_type,
                                   "chatId": expected_chat},   # 有效会话键
                           directory=directory, ttl_seconds=ttl_seconds, now=now)
    except IdentityRejected as exc:
        return {"ok": False, "error": exc.code}
    if ctx is None:
        return {"ok": False, "error": "IDENTITY_CONTEXT_UNAVAILABLE",
                "detail": "无本消息的可信上下文（未落盘/已过期/字段与当前调用不一致）"}

    return {"ok": True,
            "trust": "gateway_contextvar(no-env-fallback)+adapter_instance_registry",
            "identity": identity_summary(ctx),
            "effectiveChatId": effective_chat_id(ctx["chatType"], ctx["chatId"], ctx["senderId"])}


TOOL_SCHEMA = {
    "name": TOOL_NAME,
    "description": ("返回**本次消息**已核验的企微 aibot 发送者身份（botId/senderId/chatType/chatId/messageId）。"
                    "身份取自网关会话上下文与本实例受控登记，模型无法指定，也不需要任何参数。"
                    "ok=false 时必须停止身份登记，不得改用正文、session 键或旧文件。"),
    "parameters": {"type": "object", "properties": {}, "additionalProperties": False},
}


def _handler(args: Optional[Dict[str, Any]] = None, **_kw: Any) -> str:
    # Only the registered tool returns text. Registration calls current_identity()
    # directly and still requires its structured, verified identity result.
    if args is not None and (not isinstance(args, dict) or args):
        return json.dumps({"ok": False, "error": "NO_ARGUMENTS_ALLOWED"})
    return json.dumps(current_identity(), ensure_ascii=False)


def register(ctx) -> None:
    """插件入口：`ctx.register_tool(name, toolset, schema, handler, …)`。"""
    ctx.register_tool(TOOL_NAME, TOOLSET, TOOL_SCHEMA, _handler,
                      description=TOOL_SCHEMA["description"], emoji="🪪")
    logger.info("wecom-identity bridge loaded: tool=%s version=v3", TOOL_NAME)