105 lines
4.3 KiB
Python
105 lines
4.3 KiB
Python
"""验证 protodef 解释器能真跑协议——用真实握手字节。"""
|
||
import sys, os, json, struct
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from mcfold.protodef import ProtoDef, Reader, Writer, ProtoError, for_phase
|
||
|
||
DATA='/home/newqqagent/plugin_data/mc/runtime/node_modules/minecraft-data/minecraft-data/data/pc/1.20/protocol.json'
|
||
|
||
def load():
|
||
d=json.load(open(DATA))
|
||
return d
|
||
|
||
def pd_for(d, phase, side):
|
||
return for_phase(d, phase, side)
|
||
|
||
def test_varint_roundtrip():
|
||
"""varint 是 MC 最核心的原语；含负数回绕。"""
|
||
load()
|
||
for v in [0,1,127,128,255,300,25565,2097151,-1,-2147483648,2147483647]:
|
||
w=Writer(); w.varint(v)
|
||
got=Reader(w.getvalue()).varint()
|
||
assert got==v, f"varint {v} -> {got}"
|
||
print(" ✓ varint 往返（含负数/边界）")
|
||
|
||
def test_pstring():
|
||
load()
|
||
for s in ["", "abc", "中文名字", "x"*300]:
|
||
w=Writer(); w.pstring(s)
|
||
assert Reader(w.getvalue()).pstring()==s
|
||
print(" ✓ pstring 往返（含 UTF-8/长串）")
|
||
|
||
def test_real_handshake():
|
||
"""构造真实的 handshake+login_start 字节并解出（最硬的检验）。"""
|
||
d=load()
|
||
pd=pd_for(d,'handshaking','toServer')
|
||
# handshaking toServer packet_set_protocol
|
||
w=Writer()
|
||
w.varint(0x00) # packet id (handshaking)
|
||
w.varint(763) # protocol version = 1.20.1
|
||
w.pstring("127.0.0.1") # server address
|
||
w.write(struct.pack(">H", 25565)) # port
|
||
w.varint(2) # next state = login
|
||
raw=w.getvalue()
|
||
r=Reader(raw)
|
||
pkt=pd.read(d['handshaking']['toServer']['types']['packet'], r)
|
||
assert pkt['name']=='set_protocol', pkt
|
||
p = pkt['params'] # switch 的参数落在 params 下
|
||
assert p['protocolVersion']==763, p
|
||
assert p['serverHost']=='127.0.0.1', p
|
||
assert p['serverPort']==25565, p
|
||
assert p['nextState']==2, p
|
||
assert r.remaining()==0, f"有残留字节: {r.remaining()}"
|
||
print(f" ✓ 真实 handshake 解码: proto={p['protocolVersion']} host={p['serverHost']}:{p['serverPort']} nextState={p['nextState']}")
|
||
|
||
def test_login_success_decode():
|
||
"""packet_success: uuid + username + properties（真实形状）。"""
|
||
d=load()
|
||
pd=pd_for(d,'login','toClient')
|
||
w=Writer()
|
||
w.varint(0x02) # login toClient packet_success id
|
||
w.write(bytes(range(16))) # uuid
|
||
w.pstring("BotName")
|
||
w.varint(0) # properties count = 0
|
||
r=Reader(w.getvalue())
|
||
pkt=pd.read(d['login']['toClient']['types']['packet'], r)
|
||
assert pkt['name']=='success', pkt
|
||
p = pkt['params']
|
||
assert p['username']=='BotName', p
|
||
assert p['properties']==[], p
|
||
print(f" ✓ login success 解码: uuid={p['uuid'][:8]}… username={p['username']}")
|
||
|
||
def test_play_packet_mapper():
|
||
"""play 阶段 111 个包由 mapper 分发 —— 验证能正确路由。"""
|
||
d=load()
|
||
pd=pd_for(d,'play','toClient')
|
||
w=Writer(); w.varint(0x00)
|
||
# 0x00 = bundle_delimiter（无字段）
|
||
r=Reader(w.getvalue())
|
||
pkt=pd.read(d['play']['toClient']['types']['packet'], r)
|
||
assert pkt['name']=='bundle_delimiter', pkt
|
||
print(f" ✓ play 包路由: 0x00 -> {pkt['name']}")
|
||
# 0x10 = declare_commands（我们做 mod 命令发现要用的包）
|
||
w2=Writer(); w2.varint(0x10)
|
||
r2=Reader(w2.getvalue())
|
||
try:
|
||
pkt2=pd.read(d['play']['toClient']['types']['packet'], r2)
|
||
print(f" ✓ declare_commands 路由: 0x10 -> {pkt2.get('name')}")
|
||
except ProtoError as e:
|
||
print(f" ~ declare_commands 需要更多字节（预期，因为只喂了 id）: {str(e)[:40]}")
|
||
|
||
def test_no_overshoot():
|
||
"""越界必须报错而不是静默读错 —— 这是我实测 mineflayer 静默失真的对照。"""
|
||
d=load()
|
||
pd=pd_for(d,'login','toClient')
|
||
r=Reader(b"\x05") # varint 说 5 字节字符串，实际没有
|
||
try:
|
||
pd.read("string", r); assert False, "应当抛错"
|
||
except ProtoError as e:
|
||
print(f" ✓ 越界显式报错（不静默）: {str(e)[:50]}")
|
||
|
||
if __name__=="__main__":
|
||
for fn in [test_varint_roundtrip, test_pstring, test_real_handshake,
|
||
test_login_success_decode, test_play_packet_mapper, test_no_overshoot]:
|
||
fn()
|
||
print("\nprotodef 解释器：全部通过")
|