269 lines
8.3 KiB
Python
269 lines
8.3 KiB
Python
"""
|
|
Configuration model and loading for Commonwealth Online server.
|
|
|
|
Supports JSON-based configuration files for hosted deployment.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SERVER_NAME_MAX_LENGTH = 64
|
|
SERVER_DESCRIPTION_MAX_LENGTH = 256
|
|
MAX_PLAYERS_HARD_LIMIT = 256
|
|
|
|
DEFAULT_ADMIN_PORT = 7779
|
|
|
|
|
|
|
|
def _parse_bool(value: Any, *, field_name: str, default: bool = False) -> bool:
|
|
if value is None:
|
|
return default
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, int) and not isinstance(value, bool) and value in (0, 1):
|
|
return bool(value)
|
|
if isinstance(value, str):
|
|
text = value.strip().lower()
|
|
if text in {"1", "true", "yes", "on"}:
|
|
return True
|
|
if text in {"0", "false", "no", "off", ""}:
|
|
return False
|
|
raise ValueError(
|
|
f"{field_name} must be a boolean or one of true/false, yes/no, on/off, 1/0"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
"""Server configuration model."""
|
|
host: str = "0.0.0.0"
|
|
port: int = 7777
|
|
server_name: str = "Commonwealth Online Server"
|
|
server_description: str = ""
|
|
max_players: int = 16
|
|
log_verbosity: str = "info"
|
|
admin_port: int = DEFAULT_ADMIN_PORT
|
|
enable_gns_transport: bool = False
|
|
gns_bridge_path: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Convert config to dictionary."""
|
|
return {
|
|
"host": self.host,
|
|
"port": self.port,
|
|
"server_name": self.server_name,
|
|
"server_description": self.server_description,
|
|
"max_players": self.max_players,
|
|
"log_verbosity": self.log_verbosity,
|
|
"admin_port": self.admin_port,
|
|
"enable_gns_transport": self.enable_gns_transport,
|
|
"gns_bridge_path": self.gns_bridge_path,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> Config:
|
|
"""Create config from dictionary."""
|
|
try:
|
|
port = int(data.get("port", 7777))
|
|
max_players = int(data.get("max_players", 16))
|
|
admin_port = int(data.get("admin_port", DEFAULT_ADMIN_PORT))
|
|
except (TypeError, ValueError) as error:
|
|
raise ValueError(f"Invalid numeric config field: {error}") from error
|
|
|
|
enable_gns_transport = _parse_bool(
|
|
data.get("enable_gns_transport"),
|
|
field_name="enable_gns_transport",
|
|
default=False,
|
|
)
|
|
raw_bridge_path = data.get("gns_bridge_path")
|
|
if raw_bridge_path is None:
|
|
gns_bridge_path = None
|
|
elif isinstance(raw_bridge_path, str):
|
|
normalized_bridge_path = raw_bridge_path.strip()
|
|
gns_bridge_path = normalized_bridge_path or None
|
|
else:
|
|
raise ValueError("gns_bridge_path must be a string or null")
|
|
|
|
return cls(
|
|
host=str(data.get("host", "0.0.0.0")),
|
|
port=port,
|
|
server_name=str(data.get("server_name", "Commonwealth Online Server")),
|
|
server_description=str(data.get("server_description", "")),
|
|
max_players=max_players,
|
|
log_verbosity=str(data.get("log_verbosity", "info")),
|
|
admin_port=admin_port,
|
|
enable_gns_transport=enable_gns_transport,
|
|
gns_bridge_path=gns_bridge_path,
|
|
)
|
|
|
|
|
|
def load_config(config_path: str | None = None) -> Config:
|
|
"""
|
|
Load configuration from file or use defaults.
|
|
|
|
Args:
|
|
config_path: Path to config.json file. If None, returns defaults.
|
|
|
|
Returns:
|
|
Config instance.
|
|
|
|
Raises:
|
|
FileNotFoundError: If config_path is provided but file does not exist.
|
|
json.JSONDecodeError: If config file is invalid JSON.
|
|
ValueError: If config root is not an object or values are invalid.
|
|
"""
|
|
if config_path is None:
|
|
return Config()
|
|
|
|
path = Path(config_path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Config file not found: {config_path}")
|
|
|
|
with open(path, "r", encoding="utf-8", newline=None) as f:
|
|
data = json.load(f)
|
|
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Config file must contain a JSON object at root level.")
|
|
|
|
return Config.from_dict(data)
|
|
|
|
|
|
def save_config(config: Config, config_path: str) -> None:
|
|
"""
|
|
Save configuration to a JSON file.
|
|
|
|
Args:
|
|
config: Config instance to save.
|
|
config_path: Path to write config.json.
|
|
"""
|
|
path = Path(config_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(path, "w", encoding="utf-8", newline="\n") as f:
|
|
json.dump(config.to_dict(), f, indent=2, ensure_ascii=False)
|
|
f.write("\n")
|
|
|
|
|
|
def generate_default_config(config_path: str) -> Config:
|
|
"""
|
|
Generate and save a default configuration file.
|
|
|
|
Args:
|
|
config_path: Path where default config.json will be written.
|
|
|
|
Returns:
|
|
The generated Config instance.
|
|
"""
|
|
config = Config()
|
|
save_config(config, config_path)
|
|
return config
|
|
|
|
|
|
def _is_valid_bind_host(host: str) -> bool:
|
|
"""Return True when host can be used as an AF_INET bind address."""
|
|
text = host.strip()
|
|
if not text:
|
|
return False
|
|
if text in ("0.0.0.0", "127.0.0.1", "localhost"):
|
|
return True
|
|
try:
|
|
socket.inet_aton(text)
|
|
return True
|
|
except OSError:
|
|
pass
|
|
try:
|
|
socket.getaddrinfo(text, None, family=socket.AF_INET, type=socket.SOCK_STREAM)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _is_valid_gns_bind_host(host: str) -> bool:
|
|
"""Return True for the strict IPv4 bind form used by the native GNS bridge."""
|
|
text = host.strip()
|
|
if not text:
|
|
return False
|
|
try:
|
|
socket.inet_pton(socket.AF_INET, text)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def validate_config(config: Config) -> tuple[bool, list[str]]:
|
|
"""
|
|
Validate configuration values.
|
|
|
|
Args:
|
|
config: Config to validate.
|
|
|
|
Returns:
|
|
(is_valid, list_of_errors). Empty list if valid.
|
|
"""
|
|
errors: list[str] = []
|
|
|
|
if not config.host or not str(config.host).strip():
|
|
errors.append("host cannot be empty")
|
|
elif not _is_valid_bind_host(str(config.host)):
|
|
errors.append(
|
|
f"host '{config.host}' is not a valid IPv4 address or resolvable hostname"
|
|
)
|
|
elif config.enable_gns_transport and not _is_valid_gns_bind_host(str(config.host)):
|
|
errors.append(
|
|
"enable_gns_transport requires host to be an explicit IPv4 bind address "
|
|
"such as 0.0.0.0 or 127.0.0.1"
|
|
)
|
|
|
|
if config.port < 1 or config.port > 65535:
|
|
errors.append(f"port must be 1-65535, got {config.port}")
|
|
|
|
if config.admin_port < 1 or config.admin_port > 65535:
|
|
errors.append(f"admin_port must be 1-65535, got {config.admin_port}")
|
|
elif config.admin_port == config.port:
|
|
errors.append("admin_port must differ from the game port")
|
|
|
|
if not config.server_name:
|
|
errors.append("server_name cannot be empty")
|
|
elif len(config.server_name) > SERVER_NAME_MAX_LENGTH:
|
|
errors.append(
|
|
f"server_name must be <= {SERVER_NAME_MAX_LENGTH} characters, "
|
|
f"got {len(config.server_name)}"
|
|
)
|
|
|
|
if len(config.server_description) > SERVER_DESCRIPTION_MAX_LENGTH:
|
|
errors.append(
|
|
f"server_description must be <= {SERVER_DESCRIPTION_MAX_LENGTH} characters, "
|
|
f"got {len(config.server_description)}"
|
|
)
|
|
|
|
if config.max_players < 1:
|
|
errors.append(f"max_players must be >= 1, got {config.max_players}")
|
|
elif config.max_players > MAX_PLAYERS_HARD_LIMIT:
|
|
errors.append(
|
|
f"max_players must be <= {MAX_PLAYERS_HARD_LIMIT}, got {config.max_players}"
|
|
)
|
|
|
|
if config.log_verbosity not in ("debug", "info", "warning", "error"):
|
|
errors.append(f"log_verbosity must be debug/info/warning/error, got {config.log_verbosity}")
|
|
|
|
return len(errors) == 0, errors
|
|
|
|
|
|
def ensure_writable_directory(path: Path) -> None:
|
|
"""Raise PermissionError when the directory cannot be created or written."""
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
probe = path / ".commonwealth-write-probe"
|
|
try:
|
|
probe.write_text("ok\n", encoding="utf-8", newline="\n")
|
|
finally:
|
|
try:
|
|
probe.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|