announcer/config.py

#!/usr/bin/env python3
"""Konfigurasjon: valgfri config.toml pluss miljøvariabler.

Appen virker helt uten config.toml – da gjelder DEFAULTS, som er den gamle
oppførselen: ingen innlogging, norsk tekst, samme mapper som før.

Tre lag, der det siste vinner:

    DEFAULTS  ->  config.toml  ->  ANNOUNCER_*-miljøvariabler

Hemmeligheter (PIN, JWT-nøkkel, OIDC-client_secret) hører hjemme i det siste
laget, så de kan stå i systemd-enheten istedenfor i en fil som ligger i repoet.

Egen modul fordi både announcer.py og auth.py trenger den; ville blitt en
import-sirkel om den bodde i announcer.py.
"""

from __future__ import annotations

import copy
import os
import tomllib
from pathlib import Path
from typing import Any, Callable

HERE = Path(__file__).resolve().parent

# Alt man kan få lov til å gjøre. Roller er navngitte knipper av disse.
ACTIONS = ("view", "preview", "announce", "stop", "clear")

DEFAULTS: dict[str, Any] = {
    "app": {
        "title": "Announcer",
        "subtitle": "Skriv meldingen, forhåndslytt, og send den ut på anlegget.",
        "accent": "#4fd1c5",
        "idle_text": "Stille · ingenting i kø",
        "message_placeholder": "Skriv meldingen som skal leses opp …",
        "max_chars": 600,
        "quick_phrases": [],
        "default_voice": "finn",
        "default_chime": "dingdong",
    },
    "paths": {
        "voices": str(HERE / "voices"),
        "cache": str(HERE / "cache"),
        "chimes": str(HERE / "chimes"),
        "log": str(HERE / "logs" / "announcements.jsonl"),
        "state": str(HERE / "instance"),
    },
    "audio": {
        # pw-play --target / paplay --device. Tom = systemets standardutgang.
        "sink": "",
    },
    "server": {
        "host": "0.0.0.0",
        "port": 8090,
        # Settes bare når appen står bak HTTPS. Styrer Secure-flagget på
        # sesjonscookien; på i rå HTTP betyr ingen innlogging i det hele tatt.
        "behind_tls": False,
        "secret_key": "",
        # Full URL utenfra, trengs for OIDC redirect_uri bak en proxy.
        "public_url": "",
    },
    "auth": {
        # Tom liste = alt åpent. Se active_providers(): en leverandør er også
        # aktiv når den er ferdig konfigurert, så det holder å sette
        # ANNOUNCER_PIN eller ANNOUNCER_JWT_SECRET.
        "providers": [],
    },
    "auth.pin": {
        "pin": "",
        # admin, så dagens PIN-brukere beholder stopp og tøm kø.
        "role": "admin",
        "label": "PIN",
        "hint": "",
    },
    "auth.jwt": {
        "issuer": "",
        "audience": "",
        "secret": "",
        "jwks_url": "",
        "algorithms": ["HS256"],
        "role_claim": "roles",
        "name_claim": "",
        "default_role": "",
        # Hvor lenge en sesjon fra /auth/token varer, i sekunder.
        "session_seconds": 43200,
        "leeway": 30,
    },
    "auth.oidc": {
        "issuer": "",
        "client_id": "",
        "client_secret": "",
        "scopes": ["openid", "profile", "email"],
        "role_claim": "roles",
        "name_claim": "",
        "default_role": "",
        # Settes bare hvis panelet mangler discovery.
        "authorization_endpoint": "",
        "token_endpoint": "",
        "jwks_url": "",
        "end_session_endpoint": "",
        "label": "Logg inn",
        "verify_signature": True,
    },
    "roles": {
        "anonymous": [],
        "crew": ["view", "preview", "announce"],
        "admin": ["*"],
    },
}


def _bool(v: str) -> bool:
    return v.strip().lower() in ("1", "true", "yes", "on", "ja")


def _list(v: str) -> list[str]:
    return [p.strip() for p in v.replace(",", " ").split() if p.strip()]


# Miljøvariabel -> (sti i konfigtreet, konvertering).
ENV: dict[str, tuple[tuple[str, str], Callable[[str], Any]]] = {
    "ANNOUNCER_TITLE": (("app", "title"), str),
    "ANNOUNCER_SUBTITLE": (("app", "subtitle"), str),
    "ANNOUNCER_ACCENT": (("app", "accent"), str),
    "ANNOUNCER_MAX_CHARS": (("app", "max_chars"), int),
    "ANNOUNCER_DEFAULT_VOICE": (("app", "default_voice"), str),
    "ANNOUNCER_DEFAULT_CHIME": (("app", "default_chime"), str),
    "ANNOUNCER_VOICES": (("paths", "voices"), str),
    "ANNOUNCER_CACHE": (("paths", "cache"), str),
    "ANNOUNCER_CHIMES": (("paths", "chimes"), str),
    "ANNOUNCER_LOG": (("paths", "log"), str),
    "ANNOUNCER_STATE": (("paths", "state"), str),
    "ANNOUNCER_SINK": (("audio", "sink"), str),
    "ANNOUNCER_HOST": (("server", "host"), str),
    "ANNOUNCER_PORT": (("server", "port"), int),
    "ANNOUNCER_BEHIND_TLS": (("server", "behind_tls"), _bool),
    "ANNOUNCER_SECRET_KEY": (("server", "secret_key"), str),
    "ANNOUNCER_PUBLIC_URL": (("server", "public_url"), str),
    "ANNOUNCER_AUTH": (("auth", "providers"), _list),
    "ANNOUNCER_PIN": (("auth.pin", "pin"), str),
    "ANNOUNCER_PIN_ROLE": (("auth.pin", "role"), str),
    "ANNOUNCER_JWT_ISSUER": (("auth.jwt", "issuer"), str),
    "ANNOUNCER_JWT_AUDIENCE": (("auth.jwt", "audience"), str),
    "ANNOUNCER_JWT_SECRET": (("auth.jwt", "secret"), str),
    "ANNOUNCER_JWT_JWKS_URL": (("auth.jwt", "jwks_url"), str),
    "ANNOUNCER_JWT_ALGORITHMS": (("auth.jwt", "algorithms"), _list),
    "ANNOUNCER_JWT_ROLE_CLAIM": (("auth.jwt", "role_claim"), str),
    "ANNOUNCER_JWT_DEFAULT_ROLE": (("auth.jwt", "default_role"), str),
    "ANNOUNCER_OIDC_ISSUER": (("auth.oidc", "issuer"), str),
    "ANNOUNCER_OIDC_CLIENT_ID": (("auth.oidc", "client_id"), str),
    "ANNOUNCER_OIDC_CLIENT_SECRET": (("auth.oidc", "client_secret"), str),
    "ANNOUNCER_OIDC_SCOPES": (("auth.oidc", "scopes"), _list),
    "ANNOUNCER_OIDC_ROLE_CLAIM": (("auth.oidc", "role_claim"), str),
    "ANNOUNCER_OIDC_DEFAULT_ROLE": (("auth.oidc", "default_role"), str),
}


def _merge(base: dict, over: dict) -> dict:
    """Dyp fletting. Lister erstattes, de slås ikke sammen."""
    out = dict(base)
    for k, v in over.items():
        if isinstance(v, dict) and isinstance(out.get(k), dict):
            out[k] = _merge(out[k], v)
        else:
            out[k] = v
    return out


def _flatten_tables(raw: dict) -> dict:
    """[auth.pin] i TOML blir {"auth": {"pin": …}}; vi vil ha "auth.pin" flatt.

    Det gjør oppslag i resten av koden kjedelig enkelt, og det er bare auth
    som har undertabeller.
    """
    out = dict(raw)
    auth = out.get("auth")
    if isinstance(auth, dict):
        rest = {}
        for k, v in auth.items():
            if isinstance(v, dict):
                out[f"auth.{k}"] = _merge(out.get(f"auth.{k}", {}), v)
            else:
                rest[k] = v
        out["auth"] = rest
    return out


class Config:
    def __init__(self, data: dict[str, Any], source: Path | None = None) -> None:
        self.data = data
        self.source = source

    # -- oppslag ----------------------------------------------------------
    def section(self, name: str) -> dict[str, Any]:
        return self.data.get(name, {})

    def get(self, section: str, key: str, default: Any = None) -> Any:
        return self.data.get(section, {}).get(key, default)

    @property
    def app(self) -> dict[str, Any]:
        return self.section("app")

    @property
    def roles(self) -> dict[str, list[str]]:
        return self.section("roles")

    def path(self, key: str) -> Path:
        return Path(self.get("paths", key)).expanduser()

    # -- auth -------------------------------------------------------------
    def active_providers(self) -> list[str]:
        """Hvilke innloggingsmetoder som gjelder, i rekkefølge.

        En metode er aktiv når den er nevnt i auth.providers *eller* når den er
        ferdig konfigurert. Den siste halvdelen er det som gjør at
        ANNOUNCER_PIN=1234 alene fortsatt slår på PIN, slik det har virket før,
        og at et JWT-oppsett ikke må nevnes to steder.
        """
        listed = [str(p).strip().lower() for p in self.get("auth", "providers", [])]
        out: list[str] = []
        for name in ("pin", "jwt", "oidc"):
            if name in listed or self.configured(name):
                out.append(name)
        # Rekkefølgen i config vinner for de som er nevnt der.
        out.sort(key=lambda n: listed.index(n) if n in listed else len(listed))
        return out

    def configured(self, provider: str) -> bool:
        s = self.section(f"auth.{provider}")
        if provider == "pin":
            return bool(str(s.get("pin", "")).strip())
        if provider == "jwt":
            return bool(s.get("secret") or s.get("jwks_url"))
        if provider == "oidc":
            return bool(s.get("client_id") and (s.get("issuer") or s.get("authorization_endpoint")))
        return False

    def permissions(self, roles: tuple[str, ...]) -> frozenset[str]:
        """Rollenavn -> handlinger. "*" betyr alt."""
        out: set[str] = set()
        for role in roles:
            for action in self.roles.get(role, []):
                if action == "*":
                    return frozenset(ACTIONS)
                if action in ACTIONS:
                    out.add(action)
        return frozenset(out)


def load(path: str | Path | None = None) -> Config:
    """Leser konfigurasjonen. Manglende config.toml er helt greit."""
    if path is None:
        path = os.environ.get("ANNOUNCER_CONFIG", HERE / "config.toml")
    path = Path(path).expanduser()

    data = copy.deepcopy(DEFAULTS)
    source: Path | None = None
    if path.is_file():
        with path.open("rb") as fh:
            data = _merge(data, _flatten_tables(tomllib.load(fh)))
        source = path

    for env, ((section, key), cast) in ENV.items():
        raw = os.environ.get(env)
        if raw is None or not raw.strip():
            continue
        try:
            data[section][key] = cast(raw.strip())
        except (TypeError, ValueError) as e:
            raise SystemExit(f"{env}: ugyldig verdi {raw!r} ({e})") from e

    return Config(data, source)