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