add build step

2026-09-05 12:49:01 -05:00
parent 0c7783309c
commit f366201a87
4 changed files with 10 additions and 2 deletions
@@ -0,0 +1,68 @@
+from socket import socket
+
+import rcon
+from rcon import login, send_command
+from utils import parse_args, safe_input, load_script
+
+# terminal color codes
+# the readline module requires \001 and \002 because it counts characters
+# when displaying, but those color codes are zero-width
+RST = '\001\x1b[0m\002'
+GRN = '\001\x1b[0;32m\002'
+BLU = '\001\x1b[0;34m\002'
+GRY = '\001\x1b[0;37m\002'
+
+def main():
+    args = parse_args()
+    RCON_ADDRESS = None
+    if args.host:
+        RCON_ADDRESS = args.host
+        args.port = args.port or 25575
+    else:
+        RCON_ADDRESS = safe_input(f"{GRN} RCON hostname (IP address)> {RST}")
+
+    RCON_PORT = args.port or int(safe_input(f"{GRN} RCON port (default 25575)> {RST}") or 25575)
+    RCON_PASSWORD = args.password or safe_input(f"{GRN} RCON server password> {RST}") 
+
+    # open TCP connection to server
+    with socket() as sock:
+        try:
+            sock.connect((RCON_ADDRESS, RCON_PORT))
+        except:
+            print(f"{GRY} Connection to server failed. {RST}")
+            print(f"{GRY} Please check the server hostname and port, then try again. {RST}")
+            exit(1)
+
+        sock.settimeout(args.socket_timeout or 30)
+
+        # login attempts
+        login_success = login(sock, RCON_PASSWORD)
+        while (not login_success):
+            print(f"{GRY} Authentication failed. {RST}")
+            # if not run interactively, quit b/c we don't want to prompt the user for more passwords
+            if args.password: exit(1) 
+            login_success = login(sock, safe_input(f"{GRN} Please re-enter the RCON password> {RST}"))
+
+        if args.execute_command:
+            if args.echo: rcon.ECHO_COMMANDS = True
+            for command in args.execute_command:
+                print(send_command(sock, command))
+            exit()
+
+        elif args.execute_file:
+            if args.echo: rcon.ECHO_COMMANDS = True
+            script = load_script(args.execute_file)
+            # load_script already verifies the user script has a `run` method
+            script.run(sock)
+            exit()
+
+        else: # command REPL
+            print(f"{GRY} Successfully logged in. Press Ctrl-C for a fresh prompt or Ctrl-D to exit. {RST}")
+            while True: # safe_input will terminate the program when the user pressed Ctrl-D
+                command = safe_input(f"{BLU} RCON@{RCON_ADDRESS} > {RST}")
+                print(f"{GRY}", send_command(sock, command))
+
+
+if __name__ == '__main__':
+    main()
+
@@ -0,0 +1,94 @@
+from dataclasses import dataclass
+from random import randint
+from socket import socket
+import struct
+
+MAX_OUTGOING_PAYLOAD_LENGTH = 1446
+MAX_INCOMING_PAYLOAD_LENGTH = 4096
+INVALID_PACKET_REQUEST_ID = 32767
+
+# can be overridden to make send_command echo all commands before sending them; useful for troubleshooting
+ECHO_COMMANDS = False 
+
+# packet types
+PKT_LOGIN = 3
+PKT_LOGIN_RESPONSE = 2
+PKT_COMMAND = 2
+PKT_COMMAND_RESPONSE = 0
+
+@dataclass
+class Packet:
+    """A single RCON packet, either sent or received."""
+    packet_size: int
+    request_id: int
+    packet_type: int
+    payload: bytes
+
+    @staticmethod
+    def recv(sock: socket) -> Packet:
+        """Read and parse the next full packet from sock, blocking until complete."""
+        # first four bytes (little-endian) contain the size of the remainder of the packet
+        [packet_size] = struct.unpack('<i', sock.recv(4))
+        # then, 4 bytes each for ID & type, arbitrary-length string payload, and two null bytes
+        request_id, packet_type, payload, null = struct.unpack(f'<ii{packet_size-4-4-2}sh',
+                                                               Packet._recvall(sock, packet_size))
+        
+        return Packet(packet_size, request_id, packet_type, payload)
+
+    @staticmethod
+    def _recvall(sock: socket, n: int) -> bytes:
+        """Block until exactly n bytes have been read from sock."""
+        buf = b''
+        while len(buf) < n:
+            chunk = sock.recv(n - len(buf))
+            if not chunk:
+                raise ConnectionError("socket closed before full packet received")
+            buf += chunk
+        return buf
+
+    @staticmethod
+    def send(sock: socket, packet_type: int, payload: str, request_id: int | None = None):
+        """Build a packet from payload and transmit it on sock.
+
+        Raises ValueError if the encoded payload exceeds MAX_OUTGOING_PAYLOAD_LENGTH.
+        """
+        payload_bytes = bytes(payload, encoding='utf-8')
+
+        if len(payload_bytes) > MAX_OUTGOING_PAYLOAD_LENGTH:
+            raise ValueError(f"attempted to send payload of length {len(payload_bytes)} bytes, which exceeds the maximum of {MAX_OUTGOING_PAYLOAD_LENGTH} bytes")
+
+        # 4 bytes each for ID & type, arbitrary-length string payload, and two null bytes
+        packet_size = 4 + 4 + len(payload_bytes) + 2
+
+        if request_id == None:
+            request_id = randint(1, 32766)
+
+        data = struct.pack(f'<iii{len(payload_bytes)}sh', packet_size, request_id, packet_type, payload_bytes, 0)
+        sock.sendall(data)
+
+
+def login(sock: socket, password: str) -> bool:
+    """Authenticate on an already-connected socket. Returns a bool representing whether login succeeded."""
+    Packet.send(sock, PKT_LOGIN, password)
+    response = Packet.recv(sock)
+
+    if response.request_id == -1:
+        return False # auth failure
+    return True
+
+def send_command(sock: socket, command: str) -> str:
+    """Run command on an already-authenticated socket and return the decoded response."""
+    # Some commands can have responses greater than the max response length of 4096, so the server will send the response fragmented across several packets. However, the Notchian server doesn't properly implement that fragmentation logic, meaning there can be ambiguity about when the actual last packet is. 
+    # To handle this, we send two packets with distinct IDs. The first packet has the command we want to run; the second has an invalid packet type. Since the server responds to packets in the order they were received, we read from the socket until we get a response that matches the second packet; everything before was part of the first packet's response. 
+    if ECHO_COMMANDS: print(command)
+    Packet.send(sock, PKT_COMMAND, command)
+    Packet.send(sock, 200, "", request_id=INVALID_PACKET_REQUEST_ID)
+
+    response = b''
+    while True:
+        received = Packet.recv(sock)
+        if received.request_id == INVALID_PACKET_REQUEST_ID: break # we got a response from our second packet
+        response += received.payload
+
+    return response.decode('utf-8', errors='replace')
+
@@ -0,0 +1,58 @@
+import argparse
+import readline
+from importlib.util import spec_from_file_location, module_from_spec
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+            prog="Minecraft RCON Client",
+            description="A simple RCON client for remotely controlling a Minecraft server. Can be used interactively or to run prespecified commands. Also supports running arbitrary Python scripts for more complex logic.")
+
+    parser.add_argument("-H", "--host", help="The hostname of the RCON server.")
+    parser.add_argument("-P", "--port", help="The port number of the RCON server. Default is 25575 if omitted and -H is specified. ", type=int)
+    parser.add_argument("-p", "--password", help="The password to the RCON server.")
+
+    execute_arg_group = parser.add_mutually_exclusive_group()
+    execute_arg_group.add_argument("-e", "--execute-command", help="Send a single command on the RCON server. Can be passed multiple times to run multiple commands. Does not abort on server-side command failures. Requires -H and -p to be specified. Mutually exclusive with -E.", metavar="COMMAND", action='append')
+    execute_arg_group.add_argument("-E", "--execute-file", help="Execute a Python file. Requires -H and -p to be specified. Mutually exclusive with -e.", metavar="FILEPATH")
+
+    parser.add_argument("--echo", help="Echo a command before sending it. Does nothing when neither -e nor -E are not present.", action="store_true")
+    parser.add_argument("--socket-timeout", help="Set the timeout for the socket when waiting to receive data. Default 30 seconds.", type=float, metavar="TIMEOUT_SECONDS")
+
+    args = parser.parse_args()
+
+    if (args.execute_command or args.execute_file) and not (args.host and args.password):
+        parser.error("-e/-E require -H and -p to be specified")
+
+    return args
+
+
+def safe_input(prompt: str) -> str:
+    """Like `input()`, but Ctrl-C prints a fresh prompt and Ctrl-D cleanly exits"""
+    while True:
+        try:
+            return input(prompt)
+        except KeyboardInterrupt:
+            print('^C')
+        except EOFError:
+            print()
+            exit(0)
+
+
+def load_script(script_path: str):
+    # three stages to importing here: find the thing to import, initialize an empty module, and execute the top-level code in that module
+    # here, we're essentially deciding which tool to use to read the script
+    spec = spec_from_file_location("user_script", script_path)
+    if spec is None:
+        raise ValueError(f"{script_path} is not a valid Python file path")
+
+    # initialize an empty "box" that we'll populate with all of our code
+    module = module_from_spec(spec)
+    # execute the top-level code (function definitions, global variables, etc.)
+    spec.loader.exec_module(module) # pyright: ignore
+
+    if not hasattr(module, "run"):
+        raise ValueError(f"Provided script {script_path} must define a run(sock) function")
+
+    return module
+