protobuf编码规则分析

分析protobuf二进制编码#

比如一个最简单的pb：

1
syntax = "proto3";
2

3
message ResourceModify {
4
    int32 type = 1;
5
    int32 amount = 2;
6
}

我们假设数据是：

1
{
2
  "type": 1,
3
  "amount": 26526
4
}

如何得到其对应的二进制编码呢？

tag + value#

可以简单的把pb的二进制编码看作是 tag + value ，首先看tag怎么计算

公式：

1
tag = (field_number << 3) | wire_type

field_num 即是序号，左移三位，然后和wire_type计算或

这个wire_type即代表了该字段的类型，定义如下

因此，第一个字段 type = 1 对应的key就是：

1
1 << 3 | 0 = 0x8

接着再拼接上它的value，最后就是0x0801

msb +小端序编码#

接着看第二字段

1
"amount": 26526

计算tag：2 <<3 |0 = 0x10

对应的value，则是：9ecf01 ，你是不是很好奇怎么来的，为什么不直接是26526的十六进制编码，这是因为pb采用了msb +小端序的规则

msb即最高位置作为符号位信息

我们将其转为对应的二进制，第一位为msb符号位直接去掉，然后按照小端序来拼接

1
9e = 10011110
2
cf = 11001111
3
01 = 00000001
4

5
拼接过程， 先取01 -> 00000001 （注意MSB，第一位直接丢掉
6
然后取 cf -> 1001111(同样第一位被丢掉了)
7
再取 9e -> 0011110
8
拼接起来就成了  1 1001111 0011110
9
用py代码转一下就有了
10
int('1 1001111 0011110'.replace(' ', ''), 2) # 26526

vibe coding一下

1
def int_to_varint(n):
2
    """
3
    将一个大于 255 的整数转换为 Protobuf varint 编码。
4

5
    :param n: 需要转换的整数。
6
    :return: varint 编码后的字节数组（以 list 的形式返回）。
7
    """
8
    result = []
9
    while n > 0x7F:  # 0x7F = 127
10
        # 取低 7 位，并将高位设为 1 表示后续还有字节
11
        result.append((n & 0x7F) | 0x80)
12
        n >>= 7  # 将 n 右移 7 位
13
    result.append(n & 0x7F)  # 最后一个字节，最高位为 0
14
    return result
15

16
# 示例：将一
17
# 个大于 255 的数转换为 varint 编码
18
number = 26581
19
varint_encoded = int_to_varint(number)
20
print(f"Varint encoding for {number}: {[hex(b) for b in varint_encoded]}")

嵌套消息结构#

如果遇到了这段编码

1
0A 06 08 01 10 9E CF 01

可以看到就是多了前两个字节： 0A 06

其中 0A 是第一个字段，并且代表后面是一个字符串或者嵌套消息，06代表长度，接着后面刚好6个字节， 08 01 10 9E CF 01，其实就回到前面了，所以其消息结构应该是：

1
{
2
  "1": {
3
    "1": 1,
4
    "2": 26526
5
  }
6
}

一键解码pb二进制#

可以直接调用 protoc --decode_raw 来解码，不过pb编码后会丢失字段名，所以解码后也是看不到字段名的

1
❯ echo "0A 06 08 01 10 9E CF 01" | xxd -r -p | protoc --decode_raw
2

3
1 {
4
  1: 1
5
  2: 26526
6
}

或者这个例子：

1
❯ echo "0a1074645f6f70656e65775f64616d656e67120864616d656e6777781a08313139373230363922a0017b226f70656e6964223a226f33396867347a5155484b47376f414b57316655384b39425674426f222c22746f6b656e223a226a6434747165564664484b4c65366b31576d4a346550427875565137564e6e724c6c55413847442f6a5a2b4e3636535464655039534c37556936666a6176676a67714b735378646d61396d62644a6d2f6634615352517646415a414f50545578414c6b49664961513776513d227d2a00320864616d656e6777783a140a0012001a0022002a0032003a0631342e322e304202636e4804" | xxd -r -p  | protoc --decode_raw
2
1: "td_openew_dameng"
3
2: "damengwx"
4
3: "11972069"
5
4: "{\"openid\":\"o39hg4zQUHKG7oAKW1fU8K9BVtBo\",\"token\":\"jd4tqeVFdHKLe6k1WmJ4ePBxuVQ7VNnrLlUA8GD/jZ+N66STdeP9SL7Ui6fjavgjgqKsSxdma9mbdJm/f4aSRQvFAZAOPTUxALkIfIaQ7vQ=\"}"
6
5: ""
7
6: "damengwx"
8
7 {
9
  1: ""
10
  2: ""
11
  3: ""
12
  4: ""
13
  5: ""
14
  6: ""
15
  7: "14.2.0"
16
}
17
8: "cn"
18
9: 4

一键编码pb二进制#

protoc 提供了解码，但是并没有提供编码，这个让我百思不得其解，实战中经常会有需要解码后修改数据再编码回去的，这个过程是完全可逆的，不会损失任何信息

于是干脆自己vibe coding了一个，如下，非常方便

1
'''
2
python pb_test.py # 解码，默认输出到output.txt
3
python pb_test.py encode # 编码，默认读取output.txt
4
'''
5

6
from google.protobuf.internal import decoder, encoder
7
import base64
8
import struct
9

10
# 在这里粘贴base64编码的pb二进制数据
11
a = '''CA8SyQEIARABGAAgACgAMAA4AEAASABQAFgAYABoAHC+T3gAgAEAiAEAkAEAmAEAoAEAqAEDsAEAuAEAwgEIAQg2ONAP0Q/IAQDQAdvctKeiM9gBAOABAOgBAPABAPgBAYACAIgC0A+SAgCYAgKgAgCoAgywAgC4AgDAAgDIAgDYAgDgAgDoAgDyAgoIBhAAGAAgASgA+AIAgAMAigMICJvbgMgGIAGSAxUKCAiAi/nHBhAACgkIgK7+xwYQryuaAwYoADABOGSiAwCoAwAaCAix2wYQABgAGggIstsGEAAYABoICLPbBhAAGAAaCAi02wYQABgAIgYIwZoMEAAiCgihjQYQgNKTrQMiCgiijQYQgNKTrQMqvQcKIwgBEAEYASABKgIBBTC7gMHGojM4AEAASABQAGAAaABwAHgBEgMI8AcaAghlKgwIoY0GEGUYACAAKAA6DQiRThBlGAEgACgAMAA6DQiSThBkGAAgACgAMAM6DQiTThBkGAAgACgAMANCDAihnAEQZBgAIAAoAEIMCKKcARBkGAAgACgAQgwIo5wBEGQYACAAKABCDAiknAEQZBgAIAAoAEIMCKWcARBkGAAgACgAQgwIppwBEGQYACAAKABCDAipnAEQZRgBIAAoAEoMCLHqARBlGAEgACgAUg4IwbgCEGQYACAAKAAwBFoICNGGAxBlGABiCAjh1AMQZRgBaggI8aIEEGUYAXIMCIHxBBBlGAEgACgAegoIkb8FEGQYACAAggEECAEQAIIBBAgCEACCAQQIAxAAggEECAQQAIIBBAgFEACCAQQIBhAAggEECAcQAIIBBAgIEACCAQQICRAAigEMCLHbBhBkGAAgACgAigEMCLLbBhBkGAAgACgAigEMCLPbBhBkGAAgACgAkgEMCMGpBxBkGAAgACgAmgEMCOHFCBBkGAAgACgAmgEMCOLFCBBkGAAgACgAmgEMCOPFCBBkGAAgACgAmgEMCOTFCBBkGAAgACgAmgEMCOXFCBBkGAAgACgAmgEMCObFCBBkGAAgACgAmgEMCOfFCBBkGAAgACgAmgEMCOjFCBBkGAAgACgAmgEMCOnFCBBkGAAgACgAqgEMCIHiCRBkGAAgACgAqgEMCILiCRBkGAAgACgAqgEMCIPiCRBkGAAgACgAsgEOCJOwChBkGAAgACgAMAKyAQ4IkrAKEGQYACAAKAAwArIBDgiRsAoQZBgAIAAoADACugEICKH+ChBlGADCAQwIscwLEGQYACAAKADCAQwIsswLEGQYACAAKADCAQwIs8wLEGQYACAAKADCAQwItMwLEGQYACAAKADKAQgIwZoMEGQYANIBDAjU6AwQZBgAIAAoANIBDAjR6AwQZBgAIAAoANIBDAjS6AwQZBgAIAAoANIBDAjT6AwQZBgAIAAoAOIBDAjxhA4QZBgAIAAoAOIBDAjyhA4QZBgAIAAoAOIBDAjzhA4QZBgAIAAoAOIBDAj0hA4QZBgAIAAoAOoBDgiB0w4QZBgAIAAoADAD6gEOCILTDhBkGAAgACgAMAPyAQwI4ZgXEGQYACAAKAD6AQwIgbUYEGQYACAAKACCAgwI8eYXEGQYACAAKACKAh4IARACGAAgACgAMgA6AEIASABQAFgAYABoAHAAeAEyADoKCAEQy+qAyAYqAEIA'''
12

13
# base64解码
14
data = base64.b64decode(a)
15

16
def is_valid_utf8(data):
17
    """检查数据是否为有效的UTF-8字符串"""
18
    try:
19
        data.decode('utf-8')
20
        return True
21
    except:
22
        return False
23

24
def parse_message(data, depth=0):
25
    """
26
    完整解析 protobuf message，支持所有wire types
27
    返回 list of (field_number, value)，保留重复字段
28
    """
29
    pos = 0
30
    result = []
31
    while pos < len(data):
32
        try:
33
            key, pos = decoder._DecodeVarint(data, pos)
34
        except:
35
            break
36

37
        field_number = key >> 3
38
        wire_type = key & 0x7
39

40
        if wire_type == 0:  # varint
41
            value, pos = decoder._DecodeVarint(data, pos)
42
            result.append((field_number, value))
43

44
        elif wire_type == 1:  # 64-bit (fixed64, double)
45
            if pos + 8 > len(data):
46
                break
47
            value = struct.unpack('<Q', data[pos:pos+8])[0]
48
            result.append((field_number, value))
49
            pos += 8
50

51
        elif wire_type == 2:  # length-delimited (字符串/bytes/嵌套消息)
52
            length, pos = decoder._DecodeVarint(data, pos)
53
            if pos + length > len(data):
54
                break
55
            value = data[pos:pos+length]
56

57
            # 尝试解析为嵌套消息
58
            try:
59
                nested = parse_message(value, depth+1)
60
                if nested and len(nested) > 0:
61
                    # 如果成功解析出字段，认为是嵌套消息
62
                    result.append((field_number, nested))
63
                else:
64
                    # 否则作为bytes/string
65
                    result.append((field_number, value))
66
            except:
67
                # 解析失败，作为bytes/string
68
                result.append((field_number, value))
69
            pos += length
70

71
        elif wire_type == 5:  # 32-bit (fixed32, float)
72
            if pos + 4 > len(data):
73
                break
74
            value = struct.unpack('<I', data[pos:pos+4])[0]
75
            result.append((field_number, value))
76
            pos += 4
77

78
        else:
79
            # 不支持的wire type，跳过
80
            break
81

82
    return result
83

84
def format_bytes(b):
85
    """格式化bytes为带引号和转义的字符串"""
86
    result = []
87
    for byte in b:
88
        if 32 <= byte <= 126 and byte != ord('\\') and byte != ord('"'):
89
            # 可打印ASCII字符
90
            result.append(chr(byte))
91
        else:
92
            # 转义表示
93
            if byte < 8:
94
                result.append(f'\\{byte:03o}')
95
            else:
96
                result.append(f'\\{byte:03o}')
97
    return '"' + ''.join(result) + '"'
98

99
def format_value(value, indent=0):
100
    """格式化单个值"""
101
    if isinstance(value, list):
102
        # 嵌套消息
103
        lines = []
104
        lines.append(' ' * indent + '{')
105
        for field_num, field_val in value:
106
            field_lines = format_field(field_num, field_val, indent + 2)
107
            lines.extend(field_lines)
108
        lines.append(' ' * indent + '}')
109
        return lines
110
    elif isinstance(value, bytes):
111
        # bytes类型，判断是否为空字符串或二进制数据
112
        if len(value) == 0:
113
            return [' ' * indent + '""']
114
        # 检查是否全是可打印字符
115
        try:
116
            decoded = value.decode('utf-8')
117
            # 如果只包含数字字母和一些常见字符，直接显示
118
            if all(32 <= b <= 126 for b in value):
119
                return [' ' * indent + f'"{decoded}"']
120
        except:
121
            pass
122
        # 否则使用转义格式
123
        return [' ' * indent + format_bytes(value)]
124
    else:
125
        # 整数
126
        return [' ' * indent + str(value)]
127

128
def format_field(field_num, value, indent=0):
129
    """格式化字段"""
130
    lines = []
131
    if isinstance(value, list):
132
        # 嵌套消息
133
        lines.append(' ' * indent + f'{field_num} {{')
134
        for sub_field_num, sub_value in value:
135
            sub_lines = format_field(sub_field_num, sub_value, indent + 2)
136
            lines.extend(sub_lines)
137
        lines.append(' ' * indent + '}')
138
    elif isinstance(value, bytes):
139
        # bytes类型
140
        if len(value) == 0:
141
            lines.append(' ' * indent + f'{field_num}: ""')
142
        else:
143
            # 尝试解码
144
            try:
145
                decoded = value.decode('utf-8')
146
                if all(32 <= b <= 126 for b in value):
147
                    lines.append(' ' * indent + f'{field_num}: "{decoded}"')
148
                else:
149
                    lines.append(' ' * indent + f'{field_num}: {format_bytes(value)}')
150
            except:
151
                lines.append(' ' * indent + f'{field_num}: {format_bytes(value)}')
152
    else:
153
        # 整数
154
        lines.append(' ' * indent + f'{field_num}: {value}')
155
    return lines
156

157
def format_message(parsed_list):
158
    """格式化整个消息为res1.txt的格式"""
159
    lines = []
160
    for field_num, value in parsed_list:
161
        field_lines = format_field(field_num, value, 0)
162
        lines.extend(field_lines)
163
    return '\n'.join(lines)
164

165
def parse_escaped_string(s):
166
    """解析带转义的字符串，返回bytes"""
167
    if not s.startswith('"') or not s.endswith('"'):
168
        raise ValueError(f"字符串必须以引号包围: {s}")
169

170
    s = s[1:-1]  # 去掉首尾引号
171
    result = []
172
    i = 0
173
    while i < len(s):
174
        if s[i] == '\\' and i + 3 < len(s):
175
            # 八进制转义 \xxx
176
            try:
177
                octal = s[i+1:i+4]
178
                byte_val = int(octal, 8)
179
                result.append(byte_val)
180
                i += 4
181
            except:
182
                result.append(ord(s[i]))
183
                i += 1
184
        else:
185
            result.append(ord(s[i]))
186
            i += 1
187
    return bytes(result)
188

189
def parse_text_format(lines):
190
    """
191
    从文本格式解析回数据结构
192
    返回 list of (field_number, value)
193
    """
194
    result = []
195
    i = 0
196

197
    while i < len(lines):
198
        line = lines[i].rstrip()
199
        if not line or line.isspace():
200
            i += 1
201
            continue
202

203
        # 计算缩进
204
        indent = len(line) - len(line.lstrip())
205
        line = line.strip()
206

207
        if line == '}':
208
            # 消息结束
209
            break
210
        elif ':' in line and not line.endswith('{'):
211
            # 字段: 值
212
            field_str, value_str = line.split(':', 1)
213
            field_num = int(field_str.strip())
214
            value_str = value_str.strip()
215

216
            if value_str.startswith('"'):
217
                # 字符串/bytes值
218
                value = parse_escaped_string(value_str)
219
            else:
220
                # 整数值
221
                value = int(value_str)
222

223
            result.append((field_num, value))
224
            i += 1
225
        elif line.endswith('{'):
226
            # 嵌套消息开始
227
            field_str = line[:-1].strip()
228
            field_num = int(field_str)
229

230
            # 找到匹配的结束括号
231
            bracket_count = 1
232
            j = i + 1
233
            start_line = j
234
            while j < len(lines) and bracket_count > 0:
235
                sub_line = lines[j].strip()
236
                if sub_line.endswith('{'):
237
                    bracket_count += 1
238
                elif sub_line == '}':
239
                    bracket_count -= 1
240
                j += 1
241

242
            # 递归解析嵌套消息
243
            nested_lines = lines[start_line:j-1]
244
            nested_value = parse_text_format(nested_lines)
245
            result.append((field_num, nested_value))
246
            i = j
247
        else:
248
            i += 1
249

250
    return result
251

252
def encode_message(parsed_list):
253
    """
254
    将解析后的list编码回protobuf二进制
255
    """
256
    parts = []
257
    for field_number, value in parsed_list:
258
        if isinstance(value, int):  # varint
259
            key = (field_number << 3) | 0
260
            parts.append(encoder._VarintBytes(key))
261
            parts.append(encoder._VarintBytes(value))
262
        elif isinstance(value, list):  # 嵌套消息
263
            key = (field_number << 3) | 2
264
            nested_bytes = encode_message(value)
265
            parts.append(encoder._VarintBytes(key))
266
            parts.append(encoder._VarintBytes(len(nested_bytes)))
267
            parts.append(nested_bytes)
268
        elif isinstance(value, bytes):
269
            key = (field_number << 3) | 2
270
            parts.append(encoder._VarintBytes(key))
271
            parts.append(encoder._VarintBytes(len(value)))
272
            parts.append(value)
273
        else:
274
            raise TypeError(f"不支持的类型: {type(value)}")
275
    return b''.join(parts)
276

277
if __name__ == "__main__":
278
    import sys
279

280
    if len(sys.argv) > 1 and sys.argv[1] == 'encode':
281
        # 反向模式：从文本文件编码回base64
282
        input_file = sys.argv[2] if len(sys.argv) > 2 else 'output.txt'
283

284
        print(f"正在从 {input_file} 读取并编码...")
285
        with open(input_file, 'r', encoding='utf-8') as f:
286
            lines = f.readlines()
287

288
        # 解析文本格式
289
        parsed = parse_text_format(lines)
290

291
        # 编码回二进制
292
        encoded_data = encode_message(parsed)
293

294
        # base64编码
295
        b64_result = base64.b64encode(encoded_data).decode('ascii')
296

297
        print("\n编码结果（base64）:")
298
        print(b64_result)
299
        print(f"\n原始长度: {len(data)} bytes")
300
        print(f"编码长度: {len(encoded_data)} bytes")
301
        print(f"是否匹配: {encoded_data == data}")
302

303
        # 保存到文件
304
        with open('encoded_result.txt', 'w') as f:
305
            f.write(b64_result)
306
        print(f"\n已保存到 encoded_result.txt")
307

308
    else:
309
        # 正常模式：解析base64并输出文本
310
        print("正在解析protobuf数据...")
311
        msg = parse_message(data)
312

313
        # 格式化输出
314
        formatted = format_message(msg)
315
        print(formatted)
316

317
        # 保存到文件
318
        with open('output1.txt', 'w', encoding='utf-8') as f:
319
            f.write(formatted + '\n')
320
        print("\n已保存到 output.txt")
321

322
        # 验证：编码回去看是否一致
323
        print("\n验证：重新编码...")
324
        re_encoded = encode_message(msg)
325
        print(f"原始数据: {len(data)} bytes")
326
        print(f"重新编码: {len(re_encoded)} bytes")
327
        print(f"数据一致: {re_encoded == data}")
328

329
        if re_encoded == data:
330
            print("✅ 编码解码完全可逆！")
331
        else:
332
            print("❌ 数据不一致")
333
            # 找出差异
334
            for i, (a, b) in enumerate(zip(data, re_encoded)):
335
                if a != b:
336
                    print(f"第一个差异在位置 {i}: 原始={a:02x}, 重编码={b:02x}")
337
                    break