chuanqi/cq2_sdk/scripts/hyp_test.py

"""Two-hypothesis test for the CS 2001 rejection:

H-A (watermark): In proxy-port mode, param1 MUST come from the NetStamp
     sequence table (server validates it). Our random param1 gets rejected.
     -> Try harder to fetch the sequence: uppercase MD5, no-.txt, different
        serverID encodings, alternate hosts.

H-B (32-byte header): The doc says '24-byte header becomes 32 when encrypted'.
     Maybe even with SData=0 the header is padded to 32 bytes on the wire.
     -> Send 24B header + 8 zero bytes + body (=110B total) and see if the
        server response differs.
"""
import asyncio
import base64
import hashlib
import json
import random
import struct
import sys
from pathlib import Path

sys.path.insert(0, "src")
from cq2_sdk.utils.encoding import encode6bit

import httpx

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"

HOST_IP = "47.118.172.233"
PROXY_PORT = 12000
SID = 14414738
SPORT = 8021

NS_BASE = "https://engine-game.dhsf.996db.cn/list/"


async def fetch_netstamp_variants() -> int | None:
    """Try every plausible md5/url variant for the sequence table."""
    print("=== H-A: NetStamp variants ===")
    cands: list[tuple[str, str]] = []
    for sid in [SID, 14414736, 14414739, 14414741, 14301060]:
        s = str(sid)
        md5_lower = hashlib.md5(s.encode()).hexdigest()
        md5_upper = md5_lower.upper()
        for tag, h in [("lower", md5_lower), ("upper", md5_upper)]:
            cands.append((f"sid{sid}-{tag}.txt", f"{NS_BASE}{h}.txt"))
            cands.append((f"sid{sid}-{tag}-noext", f"{NS_BASE}{h}"))
    # maybe it's md5 of the whole URL path or the port
    cands.append(("port8021", f"{NS_BASE}{hashlib.md5(b'8021').hexdigest()}.txt"))
    cands.append(("proxyport12000", f"{NS_BASE}{hashlib.md5(b'12000').hexdigest()}.txt"))
    # maybe list endpoint itself
    cands.append(("index", NS_BASE))
    cands.append(("index-no-slash", NS_BASE.rstrip("/")))

    async with httpx.AsyncClient(timeout=6, follow_redirects=True) as c:
        for label, url in cands:
            try:
                r = await c.get(url)
                if r.status_code == 200 and r.text.strip():
                    txt = r.text.strip()
                    print(f"  *** HIT {label}: {url}")
                    print(f"      content[:200]: {txt[:200]}")
                    try:
                        data = json.loads(txt)
                        # Lua: index = L8Bit(UnixTime)+1 ; param1 = Sequence[index]
                        seq_list = data.get("Sequence") or []
                        unix = data.get("UnixTime") or 0
                        if seq_list:
                            i8 = unix & 0xFF
                            idx_seq = (i8 % len(seq_list)) if seq_list else 0
                            val = int(seq_list[idx_seq])
                            print(f"      Sequence[{idx_seq}] = {val} (UnixTime low8={i8})")
                            return val
                    except Exception as e:
                        print(f"      parse err: {e}")
                else:
                    print(f"  {r.status_code} {label}")
            except Exception as e:
                print(f"  ERR {label}: {type(e).__name__}")
    return None


async def send_frame(frame: bytes, label: str) -> None:
    try:
        r, w = await asyncio.wait_for(asyncio.open_connection(HOST_IP, PROXY_PORT), timeout=5)
    except Exception as e:
        print(f"  {label}: connect {type(e).__name__}")
        return
    try:
        w.write(frame)
        await w.drain()
        total = b""
        t0 = asyncio.get_event_loop().time()
        while asyncio.get_event_loop().time() - t0 < 4:
            try:
                chunk = await asyncio.wait_for(r.read(4096), timeout=1.2)
                if not chunk:
                    break
                total += chunk
                if len(total) >= 6:
                    break
            except asyncio.TimeoutError:
                if total:
                    break
        if total:
            note = ""
            if total != b"\x00" * 6:
                note = "  *** DIFFERENT FROM 6B-ZERO !!! ***"
            print(f"  {label}: reply {len(total)}B {total.hex()}{note}")
        else:
            print(f"  {label}: no reply")
    finally:
        w.close()
        try:
            await w.wait_closed()
        except Exception:
            pass


async def main():
    # H-A: try to get a valid watermark
    seq_val = await fetch_netstamp_variants()
    print()

    # H-B: 32-byte header variant
    print("=== H-B: frame format variants ===")
    variants = []

    # v1: baseline (what we've been sending) — 24B header
    p1_rand = random.randint(100_000_000, 2_100_000_000) & 0xFFFF00FF
    variants.append((
        struct.pack("<6i", 2001, 900000, p1_rand, SID, SPORT, len(BODY.encode())) + BODY.encode(),
        "v1 24B-header random-p1 (baseline)",
    ))

    # v2: 24B header + 8 zero bytes padding (32B header hypothesis)
    variants.append((
        struct.pack("<6i", 2001, 900000, p1_rand, SID, SPORT, len(BODY.encode()))
        + b"\x00" * 8
        + BODY.encode(),
        "v2 32B-header (8 zero pad)",
    ))

    # v3: 24B header + 8 bytes = SData(0) repeated as int32 x2
    variants.append((
        struct.pack("<6i", 2001, 900000, p1_rand, SID, SPORT, len(BODY.encode()))
        + struct.pack("<ii", 0, 0)
        + BODY.encode(),
        "v3 32B-header (int32 0,0)",
    ))

    # v4: watermark = 0 exactly (not random)
    variants.append((
        struct.pack("<6i", 2001, 900000, 0, SID, SPORT, len(BODY.encode())) + BODY.encode(),
        "v4 24B-header p1=0",
    ))

    # v5: if we got a netstamp value, use it
    if seq_val is not None:
        variants.append((
            struct.pack("<6i", 2001, 900000, seq_val, SID, SPORT, len(BODY.encode())) + BODY.encode(),
            f"v5 24B-header p1=netstamp({seq_val})",
        ))

    for frame, label in variants:
        await send_frame(frame, label)
        await asyncio.sleep(0.5)
    print()
    print("Baseline reply is 6B zeros + FIN. Any DIFFERENT reply = breakthrough.")


asyncio.run(main())