chuanqi/cpp_reversed/frame.py

#!/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.")