chuanqi/cpp_reversed/msg_codec.py

#!/usr/bin/env python3
"""
996传奇客户端 协议编解码 统一 API
来源: cpp_reversed/ 目录下各模块汇总
  - hdr_codec:  24B 头编解码 (Encode6BitBuf 算法核心)
  - frame:      36B 头 + body 完整帧装拆
  - xxtea:      脚本密文加解密 (LuaBridgeCtl::SetXXTEAKeyGM 配套)
  - rsa:        客户端本地 RSA 密钥对 (RsaEncryptPub / RsaDecryptPri)

定位: 这是 cpp_reversed/ 体系的"使用层入口", 所有"我想发一条消息" / "我想解一条抓包"
的操作都从这里走。底层模块仍可单独 import 使用。

依赖: 同目录的 hdr_codec.py / frame.py / xxtea.py / rsa.py
      跨目录 (protocol_bot/) 仅在 self-test 交叉验证时使用
"""
from __future__ import annotations
import os
import sys
from typing import Optional, Dict  # noqa: F401

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from hdr_codec import (  # type: ignore
    ENC_TABLE,
    encode_header, decode_header, parse_header,
)
from frame import (  # type: ignore
    Frame, HDR_STRUCT, HEADER_PLAIN_SIZE, HEADER_ENC_SIZE,
    pack_frame, unpack_stream,
    MSG_CS_LOGIN, MSG_CS_CREATE_ROLE, MSG_CS_SELECT_ROLE,
    MSG_SC_ROLE_LIST, MSG_SC_CREATE_ROLE_OK, MSG_SC_CREATE_ROLE_FAIL,
    MSG_SC_GAME_SERVER_ADDR, MSG_CS_HEARTBEAT,
    MSG_SC_DISCONNECTED, MSG_ILLEGAL, MSG_BATCH,
)
from xxtea import (  # type: ignore
    encrypt as xxtea_encrypt, decrypt as xxtea_decrypt,
    encrypt_padded, decrypt_padded,
    SCRIPT_KEY_RAW, SCRIPT_SIGN,
)
from rsa import encrypt_pub, decrypt_pri, PUB_KEY_N, PUB_KEY_E, PRI_KEY_D  # type: ignore


__all__ = [
    # 头编解码 (24B 内部表示)
    "encode_header", "decode_header", "parse_header", "ENC_TABLE",
    # 完整帧
    "Frame", "pack_frame", "unpack_stream", "send", "receive",
    # XXTEA
    "xxtea_encrypt", "xxtea_decrypt", "encrypt_padded", "decrypt_padded",
    "SCRIPT_KEY_RAW", "SCRIPT_SIGN",
    # RSA
    "encrypt_pub", "decrypt_pri",
    # 常用消息号
    "MSG_CS_LOGIN", "MSG_CS_CREATE_ROLE", "MSG_CS_SELECT_ROLE",
    "MSG_SC_ROLE_LIST", "MSG_SC_CREATE_ROLE_OK", "MSG_SC_CREATE_ROLE_FAIL",
    "MSG_SC_GAME_SERVER_ADDR", "MSG_CS_HEARTBEAT",
    "MSG_SC_DISCONNECTED", "MSG_ILLEGAL", "MSG_BATCH",
    # 头结构常量
    "HDR_STRUCT", "HEADER_PLAIN_SIZE", "HEADER_ENC_SIZE",
]


# ============== 高层 API ==============

def send(msg_id: int, recog: int, p1: int, p2: int, p3: int,
         body: bytes = b"", tail: int = 0) -> bytes:
    """
    构造一条要发送的线协议消息
    :return: 48B 编码头 + body, 可直接 sendall
    """
    return pack_frame(msg_id, recog, p1, p2, p3, tail, body)


def receive(buffer: bytes) -> tuple[list[Frame], bytes]:
    """
    从累积的字节流中切出所有完整帧
    :return: (frames, rest) — frames 是完整帧, rest 是不足一帧的尾部残留 (供下次 recv 拼)
    """
    return unpack_stream(buffer)


def frame_to_dict(f: Frame) -> dict:
    """Frame → 人类可读的 dict (供日志/调试)"""
    return {
        "msg_id": f.msg_id,
        "recog": f.recog,
        "p1": f.p1,
        "p2": f.p2,
        "p3": f.p3,
        "tail": f.tail,
        "body_len": len(f.body),
        "body_preview": f.body[:32].hex() + ("..." if len(f.body) > 32 else ""),
    }


def decode_captured(enc_bytes: bytes) -> dict:
    """
    抓包分析: 编码字节 → 明文头 dict (供 Wireshark 抓 TCP 流后 1-行解析)

    接受两种输入:
      - 32B (旧 24B 头, 用 hdr_codec 解析)
      - 48B (新 36B 头, 用 frame 解析, 包含 body_len / tail)

    自动识别: 长度 >= 48 → 走 wire 格式; 否则按 hdr_codec 24B 格式
    """
    if len(enc_bytes) >= HEADER_ENC_SIZE:
        # wire 格式 36B 头: 0, msg_id, recog, p1, p2, p3, 0, body_len, tail
        plain36 = _decode_full_to_36(enc_bytes[:HEADER_ENC_SIZE])
        zero, msg_id, recog, p1, p2, p3, zero2, body_len, tail = HDR_STRUCT.unpack(plain36)
        return {
            "msgId": msg_id, "recog": recog,
            "p1": p1, "p2": p2, "p3": p3,
            "body_len": body_len, "tail": tail,
        }
    elif len(enc_bytes) >= 32:
        # 24B 旧格式 (hdr_codec)
        plain24 = decode_header(bytes(enc_bytes[:32]))
        return parse_header(plain24)
    else:
        raise ValueError(f"至少需要 32B 编码头, 实际={len(enc_bytes)}")


def _decode_full_to_36(enc48: bytes) -> bytes:
    """48B 编码头 → 36B 明文头 (与 frame._decode_full 相同, 此处独立以避免循环引用)"""
    from frame import _decode_full
    return _decode_full(enc48, HEADER_PLAIN_SIZE)


# ============== 反汇编地址速查 (供"我想精读某函数实现"用) ==============

ADDR = {
    "encode_header": "0x10004b4e4",   # 头编码 (24B→32B)
    "decode_header": "0x10004b594",   # 头解码 (32B→24B)
    "SendNetMessage": "0x10003d408",  # 加密+入队
    "LuaNetworkController_SendMsg": "0x10005f8cc",
    "tolua_SendMsg_binding": "0x10007cd54",
    "netClient_recv_loop": "0x100023df8",
    "netClient_recv_dispatch": "0x10003cd30",
    "LuaBridgeCtl_SetXXTEAKeyGM": "0x100044b88",
    "init_keys_and_sign": "0x100043bdc",
    "xor_deobfuscate": "0x100043b34",
}


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 build_frame, encode6
    from frame import HDR_STRUCT

    print("msg_codec.py — 996传奇客户端 协议编解码 统一 API")
    print("=" * 60)
    print()

    # 测试 1: send → receive roundtrip
    body = b'{"chr":"\xe8\x90\x8c\xe6\x96\xb0","job":0,"sex":1}'
    sent = send(MSG_CS_CREATE_ROLE, 900000, 12345, 37504, 0, body)
    print(f"[1] send():  msg_id={MSG_CS_CREATE_ROLE}, body={len(body)}B")
    print(f"    wire:    {len(sent)}B = 48 header + {len(body)} body")
    print(f"    hex:     {sent.hex()}")
    print(f"    ascii:   {''.join(chr(b) if 0x3C<=b<=0x7B else '.' for b in sent[:48])}")
    print()

    frames, rest = receive(sent)
    assert rest == b""
    assert len(frames) == 1
    f = frames[0]
    print(f"[2] receive(): {len(frames)} frame, rest={rest!r}")
    print(f"    decoded: {frame_to_dict(f)}")
    assert f.body == body
    print()

    # 测试 3: 编码头与 protocol_bot 字节级一致
    sent_no_body = send(MSG_CS_CREATE_ROLE, 900000, 12345, 37504, 0, b"")
    other = build_frame(MSG_CS_CREATE_ROLE, 900000, 12345, 37504, 0)
    assert sent_no_body == other, "与 protocol_bot 不一致"
    print(f"[3] wire format 与 protocol_bot 字节级一致 ✓")
    print()

    # 测试 4: parse_header 抓包分析
    enc = sent[:48]  # 前 48B 编码头 (wire 格式)
    fields = decode_captured(enc)
    print(f"[4] decode_captured(48B wire): {fields}")
    assert fields["msgId"] == MSG_CS_CREATE_ROLE
    assert fields["recog"] == 900000
    assert fields["p1"] == 12345
    print()

    # 测试 4b: decode_captured 也支持 24B 旧格式
    legacy24 = encode_header(b"\x00" * 24)  # 占位
    legacy_dict = decode_captured(legacy24)
    print(f"[4b] decode_captured(32B legacy): keys={sorted(legacy_dict.keys())} ✓")
    print()

    # 测试 5: XXTEA 自洽 (脚本密文场景)
    from xxtea import encrypt as xe, decrypt as xd
    test_key = b"0123456789abcdef"
    test_plain = b"Hello XXTEA!!!  "  # 16 字节 (8 倍数)
    enc = xe(test_plain, test_key)
    dec = xd(enc, test_key)
    assert dec == test_plain
    print(f"[5] XXTEA roundtrip: '{test_plain.decode()}' → {enc.hex()} → '{dec.decode()}' ✓")
    print()

    # 测试 6: 多包粘包
    f1 = send(MSG_CS_LOGIN, 0, 0, 0, 0, b"*#acct/role/12345/9/1/0", 9)
    f2 = send(MSG_SC_ROLE_LIST, 0, 0, 0, 0, b"key32/session123", 0)
    multi = f1 + f2
    frames, rest = receive(multi)
    assert rest == b""
    assert len(frames) == 2
    assert frames[0].msg_id == MSG_CS_LOGIN
    assert frames[1].msg_id == MSG_SC_ROLE_LIST
    print(f"[6] 多包粘包: 2 帧 [{frames[0].msg_id}, {frames[1].msg_id}] ✓")
    print()

    # 测试 7: RSA 接口(应抛 NotImplementedError, 因密钥未提取)
    try:
        encrypt_pub(b"test")
        print("[7] FAIL: 应抛 NotImplementedError")
    except NotImplementedError as e:
        print(f"[7] RSA encrypt_pub 接口: NotImplementedError (密钥待提取) ✓")
    print()

    print("=" * 60)
    print("All msg_codec self-tests passed.")
    print()
    print("可用入口:")
    print("  send(msg_id, recog, p1, p2, p3, body) → 48B 头 + body")
    print("  receive(buffer) → (frames, rest)")
    print("  frame_to_dict(f) → 调试 dict")
    print("  decode_captured(32B 编码头) → 抓包解析 dict")