chuanqi/sniff_live.py

# -*- coding: utf-8 -*-
"""
sniff_live.py —— 等待图交互抓包: 监听 996盒子真实客户端 与 游戏网关的 TCP 流,
完整记录 CS/SC 帧(时间戳+方向+msgId+头字段+body), 用于对照找出"进新手村"缺的帧。

用法(需要 sudo):
  sudo protocol_bot/.venv/bin/python sniff_live.py --out /tmp/cap.jsonl
过滤: 目标/来源端口 8001-8099 的 TCP 载荷, 按流缓冲, 帧格式=48B encode6头+body。
"""
import argparse
import json
import socket
import struct
import sys
import time

sys.path.insert(0, "protocol_bot")
sys.path.insert(0, "protocol_bot_backup")
from encode6bitbuf_final import encode6, decode6  # noqa: E402


def sniff(port_lo, port_hi, out_path, duration):
    from scapy.all import sniff as sc_sniff, TCP, IP, Raw  # type: ignore
    flows = {}
    t0 = time.time()
    fh = open(out_path, "w", buffering=1)     # 行缓冲实时落盘
    n = [0]

    def on_pkt(pkt):
        if TCP not in pkt or Raw not in pkt or IP not in pkt:
            return
        sport, dport = pkt[TCP].sport, pkt[TCP].dport
        if not (port_lo <= sport <= port_hi or port_lo <= dport <= port_hi):
            return
        payload = bytes(pkt[Raw].load)
        if not payload:
            return
        # 方向判定修正(2026-09-23): macOS 客户端临时端口 49152+ 全部 >=8001,
        # 旧规则 "dport>=port_lo" 会把服务器下行帧误标 CS。
        # 正确: 服务器端口固定 8001-8099 → 该侧端口在范围内的=服务器侧:
        #   客户端→服务器(dport 在 8001-8099)= 真上行 "CS"
        #   服务器→客户端(sport 在 8001-8099)= 真下行 "SC"
        if port_lo <= dport <= port_hi:
            d = "CS"
        elif port_lo <= sport <= port_hi:
            d = "SC"
        else:
            return
        key = (pkt[IP].src, sport, pkt[IP].dst, dport, d)
        buf = flows.get(key, b"") + payload
        pos = 0
        while len(buf) - pos >= 48:
            try:
                h = struct.unpack("<9i", decode6(buf[pos:pos + 48]))
            except Exception:
                break
            blen = h[7]
            if blen < 0 or blen > 300000 or len(buf) - pos < 48 + blen:
                break
            body = buf[pos + 48: pos + 48 + blen]
            fh.write(json.dumps({
                "t": round(time.time() - t0, 2), "dir": d,
                "msg": h[1], "recog": h[2], "p1": h[3], "p2": h[4],
                "p3": h[5], "p6": h[6], "tail": h[8], "blen": blen,
                "body_hex": body.hex(),  # 全量不截断(旧 2048 截断吞了进图大帧)
            }) + "\n")
            n[0] += 1
            pos += 48 + blen
        flows[key] = buf[pos:]

    print(f"[sniff] ports {port_lo}-{port_hi} → {out_path} (实时写入)", flush=True)
    try:
        sc_sniff(filter=f"tcp portrange {port_lo}-{port_hi}", prn=on_pkt,
                 store=False, timeout=duration or None)
    except KeyboardInterrupt:
        pass
    fh.close()
    print(f"[done] {n[0]} frames → {out_path}")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default="/tmp/cap.jsonl")
    ap.add_argument("--lo", type=int, default=8001)
    ap.add_argument("--hi", type=int, default=8099)
    ap.add_argument("--duration", type=int, default=0)
    a = ap.parse_args()
    sniff(a.lo, a.hi, a.out, a.duration)