547f4beb7e
M1（构建）： - build.sh 编译 CP 加入 $OUT/fabric/*.jar，解决 net.fabricmc.* 找不到 - gen_1_21 内联 Sodium 通知改为委托共用 RenderCompat，补 reportRenderCompat - 1.21.1 全量七步构建产物齐备（classes-intermediary.jar / names*.json / srg / inheritance.map） M2（协议）： - login_start 按协议数据发 UUID（1.21 必填；1.20 可选） - login_acknowledged / serverbound 包 id 改为查表，不再写死 0x00 - send() 实现压缩帧格式（varint(dataLen)+数据，小包不压） - configuration 双向协商：发 settings、回 select_known_packs、拒 resource_pack - protodef 补 bitflags / entityMetadataLoop / count(ctx引用) / compareTo 路径表达式 - NBT 读取器由占位符改为真实现（anonymousNbt 支撑 registry_data） - 坏包容错：长度前缀是帧层权威，解不开整包丢弃并布告，不断流 实测：login→configuration(11 注册表)→play，位置 (5.5,-60.0,-0.5)，map_chunk×9 判据：protodef 6→13 项（含线上真实 entity_metadata 字节回归）；全量 77 项通过
548 lines
23 KiB
Python
548 lines
23 KiB
Python
"""最小 MC 客户端（Python 净室实现）—— 跨实现互操作验证。
|
||
|
||
用它去连 Node 的 minecraft-protocol 服务端：
|
||
两边是**完全独立的实现**，能握手成功才证明协议层真的通了。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import socket
|
||
import struct
|
||
from pathlib import Path
|
||
|
||
from .protodef import Reader, Writer, ProtoError, for_phase
|
||
|
||
|
||
class ProtocolError(Exception):
|
||
pass
|
||
|
||
|
||
class Client:
|
||
"""一个最小 MC 客户端：握手 → 登录 → 读包。"""
|
||
|
||
def __init__(
|
||
self,
|
||
host: str,
|
||
port: int,
|
||
username: str,
|
||
version: str = "1.20.1",
|
||
protocol_path: str | Path | None = None,
|
||
):
|
||
self.host = host
|
||
self.port = port
|
||
self.username = username
|
||
self.version = version
|
||
self.sock: socket.socket | None = None
|
||
self.proto: dict = {}
|
||
self.proto_num: int = 0
|
||
self.compression: int = -1
|
||
# 当前所处协议阶段：login → configuration → play
|
||
# ★ 1.20.2 起才有 configuration 阶段（1.20.1 跳过它）
|
||
self.phase: str = "login"
|
||
# ★ configuration 阶段拿到的**服务端权威注册表**（mod 方块/物品在这里）
|
||
self.registries: dict[str, Any] = {}
|
||
self._pkt_names: dict[tuple[str, str], dict[int, str]] = {}
|
||
self._pid_cache: dict[tuple[str, str, str], int] = {}
|
||
# ★ 解不开的包（上游数据缺口）记在这里，不断流。
|
||
# 实测动机：declare_recipes 在官方数据上同样解不开。
|
||
self.decode_failures: list[dict[str, Any]] = []
|
||
# 允许直接指定协议数据文件（优先），否则按 data_root 找
|
||
self._path = Path(protocol_path) if protocol_path else None
|
||
|
||
# ---------- 协议数据 ----------
|
||
def load_protocol(self) -> None:
|
||
"""按版本号找 minecraft-data 的 protocol.json。
|
||
|
||
实测关键：1.20.1 的数据靠 majorVersion 继承 1.20（1.20.1/ 目录里只有
|
||
version.json）。所以要先读 version.json 拿 majorVersion 再取协议。
|
||
"""
|
||
if self.proto is not None and self.proto:
|
||
return
|
||
root = self._data_root()
|
||
vdir = root / "pc" / self.version
|
||
if not vdir.is_dir():
|
||
# 版本目录不存在时，尝试按 majorVersion 找
|
||
raise ProtocolError(f"无该版本数据: {self.version}")
|
||
|
||
vj = vdir / "version.json"
|
||
meta = json.loads(vj.read_text()) if vj.is_file() else {}
|
||
self.proto_num = meta.get("version", 0)
|
||
major = meta.get("majorVersion", self.version)
|
||
|
||
pj = root / "pc" / major / "protocol.json"
|
||
if not pj.is_file():
|
||
pj = vdir / "protocol.json"
|
||
self.proto = json.loads(pj.read_text())
|
||
self._protocol_file = pj
|
||
|
||
def _data_root(self) -> Path:
|
||
if self._path:
|
||
# 传进来的可能是 data 根，也可能是 data/pc —— 都容忍
|
||
for c in (self._path, self._path / "pc", self._path.parent):
|
||
if (c / "pc").is_dir():
|
||
return c
|
||
if c.name == "pc" and c.is_dir():
|
||
return c.parent
|
||
return self._path
|
||
cands = [
|
||
Path("/home/newqqagent/plugin_data/mc/runtime/node_modules/"
|
||
"minecraft-data/minecraft-data/data"),
|
||
Path("/home/newqqagent/plugindev/mcfold/data"),
|
||
]
|
||
for c in cands:
|
||
if (c / "pc").is_dir():
|
||
return c
|
||
raise ProtocolError("找不到 minecraft-data 数据目录")
|
||
|
||
def _pd(self, phase: str, side: str):
|
||
return for_phase(self.proto, phase, side)
|
||
|
||
# ---------- 网络 ----------
|
||
def connect(self) -> None:
|
||
self.sock = socket.create_connection((self.host, self.port), timeout=15)
|
||
self.sock.settimeout(15)
|
||
|
||
def send(self, payload: bytes) -> None:
|
||
"""发一个包（自动适配压缩帧格式）。
|
||
|
||
★★ 压缩启用后的包体格式（这正是 M2 卡住我的地方）：
|
||
包 = varint(总长) + varint(dataLength) + 数据
|
||
其中 dataLength：
|
||
== 0 → 数据**未压缩**，原样跟在后面（小包）
|
||
> 0 → 数据是 zlib，解压后长度为 dataLength
|
||
★ 服务端的判据是「未压缩长度 >= threshold 才压缩」：
|
||
包小于 threshold ⇒ dataLength 必须为 0（原样发）
|
||
我先前没加这个前缀，直接发裸包体 ⇒ 服务端把包体第 1 字节
|
||
当成 dataLength 读 ⇒ `Badly compressed packet - size of 3
|
||
is below server threshold of 256` ⇒ 断连。
|
||
反过来：把小于 threshold 的包也 zlib 压了同样会被拒。
|
||
"""
|
||
assert self.sock
|
||
if self.compression >= 0:
|
||
fw = Writer()
|
||
if self.compression > 0 and len(payload) >= self.compression:
|
||
import zlib
|
||
comp = zlib.compress(payload)
|
||
# 压缩后反而变大（极小包）时依然走未压缩分支
|
||
if len(comp) < len(payload):
|
||
fw.varint(len(payload))
|
||
fw.write(comp)
|
||
body = fw.getvalue()
|
||
else:
|
||
fw2 = Writer()
|
||
fw2.varint(0)
|
||
fw2.write(payload)
|
||
body = fw2.getvalue()
|
||
else:
|
||
fw.varint(0)
|
||
fw.write(payload)
|
||
body = fw.getvalue()
|
||
else:
|
||
body = payload
|
||
out = Writer()
|
||
out.varint(len(body))
|
||
out.write(body)
|
||
self.sock.sendall(out.getvalue())
|
||
|
||
def read_packet(self) -> bytes:
|
||
"""读一个包体（已去长度前缀；压缩未启用时即原始包）。"""
|
||
assert self.sock
|
||
length = self._read_varint_sock()
|
||
body = self._recv_exact(length)
|
||
if self.compression >= 0:
|
||
r = Reader(body)
|
||
data_len = r.varint()
|
||
if data_len == 0:
|
||
return body[r.pos:]
|
||
# 真压缩（zlib）时这里需要解压；测试服通常不压缩登录前
|
||
import zlib
|
||
return zlib.decompress(body[r.pos:])
|
||
return body
|
||
|
||
def _recv_exact(self, n: int) -> bytes:
|
||
assert self.sock
|
||
buf = b""
|
||
while len(buf) < n:
|
||
chunk = self.sock.recv(n - len(buf))
|
||
if not chunk:
|
||
raise ProtocolError(f"连接被关闭（还需 {n - len(buf)} 字节）")
|
||
buf += chunk
|
||
return buf
|
||
|
||
def _read_varint_sock(self) -> int:
|
||
result = 0
|
||
shift = 0
|
||
while True:
|
||
b = self._recv_exact(1)[0]
|
||
result |= (b & 0x7F) << shift
|
||
if not (b & 0x80):
|
||
break
|
||
shift += 7
|
||
return result
|
||
|
||
# ---------- 流程 ----------
|
||
def handshake(self, state: int = 2) -> None:
|
||
"""handshaking: set_protocol → 宣告协议版本与目标状态。"""
|
||
w = Writer()
|
||
w.varint(0x00)
|
||
w.varint(self.proto_num)
|
||
w.pstring(self.host)
|
||
w.write(struct.pack(">H", self.port))
|
||
w.varint(state)
|
||
self.send(w.getvalue())
|
||
|
||
def login_start(self) -> None:
|
||
"""login: login_start（离线模式只发用户名）。
|
||
|
||
★ **世代差异**（实测 2026-09-26，两种都撞过）：
|
||
1.20.x playerUUID: ['option', 'UUID'] ⇒ 可选，发 varint(0) 表示「无」
|
||
1.21.1 playerUUID: UUID ⇒ **必填 16 字节**
|
||
之前我无条件发 `varint(0)`，在 1.21.1 上服务端直接回
|
||
Failed to decode packet 'serverbound/minecraft:hello'
|
||
⇒ 这类差异必须**从协议数据里读**，不能凭印象写死。
|
||
"""
|
||
w = Writer()
|
||
w.varint(0x00)
|
||
w.pstring(self.username)
|
||
fields = self._login_start_fields()
|
||
if fields is not None and fields.get("playerUUID") == "UUID":
|
||
# 1.21.1+：必填 16 字节。离线模式下用**由用户名派生的确定性 UUID**
|
||
# （与服务端算法一致，见 _offline_uuid），避免每次连都换身份。
|
||
w.write(self._offline_uuid(self.username))
|
||
else:
|
||
# 1.20.x：可选，varint(0) = 无
|
||
w.varint(0)
|
||
self.send(w.getvalue())
|
||
|
||
def _login_start_fields(self) -> dict[str, str] | None:
|
||
"""从 protocol.json 读 login_start 的字段类型（不凭印象）。"""
|
||
try:
|
||
types = self.proto["login"]["toServer"]["types"]
|
||
for name in ("packet_login_start", "packet_hello"):
|
||
c = types.get(name)
|
||
if c and isinstance(c, list) and len(c) > 1:
|
||
return {f["name"]: f["type"] for f in c[1]}
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
@staticmethod
|
||
def _offline_uuid(username: str) -> bytes:
|
||
"""离线模式的 UUID = UUID.nameUUIDFromBytes("OfflinePlayer:" + name)。
|
||
|
||
★ 与官方一致：服务端对 offline-mode 也用这个算玩家身份
|
||
（GameProfile / ServerPlayer 里的同名算法）⇒ 我们发一样的值，
|
||
服务端认这个玩家，不是新建一个。
|
||
"""
|
||
import hashlib
|
||
h = bytearray(hashlib.md5(f"OfflinePlayer:{username}".encode()).digest())
|
||
h[6] = (h[6] & 0x0F) | 0x30 # version 3
|
||
h[8] = (h[8] & 0x3F) | 0x80 # IETF variant
|
||
return bytes(h)
|
||
|
||
def login_acknowledged(self) -> None:
|
||
"""login: acknowledged —— **1.21+ 才有**（1.20.x 没有这个包）。
|
||
|
||
★ 这是从 login 切到 configuration 的信号（实测）。
|
||
★★ 包 id **必须查表**，不能写死 `0x00`：
|
||
1.21.1 的 login/login_acknowledged 是 **0x03**
|
||
（0x00=login_start, 0x01=encryption_begin,
|
||
0x02=login_plugin_response, 0x03=login_acknowledged）
|
||
我第一版写死 0x00 ⇒ 服务端收到一个新 login_start，
|
||
停在 login 阶段等正确信号 ⇒ configuration 阶段永远收不到包
|
||
（时间超时，日志上看起来像"挂住"）。
|
||
"""
|
||
try:
|
||
types = self.proto["login"]["toServer"]["types"]
|
||
except Exception:
|
||
types = {}
|
||
if "packet_login_acknowledged" not in types:
|
||
return
|
||
w = Writer()
|
||
w.varint(self._serverbound_id("login", "packet_login_acknowledged"))
|
||
self.send(w.getvalue())
|
||
|
||
def _recv_packet(self, phase: str) -> dict:
|
||
"""读一包并解成 {name, params, _id, _len}。
|
||
|
||
★★ 关键韧性设计（实测下来这是必需的）：**用一个包坏掉不能断流**。
|
||
|
||
长度前缀是**帧层权威**，所以无论一个包的载荷能不能解，
|
||
下一个包的边界都精确可知 ⇒ 解不开就整包丢弃并标记
|
||
`_undecoded`，而不是抛异常让整条连接作废。
|
||
|
||
实测依据（2026-09-26）：
|
||
`declare_recipes`（108KB，1290 条配方）在
|
||
**minecraft-protocol + 官方 minecraft-data 3.117 上同样报零**
|
||
（PartialReadError: found 108271, expected 114153）
|
||
⇒ 这是**上游数据定义的缺口**，不是我们的解析器 bug。
|
||
若因它抛异常，play 阶段的 position/map_chunk 就永远收不到 ——
|
||
agent 会完全看不到世界，因为一条配方列表解不开。
|
||
|
||
⇒ 此路：坏包容错 + 布告（`undecoded` 计数），连接继续。
|
||
"""
|
||
body = self.read_packet()
|
||
try:
|
||
pkt = self.decode(phase, "toClient", body)
|
||
except Exception as e:
|
||
pid = -1
|
||
try:
|
||
pid = Reader(body).varint()
|
||
except Exception:
|
||
pass
|
||
name = self._name_by_id(phase, "toClient", pid)
|
||
self.decode_failures.append({"phase": phase, "name": name,
|
||
"len": len(body), "error": str(e)[:160]})
|
||
return {"name": name, "params": {}, "_raw": None,
|
||
"_len": len(body), "_undecoded": True, "_error": str(e)[:160]}
|
||
name = pkt.get("name")
|
||
if name in (None, "unknown"):
|
||
name = self._name_by_id(phase, "toClient", pkt.get("_pid", -1))
|
||
return {"name": name, "params": pkt.get("params", {}),
|
||
"_raw": pkt, "_len": len(body)}
|
||
|
||
def _name_by_id(self, phase: str, side: str, pid: int) -> str:
|
||
return self._packet_names(phase, side).get(pid, f"unknown_0x{pid:02x}")
|
||
|
||
def _decode_any(self, phase: str, side: str, body: bytes) -> dict:
|
||
"""解一个包并回包名（保留 params 形状，供流程控制）。"""
|
||
pd = self._pd(phase, side)
|
||
pkt = pd.read(self.proto[phase][side]["types"]["packet"], Reader(body))
|
||
if not isinstance(pkt, dict):
|
||
return {"name": None, "params": {}, "_raw": pkt}
|
||
pid = 0
|
||
try:
|
||
pid = Reader(body).varint()
|
||
except Exception:
|
||
pass
|
||
pkt["_id"] = pid
|
||
return pkt
|
||
|
||
def _packet_names(self, phase: str, side: str) -> dict[int, str]:
|
||
"""包 id → 包名（从 protocol.json 的 mapper 读；缓存）。
|
||
|
||
★ mapper 的实际形状（实测）：
|
||
['mapper', {'type': 'varint', 'mappings': {'0x00': 'disconnect', ...}}, 'string']
|
||
映射在 [1]['mappings']，**不是** [1] 本身 —— 我第一版把 [1] 当映射表，
|
||
结果所有包名都是 unknown。
|
||
"""
|
||
key = (phase, side)
|
||
if key in self._pkt_names:
|
||
return self._pkt_names[key]
|
||
out: dict[int, str] = {}
|
||
try:
|
||
tm = self.proto[phase][side]["types"].get("packet", {})
|
||
for f in tm[1]:
|
||
if f["name"] != "name":
|
||
continue
|
||
spec = f["type"] # ['mapper', {...}, 'string']
|
||
d = spec[1] if isinstance(spec, list) and len(spec) > 1 else {}
|
||
m = d.get("mappings", d) if isinstance(d, dict) else {}
|
||
for k, v in m.items():
|
||
if k == "type":
|
||
continue
|
||
out[int(k, 0) if isinstance(k, str) else int(k)] = v
|
||
break
|
||
except Exception:
|
||
pass
|
||
self._pkt_names[key] = out
|
||
return out
|
||
|
||
def _recv(self, phase: str) -> dict:
|
||
return self._recv_packet(phase)
|
||
|
||
def configuration_handshake(self) -> dict:
|
||
"""走完 configuration 阶段，拿到服务端的注册表（含 mod 方块）。
|
||
|
||
## 为什么这一步是 M2 的核心
|
||
|
||
1.20.2 起服务端在 login 后插入一个 **configuration 阶段**，
|
||
里面发 `registry_data`。它是**方块/物品注册表的权威来源** ——
|
||
而客户端本机 jar 里的注册表只包含自己装的东西：
|
||
**mod 方块的 state id 只有服务端知道**。
|
||
|
||
这就是「与服务端 mod 物品交互」的技术核心：
|
||
不读这个阶段，我们看到 mod 方块只会是 `unknown(id)`。
|
||
|
||
## 流程（实测 1.21.1）
|
||
|
||
login: success
|
||
→ 发 login_acknowledged （1.21+ 才有）
|
||
configuration:
|
||
收到 feature_flags / registry_data×N / tags / ...
|
||
keep_alive → 必须回 pong，否则超时断
|
||
→ 发 finish_configuration
|
||
configuration: finish_configuration（服务端回）
|
||
→ 进 play 阶段
|
||
"""
|
||
self.login_acknowledged()
|
||
self.phase = "configuration"
|
||
# ★ 先报本地设置（vanilla 行为，否则服务端等不到 finish_configuration）
|
||
self.send_settings()
|
||
raw: dict[str, Any] = {}
|
||
keepalive = 0
|
||
seen: list[str] = []
|
||
while True:
|
||
pkt = self._recv("configuration")
|
||
nm = pkt.get("name")
|
||
pr = pkt.get("params", {}) or {}
|
||
seen.append(str(nm))
|
||
if nm == "keep_alive":
|
||
keepalive += 1
|
||
kid = pr.get("keepAliveId", pr.get("keepAliveId", 0))
|
||
w = Writer()
|
||
w.varint(self._serverbound_id("configuration", "packet_keep_alive"))
|
||
w.int64(int(kid))
|
||
self.send(w.getvalue())
|
||
continue
|
||
if nm == "select_known_packs":
|
||
# ★ 必须回应，否则服务端不再发 registry_data（实测）
|
||
self.select_known_packs()
|
||
continue
|
||
if nm == "add_resource_pack":
|
||
# 资源包：同意/拒绝二选一。我们不装，**拒绝**
|
||
# （约定值 3 = 已成功加载的包集合里没有它，即 declined）
|
||
w = Writer()
|
||
w.varint(self._serverbound_id("configuration", "packet_resource_pack_receive"))
|
||
w.byte(3)
|
||
w.varint(0) # 已加载包数 = 0
|
||
self.send(w.getvalue())
|
||
continue
|
||
if nm == "registry_data":
|
||
rid = pr.get("id")
|
||
ents = pr.get("entries")
|
||
n = len(ents) if isinstance(ents, (list, tuple)) else 0
|
||
raw.setdefault("registries", {})[str(rid)] = n
|
||
# ★ 保留实际条目供上层建表（这是**权威**注册表）
|
||
self.registries[str(rid)] = ents
|
||
continue
|
||
if nm in ("finish_configuration", "disconnect"):
|
||
raw["finished"] = (nm == "finish_configuration")
|
||
if nm == "disconnect":
|
||
raw["reason"] = pr.get("reason")
|
||
break
|
||
if raw.get("finished"):
|
||
w = Writer()
|
||
w.varint(self._serverbound_id("configuration", "packet_finish_configuration"))
|
||
self.send(w.getvalue())
|
||
self.phase = "play"
|
||
raw["keepAlive"] = keepalive
|
||
raw["registryCount"] = len(self.registries)
|
||
raw["seenKinds"] = sorted(set(seen))
|
||
return raw
|
||
|
||
def send_settings(self) -> None:
|
||
"""configuration: settings（Client Information）—— **vanilla 客户端会发**。
|
||
|
||
★ 必需的原因（实测）：configuration 阶段是**双向协商**，
|
||
服务端等客户端报上本地化/视距/渲染偏好后才收尾发
|
||
finish_configuration；不发就一直等（表现为"挂住"）。
|
||
参照实现 minecraft-protocol/src/client/play.js:63-72 同样在
|
||
进 configuration 状态时立即写一份 settings。
|
||
|
||
★ 字段定义不在 configuration.toServer.types 里，而在**共享顶层**
|
||
`types.packet_common_settings` —— 我先前在阶段内找 `packet_settings`
|
||
得到 null，差点当成「不需要发」。
|
||
★ 配置值取 vanilla 默认（en_us / 视距 10 / 全部皮肤层 / 主手右手）。
|
||
"""
|
||
w = Writer()
|
||
w.varint(self._serverbound_id("configuration", "packet_settings"))
|
||
w.pstring("en_us") # locale
|
||
w.int8(8) # viewDistance（i8）
|
||
w.varint(0) # chatFlags
|
||
w.boolean(True) # chatColors
|
||
w.uint8(0x7F) # skinParts 127 = 全部
|
||
w.varint(1) # mainHand 1 = 右手
|
||
w.boolean(False) # enableTextFiltering
|
||
w.boolean(True) # enableServerListing
|
||
self.send(w.getvalue())
|
||
|
||
def select_known_packs(self) -> None:
|
||
"""configuration: select_known_packs → 回一份**空列表**（我们不认它的包）。
|
||
|
||
★ 实测（2026-09-26，服务端发完这个就不发 registry_data 了）：
|
||
configuration 阶段 = 双向协商，服务端会**等回应**才继续：
|
||
select_known_packs → 必须回（同理 cookie_request / resource_pack）
|
||
settings → 必须发
|
||
registry_data ×N → 只需要收
|
||
finish_configuration → 必须回
|
||
只收不回，registry_data 永远不来（实测卡在这）。
|
||
★ 字段定义在 protocol.json 里是 null（数据缺口），按参照实现
|
||
minecraft-protocol/src/client/play.js:78-80 用 `packs: []`。
|
||
"""
|
||
w = Writer()
|
||
w.varint(self._serverbound_id("configuration", "packet_select_known_packs"))
|
||
w.varint(0) # packs 数量 = 0（一个都不认）
|
||
self.send(w.getvalue())
|
||
|
||
def _serverbound_id(self, phase: str, pname: str) -> int:
|
||
"""包名 → 包 id（serverbound）。拿不到就报错（不猜）。
|
||
|
||
★ 包名有两种写法（实测）：
|
||
mapper 里是**短名** `login_acknowledged`
|
||
类型定义里是**长名** `packet_login_acknowledged`
|
||
两者都要能吃 —— 写死一种会在另一处失配
|
||
（我第一版只认长名，`packet_login_acknowledged` 查不到）。
|
||
"""
|
||
key = ("toServer", phase, pname)
|
||
if key in self._pid_cache:
|
||
return self._pid_cache[key]
|
||
short = pname[7:] if pname.startswith("packet_") else pname
|
||
tm = self.proto[phase]["toServer"]["types"].get("packet", {})
|
||
found = None
|
||
for f in tm[1]:
|
||
if f["name"] != "name":
|
||
continue
|
||
spec = f["type"]
|
||
d = spec[1] if isinstance(spec, list) and len(spec) > 1 else {}
|
||
m = d.get("mappings", d) if isinstance(d, dict) else {}
|
||
for k, v in m.items():
|
||
if v in (pname, short):
|
||
found = int(k, 0) if isinstance(k, str) else int(k)
|
||
break
|
||
break
|
||
if found is None:
|
||
raise ProtocolError(f"{phase} 阶段无此 serverbound 包: {pname}")
|
||
self._pid_cache[key] = found
|
||
return found
|
||
|
||
def decode(self, phase: str, side: str, body: bytes) -> dict:
|
||
pd = self._pd(phase, side)
|
||
pkt = pd.read(self.proto[phase][side]["types"]["packet"], Reader(body))
|
||
return pkt
|
||
|
||
def run_smoke(self) -> dict:
|
||
"""握手 → 登录 → 读若干包，返回摘要。"""
|
||
self.load_protocol()
|
||
self.connect()
|
||
self.handshake(state=2)
|
||
self.login_start()
|
||
|
||
got: list[str] = []
|
||
login_ok = False
|
||
for _ in range(20):
|
||
body = self.read_packet()
|
||
try:
|
||
pkt = self.decode("login", "toClient", body)
|
||
except ProtoError:
|
||
break
|
||
name = pkt.get("name")
|
||
got.append(name)
|
||
if name == "compress":
|
||
self.compression = pkt.get("params", {}).get("threshold", 256)
|
||
elif name == "success":
|
||
login_ok = True
|
||
break
|
||
elif name == "disconnect":
|
||
raise ProtocolError(f"服务端拒绝: {pkt.get('params', {}).get('reason')}")
|
||
return {"login_ok": login_ok, "packets": got, "protocol": self.proto_num}
|
||
|
||
def close(self) -> None:
|
||
if self.sock:
|
||
try:
|
||
self.sock.close()
|
||
except OSError:
|
||
pass
|
||
self.sock = None
|