onyx_online / Q3 Rcon CLI - single file, zero dependencies

#!/usr/bin/env python3 # MIT License # # Copyright (c) 2025 Onyx and Iris # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import argparse import re import socket class Packet: MAGIC = bytearray([0xFF, 0xFF, 0xFF, 0xFF]) class Request(Packet): """ Class used to encode a request packet for remote console (RCON) communication. Attributes: password (str): The password used for authentication with the RCON server. Methods: _header() -> bytes: Generates the header for the RCON packet. encode(cmd: str) -> bytes: Encodes the command into a byte string suitable for sending to the RCON server. Args: cmd (str): The command to be sent to the RCON server. Returns: bytes: The encoded command as a byte string. """ def __init__(self, password: str): self.password = password def _header(self) -> bytes: return Packet.MAGIC + b"rcon" def encode(self, cmd: str) -> bytes: return self._header() + f" {self.password} {cmd}".encode() class ResponseBuilder(Packet): """ Class used to build and decode a response packet from multiple fragments. Attributes: _fragments (bytearray): A bytearray to store the concatenated fragments. Methods: _header() -> bytes: Returns the header bytes that each fragment should start with. is_valid_fragment(fragment: bytes) -> bool: Checks if the given fragment starts with the expected header. add_fragment(fragment: bytes): Adds a fragment to the internal storage after removing the header. build() -> str: Decodes and returns the concatenated fragments as a string. """ def __init__(self): self._fragments = bytearray() def _header(self) -> bytes: return Packet.MAGIC + b"print\n" def is_valid_fragment(self, fragment: bytes) -> bool: return fragment.startswith(self._header()) def add_fragment(self, fragment: bytes): self._fragments.extend(fragment.removeprefix(self._header())) def build(self) -> str: return self._fragments.decode() def send(sock: socket.socket, request: Request, host: str, port: int, cmd: str) -> str: """ Send a single RCON (Remote Console) command to the server and return the response. This function sends a command to a game server using the RCON protocol over a UDP socket. It waits for the server's response, collects it in fragments if necessary, and returns the complete response. Args: sock (socket.socket): The UDP socket object used to send and receive data. request (Request): An instance of the Request class used to encode the command. host (str): The hostname or IP address of the server. port (int): The port number on which the server is listening. cmd (str): The RCON command to be sent to the server. Returns: str: The complete response from the server after processing all received fragments. Raises: socket.timeout: If the socket times out while waiting for a response from the server. """ sock.sendto(request.encode(cmd), (host, port)) response_builder = ResponseBuilder() while True: try: data, _ = sock.recvfrom(4096) except socket.timeout: break if response_builder.is_valid_fragment(data): response_builder.add_fragment(data) return response_builder.build() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--host", default="localhost") parser.add_argument("--port", type=int, default=27960) parser.add_argument("--password", default="secret") parser.add_argument("--timeout", type=float, default=0.2) parser.add_argument("--interactive", action="store_true") parser.add_argument("cmd", nargs="?", default="status") args = parser.parse_args() return args def main(): """ Main function to handle command-line arguments and interact with a remote server using RCON protocol. This function parses command-line arguments, creates a UDP socket, and sends requests to a remote server. It supports both interactive and non-interactive modes. In non-interactive mode, it sends a single command to the server and prints the response. In interactive mode, it continuously prompts the user for commands until the user quits by entering 'Q'. Args: None Returns: None """ def cleaned_response(resp: str) -> str: return re.sub(r"\^[0-9]", "", resp) args = parse_args() with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.settimeout(args.timeout) request = Request(args.password) if not args.interactive: if resp := send(sock, request, args.host, args.port, args.cmd): print(cleaned_response(resp)) return while cmd := input("cmd: ").strip(): if cmd == "Q": break if resp := send(sock, request, args.host, args.port, cmd): print(cleaned_response(resp)) if __name__ == "__main__": main()