chuanqi/cq2_sdk/scripts/capture_cq2.py

"""Capture cq2 TCP payloads via Npcap. NO BPF — Python-side filtering.

Usage:
  uv run python scripts/capture_cq2.py --duration 120
  uv run python scripts/capture_cq2.py --show cq2_wire.jsonl
"""
import argparse
import json
import struct
import sys
import time

GAME_HOSTS = {"47.118.169.147", "47.118.169.153", "47.118.172.233"}
GAME_PORT = 12000


def make_handler(out: str, counter: dict):
    from scapy.layers.l2 import Ether
    from scapy.layers.inet import IP, TCP

    def on_pkt(pkt):
        try:
            # The WLAN iface reports linktype=1 which confuses scapy's
            # auto-detection; packets arrive as Raw. Re-parse as Ether.
            if not pkt.haslayer(IP):
                try:
                    raw = bytes(pkt)
                    if len(raw) < 34:
                        return
                    # try Ether parse of the raw bytes
                    pkt = Ether(raw)
                    if not pkt.haslayer(IP):
                        return
                except Exception:
                    return
            if not pkt.haslayer(TCP):
                return
            ip, tcp = pkt[IP], pkt[TCP]
            src_ip, dst_ip = ip.src, ip.dst
            sport, dport = int(tcp.sport), int(tcp.dport)
            if not (sport == GAME_PORT or dport == GAME_PORT
                    or src_ip in GAME_HOSTS or dst_ip in GAME_HOSTS):
                return
            payload = bytes(tcp.payload)
            flags = int(tcp.flags)
            seq = int(tcp.seq)
            # Record control packets (FIN/RST/SYN) even without payload —
            # needed to understand the server's close semantics.
            if not payload:
                if flags & 0x01 or flags & 0x04 or flags & 0x02:  # FIN/RST/SYN
                    rec = {
                        "ts": round(time.time(), 3),
                        "src": f"{src_ip}:{sport}",
                        "dst": f"{dst_ip}:{dport}",
                        "len": 0,
                        "flags": flags,
                        "seq": seq,
                        "ctrl": True,
                    }
                    with open(out, "a", encoding="utf-8") as f:
                        f.write(json.dumps(rec) + "\n")
                    fnames = []
                    if flags & 0x01:
                        fnames.append("FIN")
                    if flags & 0x04:
                        fnames.append("RST")
                    if flags & 0x02:
                        fnames.append("SYN")
                    print(f"  [CTRL] {rec['src']:>22} -> {rec['dst']:<22} {'+'.join(fnames)} seq={seq}")
                return
            msg_id = None
            if len(payload) >= 4:
                (mid,) = struct.unpack("<i", payload[:4])
                msg_id = mid
            rec = {
                "ts": round(time.time(), 3),
                "src": f"{src_ip}:{sport}",
                "dst": f"{dst_ip}:{dport}",
                "len": len(payload),
                "flags": flags,
                "seq": seq,
                "msg_id_guess": msg_id,
                "hex": payload.hex(),
            }
            with open(out, "a", encoding="utf-8") as f:
                f.write(json.dumps(rec) + "\n")
            counter["n"] += 1
            m = msg_id
            tag = ""
            if m == 2001:
                tag = "  *** CS 2001 ***"
            elif m == 529:
                tag = "  *** SC 529 ***"
            elif m and 0 < m < 100000:
                tag = f"  (msg {m})"
            print(f"  [{counter['n']:>4}] {rec['src']:>22} -> {rec['dst']:>22} len={rec['len']:>4}{tag}")
        except Exception as e:
            print("pkt err:", e)
    return on_pkt


def capture(out: str, duration: int) -> int:
    from scapy.config import conf
    from scapy.sendrecv import sniff

    dev = conf.iface
    print(f"capturing on: {dev}")
    print(f"python filter: port {GAME_PORT} / game hosts; {duration}s -> {out}")
    print()
    print(">>> NOW: open 996box, login, 玖扬复古176, 启动 <<<")
    print()
    counter = {"n": 0}
    sniff(iface=dev, prn=make_handler(out, counter), timeout=duration, store=False)
    print(f"\ntotal game payloads: {counter['n']} -> {out}")
    return 0


def show(path: str) -> int:
    from pathlib import Path

    p = Path(path)
    if not p.exists():
        print(f"{path} not found", file=sys.stderr)
        return 1
    print(f"=== {path} ===")
    i = 0
    with p.open(encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError:
                continue
            i += 1
            m = rec.get("msg_id_guess")
            tag = ""
            if m == 2001:
                tag = " *** CS 2001 ***"
            elif m == 529:
                tag = " *** SC 529 ***"
            elif m and 0 < m < 100000:
                tag = f" (msg {m})"
            print(f"  [{i:>4}] {rec['src']:>22} -> {rec['dst']:>22} len={rec['len']:>4}{tag}")
            if m == 2001:
                print(f"         HEX: {rec['hex']}")
                try:
                    print(f"         ASCII: {bytes.fromhex(rec['hex'])!r}")
                except Exception:
                    pass
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default="cq2_wire.jsonl")
    ap.add_argument("--duration", type=int, default=120)
    ap.add_argument("--show", default=None)
    args = ap.parse_args()
    if args.show:
        return show(args.show)
    return capture(args.out, args.duration)


if __name__ == "__main__":
    sys.exit(main())