"""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()