Add env-driven config, service models and atomic JSON store

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),
+        )

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

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