VoxIP Kiosk OS 1.0.0: Raspberry Pi kiosk with web management

Chromium kiosk without a desktop for the Raspberry Pi 3, a Flask web
interface on port 8080 (kiosk, system and network management), a
restricted root helper, network changes with automatic rollback, a
healthcheck timer, install/update/uninstall scripts and a pi-gen image
build.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/voxip_kiosk/config.py

@@ -0,0 +1,249 @@
+"""Kiosk configuration: defaults, validation and atomic persistence.
+
+``load_config`` never raises: a missing, corrupt or partially invalid file
+falls back to the ``.bak`` copy and then to defaults per key, so a bad edit
+can never keep the kiosk from booting. ``save_config`` validates the whole
+result before anything touches the disk.
+"""
+
+from __future__ import annotations
+
+import copy
+import fcntl
+import json
+import logging
+import os
+import tempfile
+import threading
+from pathlib import Path
+from typing import Any, Callable
+
+from . import validation as v
+
+log = logging.getLogger(__name__)
+
+DEFAULTS: dict[str, Any] = {
+    "device_name": "VoxIP Kiosk",
+    "url": "https://schermen-ijsselflat.voxip.nl",
+    "refresh_interval": 0,
+    "screen_always_on": True,
+    "screen_off_minutes": 10,
+    "rotation": 0,
+    "resolution": "auto",
+    "zoom": 1.0,
+    "browser_restart": True,
+    "browser_profile": "persistent",
+    "admin_port": 8080,
+    "admin_https": False,
+    "admin_allow_public": False,
+}
+
+
+def _bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    raise v.ValidationError("Verwacht ja/nee.")
+
+
+def _int_range(lo: int, hi: int, message: str, allow_zero: bool = False) -> Callable[[Any], int]:
+    def check(value: Any) -> int:
+        if isinstance(value, bool) or not isinstance(value, (int, float, str)):
+            raise v.ValidationError(message)
+        try:
+            number = int(value)
+        except (TypeError, ValueError):
+            raise v.ValidationError(message) from None
+        if allow_zero and number == 0:
+            return 0
+        if not lo <= number <= hi:
+            raise v.ValidationError(message)
+        return number
+
+    return check
+
+
+def _rotation(value: Any) -> int:
+    try:
+        number = int(value)
+    except (TypeError, ValueError):
+        number = -1
+    if isinstance(value, bool) or number not in (0, 90, 180, 270):
+        raise v.ValidationError("Rotatie moet 0, 90, 180 of 270 graden zijn.")
+    return number
+
+
+def _zoom(value: Any) -> float:
+    try:
+        number = round(float(value), 2)
+    except (TypeError, ValueError):
+        number = -1.0
+    if isinstance(value, bool) or not 0.25 <= number <= 4.0:
+        raise v.ValidationError("Zoomfactor moet tussen 0,25 en 4 liggen.")
+    return number
+
+
+def _choice(*options: str) -> Callable[[Any], str]:
+    def check(value: Any) -> str:
+        if value not in options:
+            raise v.ValidationError("Kies een van: " + ", ".join(options) + ".")
+        return value
+
+    return check
+
+
+def _str(fn: Callable[[str], Any]) -> Callable[[Any], Any]:
+    def check(value: Any) -> Any:
+        if not isinstance(value, str):
+            raise v.ValidationError("Verwacht tekst.")
+        return fn(value)
+
+    return check
+
+
+VALIDATORS: dict[str, Callable[[Any], Any]] = {
+    "device_name": _str(v.device_name),
+    "url": _str(v.kiosk_url),
+    "refresh_interval": _int_range(30, 86400, "Verversinterval: 0 (uit) of 30 tot 86400 seconden.", allow_zero=True),
+    "screen_always_on": _bool,
+    "screen_off_minutes": _int_range(1, 240, "Scherm uit na: 1 tot 240 minuten."),
+    "rotation": _rotation,
+    "resolution": _str(v.resolution),
+    "zoom": _zoom,
+    "browser_restart": _bool,
+    "browser_profile": _choice("persistent", "ephemeral"),
+    "admin_port": _int_range(1024, 65535, "Poort moet tussen 1024 en 65535 liggen."),
+    "admin_https": _bool,
+    "admin_allow_public": _bool,
+}
+
+
+class ConfigError(Exception):
+    def __init__(self, errors: dict[str, str]):
+        super().__init__("; ".join(f"{k}: {m}" for k, m in errors.items()))
+        self.errors = errors
+
+
+def validate(updates: dict[str, Any]) -> dict[str, Any]:
+    """Validate known keys in ``updates``; raise ConfigError listing every problem."""
+    clean: dict[str, Any] = {}
+    errors: dict[str, str] = {}
+    for key, value in updates.items():
+        check = VALIDATORS.get(key)
+        if check is None:
+            errors[key] = "Onbekende instelling."
+            continue
+        try:
+            clean[key] = check(value)
+        except v.ValidationError as exc:
+            errors[key] = str(exc)
+    if errors:
+        raise ConfigError(errors)
+    return clean
+
+
+def _read_json(path: Path) -> dict[str, Any] | None:
+    try:
+        with path.open(encoding="utf-8") as fh:
+            data = json.load(fh)
+    except FileNotFoundError:
+        return None
+    except (OSError, ValueError) as exc:
+        log.warning("Kan %s niet lezen: %s", path, exc)
+        return None
+    if not isinstance(data, dict):
+        log.warning("%s bevat geen JSON-object", path)
+        return None
+    return data
+
+
+def _read_raw(path: Path) -> dict[str, Any]:
+    data = _read_json(path)
+    if data is None and path.exists():
+        data = _read_json(path.with_suffix(path.suffix + ".bak"))
+        if data is not None:
+            log.warning("Configuratie hersteld uit %s.bak", path.name)
+    return data or {}
+
+
+def load_config(path: Path) -> dict[str, Any]:
+    """Effective configuration: defaults overlaid with every valid stored value."""
+    raw = _read_raw(path)
+    config = copy.deepcopy(DEFAULTS)
+    for key, value in raw.items():
+        check = VALIDATORS.get(key)
+        if check is None:
+            continue
+        try:
+            config[key] = check(value)
+        except v.ValidationError as exc:
+            log.warning("Ongeldige waarde voor %s in %s (%s); standaard gebruikt", key, path, exc)
+    return config
+
+
+def atomic_write(path: Path, content: bytes, mode: int = 0o640, owner: tuple[int, int] | None = None) -> None:
+    """Write ``content`` to ``path`` so readers see either the old or the new file, never a partial one."""
+    path = Path(path)
+    if owner is None and path.exists():
+        st = path.stat()
+        owner = (st.st_uid, st.st_gid)
+    fd, tmp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "wb") as fh:
+            fh.write(content)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.chmod(tmp, mode)
+        if owner is not None and os.geteuid() == 0:
+            os.chown(tmp, *owner)
+        os.replace(tmp, path)
+    except BaseException:
+        try:
+            os.unlink(tmp)
+        except FileNotFoundError:
+            pass
+        raise
+    dir_fd = os.open(str(path.parent), os.O_RDONLY)
+    try:
+        os.fsync(dir_fd)
+    finally:
+        os.close(dir_fd)
+
+
+def atomic_write_json(path: Path, data: Any, mode: int = 0o640, owner: tuple[int, int] | None = None) -> None:
+    content = (json.dumps(data, indent=2, ensure_ascii=False) + "\n").encode("utf-8")
+    atomic_write(path, content, mode=mode, owner=owner)
+
+
+_thread_lock = threading.Lock()
+
+
+def save_config(path: Path, updates: dict[str, Any]) -> dict[str, Any]:
+    """Validate ``updates``, merge them into the stored file and write it atomically.
+
+    Unknown keys already in the file are preserved; the previous version is
+    kept as ``config.json.bak``. Returns the new effective configuration.
+    """
+    clean = validate(updates)
+    path = Path(path)
+    lock_path = path.with_name(".config.lock")
+    with _thread_lock:
+        # Read-only is enough for flock and works for every writer (the web
+        # interface as voxip, the helper as root), whoever created the file.
+        lock_fd = os.open(str(lock_path), os.O_RDONLY | os.O_CREAT, 0o640)
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_EX)
+            return _save_locked(path, clean)
+        finally:
+            os.close(lock_fd)
+
+
+def _save_locked(path: Path, clean: dict[str, Any]) -> dict[str, Any]:
+    raw = _read_raw(path)
+    if path.exists():
+        try:
+            atomic_write(path.with_suffix(path.suffix + ".bak"), path.read_bytes())
+        except OSError as exc:
+            log.warning("Kon geen back-up van de configuratie maken: %s", exc)
+    raw.update(clean)
+    atomic_write_json(path, raw)
+    return load_config(path)