chuanqi/cq2_sdk/scripts/probe_gateway2.py

"""Try the 3 DNS IPs of gfgame.dhsf.xqhuyu.com directly, on multiple ports,
with a minimal CS 2001 probe. Some gateways reply with an error frame (503)
even for malformed logins — that alone proves the framing is right.

Also try a pure 8-byte or empty probe to see if the gateway sends anything
at all upon connect (banner).
"""
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
from cq2_sdk.transport.framing import encode_frame

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

IPS = ["47.118.169.147", "47.118.169.153", "47.118.172.233"]


def get_creds() -> tuple[str, str]:
    text = Path(r"D:\ProgramFiles\996box\GameBoxDownload\1\env.json").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 attempt(ip: str, port: int, uid: str, token: str, sid: int) -> None:
    label = f"{ip}:{port}"
    try:
        reader, writer = await asyncio.wait_for(asyncio.open_connection(ip, port), timeout=5)
    except Exception as e:
        print(f"{label:24s} connect: {type(e).__name__}")
        return
    try:
        # Send minimal CS 2001
        attach = f"{uid}/{token}/1/0".encode()
        frame = encode_frame(2001, 900000, 0, sid, 0, attach)
        writer.write(frame)
        await writer.drain()
        try:
            data = await asyncio.wait_for(reader.read(4096), timeout=6)
            if data:
                print(f"{label:24s} REPLY {len(data)}B: {data[:60].hex()}")
            else:
                print(f"{label:24s} connection closed by server (0 bytes)")
        except asyncio.TimeoutError:
            print(f"{label:24s} no reply (6s)")
    finally:
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass


async def main() -> None:
    uid, token = get_creds()
    sid = 14414738
    print(f"uid={uid} token_len={len(token)} sid={sid}")
    print()
    for ip in IPS:
        for port in (12000, 8001, 8021):
            await attempt(ip, port, uid, token, sid)
    print()
    print("--- also try raw banner read (connect, send nothing) on 12000 ---")
    try:
        reader, writer = await asyncio.wait_for(asyncio.open_connection(IPS[0], 12000), timeout=5)
        try:
            data = await asyncio.wait_for(reader.read(4096), timeout=4)
            print(f"banner: {data[:60]!r}" if data else "no banner (server waits for client)")
        except asyncio.TimeoutError:
            print("no banner within 4s (normal — server waits)")
        finally:
            writer.close()
    except Exception as e:
        print(f"banner connect failed: {e}")


asyncio.run(main())