190 lines
6.1 KiB
Python
190 lines
6.1 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
|
||
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:
|
||
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:
|
||
assert self.sock
|
||
# 未启用压缩前：包 = varint(长度) + 包体
|
||
w = Writer()
|
||
w.varint(len(payload))
|
||
w.write(payload)
|
||
self.sock.sendall(w.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（离线模式只发用户名）。"""
|
||
w = Writer()
|
||
w.varint(0x00)
|
||
w.pstring(self.username)
|
||
w.varint(0) # uuid 可选，0 表示无
|
||
self.send(w.getvalue())
|
||
|
||
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
|