feat(bootstrap): 接入区服端口表三级策略 — serverlist XXTEA解密按区直取端口, 新区不再错端口断连

pipeline_bot/pipeline/bootstrap.py

@@ -13,6 +13,7 @@ import requests
 from pbot.config import Account, RoleSpec
 from pbot.game.session import enter_world
 from pbot.login.box_client import enter_game_client, login
+from pbot.login.zone_ports import fetch_zone_ports
 from pbot.transport.gateway import create_role_needed, login_gate_with_retry
 from pbot.types import ZoneId
 from pbot.world.update import as_dict, to_float, to_int, typed_json
@@ -82,6 +83,25 @@ def _resolve_zone(force_zone: int) -> tuple[ZoneId, str]:
     return ZoneId(sid), _zone_label(list_name)
 
 
+def _resolve_login_ports(zone: ZoneId, login_port: int) -> tuple[int, ...]:
+    """三级端口策略(2026-10-01 定案: 新物理机端口与旧表不同, 错端口 CS100 被裸断连)。
+
+    1. 显式 --login-port 最高优先;
+    2. serverlist XXTEA 解密(zone_ports)按区直取 — 秒级全自动;
+    3. 旧端口表轮询兜底(解密链失败时)。
+    """
+    if login_port:
+        return (login_port,)
+    try:
+        info = fetch_zone_ports().get(int(zone))
+        if info and info.get("port"):
+            _LOG.info("区服端口表: %s 端口=%d", info.get("name"), info["port"])
+            return (info["port"],)
+    except Exception as e:  # noqa: BLE001 - 端口表失败走兜底轮询
+        _LOG.warning("区服端口表获取失败(%s) — 回退旧端口轮询", e)
+    return DEFAULT_PORTS
+
+
 def bootstrap_session(user: str, password: str, woma_minutes: float,
                       login_port: int = 0, force_zone: int = 0,
                       role_name: str = "",
@@ -104,7 +124,7 @@ def bootstrap_session(user: str, password: str, woma_minutes: float,
                 raise
             time.sleep(_ENTER_GAME_RETRY_SLEEP_S)
 
-    ports: tuple[int, ...] = (login_port,) if login_port else DEFAULT_PORTS
+    ports: tuple[int, ...] = _resolve_login_ports(zone, login_port)
     _, sid_game, real_name, ticket_key, game_port = login_gate_with_retry(
         zone, jwt, uid, role, ports)
     if real_name != role:

pipeline_bot/src/pbot/login/zone_ports.py

@@ -0,0 +1,120 @@
+"""zone_ports.py — 996盒子区服端口获取(纯 HTTP, 无需客户端文件)。
+
+链路(2026-10-01 定案, 逆向 lua脚本解密源码/mod_launcher):
+    1. GET https://list-box.dhsf.xqhuyu.com/modlist/modlist_{channel}.txt
+       → modData[].srvlist = serverlist URL
+    2. GET serverlist → base64 → b"4meJnPyl"(XXTEA sign) + XXTEA 密文
+    3. XXTEA 解密(key=34UFXMgtkz, cpp_reversed/keys_inventory.md 定案)
+    4. JSON: serverData[] = {d: serverid, p: 登录服端口, s: state(1=已开),
+       n: 区名, t: 分组tag}
+约束: 解密输出 JSON 后有 ≤4B 密文残渣, 用 raw_decode 截取。
+"""
+from __future__ import annotations
+
+import base64
+import json
+import logging
+import struct
+from typing import Final
+
+import requests
+
+_LOG = logging.getLogger("pbot.zone_ports")
+
+MODLIST_URL: Final[str] = \
+    "https://list-box.dhsf.xqhuyu.com/modlist/modlist_{channel}.txt"
+XXTEA_SIGN: Final[bytes] = b"4meJnPyl"
+XXTEA_KEY: Final[bytes] = b"34UFXMgtkz"
+_DELTA: Final[int] = 0x9E3779B9
+_M32: Final[int] = 0xFFFFFFFF
+
+
+def _xxtea_decrypt(ct: bytes, key16: bytes) -> bytes:
+    """cocos2d-x xxtea.c 标准解密(key 补零至 16B)。"""
+    v = list(struct.unpack(f"<{len(ct) // 4}I", ct[: len(ct) // 4 * 4]))
+    k = list(struct.unpack("<4I", key16))
+    n = len(v)
+    if n < 2:
+        return b""
+    rounds = 6 + 52 // n
+    sum_ = (rounds * _DELTA) & _M32
+    y = v[0]
+    for _ in range(rounds):
+        e = (sum_ >> 2) & 3
+        for p in range(n - 1, 0, -1):
+            z = v[p - 1]
+            mx = (((z >> 5) ^ (y << 2)) + ((y >> 3) ^ (z << 4))) \
+                ^ ((sum_ ^ y) + (k[(p & 3) ^ e] ^ z))
+            v[p] = (v[p] - mx) & _M32
+            y = v[p]
+        z = v[n - 1]
+        mx = (((z >> 5) ^ (y << 2)) + ((y >> 3) ^ (z << 4))) \
+            ^ ((sum_ ^ y) + (k[e] ^ z))
+        v[0] = (v[0] - mx) & _M32
+        y = v[0]
+        sum_ = (sum_ - _DELTA) & _M32
+    return struct.pack(f"<{n}I", *v)
+
+
+def _fetch(url: str, s: requests.Session) -> bytes:
+    r = s.get(url, timeout=20, verify=False)
+    r.raise_for_status()
+    return r.content
+
+
+def fetch_zone_ports(channel: int = 110585,
+                     s: requests.Session | None = None) -> dict[int, dict]:
+    """区服端口表: {serverid: {port, state, name}}。纯 HTTP, 无客户端依赖。"""
+    own = s is None
+    sess = s or requests.Session()
+    if own:
+        sess.trust_env = False
+        sess.verify = False
+    try:
+        mod = json.loads(_fetch(MODLIST_URL.format(channel=channel), sess))
+        srv_url = ""
+        for m in mod.get("modData", []):
+            srv_url = str(m.get("srvlist", "")) or srv_url
+        if not srv_url:
+            raise ValueError("modlist 无 srvlist 字段")
+        raw = _fetch(srv_url, sess)
+        blob = base64.b64decode(raw.strip() + b"=" * (-len(raw.strip()) % 4))
+        if not blob.startswith(XXTEA_SIGN):
+            sign = f"serverlist sign 不匹配: {blob[:8]!r}"
+            raise ValueError(sign)
+        key16 = XXTEA_KEY.ljust(16, b"\x00")
+        pt = _xxtea_decrypt(blob[8:], key16).rstrip(b"\x00")
+        txt = pt.decode("utf-8", "replace")
+        data, _end = json.JSONDecoder().raw_decode(txt)
+        zones: dict[int, dict] = {}
+        for item in data.get("serverData", []):
+            sid = int(item.get("d", 0))
+            if sid:
+                zones[sid] = {"port": int(item.get("p", 0)),
+                              "state": int(item.get("s", 0)),
+                              "name": str(item.get("n", ""))}
+        _LOG.info("[区服端口] %d 区获取成功(源=%s)", len(zones), srv_url)
+        return zones
+    finally:
+        if own:
+            sess.close()
+
+
+def port_for(zone: int, channel: int = 110585) -> int:
+    """单区端口便捷入口; 0 = 未查到。"""
+    try:
+        info = fetch_zone_ports(channel).get(zone)
+        return info["port"] if info else 0
+    except Exception as e:  # noqa: BLE001 - 便捷入口整体容错
+        _LOG.warning("[区服端口] 获取失败: %s", e)
+        return 0
+
+
+if __name__ == "__main__":
+    import sys
+    from pathlib import Path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src"))
+    logging.basicConfig(level=logging.INFO, stream=sys.stdout)
+    zones = fetch_zone_ports()
+    for zid in (14605092, 14605093, 14605094):
+        _LOG.info("冒烟: %d -> %s", zid, zones.get(zid))