即时通讯机器人API协议剖析与Python原生复刻实践

一、为什么要“扒”协议：插件好用，但黑盒难集成

近年来，不少即时通讯平台在开放机器人（Bot）能力时，首选交付形态是“官方插件”：用户扫码即可登录，插件随即代理消息收发。对普通使用者，这确实开箱即用；但对开发者与企业用户，这种形态存在三个现实痛点：

第一，黑盒运行。 插件把全部通信细节封装在内部，开发者既无法观察其与服务端的交互过程，也无法干预其行为策略，出了问题只能重启，无从定位。

第二，集成受限。 插件通常绑定特定宿主运行环境，很难嵌入自有业务系统、服务端任务编排或自动化流水线。当业务方希望“机器人收到的消息直接进入我的事件总线”时，插件形态就无能为力。

第三，可控性弱。 会话保持、断线重连、消息去重、错误恢复等策略均被内置实现锁定，无法按业务诉求定制。

换一个角度看问题：插件本身就是该机器人协议的一个完整参考实现。只要插件以未经混淆、未经打包的源码形态发布，那么协议的全部细节，接口地址、请求头规则、鉴权流程、加密方式，实际上都写在源码里。此时所谓的“逆向”，就退化成一次认真的源码阅读与归纳。本文记录的正是这样一次完整分析：从源码结构入手梳理通信全貌，再用 Python 从零复刻。

二、源码阅读方法论：先画地图，再走调用链

本次分析的样本是一套以源码形态直接发布的插件：数十个文件、带完整类型注解、无混淆、无打包压缩。这种“透明发布”在工程实践中并不多见，分析成本因此大幅降低。拿到源码后，建议按以下顺序推进：

1. 看目录结构，建立功能地图。 目录划分通常与模块职责严格对应：

src/
├── api/        # HTTP 请求层：接口封装、类型定义、会话保护
├── auth/       # 登录：扫码流程、账号凭据持久化
├── messaging/  # 消息：接收、发送、完整处理链路
├── cdn/        # 媒体：加密、上传
└── 入口文件     # 插件生命周期与事件分发

1. 读主入口，确认生命周期。 入口文件回答两个问题：插件何时被唤醒、事件回调如何触发。这决定了协议的“驱动方式”，是被动等推送，还是主动轮询。

2. 啃请求层，拿到接口清单。 所有 HTTP 接口往往集中封装在同一处，相当于协议的目录页，先全部罗列出来。

3. 精读类型定义。 完整的数据结构是理解协议语义最快的途径：哪些字段必填、哪些是枚举、嵌套结构如何组织，一目了然。

4. 逐个击破业务模块。 登录、收发、上传各自独立成章，按状态机去理解每个模块。

5. 交叉验证。 有条件的话让插件真实运行，用通用抓包代理工具对照实际报文验证推断。源码是静态事实，报文是动态事实，两者互证才算闭环。

分析产出应固化成三样东西：一份接口清单、一套请求头构造规则、一组登录与消息收发的状态机图。三样齐备，协议就算“扒”完了。

三、协议全貌：五个核心接口构成最小闭环

梳理下来的结论非常简洁：全部业务接口均为 POST JSON，指向统一基地址，配合一套固定通用请求头即可完成所有交互。

3.1 通用请求头逐字段解析

import base64
import json
import secrets
import urllib.request
BASE = "https://<机器人服务基地址>"  # 以实际分析结果为准
def random_uin() -> str:
    # 随机无符号 32 位整数 → Base64 编码
    return base64.b64encode(str(secrets.randbits(32)).encode()).decode()
def build_headers(bot_token: str, body: bytes) -> dict:
    return {
        "Content-Type": "application/json",
        "AuthorizationType": "<固定鉴权标识>",    # 常量，标识鉴权方案
        "Authorization": f"Bearer {bot_token}",  # 扫码登录后获得
        "X-BOT-UIN": random_uin(),               # 每次请求随机生成
        "Content-Length": str(len(body)),         # 必须与请求体精确一致
    }

几个细节值得展开：

• AuthorizationType 是固定常量，用于向服务端标识“当前使用哪种鉴权方案”。这类字段在多鉴权方案并存的平台中很常见，起到协议路由作用。
• Authorization 采用业界标准的 Bearer 方案，令牌在扫码登录成功后获得，后续所有请求复用。
• 随机请求标识字段颇为玩味：每次请求都随机生成一个无符号 32 位整数再做 Base64 编码。既然是随机的，就不可能承担身份标识职责，更可能是请求去重、风控埋点或负载均衡哈希的输入。分析协议时遇到“看不懂的字段”，先记录行为特征、再推测语义，是稳妥的顺序。
• Content-Length 必须与请求体字节数精确一致。这提示服务端很可能对请求体做了完整性校验。手写 HTTP 客户端时这里最易翻车：若先设头、后改请求体，长度对不上就会被直接拒绝。

3.2 接口清单

能力	示意路径	作用
消息拉取	/bot/getupdates	长轮询阻塞式获取新消息
消息发送	/bot/sendmessage	向指定会话发送文本/媒体消息
上传凭证	/bot/getuploadurl	换取媒体上传的目标地址
配置获取	/bot/getconfig	获取输入状态等附加能力所需凭证
输入状态	/bot/sendtyping	上报“正在输入”提示

（注：路径为示意写法，实际以源码分析结果为准。）

除上述五个业务接口外，登录流程另有两条 GET 请求：获取登录二维码、轮询扫码状态。它们独立于业务路径，构成单独的登录通道。

3.3 为什么恰好是这五个接口

这五个接口恰好构成机器人会话的“最小闭环”：能收（长轮询拉取）、能发（消息发送）、能传媒体（凭证+上传）、能有基础体验（输入状态与配置获取），不多一个冗余接口，不少一个关键环节。

传输层选择同样值得玩味：平台选择了 HTTP 长轮询而非 WebSocket 长连接。长轮询的实现与运维成本更低，服务端可近乎无状态地横向扩容，对防火墙与代理环境也更友好；代价是收发延迟略高、存在空轮询开销。对消息量不大的机器人场景，这是一个工程上完全说得通的取舍。

四、扫码登录：两条 GET 组成的状态机

登录流程可抽象为一个小状态机：发起 → 展示二维码 → 等待扫描 → 等待确认 → 换取令牌。用 Python 标准库即可完整实现：

import json
import time
import urllib.request
def get_json(url: str) -> dict:
    with urllib.request.urlopen(url, timeout=10) as resp:
        return json.loads(resp.read().decode())
def login_by_qrcode() -> str:
    # Step 1：发起登录，获取二维码
    qr = get_json(f"{BASE}/bot/get_bot_qrcode?bot_type=3")
    qrcode_id = qr["qrcode"]
    # 将二维码内容交给任意渲染方式展示，等待用户扫码
    # Step 2：轮询扫码状态，确认后换取令牌
    while True:
        status = get_json(f"{BASE}/bot/get_qrcode_status?qrcode={qrcode_id}")
        if status.get("state") == "confirmed":
            return status["token"]
        time.sleep(2)

代码不到三十行，但有几个工程细节不能省：

• 轮询要有节奏。 示例采用 2 秒间隔，实际应结合接口返回的状态提示动态调整，并在长时间未扫码时主动作废旧二维码重新发起，避免无限等待。
• 令牌要持久化。 源码中存在独立的账号持久化模块，说明令牌并非永久有效：需要落盘保存、重启后恢复，失效时自动触发重新扫码。自研实现应把“令牌有效期管理”独立成模块，而不是散落在业务代码里。
• 并发要有保护。 源码中专门的会话保护模块用于解决并发请求下的令牌竞态：当令牌正在刷新时，其他请求应排队等待，而非带着旧令牌冲出去。

五、消息收发与上下文令牌

5.1 长轮询的正确姿势

长轮询的核心在于“服务端 hold 住请求直到有数据或超时”。客户端实现需注意三点：客户端超时要大于服务端超时（否则请求会被客户端提前掐断）；循环要有异常捕获与退避；拉取之间保持游标连续，避免漏消息或重复处理。

def post_json(path: str, payload: dict, token: str) -> dict:
    body = json.dumps(payload).encode()
    req = urllib.request.Request(
        f"{BASE}{path}", data=body,
        headers=build_headers(token, body), method="POST",
    )
    with urllib.request.urlopen(req, timeout=35) as resp:
        return json.loads(resp.read().decode())
def listen_forever(token: str):
    while True:
        # 服务端超时 25 秒，客户端超时需大于它
        updates = post_json("/bot/getupdates", {"timeout": 25}, token)
        for msg in updates.get("messages", []):
            handle_message(token, msg)
def handle_message(token: str, msg: dict):
    reply = f"已收到：{msg.get('text', '')}"
    post_json("/bot/sendmessage", {
        "chat_id": msg["chat_id"],
        "text": reply,
        "context_token": msg["context_token"],  # 关联会话上下文
    }, token)

5.2 context_token：被低估的关键设计

源码中消息接收模块专门管理一个 context_token：它随收到的消息事件下发，回复该会话时必须原样携带。设计意图是让服务端把“回复”与“触发上下文”精确关联，在多轮对话中尤为关键：用户连续发来多条消息时，服务端可据此判断机器人当前回应的是哪一条，从而避免乱序与答非所问。实现层面，建议为每个会话维护独立的 context_token 状态，收到新事件即更新、发送时取最新值。

六、媒体上传：先取凭证，再传密文

媒体上传采用业界常见的两段式：先调用上传凭证接口换取临时上传地址，再上传数据，最后在消息中引用上传结果。临时凭证模式的好处是权限收敛：上传权限由短时效凭证控制，主令牌不直接暴露在上传链路中。

特别之处在于上传内容并非明文，而是先做了一层 AES-ECB 加密。为什么选 ECB？从实现角度看有现实理由：不需要初始向量、加解密可按块独立进行、实现简单。但必须指出，ECB 是密码学公认的最弱工作模式，相同明文块会加密成相同密文块，对图像等规律性强的数据，密文可能泄露轮廓信息。通用最佳实践是：学习与原理分析场景理解它没有问题；自建系统选型时应优先选择 GCM 这类带完整性认证的模式。用 Python 复刻该环节的思路是：对媒体数据先做 PKCS7 填充、再按块加密，密钥通常由登录流程或配置接口下发，其确切来源可从源码中确认。

七、工程化复刻要点

协议打通只是第一步，要长期稳定运行，还有三件事要做：

1. 分层组织代码。 按“传输层（HTTP 封装与请求头构造）/ 协议层（接口封装）/ 业务层（消息处理链）”三层拆分，与源码插件的模块划分保持同构，便于后续对照升级。
2. 建立错误处理矩阵。 至少覆盖四类情况：网络错误（重试与退避）、鉴权失效（触发重登）、限频（按返回提示降速）、数据异常（跳过并记日志）。
3. 做好可观测性。 长轮询是个“安静”的进程，出问题时往往没有异常抛出，只能靠日志与心跳指标发现。若选择云上部署，可将复刻服务托管于容器平台，配合日志服务留存运行痕迹、监控告警盯住轮询存活状态（这类通用能力在百度智能云等云厂商均有对应的产品类目可选）。

最后是合规边界：插件协议的归属权在平台方。本文所述分析与复刻的价值在于理解通信设计、沉淀方法论，适合学习研究与内部原理验证；任何面向生产的正式集成，都应回归平台官方开放的接口与授权渠道。

八、写在最后

回顾整个过程，方法论比结论更有价值：遇到未混淆源码，先建目录地图，再啃请求层拿接口清单，用类型定义补语义，用抓包互证；产出接口清单、请求头规则与状态机三件套之后，任何此类协议都能用一个周末完成原生复刻。掌握这套方法，下次面对任何“黑盒插件”，你都有了打开它的钥匙。