feat(apilot): host with verified login-chain drivers

Host(zone,user,password,state_dir) wires HTTP login -> gateway -> create role
on top of Task1-6 stubs. build_2001_body / build_101_body are pure and
unit-tested; run_create_role walks the cq2_pure_enter.login_gate() frame
sequence verbatim (2001/529/104/530/100/520/101/521/522) with explicit
per-frame comments mapping recog/p1/p2/p3/tail/body back to the source.

Frame header construction goes through nc.send_frame -> pack_frame only,
no duplicate header assembly. SC530 sid is recovered via _recv_with_pre
since apilot Frame discards m[0]; other SC frames are read via the normal
pump/dispatch path.

run_zone is the CLI one-shot (HTTP -> gate -> role) wrapper.

apilot/host.py

@@ -0,0 +1,248 @@
+"""宿主: HTTP 登录 + 网关连接 + 创角驱动。复用 protocol_bot 已验证资产。"""
+from __future__ import annotations
+
+import json
+import os
+import random
+import sys
+import time
+from typing import Any
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.join(ROOT, "protocol_bot"))
+
+import requests as rq  # noqa: E402
+import login_996pc as L  # type: ignore  # noqa: E402
+import cq2_pure_enter as P  # type: ignore  # noqa: E402
+from encode6bitbuf_final import encode6, decode6  # type: ignore  # noqa: E402
+
+from apilot.bridge import bootstrap_runtime  # noqa: E402
+from apilot.stubs.env import EnvData  # noqa: E402
+from apilot.stubs.netclient import NetClientStub  # noqa: E402
+from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame  # noqa: E402
+from apilot.stubs.scheduler import Scheduler  # noqa: E402
+from apilot.wire import Frame  # noqa: E402
+
+GW_HOST = "s{zone}-jycq-206923.dhsf.xqhuyu.com"
+CHANNEL = "110585"
+GAME_ID = 206923
+
+
+class Host:
+    def __init__(self, zone_id: int, user: str, password: str,
+                 state_dir: str = ".apilot_state") -> None:
+        self.zone = zone_id
+        self.user, self.password = user, password
+        self.state_dir = state_dir
+        self.nc = NetClientStub()
+        self.ctl = NetworkCtlStub(self.nc)
+        self.sched = Scheduler()
+        self.env = EnvData({})
+        self.lua: Any = None
+        self.jwt = ""
+        self.uid = ""
+        self._login_port = 0
+
+    def build_2001_body(self, uid: str, jwt: str) -> bytes:
+        return (f"{uid}/".encode() + encode6(jwt.encode()) + b"/" +
+                encode6(CHANNEL.encode()) + b"/1/0")
+
+    def build_101_body(self, name: str, uid: str, job: int = 0, sex: int = 0) -> bytes:
+        return json.dumps({"server_name": self.zone, "chr": name, "secsdk": "0",
+                           "job": job, "sex": sex, "account": uid,
+                           "sdkid": CHANNEL}, ensure_ascii=False).encode()
+
+    def http_login(self) -> dict:
+        s = rq.Session()
+        j, _ = L.do_login(s, self.user, self.password)
+        if j.get("code") != 200:
+            raise SystemExit(f"[!] 登录失败: {j.get('msg')}")
+        self.jwt = j["data"]["token"]
+        self.uid = str((j["data"].get("userInfo") or {}).get("id") or "")
+        self.env.data.update({"uid": self.uid, "token": self.jwt,
+                              "new_boxtoken": self.jwt, "isBoxLogin": 1})
+        return j
+
+    def boot_lua(self) -> None:
+        self.lua = bootstrap_runtime(self.nc, self.ctl, self.sched, self.env, self.state_dir)
+
+    def connect_login_gate(self, port: int = 8041) -> None:
+        self._login_port = port
+        self.nc.connect(GW_HOST.format(zone=self.zone), port)
+
+    def pump(self, seconds: float, on_frame=None) -> list:
+        """收帧→分发→调度, 驱动 seconds 墙钟。"""
+        end = time.time() + seconds
+        got: list = []
+        while time.time() < end:
+            for f in self.nc.tick(0.2):
+                if self.lua is not None:
+                    dispatch_frame(self.lua, self.ctl, f)
+                got.append(f)
+                if on_frame:
+                    on_frame(f)
+            self.sched.tick(0.05)
+        return got
+
+    def tick(self, dt: float = 0.05) -> None:
+        for f in self.nc.tick(dt):
+            if self.lua is not None:
+                dispatch_frame(self.lua, self.ctl, f)
+        self.sched.tick(dt)
+
+    def _send(self, mid: int, recog: int, p1: int, p2: int, p3: int,
+              tail: int, body: bytes) -> None:
+        """登录链统一发包: body 字节由 host 构造,帧头走 pack_frame(wire.py)唯一真源。"""
+        self.nc.send_frame(Frame(recog=recog, msg_id=mid, p1=p1, p2=p2,
+                                  p3=p3, tail=tail, body=body))
+
+    def _wait_for(self, mid: int, timeout: float):
+        """收帧直到命中 mid 或超时。返回 (帧, 已收列表)。"""
+        deadline = time.time() + timeout
+        got: list = []
+        while time.time() < deadline:
+            remain = max(0.05, deadline - time.time())
+            for f in self.pump(min(remain, 0.2)):
+                got.append(f)
+                if f.msg_id == mid:
+                    return f, got
+        return None, got
+
+    def _recv_with_pre(self, target_mid: int, timeout: float):
+        """等帧时同时拿回 m[0]("pre" 字段,apilot Frame 抽象丢弃)。
+        仅 SC530 用 — cq2_pure_enter.login_gate:138 sid=m[0]。"""
+        import struct
+        wire = self.nc.wire
+        if wire is None:
+            return None, None
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            remain = max(0.05, deadline - time.time())
+            wire.sock.settimeout(min(0.2, remain))
+            try:
+                d = wire.sock.recv(65536)
+                if not d:
+                    break
+            except (socket.timeout, ConnectionResetError, OSError):
+                d = b""
+            consumed = 0
+            found_frame: Frame | None = None
+            found_pre: int | None = None
+            while consumed + 48 <= len(d):
+                try:
+                    m = struct.unpack("<9i", decode6(d[consumed:consumed + 48]))
+                except Exception:
+                    break
+                blen = m[7]
+                if m[1] == target_mid:
+                    found_frame = Frame(recog=m[2], msg_id=m[1], p1=m[3], p2=m[4],
+                                        p3=m[5], tail=m[8],
+                                        body=d[consumed + 48:consumed + 48 + blen])
+                    found_pre = m[0]
+                    after = d[consumed + 48 + blen:]
+                    wire._buf = wire._buf + d[:consumed] + after
+                    return found_frame, found_pre
+                consumed += 48 + blen
+            wire._buf = wire._buf + d
+        return None, None
+
+    def run_create_role(self, name: str | None = None, timeout: float = 30) -> dict:
+        """登录链逐帧驱动(对照 cq2_pure_enter.login_gate 唯一真源,逐帧注释)。
+
+        2001(tail=0,recog=random) → 等 529 取 key32(decode6 split /) →
+        104(body=encode6(key32),tail=1) → 等 530 取 sid(recog 字段,cq2_pure_enter:138) →
+        nc.set_sdata(sid) → 100(uid/sid/zone/channel/gameid/key32,tail=2) →
+        等 520 角色 JSON(decode6 后 json.loads) →
+        角色列表含 name→复用;否则 101(tail=3) → 等 521(decode6 后 split /)
+        /522(换名重试 ≤5 次)。
+        """
+        port = self._login_port
+        # CS2001 ─ 登录服首帧,带 uid/encode6(jwt)/encode6(channel)/1/0
+        self._send(2001, random.randint(1, 2**30), self.zone, port, 0, 0,
+                   self.build_2001_body(self.uid, self.jwt))
+        # 等 SC529 → 会话密钥 key32(body 形如 "key32/...")
+        f529, _ = self._wait_for(529, timeout)
+        if f529 is None:
+            raise RuntimeError("[!] 登录服未下发 529 key")
+        key32 = decode6(f529.body).split(b"/")[0]
+        # CS104 ─ 回传 key32 协商
+        self._send(104, 0, 0, 0, 0, 1, encode6(key32))
+        # 等 SC530 → sid 真实位置 m[0](首 int,"pre" 字段),不是 Frame.recog(m[2])。
+        # cq2_pure_enter.login_gate:138 sid=m[0] 唯一真源;apilot Frame 抽象丢弃 m[0],
+        # 这里 _recv_with_pre 直读 raw 字节补回。未命中超时即报错。
+        _, sid = self._recv_with_pre(530, timeout)
+        if sid is None:
+            raise RuntimeError("[!] 登录服未下发 530 sid")
+        self.nc.set_sdata(sid)
+        # CS100 ─ 拉角色列表(body 形如 "uid/sid/zone/channel/gameid/"+key32)
+        body100 = (f"{self.uid}/{sid}/{self.zone}/{CHANNEL}/{GAME_ID}/"
+                   .encode() + key32)
+        self._send(100, random.randint(1, 2**30), self.zone, port, sid, 2, body100)
+        # 等 SC520 → 角色列表 JSON(cq2_pure_enter 兼容: encode6 包裹时先 decode6)
+        roles: dict | None = None
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            remain = max(0.05, deadline - time.time())
+            for f in self.pump(min(remain, 0.2)):
+                if f.msg_id == 520 and f.body:
+                    try:
+                        b = f.body
+                        if all(0x3C <= c <= 0x7B for c in b[:200]):
+                            b = decode6(b)
+                        roles = json.loads(b)
+                    except Exception:
+                        roles = None
+                    if roles is not None:
+                        break
+            if roles is not None:
+                break
+        # 选/创角
+        if name is None:
+            name = "萌新" + str(random.randint(1000, 9999))
+        chr_info = next((c for c in (roles or {}).get("list", [])
+                          if c.get("chr") == name), None)
+        retries = 0
+        while chr_info is None and retries <= 5:
+            self._send(101, 0, 0, 0, sid, 3,
+                       self.build_101_body(name, self.uid))
+            ticket: str | None = None
+            got_522 = False
+            sub_deadline = time.time() + timeout
+            while time.time() < sub_deadline:
+                remain = max(0.05, sub_deadline - time.time())
+                for f in self.pump(min(remain, 0.2)):
+                    if f.msg_id == 521 and f.body:
+                        try:
+                            ticket = decode6(f.body).decode("utf-8", "replace")
+                        except Exception:
+                            ticket = f.body.decode("utf-8", "replace")
+                    if f.msg_id == 522:
+                        got_522 = True
+                if ticket or got_522:
+                    break
+            if got_522:
+                retries += 1
+                name = "萌新" + str(random.randint(1000, 9999))
+                continue
+            if ticket:
+                chr_info = {"chr": name, "UserID": ticket.split("/", 1)[1]}
+                break
+            break
+        if chr_info is None:
+            lst = (roles or {}).get("list") or []
+            if lst:
+                chr_info = lst[-1]
+        if chr_info is None:
+            raise RuntimeError("[!] 无角色且创角失败")
+        return {"name": chr_info["chr"], "sid": sid,
+                "key32": key32, "user_id": chr_info["UserID"]}
+
+
+def run_zone(zone_id: int, user: str, password: str, name: str | None = None,
+             port: int = 8041) -> dict:
+    """一站式: HTTP 登录 → 网关连接 → 创角 → 返回票据。"""
+    h = Host(zone_id, user, password)
+    h.http_login()
+    h.connect_login_gate(port=port)
+    info = h.run_create_role(name)
+    return {"uid": h.uid, "jwt": h.jwt, "zone": h.zone, **info}

tests/test_host_unit.py

@@ -0,0 +1,28 @@
+# tests/test_host_unit.py
+"""Task 7: host.py — 登录链拼包纯函数(零真实网络)。"""
+import json
+
+from apilot.host import Host
+
+
+def test_login_2001_body_shape(tmp_path):
+    """2001 body = b"uid/" + encode6(jwt) + b"/" + encode6(channel) + b"/1/0"。
+    与 cq2_pure_enter.login_gate() 第 124-125 行 逐字节一致。"""
+    h = Host(14481607, "longwu988", "Longwu1988", state_dir=str(tmp_path))
+    uid, jwt = "1186115037", "eyJfake.jwt.token"
+    body = h.build_2001_body(uid, jwt)
+    import sys
+    sys.path.insert(0, "protocol_bot")
+    from encode6bitbuf_final import encode6  # type: ignore
+    assert body == f"{uid}/".encode() + encode6(jwt.encode()) + b"/" + encode6(b"110585") + b"/1/0"
+
+
+def test_create_role_101_body(tmp_path):
+    """101 body = {"server_name":zone,"chr":name,"secsdk":"0","job":0,"sex":0,
+    "account":uid,"sdkid":"110585"};中文/账号/区号全保留。"""
+    h = Host(14481607, "u", "p", state_dir=str(tmp_path))
+    body = json.loads(h.build_101_body("萌新1234", uid="u1", job=0, sex=0))
+    assert body["chr"] == "萌新1234" and body["sdkid"] == "110585"
+    assert body["server_name"] == 14481607
+    assert body["account"] == "u1" and body["job"] == 0 and body["sex"] == 0
+    assert body["secsdk"] == "0"