chuanqi/cq2_sdk/scripts/login_attempt3.py

"""Full CS 2001 login attempt following cq2-01 protocol exactly:

  1. Fetch watermark Sequence from NetStampHost: engine-game.dhsf.996db.cn/{md5(sid)}.txt
  2. Encode password with Encode6BitBuf (our SDK implementation)
  3. attach = "account/Encode6BitBuf(token)/channel-encoded/otpType/otpCode"
  4. recog=900000, param1=<from sequence table>, param2=sid, param3=port
  5. Send to gateway (gfgame.dhsf.xqhuyu.com:12000 proxy mode)
"""
import asyncio
import base64
import hashlib
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()

import httpx
from cq2_sdk.utils.encoding import encode6bit
from cq2_sdk.transport.framing import encode_frame

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

NETSTAMP_HOST = "http://engine-game.dhsf.996db.cn"
GATEWAY = "gfgame.dhsf.xqhuyu.com"
PROXY_PORT = 12000
SID = 14414738
CHANNEL = "110585"


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 fetch_sequence(sid: int) -> dict | None:
    md5 = hashlib.md5(str(sid).encode()).hexdigest()
    url = f"{NETSTAMP_HOST}/{md5}.txt"
    print(f"netstamp: GET {url}")
    try:
        async with httpx.AsyncClient(timeout=10) as c:
            r = await c.get(url)
            print(f"  -> {r.status_code} len={len(r.text)}")
            if r.status_code == 200 and r.text.strip():
                print(f"  content[:300]: {r.text[:300]}")
                try:
                    return r.json()
                except Exception:
                    return {"raw": r.text}
    except Exception as e:
        print(f"  -> error: {e}")
    return None


async def attempt(uid: str, token: str, seq_val: int | None) -> None:
    # Encode password with Encode6BitBuf
    enc_token = encode6bit(token.encode()).decode("ascii")
    # attach per cq2-01: "account/password(Encode6BitBuf)/channel(encoded)/otpType/otpCode"
    # channel also "encoded" — try Encode6BitBuf on channel too
    enc_channel = encode6bit(CHANNEL.encode()).decode("ascii")
    attach = f"{uid}/{enc_token}/{enc_channel}/1/0"
    print(f"attach ({len(attach)} chars): {uid}/{enc_token[:20]}.../{enc_channel[:12]}.../1/0")

    frame = encode_frame(2001, 900000, seq_val or 0, SID, PROXY_PORT, attach.encode())
    print(f"frame: {len(frame)}B  msg=2001 recog=900000 p1={seq_val} p2={SID} p3={PROXY_PORT}")

    reader, writer = await asyncio.wait_for(asyncio.open_connection(GATEWAY, PROXY_PORT), timeout=5)
    try:
        writer.write(frame)
        await writer.drain()
        data = await asyncio.wait_for(reader.read(4096), timeout=8)
        if data:
            print(f"REPLY {len(data)}B hex[:60]: {data[:60].hex()}")
            # Try parse as frame
            from cq2_sdk.transport.framing import decode_frame
            try:
                hdr, body = decode_frame(data)
                print(f"  msg_id={hdr.msg_id} recog={hdr.recog} p2={hdr.param2}")
                print(f"  body: {body[:150]!r}")
                if hdr.msg_id == 529:
                    print("*** SC 529 — LOGIN KEY RECEIVED! ***")
                elif hdr.msg_id == 503:
                    print("*** 503 auth rejected — framing WORKS, creds may need refresh ***")
            except Exception:
                print("  (does not parse as plaintext frame — may be encrypted)")
        else:
            print("server closed connection (0 bytes)")
    finally:
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass


async def main() -> None:
    uid, token = get_creds()
    print(f"uid={uid} token_len={len(token)}")
    print()
    seq = await fetch_sequence(SID)
    print()
    seq_val = None
    if seq:
        # sequence table format unknown — try common shapes
        if isinstance(seq, dict):
            for k in ("sequence", "seq", "value", "s"):
                if k in seq:
                    try:
                        seq_val = int(seq[k])
                        break
                    except Exception:
                        pass
            if seq_val is None and seq.get("raw"):
                try:
                    seq_val = int(str(seq["raw"]).strip())
                except Exception:
                    pass
        elif isinstance(seq, list) and seq:
            try:
                seq_val = int(seq[0])
            except Exception:
                pass
    print(f"sequence value: {seq_val}")
    print()
    await attempt(uid, token, seq_val)


asyncio.run(main())