如何在Python中无需.proto文件解码Protobuf数据？

如果没有原始.proto文件，想要实现类似在线工具的Protobuf解码效果，核心是基于Protobuf的二进制编码规则直接解析字段标签、类型和对应值，无需预先定义消息结构。以下是两种可行方案：

方法一：手动实现解析逻辑（理解底层原理） #

Protobuf每个字段由标签(tag)和值(value)组成：

• 标签格式：(字段号 << 3) | wireType，其中wireType决定值的编码方式
• 值的编码分为varint、固定长度、长度前缀等类型

代码示例 #

import struct

def decode_varint(data, pos):
    """解析varint类型值，返回结果和新的指针位置"""
    result = 0
    shift = 0
    while True:
        byte = data[pos]
        pos += 1
        result |= (byte & 0x7F) << shift
        if not (byte & 0x80):
            break
        shift += 7
    return result, pos

def decode_protobuf_hex(hex_str):
    """解析十六进制格式的Protobuf数据"""
    data = bytes.fromhex(hex_str)
    pos = 0
    decoded = []
    while pos < len(data):
        # 解析标签
        tag, pos = decode_varint(data, pos)
        wire_type = tag & 0x07
        field_number = tag >> 3

        # 根据wire type解析对应值
        if wire_type == 0:  # varint类型
            value, pos = decode_varint(data, pos)
            decoded.append({"field_number": field_number, "wire_type": "varint", "value": value})
        elif wire_type == 1:  # 64位固定长度
            value = struct.unpack("<Q", data[pos:pos+8])[0]
            pos += 8
            decoded.append({"field_number": field_number, "wire_type": "fixed64", "value": value})
        elif wire_type == 2:  # 长度前缀类型（嵌套消息、字符串、字节）
            length, pos = decode_varint(data, pos)
            sub_data = data[pos:pos+length]
            pos += length
            # 递归解析嵌套消息，失败则按字节/字符串处理
            try:
                nested = decode_protobuf_hex(sub_data.hex())
                decoded.append({"field_number": field_number, "wire_type": "length_delimited", "value": nested})
            except:
                try:
                    str_value = sub_data.decode('utf-8')
                    decoded.append({"field_number": field_number, "wire_type": "length_delimited", "value": str_value})
                except:
                    decoded.append({"field_number": field_number, "wire_type": "length_delimited", "value": sub_data.hex()})
        elif wire_type == 5:  # 32位固定长度
            value = struct.unpack("<I", data[pos:pos+4])[0]
            pos += 4
            decoded.append({"field_number": field_number, "wire_type": "fixed32", "value": value})
        else:
            decoded.append({"field_number": field_number, "wire_type": f"unknown({wire_type})", "value": f"skipped, pos: {pos}"})
            break
    return decoded

# 解析你的示例数据
hex_data = "0a06282c0241057a10011805220d080510bea3f493062a03010c1628f1a6f493063002382b4001481482010f383634333233303532343736343839"
result = decode_protobuf_hex(hex_data)
for item in result:
    print(f"字段号: {item['field_number']}, 类型: {item['wire_type']}, 值: {item['value']}")

方法二：使用现成第三方库 #

可以用专门支持无Schema解析的protobuf-decoder库，封装了完整的Protobuf解析逻辑，使用更简洁。

步骤 #

1. 安装库：

pip install protobuf-decoder

2. 解析示例数据：

from protobuf_decoder import decode_hex

hex_data = "0a06282c0241057a10011805220d080510bea3f493062a03010c1628f1a6f493063002382b4001481482010f383634333233303532343736343839"
decoded = decode_hex(hex_data)

# 格式化打印解析结果
def print_decoded(decoded_items, indent=0):
    prefix = "  " * indent
    for item in decoded_items:
        print(f"{prefix}字段号: {item.field_number}, 类型: {item.wire_type.name}, 值: ", end="")
        if item.wire_type == item.wire_type.LENGTH_DELIMITED:
            if isinstance(item.value, list):
                print()
                print_decoded(item.value, indent + 1)
            else:
                try:
                    print(item.value.decode('utf-8'))
                except:
                    print(item.value.hex())
        else:
            print(item.value)

print_decoded(decoded)

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