某方数据protobuf协议逆向#

1
data = {
2
    "\\u0000\\u0000\\u0000\\u0000": "\n$\n\\u0005paper\\u0012\\u0006爬虫(\\u00010\\u0014B\\u0001\\u0000H\\u0001b\\u0002pcj\\u0006search\\u0010\\u0001\"\\u0007AI_READ\"\nAI_EXTRACT"
3
}

1
import blackboxprotobuf
2

3
import codecs
4

5
# 示例 protobuf 字节串
6
protobuf_bytes = b'\x08\x96\x01'
7
message_data = {
8
    '1': {                          # 嵌套对象
9
        '1': b'paper',              # 字符串用 bytes
10
        '2': '爬虫'.encode('utf-8'),  # 中文必须编码为 UTF-8
11
        '5': 1,                     # int
12
        '6': 20,                    # int
13
        '8': b'\x00',              # 空字节
14
        '9': 1,                     # int
15
        '12': {                     # 嵌套对象
16
            '14': 99               # int (ascii 'c')
17
        },
18
        '13': b'search',           # 字符串
19
    },
20
    '2': 1,                         # int
21
    '4': [                          # 数组
22
        b'AI_READ',
23
        b'AI_EXTRACT'
24
    ]
25
}
26
s = "\n$\n\\u0005paper\\u0012\\u0006爬虫(\\u00010\\u0014B\\u0001\\u0000H\\u0001b\\u0002pcj\\u0006search\\u0010\\u0001\"\\u0007AI_READ\"\nAI_EXTRACT"
27
decoded = codecs.decode(s, 'unicode_escape')  # 把 \\u0005 变成真正的 \x05
28
mybytes = decoded.encode('latin-1')
29
# 解码 protobuf 消息
30
decoded_message, type_definition = blackboxprotobuf.decode_message(mybytes)
31
print('type:',type_definition)
32
print("Decoded Message:", decoded_message)
33

34
# 重新编码消息
35
encoded_bytes = blackboxprotobuf.encode_message(decoded_message, type_definition)
36
newencoded_bytes = blackboxprotobuf.encode_message(message_data, type_definition)
37
print("Encoded Bytes:", encoded_bytes)
38
print('new encoded:',newencoded_bytes)

1
for (o = (t = (0,
2
                o.a)(t.getRequestMessage())).length,
3
                d = [0, 0, 0, 0],
4
                s = new Uint8Array(5 + o),
5
                a = 3; 0 <= a; a--)
6
                    d[a] = o % 256,
7
                    o >>>= 8;
8
                if (s.set(new Uint8Array(d), 1),
9
                s.set(t, 5),
10
                t = s,
11
                "text" == e.a)

1
a = 3; 0 <= a; a--)
2
                    d[a] = o % 256,
3
                    o >>>= 8;

1
[0, 0, 0, 0, 61, 10, 36, 10, 5, 112, 97, 112, 101, 114, 18, 6, 231, 136, 172, 232, 153, 171, 40, 1, 48, 20, 66, 1, 0, 72, 1, 98, 2, 112, 99, 106, 6, 115, 101, 97, 114, 99, 104, 16, 1, 34, 7, 65, 73, 95, 82, 69, 65, 68, 34, 10, 65, 73, 95, 69, 88, 84, 82, 65, 67, 84]

1
[0, 0, 0, 0, 61]

1
def encode_varint(value):
2
    result = []
3
    while value > 0x7f:
4
        result.append((value & 0x7f) | 0x80)
5
        value >>= 7
6
    result.append(value & 0x7f)
7
    return bytes(result)
8

9
def encode_length_delimited(field_number, data):
10
    tag = (field_number << 3) | 2
11
    return encode_varint(tag) + encode_varint(len(data)) + data
12

13
def encode_varint_field(field_number, value):
14
    tag = (field_number << 3) | 0
15
    return encode_varint(tag) + encode_varint(value)
16

17
query = '美女'.encode('utf-8')
18
inner = b''
19
inner += encode_length_delimited(1, b'paper')
20
inner += encode_length_delimited(2, query)
21
inner += encode_varint_field(5, 1)
22
inner += encode_varint_field(6, 20)
23
inner += encode_length_delimited(8, b'\x00')
24
inner += encode_varint_field(9, 1)
25
inner12 = encode_varint_field(14, 99)
26
inner += encode_length_delimited(12, inner12)
27
inner += encode_length_delimited(13, b'search')
28

29
body = b''
30
body += encode_length_delimited(1, inner)
31
body += encode_varint_field(2, 1)
32
body += encode_length_delimited(4, b'AI_READ')
33
body += encode_length_delimited(4, b'AI_EXTRACT')
34

35
grpc_frame = b'\x00' + struct.pack('>I', len(body)) + body

1
import json
2
import struct
3
import re
4

5
with open('response_raw.json', 'r', encoding='utf-8') as f:
6
    raw_data = json.load(f)
7

8
data = bytes.fromhex(raw_data['content_hex'])
9

10
# 解析 gRPC 帧头
11
msg_len = struct.unpack('>I', data[1:5])[0]
12
proto_data = data[5:5 + msg_len]
13

14

15
def read_varint(data, pos):
16
    result = 0
17
    shift = 0
18
    while pos < len(data):
19
        byte = data[pos]
20
        pos += 1
21
        result |= (byte & 0x7f) << shift
22
        if not (byte & 0x80):
23
            break
24
        shift += 7
25
    return result, pos
26

27

28
def parse_protobuf(data, max_depth=5):
29
    if max_depth <= 0:
30
        return {}
31
    result = {}
32
    pos = 0
33
    while pos < len(data):
34
        try:
35
            tag, pos = read_varint(data, pos)
36
        except:
37
            break
38
        field_num = tag >> 3
39
        wire_type = tag & 0x07
40

41
        if wire_type == 0:
42
            value, pos = read_varint(data, pos)
43
        elif wire_type == 2:
44
            length, pos = read_varint(data, pos)
45
            value = data[pos:pos + length]
46
            pos += length
47
            try:
48
                text = value.decode('utf-8')
49
                if any('\u4e00' <= c <= '\u9fff' for c in text[:50]):
50
                    value = text
51
                else:
52
                    sub = parse_protobuf(value, max_depth - 1)
53
                    value = sub if sub else text
54
            except:
55
                sub = parse_protobuf(value, max_depth - 1)
56
                value = sub if sub else value
57
        else:
58
            break
59

60
        if field_num in result:
61
            if not isinstance(result[field_num], list):
62
                result[field_num] = [result[field_num]]
63
            result[field_num].append(value)
64
        else:
65
            result[field_num] = value
66
    return result
67

68

69
top = parse_protobuf(proto_data)
70
papers = top.get(4, [])
71

72
print(f"共 {len(papers)} 篇论文:\n")
73

74
for i, paper in enumerate(papers):
75
    inner = paper.get(101, {})
76
    if not isinstance(inner, dict) or not inner:
77
        continue
78

79
    # 标题：优先取中文，否则取第一个
80
    title_list = inner.get(2, '')
81
    if not isinstance(title_list, list):
82
        title_list = [title_list]
83
    title = ''
84
    for t in title_list:
85
        t_str = str(t)
86
        if any('\u4e00' <= c <= '\u9fff' for c in t_str):
87
            title = t_str
88
            break
89
    if not title and title_list:
90
        title = str(title_list[0])
91
    title = re.sub(r'<[^>]+>', '', title)
92

93
    # 作者
94
    authors = inner.get(3, [])
95
    if not isinstance(authors, list):
96
        authors = [authors]
97
    authors = [str(a) for a in authors if isinstance(a, str) and any('\u4e00' <= c <= '\u9fff' for c in a)]
98

99
    # 期刊
100
    journal_list = inner.get(23, '')
101
    if not isinstance(journal_list, list):
102
        journal_list = [journal_list]
103
    journal = ''
104
    for j in journal_list:
105
        j_str = str(j)
106
        if any('\u4e00' <= c <= '\u9fff' for c in j_str):
107
            journal = j_str
108
            break
109
    if not journal and journal_list:
110
        journal = str(journal_list[0])
111

112
    year = inner.get(33, '')
113
    page = inner.get(36, '')
114
    doi = inner.get(41, '')
115

116
    print(f"{i + 1}. {title}")
117
    print(f"   作者: {', '.join(authors[:3]) if authors else '未知'}")
118
    print(f"   期刊: {journal}")
119
    print(f"   年份: {year}  页码: {page}")
120
    print(f"   DOI: {doi}")
121
    print()