From 8998f94c42e053c2a89f54f8d028dfb0e0199ccb Mon Sep 17 00:00:00 2001 From: chiguyong Date: Thu, 25 Jun 2026 20:45:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels):=20U12=20=E2=80=94=20DingTalk/We?= =?UTF-8?q?Com/Slack=20adapters=20+=20multi-channel=20webhook=20dispatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/agentkit/channels/dingtalk.py | 281 ++++++++++++++++++ src/agentkit/channels/slack.py | 287 ++++++++++++++++++ src/agentkit/channels/wecom.py | 390 +++++++++++++++++++++++++ src/agentkit/server/routes/channels.py | 131 +++++++-- tests/unit/channels/test_dingtalk.py | 304 +++++++++++++++++++ tests/unit/channels/test_feishu.py | 8 +- tests/unit/channels/test_slack.py | 320 ++++++++++++++++++++ tests/unit/channels/test_wecom.py | 323 ++++++++++++++++++++ 8 files changed, 2014 insertions(+), 30 deletions(-) create mode 100644 src/agentkit/channels/dingtalk.py create mode 100644 src/agentkit/channels/slack.py create mode 100644 src/agentkit/channels/wecom.py create mode 100644 tests/unit/channels/test_dingtalk.py create mode 100644 tests/unit/channels/test_slack.py create mode 100644 tests/unit/channels/test_wecom.py diff --git a/src/agentkit/channels/dingtalk.py b/src/agentkit/channels/dingtalk.py new file mode 100644 index 0000000..33a43c9 --- /dev/null +++ b/src/agentkit/channels/dingtalk.py @@ -0,0 +1,281 @@ +"""钉钉 IM 适配器 (U12)。 + +实现 :class:`MessageAdapter` 协议，对接钉钉企业内机器人 outgoing/webhook 回调。 + +关键设计决策： +- 签名校验 fail-closed：``Sign`` + ``Timestamp`` 头缺失时仅当 token 校验通过才放行。 +- 时间戳窗口 3600s（钉钉官方窗口 1 小时）。 +- ``accessToken`` 简单 TTL 缓存（7200s）；钉钉 token 有效期 2 小时。 +- httpx 客户端懒构造，避免未使用的适配器持有连接池。 +- 钉钉无 URL verification challenge 流程 — 合法签名请求直接 200。 +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import logging +import re +import time +from typing import Any + +import httpx + +from agentkit.channels.base import ( + ChannelType, + IncomingMessage, + MessageAdapter, + OutgoingMessage, +) + +logger = logging.getLogger(__name__) + +# 签名时间戳允许的最大偏移（秒）— 钉钉官方窗口 1 小时 +_SIGNATURE_MAX_AGE_SECONDS = 3600 +# accessToken 缓存 TTL（秒）— 钉钉 token 有效期 2 小时 +_TOKEN_CACHE_TTL = 7200.0 + +# 钉钉 API 端点 +_ACCESS_TOKEN_URL = "https://api.dingtalk.com/v1.0/oauth2/accessToken" +_SEND_MESSAGE_URL = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend" + +# 钉钉 @ 机器人前缀剥离 — 内容形如 "@robotName 实际消息" +# ponytail: 简单正则剥离首个 @token。天花板：无法区分 @ 机器人与 @ 用户； +# 升级路径：从 config 注入 robot_name 精确匹配。 +_MENTION_PREFIX_RE = re.compile(r"^@\S+\s*") + + +class DingTalkMessageAdapter(MessageAdapter): + """钉钉 IM 适配器。 + + 生命周期： + ``__init__`` → :meth:`verify_signature` → :meth:`receive_message` + → :meth:`send_message` → :meth:`close` + + Args: + app_key: 钉钉应用 App Key。 + app_secret: 钉钉应用 App Secret（同时作为签名密钥）。 + robot_code: 机器人 robotCode（发送消息时使用）。 + token: 可选 token 校验值（部分机器人通过 ``Token`` 头校验）。 + """ + + def __init__( + self, + app_key: str, + app_secret: str, + robot_code: str, + token: str | None = None, + ) -> None: + self.app_key = app_key + self.app_secret = app_secret + self.robot_code = robot_code + self.token = token + # 懒加载 httpx 客户端 + self._client: httpx.AsyncClient | None = None + # ponytail: 简单 TTL 缓存 (token, expiry)。天花板：单实例内存； + # 升级路径：Redis 缓存共享给多实例。 + self._token_cache: tuple[str, float] | None = None + + # ------------------------------------------------------------------ + # httpx 客户端懒加载 + # ------------------------------------------------------------------ + + def _get_client(self) -> httpx.AsyncClient: + if self._client is None: + self._client = httpx.AsyncClient(timeout=10.0) + return self._client + + # ------------------------------------------------------------------ + # 签名验证 + # ------------------------------------------------------------------ + + async def verify_signature(self, headers: dict[str, str], body: bytes) -> bool: + """验证钉钉 webhook 签名。 + + - 若配置了 ``token``：校验 ``Token`` 头，不匹配返回 False。 + - 若存在 ``Sign`` + ``Timestamp`` 头：校验 HMAC-SHA256 签名与时间戳新鲜度。 + - 两者皆无：仅当 token 已校验通过才放行（token-only 模式）。 + + Args: + headers: HTTP 请求头（键大小写不敏感查找）。 + body: 原始请求体字节（钉钉签名不依赖 body）。 + + Returns: + True 表示签名校验通过。 + """ + # Token 校验（若配置） + if self.token is not None: + token_header = _header_get(headers, "Token") + if token_header is None or not hmac.compare_digest(token_header, self.token): + return False + + sign = _header_get(headers, "Sign") + timestamp_str = _header_get(headers, "Timestamp") + + if sign is None and timestamp_str is None: + # 无签名头：仅当 token 已校验通过才放行 + return self.token is not None + + if sign is None or timestamp_str is None: + return False + + # 时间戳新鲜度（毫秒 → 秒） + try: + ts_ms = int(timestamp_str) + except ValueError: + return False + ts_sec = ts_ms / 1000.0 + now = time.time() + if abs(now - ts_sec) > _SIGNATURE_MAX_AGE_SECONDS: + logger.warning("钉钉 webhook 时间戳超出 %ds 窗口 — 拒绝", _SIGNATURE_MAX_AGE_SECONDS) + return False + + # 计算签名：base64(hmac-sha256(key=app_secret, msg="{timestamp}\n{app_secret}")) + string_to_sign = f"{timestamp_str}\n{self.app_secret}" + expected = base64.b64encode( + hmac.new( + self.app_secret.encode("utf-8"), + string_to_sign.encode("utf-8"), + hashlib.sha256, + ).digest() + ).decode("utf-8") + return hmac.compare_digest(sign, expected) + + # ------------------------------------------------------------------ + # 消息接收 / 解析 + # ------------------------------------------------------------------ + + async def receive_message(self, headers: dict[str, str], body: bytes) -> IncomingMessage: + """解析钉钉 webhook 事件为标准化 :class:`IncomingMessage`。 + + 钉钉无 URL verification challenge 流程 — 合法请求直接解析为消息。 + + Raises: + ValueError: 事件 body 不是合法 JSON。 + """ + try: + data: dict[str, Any] = json.loads(body) + except json.JSONDecodeError as exc: + raise ValueError(f"钉钉事件 body 不是合法 JSON: {exc}") from exc + + chat_id = data.get("conversationId", "") + user_id = data.get("senderStaffId") or data.get("senderId") or data.get("staffId") or "" + msg_id = data.get("msgId", "") + session_expired = data.get("sessionWebhookExpiredTime", "") + timestamp = str(session_expired) if session_expired else "" + + content = self._extract_content(data) + + return IncomingMessage( + channel=ChannelType.DINGTALK, + platform_message_id=msg_id, + user_id=user_id, + chat_id=chat_id, + content=content, + raw_event=data, + timestamp=timestamp, + ) + + def _extract_content(self, data: dict[str, Any]) -> str: + """从钉钉事件提取文本内容。 + + - text 类型：解析 ``text.content``，剥离 @ 机器人前缀。 + - 其他类型：返回 ``[unsupported message type: {type}]``。 + """ + msgtype = data.get("msgtype", "") + if msgtype == "text": + text_obj = data.get("text", {}) + content = text_obj.get("content", "") if isinstance(text_obj, dict) else "" + # 剥离 @ 机器人前缀 + return _MENTION_PREFIX_RE.sub("", content, count=1).strip() + return f"[unsupported message type: {msgtype}]" + + # ------------------------------------------------------------------ + # 消息发送 + # ------------------------------------------------------------------ + + async def send_message(self, message: OutgoingMessage) -> bool: + """向钉钉发送文本消息（oToMessages/batchSend 单聊）。 + + Returns: + True 表示 HTTP 200。 + """ + try: + token = await self._get_access_token() + if not token: + return False + + client = self._get_client() + payload = { + "robotCode": self.robot_code, + "conversationId": message.chat_id, + "msgKey": "sampleText", + "msgParam": json.dumps({"content": message.content}), + } + resp = await client.post( + _SEND_MESSAGE_URL, + json=payload, + headers={"x-acs-dingtalk-access-token": token}, + ) + if resp.status_code != 200: + logger.error("钉钉 send_message HTTP %d: %s", resp.status_code, resp.text[:200]) + return False + return True + except httpx.HTTPError as exc: + logger.error("钉钉 send_message 网络错误: %s", exc) + return False + + async def _get_access_token(self) -> str | None: + """获取并缓存钉钉 ``accessToken``。""" + # 命中缓存 + if self._token_cache is not None: + token, expiry = self._token_cache + if time.monotonic() < expiry: + return token + + try: + client = self._get_client() + resp = await client.post( + _ACCESS_TOKEN_URL, + json={"appKey": self.app_key, "appSecret": self.app_secret}, + ) + if resp.status_code != 200: + logger.error("钉钉 accessToken HTTP %d: %s", resp.status_code, resp.text[:200]) + return None + data = resp.json() + token = data.get("accessToken", "") + if not token: + return None + self._token_cache = (token, time.monotonic() + _TOKEN_CACHE_TTL) + return token + except httpx.HTTPError as exc: + logger.error("钉钉 accessToken 网络错误: %s", exc) + return None + + # ------------------------------------------------------------------ + # 资源释放 + # ------------------------------------------------------------------ + + async def close(self) -> None: + """关闭 httpx 客户端（如已创建）。""" + if self._client is not None: + await self._client.aclose() + self._client = None + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _header_get(headers: dict[str, str], name: str) -> str | None: + """大小写不敏感的 header 查找。""" + if name in headers: + return headers[name] + lower = name.lower() + for k, v in headers.items(): + if k.lower() == lower: + return v + return None diff --git a/src/agentkit/channels/slack.py b/src/agentkit/channels/slack.py new file mode 100644 index 0000000..e9cf6fd --- /dev/null +++ b/src/agentkit/channels/slack.py @@ -0,0 +1,287 @@ +"""Slack IM 适配器 (U12)。 + +实现 :class:`MessageAdapter` 协议，对接 Slack Events API 与 Slash Commands。 + +关键设计决策： +- 签名：``v0:{timestamp}:{body}`` 的 HMAC-SHA256，与 ``X-Slack-Signature`` 比对。 +- 时间戳窗口 300s（Slack 官方推荐 5 分钟）。 +- URL 验证：``url_verification`` 事件抛 :class:`URLVerificationChallenge`。 +- Events API 与 Slash Commands 双流程：JSON body → Events API；form-encoded → Slash。 +- ``<@U12345>`` 提及标记剥离。 +- httpx 客户端懒构造。 +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import logging +import re +import time +from typing import Any +from urllib.parse import parse_qs +from uuid import uuid4 + +import httpx + +from agentkit.channels.base import ( + ChannelType, + IncomingMessage, + MessageAdapter, + OutgoingMessage, +) + +logger = logging.getLogger(__name__) + +# 签名时间戳允许的最大偏移（秒）— Slack 官方推荐 5 分钟 +_SIGNATURE_MAX_AGE_SECONDS = 300 + +# Slack API 端点 +_SEND_MESSAGE_URL = "https://slack.com/api/chat.postMessage" + +# Slack <@U12345> / <@U12345|name> 提及标记剥离 +_MENTION_RE = re.compile(r"<@[^>]+>\s*") + + +class URLVerificationChallenge(Exception): + """Slack URL 验证事件 — webhook 端点需返回 ``{"challenge": ...}`` 响应。 + + Slack 在配置 Events API 时发送 ``url_verification`` 事件，要求服务端 + 原样返回 ``challenge`` 字段以验证 URL 可达。 + """ + + def __init__(self, challenge: str) -> None: + super().__init__(f"Slack URL verification challenge: {challenge}") + self.challenge = challenge + + +class SignatureVerificationError(Exception): + """``verification_token`` 校验失败 — 拒绝处理。""" + + +class SlackMessageAdapter(MessageAdapter): + """Slack IM 适配器。 + + 生命周期： + ``__init__`` → :meth:`verify_signature` → :meth:`receive_message` + → :meth:`send_message` → :meth:`close` + + Args: + bot_token: Slack Bot User OAuth Token（``xoxb-`` 开头）。 + signing_secret: Slack 应用 Signing Secret（签名校验）。 + verification_token: 可选旧版 Verification Token（URL 验证时校验）。 + """ + + def __init__( + self, + bot_token: str, + signing_secret: str, + verification_token: str | None = None, + ) -> None: + self.bot_token = bot_token + self.signing_secret = signing_secret + self.verification_token = verification_token + # 懒加载 httpx 客户端 + self._client: httpx.AsyncClient | None = None + + # ------------------------------------------------------------------ + # httpx 客户端懒加载 + # ------------------------------------------------------------------ + + def _get_client(self) -> httpx.AsyncClient: + if self._client is None: + self._client = httpx.AsyncClient(timeout=10.0) + return self._client + + # ------------------------------------------------------------------ + # 签名验证 + # ------------------------------------------------------------------ + + async def verify_signature(self, headers: dict[str, str], body: bytes) -> bool: + """验证 Slack webhook 签名。 + + 签名 = ``v0=`` + hmac-sha256(key=signing_secret, msg="v0:{timestamp}:{body}")， + 与 ``X-Slack-Signature`` 头比对。时间戳超过 5 分钟视为重放攻击。 + + Args: + headers: HTTP 请求头（键大小写不敏感查找）。 + body: 原始请求体字节。 + + Returns: + True 表示签名校验通过。 + """ + signature = _header_get(headers, "X-Slack-Signature") + timestamp_str = _header_get(headers, "X-Slack-Request-Timestamp") + if not signature or not timestamp_str: + return False + + # 时间戳重放保护 + try: + ts = int(timestamp_str) + except ValueError: + return False + now = time.time() + if abs(now - ts) > _SIGNATURE_MAX_AGE_SECONDS: + logger.warning("Slack webhook 时间戳超出 %ds 窗口 — 拒绝", _SIGNATURE_MAX_AGE_SECONDS) + return False + + # 计算签名：v0={timestamp}:{body} + base = f"v0:{timestamp_str}:{body.decode('utf-8')}" + expected = ( + "v0=" + + hmac.new( + self.signing_secret.encode("utf-8"), + base.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + ) + return hmac.compare_digest(signature, expected) + + # ------------------------------------------------------------------ + # 消息接收 / 解析 + # ------------------------------------------------------------------ + + async def receive_message(self, headers: dict[str, str], body: bytes) -> IncomingMessage: + """解析 Slack webhook 事件为标准化 :class:`IncomingMessage`。 + + - JSON body：Events API 或 URL 验证。 + - form-encoded body：Slash Command。 + + Raises: + URLVerificationChallenge: URL 验证事件。 + SignatureVerificationError: ``verification_token`` 不匹配。 + """ + # 优先尝试 JSON 解析（Events API / URL 验证） + try: + data: dict[str, Any] = json.loads(body) + except (json.JSONDecodeError, UnicodeDecodeError): + data = {} + + if data: + return self._parse_event(data) + + # 非 JSON — 视为 Slash Command（form-encoded） + return self._parse_slash_command(body) + + def _parse_event(self, data: dict[str, Any]) -> IncomingMessage: + """解析 Events API 事件。""" + # URL 验证流程 + if data.get("type") == "url_verification": + challenge = data.get("challenge", "") + token = data.get("token", "") + if self.verification_token is not None and not hmac.compare_digest( + token, self.verification_token + ): + raise SignatureVerificationError("verification_token 不匹配") + raise URLVerificationChallenge(challenge) + + event = data.get("event", {}) + event_id = data.get("event_id", "") + event_type = event.get("type", "") + user = str(event.get("user", "")) + channel = event.get("channel", "") + text = event.get("text", "") + ts = event.get("ts", "") + + # 仅支持 message / app_mention 事件 + if event_type not in ("message", "app_mention"): + return IncomingMessage( + channel=ChannelType.SLACK, + platform_message_id=event_id, + user_id=user, + chat_id=channel, + content=f"[unsupported event type: {event_type}]", + raw_event=data, + timestamp=ts, + ) + + # 剥离 <@U12345> 提及标记 + text = _MENTION_RE.sub("", text).strip() + + return IncomingMessage( + channel=ChannelType.SLACK, + platform_message_id=event_id, + user_id=user, + chat_id=channel, + content=text, + raw_event=data, + timestamp=ts, + ) + + def _parse_slash_command(self, body: bytes) -> IncomingMessage: + """解析 Slash Command（form-encoded body）。""" + params = parse_qs(body.decode("utf-8")) + text = params.get("text", [""])[0] + user_id = params.get("user_id", [""])[0] + channel_id = params.get("channel_id", [""])[0] + command = params.get("command", [""])[0] + + return IncomingMessage( + channel=ChannelType.SLACK, + platform_message_id=f"slash-{uuid4()}", + user_id=user_id, + chat_id=channel_id, + content=text, + raw_event={"command": command, "form": {k: v[0] for k, v in params.items()}}, + timestamp=str(time.time()), + ) + + # ------------------------------------------------------------------ + # 消息发送 + # ------------------------------------------------------------------ + + async def send_message(self, message: OutgoingMessage) -> bool: + """向 Slack 发送文本消息（chat.postMessage）。 + + Returns: + True 表示 HTTP 200 且响应 ``ok == true``。 + """ + try: + client = self._get_client() + resp = await client.post( + _SEND_MESSAGE_URL, + headers={"Authorization": f"Bearer {self.bot_token}"}, + json={"channel": message.chat_id, "text": message.content}, + ) + if resp.status_code != 200: + logger.error("Slack send_message HTTP %d: %s", resp.status_code, resp.text[:200]) + return False + data = resp.json() + if not data.get("ok"): + logger.error( + "Slack send_message 业务失败 ok=%s error=%s", + data.get("ok"), + data.get("error", "")[:200], + ) + return False + return True + except httpx.HTTPError as exc: + logger.error("Slack send_message 网络错误: %s", exc) + return False + + # ------------------------------------------------------------------ + # 资源释放 + # ------------------------------------------------------------------ + + async def close(self) -> None: + """关闭 httpx 客户端（如已创建）。""" + if self._client is not None: + await self._client.aclose() + self._client = None + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _header_get(headers: dict[str, str], name: str) -> str | None: + """大小写不敏感的 header 查找。""" + if name in headers: + return headers[name] + lower = name.lower() + for k, v in headers.items(): + if k.lower() == lower: + return v + return None diff --git a/src/agentkit/channels/wecom.py b/src/agentkit/channels/wecom.py new file mode 100644 index 0000000..49aa1e7 --- /dev/null +++ b/src/agentkit/channels/wecom.py @@ -0,0 +1,390 @@ +"""企业微信 IM 适配器 (U12)。 + +实现 :class:`MessageAdapter` 协议，对接企业微信回调 webhook。 + +关键设计决策： +- EncodingAESKey（43 字符 base64）→ 32 字节 AES-256-CBC 密钥。 +- 签名：sha1(sorted([token, timestamp, nonce, encrypt])) 与 ``msg_signature`` 比对。 +- 加密协议：AES-256-CBC，明文 = random(16) + msg_len(4, 大端) + msg + app_id。 + app_id 校验需匹配 ``corp_id``。 +- ``access_token`` 简单 TTL 缓存（7200s）。 +- XML 解析使用 stdlib ``xml.etree.ElementTree``。 +- httpx 客户端懒构造。 +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import logging +import os +import time +import xml.etree.ElementTree as ET + +import httpx + +from agentkit.channels.base import ( + ChannelType, + IncomingMessage, + MessageAdapter, + OutgoingMessage, +) + +logger = logging.getLogger(__name__) + +# access_token 缓存 TTL（秒）— 企微 token 有效期 2 小时 +_TOKEN_CACHE_TTL = 7200.0 + +# 企微 API 端点 +_GET_TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" +_SEND_MESSAGE_URL = "https://qyapi.weixin.qq.com/cgi-bin/message/send" + + +class WeComURLVerification(Exception): + """企微 URL 验证事件 — webhook 端点需返回 XML 响应。 + + 企微在配置回调时发送含 ``EchoStr`` 的加密事件，服务端需解密、 + 重新加密后包装成 XML 响应返回。 + """ + + def __init__(self, response_xml: str) -> None: + super().__init__("WeCom URL verification") + self.response_xml = response_xml + + +class WeComError(Exception): + """企微消息处理错误（解密失败、app_id 不匹配等）。""" + + +class WeComMessageAdapter(MessageAdapter): + """企业微信 IM 适配器。 + + 生命周期： + ``__init__`` → :meth:`verify_signature` → :meth:`receive_message` + → :meth:`send_message` → :meth:`close` + + Args: + corp_id: 企业 ID（解密时校验 app_id）。 + agent_id: 应用 AgentID（发送消息时使用）。 + corp_secret: 应用凭证密钥（获取 access_token）。 + token: 回调配置的 Token（签名校验）。 + encoding_aes_key: 43 字符 EncodingAESKey（AES 密钥来源）。 + """ + + def __init__( + self, + corp_id: str, + agent_id: int, + corp_secret: str, + token: str, + encoding_aes_key: str, + ) -> None: + self.corp_id = corp_id + self.agent_id = agent_id + self.corp_secret = corp_secret + self.token = token + self.encoding_aes_key = encoding_aes_key + # 懒加载 httpx 客户端 + self._client: httpx.AsyncClient | None = None + # ponytail: 简单 TTL 缓存。天花板：单实例内存；升级路径：Redis 共享。 + self._token_cache: tuple[str, float] | None = None + + # ------------------------------------------------------------------ + # httpx 客户端懒加载 + # ------------------------------------------------------------------ + + def _get_client(self) -> httpx.AsyncClient: + if self._client is None: + self._client = httpx.AsyncClient(timeout=10.0) + return self._client + + # ------------------------------------------------------------------ + # AES 密钥 / 加解密 + # ------------------------------------------------------------------ + + def _decode_aes_key(self) -> bytes: + """EncodingAESKey（43 字符）→ 32 字节 AES 密钥。""" + return base64.b64decode(self.encoding_aes_key + "=") + + def _encrypt(self, plaintext_str: str) -> str: + """AES-256-CBC 加密 — 返回 base64(IV + 密文)。 + + 明文结构：random(16) + msg_len(4, 大端) + msg + corp_id，PKCS7 填充。 + """ + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives.padding import PKCS7 + + key = self._decode_aes_key() + iv = os.urandom(16) + random_prefix = os.urandom(16) + msg_bytes = plaintext_str.encode("utf-8") + msg_len = len(msg_bytes).to_bytes(4, "big") + app_id = self.corp_id.encode("utf-8") + plaintext = random_prefix + msg_len + msg_bytes + app_id + + padder = PKCS7(algorithms.AES.block_size).padder() + padded = padder.update(plaintext) + padder.finalize() + + cipher = Cipher(algorithms.AES(key), modes.CBC(iv)) + encryptor = cipher.encryptor() + ciphertext = encryptor.update(padded) + encryptor.finalize() + + return base64.b64encode(iv + ciphertext).decode("utf-8") + + def _decrypt(self, encrypt_b64: str) -> str: + """AES-256-CBC 解密 — 校验 app_id 后返回 msg 明文。 + + Raises: + WeComError: app_id 不匹配 corp_id，或密文损坏。 + """ + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives.padding import PKCS7 + + key = self._decode_aes_key() + ciphertext = base64.b64decode(encrypt_b64) + if len(ciphertext) < 17: # IV(16) + 至少 1 字节密文 + raise WeComError("企微密文长度不足") + + iv = ciphertext[:16] + encrypted = ciphertext[16:] + + cipher = Cipher(algorithms.AES(key), modes.CBC(iv)) + decryptor = cipher.decryptor() + padded = decryptor.update(encrypted) + decryptor.finalize() + + unpadder = PKCS7(algorithms.AES.block_size).unpadder() + plaintext = unpadder.update(padded) + unpadder.finalize() + + # plaintext = random(16) + msg_len(4, 大端) + msg + app_id + if len(plaintext) < 20: + raise WeComError("企微解密后明文长度不足") + msg_len = int.from_bytes(plaintext[16:20], "big") + msg = plaintext[20 : 20 + msg_len] + app_id = plaintext[20 + msg_len :].decode("utf-8", errors="replace") + + if app_id != self.corp_id: + raise WeComError(f"app_id 不匹配: {app_id} != {self.corp_id}") + + return msg.decode("utf-8") + + # ------------------------------------------------------------------ + # XML 解析辅助 + # ------------------------------------------------------------------ + + def _extract_encrypt(self, body: bytes) -> str | None: + """从外层 XML body 提取 ``Encrypt`` 字段。""" + try: + root = ET.fromstring(body.decode("utf-8")) + except (ET.ParseError, UnicodeDecodeError): + return None + encrypt_elem = root.find("Encrypt") + if encrypt_elem is None or encrypt_elem.text is None: + return None + return encrypt_elem.text + + def _parse_xml(self, xml_str: str) -> dict[str, str]: + """解析 XML 为 ``{tag: text}`` 字典（无命名空间）。""" + root = ET.fromstring(xml_str) + result: dict[str, str] = {} + for child in root: + result[child.tag] = child.text or "" + return result + + # ------------------------------------------------------------------ + # 签名验证 + # ------------------------------------------------------------------ + + async def verify_signature(self, headers: dict[str, str], body: bytes) -> bool: + """验证企微 webhook 签名。 + + 签名 = sha1(sorted([token, timestamp, nonce, encrypt]).join(""))， + 与 query 参数 ``msg_signature`` 比对。timestamp/nonce/msg_signature + 由 webhook 端点从 query 参数合并到 headers dict 中。 + + Args: + headers: HTTP 请求头（含已合并的 query 参数）。 + body: 原始请求体字节（XML）。 + + Returns: + True 表示签名校验通过。 + """ + msg_signature = _header_get(headers, "msg_signature") + timestamp = _header_get(headers, "timestamp") + nonce = _header_get(headers, "nonce") + if not msg_signature or not timestamp or not nonce: + return False + + encrypt = self._extract_encrypt(body) + if not encrypt: + return False + + # sha1(sorted([token, timestamp, nonce, encrypt]).join("")) + parts = sorted([self.token, timestamp, nonce, encrypt]) + expected = hashlib.sha1("".join(parts).encode("utf-8")).hexdigest() + return hmac.compare_digest(msg_signature, expected) + + # ------------------------------------------------------------------ + # 消息接收 / 解析 + # ------------------------------------------------------------------ + + async def receive_message(self, headers: dict[str, str], body: bytes) -> IncomingMessage: + """解析企微 webhook 事件为标准化 :class:`IncomingMessage`。 + + - URL 验证流程（解密后含 ``EchoStr``）：抛 :class:`WeComURLVerification`。 + - 普通消息：解密后解析内部 XML，提取 FromUserName/MsgId/Content。 + + Raises: + WeComURLVerification: URL 验证事件，携带 XML 响应。 + WeComError: 缺少 Encrypt 字段或解密失败。 + """ + encrypt = self._extract_encrypt(body) + if not encrypt: + raise WeComError("企微 XML body 缺少 Encrypt 字段") + + plaintext = self._decrypt(encrypt) + inner = self._parse_xml(plaintext) + + # URL 验证流程 — 内部 XML 包含 EchoStr + if "EchoStr" in inner: + timestamp = _header_get(headers, "timestamp") or "" + nonce = _header_get(headers, "nonce") or "" + encrypted_echo = self._encrypt(inner["EchoStr"]) + # 计算响应签名 + sig_parts = sorted([self.token, timestamp, nonce, encrypted_echo]) + sig = hashlib.sha1("".join(sig_parts).encode("utf-8")).hexdigest() + response_xml = ( + f"" + f"" + f"{timestamp}" + f"" + ) + raise WeComURLVerification(response_xml) + + # 普通消息 + from_user = inner.get("FromUserName", "") + msg_id = inner.get("MsgId", "") + content = inner.get("Content", "") + create_time = inner.get("CreateTime", "") + + # 群聊消息内容以 ":" 开头 — 剥离 + if content.startswith(":"): + content = content[1:].strip() + + # chat_id: 使用 corp_id + from_user 复合键 + # ponytail: 简单复合键。天花板：群聊场景需独立 ChatId； + # 升级路径：从内部 XML 的 ChatId 字段提取（群聊事件携带）。 + chat_id = f"{self.corp_id}:{from_user}" + + return IncomingMessage( + channel=ChannelType.WECOM, + platform_message_id=msg_id, + user_id=from_user, + chat_id=chat_id, + content=content, + raw_event=inner, + timestamp=create_time, + ) + + # ------------------------------------------------------------------ + # 消息发送 + # ------------------------------------------------------------------ + + async def send_message(self, message: OutgoingMessage) -> bool: + """向企微发送文本消息。 + + Returns: + True 表示 HTTP 200 且 ``errcode == 0``。 + """ + try: + token = await self._get_access_token() + if not token: + return False + + client = self._get_client() + payload = { + "touser": message.chat_id, + "msgtype": "text", + "agentid": self.agent_id, + "text": {"content": message.content}, + } + resp = await client.post( + _SEND_MESSAGE_URL, + params={"access_token": token}, + json=payload, + ) + if resp.status_code != 200: + logger.error("企微 send_message HTTP %d: %s", resp.status_code, resp.text[:200]) + return False + data = resp.json() + if data.get("errcode") != 0: + logger.error( + "企微 send_message 业务失败 errcode=%s errmsg=%s", + data.get("errcode"), + data.get("errmsg", "")[:200], + ) + return False + return True + except httpx.HTTPError as exc: + logger.error("企微 send_message 网络错误: %s", exc) + return False + + async def _get_access_token(self) -> str | None: + """获取并缓存企微 ``access_token``。""" + # 命中缓存 + if self._token_cache is not None: + token, expiry = self._token_cache + if time.monotonic() < expiry: + return token + + try: + client = self._get_client() + resp = await client.get( + _GET_TOKEN_URL, + params={"corpid": self.corp_id, "corpsecret": self.corp_secret}, + ) + if resp.status_code != 200: + logger.error("企微 access_token HTTP %d: %s", resp.status_code, resp.text[:200]) + return None + data = resp.json() + if data.get("errcode") != 0: + logger.error( + "企微 access_token 业务失败 errcode=%s errmsg=%s", + data.get("errcode"), + data.get("errmsg", "")[:200], + ) + return None + token = data.get("access_token", "") + if not token: + return None + self._token_cache = (token, time.monotonic() + _TOKEN_CACHE_TTL) + return token + except httpx.HTTPError as exc: + logger.error("企微 access_token 网络错误: %s", exc) + return None + + # ------------------------------------------------------------------ + # 资源释放 + # ------------------------------------------------------------------ + + async def close(self) -> None: + """关闭 httpx 客户端（如已创建）。""" + if self._client is not None: + await self._client.aclose() + self._client = None + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _header_get(headers: dict[str, str], name: str) -> str | None: + """大小写不敏感的 header 查找。""" + if name in headers: + return headers[name] + lower = name.lower() + for k, v in headers.items(): + if k.lower() == lower: + return v + return None diff --git a/src/agentkit/server/routes/channels.py b/src/agentkit/server/routes/channels.py index 1d549ba..941a169 100644 --- a/src/agentkit/server/routes/channels.py +++ b/src/agentkit/server/routes/channels.py @@ -24,11 +24,18 @@ import time from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import Response from pydantic import BaseModel, Field from agentkit.channels.base import ChannelType, MessageAdapter, OutgoingMessage +from agentkit.channels.dingtalk import DingTalkMessageAdapter from agentkit.channels.feishu import FeishuMessageAdapter, URLVerificationChallenge from agentkit.channels.secrets import SecretsStore +from agentkit.channels.slack import ( + SlackMessageAdapter, + URLVerificationChallenge as SlackURLVerificationChallenge, +) +from agentkit.channels.wecom import WeComMessageAdapter, WeComURLVerification from agentkit.server.auth.dependencies import require_permission from agentkit.server.auth.permissions import Permission @@ -315,39 +322,100 @@ async def delete_channel( async def _build_adapter(channel_id: str) -> MessageAdapter: """根据渠道配置与 secrets 构造适配器实例。 + 支持飞书 / 钉钉 / 企微 / Slack 四种渠道类型，按 ``channel_type`` 分发。 + Raises: - HTTPException: 渠道不存在（404）、渠道类型非飞书（400）、缺少必要凭证（500）。 + HTTPException: 渠道不存在（404）、渠道类型不支持（400）、缺少必要凭证（500）。 """ cfg = _channels.get(channel_id) if cfg is None: raise HTTPException(status_code=404, detail=f"渠道 '{channel_id}' 不存在") - if cfg["channel_type"] != ChannelType.FEISHU.value: - raise HTTPException(status_code=400, detail=f"渠道 '{channel_id}' 不是飞书渠道") store = _get_secrets_store() - app_id = await store.get_secret(f"{channel_id}:app_id") - app_secret = await store.get_secret(f"{channel_id}:app_secret") - encrypt_key = await store.get_secret(f"{channel_id}:encrypt_key") - verification_token = await store.get_secret(f"{channel_id}:verification_token") - if not app_id or not app_secret: - raise HTTPException( - status_code=500, detail=f"渠道 '{channel_id}' 缺少 app_id 或 app_secret" + channel_type = cfg["channel_type"] + + if channel_type == ChannelType.FEISHU.value: + app_id = await store.get_secret(f"{channel_id}:app_id") + app_secret = await store.get_secret(f"{channel_id}:app_secret") + encrypt_key = await store.get_secret(f"{channel_id}:encrypt_key") + verification_token = await store.get_secret(f"{channel_id}:verification_token") + if not app_id or not app_secret: + raise HTTPException( + status_code=500, detail=f"渠道 '{channel_id}' 缺少 app_id 或 app_secret" + ) + return FeishuMessageAdapter( + app_id=app_id, + app_secret=app_secret, + encrypt_key=encrypt_key, + verification_token=verification_token, ) - return FeishuMessageAdapter( - app_id=app_id, - app_secret=app_secret, - encrypt_key=encrypt_key, - verification_token=verification_token, - ) + + if channel_type == ChannelType.DINGTALK.value: + app_key = await store.get_secret(f"{channel_id}:app_key") + app_secret = await store.get_secret(f"{channel_id}:app_secret") + robot_code = await store.get_secret(f"{channel_id}:robot_code") + token = await store.get_secret(f"{channel_id}:token") + if not all([app_key, app_secret, robot_code]): + raise HTTPException( + status_code=500, detail=f"渠道 '{channel_id}' 缺少 dingtalk 凭证" + ) + return DingTalkMessageAdapter( + app_key=app_key, + app_secret=app_secret, + robot_code=robot_code, + token=token, + ) + + if channel_type == ChannelType.WECOM.value: + corp_id = await store.get_secret(f"{channel_id}:corp_id") + corp_secret = await store.get_secret(f"{channel_id}:corp_secret") + token = await store.get_secret(f"{channel_id}:token") + encoding_aes_key = await store.get_secret(f"{channel_id}:encoding_aes_key") + agent_id_raw = await store.get_secret(f"{channel_id}:agent_id") + if not all([corp_id, corp_secret, token, encoding_aes_key, agent_id_raw]): + raise HTTPException( + status_code=500, detail=f"渠道 '{channel_id}' 缺少 wecom 凭证" + ) + try: + agent_id = int(agent_id_raw) + except (TypeError, ValueError) as exc: + raise HTTPException( + status_code=500, + detail=f"渠道 '{channel_id}' agent_id 不是合法整数", + ) from exc + return WeComMessageAdapter( + corp_id=corp_id, + agent_id=agent_id, + corp_secret=corp_secret, + token=token, + encoding_aes_key=encoding_aes_key, + ) + + if channel_type == ChannelType.SLACK.value: + bot_token = await store.get_secret(f"{channel_id}:bot_token") + signing_secret = await store.get_secret(f"{channel_id}:signing_secret") + verification_token = await store.get_secret(f"{channel_id}:verification_token") + if not bot_token or not signing_secret: + raise HTTPException( + status_code=500, detail=f"渠道 '{channel_id}' 缺少 slack 凭证" + ) + return SlackMessageAdapter( + bot_token=bot_token, + signing_secret=signing_secret, + verification_token=verification_token, + ) + + raise HTTPException(status_code=400, detail=f"不支持的渠道类型: {channel_type}") async def _process_inbound_message( - app_state: Any, adapter: FeishuMessageAdapter, message: Any + app_state: Any, adapter: MessageAdapter, message: Any ) -> None: """后台处理入站消息 — 调用 chat 链路并通过适配器回复。 整个流程 try/except 包裹，任何异常仅记录日志，不向上抛出 （webhook 必须保持响应能力）。``adapter.close()`` 在 finally 中调用。 + 适配器类型不限 — 出站消息的 ``channel`` 取自入站消息以匹配平台。 """ try: request_preprocessor = getattr(app_state, "request_preprocessor", None) @@ -395,13 +463,13 @@ async def _process_inbound_message( return outgoing = OutgoingMessage( - channel=ChannelType.FEISHU, + channel=message.channel, chat_id=message.chat_id, content=final_content, ) await adapter.send_message(outgoing) except Exception as exc: # noqa: BLE001 — webhook 必须保持响应能力 - logger.exception("处理飞书入站消息失败: %s", exc) + logger.exception("处理入站消息失败: %s", exc) finally: try: await adapter.close() @@ -421,16 +489,19 @@ def _get_client_ip(request: Request) -> str: @router.post("/channels/{channel_id}/webhook") -async def channel_webhook(channel_id: str, request: Request) -> dict[str, Any]: - """飞书 webhook 端点 — 接收平台事件。 +async def channel_webhook(channel_id: str, request: Request) -> Any: + """渠道 webhook 端点 — 接收平台事件（飞书/钉钉/企微/Slack）。 安全流程（按顺序）： 1. Per-IP 限流（100 req/min）— 超限 429 2. 读取原始 body（未解析） 3. 签名验证 — 失败 401 4. Nonce dedup — 重复返回 200（飞书要求 3s 内响应） - 5. URL verification challenge — 返回 ``{"challenge": ...}`` + 5. URL verification — 飞书/Slack 返回 challenge；企微返回 XML 6. 解析消息 → 后台异步处理 → 立即返回 200 + + 企微通过 query 参数传递 ``msg_signature``/``timestamp``/``nonce``， + 合并到 headers dict 供适配器读取。 """ client_ip = _get_client_ip(request) if not _check_rate_limit(client_ip): @@ -441,21 +512,29 @@ async def channel_webhook(channel_id: str, request: Request) -> dict[str, Any]: adapter = await _build_adapter(channel_id) headers_dict = dict(request.headers) + # 企微等平台通过 query 参数传递签名信息 — 合并到 headers dict 供适配器读取 + for key, value in request.query_params.items(): + if key not in headers_dict: + headers_dict[key] = value + if not await adapter.verify_signature(headers_dict, body): raise HTTPException(status_code=401, detail="签名校验失败") - # Nonce dedup（可选 — 若头不存在则跳过去重） + # Nonce dedup（可选 — 若头不存在则跳过去重；仅飞书携带该头） nonce = request.headers.get("x-lark-request-nonce") if nonce and not _check_nonce_dedup(nonce): return {"code": 0, "msg": "duplicate"} try: message = await adapter.receive_message(headers_dict, body) - except URLVerificationChallenge as e: - # URL 验证流程 — 飞书配置 webhook 时发送 + except (URLVerificationChallenge, SlackURLVerificationChallenge) as e: + # URL 验证流程 — 飞书 / Slack 配置 webhook 时发送 return {"challenge": e.challenge} + except WeComURLVerification as e: + # 企微 URL 验证 — 返回 XML 响应 + return Response(content=e.response_xml, media_type="application/xml") - # 异步处理 — 不阻塞 webhook 响应（飞书要求 3s 内返回 200） + # 异步处理 — 不阻塞 webhook 响应（平台要求快速返回 200） asyncio.create_task(_process_inbound_message(request.app.state, adapter, message)) return {"code": 0} diff --git a/tests/unit/channels/test_dingtalk.py b/tests/unit/channels/test_dingtalk.py new file mode 100644 index 0000000..2257169 --- /dev/null +++ b/tests/unit/channels/test_dingtalk.py @@ -0,0 +1,304 @@ +"""钉钉 IM 适配器单元测试 (U12)。 + +覆盖场景： +- 签名校验（有效/无效/过期/缺签名头） +- Token 校验（匹配/不匹配/未配置） +- 文本消息解析（含 @ 提及剥离） +- 不支持消息类型 +- send_message 成功/失败 +- accessToken 缓存 +- senderStaffId/senderId/staffId 回退 +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import time +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from agentkit.channels.base import ChannelType, IncomingMessage, OutgoingMessage +from agentkit.channels.dingtalk import DingTalkMessageAdapter + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _sign(app_secret: str, timestamp_ms: int) -> str: + """构造钉钉 Sign 头值：base64(hmac-sha256(key=app_secret, msg="{ts}\n{secret}"))。""" + string_to_sign = f"{timestamp_ms}\n{app_secret}" + digest = hmac.new( + app_secret.encode("utf-8"), + string_to_sign.encode("utf-8"), + hashlib.sha256, + ).digest() + return base64.b64encode(digest).decode("utf-8") + + +def _make_body( + *, + text: str = "hello", + msgtype: str = "text", + conversation_id: str = "cid1", + sender_staff_id: str | None = "u1", + sender_id: str | None = None, + staff_id: str | None = None, + msg_id: str = "m1", + session_expired: int = 1700000000000, +) -> dict[str, Any]: + """构造钉钉 webhook 事件 JSON 体。""" + body: dict[str, Any] = { + "conversationId": conversation_id, + "msgId": msg_id, + "msgtype": msgtype, + "sessionWebhookExpiredTime": session_expired, + } + if sender_staff_id is not None: + body["senderStaffId"] = sender_staff_id + if sender_id is not None: + body["senderId"] = sender_id + if staff_id is not None: + body["staffId"] = staff_id + if msgtype == "text": + body["text"] = {"content": text} + else: + body["richText"] = {"content": text} + return body + + +# --------------------------------------------------------------------------- +# 签名校验 +# --------------------------------------------------------------------------- + + +class TestSignatureVerification: + """钉钉签名校验。""" + + async def test_valid_signature(self): + """正确 Sign + Timestamp 返回 True。""" + app_secret = "secret123" + adapter = DingTalkMessageAdapter(app_key="k", app_secret=app_secret, robot_code="r") + ts_ms = int(time.time() * 1000) + headers = {"Sign": _sign(app_secret, ts_ms), "Timestamp": str(ts_ms)} + assert await adapter.verify_signature(headers, b"{}") is True + + async def test_invalid_signature(self): + """篡改 Sign 返回 False。""" + app_secret = "secret123" + adapter = DingTalkMessageAdapter(app_key="k", app_secret=app_secret, robot_code="r") + ts_ms = int(time.time() * 1000) + headers = {"Sign": "tampered_signature", "Timestamp": str(ts_ms)} + assert await adapter.verify_signature(headers, b"{}") is False + + async def test_expired_timestamp(self): + """时间戳超过 1 小时返回 False。""" + app_secret = "secret123" + adapter = DingTalkMessageAdapter(app_key="k", app_secret=app_secret, robot_code="r") + ts_ms = int((time.time() - 3700) * 1000) + headers = {"Sign": _sign(app_secret, ts_ms), "Timestamp": str(ts_ms)} + assert await adapter.verify_signature(headers, b"{}") is False + + async def test_missing_signature_headers(self): + """缺 Sign + Timestamp 头且未配置 token 返回 False。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + assert await adapter.verify_signature({}, b"{}") is False + + +class TestTokenVerification: + """Token 校验。""" + + async def test_token_mismatch(self): + """配置 token 后 Token 头不匹配返回 False。""" + adapter = DingTalkMessageAdapter( + app_key="k", app_secret="s", robot_code="r", token="abc" + ) + headers = {"Token": "wrong"} + assert await adapter.verify_signature(headers, b"{}") is False + + async def test_token_match_without_sign(self): + """配置 token 且 Token 头匹配（无 Sign 头）返回 True。""" + adapter = DingTalkMessageAdapter( + app_key="k", app_secret="s", robot_code="r", token="abc" + ) + headers = {"Token": "abc"} + assert await adapter.verify_signature(headers, b"{}") is True + + async def test_token_none_skips_token_check(self): + """token=None 时无需 Token 头，仅凭签名放行。""" + app_secret = "secret123" + adapter = DingTalkMessageAdapter( + app_key="k", app_secret=app_secret, robot_code="r", token=None + ) + ts_ms = int(time.time() * 1000) + headers = {"Sign": _sign(app_secret, ts_ms), "Timestamp": str(ts_ms)} + assert await adapter.verify_signature(headers, b"{}") is True + + +# --------------------------------------------------------------------------- +# 消息解析 +# --------------------------------------------------------------------------- + + +class TestMessageParsing: + """文本消息解析。""" + + async def test_text_message_parsing(self): + """文本事件解析为 IncomingMessage。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + body = json.dumps(_make_body(text="hello world")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert isinstance(msg, IncomingMessage) + assert msg.channel == ChannelType.DINGTALK + assert msg.content == "hello world" + assert msg.chat_id == "cid1" + assert msg.user_id == "u1" + assert msg.platform_message_id == "m1" + assert msg.timestamp == "1700000000000" + + async def test_mention_stripping(self): + """@ 机器人前缀被剥离。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + body = json.dumps(_make_body(text="@robotName hello")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.content == "hello" + + async def test_unsupported_message_type(self): + """非 text 类型返回 unsupported 占位内容。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + body = json.dumps(_make_body(msgtype="image")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.content.startswith("[unsupported message type: image]") + + async def test_senderid_fallback(self): + """缺少 senderStaffId 时回退到 senderId。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + body = json.dumps( + _make_body(sender_staff_id=None, sender_id="fallback_user") + ).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.user_id == "fallback_user" + + async def test_staffid_fallback(self): + """缺少 senderStaffId 与 senderId 时回退到 staffId。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + body = json.dumps( + _make_body(sender_staff_id=None, sender_id=None, staff_id="staff_user") + ).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.user_id == "staff_user" + + +# --------------------------------------------------------------------------- +# send_message +# --------------------------------------------------------------------------- + + +class TestSendMessage: + """send_message 行为。""" + + async def test_send_message_success(self): + """HTTP 200 返回 True，且 send 调用携带正确的 access token 头。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + mock_token = MagicMock() + mock_token.status_code = 200 + mock_token.json.return_value = {"accessToken": "tok_123"} + + mock_send = MagicMock() + mock_send.status_code = 200 + mock_send.json.return_value = {"processQueryKey": "x"} + + mock_client = AsyncMock() + mock_client.post = AsyncMock(side_effect=[mock_token, mock_send]) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.DINGTALK, chat_id="c1", content="hi") + assert await adapter.send_message(out) is True + + send_call = mock_client.post.call_args_list[1] + assert "robot/oToMessages/batchSend" in send_call.args[0] + assert send_call.kwargs["headers"]["x-acs-dingtalk-access-token"] == "tok_123" + assert send_call.kwargs["json"]["robotCode"] == "r" + + async def test_send_message_failure(self): + """send 返回非 200 返回 False。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + mock_token = MagicMock() + mock_token.status_code = 200 + mock_token.json.return_value = {"accessToken": "tok_x"} + + mock_send = MagicMock() + mock_send.status_code = 400 + mock_send.text = "invalid request" + + mock_client = AsyncMock() + mock_client.post = AsyncMock(side_effect=[mock_token, mock_send]) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.DINGTALK, chat_id="c1", content="hi") + assert await adapter.send_message(out) is False + + async def test_send_message_token_fetch_failure(self): + """获取 accessToken 失败返回 False。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + mock_token = MagicMock() + mock_token.status_code = 401 + mock_token.text = "invalid" + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_token) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.DINGTALK, chat_id="c1", content="hi") + assert await adapter.send_message(out) is False + + async def test_access_token_caching(self): + """同 TTL 内的两次 send_message 只拉取一次 accessToken。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + mock_token = MagicMock() + mock_token.status_code = 200 + mock_token.json.return_value = {"accessToken": "cached_tok"} + + mock_send = MagicMock() + mock_send.status_code = 200 + mock_send.json.return_value = {"processQueryKey": "x"} + + mock_client = AsyncMock() + # 第一次：token + send；第二次：仅 send（token 走缓存） + mock_client.post = AsyncMock(side_effect=[mock_token, mock_send, mock_send]) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.DINGTALK, chat_id="c1", content="hi") + await adapter.send_message(out) + await adapter.send_message(out) + + # 仅 1 次 token 调用 + 2 次 send 调用 = 3 次（未缓存会是 4 次） + assert mock_client.post.call_count == 3 + first_url = mock_client.post.call_args_list[0].args[0] + assert "accessToken" in first_url + + +# --------------------------------------------------------------------------- +# 资源释放 +# --------------------------------------------------------------------------- + + +class TestClose: + """资源释放。""" + + async def test_close_no_client_is_noop(self): + """未创建 httpx 客户端时 close 不抛异常。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + await adapter.close() + + async def test_close_resets_client(self): + """close 后客户端引用清空。""" + adapter = DingTalkMessageAdapter(app_key="k", app_secret="s", robot_code="r") + adapter._get_client() + assert adapter._client is not None + await adapter.close() + assert adapter._client is None diff --git a/tests/unit/channels/test_feishu.py b/tests/unit/channels/test_feishu.py index 1bdc193..d29518e 100644 --- a/tests/unit/channels/test_feishu.py +++ b/tests/unit/channels/test_feishu.py @@ -574,9 +574,9 @@ class TestWebhookErrors: ) assert resp.status_code == 404 - def test_non_feishu_channel_returns_400(self, webhook_client): - """POST 到非飞书渠道 webhook 返回 400。""" - # 注册一个钉钉渠道 + def test_dingtalk_channel_without_secrets_returns_500(self, webhook_client): + """钉钉渠道未配置凭证时 webhook 返回 500（U12 后钉钉受支持）。""" + # 注册一个钉钉渠道（不配置凭证） webhook_client.post( "/api/v1/channels", json={ @@ -590,7 +590,7 @@ class TestWebhookErrors: content=b"{}", headers={"Content-Type": "application/json"}, ) - assert resp.status_code == 400 + assert resp.status_code == 500 class TestWebhookImmediateResponse: diff --git a/tests/unit/channels/test_slack.py b/tests/unit/channels/test_slack.py new file mode 100644 index 0000000..98a19f8 --- /dev/null +++ b/tests/unit/channels/test_slack.py @@ -0,0 +1,320 @@ +"""Slack IM 适配器单元测试 (U12)。 + +覆盖场景： +- 签名校验（有效/无效/过期） +- URL 验证（challenge / token 不匹配） +- Events API 消息解析 +- Slash Command 解析 +- <@U12345> 提及剥离 +- 不支持事件类型 +- send_message 成功/失败 +- bot_token 构造器存储 +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import time +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from urllib.parse import urlencode + +import pytest + +from agentkit.channels.base import ChannelType, IncomingMessage, OutgoingMessage +from agentkit.channels.slack import ( + SlackMessageAdapter, + SignatureVerificationError, + URLVerificationChallenge, +) + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _sign(signing_secret: str, timestamp: int, body: bytes) -> str: + """构造 Slack X-Slack-Signature 头值：v0= + hmac-sha256(signing_secret, "v0:{ts}:{body}")。""" + base = f"v0:{timestamp}:{body.decode('utf-8')}" + digest = hmac.new( + signing_secret.encode("utf-8"), + base.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"v0={digest}" + + +def _make_event_body( + *, + text: str = "hello", + event_type: str = "message", + event_id: str = "evt_001", + user: str = "U123", + channel: str = "C456", + ts: str = "1700000000.000123", +) -> dict[str, Any]: + """构造 Slack Events API 事件 JSON 体。""" + return { + "event_id": event_id, + "type": "event_callback", + "event": { + "type": event_type, + "user": user, + "channel": channel, + "text": text, + "ts": ts, + }, + } + + +# --------------------------------------------------------------------------- +# 签名校验 +# --------------------------------------------------------------------------- + + +class TestSignatureVerification: + """Slack 签名校验。""" + + async def test_valid_signature(self): + """正确 v0 签名返回 True。""" + signing_secret = "sh_secret" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret=signing_secret) + body = json.dumps(_make_event_body()).encode("utf-8") + ts = int(time.time()) + headers = { + "X-Slack-Signature": _sign(signing_secret, ts, body), + "X-Slack-Request-Timestamp": str(ts), + } + assert await adapter.verify_signature(headers, body) is True + + async def test_invalid_signature(self): + """篡改签名返回 False。""" + signing_secret = "sh_secret" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret=signing_secret) + body = json.dumps(_make_event_body()).encode("utf-8") + ts = int(time.time()) + headers = {"X-Slack-Signature": "v0=tampered", "X-Slack-Request-Timestamp": str(ts)} + assert await adapter.verify_signature(headers, body) is False + + async def test_expired_timestamp(self): + """时间戳超过 5 分钟返回 False。""" + signing_secret = "sh_secret" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret=signing_secret) + body = json.dumps(_make_event_body()).encode("utf-8") + old_ts = int(time.time()) - 600 + headers = { + "X-Slack-Signature": _sign(signing_secret, old_ts, body), + "X-Slack-Request-Timestamp": str(old_ts), + } + assert await adapter.verify_signature(headers, body) is False + + async def test_missing_signature_headers(self): + """缺签名头返回 False。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + assert await adapter.verify_signature({}, b"{}") is False + + +# --------------------------------------------------------------------------- +# URL 验证 +# --------------------------------------------------------------------------- + + +class TestURLVerification: + """Slack URL 验证流程。""" + + async def test_url_verification_raises_challenge(self): + """url_verification 事件抛 URLVerificationChallenge。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + body = json.dumps( + {"type": "url_verification", "challenge": "verify_abc", "token": "t"} + ).encode("utf-8") + with pytest.raises(URLVerificationChallenge) as exc_info: + await adapter.receive_message({}, body) + assert exc_info.value.challenge == "verify_abc" + + async def test_url_verification_token_mismatch_raises(self): + """verification_token 不匹配抛 SignatureVerificationError。""" + adapter = SlackMessageAdapter( + bot_token="xoxb-x", + signing_secret="s", + verification_token="right_token", + ) + body = json.dumps( + {"type": "url_verification", "challenge": "abc", "token": "wrong_token"} + ).encode("utf-8") + with pytest.raises(SignatureVerificationError): + await adapter.receive_message({}, body) + + async def test_url_verification_token_match_passes(self): + """verification_token 匹配时正常抛出 challenge。""" + adapter = SlackMessageAdapter( + bot_token="xoxb-x", + signing_secret="s", + verification_token="right_token", + ) + body = json.dumps( + {"type": "url_verification", "challenge": "ok123", "token": "right_token"} + ).encode("utf-8") + with pytest.raises(URLVerificationChallenge) as exc_info: + await adapter.receive_message({}, body) + assert exc_info.value.challenge == "ok123" + + +# --------------------------------------------------------------------------- +# 消息解析 +# --------------------------------------------------------------------------- + + +class TestEventsAPIParsing: + """Events API 消息解析。""" + + async def test_event_message_parsing(self): + """Events API 消息解析为 IncomingMessage。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + body = json.dumps(_make_event_body(text="hello slack")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert isinstance(msg, IncomingMessage) + assert msg.channel == ChannelType.SLACK + assert msg.content == "hello slack" + assert msg.platform_message_id == "evt_001" + assert msg.user_id == "U123" + assert msg.chat_id == "C456" + assert msg.timestamp == "1700000000.000123" + + async def test_mention_stripping(self): + """<@U12345> 提及标记被剥离。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + body = json.dumps(_make_event_body(text="<@U12345> hello there")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.content == "hello there" + + async def test_mention_with_name_stripped(self): + """<@U12345|name> 形式的提及标记也被剥离。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + body = json.dumps(_make_event_body(text="<@U999|bob> hi")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.content == "hi" + + async def test_unsupported_event_type(self): + """非 message/app_mention 事件返回 unsupported 占位内容。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + body = json.dumps(_make_event_body(event_type="reaction_added")).encode("utf-8") + msg = await adapter.receive_message({}, body) + assert msg.content.startswith("[unsupported event type: reaction_added]") + + +class TestSlashCommandParsing: + """Slash Command 解析。""" + + async def test_slash_command_parsing(self): + """form-encoded slash command 解析为 IncomingMessage。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + form = urlencode( + { + "text": "echo me", + "user_id": "U_slash", + "channel_id": "C_slash", + "command": "/echo", + "response_url": "https://hooks.slack.com/x", + } + ) + body = form.encode("utf-8") + msg = await adapter.receive_message( + {"Content-Type": "application/x-www-form-urlencoded"}, body + ) + assert msg.channel == ChannelType.SLACK + assert msg.content == "echo me" + assert msg.user_id == "U_slash" + assert msg.chat_id == "C_slash" + assert msg.platform_message_id.startswith("slash-") + assert msg.raw_event["command"] == "/echo" + + +# --------------------------------------------------------------------------- +# send_message +# --------------------------------------------------------------------------- + + +class TestSendMessage: + """send_message 行为。""" + + async def test_send_message_success(self): + """HTTP 200 + ok=true 返回 True。""" + adapter = SlackMessageAdapter(bot_token="xoxb-123", signing_secret="s") + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"ok": True, "channel": "C1", "ts": "1.0"} + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_resp) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.SLACK, chat_id="C1", content="hi") + assert await adapter.send_message(out) is True + + call = mock_client.post.call_args + assert "chat.postMessage" in call.args[0] + assert call.kwargs["headers"]["Authorization"] == "Bearer xoxb-123" + assert call.kwargs["json"]["channel"] == "C1" + + async def test_send_message_failure(self): + """HTTP 200 但 ok=false 返回 False。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"ok": False, "error": "channel_not_found"} + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_resp) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.SLACK, chat_id="C1", content="hi") + assert await adapter.send_message(out) is False + + async def test_send_message_http_error(self): + """非 200 HTTP 状态返回 False。""" + adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s") + mock_resp = MagicMock() + mock_resp.status_code = 500 + mock_resp.text = "server error" + + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_resp) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.SLACK, chat_id="C1", content="hi") + assert await adapter.send_message(out) is False + + +# --------------------------------------------------------------------------- +# 构造器 / 资源释放 +# --------------------------------------------------------------------------- + + +class TestConstructor: + """构造器与资源释放。""" + + def test_bot_token_stored(self): + """bot_token 由构造器存储。""" + adapter = SlackMessageAdapter( + bot_token="xoxb-secret-token", signing_secret="s", verification_token="vt" + ) + assert adapter.bot_token == "xoxb-secret-token" + assert adapter.signing_secret == "s" + assert adapter.verification_token == "vt" + + async def test_close_no_client_is_noop(self): + """未创建 httpx 客户端时 close 不抛异常。""" + adapter = SlackMessageAdapter(bot_token="x", signing_secret="s") + await adapter.close() + + async def test_close_resets_client(self): + """close 后客户端引用清空。""" + adapter = SlackMessageAdapter(bot_token="x", signing_secret="s") + adapter._get_client() + assert adapter._client is not None + await adapter.close() + assert adapter._client is None diff --git a/tests/unit/channels/test_wecom.py b/tests/unit/channels/test_wecom.py new file mode 100644 index 0000000..d0824c8 --- /dev/null +++ b/tests/unit/channels/test_wecom.py @@ -0,0 +1,323 @@ +"""企业微信 IM 适配器单元测试 (U12)。 + +覆盖场景： +- EncodingAESKey 解码（43 字符 → 32 字节） +- 签名校验（有效/无效） +- AES-256-CBC 加解密往返 +- URL 验证流程（XML 响应） +- 消息解析（FromUserName/MsgId/Content） +- send_message 成功/失败 +- access_token 缓存 +- app_id 不匹配 → 抛异常 +- chat_id 复合键 +- 群聊消息 ":" 前缀剥离 +""" + +from __future__ import annotations + +import base64 +import hashlib +from unittest.mock import AsyncMock, MagicMock +from xml.etree import ElementTree as ET + +import pytest + +from agentkit.channels.base import ChannelType, IncomingMessage, OutgoingMessage +from agentkit.channels.wecom import WeComError, WeComMessageAdapter, WeComURLVerification + + +# --------------------------------------------------------------------------- +# 辅助函数 +# --------------------------------------------------------------------------- + + +def _make_aes_key() -> str: + """生成合法的 43 字符 EncodingAESKey（解码为 32 字节）。""" + return base64.b64encode(b"\x11" * 32).decode().rstrip("=") + + +def _make_adapter( + *, + corp_id: str = "corp_test", + agent_id: int = 1000001, + corp_secret: str = "secret_test", + token: str = "token_test", + encoding_aes_key: str | None = None, +) -> WeComMessageAdapter: + """构造测试用企微适配器。""" + return WeComMessageAdapter( + corp_id=corp_id, + agent_id=agent_id, + corp_secret=corp_secret, + token=token, + encoding_aes_key=encoding_aes_key or _make_aes_key(), + ) + + +def _build_inner_xml(fields: dict[str, str]) -> str: + """由字段字典构造内部 XML 字符串。""" + parts = [""] + for tag, value in fields.items(): + parts.append(f"<{tag}>{value}") + parts.append("") + return "".join(parts) + + +def _build_body(adapter: WeComMessageAdapter, inner_xml: str) -> bytes: + """加密内部 XML 并包装为外层 webhook body。""" + encrypt = adapter._encrypt(inner_xml) + return ( + f"{adapter.corp_id}" + f"{encrypt}" + ).encode("utf-8") + + +def _signature(token: str, timestamp: str, nonce: str, encrypt: str) -> str: + """计算企微签名：sha1(sorted([token, ts, nonce, encrypt]).join(""))。""" + parts = sorted([token, timestamp, nonce, encrypt]) + return hashlib.sha1("".join(parts).encode("utf-8")).hexdigest() + + +# --------------------------------------------------------------------------- +# AES 密钥 / 加解密 +# --------------------------------------------------------------------------- + + +class TestAESKey: + """EncodingAESKey 解码。""" + + def test_decode_aes_key_to_32_bytes(self): + """43 字符 EncodingAESKey 解码为 32 字节。""" + adapter = _make_adapter(encoding_aes_key=_make_aes_key()) + key = adapter._decode_aes_key() + assert len(key) == 32 + + def test_encrypt_decrypt_roundtrip(self): + """加密后解密能恢复原文。""" + adapter = _make_adapter() + encrypted = adapter._encrypt("hello 企微") + assert adapter._decrypt(encrypted) == "hello 企微" + + +class TestSignatureVerification: + """签名校验。""" + + async def test_valid_signature(self): + """正确 msg_signature 返回 True。""" + adapter = _make_adapter() + inner_xml = _build_inner_xml({"MsgId": "m1", "Content": "hi"}) + body = _build_body(adapter, inner_xml) + encrypt = adapter._extract_encrypt(body) or "" + ts, nonce = "1609459200", "n1" + sig = _signature(adapter.token, ts, nonce, encrypt) + headers = {"msg_signature": sig, "timestamp": ts, "nonce": nonce} + assert await adapter.verify_signature(headers, body) is True + + async def test_invalid_signature(self): + """篡改 msg_signature 返回 False。""" + adapter = _make_adapter() + inner_xml = _build_inner_xml({"MsgId": "m1", "Content": "hi"}) + body = _build_body(adapter, inner_xml) + ts, nonce = "1609459200", "n1" + headers = {"msg_signature": "tampered", "timestamp": ts, "nonce": nonce} + assert await adapter.verify_signature(headers, body) is False + + async def test_missing_query_params(self): + """缺少 msg_signature/timestamp/nonce 返回 False。""" + adapter = _make_adapter() + body = b"x" + assert await adapter.verify_signature({}, body) is False + + +# --------------------------------------------------------------------------- +# URL 验证流程 +# --------------------------------------------------------------------------- + + +class TestURLVerification: + """企微 URL 验证流程。""" + + async def test_url_verification_raises_with_xml(self): + """含 EchoStr 的加密事件抛 WeComURLVerification，响应 XML 可解密回 echo。""" + adapter = _make_adapter() + inner_xml = _build_inner_xml({"EchoStr": "echo123"}) + body = _build_body(adapter, inner_xml) + ts, nonce = "1609459200", "n1" + headers = {"timestamp": ts, "nonce": nonce} + + with pytest.raises(WeComURLVerification) as exc_info: + await adapter.receive_message(headers, body) + + # 验证响应 XML 可解密回 echo 值 + response_xml = exc_info.value.response_xml + root = ET.fromstring(response_xml) + encrypted_echo = root.find("Encrypt").text # type: ignore[union-attr] + assert adapter._decrypt(encrypted_echo) == "echo123" + + async def test_url_verification_missing_encrypt_raises(self): + """body 缺少 Encrypt 字段抛 WeComError。""" + adapter = _make_adapter() + body = b"x" + with pytest.raises(WeComError): + await adapter.receive_message({}, body) + + +# --------------------------------------------------------------------------- +# 消息解析 +# --------------------------------------------------------------------------- + + +class TestMessageParsing: + """消息解析。""" + + async def test_receive_message_fields(self): + """解析普通消息提取 FromUserName/MsgId/Content。""" + adapter = _make_adapter() + inner_xml = _build_inner_xml( + { + "FromUserName": "user1", + "MsgId": "msg_001", + "Content": "hello wecom", + "CreateTime": "1609459200", + "MsgType": "text", + } + ) + body = _build_body(adapter, inner_xml) + msg = await adapter.receive_message({}, body) + assert isinstance(msg, IncomingMessage) + assert msg.channel == ChannelType.WECOM + assert msg.user_id == "user1" + assert msg.platform_message_id == "msg_001" + assert msg.content == "hello wecom" + assert msg.timestamp == "1609459200" + + async def test_chat_id_composition(self): + """chat_id 由 corp_id + from_user 复合而成。""" + adapter = _make_adapter(corp_id="my_corp") + inner_xml = _build_inner_xml({"FromUserName": "userA", "MsgId": "m1", "Content": "hi"}) + body = _build_body(adapter, inner_xml) + msg = await adapter.receive_message({}, body) + assert msg.chat_id == "my_corp:userA" + + async def test_chat_room_mention_stripping(self): + """群聊消息内容以 ":" 开头时被剥离。""" + adapter = _make_adapter() + inner_xml = _build_inner_xml( + {"FromUserName": "u1", "MsgId": "m1", "Content": ":hello room"} + ) + body = _build_body(adapter, inner_xml) + msg = await adapter.receive_message({}, body) + assert msg.content == "hello room" + + +class TestAppIDMismatch: + """app_id 校验。""" + + async def test_app_id_mismatch_raises(self): + """解密后 app_id 不匹配 corp_id 抛 WeComError。""" + adapter = _make_adapter(corp_id="corp_a") + # 用不同 corp_id 的适配器加密，AES 密钥相同 + other = _make_adapter(corp_id="corp_b") + inner_xml = _build_inner_xml({"FromUserName": "u1", "MsgId": "m1", "Content": "hi"}) + body = _build_body(other, inner_xml) + + with pytest.raises(WeComError): + await adapter.receive_message({}, body) + + +# --------------------------------------------------------------------------- +# send_message +# --------------------------------------------------------------------------- + + +class TestSendMessage: + """send_message 行为。""" + + async def test_send_message_success(self): + """HTTP 200 + errcode=0 返回 True。""" + adapter = _make_adapter() + mock_token = MagicMock() + mock_token.status_code = 200 + mock_token.json.return_value = {"errcode": 0, "access_token": "tok_123"} + + mock_send = MagicMock() + mock_send.status_code = 200 + mock_send.json.return_value = {"errcode": 0, "errmsg": "ok"} + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_token) + mock_client.post = AsyncMock(return_value=mock_send) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.WECOM, chat_id="c1", content="hi") + assert await adapter.send_message(out) is True + + send_call = mock_client.post.call_args + assert "message/send" in send_call.args[0] + assert send_call.kwargs["params"]["access_token"] == "tok_123" + assert send_call.kwargs["json"]["agentid"] == 1000001 + + async def test_send_message_business_failure(self): + """HTTP 200 但 errcode != 0 返回 False。""" + adapter = _make_adapter() + mock_token = MagicMock() + mock_token.status_code = 200 + mock_token.json.return_value = {"errcode": 0, "access_token": "tok_x"} + + mock_send = MagicMock() + mock_send.status_code = 200 + mock_send.json.return_value = {"errcode": 40014, "errmsg": "invalid token"} + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_token) + mock_client.post = AsyncMock(return_value=mock_send) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.WECOM, chat_id="c1", content="hi") + assert await adapter.send_message(out) is False + + async def test_access_token_caching(self): + """同 TTL 内两次 send_message 只拉取一次 access_token。""" + adapter = _make_adapter() + mock_token = MagicMock() + mock_token.status_code = 200 + mock_token.json.return_value = {"errcode": 0, "access_token": "cached_tok"} + + mock_send = MagicMock() + mock_send.status_code = 200 + mock_send.json.return_value = {"errcode": 0} + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_token) + mock_client.post = AsyncMock(return_value=mock_send) + adapter._client = mock_client + + out = OutgoingMessage(channel=ChannelType.WECOM, chat_id="c1", content="hi") + await adapter.send_message(out) + await adapter.send_message(out) + + # 仅 1 次 token（GET）+ 2 次 send（POST） + assert mock_client.get.call_count == 1 + assert mock_client.post.call_count == 2 + + +# --------------------------------------------------------------------------- +# 资源释放 +# --------------------------------------------------------------------------- + + +class TestClose: + """资源释放。""" + + async def test_close_no_client_is_noop(self): + """未创建 httpx 客户端时 close 不抛异常。""" + adapter = _make_adapter() + await adapter.close() + + async def test_close_resets_client(self): + """close 后客户端引用清空。""" + adapter = _make_adapter() + adapter._get_client() + assert adapter._client is not None + await adapter.close() + assert adapter._client is None