Commonwealth-Online-Server/server/config.py

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