无Proto文件时如何用Python正确序列化发送自定义gRPC请求

gRPC 基于 HTTP/2 传输的报文结构是完全公开的，即便没有 .proto 定义文件，只要能拿到真实通信的二进制报文，就可以手动完成序列化和请求发送。

核心报文规则 #

gRPC 请求的结构固定，不存在黑盒：

• 请求头必须携带 Content-Type: application/grpc+proto、TE: trailers 两个必填标识，请求路径、域名直接复用抓包拿到的内容即可，本次目标路径为 /co.znly.users.services.sessions.SessionsService/SessionCreate，域名为 secret.zenly.rpc.com
• 请求体结构为：1字节压缩标记（无压缩固定为 0x00，gzip压缩为 0x01） + 4字节大端序存储的载荷长度 + Protobuf 序列化后的二进制载荷
• 响应结构和请求体完全一致，解析时跳过前5字节帧头，剩余内容就是Protobuf格式的响应数据
• grpc_requests、grpcurl 工具失效，基本是因为工具会自动生成标准gRPC请求头，和Zenly客户端实际携带的自定义签名、设备标识头不匹配，被服务端校验拦截，和是否有Proto文件无关。

第一步：从抓包数据反推字段对应关系 #

Protobuf 编码本身是「字段Tag+类型+值」的结构，不需要Proto文件也能直接解析二进制内容：

1. 从mitmproxy中导出该接口成功请求的原始二进制报文，跳过前5字节gRPC帧头，得到纯Protobuf二进制载荷
2. 使用 blackboxprotobuf 库直接解码该二进制载荷，就能得到带Tag编号的结构化数据，和你已经拿到的JSON结构一一对应，就能确认每个字段对应的Tag号、字段类型（字符串、嵌套结构体等）

解码示例输出参考：

{
  '1': b'对应手机号字段（string类型）',
  '2': { # 对应device嵌套结构体
      '1': b'4.63.14', # 对应appVersion
      '2': b'ANDROID', # 对应type
      # 其余字段按解码结果逐一和JSON字段映射即可
  },
  '3': b'对应deviceOsUuid字段',
  '4': {} # 对应carrierInformations嵌套结构体
}

第二步：手动序列化Protobuf载荷 #

不需要编写Proto文件，直接用 blackboxprotobuf 按照映射好的Tag结构序列化即可，示例代码如下：

import blackboxprotobuf
import struct

# 注意：以下Tag编号为示例，必须替换为你自己抓包解码得到的真实Tag值
message_type_def = {
    '1': {'type': 'bytes', 'name': 'PhoneNumber'},
    '2': {
        'type': 'message',
        'name': 'device',
        'message_typedef': {
            '1': {'type': 'bytes', 'name': 'appVersion'},
            '2': {'type': 'bytes', 'name': 'type'},
            '3': {'type': 'bytes', 'name': 'osVersion'},
            '4': {'type': 'bytes', 'name': 'model'},
            '5': {'type': 'bytes', 'name': 'acceptLanguages'},
            '6': {'type': 'bytes', 'name': 'coreVersion'},
            '7': {'type': 'bytes', 'name': 'appBundle'}
        }
    },
    '3': {'type': 'bytes', 'name': 'deviceOsUuid'},
    '4': {
        'type': 'message',
        'name': 'carrierInformations',
        'message_typedef': {
            '1': {'type': 'bytes', 'name': 'networkOperatorCode'},
            '2': {'type': 'bytes', 'name': 'networkOperatorName'},
            '3': {'type': 'bytes', 'name': 'networkCountryIso'},
            '4': {'type': 'bytes', 'name': 'simOperatorCode'},
            '5': {'type': 'bytes', 'name': 'simOperatorName'},
            '6': {'type': 'bytes', 'name': 'simCountryIso'}
        }
    }
}

# 填入自定义请求参数
request_data = {
    '1': '你的目标手机号'.encode('utf-8'),
    '2': {
        '1': b'4.63.14',
        '2': b'ANDROID',
        '3': b'12',
        '4': '你的设备model值'.encode('utf-8'),
        '5': b'en-US;q=1.0',
        '6': b'1.96.7',
        '7': b'app.zenly.locator'
    },
    '3': '你的deviceOsUuid值'.encode('utf-8'),
    '4': {
        '1': b'25001',
        '2': b'MTS',
        '3': b'ru',
        '4': b'25001',
        '5': b'MTS RUS',
        '6': b'ru'
    }
}

# 序列化为Protobuf二进制
pb_payload, _ = blackboxprotobuf.encode_message(request_data, message_type_def)
# 拼接gRPC要求的5字节帧头
grpc_request_body = b'\x00' + struct.pack('>I', len(pb_payload)) + pb_payload

第三步：发送HTTP/2请求 #

直接使用支持HTTP/2的HTTP客户端发送请求即可，不要使用官方gRPC客户端，避免自动修改请求头：

import httpx

# 注意：必须把mitmproxy抓到的该接口所有请求头原封不动全部带上，包括x-zenly开头的签名、鉴权类自定义头
request_headers = {
    'Content-Type': 'application/grpc+proto',
    'TE': 'trailers',
    'User-Agent': '抓包拿到的真实UA值'
    # 其余自定义头全部补全
}

with httpx.Client(http2=True, timeout=10) as client:
    resp = client.post(
        url='https://secret.zenly.rpc.com/co.znly.users.services.sessions.SessionsService/SessionCreate',
        headers=request_headers,
        content=grpc_request_body
    )
    # 解析响应
    resp_pb_content = resp.content[5:] # 跳过5字节gRPC帧头
    resp_data, _ = blackboxprotobuf.decode_message(resp_pb_content)
    print(resp_data)

注意事项 #

• 绝对不要猜测字段Tag编号，必须从真实成功的抓包报文中解码获取，Tag错误会导致服务端完全无法解析请求
• 所有抓包可见的请求头必须全量携带，缺失签名、设备校验类头会直接被服务端拦截
• 如果抓包看到请求开启了压缩，帧头第一个字节为0x01，需要先将Protobuf载荷用gzip压缩后再拼接帧头发送

内容的提问来源于stack exchange，提问作者Danila