chuanqi/cq2_sdk/scripts/probe_game_login.py

"""Decode env.json (robust) and attempt cq2 TCP game login with live credentials.

Strategy:
  1. Read uid/token from env.json (account-level credentials, still alive)
  2. Probe the game gateway for BOTH gameids (206923 = 玖扬复古176 from srvlist,
     218620 = whatever was launched last)
  3. Send CS 2001 login frame with the env.json credentials
  4. See if the server replies with SC 529 (login key) — proving the TCP path works
"""
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")


def decode_env() -> dict:
    p = Path(r"D:\ProgramFiles\996box\GameBoxDownload\1\env.json")
    text = p.read_bytes().decode("utf-8", errors="replace")
    idx = text.find("eyJ")
    if idx < 0:
        idx = text.find("MeyJ")
    if idx < 0:
        return {}
    b64 = text[idx:].rstrip()
    padded = b64 + "=" * (-len(b64) % 4)
    dec = base64.urlsafe_b64decode(padded).decode("utf-8", errors="replace")
    obj, _ = json.JSONDecoder().raw_decode(dec)
    return obj if isinstance(obj, dict) else {}


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, uid: str, token: str, server_id: int, label: str) -> None:
    print(f"--- {label}: {host}:{port} serverId={server_id} ---")
    tcp = TcpClient(host, port)
    try:
        await tcp.connect()
    except Exception as e:
        print(f"  connect failed: {e}")
        return
    try:
        # CS 2001: attach = "account/password/otpType/otpCode"
        attach = f"{uid}/{token}/1/0"
        await tcp.send(2001, recog=900000, param1=0, param2=server_id, 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}")
            print(f"  body[:150]: {body[:150]!r}")
            if hdr.msg_id == 529:
                print("  >>> SUCCESS: SC 529 login key received!")
            elif hdr.msg_id == 503:
                print("  >>> 503 = auth failed (token expired/wrong format) — but TCP protocol WORKS")
        except asyncio.TimeoutError:
            print("  TIMEOUT: no reply (frame format or encryption mismatch)")
    finally:
        await tcp.close()


async def main() -> None:
    env = decode_env()
    uid = str(env.get("uid", ""))
    token = str(env.get("token", ""))
    print(f"credentials: uid={uid} token=***REDACTED***(len={len(token)})")
    print()

    # Gateway candidates
    targets = [
        ("jycq-206923.dhsf.xqhuyu.com", 8001, 14414738, "玖扬复古176 (srvlist serverId)"),
        ("jycq-218620.dhsf.xqhuyu.com", 8001, 0, "last-launched game"),
    ]

    print("=== Port probe ===")
    for host, port, sid, label in targets:
        reachable = await probe(host, port)
        print(f"  {host}:{port}  {'OPEN' if reachable else 'CLOSED'}   ({label})")
    print()

    print("=== TCP login attempts ===")
    for host, port, sid, label in targets:
        if await probe(host, port):
            await try_login(host, port, uid, token, sid, label)
            print()


asyncio.run(main())