diff --git a/opengsq/protocols/raknet.py b/opengsq/protocols/raknet.py index 4393871..a483cf2 100644 --- a/opengsq/protocols/raknet.py +++ b/opengsq/protocols/raknet.py @@ -1,3 +1,4 @@ +from opengsq.responses.raknet import Status from opengsq.binary_reader import BinaryReader from opengsq.exceptions import InvalidPacketException from opengsq.protocol_base import ProtocolBase @@ -5,17 +6,33 @@ class RakNet(ProtocolBase): - """RakNet Protocol (https://wiki.vg/Raknet_Protocol)""" - full_name = 'RakNet Protocol' - - __ID_UNCONNECTED_PING = b'\x01' - __ID_UNCONNECTED_PONG = b'\x1C' - __TIMESTAMP = b'\x12\x23\x34\x45\x56\x67\x78\x89' - __OFFLINE_MESSAGE_DATA_ID = b'\x00\xFF\xFF\x00\xFE\xFE\xFE\xFE\xFD\xFD\xFD\xFD\x12\x34\x56\x78' - __CLIENT_GUID = b'\x00\x00\x00\x00\x00\x00\x00\x00' - - async def get_status(self) -> dict: - request = self.__ID_UNCONNECTED_PING + self.__TIMESTAMP + self.__OFFLINE_MESSAGE_DATA_ID + self.__CLIENT_GUID + """ + This class represents the RakNet Protocol. It provides methods to interact with the RakNet API. + (https://wiki.vg/Raknet_Protocol) + """ + + full_name = "RakNet Protocol" + + __ID_UNCONNECTED_PING = b"\x01" + __ID_UNCONNECTED_PONG = b"\x1C" + __TIMESTAMP = b"\x12\x23\x34\x45\x56\x67\x78\x89" + __OFFLINE_MESSAGE_DATA_ID = ( + b"\x00\xFF\xFF\x00\xFE\xFE\xFE\xFE\xFD\xFD\xFD\xFD\x12\x34\x56\x78" + ) + __CLIENT_GUID = b"\x00\x00\x00\x00\x00\x00\x00\x00" + + async def get_status(self) -> Status: + """ + Asynchronously retrieves the status of the game server. + + :return: A Status object containing the status of the game server. + """ + request = ( + self.__ID_UNCONNECTED_PING + + self.__TIMESTAMP + + self.__OFFLINE_MESSAGE_DATA_ID + + self.__CLIENT_GUID + ) response = await UdpClient.communicate(self, request) br = BinaryReader(response) @@ -23,51 +40,51 @@ async def get_status(self) -> dict: if header != self.__ID_UNCONNECTED_PONG: raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {}.' - .format(header, self.__ID_UNCONNECTED_PONG) + "Packet header mismatch. Received: {}. Expected: {}.".format( + header, self.__ID_UNCONNECTED_PONG + ) ) - br.read_bytes(len(self.__TIMESTAMP) + len(self.__CLIENT_GUID)) # skip timestamp and guid + br.read_bytes( + len(self.__TIMESTAMP) + len(self.__CLIENT_GUID) + ) # skip timestamp and guid magic = br.read_bytes(len(self.__OFFLINE_MESSAGE_DATA_ID)) if magic != self.__OFFLINE_MESSAGE_DATA_ID: raise InvalidPacketException( - 'Magic value mismatch. Received: {}. Expected: {}.' - .format(magic, self.__OFFLINE_MESSAGE_DATA_ID) + "Magic value mismatch. Received: {}. Expected: {}.".format( + magic, self.__OFFLINE_MESSAGE_DATA_ID + ) ) br.read_short() # skip remaining packet length - d = [b';'] # delimiter - result = {} - result['edition'] = br.read_string(d) - result['motd_line_1'] = br.read_string(d) - result['protocol_version'] = int(br.read_string(d)) - result['version'] = br.read_string(d) - result['num_players'] = int(br.read_string(d)) - result['max_players'] = int(br.read_string(d)) - - # Try to read all data - try: - result['server_uid'] = br.read_string(d) - result['motd_line_2'] = br.read_string(d) - result['gamemode'] = br.read_string(d) - result['gamemode_numeric'] = int(br.read_string(d)) - result['port_ipv4'] = int(br.read_string(d)) - result['port_ipv6'] = int(br.read_string(d)) - except Exception: - pass - - return result - - -if __name__ == '__main__': + d = [b";"] # delimiter + + return Status( + edition=br.read_string(d), + motd_line1=br.read_string(d), + protocol_version=int(br.read_string(d)), + version_name=br.read_string(d), + num_players=int(br.read_string(d)), + max_players=int(br.read_string(d)), + server_unique_id=br.read_string(d), + motd_line2=br.read_string(d), + game_mode=br.read_string(d), + game_mode_numeric=int(br.read_string(d)), + port_ipv4=int(br.read_string(d)), + port_ipv6=int(br.read_string(d)), + ) + + +if __name__ == "__main__": import asyncio import json + from dataclasses import asdict async def main_async(): - raknet = RakNet(host='mc.advancius.net', port=19132, timeout=5.0) + raknet = RakNet(host="mc.advancius.net", port=19132, timeout=5.0) status = await raknet.get_status() - print(json.dumps(status, indent=None) + '\n') + print(json.dumps(asdict(status), indent=None) + "\n") asyncio.run(main_async()) diff --git a/opengsq/responses/raknet/__init__.py b/opengsq/responses/raknet/__init__.py new file mode 100644 index 0000000..2f91dae --- /dev/null +++ b/opengsq/responses/raknet/__init__.py @@ -0,0 +1 @@ +from .status import Status diff --git a/opengsq/responses/raknet/status.py b/opengsq/responses/raknet/status.py new file mode 100644 index 0000000..de3f1a0 --- /dev/null +++ b/opengsq/responses/raknet/status.py @@ -0,0 +1,44 @@ +from dataclasses import dataclass + + +@dataclass +class Status: + """ + Represents the status response from a Minecraft server. + """ + + edition: str + """The edition of the server (MCPE or MCEE for Education Edition).""" + + motd_line1: str + """The first line of the Message of the Day (MOTD).""" + + protocol_version: int + """The protocol version of the server.""" + + version_name: str + """The version name of the server.""" + + num_players: int + """The number of players currently on the server.""" + + max_players: int + """The maximum number of players that can join the server.""" + + server_unique_id: str + """The unique ID of the server.""" + + motd_line2: str + """The second line of the Message of the Day (MOTD).""" + + game_mode: str + """The game mode of the server.""" + + game_mode_numeric: int + """The numeric representation of the game mode.""" + + port_ipv4: int + """The IPv4 port of the server.""" + + port_ipv6: int + """The IPv6 port of the server."""