""" 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 @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 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, } @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 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, ) 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 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" ) 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