From 7440a1061c903a9dceeb2b6933af71f9094dcba5 Mon Sep 17 00:00:00 2001
From: rekushi <1+rekushi@noreply.flipnote.world>
Date: Thu, 29 Oct 2026 04:38:57 +0000
Subject: [PATCH] feat(config): add validated server settings

Provide defaults, validate configuration and save recognised settings atomically.
---
 server_config.py | 129 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 129 insertions(+)
 create mode 100644 server_config.py

diff --git a/server_config.py b/server_config.py
new file mode 100644
index 0000000..646ccd8
--- /dev/null
+++ b/server_config.py
@@ -0,0 +1,129 @@
+"""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