chuanqi/cq2_sdk/scripts/probe_gateway.py

"""Find 玖扬复古176 group servers (t=1000514) and attempt TCP login to the
real game gateway: gfgame.dhsf.xqhuyu.com.

Field map (decoded from cached serverlist):
  d = serverId, p = port, s = state(1=open), t = group tag,
  i = gateway host (optional, default gfgame.dhsf.xqhuyu.com),
  x = alt port, n = server name, u = version
"""
import asyncio
import base64
import json
import os
import sys
from pathlib import Path

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

from cq2_sdk.env_loader import load_env
load_env()

from cq2_sdk.transport.tcp import TcpClient

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")

SRVLIST = Path(r"D:\ProgramFiles\996box\GameBoxDownload\1\cq\cache_206923\ResServerlist.txt_206923")
DEFAULT_HOST = "gfgame.dhsf.xqhuyu.com"


def get_creds() -> tuple[str, str]:
    env_p = Path(r"D:\ProgramFiles\996box\GameBoxDownload\1\env.json")
    text = env_p.read_bytes().decode("utf-8", errors="replace")
    idx = text.find("eyJ")
    b64 = text[idx:].rstrip()
    padded = b64 + "=" * (-len(b64) % 4)
    obj, _ = json.JSONDecoder().raw_decode(base64.urlsafe_b64decode(padded).decode("utf-8", errors="replace"))
    return str(obj.get("uid", "")), str(obj.get("token", ""))


async def probe(host: str, port: int) -> bool:
    try:
        r, w = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=5)
        w.close()
        return True
    except Exception:
        return False


async def try_login(host: str, port: int, sid: int, uid: str, token: str, label: str) -> None:
    print(f"--- {label}: {host}:{port} sid={sid} ---")
    tcp = TcpClient(host, port)
    try:
        await tcp.connect()
    except Exception as e:
        print(f"  connect failed: {type(e).__name__}: {e}")
        return
    try:
        attach = f"{uid}/{token}/1/0"
        await tcp.send(2001, recog=900000, param1=0, param2=sid, param3=0, body=attach.encode())
        try:
            hdr, body = await tcp.recv(timeout=8.0)
            print(f"  REPLY msg_id={hdr.msg_id} recog={hdr.recog} p2={hdr.param2} p3={hdr.param3}")
            print(f"  body[:200]: {body[:200]!r}")
            if hdr.msg_id == 529:
                print("  >>> SC 529 LOGIN KEY — TCP LOGIN WORKS!!!")
            elif hdr.msg_id == 503:
                print("  >>> 503 auth failed — but protocol/framing is CORRECT (server understood us)")
            else:
                print(f"  >>> unexpected reply (msg {hdr.msg_id}) — framing may be off")
        except asyncio.TimeoutError:
            print("  TIMEOUT — no reply")
    finally:
        await tcp.close()


async def main() -> None:
    data = json.loads(SRVLIST.read_text(encoding="utf-8"))
    servers = data["serverData"]

    # 1. 玖扬复古176 group (t=1000514)
    jiuyang = [s for s in servers if str(s.get("t")) == "1000514"]
    print(f"玖扬复古176 group servers: {len(jiuyang)}")
    for s in jiuyang[:8]:
        host = s.get("i") or DEFAULT_HOST
        print(f"  sid={s.get('d')} port={s.get('p')} state={s.get('s')} host={host} name={s.get('n')}")

    # 2. User's last server
    mine = [s for s in servers if str(s.get("d")) == "14414738"]
    if mine:
        s = mine[0]
        host = s.get("i") or DEFAULT_HOST
        print(f"\nuser's last server: sid={s['d']} port={s['p']} state={s['s']} host={host}")

    # 3. creds
    uid, token = get_creds()
    print(f"creds: uid={uid} token=***(len={len(token)})")
    print()

    # 4. probe gateway ports
    print("=== probe gfgame.dhsf.xqhuyu.com ===")
    for port in (8001, 8021, 8041, 8061, 8081, 12000):
        ok = await probe(DEFAULT_HOST, port)
        print(f"  :{port}  {'OPEN' if ok else 'closed'}")
    print()

    # 5. TCP login attempts — user's server + first open 玖扬 server
    print("=== TCP login attempts ===")
    if mine:
        s = mine[0]
        host = s.get("i") or DEFAULT_HOST
        await try_login(host, int(s["p"]), int(s["d"]), uid, token, "user's server 14414738")
        print()

    for s in jiuyang:
        if str(s.get("s")) == "1":
            host = s.get("i") or DEFAULT_HOST
            await try_login(host, int(s["p"]), int(s["d"]), uid, token, f"玖扬 {s.get('n')}")
            break


asyncio.run(main())