chuanqi/protocol_bot/docs/sniff_live_v3.py

"""sniff_live_v3 — 引导序列抓包(2026-09-26, 解锁路线①)。

监听 996盒子 Mac 客户端 与 游戏网关(8001-8099)的 TCP 流, 逐帧落盘 JSONL。
目标: 抓真客户端从出生到"能走动"的完整引导序列(1011/3199 等), 供按序重放。

相对 sniff_live_fixed.py 升级:
1. frame_hex: 48B encode6头+body 的原始字节 → 离线可字节级重放(旧版只有解析字段)
2. body_utf8: body 尽力 UTF-8 解码 → @yindao_chiyao/@chiyao_kaiqi 等引导串肉眼可读
3. ts: 绝对 epoch 时间戳(对齐客户端启动/登录操作), t: 相对秒
4. flow: 四元组保留 → 验证"移动帧独立通道"假设(3011 与 3031 是否同流)

用法(root):
  sudo protocol_bot/.venv/bin/python protocol_bot/docs/sniff_live_v3.py \
      --out /tmp/cap_guide.jsonl --duration 2400
行缓冲实时落盘, 被杀也不丢已收帧。
"""
from __future__ import annotations

import argparse
import json
import struct
import time
from pathlib import Path
from typing import TYPE_CHECKING, Final

from pbot.codec.bitbuf import decode6

if TYPE_CHECKING:
    from collections.abc import Callable
    from typing import TextIO

PORT_LO: Final[int] = 8001
PORT_HI: Final[int] = 8099
HDR_LEN: Final[int] = 48
MAX_BODY: Final[int] = 300000


def split_frames(
    buf: bytes,
) -> tuple[list[tuple[list[int], bytes]], bytes]:
    """从流缓冲切出完整帧: 返回 ([(9int头, body)...], 剩余字节)。

    头 48B encode6 编码; blen 非法或帧不完整即停(等待更多数据)。
    """
    out: list[tuple[list[int], bytes]] = []
    pos = 0
    while len(buf) - pos >= HDR_LEN:
        try:
            h = list(struct.unpack("<9i", decode6(buf[pos : pos + HDR_LEN])))
        except Exception:  # noqa: BLE001 非法头=流边界错位, 截断等待重同步
            break
        blen = h[7]
        if blen < 0 or blen > MAX_BODY or len(buf) - pos < HDR_LEN + blen:
            break
        out.append((h, buf[pos + HDR_LEN : pos + HDR_LEN + blen]))
        pos += HDR_LEN + blen
    return out, buf[pos:]


def body_utf8(body: bytes) -> str:
    """尽力解码引导串; 不可解码部分以 . 替换。"""
    return body.decode("utf-8", errors="replace").replace("\x00", "")


def make_handler(
    fh: TextIO,
    t0: float | None = None,
) -> tuple[Callable[[object], None], dict[str, int]]:
    """构造 scapy prn 回调。fh 为逐行落盘函数(行缓冲)。返回 (handler, 计数器)。"""
    flows: dict[tuple[str, int, str, int], bytes] = {}
    stats = {"frames": 0, "pkts": 0}
    start = t0 if t0 is not None else time.time()

    def on_pkt(pkt: object) -> None:
        tcp = getattr(pkt, "TCP", None)
        ip = getattr(pkt, "IP", None)
        raw = getattr(pkt, "Raw", None)
        if tcp is None or ip is None or raw is None:
            return
        payload = bytes(getattr(raw, "load", b""))
        if not payload:
            return
        sport = int(tcp.sport)
        dport = int(tcp.dport)
        if PORT_LO <= dport <= PORT_HI:
            d = "CS"  # 客户端→服务器(服务器端口在 8001-8099)
        elif PORT_LO <= sport <= PORT_HI:
            d = "SC"
        else:
            return
        stats["pkts"] += 1
        src = str(ip.src)
        dst = str(ip.dst)
        key = (src, sport, dst, dport)
        buf = flows.get(key, b"") + payload
        frames, rest = split_frames(buf)
        now = time.time()
        flow = f"{src}:{sport}>{dst}:{dport}"
        for h, body in frames:
            rec = {
                "t": round(now - start, 2),
                "ts": round(now, 3),
                "dir": d,
                "flow": flow,
                "frame_hex": (
                    # 原始 48B 头在 buf 中已消费, 从解析字段重编码等价;
                    # 这里直接存 48B 头的 encode6 输出 + body, 保证可重放
                    _reencode_header_hex(h) + body.hex()
                ),
                "pre": h[0],
                "msg": h[1],
                "recog": h[2],
                "p1": h[3],
                "p2": h[4],
                "p3": h[5],
                "p6": h[6],
                "blen": h[7],
                "tail": h[8],
                "body_hex": body.hex(),
                "body_utf8": body_utf8(body),
            }
            fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
            stats["frames"] += 1
        flows[key] = rest

    return on_pkt, stats


def _reencode_header_hex(h: list[int]) -> str:
    from pbot.codec.bitbuf import encode6

    return encode6(struct.pack("<9i", *h)).hex()


def main() -> None:
    """主入口: 解析参数, 启动 scapy 抓包循环(需 root)。"""
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default="/tmp/cap_guide.jsonl")  # noqa: S108
    ap.add_argument("--duration", type=int, default=0, help="秒, 0=直到 Ctrl-C")
    a = ap.parse_args()

    from scapy.all import sniff  # noqa: PLC0415  (ruff 版本差异, 保留防误报)

    fh = Path(a.out).open("w", buffering=1)  # noqa: SIM115 行缓冲实时落盘
    handler, stats = make_handler(fh)
    print(  # noqa: T201 抓包工具 CLI 输出
        f"[sniff_v3] ports {PORT_LO}-{PORT_HI} → {a.out} (实时写入)",
        flush=True,
    )
    try:
        sniff(
            filter=f"tcp portrange {PORT_LO}-{PORT_HI}",
            prn=handler,
            store=False,
            timeout=a.duration or None,
        )
    except KeyboardInterrupt:
        pass
    finally:
        fh.close()
    print(f"[done] {stats['frames']} frames / {stats['pkts']} pkts → {a.out}")  # noqa: T201


if __name__ == "__main__":
    main()