chuanqi/protocol_bot/scripts/scan_sourid_pcap.py

#!/usr/bin/env python3
"""scan_sourid_pcap.py — 从 pcap 重组 TCP 流, 提取全部 SC51/SC634 地图属性。

方法(交接文档定案): scapy 重组 sport=8043 流 → encode6 帧头切帧 →
SC-1 批量帧 zlib 解压子帧 → 全部 SC51/SC634 decode6 → sourid 提取。

用法: sudo .venv/bin/python scripts/scan_sourid_pcap.py /tmp/cap_full.pcap
"""
from __future__ import annotations

import json
import sys
import zlib

sys.path.insert(0, "src")
from pbot.codec.bitbuf import decode6  # noqa: E402


def frames_from_stream(buf: bytes) -> list[tuple[int, bytes]]:
    """从 TCP 字节流按 encode6 帧头切帧(FrameSplitter: 粘包/半包安全)。"""
    from pbot.codec.frame import FrameSplitter
    sp = FrameSplitter()
    return [(f.msg_id, f.body) for f in sp.feed(buf)]


def extract_map_bodies(frames: list[tuple[int, bytes]]) -> list[dict]:
    out: list[dict] = []
    for msg, body in frames:
        if msg == -1:
            # 批量帧: zlib 压缩子帧流
            try:
                raw = zlib.decompress(body)
            except zlib.error:
                continue
            out.extend(extract_map_bodies(frames_from_stream(raw)))
            continue
        if msg not in (51, 634):
            continue
        try:
            text = decode6(body).decode("utf-8", errors="replace")
            start = text.find("{")
            if start < 0:
                continue
            j = json.loads(text[start:text.rfind("}") + 1])
            j["_msg"] = f"SC{msg}"
            out.append(j)
        except Exception:
            continue
    return out


def main() -> None:
    pcap = sys.argv[1] if len(sys.argv) > 1 else "/tmp/cap_full.pcap"
    from scapy.all import rdpcap, TCP, Raw  # type: ignore
    packets = rdpcap(pcap)
    streams: dict[tuple, bytearray] = {}
    for pkt in packets:
        if TCP not in pkt or Raw not in pkt:
            continue
        tcp = pkt[TCP]
        sport, dport = tcp.sport, tcp.dport
        # 游戏服端口 8043, 取服务器→客户端方向(sport=8043)
        if sport != 8043:
            continue
        key = (tcp.ack,)
        streams.setdefault(key, bytearray()).extend(bytes(pkt[Raw].load))
    print(f"流数: {len(streams)}, 总字节: {sum(len(v) for v in streams.values())}")

    results: dict[str, dict] = {}
    for buf in streams.values():
        for j in extract_map_bodies(frames_from_stream(bytes(buf))):
            name = str(j.get("MapName") or j.get("MapId") or "?")
            # SC51 优先(含 sourid), SC634 兜底
            if name in results and j["_msg"] == "SC634" \
                    and results[name]["_msg"] == "SC51":
                continue
            results[name] = j

    print(f"\n{'显示名':<14} {'MapId':<14} {'sourid':<8} {'Minimap':<8} 来源")
    print("-" * 72)
    for name in sorted(results):
        j = results[name]
        print(f"{name:<14} {str(j.get('MapId','')):<14} "
              f"{str(j.get('sourid','')):<8} {str(j.get('Minimap','')):<8} {j['_msg']}")


if __name__ == "__main__":
    main()