#971 微信Bot P2 - 接收侧媒体下载解密

实现CDN媒体下载解密(parse_aes_key支持base64(raw16B)和base64(hex32)两种编码)+extract_inbound_media(图片/语音/文件/视频)。接收循环检测媒体消息下载解密保存到/tmp。图片已保存供后续vision模型接入。

backend/python/app/api/im_ilink_media.py

@@ -19,6 +19,7 @@ import base64
 import hashlib
 import math
 import os
+import re
 from urllib.parse import quote
 
 import requests
@@ -212,3 +213,118 @@ def build_video_item(uploaded: dict) -> dict:
             "video_size": uploaded["filesize"],
         },
     }
+
+
+# ============================================================
+# 接收侧媒体下载解密（P2）
+# 参考 openclaw cdn/pic-decrypt.ts + media/media-download.ts
+# ============================================================
+
+def parse_aes_key(aes_key_base64: str) -> bytes:
+    """解析 CDNMedia.aes_key 为原始 16 字节 AES key。
+
+    两种真实编码（openclaw parseAesKey）：
+      - base64(raw 16 bytes)          → 图片 (media.aes_key)
+      - base64(hex 32 chars 字符串)    → 文件/语音/视频
+    第二种需 base64 解码后得到 32 个 ASCII hex 字符，再 hex 解码成 16 字节。
+    """
+    decoded = base64.b64decode(aes_key_base64)
+    if len(decoded) == 16:
+        return decoded
+    if len(decoded) == 32 and re.match(r"^[0-9a-fA-F]{32}$", decoded.decode("ascii", "ignore")):
+        return bytes.fromhex(decoded.decode("ascii"))
+    raise ValueError(
+        f"aes_key 解码后应为 16 字节或 32 字符 hex，实际 {len(decoded)} 字节"
+    )
+
+
+def download_and_decrypt_media(
+    encrypted_query_param: str,
+    aes_key_base64: str,
+    cdn_base_url: str = WECHAT_CDN_BASE,
+    full_url: str = None,
+) -> bytes:
+    """下载并 AES-128-ECB 解密 CDN 媒体文件，返回明文字节。"""
+    key = parse_aes_key(aes_key_base64)
+    if full_url:
+        url = full_url
+    else:
+        url = (
+            f"{cdn_base_url}/download"
+            f"?encrypted_query_param={quote(encrypted_query_param, safe='')}"
+        )
+    resp = requests.get(url, timeout=60)
+    if not resp.ok:
+        raise RuntimeError(f"CDN download failed HTTP {resp.status_code}: {resp.text[:150]}")
+    encrypted = resp.content
+    return decrypt_aes_ecb(encrypted, key)
+
+
+def extract_inbound_media(item: dict) -> tuple:
+    """从入站 MessageItem 提取媒体并下载解密。
+
+    返回 (media_kind, media_bytes)，不支持的类型或失败返回 (None, None)。
+    media_kind ∈ {"image", "voice", "file", "video"}
+    """
+    try:
+        t = item.get("type")
+        if t == 2:  # 图片
+            img = item.get("image_item") or {}
+            media = img.get("media") or {}
+            eqp = media.get("encrypt_query_param") or ""
+            full = media.get("full_url")
+            if not eqp and not full:
+                return None, None
+            # 图片 key 优先用 image_item.aeskey（hex string）转 base64
+            if img.get("aeskey"):
+                aes_key_b64 = base64.b64encode(bytes.fromhex(img["aeskey"])).decode()
+            else:
+                aes_key_b64 = media.get("aes_key") or ""
+            if aes_key_b64:
+                return "image", download_and_decrypt_media(eqp, aes_key_b64, full_url=full)
+            return "image", _download_plain(eqp, full)
+        elif t == 3:  # 语音
+            voice = item.get("voice_item") or {}
+            media = voice.get("media") or {}
+            eqp = media.get("encrypt_query_param") or ""
+            full = media.get("full_url")
+            aes_key_b64 = media.get("aes_key") or ""
+            if not aes_key_b64:
+                return None, None
+            return "voice", download_and_decrypt_media(eqp, aes_key_b64, full_url=full)
+        elif t == 4:  # 文件
+            fitem = item.get("file_item") or {}
+            media = fitem.get("media") or {}
+            eqp = media.get("encrypt_query_param") or ""
+            full = media.get("full_url")
+            aes_key_b64 = media.get("aes_key") or ""
+            if not aes_key_b64:
+                return None, None
+            return "file", download_and_decrypt_media(eqp, aes_key_b64, full_url=full)
+        elif t == 5:  # 视频
+            vitem = item.get("video_item") or {}
+            media = vitem.get("media") or {}
+            eqp = media.get("encrypt_query_param") or ""
+            full = media.get("full_url")
+            aes_key_b64 = media.get("aes_key") or ""
+            if not aes_key_b64:
+                return None, None
+            return "video", download_and_decrypt_media(eqp, aes_key_b64, full_url=full)
+    except Exception:
+        return None, None
+    return None, None
+
+
+def _download_plain(encrypted_query_param: str, full_url: str = None) -> bytes:
+    """下载未加密的 CDN 媒体（无 aes_key 时）。"""
+    if full_url:
+        url = full_url
+    else:
+        url = (
+            f"{WECHAT_CDN_BASE}/download"
+            f"?encrypted_query_param={quote(encrypted_query_param, safe='')}"
+        )
+    resp = requests.get(url, timeout=60)
+    if not resp.ok:
+        raise RuntimeError(f"CDN plain download failed HTTP {resp.status_code}")
+    return resp.content