server/tests/test_runtime_portability.py

@@ -8,7 +8,7 @@ from pathlib import Path
 
 import pytest
 
-from admin_server import send_admin_command
+from admin_server import DEFAULT_ADMIN_TOKEN_PATH, send_admin_command
 from lan_discovery import LanDiscoveryResponder
 from server_core import FalloutTogetherServer, get_lan_addresses
 from server_service import ServerConfig, ServerService
@@ -125,6 +125,41 @@ def test_headless_admin_roundtrip(tmp_path: Path) -> None:
     thread.join(timeout=3.0)
 
 
+def test_admin_rejects_unauthenticated_requests_and_does_not_disclose_token(tmp_path: Path) -> None:
+    game_port = _free_port()
+    admin_port = _free_port()
+    service = ServerService(
+        ServerConfig(
+            host="127.0.0.1",
+            port=game_port,
+            admin_port=admin_port,
+            bans_path=str(tmp_path / "bans.json"),
+        )
+    )
+    thread = threading.Thread(target=service.serve_forever, daemon=True)
+    thread.start()
+    deadline = time.time() + 5.0
+    while time.time() < deadline and not service.is_running():
+        time.sleep(0.05)
+    assert service.is_running()
+
+    with socket.create_connection(("127.0.0.1", admin_port), timeout=2.0) as conn:
+        conn.sendall(b'{"cmd":"ping"}\n')
+        raw = conn.recv(4096)
+    unauthorized = json.loads(raw.split(b"\n", 1)[0].decode("utf-8"))
+    assert unauthorized.get("ok") is False
+    assert unauthorized.get("error") == "Unauthorized admin request."
+
+    authenticated = send_admin_command({"cmd": "status"}, port=admin_port)
+    assert authenticated.get("ok") is True
+    token = DEFAULT_ADMIN_TOKEN_PATH.read_text(encoding="utf-8").strip()
+    assert token
+    assert token not in json.dumps(authenticated, sort_keys=True)
+
+    service.stop()
+    thread.join(timeout=3.0)
+
+
 def test_lan_discovery_sets_broadcast_option() -> None:
     class DummyServer:
         def get_stats(self):
@@ -139,7 +174,6 @@ def test_lan_discovery_sets_broadcast_option() -> None:
         def _log(self, message: str, *, level: str = "info") -> None:
             return None
 
-    # Bind an ephemeral discovery port to avoid colliding with a real server.
     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     sock.bind(("127.0.0.1", 0))
     port = int(sock.getsockname()[1])
@@ -149,7 +183,6 @@ def test_lan_discovery_sets_broadcast_option() -> None:
     responder.start()
     try:
         assert responder._socket is not None
-        # SO_BROADCAST should be enabled; querying may return 0/1 depending on OS.
         value = responder._socket.getsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST)
         assert value in (0, 1)
         probe = {