"""验证 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]}") # ---------- M2 新增：configuration/play 阶段真正用到的复合类型 ---------- # # 这些判据用**构造的字节**（不是观察当前输出），所以能真失败： # 每条都先验输入 → 期望输出，换个实现形态就红。 DATA1211='/home/newqqagent/plugin_data/mc/runtime/node_modules/minecraft-data/minecraft-data/data/pc/1.21.1/protocol.json' def _pd_1211(): d = json.load(open(DATA1211)) return d, for_phase(d, "play", "toClient") def test_bitflags_splits_bits(): """bitflags：一个整数拆成多个布尔位（position 包的 flags 靠它）。 ★ 判据动机：实现它之前 position 整个包解不开 ⇒ 看不到玩家位置。 """ d, pd = _pd_1211() t = d["play"]["toClient"]["types"]["PositionUpdateRelatives"] # 0b10101 = x(1) + z(4) + pitch(16) 置位 r = Reader(bytes([0b10101])) got = pd.read(t, r, {}) assert got["_value"] == 0b10101 assert got["x"] is True and got["z"] is True and got["pitch"] is True assert got["y"] is False and got["yaw"] is False # 负向对照：0 必须全 false（否则「恒真」实现会假通过） got0 = pd.read(t, Reader(bytes([0])), {}) assert not any(v for k, v in got0.items() if k != "_value") print(" ✓ bitflags 拆位（含 0 的负向对照）") def test_bitfield_slices_fields(): """bitfield：按位宽切子字段（declare_commands 的 flags 靠它）。 ★ 位序是 **MSB 优先**（高位先给前面的字段）——实测判据： 线上 declare_commands 解出 25 个节点，/help 那条正好是 has_command=1 且 command_node_type=1，组合自洽。 我一开始按 LSB 假设写判据，结果在**正确实现**上判红 —— 又一次 「判据的假设必须来自被测事实，不能来自我的直觉」。 构造（MSB 优先）：unused(3)=0b101, has_custom_suggestions=1, has_redirect_node=0, has_command=1, command_node_type=0b10 → 0b101 1 0 1 10 = 0b10110110 = 182 """ d, pd = _pd_1211() spec = ["bitfield", [ {"name": "unused", "size": 3}, {"name": "has_custom_suggestions", "size": 1}, {"name": "has_redirect_node", "size": 1}, {"name": "has_command", "size": 1}, {"name": "command_node_type", "size": 2}, ]] got = pd.read(spec, Reader(bytes([0b10110110])), {}) assert got["unused"] == 0b101, got assert got["has_custom_suggestions"] == 1, got assert got["has_redirect_node"] == 0, got assert got["has_command"] == 1, got assert got["command_node_type"] == 0b10, got # 负向对照：全 0 必须全 0（防「恒真」实现） z = pd.read(spec, Reader(bytes([0])), {}) assert all(v == 0 for v in z.values()), z print(" ✓ bitfield 按位宽切分（MSB 优先，含全 0 负向对照）") def test_array_count_uses_ctx_reference(): """数组的 count 可以引用**已读字段**，此时不消费字节。 ★ 这是 declare_recipes 流错位的根因：我只认 countType， 遇到 count 就多读一个字节 ⇒ 后面全乱（报 switch type=948/101）。 构造：先读 varint=3（n），再用 count:'n' 读 3 个 u8。 """ d, pd = _pd_1211() spec = ["container", [ {"name": "n", "type": "varint"}, {"name": "vals", "type": ["array", {"count": "n", "type": "u8"}]}, {"name": "tail", "type": "u8"}, ]] payload = bytes([3, 10, 20, 30, 99]) # n=3, vals=[10,20,30], tail=99 got = pd.read(spec, Reader(payload), {}) assert got["n"] == 3 assert got["vals"] == [10, 20, 30], got # ★ 关键判据：count 不消耗流 ⇒ tail 必须是 99 而不是错位读到的值 assert got["tail"] == 99, f"流未对齐，tail={got['tail']}" assert Reader(payload).varint() == 3 print(" ✓ array count 引用 ctx（不消费字节，tail 对齐）") def test_switch_compareto_path_expression(): """switch 的 compareTo 支持路径：`flags/x`、`../flags/x`。 ★ 全协议 12 处路径判据；只做平铺 ctx.get 会让 declare_commands / player_info 整包解不开。 """ d, pd = _pd_1211() spec = ["container", [ {"name": "flags", "type": ["bitfield", [ {"name": "has_x", "size": 1}, {"name": "unused", "size": 7}]]}, {"name": "payload", "type": ["switch", { "compareTo": "flags/has_x", "fields": {"1": "u8"}, "default": "void"}]}, ]] # MSB 优先：has_x 是最高位 ⇒ 0x80 时 has_x=1 ⇒ payload 读 1 字节 got = pd.read(spec, Reader(bytes([0x80, 42])), {}) assert got["flags"]["has_x"] == 1, got assert got["payload"] == 42, got # 负向对照：has_x=0 ⇒ payload 走 void，不消费 got0 = pd.read(spec, Reader(bytes([0x00, 42])), {}) assert got0["flags"]["has_x"] == 0, got0 assert got0["payload"] is None print(" ✓ switch compareTo 路径表达式（含 void 负向对照）") def test_entity_metadata_loop_terminates_on_endval(): """entityMetadataLoop 读到 key==endVal 就停（不是读到流尽头）。 ★ 终止符是**光秃秃一个字节** 0xFF，后面**没有** type/value。 我第一版写成「读完整个条目再比 key」⇒ 终止时多读 2 字节 ⇒ 越界， 是线上真实包（entityId=144，13 字节）抓出来的。 参照 minecraft-protocol/src/datatypes/minecraft.js:120-128。 构造：条目1 = {key:0, type:0(byte), val:1}；然后单个 0xFF 终止； 后面还剩 3 字节必须**留着不动**。 """ d, pd = _pd_1211() spec = ["entityMetadataLoop", {"endVal": 255, "type": "entityMetadataEntry"}] payload = bytes([0x00, 0x00, 0x01, # key=0, type=0(byte), val=1 0xFF]) + bytes([7, 7, 7]) r = Reader(payload) got = pd.read(spec, r, {}) # ★ 只有 1 个真条目：0xFF 是**终止符本身**，不是「key=255 的条目」 assert isinstance(got, list) and len(got) == 1, got assert got[0]["key"] == 0 and got[0]["value"] == 1, got # 终止只消费那 1 字节 ⇒ 后续 3 字节必须留着 assert r.remaining() == 3, f"提前吃掉了后续数据，剩 {r.remaining()}" print(" ✓ entityMetadataLoop 按光秃终止符停止（后续字节不越界）") def test_anonymous_nbt_has_no_root_name(): """anonymousNbt：无根名字的 compound（1.20.2+ 网络 NBT 形态）。 ★ registry_data 的条目就是它。若按具名 NBT 读（多读 2 字节名长 + 名） 会整体错位。 """ d, pd = _pd_1211() # TAG_Compound(10) + 一个 TAG_Int(3) 字段 "n" = 7 + payload # 载荷: 03 0001 6e 00000007 00 payload = bytes([0x0A, 0x03, 0x00, 0x01, 0x6E, 0x00, 0x00, 0x00, 0x07, 0x00]) got = pd.read("anonymousNbt", Reader(payload), {}) assert got == {"n": 7}, got # 负向对照：同一段字节按具名 nbt 读会因为多读名字而错 print(" ✓ anonymousNbt 无根名 compound") def test_entity_metadata_real_server_bytes(): """线上真实包回归：entity_metadata（entityId=144，13 字节）。 ★ 抓 bug 的就是这个包 —— 我的 entityMetadataLoop 早前把终止符 当成「带 type/value 的条目」⇒ 多读 2 字节 ⇒ 越界 ⇒ 整包丢弃。 判据直接用线上抓到的字节，不自己构造。 """ d, pd = _pd_1211() body = bytes.fromhex("589001090341a0000011007fff") r = Reader(body) pid = r.varint() assert pid == 0x58, f"entity_metadata 的包 id 应为 0x58，实得 {pid:#x}" eid = r.varint() assert eid == 144, eid got = pd.read("entityMetadata", r, {}) assert isinstance(got, list) and len(got) == 2, got assert [e["key"] for e in got] == [9, 17], got assert r.remaining() == 0, f"未完全对齐，剩 {r.remaining()}" print(" ✓ 线上真实 entity_metadata 字节完整解开（0 余字节）") 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, # M2：configuration/play 阶段依赖的复合类型 test_bitflags_splits_bits, test_bitfield_slices_fields, test_array_count_uses_ctx_reference, test_switch_compareto_path_expression, test_entity_metadata_loop_terminates_on_endval, test_anonymous_nbt_has_no_root_name, test_entity_metadata_real_server_bytes]: fn() print("\nprotodef 解释器：全部通过")