From 46e9f1e1e675b3dd3b84988c0f6b2cb2a378c074 Mon Sep 17 00:00:00 2001 From: qihu-bot Date: Sat, 15 Aug 2026 17:45:47 +0800 Subject: [PATCH] =?UTF-8?q?#969=20=E5=BE=AE=E4=BF=A1Bot=E5=AA=92=E4=BD=93?= =?UTF-8?q?=E5=86=85=E8=81=94=E5=8F=91=E9=80=81=20-=20iLink=20CDN=20AES?= =?UTF-8?q?=E5=8A=A0=E5=AF=86=E4=B8=8A=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 参考微信官方openclaw-weixin源码实现媒体内联发送链路：下载媒体→getuploadurl→AES-128-ECB加密→上传微信CDN→内联发送(type=2/5)。新增im_ilink_media.py模块。失败降级文本链接兜底。修复pg18/casdoor容器未启动问题。 --- backend/python/app/api/im_bot_user.py | 45 +++-- backend/python/app/api/im_ilink_media.py | 214 +++++++++++++++++++++++ 2 files changed, 248 insertions(+), 11 deletions(-) create mode 100644 backend/python/app/api/im_ilink_media.py diff --git a/backend/python/app/api/im_bot_user.py b/backend/python/app/api/im_bot_user.py index 71f1140..a246682 100644 --- a/backend/python/app/api/im_bot_user.py +++ b/backend/python/app/api/im_bot_user.py @@ -319,21 +319,44 @@ def _platform_relay_url(raw_url: str) -> str: def _wechat_send_media(baseurl: str, bot_token: str, to_user_id: str, context_token: str, media_type: str, url: str): - """发送媒体（图片/视频）。 + """发送媒体（图片/视频）。#969 iLink CDN 内联发送 - 问题：AGNES 媒体 URL 位于境外，iLink(中国大陆服务器)直连常被墙/超时。 - 解决：改写为平台中转地址(经香港代理可达 AGNES)。 - - 2026-07-31 兼容修复：iLink 对 type=2/5 的媒体 item 返回 ret=-2 invalid arguments， - 且旧代码把 ret=-2 误判为 token 失效，导致连接假性断开。先以纯文本链接形式发送， - 确保用户可点击访问且不会破坏会话；待拿到 iLink 媒体协议文档后再恢复内联 item。 + 优先走 iLink 官方 CDN 加密上传链路，实现图片/视频真正内联显示； + 失败时降级为纯文本链接（点击访问），确保会话不中断。 """ relay_url = _platform_relay_url(url) + # 尝试内联发送（iLink CDN 加密上传） try: - with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f: - _f.write(_beijing_now().isoformat() + " MEDIA_URL_LEN=%s type=%s\n" % (len(relay_url), media_type)) - except Exception: - pass + from app.api.im_ilink_media import ( + MEDIA_TYPE_IMAGE, MEDIA_TYPE_VIDEO, + download_media_bytes, upload_media_to_wechat_cdn, + build_image_item, build_video_item, + ) + media_bytes = download_media_bytes(url) + mt = MEDIA_TYPE_IMAGE if media_type == "image" else MEDIA_TYPE_VIDEO + uploaded = upload_media_to_wechat_cdn(baseurl, bot_token, to_user_id, media_bytes, mt) + item = build_image_item(uploaded) if media_type == "image" else build_video_item(uploaded) + ok, ret, _ = _wechat_post_msg(baseurl, bot_token, to_user_id, context_token, [item]) + if ok: + try: + with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " MEDIA_INLINE_OK type=%s size=%d\n" + % (media_type, len(media_bytes))) + except Exception: + pass + return True, ret, {} + try: + with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " MEDIA_INLINE_FAIL ret=%s → fallback text\n" % (ret,)) + except Exception: + pass + except Exception as _e: + try: + with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " MEDIA_INLINE_EXC %s → fallback text\n" % (_e,)) + except Exception: + pass + # 降级：纯文本链接 if media_type == "image": caption = "🖼️ 图片已生成，点击查看：" + relay_url else: diff --git a/backend/python/app/api/im_ilink_media.py b/backend/python/app/api/im_ilink_media.py new file mode 100644 index 0000000..1039697 --- /dev/null +++ b/backend/python/app/api/im_ilink_media.py @@ -0,0 +1,214 @@ +""" +iLink CDN 媒体加解密 —— 微信官方 iLink(ClawBot) 协议媒体内联发送核心模块 +im_ilink_media.py + +参考微信官方 OpenClaw 插件 (@tencent-weixin/openclaw-weixin) 源码实现： +- cdn/aes-ecb.ts: AES-128-ECB + PKCS7 加解密 +- cdn/upload.ts: 六步上传流程 +- messaging/send.ts: image_item / video_item 构造 + +严格照抄官方行为，保证字段名与编码方式与微信服务器一致： +- aeskey = 随机 16 字节 +- getUploadUrl 传 aeskey.hex()（32 字符 hex） +- 加密用 16 字节原始 key +- sendmessage 的 media.aes_key = base64(hex 字符串的 UTF-8 字节) + +媒体类型映射: IMAGE=1, VIDEO=2, FILE=3, VOICE=4 +""" +import base64 +import hashlib +import math +import os +from urllib.parse import quote + +import requests +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives import padding + +# 微信 CDN 域名（图片/语音/视频媒体文件走这里） +WECHAT_CDN_BASE = "https://novac2c.cdn.weixin.qq.com/c2c" + +# 媒体类型枚举（与 iLink UploadMediaType 一致） +MEDIA_TYPE_IMAGE = 1 +MEDIA_TYPE_VIDEO = 2 +MEDIA_TYPE_FILE = 3 +MEDIA_TYPE_VOICE = 4 + + +def encrypt_aes_ecb(plaintext: bytes, key: bytes) -> bytes: + """AES-128-ECB + PKCS7 加密（微信指定，非标准 CBC）。""" + padder = padding.PKCS7(128).padder() + padded = padder.update(plaintext) + padder.finalize() + cipher = Cipher(algorithms.AES(key), modes.ECB()) + enc = cipher.encryptor() + return enc.update(padded) + enc.finalize() + + +def decrypt_aes_ecb(ciphertext: bytes, key: bytes) -> bytes: + """AES-128-ECB + PKCS7 解密。""" + cipher = Cipher(algorithms.AES(key), modes.ECB()) + dec = cipher.decryptor() + padded = dec.update(ciphertext) + dec.finalize() + unpadder = padding.PKCS7(128).unpadder() + return unpadder.update(padded) + unpadder.finalize() + + +def aes_ecb_padded_size(plaintext_size: int) -> int: + """AES-128-ECB PKCS7 填充后的密文大小。""" + return math.ceil((plaintext_size + 1) / 16) * 16 + + +def _wechat_headers(bot_token: str) -> dict: + """iLink 协议请求头（复制 im_bot_user 逻辑，避免循环依赖）。""" + uin = base64.b64encode(str(random_int()).encode("utf-8")).decode("utf-8") + return { + "Content-Type": "application/json", + "Authorization": f"Bearer {bot_token}", + "AuthorizationType": "ilink_bot_token", + "X-WECHAT-UIN": uin, + } + + +def random_int(): + import random + return random.randint(0, 2 ** 32 - 1) + + +def download_media_bytes(url: str) -> bytes: + """下载 AGNES/供应商媒体字节（走 HK 代理，绕过劫持）。 + + 复用 chat_utils._proxy_for_url 的代理选择逻辑；对境外媒体走 HK 代理。 + """ + from app.api.chat_utils import _proxy_for_url + + proxy = _proxy_for_url(url) + proxies = None + if proxy: + proxies = {"http": proxy, "https": proxy} + # trust_env=False 语义等价：requests 不读环境变量代理 + # 显式传 proxies=None 且环境变量存在时 requests 会用环境变量，故需清空 + import os as _os + env_bak = {} + for _k in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"): + if _k in _os.environ: + env_bak[_k] = _os.environ.pop(_k) + try: + resp = requests.get(url, proxies=proxies, timeout=180, allow_redirects=True) + finally: + _os.environ.update(env_bak) + if not resp.ok: + raise RuntimeError(f"media download failed: {resp.status_code} url={url[:120]}") + return resp.content + + +def upload_media_to_wechat_cdn( + baseurl: str, + bot_token: str, + to_user_id: str, + media_bytes: bytes, + media_type: int, +) -> dict: + """六步上传流程：读文件→hash→gen aeskey→getUploadUrl→加密上传CDN→返回引用。 + + 返回 uploaded dict: { filekey, download_param, aeskey_hex, filesize } + """ + rawsize = len(media_bytes) + rawfilemd5 = hashlib.md5(media_bytes).hexdigest() + filesize = aes_ecb_padded_size(rawsize) + filekey = os.urandom(16).hex() # 32 字符 hex + aeskey = os.urandom(16) # 16 字节原始 key + aeskey_hex = aeskey.hex() # 32 字符 hex + + # 1. getUploadUrl + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/getuploadurl", + headers=_wechat_headers(bot_token), + json={ + "filekey": filekey, + "media_type": media_type, + "to_user_id": to_user_id, + "rawsize": rawsize, + "rawfilemd5": rawfilemd5, + "filesize": filesize, + "no_need_thumb": True, + "aeskey": aeskey_hex, + }, + timeout=30, + ) + if not resp.ok: + raise RuntimeError(f"getuploadurl HTTP {resp.status_code}: {resp.text[:200]}") + body = resp.json() if resp.content else {} + upload_param = body.get("upload_param") + upload_full_url = (body.get("upload_full_url") or "").strip() + + if not upload_param and not upload_full_url: + raise RuntimeError(f"getuploadurl returned no upload URL: {str(body)[:200]}") + + # 2. AES-128-ECB 加密 + ciphertext = encrypt_aes_ecb(media_bytes, aeskey) + + # 3. 上传 CDN + if upload_full_url: + cdn_url = upload_full_url + else: + cdn_url = ( + f"{WECHAT_CDN_BASE}/upload" + f"?encrypted_query_param={quote(upload_param, safe='')}" + f"&filekey={quote(filekey, safe='')}" + ) + up_resp = requests.post( + cdn_url, + data=ciphertext, + headers={"Content-Type": "application/octet-stream"}, + timeout=120, + ) + if up_resp.status_code not in (200, 201): + raise RuntimeError(f"CDN upload failed HTTP {up_resp.status_code}: {up_resp.text[:200]}") + download_param = up_resp.headers.get("x-encrypted-param") + if not download_param: + raise RuntimeError("CDN upload response missing x-encrypted-param header") + + return { + "filekey": filekey, + "download_param": download_param, + "aeskey_hex": aeskey_hex, + "filesize": filesize, + } + + +def _aes_key_b64(aeskey_hex: str) -> str: + """media.aes_key 编码：base64(hex 字符串的 UTF-8 字节)。 + + 严格对齐 openclaw: Buffer.from(hexString).toString('base64') + """ + return base64.b64encode(aeskey_hex.encode("utf-8")).decode("utf-8") + + +def build_image_item(uploaded: dict) -> dict: + """构造图片 MessageItem（type=2）。""" + return { + "type": 2, + "image_item": { + "media": { + "encrypt_query_param": uploaded["download_param"], + "aes_key": _aes_key_b64(uploaded["aeskey_hex"]), + "encrypt_type": 1, + }, + "mid_size": uploaded["filesize"], + }, + } + + +def build_video_item(uploaded: dict) -> dict: + """构造视频 MessageItem（type=5）。""" + return { + "type": 5, + "video_item": { + "media": { + "encrypt_query_param": uploaded["download_param"], + "aes_key": _aes_key_b64(uploaded["aeskey_hex"]), + "encrypt_type": 1, + }, + "video_size": uploaded["filesize"], + }, + }