Commonwealth-Online-Server/server/admin_server.py

"""
Localhost-only 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.
"""

from __future__ import annotations

import json
import socket
import threading
from collections.abc import Callable
from typing import Any


DEFAULT_ADMIN_PORT = 7779
ADMIN_HOST = "127.0.0.1"
ACCEPT_TIMEOUT_SECONDS = 0.5


class AdminServer:
    """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,
    ) -> None:
        self._handler = handler
        self.port = port
        self._log = log or (lambda _message: None)
        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 only)")

    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
                    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)
                        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."}

        try:
            response = self._handler(request)
        except Exception as error:  # noqa: BLE001 - admin channel must not kill the server
            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,
) -> 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"
    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
        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