Authenticate localhost admin IPC

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,9 +202,19 @@ 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 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)
@@ -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):