From b504c94b907c8f588373c009e531583e1a9c0613 Mon Sep 17 00:00:00 2001 From: Nomads_Reach <144523850+NomadsReach@users.noreply.github.com> Date: Sat, 15 Aug 2026 21:27:43 -0400 Subject: [PATCH] Authenticate localhost admin IPC --- server/admin_server.py | 109 ++++++++++++++++++++++++++++++++++++----- 1 file changed, 98 insertions(+), 11 deletions(-) diff --git a/server/admin_server.py b/server/admin_server.py index 134df95..21b9a26 100644 --- a/server/admin_server.py +++ b/server/admin_server.py @@ -1,36 +1,104 @@ """ -Localhost-only admin control channel for a running Commonwealth Online server. +Authenticated localhost admin control channel for a running Commonwealth Online server. Binds to 127.0.0.1 and speaks newline-delimited JSON request/response messages. -Not a public game API. +The authentication token is generated locally and is never exposed by the game protocol. """ from __future__ import annotations +import hmac import json +import os +import secrets import socket import threading from collections.abc import Callable +from pathlib import Path from typing import Any DEFAULT_ADMIN_PORT = 7779 ADMIN_HOST = "127.0.0.1" ACCEPT_TIMEOUT_SECONDS = 0.5 +MAX_ADMIN_LINE_BYTES = 64 * 1024 +DEFAULT_ADMIN_TOKEN_PATH = Path(__file__).resolve().parent / ".admin-token" + + +def load_or_create_admin_token(path: str | Path | None = None) -> str: + """Load the local admin token, creating it with restrictive permissions when absent.""" + token_path = Path(path) if path is not None else DEFAULT_ADMIN_TOKEN_PATH + token_path.parent.mkdir(parents=True, exist_ok=True) + + try: + existing = token_path.read_text(encoding="utf-8").strip() + except FileNotFoundError: + existing = "" + except OSError as error: + raise RuntimeError(f"Could not read admin token file {token_path}: {error}") from error + + if existing: + return existing + + token = secrets.token_urlsafe(32) + temp_path = token_path.with_name(f".{token_path.name}.{os.getpid()}.tmp") + try: + with open(temp_path, "x", encoding="utf-8", newline="\n") as handle: + handle.write(token) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + try: + os.chmod(temp_path, 0o600) + except OSError: + pass + os.replace(temp_path, token_path) + try: + os.chmod(token_path, 0o600) + except OSError: + pass + except FileExistsError: + try: + temp_path.unlink() + except OSError: + pass + return load_or_create_admin_token(token_path) + except OSError as error: + try: + temp_path.unlink() + except OSError: + pass + raise RuntimeError(f"Could not create admin token file {token_path}: {error}") from error + + return token + + +def load_admin_token(path: str | Path | None = None) -> str: + token_path = Path(path) if path is not None else DEFAULT_ADMIN_TOKEN_PATH + try: + token = token_path.read_text(encoding="utf-8").strip() + except OSError as error: + raise RuntimeError(f"Could not read admin token file {token_path}: {error}") from error + if not token: + raise RuntimeError(f"Admin token file is empty: {token_path}") + return token class AdminServer: - """JSON-lines admin TCP server bound to loopback only.""" + """Authenticated JSON-lines admin TCP server bound to loopback only.""" def __init__( self, handler: Callable[[dict[str, Any]], dict[str, Any]], port: int = DEFAULT_ADMIN_PORT, log: Callable[[str], None] | None = None, + token_path: str | Path | None = None, ) -> None: self._handler = handler self.port = port self._log = log or (lambda _message: None) + self._token_path = Path(token_path) if token_path is not None else DEFAULT_ADMIN_TOKEN_PATH + self._admin_token = load_or_create_admin_token(self._token_path) self._lock = threading.RLock() self._server_socket: socket.socket | None = None self._thread: threading.Thread | None = None @@ -62,7 +130,7 @@ class AdminServer: self._running = True self._thread = threading.Thread(target=self._accept_loop, daemon=True) self._thread.start() - self._log(f"Admin control listening on {ADMIN_HOST}:{self.port} (localhost only)") + self._log(f"Admin control listening on {ADMIN_HOST}:{self.port} (localhost, authenticated)") def stop(self) -> None: with self._lock: @@ -134,12 +202,22 @@ class AdminServer: if not chunk: break buffer += chunk + if len(buffer) > MAX_ADMIN_LINE_BYTES and b"\n" not in buffer: + self._log(f"Rejected oversized admin request from {peer}") + return while b"\n" in buffer: line_bytes, buffer = buffer.split(b"\n", 1) - line = line_bytes.decode("utf-8", errors="replace").strip() - if not line: - continue - response = self._dispatch_line(line) + if len(line_bytes) > MAX_ADMIN_LINE_BYTES: + response = {"ok": False, "error": "Admin request is too large."} + else: + try: + line = line_bytes.decode("utf-8", errors="strict").strip() + except UnicodeDecodeError: + response = {"ok": False, "error": "Admin request must be valid UTF-8."} + else: + if not line: + continue + response = self._dispatch_line(line) encoded = json.dumps(response, separators=(",", ":")).encode("utf-8") + b"\n" connection.sendall(encoded) except OSError as error: @@ -157,9 +235,13 @@ class AdminServer: if not isinstance(request, dict): return {"ok": False, "error": "Admin request must be a JSON object."} + supplied_token = request.pop("adminToken", None) + if not isinstance(supplied_token, str) or not hmac.compare_digest(supplied_token, self._admin_token): + return {"ok": False, "error": "Unauthorized admin request."} + try: response = self._handler(request) - except Exception as error: # noqa: BLE001 - admin channel must not kill the server + except Exception as error: return {"ok": False, "error": str(error)} if not isinstance(response, dict): @@ -173,9 +255,12 @@ def send_admin_command( host: str = ADMIN_HOST, port: int = DEFAULT_ADMIN_PORT, timeout_seconds: float = 3.0, + token_path: str | Path | None = None, ) -> dict[str, Any]: - """Send one admin command to a running server and return the JSON response.""" - encoded = json.dumps(request, separators=(",", ":")).encode("utf-8") + b"\n" + """Send one authenticated admin command to a running server and return the JSON response.""" + authenticated_request = dict(request) + authenticated_request["adminToken"] = load_admin_token(token_path) + encoded = json.dumps(authenticated_request, separators=(",", ":")).encode("utf-8") + b"\n" with socket.create_connection((host, port), timeout=timeout_seconds) as connection: connection.sendall(encoded) buffer = b"" @@ -184,6 +269,8 @@ def send_admin_command( if not chunk: raise ConnectionError("Admin server closed the connection without a response.") buffer += chunk + if len(buffer) > MAX_ADMIN_LINE_BYTES: + raise ValueError("Admin response exceeded the maximum size.") line_bytes, _rest = buffer.split(b"\n", 1) response = json.loads(line_bytes.decode("utf-8")) if not isinstance(response, dict):