当前位置：首页 > news >正文

Python逆向实战：手把手教你破解某吧私信的WebSocket+Protobuf加密（附完整代码）

news 2026/5/12 12:58:38

Python逆向工程实战：WebSocket与Protobuf加密协议解析

在当今互联网应用中，实时通信功能已成为标配，而WebSocket协议因其全双工通信特性被广泛采用。当这类通信叠加了Protobuf序列化和AES/RSA加密层时，协议分析就变得极具挑战性。本文将带您深入一个典型商业级应用的通信协议逆向过程，从抓包分析到完整实现，掌握现代网络协议逆向的核心方法论。

1. 逆向工程基础准备

逆向网络协议首先需要搭建完整的分析环境。不同于常规开发，逆向工程更强调工具的灵活组合与多角度验证。

必备工具链配置：

# Wireshark抓包分析 sudo apt install wireshark # Protobuf编译器 sudo apt install protobuf-compiler # Python逆向辅助工具 pip install frida objection r2pipe

关键设备建议使用安卓模拟器（如Genymotion）配合真机进行对比测试。模拟器便于动态调试，而真机可获取更真实的网络行为。在抓包环节，Wireshark需要配置SSL密钥日志才能解密HTTPS流量：

提示：设置SSLKEYLOGFILE环境变量后，浏览器和部分应用会将TLS会话密钥导出，供Wireshark解密加密流量

逆向工程通常遵循"由外而内"的分析路径：

网络层抓包确定通信模式
应用层分析协议格式
运行时动态验证猜测
代码级静态分析确认

2. WebSocket协议深度解析

目标应用采用WebSocket作为通信载体，但实现了自定义的二进制封包格式。通过Wireshark捕获的原始数据包显示，每个帧都包含9字节头部：

字节位置	长度	含义
0	1	标志位（加密/压缩/扩展数据）
1-4	4	命令码（大端序）
5-8	4	日志ID（请求响应关联）

标志位字节的各bit具有特定含义：

Bit7(0x80): AES加密标志
Bit6(0x40): Gzip压缩标志
Bit3(0x08): 扩展数据存在标志

Python封包实现：

def build_ws_packet(cmd: int, payload: bytes, encrypt=False, compress=False) -> bytes: flags = 0 if encrypt: flags |= 0x80 if compress: flags |= 0x40 header = struct.pack('>BI', flags, cmd) log_id = random.randint(0, 0xFFFFFFFF) header += struct.pack('>I', log_id) return header + payload

实际通信中观察到的典型流程：

建立WebSocket连接（HTTP 101切换协议）
发送初始化包（cmd=1001）
接收服务端确认
开始业务通信（如cmd=205001对应私信）

3. Protobuf消息逆向实战

目标应用使用Protobuf作为数据序列化方案，这需要从二进制数据反推.proto定义。通过拦截初始化阶段的数据包，发现可解析的Protobuf结构：

关键字段分析：

client_version: 客户端版本标识
device_id: 设备唯一标识
secret_key: RSA加密的临时密钥
bduss: 用户会话凭证

通过JADX反编译APK，定位到关键类UpdateClientInfoReqIdl，从中提取出原始.proto定义：

message DataReq { optional string client_version = 1; optional string device_id = 2; optional bytes secret_key = 3; optional string bduss = 4; // ...其他字段省略 }

Python解析实现：

from google.protobuf import descriptor_pool from google.protobuf.message_factory import GetMessageClass # 动态创建Protobuf消息类 pool = descriptor_pool.Default() factory = MessageFactory(pool) UpdateClientInfoReq = factory.GetPrototype( pool.FindMessageTypeByName('your.package.DataReq')) # 解析二进制数据 req = UpdateClientInfoReq() req.ParseFromString(raw_data[9:]) # 跳过9字节头部 print(req.client_version)

逆向Protobuf时常见难点：

字段编号不连续可能暗示不同版本协议
某些字段可能是废弃或实验性功能
oneof结构会增加解析复杂度

4. 混合加密体系破解

目标应用采用典型的非对称加密协商密钥+对称加密传输数据的方案，具体实现为：

密钥协商阶段：
- 客户端生成36字节随机字符串eo
- 使用SHA1派生AES密钥：key = SHA1(eo + salt)
- 用硬编码RSA公钥加密eo得到secret_key
- 通过初始化消息发送secret_key
数据传输阶段：
- 服务端用RSA私钥解密获得eo
- 双方用相同方式派生AES密钥
- 后续通信使用AES-CBC模式加密

Python实现关键步骤：

import hashlib from Crypto.Cipher import AES from Crypto.PublicKey import RSA def derive_aes_key(eo: bytes) -> bytes: salt = b'ahI' # 从反编译代码获取 dk = hashlib.pbkdf2_hmac( 'sha1', eo, salt, iterations=5, # 与Java实现一致 dklen=32 # AES-256需要32字节密钥 ) return dk def rsa_encrypt(plaintext: bytes, pub_key: str) -> bytes: key = RSA.import_key(base64.b64decode(pub_key)) cipher = PKCS1_v1_5.new(key) return cipher.encrypt(plaintext)

加密实现中的几个关键细节：

PBKDF2迭代次数需与客户端一致（本例为5次）
AES采用PKCS7填充模式
IV（初始化向量）通常为全零或随消息发送

5. 完整通信框架实现

基于aiohttp构建完整的异步通信框架需要处理以下核心问题：

请求-响应关联：WebSocket是全双工通信，需要维护请求ID到响应Future的映射
超时处理：避免未响应请求积压导致内存泄漏
连接管理：自动重连和状态恢复机制

核心类设计：

class WSClient: def __init__(self): self._req_id = 0 self._pending = weakref.WeakValueDictionary() async def send_request(self, cmd: int, payload: bytes) -> bytes: request_id = self._next_req_id() packet = self._build_packet(cmd, payload, request_id) future = asyncio.Future() self._pending[request_id] = future await self._ws.send_bytes(packet) try: return await asyncio.wait_for(future, timeout=10.0) except asyncio.TimeoutError: del self._pending[request_id] raise def _handle_response(self, packet: bytes): request_id = self._parse_req_id(packet) if request_id in self._pending: self._pending[request_id].set_result(packet) def _next_req_id(self) -> int: self._req_id = (self._req_id + 1) & 0xFFFFFFFF return self._req_id

连接初始化流程：

建立WebSocket连接
生成随机密钥并RSA加密
发送初始化消息（cmd=1001）
验证服务端响应
开始业务通信

async def init_connection(self): self._ws = await aiohttp.ClientSession().ws_connect( 'wss://im.tieba.baidu.com:8000', headers={'User-Agent': 'custom client'} ) # 密钥协商 eo = os.urandom(36) secret_key = rsa_encrypt(eo, PUBLIC_KEY) req = UpdateClientInfoReq( client_version='1.0.0', device_id='test_device', secret_key=secret_key, bduss=self._bduss ) await self.send_request(1001, req.SerializeToString()) resp = await self._wait_response(1001) self._aes_key = derive_aes_key(eo)

6. 私信功能完整实现

在完成基础通信框架后，实现具体业务功能如发送私信需要：

分析私信相关Protobuf结构
确定业务命令码（205001）
处理必要的业务逻辑字段

私信Protobuf定义：

message PrivateMessageReq { optional string to_uid = 1; optional string content = 2; optional int32 msg_type = 3 [default = 1]; optional string client_id = 4; }

Python实现示例：

async def send_pm(self, to_user: str, content: str) -> bool: req = PrivateMessageReq( to_uid=to_user, content=content, client_id=str(uuid.uuid4()) ) payload = req.SerializeToString() encrypted = aes_encrypt(self._aes_key, payload) try: resp = await self.send_request(205001, encrypted) return self._parse_response(resp) except Exception as e: logger.error(f"Send PM failed: {e}") return False

实际测试中发现几个关键点：

消息需要包含唯一的client_id避免重复处理
内容长度超过140字节时需要特殊处理
服务端会校验消息发送频率

7. 逆向工程进阶技巧

完成基础功能逆向后，可以进一步优化实现：

流量混淆：分析客户端如何避免被简单特征匹配
心跳机制：维持长连接的保活策略
错误恢复：网络中断后的状态同步

心跳包实现示例：

async def start_heartbeat(self, interval=30): while True: await asyncio.sleep(interval) try: await self.send_request(1002, b'ping') except Exception: logger.warning("Heartbeat failed, reconnecting...") await self.reconnect()

流量混淆策略：