279 lines
10 KiB
Python
279 lines
10 KiB
Python
"""
|
|
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.
|
|
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:
|
|
"""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
|
|
self._client_sockets: set[socket.socket] = set()
|
|
self._running = False
|
|
|
|
def start(self) -> None:
|
|
with self._lock:
|
|
if self._running:
|
|
return
|
|
|
|
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
try:
|
|
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
|
|
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
|
|
else:
|
|
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server_socket.bind((ADMIN_HOST, self.port))
|
|
server_socket.listen()
|
|
server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
|
|
except OSError as error:
|
|
server_socket.close()
|
|
raise OSError(
|
|
f"Could not bind the admin server to {ADMIN_HOST}:{self.port}. "
|
|
f"The port may already be in use or unavailable. ({error})"
|
|
) from error
|
|
|
|
self._server_socket = server_socket
|
|
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, authenticated)")
|
|
|
|
def stop(self) -> None:
|
|
with self._lock:
|
|
self._running = False
|
|
server_socket = self._server_socket
|
|
self._server_socket = None
|
|
clients = list(self._client_sockets)
|
|
self._client_sockets.clear()
|
|
thread = self._thread
|
|
self._thread = None
|
|
|
|
if server_socket is not None:
|
|
try:
|
|
server_socket.close()
|
|
except OSError:
|
|
pass
|
|
|
|
for client in clients:
|
|
try:
|
|
client.close()
|
|
except OSError:
|
|
pass
|
|
|
|
if thread is not None and thread is not threading.current_thread():
|
|
thread.join(timeout=1.0)
|
|
|
|
def is_running(self) -> bool:
|
|
with self._lock:
|
|
return self._running
|
|
|
|
def _accept_loop(self) -> None:
|
|
try:
|
|
while True:
|
|
with self._lock:
|
|
if not self._running:
|
|
break
|
|
server_socket = self._server_socket
|
|
|
|
if server_socket is None:
|
|
break
|
|
|
|
try:
|
|
connection, address = server_socket.accept()
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
break
|
|
|
|
with self._lock:
|
|
self._client_sockets.add(connection)
|
|
|
|
thread = threading.Thread(
|
|
target=self._handle_connection,
|
|
args=(connection, address),
|
|
daemon=True,
|
|
)
|
|
thread.start()
|
|
finally:
|
|
with self._lock:
|
|
self._running = False
|
|
|
|
def _handle_connection(self, connection: socket.socket, address: tuple[str, int]) -> None:
|
|
peer = f"{address[0]}:{address[1]}"
|
|
try:
|
|
with connection:
|
|
buffer = b""
|
|
while True:
|
|
chunk = connection.recv(4096)
|
|
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)
|
|
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:
|
|
self._log(f"Admin connection error from {peer}: {error}")
|
|
finally:
|
|
with self._lock:
|
|
self._client_sockets.discard(connection)
|
|
|
|
def _dispatch_line(self, line: str) -> dict[str, Any]:
|
|
try:
|
|
request = json.loads(line)
|
|
except json.JSONDecodeError as error:
|
|
return {"ok": False, "error": f"Invalid JSON: {error}"}
|
|
|
|
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:
|
|
return {"ok": False, "error": str(error)}
|
|
|
|
if not isinstance(response, dict):
|
|
return {"ok": False, "error": "Admin handler returned a non-object response."}
|
|
return response
|
|
|
|
|
|
def send_admin_command(
|
|
request: dict[str, Any],
|
|
*,
|
|
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 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""
|
|
while b"\n" not in buffer:
|
|
chunk = connection.recv(4096)
|
|
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):
|
|
raise ValueError("Admin response must be a JSON object.")
|
|
return response
|