qihu-bot
51d8df6e6a
CI / gate (push) Has been cancelled
实现CDN媒体下载解密(parse_aes_key支持base64(raw16B)和base64(hex32)两种编码)+extract_inbound_media(图片/语音/文件/视频)。接收循环检测媒体消息下载解密保存到/tmp。图片已保存供后续vision模型接入。
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""
|
||
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
|
||
import re
|
||
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"],
|
||
},
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 接收侧媒体下载解密（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
|