chuanqi/cq2_sdk/scripts/handshake_probe2.py

"""Handshake probes H3/H5 on the game gateway (fixed, self-contained).

H3: send CS 2001 -> read 6B -> echo it back -> read more
H5: direct connect to :8021 (bypass proxy) -> send CS 2001 -> read
"""
import asyncio
import base64
import json
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(b"110585").decode("ascii")
BODY = f"{uid}/{pwd_enc}/{ch_enc}/0/0"
HOST = "gfgame.dhsf.xqhuyu.com"


def frame() -> bytes:
    return (
        struct.pack("<6i", 2001, 900000, 0x0F3DB000, 14414736, 8021, len(BODY.encode()))
        + BODY.encode()
    )


async def h3_echo() -> None:
    print("=== H3: send 2001 first, wait 6B, echo back, read more ===")
    r, w = await asyncio.wait_for(asyncio.open_connection(HOST, 12000), timeout=6)
    try:
        w.write(frame())
        await w.drain()
        try:
            first = await asyncio.wait_for(r.readexactly(6), timeout=4)
            print(f"  got 6B: {first.hex()}")
            w.write(first)
            await w.drain()
            print("  echoed back")
            try:
                data = await asyncio.wait_for(r.readexactly(6), timeout=4)
                print(f"  second reply: {data.hex()}")
            except (TimeoutError, asyncio.TimeoutError):
                print("  silence after echo")
            except asyncio.IncompleteReadError:
                print("  closed after echo")
        except (TimeoutError, asyncio.TimeoutError):
            print("  no 6B within 4s")
        except asyncio.IncompleteReadError as e:
            print(f"  closed, partial {len(e.partial)}B: {e.partial.hex()}")
    finally:
        w.close()
        try:
            await w.wait_closed()
        except Exception:
            pass


async def h5_direct() -> None:
    print("=== H5: direct :8021 bypass proxy, send 2001 ===")
    try:
        r, w = await asyncio.wait_for(asyncio.open_connection(HOST, 8021), timeout=8)
    except Exception as e:
        print(f"  connect :8021 failed: {type(e).__name__}")
        return
    try:
        w.write(frame())
        await w.drain()
        print("  sent 102B to :8021")
        try:
            data = await asyncio.wait_for(r.readexactly(6), timeout=6)
            print(f"  reply 6B: {data.hex()}")
            try:
                rest = await asyncio.wait_for(r.read(4096), timeout=3)
                if rest:
                    print(f"  more: {len(rest)}B {rest.hex()[:100]}")
            except Exception:
                pass
        except asyncio.IncompleteReadError as e:
            print(f"  closed, partial {len(e.partial)}B: {e.partial.hex()}")
        except (TimeoutError, asyncio.TimeoutError):
            print("  no reply (6s)")
    finally:
        w.close()
        try:
            await w.wait_closed()
        except Exception:
            pass


async def main() -> None:
    await h3_echo()
    print()
    await h5_direct()


asyncio.run(main())