chuanqi/cq2_sdk/scripts/tcp_semantics_probe.py

"""Full-fidelity TCP behavior analysis of the game gateway's 6-byte zero replies.

Records ALL packets (with TCP flags, seq/ack, timing) — not just payloads —
to answer:
  1. Are the 6B zero packets PSH-flagged DATA, or keepalive probes?
  2. Does the server FIN or RST after them?
  3. What's the exact timing pattern (immediate? delayed?)

Runs capture in-process while sending one CS 2001 probe.
"""
import asyncio
import base64
import json
import random
import struct
import sys
import time
from pathlib import Path

sys.path.insert(0, "src")
from cq2_sdk.utils.encoding import encode6bit

raw = Path(r"D:\ProgramFiles\996box\GameBoxDownload\1\env.json").read_bytes()
text = raw.decode("utf-8", errors="replace")
idx = text.find("eyJ")
b64 = text[idx:].rstrip()
padded = b64 + "=" * ((-len(b64)) % 4)
env = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8", errors="replace"))

uid = str(env["uid"])
pwd_enc = encode6bit(env["token"].encode()).decode("ascii")
ch_enc = encode6bit(env.get("channel", "110585").encode()).decode("ascii")
BODY = f"{uid}/{pwd_enc}/{ch_enc}/1/0"

FLAG_NAMES = {0x01: "FIN", 0x02: "SYN", 0x04: "RST", 0x08: "PSH", 0x10: "ACK", 0x20: "URG"}


def flags_str(f: int) -> str:
    return "+".join(n for b, n in FLAG_NAMES.items() if f & b) or "."


def make_probe() -> bytes:
    p1 = random.randint(100_000_000, 2_100_000_000) & 0xFFFF00FF
    return (
        struct.pack("<6i", 2001, 900000, p1, 14414738, 8021, len(BODY.encode()))
        + BODY.encode()
    )


async def main() -> None:
    from scapy.config import conf
    from scapy.sendrecv import AsyncSniffer

    events: list[dict] = []

    def on_pkt(pkt):
        try:
            # WLAN iface reports linktype=1 which confuses scapy auto-detection;
            # packets arrive as Raw — re-parse as Ether (proven fix from capture_cq2.py)
            if not pkt.haslayer("IP"):
                from scapy.layers.l2 import Ether

                raw_bytes = bytes(pkt)
                if len(raw_bytes) < 34:
                    return
                pkt = Ether(raw_bytes)
                if not pkt.haslayer("IP"):
                    return
            if not pkt.haslayer("TCP"):
                return
            tcp, ip = pkt["TCP"], pkt["IP"]
            payload = bytes(tcp.payload)
            rec = {
                "t": round(time.time(), 3),
                "src": f"{ip.src}:{tcp.sport}",
                "dst": f"{ip.dst}:{tcp.dport}",
                "flags": flags_str(int(tcp.flags)),
                "seq": int(tcp.seq),
                "payload_len": len(payload),
                "payload_hex": payload.hex()[:48] if payload else "",
            }
            events.append(rec)
            pl = f" DATA[{len(payload)}B]={payload.hex()[:32]}" if payload else ""
            print(f"  {rec['t']%1000:7.3f}  {rec['src']:>22} -> {rec['dst']:<22} {rec['flags']:<12} seq={rec['seq']:<10}{pl}")
        except Exception as e:
            print("pkt err:", e)

    sniffer = AsyncSniffer(iface=conf.iface, prn=on_pkt, store=False)
    sniffer.start()
    print("sniffer running; waiting 2s for baseline...")
    await asyncio.sleep(2)

    print(f"\nsending CS 2001 probe to gfgame:12000 ...")
    try:
        r, w = await asyncio.wait_for(asyncio.open_connection("gfgame.dhsf.xqhuyu.com", 12000), timeout=6)
    except Exception as e:
        print(f"connect failed: {e}")
        sniffer.stop()
        return

    sent_at = time.time()
    try:
        w.write(make_probe())
        await w.drain()
        print(f"probe sent at t={sent_at%1000:.3f}; listening 8s for server behavior...")
        await asyncio.sleep(8)
    finally:
        try:
            w.close()
            try:
                await w.wait_closed()
            except Exception:
                pass
        except Exception:
            pass

    await asyncio.sleep(1)
    sniffer.stop()

    # Summary analysis
    print(f"\n=== {len(events)} TCP packets captured ===")
    data_pkts = [e for e in events if e["payload_len"] > 0]
    rst = [e for e in events if "RST" in e["flags"]]
    fin = [e for e in events if "FIN" in e["flags"]]
    from_server = [e for e in events if e["src"].endswith(":12000")]

    print(f"data packets: {len(data_pkts)} (ours: 1 probe)")
    print(f"server->us data: {len([e for e in from_server if e['payload_len']>0])}")
    print(f"RST: {len(rst)}  FIN: {len(fin)}")
    print(f"server packets total: {len(from_server)}")

    zs = [e for e in from_server if e["payload_len"] == 6 and set(bytes.fromhex(e["payload_hex"])) == {0}]
    if zs:
        print(f"\n6B-zero packets from server: {len(zs)}")
        base_t = events[0]["t"]
        for z in zs:
            print(f"  t+{z['t']-base_t:6.3f}s flags={z['flags']} seq={z['seq']}")
    if rst:
        print(f"\nRST packet(s): {[e['flags'] for e in rst]} — server ABORTS the connection")
    elif fin:
        print(f"\nFIN packet(s): {[e['flags'] for e in fin]} — server closes gracefully")


asyncio.run(main())