Harden Linux server runtime and packaging

Adds end-to-end Linux compatibility work for the dedicated server: new CI workflow (Ubuntu + Arch), line-ending/executable safeguards, and a local venv-first startup flow with split dependency files for server vs optional host GUI tooling. Improves runtime resilience with better bind error messages, stricter config validation, writable-state checks, cleaner socket/thread shutdown behavior, and SIGTERM-aware graceful stop handling for headless/systemd use. Updates deployment/startup docs and adds portability/runtime integration tests to lock in these behaviors.

server/tests/conftest.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+SERVER_DIR = Path(__file__).resolve().parents[1]
+if str(SERVER_DIR) not in sys.path:
+    sys.path.insert(0, str(SERVER_DIR))

server/tests/test_config_portability.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from config import (
+    Config,
+    ensure_writable_directory,
+    load_config,
+    save_config,
+    validate_config,
+)
+
+
+def test_save_and_load_utf8_lf(tmp_path: Path) -> None:
+    path = tmp_path / "commonwealth-server.json"
+    cfg = Config(
+        host="0.0.0.0",
+        port=7777,
+        server_name="Café Server",
+        server_description="résumé",
+        max_players=8,
+        log_verbosity="info",
+        admin_port=7779,
+    )
+    save_config(cfg, str(path))
+
+    raw = path.read_bytes()
+    assert b"\r\n" not in raw
+    assert "Café Server".encode("utf-8") in raw
+
+    loaded = load_config(str(path))
+    assert loaded.server_name == "Café Server"
+    assert loaded.server_description == "résumé"
+    assert loaded.admin_port == 7779
+
+
+def test_validate_rejects_bad_ports_and_max_players() -> None:
+    cfg = Config(port=70000, admin_port=7777, max_players=0)
+    ok, errors = validate_config(cfg)
+    assert not ok
+    assert any("port must be 1-65535" in error for error in errors)
+    assert any("max_players must be >= 1" in error for error in errors)
+
+    collide = Config(port=7777, admin_port=7777)
+    ok, errors = validate_config(collide)
+    assert not ok
+    assert any("admin_port must differ" in error for error in errors)
+
+
+def test_validate_rejects_unresolvable_host() -> None:
+    cfg = Config(host="this-host-should-not-resolve.invalid")
+    ok, errors = validate_config(cfg)
+    assert not ok
+    assert any("not a valid IPv4" in error for error in errors)
+
+
+def test_ensure_writable_directory(tmp_path: Path) -> None:
+    target = tmp_path / "state"
+    ensure_writable_directory(target)
+    assert target.is_dir()
+    assert not (target / ".commonwealth-write-probe").exists()
+
+
+def test_load_invalid_json(tmp_path: Path) -> None:
+    path = tmp_path / "bad.json"
+    path.write_text("{not json", encoding="utf-8")
+    with pytest.raises(Exception):
+        load_config(str(path))

server/tests/test_runtime_portability.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import json
+import socket
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+from admin_server import send_admin_command
+from lan_discovery import LanDiscoveryResponder
+from server_core import FalloutTogetherServer, get_lan_addresses
+from server_service import ServerConfig, ServerService
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+def test_get_lan_addresses_never_returns_wildcard() -> None:
+    addresses = get_lan_addresses()
+    assert "0.0.0.0" not in addresses
+    for address in addresses:
+        assert not address.startswith("127.")
+
+
+def test_discovery_failure_does_not_stop_game_server(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    game_port = _free_port()
+    admin_port = _free_port()
+
+    def _fail_start(self: LanDiscoveryResponder) -> None:
+        raise OSError("Could not bind LAN discovery to 0.0.0.0:7778. simulated failure")
+
+    monkeypatch.setattr(LanDiscoveryResponder, "start", _fail_start)
+
+    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", game_port), timeout=2.0) as conn:
+        data = conn.recv(4096)
+        assert b"welcome" in data
+
+    service.stop()
+    thread.join(timeout=3.0)
+    assert not service.is_running()
+
+
+def test_stop_is_idempotent(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"),
+        )
+    )
+    service.start()
+    deadline = time.time() + 5.0
+    while time.time() < deadline and not service.is_running():
+        time.sleep(0.05)
+    assert service.is_running()
+
+    service.stop()
+    service.stop()
+    assert not service.is_running()
+
+
+def test_admin_and_game_bind_errors_include_address() -> None:
+    occupied = _free_port()
+    holder = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        holder.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        holder.bind(("127.0.0.1", occupied))
+        holder.listen()
+
+        server = FalloutTogetherServer(host="127.0.0.1", port=occupied)
+        with pytest.raises(OSError, match=rf"127\.0\.0\.1:{occupied}"):
+            server._prepare_server_socket()
+    finally:
+        holder.close()
+
+
+def test_headless_admin_roundtrip(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()
+
+    response = send_admin_command({"cmd": "ping"}, port=admin_port)
+    assert response.get("ok") is True
+    assert response.get("data", {}).get("pong") is True
+
+    service.stop()
+    thread.join(timeout=3.0)
+
+
+def test_lan_discovery_sets_broadcast_option() -> None:
+    class DummyServer:
+        def get_stats(self):
+            return {
+                "connectedClients": 0,
+                "port": 7777,
+                "serverName": "Test",
+                "serverDescription": "",
+                "maxPlayers": 16,
+            }
+
+        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])
+    sock.close()
+
+    responder = LanDiscoveryResponder(DummyServer(), discovery_port=port)
+    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 = {
+            "type": "discover",
+            "protocol": "commonwealth-online",
+        }
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
+            client.settimeout(2.0)
+            client.sendto(json.dumps(probe).encode("utf-8"), ("127.0.0.1", port))
+            data, _addr = client.recvfrom(2048)
+            packet = json.loads(data.decode("utf-8"))
+            assert packet["type"] == "discoverResponse"
+            assert packet["port"] == 7777
+    finally:
+        responder.stop()

server/tests/test_sigterm_integration.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import json
+import os
+import signal
+import socket
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.skipif(os.name == "nt", reason="SIGTERM integration is POSIX-only")
+
+SERVER_DIR = Path(__file__).resolve().parents[1]
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+def test_sigterm_exits_cleanly(tmp_path: Path) -> None:
+    game_port = _free_port()
+    admin_port = _free_port()
+    config_path = tmp_path / "commonwealth-server.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "host": "127.0.0.1",
+                "port": game_port,
+                "server_name": "CI Test Server",
+                "server_description": "",
+                "max_players": 4,
+                "log_verbosity": "info",
+                "admin_port": admin_port,
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+
+    env = os.environ.copy()
+    env["PYTHONUNBUFFERED"] = "1"
+    env["NO_COLOR"] = "1"
+
+    process = subprocess.Popen(
+        [
+            sys.executable,
+            "-u",
+            str(SERVER_DIR / "consumer_server_cli.py"),
+            "serve",
+            "--config",
+            str(config_path),
+        ],
+        cwd=str(SERVER_DIR),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        env=env,
+    )
+
+    try:
+        deadline = time.time() + 10.0
+        connected = False
+        while time.time() < deadline:
+            try:
+                with socket.create_connection(("127.0.0.1", game_port), timeout=0.5) as conn:
+                    welcome = conn.recv(4096)
+                    if b"welcome" in welcome:
+                        connected = True
+                        break
+            except OSError:
+                time.sleep(0.1)
+        assert connected, "server did not accept a test client in time"
+
+        # Admin probe over localhost only.
+        with socket.create_connection(("127.0.0.1", admin_port), timeout=2.0) as admin:
+            admin.sendall(b'{"cmd":"ping"}\n')
+            response = admin.recv(4096)
+            assert b'"ok":true' in response.replace(b" ", b"")
+
+        process.send_signal(signal.SIGTERM)
+        try:
+            exit_code = process.wait(timeout=10.0)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            pytest.fail("server did not exit after SIGTERM")
+
+        assert exit_code == 0
+    finally:
+        if process.poll() is None:
+            process.kill()
+            process.wait(timeout=5.0)