RexiMemo-OSE/server_config.py

"""Small, validated server settings shared by the web and DSi pages."""
from __future__ import annotations

from pathlib import Path
from urllib.parse import urlsplit
import ipaddress
import json
import os
import re
import tempfile

ROOT = Path(__file__).resolve().parent
DEFAULTS = {
    "site_name": "RexiMemo OSE",
    "tagline": "Open Source Edition",
    "welcome_message": "A place to share and enjoy Flipnotes.",
    "accent_colour": "#b13b29",
    "logo_url": "/static/RexiMemo_OSE_Logo.png",
    "registration_enabled": True,
    "guest_posting": True,
    "max_upload_bytes": 2 * 1024 * 1024,
    "host": "0.0.0.0",
    "port": 8080,
    "setup_complete": False,
}


def config_path() -> Path:
    configured = Path(os.environ.get("REXIMEMO_CONFIG", "server-settings.json"))
    return configured if configured.is_absolute() else ROOT / configured


def _boolean(value, label: str) -> bool:
    if isinstance(value, bool):
        return value
    if value in (1, "1", "true", "on", "yes"):
        return True
    if value in (0, "0", "false", "off", "no", ""):
        return False
    raise ValueError(label + " must be on or off.")


def validate_config(fields: dict) -> dict:
    result = {}
    for key, default in DEFAULTS.items():
        value = fields.get(key, default)
        if key in {"site_name", "tagline", "welcome_message"}:
            text = str(value or "").strip()
            maximum = {"site_name": 64, "tagline": 120, "welcome_message": 1000}[key]
            if (key == "site_name" and not text) or len(text) > maximum:
                raise ValueError(key.replace("_", " ").capitalize() + f" must be between {1 if key == 'site_name' else 0} and {maximum} characters.")
            if any(ord(character) < 32 and character not in "\n\t" for character in text):
                raise ValueError("Server text contains an unsupported control character.")
            result[key] = text
        elif key in {"registration_enabled", "guest_posting", "setup_complete"}:
            result[key] = _boolean(value, key.replace("_", " "))
        elif key == "accent_colour":
            value = str(value).strip()
            if not re.fullmatch(r"#[0-9A-Fa-f]{6}", value):
                raise ValueError("Choose an accent colour in #RRGGBB format.")
            result[key] = value.lower()
        elif key == "logo_url":
            value = str(value).strip()
            target = urlsplit(value)
            if value == "/branding/logo.png":
                result[key] = value
                continue
            if (target.scheme or target.netloc or target.query or target.fragment
                    or not value.startswith("/static/") or "\\" in value or "%" in value
                    or any(part in {".", ".."} for part in value.split("/"))
                    or Path(target.path).suffix.lower() not in {".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg"}):
                raise ValueError("Use an image under /static/ or upload a server logo.")
            result[key] = value
        elif key == "host":
            value = str(value).strip()
            if value != "localhost":
                try:
                    ipaddress.ip_address(value)
                except ValueError as exc:
                    raise ValueError("Listen address must be an IP address or localhost.") from exc
            result[key] = value
        else:
            try:
                number = int(value)
            except (TypeError, ValueError) as exc:
                raise ValueError(key.replace("_", " ") + " must be a whole number.") from exc
            minimum, maximum = (1024, 65535) if key == "port" else (65536, 16 * 1024 * 1024)
            if not minimum <= number <= maximum:
                raise ValueError(f"{key.replace('_', ' ').capitalize()} must be between {minimum} and {maximum}.")
            result[key] = number
    return result


def load_config() -> dict:
    path = config_path()
    if not path.exists():
        return dict(DEFAULTS)
    try:
        fields = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise ValueError(f"Cannot read {path.name}. Check that it contains valid JSON.") from exc
    if not isinstance(fields, dict):
        raise ValueError(f"{path.name} must contain a JSON object.")
    return validate_config(fields)


def save_config(fields: dict) -> dict:
    """Merge recognised fields and replace the settings file atomically."""
    if not isinstance(fields, dict):
        raise ValueError("Server settings must be a group of named fields.")
    merged = load_config()
    merged.update({key: value for key, value in fields.items() if key in DEFAULTS})
    result = validate_config(merged)
    path = config_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=path.parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
            json.dump(result, stream, ensure_ascii=False, indent=2)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        if path.exists():
            temporary.chmod(path.stat().st_mode & 0o777)
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)
    return result