#969 微信Bot媒体内联发送 - iLink CDN AES加密上传

参考微信官方openclaw-weixin源码实现媒体内联发送链路：下载媒体→getuploadurl→AES-128-ECB加密→上传微信CDN→内联发送(type=2/5)。新增im_ilink_media.py模块。失败降级文本链接兜底。修复pg18/casdoor容器未启动问题。

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(中国大陆服务器)直连常被墙/超时。
+    优先走 iLink 官方 CDN 加密上传链路，实现图片/视频真正内联显示；
-    解决：改写为平台中转地址(经香港代理可达 AGNES)。
+    失败时降级为纯文本链接（点击访问），确保会话不中断。
-
-    2026-07-31 兼容修复：iLink 对 type=2/5 的媒体 item 返回 ret=-2 invalid arguments，
-    且旧代码把 ret=-2 误判为 token 失效，导致连接假性断开。先以纯文本链接形式发送，
-    确保用户可点击访问且不会破坏会话；待拿到 iLink 媒体协议文档后再恢复内联 item。
     """
     relay_url = _platform_relay_url(url)
+    # 尝试内联发送（iLink CDN 加密上传）
     try:
-        with open("/tmp/wechat_send_debug.log", "a", encoding="utf-8") as _f:
+        from app.api.im_ilink_media import (
-            _f.write(_beijing_now().isoformat() + " MEDIA_URL_LEN=%s type=%s\n" % (len(relay_url), media_type))
+            MEDIA_TYPE_IMAGE, MEDIA_TYPE_VIDEO,
-    except Exception:
+            download_media_bytes, upload_media_to_wechat_cdn,
-        pass
+            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:

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"],
+        },
+    }