chuanqi/pipeline_bot_probe.py

"""probe_zone_login.py — 最小登录服探针: 逐帧打印 33 区登录链的应答。"""
import socket
import sys

sys.path.insert(0, "src")
from pbot.codec.bitbuf import decode6, encode6  # noqa: E402
from pbot.codec.frame import FrameSplitter  # noqa: E402
from pbot.transport.gateway import (  # noqa: E402
    CHANNEL, GAME_ID, GatewayConnection)

ZONE = int(sys.argv[1]) if len(sys.argv) > 1 else 14605092
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8041
UID = sys.argv[3] if len(sys.argv) > 3 else "1186115037"
JWT = sys.argv[4]

conn = GatewayConnection(ZONE, PORT) if False else None
import random  # noqa: E402

from pbot.types import ZoneId  # noqa: E402
from pbot.errors import TransportError  # noqa: E402


def recv_print(c: GatewayConnection, tag: str, wait: float = 6.0) -> list:
    frames = c.recv_frames(wait)
    for f in frames:
        body_preview = f.body[:60]
        try:
            dec = decode6(f.body)
            body_preview = dec[:80]
        except Exception:  # noqa: BLE001
            pass
        print(f"  [{tag}] SC{f.msg_id} recog={f.recog} p1={f.p1} p2={f.p2} "
              f"p3={f.p3} body={body_preview}")
    return frames


c = GatewayConnection(ZoneId(ZONE), PORT)
try:
    body2001 = f"{UID}/".encode() + encode6(JWT.encode()) + b"/" \
        + encode6(CHANNEL.encode()) + b"/1/0"
    _ = c.send_frame(2001, random.randint(1, 2**30), ZONE, PORT, 0, body2001, tail=0)
    frames = recv_print(c, "L1")
    key32 = None
    for f in frames:
        if f.msg_id == 529:
            key32 = decode6(f.body).split(b"/")[0]
    if not key32:
        print("!! 无 529")
        sys.exit(1)
    _ = c.send_frame(104, 0, 0, 0, 0, encode6(key32), tail=1)
    frames = recv_print(c, "L2", 4)
    sid = 0
    for f in frames:
        if f.msg_id == 530:
            sid = f.pre
    print(f"sid={sid}")
    body100 = f"{UID}/{sid}/{ZONE}/{CHANNEL}/{GAME_ID}/".encode() + key32
    _ = c.send_frame(100, random.randint(1, 2**30), ZONE, PORT, sid, body100, tail=2)
    print("-- CS100 已发, 等应答(最长 8s) --")
    frames = recv_print(c, "L3", 8)
    if not frames:
        print("!! CS100 零应答")
finally:
    c.close()