""" 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