chuanqi/cq2_sdk/scripts/login_bruteforce.py

"""Brute-force CS 2001 with multiple field-name variants.

The 0-byte close tells us params/encoding is wrong. Try all likely
account/otpType combinations.
"""
import asyncio
import base64
import json
import random
import struct
import sys
from pathlib import Path

sys.path.insert(0, "src")
from cq2_sdk.env_loader import load_env
load_env()
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"))

password = env["token"]  # back to env["token"], the most likely candidate
# also try gameAccount-style account
channel_id = env.get("channel", "110585")
pwd_enc = encode6bit(password.encode()).decode("ascii")
ch_enc = encode6bit(channel_id.encode()).decode("ascii")
main_server_id = 14414736


async def attempt(account, otpType, otpCode_raw, otpCode_send, label):
    if otpCode_raw == "null":
        otp_send = "null"
    elif otpCode_raw == "":
        otp_send = ""
    else:
        otp_send = encode6bit(otpCode_raw.encode()).decode("ascii")
    body = f"{account}/{pwd_enc}/{ch_enc}/{otpType}/{otp_send}"
    frame = struct.pack("<6i", 2001, 900_000, random.randint(100_000_000, 2_100_000_000) & 0xFFFF00FF, main_server_id, 8021, len(body.encode())) + body.encode()
    print(f"--- {label}")
    print(f"  body: {body[:100]}")
    try:
        r, w = await asyncio.wait_for(asyncio.open_connection("gfgame.dhsf.xqhuyu.com", 12000), timeout=5)
    except Exception as e:
        print(f"  connect: {e}"); return
    data = b""
    try:
        w.write(frame); await w.drain()
        data = await asyncio.wait_for(r.read(4096), timeout=6)
    except asyncio.TimeoutError:
        print("  TIMEOUT")
    except Exception as e:
        print(f"  err: {e}")
    finally:
        try:
            w.close(); await w.wait_closed()
        except Exception:
            pass
    if not data:
        print("  0 bytes (closed)")
    elif len(data) >= 24:
        msg_id, *_ = struct.unpack("<6i", data[:24])
        print(f"  REPLY msg_id={msg_id} len={len(data)}B")
        if msg_id == 529:
            print(f"  *** *** SC 529 LOGIN KEY! *** ***")
            print(f"  body: {data[24:][:200]!r}")
            sys.exit(0)
    else:
        print(f"  short {len(data)}B")
    await asyncio.sleep(0.5)


async def main():
    u = env["uid"]
    g = env["gameAccount"]
    k = env["u_key"]
    variants = [
        (u, "0", "0", "0", "uid / otp 0/0"),
        (g, "0", "0", "0", "gameAccount / otp 0/0"),
        (u, "1", "0", "0", "uid / otp 1/0 (SMS)"),
        (g, "1", "0", "0", "gameAccount / otp 1/0"),
        (u, "0", "1", "0", "uid / otp 0/1"),
        (u, k, "0", "0", "uid / otp=u_key/0"),
        (u, k[:4], "0", "0", "uid / otp=u_key[:4]/0"),
        (u, "3", k, "0", "uid / otp 3/u_key"),
        (u, "100", k, "0", "uid / otp 100/u_key"),
        (u, "0", "null", "null", "uid / otp 0/null"),
        (u, "0", "", "empty", "uid / otp 0/(empty)"),
        (u, "0", u, "0", "uid / otp 0/uid"),
        (u, "1", u, "0", "uid / otp 1/uid"),
    ]
    for acct, ot, raw, send, label in variants:
        await attempt(acct, ot, raw, send, label)


asyncio.run(main())