diff --git a/opengsq/protocols/doom3.py b/opengsq/protocols/doom3.py index 28c7225..0dee5fb 100644 --- a/opengsq/protocols/doom3.py +++ b/opengsq/protocols/doom3.py @@ -7,30 +7,41 @@ class Doom3(ProtocolBase): - """Doom3 Protocol""" - full_name = 'Doom3 Protocol' + """ + This class represents the Doom3 Protocol. It provides methods to interact with the Doom3 API. + """ + + full_name = "Doom3 Protocol" _player_fields = { - 'doom': ['id', 'ping', 'rate', 'name'], - 'quake4': ['id', 'ping', 'rate', 'name', 'clantag'], - 'etqw': ['id', 'ping', 'name', 'clantag_pos', 'clantag', 'typeflag'] + "doom": ["id", "ping", "rate", "name"], + "quake4": ["id", "ping", "rate", "name", "clantag"], + "etqw": ["id", "ping", "name", "clantag_pos", "clantag", "typeflag"], } - async def get_info(self, strip_color=True): - request = b'\xFF\xFFgetInfo\x00ogsq\x00' + async def get_info(self, strip_color=True) -> dict: + """ + Asynchronously retrieves the information of the game server. + + :param strip_color: A boolean indicating whether to strip color codes from the information. + :return: A dictionary containing the information of the game server. + """ + request = b"\xFF\xFFgetInfo\x00ogsq\x00" response = await UdpClient.communicate(self, request) # Remove the first two 0xFF br = BinaryReader(response[2:]) header = br.read_string() - if header != 'infoResponse': - raise InvalidPacketException(f'Packet header mismatch. Received: {header}. Expected: infoResponse.') + if header != "infoResponse": + raise InvalidPacketException( + f"Packet header mismatch. Received: {header}. Expected: infoResponse." + ) # Read challenge br.read_bytes(4) - if br.read_bytes(4) != b'\xff\xff\xff\xff': + if br.read_bytes(4) != b"\xff\xff\xff\xff": br.stream_position -= 4 info = {} @@ -38,7 +49,7 @@ async def get_info(self, strip_color=True): # Read protocol version minor = br.read_short() major = br.read_short() - info['version'] = f'{major}.{minor}' + info["version"] = f"{major}.{minor}" # Read packet size if br.read_long() != br.remaining_bytes(): @@ -49,7 +60,7 @@ async def get_info(self, strip_color=True): key = br.read_string().strip() val = br.read_string().strip() - if key == '' and val == '': + if key == "" and val == "": break info[key] = Doom3.strip_colors(val) if strip_color else val @@ -59,23 +70,33 @@ async def get_info(self, strip_color=True): # Try parse the fields for mod in self._player_fields.keys(): try: - info['players'] = self.__parse_player(br, self._player_fields[mod], strip_color) + info["players"] = self.__parse_player( + br, self._player_fields[mod], strip_color + ) break except Exception: - info['players'] = [] + info["players"] = [] br.stream_position = stream_position return info def __parse_player(self, br: BinaryReader, fields: list, strip_color: bool): + """ + Parses the player information from the BinaryReader object. + + :param br: The BinaryReader object containing the player information. + :param fields: A list of fields to parse from the BinaryReader object. + :param strip_color: A boolean indicating whether to strip color codes from the player information. + :return: A list of dictionaries containing the player information. + """ players = [] def value_by_field(field: str, br: BinaryReader): - if field == 'id' or field == 'clantag_pos' or field == 'typeflag': + if field == "id" or field == "clantag_pos" or field == "typeflag": return br.read_byte() - elif field == 'ping': + elif field == "ping": return br.read_short() - elif field == 'rate': + elif field == "rate": return br.read_long() string = br.read_string() @@ -85,7 +106,7 @@ def value_by_field(field: str, br: BinaryReader): while True: player = {field: value_by_field(field, br) for field in fields} - if player['id'] == 32: + if player["id"] == 32: break players.append(player) @@ -93,29 +114,34 @@ def value_by_field(field: str, br: BinaryReader): return players @staticmethod - def strip_colors(text: str): - """Strip color codes""" - return re.compile('\\^(X.{6}|.)').sub('', text) + def strip_colors(text: str) -> str: + """ + Strips color codes from the given text. + + :param text: The text to strip color codes from. + :return: The text with color codes stripped. + """ + return re.compile("\\^(X.{6}|.)").sub("", text) -if __name__ == '__main__': +if __name__ == "__main__": import asyncio import json async def main_async(): # doom3 - doom3 = Doom3(host='66.85.14.240', port=27666, timeout=5.0) + doom3 = Doom3(host="66.85.14.240", port=27666, timeout=5.0) info = await doom3.get_info() - print(json.dumps(info, indent=None) + '\n') + print(json.dumps(info, indent=None) + "\n") # etqw - doom3 = Doom3(host='178.162.135.83', port=27735, timeout=5.0) + doom3 = Doom3(host="178.162.135.83", port=27735, timeout=5.0) info = await doom3.get_info() - print(json.dumps(info, indent=None) + '\n') + print(json.dumps(info, indent=None) + "\n") # quake4 - doom3 = Doom3(host='88.99.0.7', port=28007, timeout=5.0) + doom3 = Doom3(host="88.99.0.7", port=28007, timeout=5.0) info = await doom3.get_info() - print(json.dumps(info, indent=None) + '\n') + print(json.dumps(info, indent=None) + "\n") asyncio.run(main_async()) diff --git a/opengsq/protocols/eos.py b/opengsq/protocols/eos.py index 029444e..78a845b 100644 --- a/opengsq/protocols/eos.py +++ b/opengsq/protocols/eos.py @@ -21,9 +21,9 @@ def __init__( self, host: str, port: int, + deployment_id: str, + access_token: str, timeout: float = 5, - deployment_id: str = None, - access_token: str = None, ): """ Initializes the EOS object with the given parameters. @@ -234,9 +234,9 @@ async def main_async(): eos = EOS( host="5.62.115.46", port=7783, - timeout=5.0, deployment_id=deployment_id, access_token=access_token, + timeout=5.0, ) info = await eos.get_info() diff --git a/opengsq/protocols/fivem.py b/opengsq/protocols/fivem.py index cfdef0e..efbc15d 100644 --- a/opengsq/protocols/fivem.py +++ b/opengsq/protocols/fivem.py @@ -5,37 +5,61 @@ class FiveM(ProtocolBase): - """FiveM Protocol (https://docs.fivem.net/docs/server-manual/proxy-setup/)""" - full_name = 'FiveM Protocol' + """ + This class represents the FiveM Protocol (https://docs.fivem.net/docs/server-manual/proxy-setup/). It provides methods to interact with the FiveM API. + """ + + full_name = "FiveM Protocol" async def _get(self, filename: str) -> dict: - url = f'http://{self._host}:{self._port}/{filename}.json?v={int(time.time())}' + """ + Asynchronously retrieves the JSON data from the given filename on the server. + + :param filename: The filename to retrieve data from. + :return: A dictionary containing the JSON data. + """ + url = f"http://{self._host}:{self._port}/{filename}.json?v={int(time.time())}" async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.json(content_type=None) async def get_info(self) -> dict: - return await self._get('info') + """ + Asynchronously retrieves the information of the game server. + + :return: A dictionary containing the information of the game server. + """ + return await self._get("info") async def get_players(self) -> list: - return await self._get('players') + """ + Asynchronously retrieves the list of players on the game server. + + :return: A list of players on the game server. + """ + return await self._get("players") async def get_dynamic(self) -> dict: - return await self._get('dynamic') + """ + Asynchronously retrieves the dynamic data of the game server. + + :return: A dictionary containing the dynamic data of the game server. + """ + return await self._get("dynamic") -if __name__ == '__main__': +if __name__ == "__main__": import asyncio import json async def main_async(): - fivem = FiveM(host='144.217.10.12', port=30120, timeout=5.0) + fivem = FiveM(host="144.217.10.12", port=30120, timeout=5.0) info = await fivem.get_info() - print(json.dumps(info, indent=None) + '\n') + print(json.dumps(info, indent=None) + "\n") players = await fivem.get_players() - print(json.dumps(players, indent=None) + '\n') + print(json.dumps(players, indent=None) + "\n") dynamic = await fivem.get_dynamic() - print(json.dumps(dynamic, indent=None) + '\n') + print(json.dumps(dynamic, indent=None) + "\n") asyncio.run(main_async()) diff --git a/opengsq/protocols/gamespy3.py b/opengsq/protocols/gamespy3.py index 25690af..666aea4 100644 --- a/opengsq/protocols/gamespy3.py +++ b/opengsq/protocols/gamespy3.py @@ -7,40 +7,51 @@ class GameSpy3(ProtocolBase): - """GameSpy Protocol version 3""" - full_name = 'GameSpy Protocol version 3' - challenge = False + """ + This class represents the GameSpy Protocol version 3. It provides methods to interact with the GameSpy API. + """ - async def get_status(self): - """Retrieves information about the server including, Info, Players, and Teams.""" + full_name = "GameSpy Protocol version 3" + challenge_required = False + + async def get_status(self) -> dict: + """ + Asynchronously retrieves the status of the game server. The status includes information about the server, + players, and teams. + + :return: A dictionary containing the status of the game server. + """ # Connect to remote host with UdpClient() as udpClient: udpClient.settimeout(self._timeout) await udpClient.connect((self._host, self._port)) - request_h = b'\xFE\xFD' - timestamp = b'\x04\x05\x06\x07' - challenge = b'' + request_h = b"\xFE\xFD" + timestamp = b"\x04\x05\x06\x07" + challenge = b"" - if self.challenge: + if self.challenge_required: # Packet 1: Initial request - (https://wiki.unrealadmin.org/UT3_query_protocol#Packet_1:_Initial_request) - udpClient.send(request_h + b'\x09' + timestamp) + udpClient.send(request_h + b"\x09" + timestamp) # Packet 2: First response - (https://wiki.unrealadmin.org/UT3_query_protocol#Packet_2:_First_response) response = await udpClient.recv() if response[0] != 9: raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {}.' - .format(chr(response[0]), chr(9)) + "Packet header mismatch. Received: {}. Expected: {}.".format( + chr(response[0]), chr(9) + ) ) # Packet 3: Second request - (http://wiki.unrealadmin.org/UT3_query_protocol#Packet_3:_Second_request) - challenge = int(response[5:].decode('ascii').strip('\x00')) - challenge = b'' if challenge == 0 else challenge.to_bytes(4, 'big', signed=True) + challenge = int(response[5:].decode("ascii").strip("\x00")) + challenge = ( + b"" if challenge == 0 else challenge.to_bytes(4, "big", signed=True) + ) - request_data = request_h + b'\x00' + timestamp + challenge - udpClient.send(request_data + b'\xFF\xFF\xFF\x01') + request_data = request_h + b"\x00" + timestamp + challenge + udpClient.send(request_data + b"\xFF\xFF\xFF\x01") # Packet 4: Server information response # (http://wiki.unrealadmin.org/UT3_query_protocol#Packet_4:_Server_information_response) @@ -49,30 +60,30 @@ async def get_status(self): br = BinaryReader(response) result = {} - result['info'] = {} + result["info"] = {} while True: key = br.read_string() - if key == '': + if key == "": break - result['info'][key] = br.read_string() + result["info"][key] = br.read_string() - pattern = re.compile(rb'\x00([^a-zA-Z])([a-zA-Z_]+)\x00\x00(.+?(?=\x00\x00))') + pattern = re.compile(rb"\x00([^a-zA-Z])([a-zA-Z_]+)\x00\x00(.+?(?=\x00\x00))") current_id = -1 current_name = None - for (id, name, data) in re.findall(pattern, b'\x00' + br.read()): - values = data.split(b'\x00') - name = name.decode('utf-8').split('_')[0] + for id, name, data in re.findall(pattern, b"\x00" + br.read()): + values = data.split(b"\x00") + name = name.decode("utf-8").split("_")[0] - if current_id != id and id != b'\x00': + if current_id != id and id != b"\x00": current_id, current_name = id, name result[current_name] = [{} for _ in range(len(values))] for i in range(len(values)): - result[current_name][i][name] = values[i].decode('utf-8') + result[current_name][i][name] = values[i].decode("utf-8") return result @@ -88,8 +99,9 @@ async def __read(self, udpClient: UdpClient) -> bytes: if header != 0: raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {}.' - .format(chr(header), chr(0)) + "Packet header mismatch. Received: {}. Expected: {}.".format( + chr(header), chr(0) + ) ) # Skip the timestamp and splitnum @@ -112,7 +124,7 @@ async def __read(self, udpClient: UdpClient) -> bytes: # 2 = team_t \x00\x02team_t\x00\x00 since \x02 # etc... obj_id = br.read_byte() - header = b'' + header = b"" if obj_id >= 1: # The object key name @@ -122,24 +134,28 @@ async def __read(self, udpClient: UdpClient) -> bytes: count = br.read_byte() # Set back the packet header if it didn't appear before - header = b'\x00' + bytes([obj_id]) + string.encode() + b'\x00\x00' if count == 0 else b'' + header = ( + b"\x00" + bytes([obj_id]) + string.encode() + b"\x00\x00" + if count == 0 + else b"" + ) payload = header + br.read()[:-1] # Remove the last trash string on the payload - payloads[number] = payload[:payload.rfind(b'\x00') + 1] + payloads[number] = payload[: payload.rfind(b"\x00") + 1] - response = b''.join(payloads[number] for number in sorted(payloads)) + response = b"".join(payloads[number] for number in sorted(payloads)) return response -if __name__ == '__main__': +if __name__ == "__main__": import asyncio import json async def main_async(): - gs3 = GameSpy3(host='185.107.96.59', port=29900, timeout=5.0) + gs3 = GameSpy3(host="185.107.96.59", port=29900, timeout=5.0) server = await gs3.get_status() print(json.dumps(server, indent=None)) diff --git a/opengsq/protocols/gamespy4.py b/opengsq/protocols/gamespy4.py index 581d858..e53143a 100644 --- a/opengsq/protocols/gamespy4.py +++ b/opengsq/protocols/gamespy4.py @@ -2,17 +2,20 @@ class GameSpy4(GameSpy3): - """GameSpy Protocol version 4""" - full_name = 'GameSpy Protocol version 4' - challenge = True + """ + This class represents the GameSpy Protocol version 4. It inherits from the GameSpy3 class and overrides some of its properties. + """ + full_name = "GameSpy Protocol version 4" + challenge_required = True -if __name__ == '__main__': + +if __name__ == "__main__": import asyncio import json async def main_async(): - gs4 = GameSpy4(host='play.avengetech.me', port=19132, timeout=5.0) + gs4 = GameSpy4(host="play.avengetech.me", port=19132, timeout=5.0) server = await gs4.get_status() print(json.dumps(server, indent=None)) diff --git a/opengsq/protocols/quake3.py b/opengsq/protocols/quake3.py index 83162f6..f8656dd 100644 --- a/opengsq/protocols/quake3.py +++ b/opengsq/protocols/quake3.py @@ -2,75 +2,104 @@ from opengsq.binary_reader import BinaryReader from opengsq.protocols.quake2 import Quake2 +from opengsq.exceptions import InvalidPacketException class Quake3(Quake2): - """Quake3 Protocol""" - full_name = 'Quake3 Protocol' + """ + This class represents the Quake3 Protocol. It provides methods to interact with the Quake3 API. + """ + + full_name = "Quake3 Protocol" def __init__(self, host: str, port: int, timeout: float = 5.0): - """Quake3 Query Protocol""" + """ + Initializes the Quake3 object with the given parameters. + + :param host: The host of the server. + :param port: The port of the server. + :param timeout: The timeout for the server connection. + """ super().__init__(host, port, timeout) - self._request_header = b'getstatus' - self._response_header = 'statusResponse\n' + self._request_header = b"getstatus" + self._response_header = "statusResponse\n" async def get_info(self, strip_color=True) -> dict: - """This returns server information only.""" - response_data = await self._connect_and_send(b'getinfo opengsq') + """ + Asynchronously retrieves the information of the game server. + + :param strip_color: A boolean indicating whether to strip color codes from the information. + :return: A dictionary containing the information of the game server. + """ + response_data = await self._connect_and_send(b"getinfo opengsq") br = BinaryReader(response_data) header = br.read_string(self._delimiter1) - response_header = 'infoResponse\n' + response_header = "infoResponse\n" if header != response_header: - raise Exception(f'Packet header mismatch. Received: {header}. Expected: {response_header}.') + raise InvalidPacketException( + f"Packet header mismatch. Received: {header}. Expected: {response_header}." + ) info = self._parse_info(br) if not strip_color: return info - if 'hostname' in info: - info['hostname'] = Quake3.strip_colors(info['hostname']) + if "hostname" in info: + info["hostname"] = Quake3.strip_colors(info["hostname"]) return info async def get_status(self, strip_color=True) -> dict: - """This returns server information and players.""" + """ + Asynchronously retrieves the status of the game server. The status includes information about the server and players. + + :param strip_color: A boolean indicating whether to strip color codes from the status. + :return: A dictionary containing the status of the game server. + """ br = await self._get_response_binary_reader() status = { - 'info': self._parse_info(br), - 'players': self._parse_players(br), + "info": self._parse_info(br), + "players": self._parse_players(br), } if not strip_color: return status - if 'sv_hostname' in status['info']: - status['info']['sv_hostname'] = Quake3.strip_colors(status['info']['sv_hostname']) + if "sv_hostname" in status["info"]: + status["info"]["sv_hostname"] = Quake3.strip_colors( + status["info"]["sv_hostname"] + ) - for player in status['players']: - if 'name' in player: - player['name'] = Quake3.strip_colors(player['name']) + for player in status["players"]: + if "name" in player: + player["name"] = Quake3.strip_colors(player["name"]) return status @staticmethod - def strip_colors(text: str): - """Strip color codes""" - return re.compile('\\^(X.{6}|.)').sub('', text) + def strip_colors(text: str) -> str: + """ + Strips color codes from the given text. + + :param text: The text to strip color codes from. + :return: The text with color codes stripped. + """ + return re.compile("\\^(X.{6}|.)").sub("", text) -if __name__ == '__main__': +if __name__ == "__main__": import asyncio import json async def main_async(): - quake3 = Quake3(host='85.10.197.106', port=27960, timeout=5.0) + quake3 = Quake3(host="85.10.197.106", port=27960, timeout=5.0) info = await quake3.get_info() status = await quake3.get_status() - print(json.dumps(info, indent=None) + '\n') - print(json.dumps(status, indent=None) + '\n') + print(json.dumps(info, indent=None) + "\n") + print(json.dumps(status, indent=None) + "\n") asyncio.run(main_async()) diff --git a/opengsq/protocols/source.py b/opengsq/protocols/source.py index 7c8a90e..3735036 100644 --- a/opengsq/protocols/source.py +++ b/opengsq/protocols/source.py @@ -1,27 +1,26 @@ import bz2 -import random import zlib -from enum import Enum from opengsq.binary_reader import BinaryReader -from opengsq.exceptions import AuthenticationException, InvalidPacketException +from opengsq.exceptions import InvalidPacketException from opengsq.protocol_base import ProtocolBase -from opengsq.protocol_socket import TcpClient, UdpClient +from opengsq.protocol_socket import UdpClient class Source(ProtocolBase): """Source Engine Protocol""" - full_name = 'Source Engine Protocol' - _A2S_INFO = b'\x54Source Engine Query\0' - _A2S_PLAYER = b'\x55' - _A2S_RULES = b'\x56' + full_name = "Source Engine Protocol" - class __RequestHeader(): - A2A_PING = b'\x69' - A2S_SERVERQUERY_GETCHALLENGE = b'\x57' + _A2S_INFO = b"\x54Source Engine Query\0" + _A2S_PLAYER = b"\x55" + _A2S_RULES = b"\x56" - class __ResponseHeader(): + class __RequestHeader: + A2A_PING = b"\x69" + A2S_SERVERQUERY_GETCHALLENGE = b"\x57" + + class __ResponseHeader: S2C_CHALLENGE = 0x41 S2A_INFO_SRC = 0x49 S2A_INFO_DETAILED = 0x6D @@ -41,10 +40,16 @@ async def get_info(self) -> dict: br = BinaryReader(response_data) header = br.read_byte() - if header != self.__ResponseHeader.S2A_INFO_SRC and header != self.__ResponseHeader.S2A_INFO_DETAILED: + if ( + header != self.__ResponseHeader.S2A_INFO_SRC + and header != self.__ResponseHeader.S2A_INFO_DETAILED + ): raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {} or {}.' - .format(chr(header), chr(self.__ResponseHeader.S2A_INFO_SRC), chr(self.__ResponseHeader.S2A_INFO_DETAILED)) + "Packet header mismatch. Received: {}. Expected: {} or {}.".format( + chr(header), + chr(self.__ResponseHeader.S2A_INFO_SRC), + chr(self.__ResponseHeader.S2A_INFO_DETAILED), + ) ) if header == self.__ResponseHeader.S2A_INFO_SRC: @@ -55,75 +60,75 @@ async def get_info(self) -> dict: def __parse_from_info_src(self, br: BinaryReader) -> dict: info = {} - info['Protocol'] = br.read_byte() - info['Name'] = br.read_string() - info['Map'] = br.read_string() - info['Folder'] = br.read_string() - info['Game'] = br.read_string() - info['ID'] = br.read_short() - info['Players'] = br.read_byte() - info['MaxPlayers'] = br.read_byte() - info['Bots'] = br.read_byte() - info['ServerType'] = chr(br.read_byte()) - info['Environment'] = chr(br.read_byte()) - info['Visibility'] = br.read_byte() - info['VAC'] = br.read_byte() + info["Protocol"] = br.read_byte() + info["Name"] = br.read_string() + info["Map"] = br.read_string() + info["Folder"] = br.read_string() + info["Game"] = br.read_string() + info["ID"] = br.read_short() + info["Players"] = br.read_byte() + info["MaxPlayers"] = br.read_byte() + info["Bots"] = br.read_byte() + info["ServerType"] = chr(br.read_byte()) + info["Environment"] = chr(br.read_byte()) + info["Visibility"] = br.read_byte() + info["VAC"] = br.read_byte() # These fields only exist in a response if the server is running The Ship - if info['ID'] == 2400: - info['Mode'] = br.read_byte() - info['Witnesses'] = br.read_byte() - info['Duration'] = br.read_byte() + if info["ID"] == 2400: + info["Mode"] = br.read_byte() + info["Witnesses"] = br.read_byte() + info["Duration"] = br.read_byte() - info['Version'] = br.read_string() - info['EDF'] = br.read_byte() + info["Version"] = br.read_string() + info["EDF"] = br.read_byte() - if info['EDF'] & 0x80: - info['GamePort'] = br.read_short() + if info["EDF"] & 0x80: + info["GamePort"] = br.read_short() - if info['EDF'] & 0x10: - info['SteamID'] = br.read_long_long() + if info["EDF"] & 0x10: + info["SteamID"] = br.read_long_long() - if info['EDF'] & 0x40: - info['SpecPort'] = br.read_short() - info['SpecName'] = br.read_string() + if info["EDF"] & 0x40: + info["SpecPort"] = br.read_short() + info["SpecName"] = br.read_string() - if info['EDF'] & 0x20: - info['Keywords'] = br.read_string() + if info["EDF"] & 0x20: + info["Keywords"] = br.read_string() - if info['EDF'] & 0x01: - info['GameID'] = br.read_long_long() + if info["EDF"] & 0x01: + info["GameID"] = br.read_long_long() return info def __parse_from_info_detailed(self, br: BinaryReader) -> dict: info = {} - info['Address'] = br.read_string() - info['Name'] = br.read_string() - info['Map'] = br.read_string() - info['Folder'] = br.read_string() - info['Game'] = br.read_string() - info['Players'] = br.read_byte() - info['MaxPlayers'] = br.read_byte() - info['Protocol'] = br.read_byte() - info['ServerType'] = chr(br.read_byte()) - info['Environment'] = chr(br.read_byte()) - info['Visibility'] = br.read_byte() - info['Mod'] = br.read_byte() - - if info['Mod'] == 1: - info['Link'] = br.read_string() - info['DownloadLink'] = br.read_string() + info["Address"] = br.read_string() + info["Name"] = br.read_string() + info["Map"] = br.read_string() + info["Folder"] = br.read_string() + info["Game"] = br.read_string() + info["Players"] = br.read_byte() + info["MaxPlayers"] = br.read_byte() + info["Protocol"] = br.read_byte() + info["ServerType"] = chr(br.read_byte()) + info["Environment"] = chr(br.read_byte()) + info["Visibility"] = br.read_byte() + info["Mod"] = br.read_byte() + + if info["Mod"] == 1: + info["Link"] = br.read_string() + info["DownloadLink"] = br.read_string() br.read_byte() - info['Version'] = br.read_long() - info['Size'] = br.read_long() - info['Type'] = br.read_byte() - info['DLL'] = br.read_byte() + info["Version"] = br.read_long() + info["Size"] = br.read_long() + info["Type"] = br.read_byte() + info["DLL"] = br.read_byte() - info['VAC'] = br.read_byte() - info['Bots'] = br.read_byte() + info["VAC"] = br.read_byte() + info["Bots"] = br.read_byte() return info @@ -140,8 +145,9 @@ async def get_players(self) -> list: if header != self.__ResponseHeader.S2A_PLAYER: raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {}.' - .format(chr(header), chr(self.__ResponseHeader.S2A_PLAYER)) + "Packet header mismatch. Received: {}. Expected: {}.".format( + chr(header), chr(self.__ResponseHeader.S2A_PLAYER) + ) ) player_count = br.read_byte() @@ -151,15 +157,15 @@ async def get_players(self) -> list: br.read_byte() player = {} - player['Name'] = br.read_string() - player['Score'] = br.read_long() - player['Duration'] = br.read_float() + player["Name"] = br.read_string() + player["Score"] = br.read_long() + player["Duration"] = br.read_float() players.append(player) if br.remaining_bytes() > 0: for i in range(player_count): - players[i]['Deaths'] = br.read_long() - players[i]['Money'] = br.read_long() + players[i]["Deaths"] = br.read_long() + players[i]["Money"] = br.read_long() return players @@ -176,8 +182,9 @@ async def get_rules(self) -> dict: if header != self.__ResponseHeader.S2A_RULES: raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {}.' - .format(chr(header), chr(self.__ResponseHeader.S2A_RULES)) + "Packet header mismatch. Received: {}. Expected: {}.".format( + chr(header), chr(self.__ResponseHeader.S2A_RULES) + ) ) rule_count = br.read_short() @@ -192,11 +199,11 @@ async def __connect_and_send_challenge(self, header: __RequestHeader) -> bytes: await udpClient.connect((self._host, self._port)) # Send and receive - request_base = b'\xFF\xFF\xFF\xFF' + header + request_base = b"\xFF\xFF\xFF\xFF" + header request_data = request_base if header != self._A2S_INFO: - request_data += b'\xFF\xFF\xFF\xFF' + request_data += b"\xFF\xFF\xFF\xFF" udpClient.send(request_data) @@ -266,7 +273,7 @@ async def __receive(self, udpClient: UdpClient) -> bytes: break # Combine the payloads - combined_payload = b''.join(payloads[number] for number in sorted(payloads)) + combined_payload = b"".join(payloads[number] for number in sorted(payloads)) # Decompress the payload if is_compressed: @@ -274,16 +281,22 @@ async def __receive(self, udpClient: UdpClient) -> bytes: # Check CRC32 sum if zlib.crc32(combined_payload) != crc32_checksum: - raise InvalidPacketException('CRC32 checksum mismatch of uncompressed packet data.') + raise InvalidPacketException( + "CRC32 checksum mismatch of uncompressed packet data." + ) - return combined_payload.startswith(b'\xFF\xFF\xFF\xFF') and combined_payload[4:] or combined_payload + return ( + combined_payload.startswith(b"\xFF\xFF\xFF\xFF") + and combined_payload[4:] + or combined_payload + ) def __is_gold_source_split(self, br: BinaryReader): # Upper 4 bits represent the number of the current packet (starting at 0) number = br.read_byte() >> 4 # Check is it Gold Source packet split format - return number == 0 and br.read().startswith(b'\xFF\xFF\xFF\xFF') + return number == 0 and br.read().startswith(b"\xFF\xFF\xFF\xFF") async def __parse_gold_source_packet(self, udpClient: UdpClient, packets: list): total_packets = -1 @@ -316,155 +329,26 @@ async def __parse_gold_source_packet(self, udpClient: UdpClient, packets: list): payloads[number] = br.read() # Combine the payloads - combined_payload = b''.join(payloads[number] for number in sorted(payloads)) - - return combined_payload.startswith(b'\xFF\xFF\xFF\xFF') and combined_payload[4:] or combined_payload - - class RemoteConsole(ProtocolBase): - """Source RCON Protocol""" - - full_name = 'Source RCON Protocol' - - def __init__(self, host: str, port: int = 27015, timeout: float = 5.0): - """Source RCON Protocol""" - super().__init__(host, port, timeout) - - self._tcpClient = None - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_value, traceback): - self.close() - - def close(self): - """Close the connection""" - if self._tcpClient: - self._tcpClient.close() - - async def authenticate(self, password: str): - """Authenticate the connection""" - - # Connect - self._tcpClient = TcpClient() - self._tcpClient.settimeout(self._timeout) - await self._tcpClient.connect((self._host, self._port)) - - # Send password - id = random.randrange(4096) - self._tcpClient.send(self.__Packet(id, self.__PacketType.SERVERDATA_AUTH.value, password).get_bytes()) - - # Receive and parse as Packet - response_data = await self._tcpClient.recv() - packet = self.__Packet(response_data) - - # Sometimes it will return a PacketType.SERVERDATA_RESPONSE_VALUE, so receive again - if packet.type != self.__PacketType.SERVERDATA_AUTH_RESPONSE.value: - response_data = await self._tcpClient.recv() - packet = self.__Packet(response_data) - - # Throw exception if not PacketType.SERVERDATA_AUTH_RESPONSE - if packet.type != self.__PacketType.SERVERDATA_AUTH_RESPONSE.value: - self._tcpClient.close() - raise InvalidPacketException( - 'Packet header mismatch. Received: {}. Expected: {}.' - .format(chr(packet.type), chr(self.__PacketType.SERVERDATA_AUTH_RESPONSE.value)) - ) - - # Throw exception if authentication failed - if packet.id == -1 or packet.id != id: - self._tcpClient.close() - raise AuthenticationException('Authentication failed') - - async def send_command(self, command: str): - """Send command to the server""" - - # Send the command and a empty command packet - id = random.randrange(4096) - dummy_id = id + 1 - self._tcpClient.send(self.__Packet(id, self.__PacketType.SERVERDATA_EXECCOMMAND.value, command).get_bytes()) - self._tcpClient.send(self.__Packet(dummy_id, self.__PacketType.SERVERDATA_EXECCOMMAND.value, '').get_bytes()) - - packet_bytes = bytes([]) - response = '' - - while True: - # Receive - response_data = await self._tcpClient.recv() - - # Concat to last unused bytes - packet_bytes += response_data - - # Get the packets and get the unused bytes - packets, packet_bytes = self.__get_packets(packet_bytes) - - # Loop all packets - for packet in packets: - if packet.id == dummy_id: - return response - - response += packet.body - - # Handle Multiple-packet Responses - def __get_packets(self, packet_bytes: bytes): - packets = [] - - br = BinaryReader(packet_bytes) - - # + 4 to ensure br.ReadInt32() is readable - while br.stream_position + 4 < len(packet_bytes): - size = br.read_long() - - if br.stream_position + size > len(packet_bytes): - return packets, packet_bytes[br.stream_position - 4:] - - id = br.read_long() - type = br.read_long() - body = br.read_string() - br.read_byte() - - packets.append(self.__Packet(id, type, body)) - - return packets, bytes([]) - - class __PacketType(Enum): - SERVERDATA_AUTH = 3 - SERVERDATA_AUTH_RESPONSE = 2 - SERVERDATA_EXECCOMMAND = 2 - SERVERDATA_RESPONSE_VALUE = 0 - - class __Packet: - def __init__(self, *args): - if len(args) == 3: - self.id = args[0] - self.type = args[1] - self.body = args[2] - else: - # Single-packet Responses - br = BinaryReader(args[0]) - br.read_long() - self.id = br.read_long() - self.type = br.read_long() - self.body = br.read_string() + combined_payload = b"".join(payloads[number] for number in sorted(payloads)) - def get_bytes(self): - packet_bytes = self.id.to_bytes(4, byteorder='little') - packet_bytes += self.type.to_bytes(4, byteorder='little') - packet_bytes += str.encode(self.body + '\0') - return len(packet_bytes).to_bytes(4, byteorder='little') + packet_bytes + return ( + combined_payload.startswith(b"\xFF\xFF\xFF\xFF") + and combined_payload[4:] + or combined_payload + ) -if __name__ == '__main__': +if __name__ == "__main__": import asyncio import json async def main_async(): - source = Source(host='209.205.114.187', port=27015, timeout=5.0) + source = Source(host="209.205.114.187", port=27015, timeout=5.0) info = await source.get_info() - print(json.dumps(info, indent=None, ensure_ascii=False) + '\n') + print(json.dumps(info, indent=None, ensure_ascii=False) + "\n") players = await source.get_players() - print(json.dumps(players, indent=None, ensure_ascii=False) + '\n') + print(json.dumps(players, indent=None, ensure_ascii=False) + "\n") rules = await source.get_rules() - print(json.dumps(rules, indent=None, ensure_ascii=False) + '\n') + print(json.dumps(rules, indent=None, ensure_ascii=False) + "\n") asyncio.run(main_async()) diff --git a/opengsq/protocols/teamspeak3.py b/opengsq/protocols/teamspeak3.py index 1b82059..9e0aa96 100644 --- a/opengsq/protocols/teamspeak3.py +++ b/opengsq/protocols/teamspeak3.py @@ -3,26 +3,57 @@ class TeamSpeak3(ProtocolBase): - """TeamSpeak 3 Protocol""" + """ + This class represents the TeamSpeak 3 Protocol. It provides methods to interact with the TeamSpeak 3 API. + """ full_name = 'TeamSpeak 3 Protocol' def __init__(self, host: str, port: int, voice_port: int, timeout: float = 5): + """ + Initializes the TeamSpeak3 object with the given parameters. + + :param host: The host of the server. + :param port: The port of the server. + :param voice_port: The voice port of the server. + :param timeout: The timeout for the server connection. + """ super().__init__(host, port, timeout) self._voice_port = voice_port - async def get_info(self): + async def get_info(self) -> dict: + """ + Asynchronously retrieves the information of the game server. + + :return: A dictionary containing the information of the game server. + """ response = await self.__send_and_receive(b'serverinfo') return self.__parse_kvs(response) - async def get_clients(self): + async def get_clients(self) -> list[dict]: + """ + Asynchronously retrieves the list of clients on the game server. + + :return: A list of clients on the game server. + """ response = await self.__send_and_receive(b'clientlist') return self.__parse_rows(response) - async def get_channels(self): + async def get_channels(self) -> list[dict]: + """ + Asynchronously retrieves the list of channels on the game server. + + :return: A list of channels on the game server. + """ response = await self.__send_and_receive(b'channellist -topic') return self.__parse_rows(response) async def __send_and_receive(self, data: bytes): + """ + Asynchronously sends the given data to the game server and receives the response. + + :param data: The data to send to the game server. + :return: The response from the game server. + """ with TcpClient() as tcpClient: tcpClient.settimeout(self._timeout) await tcpClient.connect((self._host, self._port)) @@ -45,9 +76,21 @@ async def __send_and_receive(self, data: bytes): return response[:-21] def __parse_rows(self, response: bytes): + """ + Parses the rows from the given response. + + :param response: The response to parse rows from. + :return: A list of dictionaries containing the parsed rows. + """ return [self.__parse_kvs(row) for row in response.split(b'|')] def __parse_kvs(self, response: bytes): + """ + Parses the key-value pairs from the given response. + + :param response: The response to parse key-value pairs from. + :return: A dictionary containing the parsed key-value pairs. + """ kvs = {} for kv in response.split(b' '): diff --git a/opengsq/rcon_protocols/__init__.py b/opengsq/rcon_protocols/__init__.py new file mode 100644 index 0000000..d05a4e7 --- /dev/null +++ b/opengsq/rcon_protocols/__init__.py @@ -0,0 +1 @@ +from .source_rcon import SourceRcon diff --git a/opengsq/rcon_protocols/source_rcon.py b/opengsq/rcon_protocols/source_rcon.py new file mode 100644 index 0000000..d2b33bb --- /dev/null +++ b/opengsq/rcon_protocols/source_rcon.py @@ -0,0 +1,201 @@ +import random + +from enum import Enum + +from opengsq.binary_reader import BinaryReader +from opengsq.exceptions import AuthenticationException, InvalidPacketException +from opengsq.protocol_base import ProtocolBase +from opengsq.protocol_socket import TcpClient + + +class SourceRcon(ProtocolBase): + """ + This class represents the Source RCON Protocol. It provides methods to interact with the Source RCON API. + """ + + full_name = "Source RCON Protocol" + + def __init__(self, host: str, port: int = 27015, timeout: float = 5.0): + """ + Initializes the SourceRcon object with the given parameters. + + :param host: The host of the server. + :param port: The port of the server. + :param timeout: The timeout for the server connection. + """ + super().__init__(host, port, timeout) + + self._tcpClient = None + + def __enter__(self): + """ + Defines what the context manager should do at the beginning of the block created by the with statement. + Returns the object to be used in the context of the with statement. + """ + return self + + def __exit__(self, exc_type, exc_value, traceback): + """ + Defines what the context manager should do after its block has been executed (or terminates). + It can suppress exceptions, if needed. + """ + self.close() + + def close(self): + """ + Closes the connection to the server. + """ + if self._tcpClient: + self._tcpClient.close() + + async def authenticate(self, password: str): + """ + Asynchronously authenticates the connection to the server using the given password. + + :param password: The password to authenticate with. + """ + + # Connect + self._tcpClient = TcpClient() + self._tcpClient.settimeout(self._timeout) + await self._tcpClient.connect((self._host, self._port)) + + # Send password + id = random.randrange(4096) + self._tcpClient.send( + self.__Packet( + id, self.__PacketType.SERVERDATA_AUTH.value, password + ).get_bytes() + ) + + # Receive and parse as Packet + response_data = await self._tcpClient.recv() + packet = self.__Packet(response_data) + + # Sometimes it will return a PacketType.SERVERDATA_RESPONSE_VALUE, so receive again + if packet.type != self.__PacketType.SERVERDATA_AUTH_RESPONSE.value: + response_data = await self._tcpClient.recv() + packet = self.__Packet(response_data) + + # Throw exception if not PacketType.SERVERDATA_AUTH_RESPONSE + if packet.type != self.__PacketType.SERVERDATA_AUTH_RESPONSE.value: + self._tcpClient.close() + raise InvalidPacketException( + "Packet header mismatch. Received: {}. Expected: {}.".format( + chr(packet.type), + chr(self.__PacketType.SERVERDATA_AUTH_RESPONSE.value), + ) + ) + + # Throw exception if authentication failed + if packet.id == -1 or packet.id != id: + self._tcpClient.close() + raise AuthenticationException("Authentication failed") + + async def send_command(self, command: str) -> str: + """ + Asynchronously sends a command to the server. + + :param command: The command to send to the server. + :return: The response from the server. + """ + + # Send the command and a empty command packet + id = random.randrange(4096) + dummy_id = id + 1 + self._tcpClient.send( + self.__Packet( + id, self.__PacketType.SERVERDATA_EXECCOMMAND.value, command + ).get_bytes() + ) + self._tcpClient.send( + self.__Packet( + dummy_id, self.__PacketType.SERVERDATA_EXECCOMMAND.value, "" + ).get_bytes() + ) + + packet_bytes = bytes([]) + response = "" + + while True: + # Receive + response_data = await self._tcpClient.recv() + + # Concat to last unused bytes + packet_bytes += response_data + + # Get the packets and get the unused bytes + packets, packet_bytes = self.__get_packets(packet_bytes) + + # Loop all packets + for packet in packets: + if packet.id == dummy_id: + return response + + response += packet.body + + # Handle Multiple-packet Responses + def __get_packets(self, packet_bytes: bytes): + """ + Retrieves the packets from the given packet bytes. + + :param packet_bytes: The packet bytes to retrieve packets from. + :return: A list of packets and any remaining bytes. + """ + packets: list[SourceRcon.__Packet] = [] + + br = BinaryReader(packet_bytes) + + # + 4 to ensure br.ReadInt32() is readable + while br.stream_position + 4 < len(packet_bytes): + size = br.read_long() + + if br.stream_position + size > len(packet_bytes): + return packets, packet_bytes[br.stream_position - 4 :] + + id = br.read_long() + type = br.read_long() + body = br.read_string() + br.read_byte() + + packets.append(self.__Packet(id, type, body)) + + return packets, bytes([]) + + class __PacketType(Enum): + SERVERDATA_AUTH = 3 + SERVERDATA_AUTH_RESPONSE = 2 + SERVERDATA_EXECCOMMAND = 2 + SERVERDATA_RESPONSE_VALUE = 0 + + class __Packet: + def __init__(self, *args): + if len(args) == 3: + self.id = args[0] + self.type = args[1] + self.body = args[2] + else: + # Single-packet Responses + br = BinaryReader(args[0]) + br.read_long() + self.id = br.read_long() + self.type = br.read_long() + self.body = br.read_string() + + def get_bytes(self): + packet_bytes = self.id.to_bytes(4, byteorder="little") + packet_bytes += self.type.to_bytes(4, byteorder="little") + packet_bytes += str.encode(self.body + "\0") + return len(packet_bytes).to_bytes(4, byteorder="little") + packet_bytes + + +if __name__ == "__main__": + import asyncio + + async def main_async(): + with SourceRcon(host="", port=27015, timeout=5.0) as rcon: + await rcon.authenticate("") + response = await rcon.send_command("cvarlist") + print(response) + + asyncio.run(main_async()) diff --git a/tests/protocols/test_eos.py b/tests/protocols/test_eos.py index 562736f..75748c6 100644 --- a/tests/protocols/test_eos.py +++ b/tests/protocols/test_eos.py @@ -31,9 +31,9 @@ async def test_get_info(): eos = EOS( host="5.62.115.46", port=7783, - timeout=5.0, deployment_id=deployment_id, access_token=access_token, + timeout=5.0, ) result = await eos.get_info()