chuanqi/cq2_sdk/scripts/npcap_capture.py

"""Capture TCP traffic between cq2.exe and gfgame.dhsf.xqhuyu.com:12000
via Npcap on the loopback adapter. Zero process injection. Zero ban risk.

Usage (after Npcap is installed and game is running):
    uv run python scripts/npcap_capture.py --out capture.pcap

Output: standard .pcap (Wireshark-compatible). Then we open it in Python
(scapy) to extract the cq2 <-> game-server TCP stream and look for the
CS 2001 login frame.
"""
import argparse
import asyncio
import os
import sys
import time
from pathlib import Path

sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir, "src"))


def require_pyshark():
    try:
        import pyshark  # noqa
    except ImportError:
        print("pyshark not installed. run: uv add pyshark", file=sys.stderr)
        sys.exit(1)


async def capture(out: Path, duration: int, host_filter: str, port: int) -> int:
    require_pyshark()
    import pyshark

    # Npcap loopback adapter captures ALL local TCP (127.0.0.1)
    # We filter for: TCP port == 12000 (cq2 game gateway)
    # OR: TCP containing dhsf.xqhuyu.com (server IP resolved to local)
    bpf = f"tcp port {port}"
    if host_filter:
        bpf = f"tcp and (port {port} or host {host_filter})"

    print(f"[npcap] capturing on loopback: '{bpf}'")
    print(f"[npcap] writing to {out} for {duration}s")
    print(f"[npcap] >>>  NOW: open 996box, login, select 玖扬复古176, click 启动  <<<")

    # pyshark LiveCapture uses tshark under the hood; needs Npcap installed
    cap = pyshark.LiveCapture(
        interface=r"\Device\NPF_Loopback",  # Npcap loopback; first/only loopback
        bpf_filter=bpf,
        output_file=str(out),
    )
    end = time.time() + duration
    cap.start()
    try:
        # cap.consume() in a thread; we wait then stop
        while time.time() < end:
            await asyncio.sleep(1)
            print(f"  ... {int(end - time.time())}s left", end="\r")
    finally:
        cap.stop()
    print()
    print(f"[npcap] done -> {out}")
    return 0


def parse_pcap_for_cs2001(pcap_path: Path) -> list[dict]:
    """Read the pcap and dump cq2 <-> game server TCP segments.

    We need to identify the CS 2001 first frame: 24-byte header, msg_id=2001
    in first int32 little-endian. If we see frames at all we know the
    network path is reachable.
    """
    import pyshark

    print(f"\n[parse] reading {pcap_path}")
    cap = pyshark.FileCapture(str(pcap_path), display_filter="tcp")
    out = []
    for i, pkt in enumerate(cap):
        try:
            ts = pkt.sniff_time.isoformat()
            src = pkt.ip.src + ":" + pkt.tcp.srcport
            dst = pkt.ip.dst + ":" + pkt.tcp.dstport
            payload_hex = bytes.fromhex(pkt.tcp.payload.replace(":", "")).hex() if pkt.tcp.payload else ""
            # Try parse header
            if len(payload_hex) >= 8:
                msg_id = int.from_bytes(bytes.fromhex(payload_hex[:8]), "little") & 0xFFFFFFFF
            else:
                msg_id = None
            entry = {
                "i": i, "ts": ts, "src": src, "dst": dst,
                "len": len(bytes.fromhex(payload_hex)) // 2 if payload_hex else 0,
                "msg_id_guess": msg_id,
                "hex": payload_hex,
            }
            out.append(entry)
        except Exception as e:
            print(f"  pkt {i} err: {e}")
        if i > 0 and i % 200 == 0:
            print(f"  parsed {i} packets...")
    return out


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--out", default="cq2_capture.pcap")
    parser.add_argument("--duration", type=int, default=120, help="seconds to capture")
    parser.add_argument("--host", default="dhsf.xqhuyu.com")
    parser.add_argument("--port", type=int, default=12000)
    parser.add_argument("--parse", action="store_true", help="parse pcap after capture")
    args = parser.parse_args()

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)

    rc = asyncio.run(capture(out, args.duration, args.host, args.port))
    if rc != 0:
        return rc

    if args.parse:
        pkts = parse_pcap_for_cs2001(out)
        print(f"\n=== {len(pkts)} TCP segments with payload ===")
        for p in pkts[:30]:
            print(f"  {p['ts']}  {p['src']:>22}  ->  {p['dst']:>22}  "
                  f"len={p['len']:>4}  msg_id?={p['msg_id_guess']}  "
                  f"hex[:40]={p['hex'][:40]}")
        # Look for any msg_id == 2001
        cs2001 = [p for p in pkts if p["msg_id_guess"] == 2001]
        if cs2001:
            print(f"\n*** Found {len(cs2001)} CS 2001 frames! ***")
            for c in cs2001[:3]:
                print(f"  hex: {c['hex']}")
        else:
            print("\n(no msg_id=2001 found in payload headers)")
    return 0


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