implement packet classes for encoding and decoding

mcclient/base_client.py

@@ -1,21 +1,14 @@
 import socket
-import struct
 
 from mcclient.address import Address
-from mcclient.encoding.packet import Packet
-from mcclient.encoding.varint import pack_varint, read_varint
+from mcclient.packet import OutboundPacket
 
 DEFAULT_HOST = "localhost"
 DEFAULT_PORT = 25565
 DEFAULT_TIMEOUT = 5
 DEFAULT_PROTO = 47
 
 
-class IncompletePacket(Exception):
-    def __init__(self, size, missing):
-        super().__init__(f"Incomplete packet: missing {missing} from {size} bytes.")
-
-
 class BaseClient:
     address: Address
     sock: socket.socket
@@ -44,35 +37,12 @@ def _connect(self) -> None:
         self.connected = True
 
     def _handshake(self, next_state: int = 1) -> None:
-        packet = Packet(
-            b"\x00",  # packet id
-            pack_varint(self.protocol_version),
-            self.address.get_host(),
-            struct.pack(">H", self.address.get_port()),
-            pack_varint(next_state)  # next state 1 for status request
-        )
-        self._send(packet)
-
-    def _send(self, packet: Packet) -> int:
-        return self.sock.send(packet.pack())
-
-    def _recv(self) -> tuple[int, bytes]:
-        length = read_varint(self.sock)
-        packet_id = read_varint(self.sock)
-        return packet_id, self._recv_bytes(length)
-
-    def _recv_bytes(self, length: int) -> bytes:
-        received = 0
-        data = b""
-        while received < length - len(pack_varint(length)):
-            chunk = self.sock.recv(length - received)
-            data += chunk
-            received += len(chunk)
-
-            if chunk == b"":
-                raise IncompletePacket(length, length - received)
-
-        return data
+        p = OutboundPacket(0)
+        p.write_varint(self.protocol_version)
+        p.write_string(self.address.get_host())
+        p.write_ushort(self.address.get_port())
+        p.write_varint(next_state)
+        p.write(self.sock)
 
     def _close(self):
         self.sock.close()

mcclient/packet/__init__.py

@@ -0,0 +1,2 @@
+from mcclient.packet.inbound import InboundPacket, IncompletePacket
+from mcclient.packet.outbound import OutboundPacket

mcclient/packet/inbound.py

@@ -0,0 +1,122 @@
+import socket
+import struct
+
+from mcclient.packet.varint import read, pack, unpack
+
+
+class IncompletePacket(Exception):
+    def __init__(self, size, missing):
+        super().__init__(f"Incomplete packet: missing {missing} from {size} bytes.")
+
+
+class InboundPacket:
+    """
+    Represents an inbound packet received from a socket.
+    """
+
+    def __init__(self, sock: socket.socket):
+        """
+        Initializes the InboundPacket by receiving data from a socket.
+
+        Args:
+            sock (socket.socket): The socket from which the packet is received.
+        """
+
+        self.sock = sock
+        self.length = read(self.sock)
+        self.id = read(self.sock)
+        id_size = len(pack(self.id))
+
+        remaining = self.length - id_size
+        self.data = b""
+        while len(self.data) < remaining:
+            chunk = self.sock.recv(remaining - len(self.data))
+            if chunk == b"":
+                packet_size = len(pack(self.length)) + id_size + self.length
+                raise IncompletePacket(packet_size, self.length - id_size - len(self.data))
+
+            self.data += chunk
+
+    def read_int(self) -> int:
+        """
+        Reads a 4-byte integer from the data.
+
+        Returns:
+            int: The read integer value.
+        """
+        return struct.unpack('>i', self.read_bytes(4))[0]
+
+    def read_short(self) -> int:
+        """
+        Reads a 2-byte short integer from the data.
+
+        Returns:
+            int: The read short integer value.
+        """
+        return struct.unpack('>h', self.read_bytes(2))[0]
+
+    def read_ushort(self) -> int:
+        """
+        Reads a 2-byte unsigned short integer from the data.
+
+        Returns:
+            int: The read unsigned short integer value.
+        """
+        return struct.unpack('>H', self.read_bytes(2))[0]
+
+    def read_long(self) -> int:
+        """
+        Reads an 8-byte long integer from the data.
+
+        Returns:
+            int: The read long integer value.
+        """
+        return struct.unpack('>q', self.read_bytes(8))[0]
+
+    def read_varint(self) -> int:
+        """
+        Reads a variable-length integer (Varint) from the data.
+
+        Returns:
+            int: The read Varint value.
+        """
+        varint, size = unpack(self.data)
+        self.read_bytes(size)
+        return varint
+
+    def read_bool(self) -> bool:
+        """
+        Reads a boolean value from the data.
+
+        Returns:
+            bool: The read boolean value.
+        """
+        value = self.data[0] == 1
+        self.data = self.data[1:]
+        return value
+
+    def read_string(self) -> str:
+        """
+        Reads a string from the data.
+
+        Returns:
+            str: The read string.
+        """
+        length = self.read_varint()
+        string = self.data[:length].decode('utf-8')
+        self.data = self.data[length:]
+        return string
+
+    def read_bytes(self, length: int) -> bytes:
+        """
+        Reads a specified number of bytes from the data.
+
+        Args:
+            length (int): The number of bytes to read.
+
+        Returns:
+            bytes: The read bytes.
+        """
+        bytes_data = self.data[:length]
+        self.data = self.data[length:]
+        return bytes_data

mcclient/packet/outbound.py

@@ -0,0 +1,117 @@
+import socket
+import struct
+
+from mcclient.packet.varint import pack
+
+
+class OutboundPacket:
+    """
+    Represents an outbound packet that can be sent over a socket.
+    """
+
+    def __init__(self, packet_id: int):
+        """
+        Initializes the OutboundPacket with the specified packet ID.
+
+        Args:
+            packet_id (int): The packet ID representing the type of packet.
+        """
+        self.id = packet_id
+        self.data = b""
+
+    def write_int(self, value: int) -> None:
+        """
+        Writes a 4-byte integer to the packet.
+
+        Args:
+            value (int): The integer value to write.
+        """
+        self.data += struct.pack('>i', value)
+
+    def write_short(self, value: int) -> None:
+        """
+        Writes a 2-byte short integer to the packet.
+
+        Args:
+            value (int): The short integer value to write.
+        """
+        self.data += struct.pack('>h', value)
+
+    def write_ushort(self, value: int) -> None:
+        """
+        Writes a 2-byte unsigned short integer to the packet.
+
+        Args:
+            value (int): The unsigned short integer value to write.
+        """
+        self.data += struct.pack('>H', value)
+
+    def write_long(self, value: int) -> None:
+        """
+        Writes an 8-byte long integer to the packet.
+
+        Args:
+            value (int): The long integer value to write.
+        """
+        self.data += struct.pack('>q', value)
+
+    def write_varint(self, varint: int) -> None:
+        """
+        Writes a variable-length integer (VarInt) to the packet.
+
+        Args:
+            varint (int): The VarInt value to write.
+        """
+        self.data += pack(varint)
+
+    def write_bool(self, value: bool) -> None:
+        """
+        Writes a boolean value to the packet as a single byte.
+
+        Args:
+            value (bool): The boolean value to write. True is represented
+                          as 0x01, and False as 0x00.
+        """
+        self.data += b"\x01" if value else b"\x00"
+
+    def write_string(self, string: str) -> None:
+        """
+        Writes a UTF-8 encoded string to the packet.
+
+        Args:
+            string (str): The string to write.
+        """
+        self.write_varint(len(string))
+        self.data += string.encode('utf-8')
+
+    def write_bytes(self, data: bytes) -> None:
+        """
+        Writes raw bytes directly to the packet.
+
+        Args:
+            data (bytes): The bytes to write.
+        """
+        self.data += data
+
+    def pack(self) -> bytes:
+        """
+        Packs the packet data, including the packet ID and length, into a
+        single byte sequence.
+
+        Returns:
+            bytes: The packed packet ready for transmission.
+        """
+        data = pack(self.id) + self.data
+        return pack(len(data)) + data
+
+    def write(self, sock: socket.socket) -> None:
+        """
+        Sends the packet over the specified socket.
+
+        This method packs the packet, including the packet ID and length,
+        and then sends it to the remote socket.
+
+        Args:
+            sock (socket.socket): The socket to send the packet over.
+        """
+        sock.send(self.pack())