chuanqi/cq2_sdk/scripts/socks5_probe.py

"""Test the SOCKS5 hypothesis on port 12000.

Evidence: cq2/ ships libsocks5-x64_v2.dll + libsocks5-x86_v2.dll.
serverlist 'x' field (12000) = proxy port; Lua connects to proxyport and the
C++ netClient (with libsocks5) likely does a SOCKS5 handshake transparently.

Probe:
  1. SOCKS5 greeting (05 01 00) -> expect (05 00)
  2. SOCKS5 CONNECT to backend game server (gfgame IP : 8021)
  3. If tunnel established -> send CS 2001 through it
  4. Read reply
"""
import asyncio
import base64
import json
import random
import struct
import sys
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"

PROXY = ("gfgame.dhsf.xqhuyu.com", 12000)
BACKEND_IP = "47.118.169.153"
BACKEND_PORT = 8021
SID = 14414738


def socks5_greeting() -> bytes:
    return bytes([0x05, 0x01, 0x00])  # VER5, 1 method, NO-AUTH


def socks5_connect(ip: str, port: int) -> bytes:
    # VER CMD RSV ATYP IPv4(4) PORT(2BE)
    parts = ip.split(".")
    return bytes([0x05, 0x01, 0x00, 0x01] + [int(p) for p in parts] + list(struct.pack(">H", port)))


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


async def read_some(r, n=64, timeout=4.0) -> bytes:
    return await asyncio.wait_for(r.read(n), timeout=timeout)


async def main():
    print(f"connecting to proxy {PROXY} ...")
    r, w = await asyncio.wait_for(asyncio.open_connection(*PROXY), timeout=6)
    try:
        # Step 1: SOCKS5 greeting
        w.write(socks5_greeting())
        await w.drain()
        print("sent greeting: 05 01 00")
        resp = await read_some(r, 2)
        print(f"greeting reply: {resp.hex()} ({len(resp)}B)")
        if len(resp) >= 2 and resp[0] == 0x05:
            print("*** SOCKS5 CONFIRMED! server speaks SOCKS5 ***")
            # Step 2: CONNECT to backend
            req = socks5_connect(BACKEND_IP, BACKEND_PORT)
            w.write(req)
            await w.drain()
            print(f"sent CONNECT to {BACKEND_IP}:{BACKEND_PORT}: {req.hex()}")
            resp2 = await read_some(r, 16)
            print(f"connect reply: {resp2.hex()} ({len(resp2)}B)")
            if len(resp2) >= 4 and resp2[1] == 0x00:
                print("*** TUNNEL ESTABLISHED! sending CS 2001 ... ***")
                w.write(frame_2001())
                await w.drain()
                print("sent 2001 through tunnel")
                # read game reply
                total = b""
                import time
                t0 = time.time()
                while time.time() - t0 < 8:
                    try:
                        chunk = await asyncio.wait_for(r.read(4096), timeout=2)
                        if not chunk:
                            print("tunnel closed after", len(total), "bytes")
                            break
                        total += chunk
                        print(f"  +{len(chunk)}B: {chunk[:40].hex()}")
                        if len(total) >= 24:
                            mid, recog, p1, p2, p3, blen = struct.unpack("<6i", total[:24])
                            print(f"  *** GAME FRAME: msg_id={mid} recog={recog} p2={p2} p3={p3} blen={blen} ***")
                            if 0 < blen < 4096 and len(total) >= 24 + blen:
                                print(f"  body: {total[24:24+blen]!r}")
                            if mid == 529:
                                print("*** *** *** SC 529 LOGIN KEY — FULL TCP LOGIN WORKS *** *** ***")
                            break
                    except asyncio.TimeoutError:
                        if total:
                            break
            else:
                print(f"CONNECT rejected (reply code {resp2[1] if len(resp2)>1 else '?'})")
    except asyncio.IncompleteReadError as e:
        print(f"connection closed during handshake; partial={e.partial.hex() if e.partial else ''}")
    finally:
        w.close()
        try:
            await w.wait_closed()
        except Exception:
            pass


asyncio.run(main())