From 137d528ad289fe98dc390313f40be906e4b6b5eb Mon Sep 17 00:00:00 2001 From: Archibold Date: Wed, 28 Oct 2026 21:44:52 +1100 Subject: [PATCH] Add env-driven config, service models and atomic JSON store --- src/uwudash_mk3/config.py | 72 +++++++++++++++++++++++ src/uwudash_mk3/models.py | 119 ++++++++++++++++++++++++++++++++++++++ src/uwudash_mk3/store.py | 108 ++++++++++++++++++++++++++++++++++ 3 files changed, 299 insertions(+) create mode 100644 src/uwudash_mk3/config.py create mode 100644 src/uwudash_mk3/models.py create mode 100644 src/uwudash_mk3/store.py diff --git a/src/uwudash_mk3/config.py b/src/uwudash_mk3/config.py new file mode 100644 index 0000000..c8662aa --- /dev/null +++ b/src/uwudash_mk3/config.py @@ -0,0 +1,72 @@ +"""Runtime configuration for the dashboard. + +Everything is driven by environment variables (with sane defaults) so the +dashboard can be deployed without editing code: + + UWUDASH_HOST, UWUDASH_PORT, UWUDASH_DATA_DIR, UWUDASH_SERVICES_FILE, + UWUDASH_PROXY_TIMEOUT, UWUDASH_STRIP_FRAME_OPTIONS +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8000 +DEFAULT_DATA_DIR = Path("data") +DEFAULT_PROXY_TIMEOUT = 30.0 + + +def _int(name: str, default: int) -> int: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + try: + return int(raw) + except ValueError as exc: # pragma: no cover - misconfiguration is loud + raise ValueError(f"{name} must be an integer, got {raw!r}") from exc + + +def _float(name: str, default: float) -> float: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + try: + return float(raw) + except ValueError as exc: # pragma: no cover - misconfiguration is loud + raise ValueError(f"{name} must be a number, got {raw!r}") from exc + + +def _bool(name: str, default: bool) -> bool: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class Config: + host: str + port: int + data_dir: Path + services_file: Path + proxy_timeout: float + # Internal apps often send X-Frame-Options, which would block the iframe. + strip_frame_options: bool + + @classmethod + def from_env(cls) -> "Config": + data_dir = Path(os.environ.get("UWUDASH_DATA_DIR", DEFAULT_DATA_DIR)) + services_file = Path( + os.environ.get("UWUDASH_SERVICES_FILE", data_dir / "services.json") + ) + return cls( + host=os.environ.get("UWUDASH_HOST", DEFAULT_HOST), + port=_int("UWUDASH_PORT", DEFAULT_PORT), + data_dir=data_dir, + services_file=services_file, + proxy_timeout=_float("UWUDASH_PROXY_TIMEOUT", DEFAULT_PROXY_TIMEOUT), + strip_frame_options=_bool("UWUDASH_STRIP_FRAME_OPTIONS", True), + ) diff --git a/src/uwudash_mk3/models.py b/src/uwudash_mk3/models.py new file mode 100644 index 0000000..fbc763c --- /dev/null +++ b/src/uwudash_mk3/models.py @@ -0,0 +1,119 @@ +"""Service definitions and validation.""" + +from __future__ import annotations + +import re +from typing import Dict, Optional +from urllib.parse import urlsplit + +from pydantic import BaseModel, Field, computed_field, field_validator + +ALLOWED_SCHEMES = {"http", "https"} +OPEN_MODES = ("iframe", "newtab") + + +def origin_of(url: str) -> str: + """Return ``scheme://host[:port]`` for an absolute URL.""" + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}" + + +def base_path_of(url: str) -> str: + """Return the path prefix of an absolute URL (``""`` when at the root).""" + parts = urlsplit(url) + path = parts.path + if path.startswith("/"): + path = path[1:] + return path + + +def slugify(name: str) -> str: + slug = re.sub(r"[^a-z0-9-]+", "-", name.strip().lower()).strip("-") + return slug or "service" + + +class ServiceInput(BaseModel): + """Payload accepted by the create/update endpoints.""" + + name: str = Field(min_length=1, max_length=120) + url: str = Field(min_length=1, max_length=500) + description: str = Field(default="", max_length=500) + icon: str = Field(default="", max_length=8) + open_in: str = Field(default="iframe") + enabled: bool = True + headers: Dict[str, str] = Field(default_factory=dict) + + @field_validator("url") + def _validate_url(cls, value: str) -> str: + parts = urlsplit(value.strip()) + if parts.scheme not in ALLOWED_SCHEMES: + raise ValueError("url must start with http:// or https://") + if not parts.netloc: + raise ValueError("url must include a host") + return value.strip() + + @field_validator("open_in") + def _validate_open_in(cls, value: str) -> str: + if value not in OPEN_MODES: + raise ValueError(f"open_in must be one of {OPEN_MODES}") + return value + + @field_validator("headers") + def _validate_headers(cls, value: Dict[str, str]) -> Dict[str, str]: + for key in value: + if not key or any(c in key for c in "\r\n"): + raise ValueError("header names must not contain newlines") + return value + + +class Service(ServiceInput): + """A stored service.""" + + id: str + created_at: str + updated_at: str + + @computed_field + @property + def origin(self) -> str: + return origin_of(self.url) + + @computed_field + @property + def base_path(self) -> str: + return base_path_of(self.url) + + @computed_field + @property + def proxy_path(self) -> str: + """Dashboard-relative path used by the reverse proxy.""" + return f"/proxy/{self.id}/{self.base_path}" + + +class ServicePatch(BaseModel): + """Partial update payload; unset fields are left alone.""" + + name: Optional[str] = Field(default=None, min_length=1, max_length=120) + url: Optional[str] = Field(default=None, min_length=1, max_length=500) + description: Optional[str] = Field(default=None, max_length=500) + icon: Optional[str] = Field(default=None, max_length=8) + open_in: Optional[str] = None + enabled: Optional[bool] = None + headers: Optional[Dict[str, str]] = None + + @field_validator("url") + def _validate_url(cls, value: Optional[str]) -> Optional[str]: + if value is None: + return value + parts = urlsplit(value.strip()) + if parts.scheme not in ALLOWED_SCHEMES or not parts.netloc: + raise ValueError("url must start with http:// or https://") + return value.strip() + + @field_validator("open_in") + def _validate_open_in(cls, value: Optional[str]) -> Optional[str]: + if value is None: + return value + if value not in OPEN_MODES: + raise ValueError(f"open_in must be one of {OPEN_MODES}") + return value diff --git a/src/uwudash_mk3/store.py b/src/uwudash_mk3/store.py new file mode 100644 index 0000000..2ae5aed --- /dev/null +++ b/src/uwudash_mk3/store.py @@ -0,0 +1,108 @@ +"""Persistence for service definitions. + +The store is a plain JSON file so it is easy to inspect, back up, or seed from +a container. Writes are atomic (temp file + ``os.replace``) and serialised with +a lock so concurrent API requests cannot interleave. +""" + +from __future__ import annotations + +import json +import os +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import List, Optional + +from .models import Service, ServiceInput, ServicePatch, slugify + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +class ServiceStore: + def __init__(self, path: Path) -> None: + self.path = Path(path) + self._lock = threading.Lock() + self.path.parent.mkdir(parents=True, exist_ok=True) + if not self.path.exists(): + self._write([]) + + # ------------------------------------------------------------------ read + def _read(self) -> List[Service]: + try: + raw = json.loads(self.path.read_text(encoding="utf-8")) + except FileNotFoundError: + return [] + except json.JSONDecodeError as exc: + raise ValueError(f"{self.path} is not valid JSON: {exc}") from exc + items = raw.get("services", []) if isinstance(raw, dict) else raw + return [Service.model_validate(item) for item in items] + + def _write(self, services: List[Service]) -> None: + payload = { + "version": 1, + "services": [ + s.model_dump(exclude={"origin", "base_path", "proxy_path"}) + for s in services + ], + } + tmp = self.path.with_suffix(".tmp") + tmp.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, self.path) + + def list(self) -> List[Service]: + with self._lock: + return self._read() + + def get(self, service_id: str) -> Optional[Service]: + with self._lock: + for service in self._read(): + if service.id == service_id: + return service + return None + + # ----------------------------------------------------------------- write + def create(self, data: ServiceInput) -> Service: + with self._lock: + services = self._read() + base = slugify(data.name) + candidate = base + suffix = 1 + existing = {s.id for s in services} + while candidate in existing: + suffix += 1 + candidate = f"{base}-{suffix}" + now = _now() + service = Service( + **data.model_dump(), id=candidate, created_at=now, updated_at=now + ) + services.append(service) + self._write(services) + return service + + def update(self, service_id: str, patch: ServicePatch) -> Optional[Service]: + with self._lock: + services = self._read() + for index, service in enumerate(services): + if service.id != service_id: + continue + updates = {k: v for k, v in patch.model_dump().items() if v is not None} + if not updates: + return service + updated = service.model_copy(update=updates) + updated.updated_at = _now() + services[index] = updated + self._write(services) + return updated + return None + + def delete(self, service_id: str) -> bool: + with self._lock: + services = self._read() + remaining = [s for s in services if s.id != service_id] + if len(remaining) == len(services): + return False + self._write(remaining) + return True