#!/usr/bin/env python3 """ 996传奇客户端 协议线层帧装拆 来源: netClient::SendNetMessage @0x10003d408 + encode_header @0x10004b4e4 反汇编还原 校核: apilot/wire.py + protocol_bot/encode6bitbuf_final.py (已实网验证) 验证: 与已知心跳帧逐字节一致 (2026-09-23) 线格式 (从 encode6bitbuf_final.build_frame + apilot/wire.HDR 反推): ┌─────────────────────────── 36B 明文头(9×i32, little-endian) ──────────────────────────┐ │ [0] zero [1] msg_id [2] recog [3] p1 [4] p2 [5] p3 [6] zero [7] body_len [8] tail │ └──────────────────────────────────────────────────────────────────────────────────────────┘ │ Encode6BitBuf (hdr_codec.encode6) ▼ ┌────── 48B 编码头 (全可打印 0x3C..0x7B, 即 "<=?@A-Z[\\]^_`a-z{") ──────┐ ┌──── N 字节 body (明文, 不加密) ────┐ │ ↑ server→client 时, body 可能是 zlib 压缩 JSON 备注: - 24B→32B (hdr_codec.py) 是 36B→48B 的前 24 字节子集, 即 netMessage 类的内部 header 表示 - 实际 socket 上跑的是 36B→48B + body - [0][6] 恒为 0 (无业务含义, 推测是协议层预留) - [8] tail 字段在不同消息上有不同含义, 常见值: 0=普通, 1=批量(-1)的分片, 9=握手串 依赖: hdr_codec.py (同目录, ENC_TABLE + encode6/decode6) """ from __future__ import annotations import os import struct import sys from dataclasses import dataclass from typing import Optional sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from hdr_codec import ENC_TABLE, decode_header, encode_header # wire format 常量 HEADER_PLAIN_SIZE = 36 # 9 × i32 明文头 HEADER_ENC_SIZE = 48 # 编码后头大小 (36B → 48B, 4/3 比例) MIN_FRAME_SIZE = HEADER_ENC_SIZE # 最小帧 = 48B 头 + 0B body # 头布局 (9×i32, little-endian) HDR_STRUCT = struct.Struct("<9i") @dataclass class Frame: """一条协议消息的完整结构""" msg_id: int recog: int = 0 # 业务识别符, 创角时 900000=水印 p1: int = 0 # param1 p2: int = 0 # param2 p3: int = 0 # param3 tail: int = 0 # 末尾字段, 见文档说明 body: bytes = b"" def pack(self) -> bytes: """组帧: 36B 明文头 → 48B 编码头 + body""" return pack_frame(self.msg_id, self.recog, self.p1, self.p2, self.p3, self.tail, self.body) def _encode_full(plain36: bytes) -> bytes: """36B 明文 → 48B 编码 (与 hdr_codec.encode_header 24B→32B 同一算法, 输入更长) 算法 (来自 netClient::encode_header @0x10004b4e4): 对每个明文字节 b: v = (ENC_TABLE[b] ^ 9) & 0xFFFF bitcnt 当前累积位 (起始 0) out_byte = (acc | (v >> (bitcnt+2))) & 0x3F + 0x3C leftover = ((v << (6-bitcnt)) >> 2) & 0x3F 若 bitcnt < 4: acc = leftover; bitcnt += 2 否则: 输出 leftover+0x3C; acc=0, bitcnt=0 末尾若 bitcnt>0: 输出 acc+0x3C """ if len(plain36) % 3 != 0: raise ValueError(f"plain 长度必须为 3 倍数 (encode_full 是 24B→32B 的扩展), 实际={len(plain36)}") out = bytearray() acc = 0 bitcnt = 0 for ch in plain36: v = (ENC_TABLE[ch] ^ 9) & 0xFFFF new_bitcnt = bitcnt + 2 acc = (acc | (v >> new_bitcnt)) & 0x3F out.append((acc + 0x3C) & 0xFF) leftover = ((v << ((6 - bitcnt) & 0xFF)) & 0xFFFFFFFF) >> 2 & 0x3F if bitcnt < 4: acc = leftover bitcnt = new_bitcnt else: out.append((leftover + 0x3C) & 0xFF) acc = 0 bitcnt = 0 if bitcnt > 0: out.append((acc + 0x3C) & 0xFF) return bytes(out) def _decode_full(enc_bytes: bytes, plain_size: int = HEADER_PLAIN_SIZE) -> bytes: """编码字节流 → 明文 (与 hdr_codec.decode_header 32B→24B 同一算法, 输出更长)""" bits = 0 nbits = 0 raw = bytearray() for ch in enc_bytes: bits = (bits << 6) | (ch - 0x3C) nbits += 6 while nbits >= 8: nbits -= 8 raw.append((bits >> nbits) & 0xFF) # 还原: raw 字节过 INV_TABLE[ENC_TABLE[i] ^ 9] = i inv = [0] * 256 for i in range(256): inv[ENC_TABLE[i] ^ 9] = i plain = bytes(inv[b] for b in raw[:plain_size]) return plain def pack_frame(msg_id: int, recog: int, p1: int, p2: int, p3: int, tail: int, body: bytes = b"") -> bytes: """ 构造一条线协议消息 :return: 48B 编码头 + body """ # 9 字段: zero, msg_id, recog, p1, p2, p3, zero, body_len, tail head36 = HDR_STRUCT.pack(0, msg_id, recog, p1, p2, p3, 0, len(body), tail) return _encode_full(head36) + body def unpack_stream(buf: bytes) -> tuple[list[Frame], bytes]: """ 从累积的字节流中切出所有完整帧 :return: (frames, rest) — frames 是切出的完整帧, rest 是不足一帧的尾部残留 """ out: list[Frame] = [] pos = 0 while pos + HEADER_ENC_SIZE <= len(buf): # 解头 head36 = _decode_full(buf[pos:pos + HEADER_ENC_SIZE], HEADER_PLAIN_SIZE) try: zero, msg_id, recog, p1, p2, p3, zero2, body_len, tail = HDR_STRUCT.unpack(head36) except struct.error: break # 检查 body 是否完整 if pos + HEADER_ENC_SIZE + body_len > len(buf): break body = buf[pos + HEADER_ENC_SIZE:pos + HEADER_ENC_SIZE + body_len] out.append(Frame(msg_id=msg_id, recog=recog, p1=p1, p2=p2, p3=p3, tail=tail, body=body)) pos += HEADER_ENC_SIZE + body_len return out, buf[pos:] # ============== 已知消息类型 (来自 netMessageDef.lua) ============== # 常用 msg_id 速查 MSG_CS_LOGIN = 20 MSG_CS_CREATE_ROLE = 101 MSG_CS_SELECT_ROLE = 103 MSG_SC_ROLE_LIST = 520 MSG_SC_CREATE_ROLE_OK = 521 MSG_SC_CREATE_ROLE_FAIL = 522 MSG_SC_GAME_SERVER_ADDR = 525 MSG_CS_HEARTBEAT = 60000 MSG_SC_DISCONNECTED = 0x1FFFFFFE MSG_ILLEGAL = 0x2FFFFFFE MSG_BATCH = -1 if __name__ == "__main__": import sys import os _PARENT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _PARENT not in sys.path: sys.path.insert(0, _PARENT) from protocol_bot.encode6bitbuf_final import encode6 as ref_encode6, build_frame as ref_build_frame head36_test = HDR_STRUCT.pack(0, 60000, 1789709744, 0, 0, 149, 0, 0, 0x70000000) enc_mine = _encode_full(head36_test) enc_ref = ref_encode6(head36_test) print(f"encoded heartbeat: {enc_mine.hex()}") print(f"reference (encode6): {enc_ref.hex()}") assert enc_mine == enc_ref, f"心跳帧编码与 reference 不一致" print("✓ 心跳帧 48B 编码与 protocol_bot/encode6bitbuf_final 字节级一致\n") # 自测 2: roundtrip f = Frame(msg_id=101, recog=900000, p1=12345, p2=37504, p3=0, tail=0, body=b'{"chr":"\xe8\x90\x8c\xe6\x96\xb0","job":0,"sex":1}') packed = f.pack() print(f"packed: {len(packed)}B = 48 header + {len(f.body)} body") frames, rest = unpack_stream(packed) assert rest == b"" assert len(frames) == 1 g = frames[0] assert g.msg_id == f.msg_id assert g.recog == f.recog assert g.p1 == f.p1 assert g.p2 == f.p2 assert g.p3 == f.p3 assert g.tail == f.tail assert g.body == f.body print(f"✓ Frame roundtrip: msg_id={g.msg_id}, recog={g.recog}, body={g.body!r}\n") # 自测 3: 部分包(模拟半包/粘包) partial = packed[:-10] # 砍掉最后 10 字节 frames, rest = unpack_stream(partial) assert frames == [] assert rest == partial print(f"✓ Partial packet: 0 frames, rest len={len(rest)} == {len(partial)}\n") # 自测 4: 多包粘包 f1 = Frame(msg_id=20, recog=0, p1=0, p2=0, p3=0, tail=9, body=b"*#acct/role/12345/9/1/0/110585/0/1/1/1/1") f2 = Frame(msg_id=529, recog=0, p1=0, p2=0, p3=0, tail=1, body=b"key32abc") multi = f1.pack() + f2.pack() frames, rest = unpack_stream(multi) assert len(frames) == 2 assert frames[0].msg_id == 20 assert frames[1].msg_id == 529 assert frames[1].body == b"key32abc" print(f"✓ Multi-packet: 2 frames [{frames[0].msg_id}, {frames[1].msg_id}], rest={rest!r}\n") other = ref_build_frame(101, 900000, 12345, 37504, 0) mine = pack_frame(101, 900000, 12345, 37504, 0, 0, b"") assert other == mine, f"interop fail:\n other: {other.hex()}\n mine: {mine.hex()}" print(f"✓ 与 protocol_bot/encode6bitbuf_final.build_frame 字节级一致\n") sample = Frame(msg_id=12345, recog=0x12345678, p1=0x0A0B0C0D, p2=0x12345678, p3=0, tail=0, body=b"x"*100) enc_header = sample.pack()[:48] printable = all(0x3C <= b <= 0x7B for b in enc_header) assert printable, f"非可打印字符出现: {enc_header.hex()}" print(f"✓ 编码头全可打印 (0x3C..0x7B): {enc_header[:8].hex()}...{enc_header[-4:].hex()}\n") print("All frame.py self-tests passed.")