chuanqi/protocol_bot_backup/cq2_pure_enter.py

# -*- coding: utf-8 -*-
"""
cq2_pure_enter.py —— 纯协议全链路: 登录 → 创角/选角 → 进游戏 (无游戏客户端)
==========================================================================
2026-09-19 抓包逆向实证 (z52full.pcap 等, 38区/52区双样本验证):

  HTTP:  登录(极验滑块)→JWT → EnterGameClient(zone) → 平台记录
  登录服(端口因区而异, 38区=8041, 52区=8001):
    CS2001 body=uid/encode6(jwt)/encode6(channel)/1/0   [pre字段随机, tail=0]
      → SC529 body=encode6(key32/...)                    会话密钥
    CS104 body=encode6(key32) [tail=1] → SC530(sid在pre字段)
    CS100 body=uid/sid/zone/channel/gameid/key32 [tail=2] → SC520 角色列表JSON
    CS101 body={"server_name":zone,"chr":"名","secsdk":"0","job":0,"sex":0,
                "account":uid,"sdkid":channel}           创角(角色已存在→SC522)
      → SC521 body="角色名/UserID"                        进区票据(UserID=zone+00+ts+0000+rand)
    CS103 body="uid/角色名" [p3=sid]                       选角
      → SC525 body="127.0.0.1/游戏端口"                   游戏网关端口
  游戏服(登录服端口+2, 38区=8043, 52区=8003):
    CS20 body="*#uid/名/UserID/sid/1/0/channel/0/1/1/1/1"
       头=(pre=900000, id=20, recog=随机, p1=zone, p2=游戏端口, p3=sid)
       ★ 流首直接encode6, 无0x00前缀(38区样本的0x00是TCP粘包残渣)
    → SC游戏数据(zlib) + 心跳CS60000(recog=unix时间戳, p3=sid, 2秒间隔)持续应答

用法:
    .venv-mac/bin/python cq2_pure_enter.py --zone 14459329 [--name 名字] [--wait 60] [--auto-heal]
"""
import argparse
import json
import random
import socket
import struct
import sys
import time
import threading
import os
import zlib

import requests as rq
import urllib3
urllib3.disable_warnings()

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)

from encode6bitbuf_final import encode6, decode6  # noqa: E402
import login_996pc as L  # noqa: E402

GAME_ID = 206923
GW_HOST = "s{zone}-jycq-206923.dhsf.xqhuyu.com"
CHANNEL = "110585"
UID = 2001849239  # 默认值; 实际以登录响应 userInfo.id 为准(login_gate 的 uid 参数)
LOGIN_PORT = 8001


def box_auth_header(method, path, body, tok=""):
    import hashlib, hmac, string
    nonce = "".join(random.choice(string.digits + string.ascii_uppercase) for _ in range(32))
    ts = str(int(time.time() * 1000))
    canonical = "\n".join([method, path, ts, nonce, "pc", L.DEVICE_ID,
                           "5.1.1.9007", tok, L.UA]) + "\n" + body + "\n"
    sig = hmac.new(b"BB0AAAC906FC4B7CADB116A213D46689",
                   canonical.encode(), hashlib.sha256).hexdigest()
    return f"BOXAPP-SHA256 NONCE={nonce};TIMESTAMP={ts};SIGNATURE={sig};"


def enter_game(session, jwt, zone_id):
    device_info = json.dumps({"type": "pc", "model": "pc", "os": "Win10",
                              "width": 1920, "height": 1032, "lang": "zh-CN",
                              "id": "43b2d8df251be893bf57a67560b1e74e",
                              "net_type": "unknown", "mac": L.DEVICE_ID,
                              "ua": L.UA}, separators=(",", ":"))
    body = json.dumps({"box_game_id": GAME_ID, "serverid": zone_id,
                       "device_info": device_info, "appid": "GAjWNTqNpOVx"},
                      separators=(",", ":"))
    h = L.box_headers(jwt)
    h["Content-Type"] = "application/json;charset=UTF-8"
    h["X-BOX-AUTHORIZATION"] = box_auth_header(
        "POST", "/v1/homeapi/gameindex/EnterGameClient", body, jwt)
    r = session.post("https://pc-api.996hezi.com/v1/homeapi/gameindex/EnterGameClient",
                     data=body.encode(), headers=h, timeout=25, verify=False)
    return r.json()


def recv_frames(sock, timeout=5):
    """收满 timeout 秒(墙钟)或连接断开为止, 解析出全部帧。"""
    deadline = time.time() + timeout
    buf = b""
    while time.time() < deadline:
        sock.settimeout(max(0.2, deadline - time.time()))
        try:
            d = sock.recv(65536)
            if not d:
                break
            buf += d
        except (socket.timeout, ConnectionResetError, OSError):
            break
    out, pos = [], 0
    while pos + 48 <= len(buf):
        try:
            m = struct.unpack("<9i", decode6(buf[pos:pos + 48]))
        except Exception:
            break
        out.append((m, buf[pos + 48:pos + 48 + m[7]]))
        pos += 48 + m[7]
    return out


def gateway_ports(zone_id, login_port):
    return login_port, login_port + 2


def login_gate(zone_id, jwt, name, login_port, uid):
    g = socket.create_connection((GW_HOST.format(zone=zone_id), login_port), timeout=10)
    tail = 0

    def send(mid, recog, p1, p2, p3, body, t=None):
        nonlocal tail
        if t is None:
            tail += 1
            t = tail
        g.sendall(encode6(struct.pack("<9i", 0, mid, recog, p1, p2, p3,
                                      0, len(body), t)) + body)

    body2001 = (f"{uid}/".encode() + encode6(jwt.encode()) + b"/" +
                encode6(CHANNEL.encode()) + b"/1/0")
    send(2001, random.randint(1, 2**30), zone_id, login_port, 0, body2001, t=0)
    key32 = sid = None
    for m, b in recv_frames(g, 6):
        if m[1] == 529:
            key32 = decode6(b).split(b"/")[0]
    if not key32:
        g.close()
        raise SystemExit("[!] 登录服未下发 529 key")

    send(104, 0, 0, 0, 0, encode6(key32))
    for m, b in recv_frames(g, 4):
        if m[1] == 530:
            sid = m[0]

    body100 = f"{uid}/{sid}/{zone_id}/{CHANNEL}/{GAME_ID}/".encode() + key32
    send(100, random.randint(1, 2**30), zone_id, login_port, sid, body100)
    roles = None
    for m, b in recv_frames(g, 6):
        if m[1] == 520 and b:
            try:
                roles = json.loads(decode6(b) if all(0x3C <= c <= 0x7B for c in b[:200]) else b)
            except Exception:
                roles = None
    print(f"[*] 角色列表: {json.dumps(roles, ensure_ascii=False)[:200] if roles else None}")

    chr_info = next((c for c in (roles or {}).get("list", []) if c["chr"] == name), None)
    if chr_info is None:
        # 真客户端创角前先发 107(要名)拿 539 确认(cap_hoe_success 2026-10-01 t=21.0-21.1),
        # 跳过则服务端对 101 静默(删角后重创实测)
        send(107, 0, 0, 0, sid, name.encode())
        for m, b in recv_frames(g, 4):
            if m[1] == 539:
                print(f"[*] 107→539 要名确认: {b[:32]}")
        body101 = json.dumps({"server_name": zone_id, "chr": name, "secsdk": "0",
                              "job": 0, "sex": 0, "account": uid,
                              "sdkid": CHANNEL}, ensure_ascii=False).encode()
        send(101, 0, 0, 0, sid, body101)
        ticket = None
        for m, b in recv_frames(g, 6):
            if m[1] == 521 and b:
                try:
                    ticket = decode6(b).decode("utf-8", "replace")
                except Exception:
                    ticket = b.decode("utf-8", "replace")
            if m[1] == 522:
                print("[!] 522: 角色名已存在或槽位已满")
        print(f"[*] 101创角 → 521票据: {ticket}")
        if ticket:
            chr_info = {"chr": name, "UserID": ticket.split("/", 1)[1]}
    if chr_info is None:
        lst = (roles or {}).get("list") or []
        if not lst:
            g.close()
            raise SystemExit("[!] 无角色且创角失败")
        chr_info = lst[-1]
    name = chr_info["chr"]
    ticket_key = chr_info["UserID"]

    send(103, 0, 0, 0, sid, f"{uid}/{name}".encode())
    game_port = None
    for m, b in recv_frames(g, 6):
        if m[1] == 525 and b:
            try:
                gw = decode6(b).decode("utf-8", "replace")
                game_port = int(gw.split("/")[-1])
            except Exception:
                pass
    g.close()
    _, default_game = gateway_ports(zone_id, login_port)
    return key32, sid, name, ticket_key, game_port or default_game


def enter_game_gate(zone_id, name, ticket_key, sid, game_port, uid):
    g = socket.create_connection((GW_HOST.format(zone=zone_id), game_port), timeout=10)
    body20 = f"*#{uid}/{name}/{ticket_key}/{sid}/1/0/{CHANNEL}/0/1/1/1/1".encode()
    head = struct.pack("<9i", 900000, 20, random.randint(1, 2**30),
                       zone_id, game_port, sid, 0, len(body20), 0)
    g.sendall(encode6(head) + body20)
    return g


def heartbeat_loop(sock, sid, lock=None, sender=None):
    """CS60000 心跳。tail 必须与主发送序号共用一条递增链
    (2026-09-23 抓包实证: 真客户端 init=tail1-18, 心跳=tail19+ 全序连续;
    独立计数器造成 tail 冲突 → 服务器判定会话非法 → SC28 拒绝一切移动)。
    sender=bot 时经 bot.send 走全局唯一 tail; 否则退回独立计数(旧危险行为)。"""
    state = {"n": 0}

    def hb():
        time.sleep(2)
        fails = 0
        while True:
            try:
                if sender is not None:
                    sender(60000, int(time.time()), 0, 0, sid, b"")
                else:
                    state["n"] += 1
                    h = struct.pack("<9i", 0, 60000, int(time.time()), 0, 0,
                                    sid, 0, 0, state["n"])
                    wire = encode6(h)
                    if lock is not None:
                        with lock:
                            sock.sendall(wire)
                    else:
                        sock.sendall(wire)
                fails = 0
            except Exception:
                fails += 1
                if fails >= 3:
                    return
            time.sleep(2)
    threading.Thread(target=hb, daemon=True).start()


CFG_1018 = ('{"cfg_unbindlist":"e806a0f429dbba2caf91ce14becfa824","packageName":"com.mac.BoxMacOS.xq.game",'
            '"gameid":206923,"cfg_callmobset":"d41d8cd98f00b204e9800998ecf8427e",'
            '"cfg_suit":"e46446bb18ccaf568d906595dd5eea51156fae9eb97a06c4add808c9bb7a621e",'
            '"cfg_stditem":"b8a98d345b8929f7ab959b5f05156c11b92ebc3e1c67a7eb298d6f0f4da276f1",'
            '"cfg_cusabilmapping":"e2a78dceea819be6821b60b1b9a8a742",'
            '"channelId":"110585","tripartite":0}')

TPL_DIR = os.path.join(HERE, "docs")


def _load_ui_templates():
    path = os.path.join(TPL_DIR, "ui60_初始化模板.json")
    try:
        with open(path) as f:
            return json.load(f)
    except OSError:
        return None


DEV_FINGERPRINT_5026 = (b'{"model":"Mac16,10","type":"mac","os":"26.6.2","width":"1230",'
                        b'"id":"FCE92589-6485-5AB5-A8D5-46E7CDC20BC3",'
                        b'"boxid":"206923:110585","height":"760"}')
CHANNEL_5028 = b'{"sdkver":"1.0.0","channel":"996box_mac","appid":"ykOL4T4d6Oav","appver":"4.1.0"}'


def send_init_sequence(sock, sid, uid):
    """真实客户端进游戏后的19帧初始化序列(完整资格版, 手册§十四/P4修正).

    要点: 145B设备指纹5026 + 996box_mac渠道5028 + 415B功能位图5277,
    缺任何一项服务器对UI交互静默(60区56613真UI对照实证)."""
    tpls = _load_ui_templates()
    body_5277 = (tpls["5277"].encode() if tpls and tpls.get("5277")
                 else None)
    frames = [
        (1018, CFG_1018),
        (1046, None), (1046, None), (1046, None), (1046, None),
        (2020, f"{uid}/3.40.4087/3.40.9112/{uid}/0/1001/110585/0/0"),
        (5026, None, "fingerprint"),
        (5028, CHANNEL_5028.decode()),
        (5027, "3"),
        (5260, None), (1078, None), (1079, None), (5148, None),
        (5277, None, "bitmap"),
        (1103, None), (1151, None), (6209, None), (6208, None),
        (5027, "3"),
    ]
    t = 2
    for item in frames:
        mid, bd = item[0], item[1]
        kind = item[2] if len(item) > 2 else None
        if kind == "fingerprint":
            body = DEV_FINGERPRINT_5026
        elif kind == "bitmap":
            body = body_5277
        else:
            body = bd.encode() if bd else b""
        p1_1018, p2_1018 = (41878640, 98395) if mid == 1018 else (0, 0)
        sock.sendall(encode6(struct.pack("<9i", 0, mid, 0, p1_1018, p2_1018, sid,
                                         0, len(body), t)) + body)
        t += 1
        time.sleep(0.15)
    deadline = time.time() + 2
    while time.time() < deadline:
        try:
            sock.settimeout(max(0.2, deadline - time.time()))
            d = sock.recv(65536)
            if not d:
                break
        except (socket.timeout, OSError):
            break


def game_action(sock, sid, act, index, tail_seq, wait=5.0):
    """发送 CS3199 UI动作并等待响应, 返回(喝药开关状态 or None, 新tail序号)。"""
    body = ('{"index":%d,"act":"%s","subid":1}' % (index, act)).encode()
    sock.sendall(encode6(struct.pack("<9i", 0, 3199, 0, 0, 0, sid,
                                     0, len(body), tail_seq)) + body)
    t0 = time.time()
    state = None
    while time.time() - t0 < wait:
        try:
            sock.settimeout(2)
            d = sock.recv(65536)
        except (socket.timeout, OSError):
            continue
        if not d:
            break
        off = 0
        while off < len(d) - 47:
            try:
                h = struct.unpack("<9i", decode6(d[off:off + 48]))
            except Exception:
                off += 1
                continue
            mid, blen = h[1], h[7]
            if blen < 0 or blen > 100000:
                off += 1
                continue
            body = d[off + 48:off + 48 + blen]
            if mid != 60000 and body[:2] == b"x\x9c":
                try:
                    dec = zlib.decompress(body)
                    blobs = [dec]
                    if all(0x3C <= c <= 0x7B for c in dec[:60]):
                        try:
                            blobs.append(decode6(dec))
                        except Exception:
                            pass
                    for blob in blobs:
                        if b"yd_cy1" in blob:
                            import re
                            m2 = re.search(rb"id=yd_cy1[^}]*?but_[01]\.png", blob, re.S)
                            if m2:
                                state = "on" if b"but_1" in m2.group(0) else "off"
                            break
                except Exception:
                    pass
            off += 48 + blen
    return state


def enable_auto_heal(sock, sid, want_on=True):
    """开启喝药保护。返回最终状态 'on'/'off'/None。

    开关是 toggle: 先读当前状态, 不符则切换一次。"""
    game_action(sock, sid, "@openmianban,23", 109, 900)
    time.sleep(1)
    state = game_action(sock, sid, "@chiyao_kaiqi", 303, 901, wait=1.5)
    if state is None:
        # 面板未刷新过, 无法得知当前态: 假设为 off, 切换一次
        state = game_action(sock, sid, "@chiyao_kaiqi", 303, 902)
    if want_on and state == "off":
        state = game_action(sock, sid, "@chiyao_kaiqi", 303, 903)
    if not want_on and state == "on":
        state = game_action(sock, sid, "@chiyao_kaiqi", 303, 903)
    return state


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("-u", "--user", default="z1098210167")
    ap.add_argument("-p", "--password", default="yang535006")
    ap.add_argument("--zone", type=int, default=14459329,
                    help="霸者52区=14459329 霸者38区=14448583")
    ap.add_argument("--port", type=int, default=LOGIN_PORT,
                    help="登录服端口(38区8041, 52区8001, 默认8001)")
    ap.add_argument("--name", default="泪无期")
    ap.add_argument("--wait", type=int, default=60)
    ap.add_argument("--auto-heal", action="store_true",
                    help="进游戏后发送初始化序列并开启喝药保护")
    args = ap.parse_args()

    print(f"[*] 登录 {args.user} ...")
    s = rq.Session()
    j, _ = L.do_login(s, args.user, args.password)
    if j.get("code") != 200:
        raise SystemExit(f"[!] 登录失败: {j.get('msg')}")
    jwt = j["data"]["token"]
    uid = (j["data"].get("userInfo") or {}).get("id") or UID

    j2 = enter_game(s, jwt, args.zone)
    print(f"[*] EnterGameClient(zone={args.zone}): code={j2.get('code')}")

    key32, sid, name, ticket_key, game_port = login_gate(
        args.zone, jwt, args.name, args.port, uid)
    print(f"[✓] key32={key32.decode(errors='replace')}")
    print(f"[✓] sid={sid} uid={uid} 角色={name}")
    print(f"[✓] 票据key={ticket_key}")
    print(f"[✓] 游戏服端口={game_port}")

    g = enter_game_gate(args.zone, name, ticket_key, sid, game_port, uid)
    heartbeat_loop(g, sid)
    print("[*] CS20 已发 + 心跳运行")

    if args.auto_heal:
        time.sleep(2)
        send_init_sequence(g, sid, uid)
        print("[*] 17帧初始化序列已发")
        state = enable_auto_heal(g, sid, want_on=True)
        if state == "on":
            print("[✓] 喝药保护已开启 (服务端确认 but_1)")
        elif state == "off":
            print("[!] 喝药保护当前为关闭 (切换未生效?)")
        else:
            print("[!] 未捕获到开关状态面板")

    print("[*] 监听服务器数据 ...")
    deadline = time.time() + args.wait
    total = 0
    shown = 0
    while time.time() < deadline:
        got = recv_frames(g, min(5, max(1, int(deadline - time.time()))))
        for m, b in got:
            if m[1] == 60000:
                continue
            total += 1
            if shown < 20:
                shown += 1
                info = ""
                if b:
                    if b[:2] == b"x\x9c":
                        try:
                            b = zlib.decompress(b)
                            info += f"zlib[{len(b)}B] "
                        except Exception:
                            pass
                    info += b[:80].decode("utf-8", "replace")
                print(f"  S< id={m[1]} p3={m[5]} {info!r}")
    print(f"[✓] 完成 — 收到{total}帧游戏数据, 纯协议角色在线 {args.wait}s")


if __name__ == "__main__":
    main()