qihuaiflow/backend/python/app/api/im_ilink_media.py

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