chuanqi/protocol_bot/docs/sniff_live_v4.py

"""sniff_live_v4 — 全端口自动发现式抓包(2026-09-26)。

背景: v3 按 8001-8099 端口过滤, 但本次真客户端会话零命中(历史端口假设失效)。
v4 策略: 去掉端口/接口假设, 靠协议自身特征发现游戏流:
  - encode6 每输出字节都在 [0x3C,0x7B] → 48B 帧头 = 48 连中, 误报率 2^-96
  - 全接口抓 (en1 默认路由 + lo0 本地代理 + utun 隧道)
  - 非 TCP(IP/UDP) 只记首包 flow 元数据(拓扑发现), TCP 流自动帧同步

输出 JSONL, 每行 kind:
  frame: 完整游戏帧(frame_hex 可重放, 含 48B 头+body)
  syn:   TCP SYN(新连接发现)
  flow:  每条流首包元数据(端口/接口/方向)

用法(root):
  sudo protocol_bot/.venv/bin/python protocol_bot/docs/sniff_live_v4.py \
      --out /tmp/cap_guide4.jsonl --duration 1800
"""
from __future__ import annotations

import argparse
import json
import struct
import subprocess
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

HDR_LEN: Final[int] = 48
MAX_BODY: Final[int] = 60000
ENC_LO: Final[int] = 0x3C  # encode6 输出偏置
ENC_HI: Final[int] = 0x7B  # +63
SYNC_SCAN_CAP: Final[int] = 8192  # 未同步流最多扫描前 8KB
INERT_CAP: Final[int] = 65536  # 未同步流缓冲超 64KB → 放弃该流

_ALPHABET: Final[bytes] = bytes(range(ENC_LO, ENC_HI + 1))


def _is_encoded_header(chunk: bytes) -> bool:
    """48B 全部落在 encode6 字母表 → 候选帧头。"""
    return all(ENC_LO <= b <= ENC_HI for b in chunk)


def try_sync(buf: bytes) -> int:
    """在 buf 里找第一个有效帧偏移。返回 >=0 偏移, -1 未找到。"""
    if len(buf) < HDR_LEN:
        return -1
    scan_end = min(len(buf), SYNC_SCAN_CAP) - HDR_LEN
    for i in range(scan_end + 1):
        chunk = buf[i : i + HDR_LEN]
        if not _is_encoded_header(chunk):
            continue
        h = struct.unpack("<9i", decode6(chunk))
        blen = h[7]
        if 0 <= blen <= MAX_BODY:
            return i
    return -1


def split_frames(buf: bytes) -> tuple[list[tuple[list[int], bytes]], bytes]:
    """从已同步流缓冲切帧: 返回 ([(9int头, body)...], 剩余字节)。"""
    out: list[tuple[list[int], bytes]] = []
    pos = 0
    while len(buf) - pos >= HDR_LEN:
        chunk = buf[pos : pos + HDR_LEN]
        if not _is_encoded_header(chunk):
            break
        h = list(struct.unpack("<9i", decode6(chunk)))
        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 _reencode_header_hex(h: list[int]) -> str:
    from pbot.codec.bitbuf import encode6

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


class FlowState:
    """单条 TCP 流的切帧状态机。"""

    __slots__ = ("buf", "frames", "inert", "synced")

    def __init__(self) -> None:  # noqa: D107 简单状态容器
        self.buf = b""
        self.synced = False
        self.inert = False
        self.frames = 0

    def feed(self, payload: bytes) -> list[tuple[list[int], bytes]]:
        """喂入流数据, 返回切出的完整帧列表。"""
        if self.inert:
            return []
        self.buf += payload
        if not self.synced:
            off = try_sync(self.buf)
            if off < 0:
                if len(self.buf) > INERT_CAP:
                    self.inert = True  # 永久放弃: TLS/图片/视频流
                    self.buf = b""
                return []
            self.buf = self.buf[off:]
            self.synced = True
        frames, rest = split_frames(self.buf)
        if not frames and len(rest) == len(self.buf) and len(rest) >= HDR_LEN:
            # 已同步态却整包解析失败 → 流边界错位, 回退重新同步
            self.synced = False
            off = try_sync(rest)
            if off >= 0:
                self.buf = rest[off:]
                self.synced = True
                frames, rest = split_frames(self.buf)
            else:
                self.buf = rest
            return frames
        self.buf = rest
        self.frames += len(frames)
        return frames


def make_handler(
    fh: TextIO, t0: float | None = None
) -> tuple[Callable[[object], None], dict[str, int]]:
    """构造 scapy prn 回调。返回 (handler, 计数器)。"""
    flows: dict[tuple[str, int, str, int], FlowState] = {}
    seen_flows: set[tuple[str, int, str, int]] = set()
    local_ips = {
        s.strip()
        for s in subprocess.run(  # noqa: PLW1510 本机 ifconfig, 无外部输入
            ["/sbin/ifconfig"], capture_output=True, text=True
        ).stdout.splitlines()
        if "inet " in s
    }
    local_ips = {ln.split()[1] for ln in local_ips}
    stats = {"frames": 0, "pkts": 0, "syn": 0}
    start = t0 if t0 is not None else time.time()

    def emit(rec: dict[str, object]) -> None:
        fh.write(json.dumps(rec, ensure_ascii=False) + "\n")

    def on_pkt(pkt: object) -> None:
        from scapy.all import IP as SC_IP
        from scapy.all import TCP as SC_TCP
        from scapy.all import Raw as SC_Raw

        try:
            ip = pkt[SC_IP]  # pyright: ignore[reportIndexIssue]
            tcp = pkt[SC_TCP]  # pyright: ignore[reportIndexIssue]
        except (KeyError, IndexError, TypeError):
            return
        stats["pkts"] += 1
        src, dst = str(ip.src), str(ip.dst)
        now = round(time.time() - start, 2)
        sport, dport = int(tcp.sport), int(tcp.dport)
        key = (src, sport, dst, dport)
        flags = int(tcp.flags)
        if flags & 0x02 and not flags & 0x10:  # SYN 无 ACK
            stats["syn"] += 1
            emit({
                "kind": "syn", "t": now, "flow": f"{src}:{sport}>{dst}:{dport}",
            })
        if key not in seen_flows:
            seen_flows.add(key)
            emit({
                "kind": "flow", "t": now,
                "flow": f"{src}:{sport}>{dst}:{dport}",
            })
        if SC_Raw not in pkt:  # pyright: ignore[reportOperatorIssue]
            return
        payload = bytes(pkt[SC_Raw].load)  # pyright: ignore[reportIndexIssue]
        if not payload:
            return
        st = flows.setdefault(key, FlowState())
        # 方向: 本机地址作源 = 客户端→服务器 = CS
        local_side = src in local_ips or src.startswith("127.")
        for h, body in st.feed(payload):
            emit({
                "kind": "frame", "t": now,
                "dir": "CS" if local_side else "SC",
                "flow": f"{src}:{sport}>{dst}:{dport}",
                "frame_hex": _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),
            })
            stats["frames"] += 1

    return on_pkt, stats


def main() -> None:
    """主入口: 全接口抓包(需 root)。"""
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default="/tmp/cap_guide4.jsonl")  # noqa: S108
    ap.add_argument("--duration", type=int, default=0, help="秒, 0=直到 Ctrl-C")
    ap.add_argument("--ifaces", default="en1,lo0,utun0",
                    help="逗号分隔抓包接口")
    a = ap.parse_args()

    from scapy.all import sniff

    ifaces = [s.strip() for s in a.ifaces.split(",") if s.strip()]
    fh = Path(a.out).open("w", buffering=1)  # noqa: SIM115 行缓冲实时落盘
    handler, stats = make_handler(fh)
    print(f"[sniff_v4] ifaces={ifaces} all-ports → {a.out}", flush=True)  # noqa: T201
    try:
        sniff(
            iface=ifaces,
            filter="ip",
            prn=handler,
            store=False,
            timeout=a.duration or None,
        )
    except KeyboardInterrupt:
        pass
    finally:
        fh.close()
    print(  # noqa: T201
        f"[done] {stats['frames']} frames / {stats['syn']} syn / "
        f"{stats['pkts']} pkts → {a.out}"
    )


if __name__ == "__main__":
    main()