chuanqi/cq2_sdk/scripts/send_probe.py

"""Send one CS 2001 probe frame to the game gateway while capture_cq2.py
listens — validates that our Npcap capture sees the exact bytes on the wire.
"""
import asyncio
import json
import random
import struct
import sys
import base64
from pathlib import Path

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

# Read env.json creds
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"))

account = env["uid"]
pwd_enc = encode6bit(env["token"].encode()).decode("ascii")
ch_enc = encode6bit(env.get("channel", "110585").encode()).decode("ascii")
body = f"{account}/{pwd_enc}/{ch_enc}/0/0"
param1 = random.randint(100_000_000, 2_100_000_000) & 0xFFFF00FF
frame = struct.pack("<6i", 2001, 900_000, param1, 14414736, 8021, len(body.encode())) + body.encode()

print(f"sending {len(frame)}B CS 2001 to gfgame:12000 ...")
print(f"frame hex[:80]: {frame.hex()[:80]}")


async def send():
    r, w = await asyncio.wait_for(
        asyncio.open_connection("gfgame.dhsf.xqhuyu.com", 12000), timeout=5
    )
    try:
        w.write(frame)
        await w.drain()
        print("sent.")
        try:
            data = await asyncio.wait_for(r.read(4096), timeout=3)
            print(f"reply: {len(data) if data else 0} bytes")
        except asyncio.TimeoutError:
            print("no reply (expected — server closes)")
    finally:
        w.close()
        try:
            await w.wait_closed()
        except Exception:
            pass


asyncio.run(send())