announcer/auth.py

#!/usr/bin/env python3
"""Pluggbar autentisering for announcer.

Hele modellen er én datatype og én funksjon:

    Principal          – hvem dette er, hvilke roller, hva rollene får gjøre
    require(action)    – hent den, eller stopp forespørselen med 401/403

Rundt det ligger leverandører som prøves i rekkefølge. Hver av dem svarer på
én forespørsel med «dette er brukeren», «jeg fant ingen legitimasjon av min
type» (None) eller AuthError («fant noe, men det var ugyldig»):

    pin   – X-Pin-header, én delt kode. Virker uten nett, derfor beholdt.
    jwt   – Authorization: Bearer <jwt> fra crew-panelet. HS256 med delt
            hemmelighet, eller RS256/ES256 mot panelets JWKS.
    oidc  – autorisasjonskode med PKCE mot crew-panelet; nettleseren logger
            inn der og får en sesjonscookie her.

Er ingen leverandør konfigurert, er alt åpent – nøyaktig som før. Det er med
vilje: tjenesten skal kunne settes opp på en kveld uten identitetsleverandør.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import logging
import json
import secrets
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from flask import Blueprint, abort, g, redirect, request, session, url_for

from config import ACTIONS, Config

log = logging.getLogger("announcer.auth")

SESSION_KEY = "principal"
NAME_CLAIMS = ("name", "preferred_username", "nickname", "given_name", "email")


class AuthError(Exception):
    """Legitimasjon fulgte med, men den holdt ikke."""


# --------------------------------------------------------------------------
# Principal
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class Principal:
    subject: str                  # stabil id: "pin", "jwt:4f2a", "oidc:kent"
    name: str                     # vises i køen og i loggen
    roles: tuple[str, ...]
    method: str                   # none | anonymous | pin | jwt | oidc
    permissions: frozenset[str]

    @property
    def anonymous(self) -> bool:
        return self.method in ("none", "anonymous")

    def can(self, action: str) -> bool:
        return action in self.permissions

    def public(self) -> dict[str, Any]:
        return {
            "subject": self.subject,
            "name": self.name,
            "roles": list(self.roles),
            "method": self.method,
            "can": sorted(self.permissions),
        }


def _claim(claims: dict, path: str) -> Any:
    """Henter et claim, også nestet: "realm_access.roles"."""
    cur: Any = claims
    for part in path.split("."):
        if not isinstance(cur, dict):
            return None
        cur = cur.get(part)
    return cur


def _roles(value: Any) -> tuple[str, ...]:
    """Roller kommer som liste, streng eller mellomromsdelt streng."""
    if isinstance(value, str):
        return tuple(v for v in value.replace(",", " ").split() if v)
    if isinstance(value, (list, tuple, set)):
        return tuple(str(v).strip() for v in value if str(v).strip())
    return ()


def _display_name(claims: dict, name_claim: str) -> str:
    if name_claim:
        value = _claim(claims, name_claim)
        if value:
            return str(value)
    for key in NAME_CLAIMS:
        if claims.get(key):
            return str(claims[key])
    return str(claims.get("sub") or "ukjent")


# --------------------------------------------------------------------------
# JWT-verifisering, delt av jwt- og oidc-leverandøren
# --------------------------------------------------------------------------


class TokenVerifier:
    """Verifiserer signatur, iss, aud og exp. JWKS caches av PyJWT selv."""

    def __init__(self, section: dict[str, Any]) -> None:
        self.issuer = str(section.get("issuer", "")).strip()
        self.audience = str(section.get("audience", "")).strip()
        self.secret = str(section.get("secret", "")).strip()
        self.jwks_url = str(section.get("jwks_url", "")).strip()
        self.leeway = int(section.get("leeway", 30))
        # Tom liste: utled av hvordan nøkkelen kommer. Delt hemmelighet er
        # symmetrisk (HS256), JWKS er asymmetrisk (RS256/ES256).
        algs = [str(a) for a in section.get("algorithms") or []]
        self.algorithms = algs or (["RS256", "ES256"] if self.jwks_url else ["HS256"])
        self._jwks: Any = None

    def _signing_key(self, token: str) -> Any:
        if not self.jwks_url:
            return self.secret
        if self._jwks is None:
            import jwt  # lazy: PIN-oppsett skal ikke trenge pyjwt

            self._jwks = jwt.PyJWKClient(self.jwks_url, cache_keys=True, lifespan=3600)
        return self._jwks.get_signing_key_from_jwt(token).key

    def verify(self, token: str) -> dict:
        import jwt

        if not (self.secret or self.jwks_url):
            raise AuthError("JWT er ikke konfigurert")
        try:
            return jwt.decode(
                token,
                self._signing_key(token),
                algorithms=self.algorithms,
                audience=self.audience or None,
                issuer=self.issuer or None,
                leeway=self.leeway,
                options={
                    "verify_aud": bool(self.audience),
                    "verify_iss": bool(self.issuer),
                    "require": ["exp"],
                },
            )
        except jwt.ExpiredSignatureError as e:
            raise AuthError("Tokenet er utløpt") from e
        except jwt.InvalidAudienceError as e:
            raise AuthError("Tokenet gjelder en annen tjeneste") from e
        except jwt.InvalidTokenError as e:
            raise AuthError(f"Ugyldig token ({e})") from e
        except urllib.error.URLError as e:
            raise AuthError(f"Fikk ikke hentet nøkler fra {self.jwks_url} ({e})") from e


# --------------------------------------------------------------------------
# Leverandører
# --------------------------------------------------------------------------


class Provider:
    name = ""

    def authenticate(self, req) -> Principal | None:  # noqa: ANN001
        return None

    def describe(self) -> dict[str, Any]:
        return {"method": self.name}


class PinProvider(Provider):
    name = "pin"

    def __init__(self, cfg: Config) -> None:
        s = cfg.section("auth.pin")
        self.pin = str(s.get("pin", "")).strip()
        self.role = str(s.get("role") or "admin")
        self.label = str(s.get("label") or "PIN")
        self.hint = str(s.get("hint", ""))
        self.cfg = cfg

    def authenticate(self, req) -> Principal | None:  # noqa: ANN001
        given = req.headers.get("X-Pin", "").strip()
        if not given:
            auth = req.headers.get("Authorization", "")
            if auth[:4].lower() == "pin ":
                given = auth[4:].strip()
        if not given:
            return None
        # compare_digest, ikke ==, så svartiden ikke lekker hvor langt man kom.
        if not hmac.compare_digest(given, self.pin):
            raise AuthError("Feil PIN")
        roles = (self.role,)
        # En delt kode peker ikke på en person, og skal ikke late som. Vil man
        # vite hvem som sendte hva, er det JWT eller OIDC man vil ha.
        return Principal("pin", "delt PIN", roles, "pin", self.cfg.permissions(roles))

    def describe(self) -> dict[str, Any]:
        return {"method": "pin", "label": self.label, "hint": self.hint}


class MeshProvider(Provider):
    """Sentralen som leverer en WAV til denne sonen.

    Egen header, ikke Authorization: Bearer – JwtProvider ville avvist en
    nøkkel som ikke er et JWT før neste leverandør fikk prøve.
    """

    name = "mesh"

    def __init__(self, cfg: Config) -> None:
        self.secret = str(cfg.get("mesh", "secret", "")).strip()
        self.cfg = cfg

    def authenticate(self, req) -> Principal | None:  # noqa: ANN001
        given = req.headers.get("X-Mesh-Key", "").strip()
        if not given:
            return None
        if not hmac.compare_digest(given, self.secret):
            raise AuthError("Feil mesh-nøkkel")
        roles = ("mesh",)
        return Principal("mesh", "sentral", roles, "mesh", self.cfg.permissions(roles))


class JwtProvider(Provider):
    name = "jwt"

    def __init__(self, cfg: Config) -> None:
        s = cfg.section("auth.jwt")
        self.cfg = cfg
        self.verifier = TokenVerifier(s)
        self.role_claim = str(s.get("role_claim") or "roles")
        self.name_claim = str(s.get("name_claim", ""))
        self.default_role = str(s.get("default_role", ""))
        self.session_seconds = int(s.get("session_seconds", 43200))

    def principal(self, claims: dict, method: str = "jwt") -> Principal:
        roles = _roles(_claim(claims, self.role_claim))
        if not roles and self.default_role:
            roles = (self.default_role,)
        sub = str(claims.get("sub") or claims.get("email") or "ukjent")
        return Principal(
            f"{method}:{sub}",
            _display_name(claims, self.name_claim),
            roles,
            method,
            self.cfg.permissions(roles),
        )

    def authenticate(self, req) -> Principal | None:  # noqa: ANN001
        auth = req.headers.get("Authorization", "")
        if auth[:7].lower() != "bearer ":
            return None
        token = auth[7:].strip()
        if not token:
            return None
        return self.principal(self.verifier.verify(token))


class OidcProvider(Provider):
    """Autorisasjonskode med PKCE. Innloggingen skjer i nettleseren, så
    authenticate() finner aldri noe i en header – sesjonen tar seg av resten."""

    name = "oidc"

    def __init__(self, cfg: Config) -> None:
        s = cfg.section("auth.oidc")
        self.cfg = cfg
        self.issuer = str(s.get("issuer", "")).strip().rstrip("/")
        self.client_id = str(s.get("client_id", "")).strip()
        self.client_secret = str(s.get("client_secret", "")).strip()
        self.scopes = [str(x) for x in s.get("scopes") or ["openid"]]
        self.label = str(s.get("label") or "Logg inn")
        self.role_claim = str(s.get("role_claim") or "roles")
        self.name_claim = str(s.get("name_claim", ""))
        self.default_role = str(s.get("default_role", ""))
        self._manual = {
            "authorization_endpoint": str(s.get("authorization_endpoint", "")).strip(),
            "token_endpoint": str(s.get("token_endpoint", "")).strip(),
            "jwks_uri": str(s.get("jwks_url", "")).strip(),
            "end_session_endpoint": str(s.get("end_session_endpoint", "")).strip(),
        }
        self._meta: dict[str, Any] | None = None
        self._verifier: TokenVerifier | None = None

    # -- oppdagelse -------------------------------------------------------
    def metadata(self) -> dict[str, Any]:
        """Discovery hentes først ved første innlogging, aldri ved oppstart –
        tjenesten skal starte selv om crew-panelet er nede."""
        if self._meta is not None:
            return self._meta
        meta = {k: v for k, v in self._manual.items() if v}
        if not (meta.get("authorization_endpoint") and meta.get("token_endpoint")):
            url = f"{self.issuer}/.well-known/openid-configuration"
            try:
                fetched = _get_json(url)
            except Exception as e:  # noqa: BLE001
                raise AuthError(f"Fikk ikke hentet OIDC-oppsett fra {url} ({e})") from e
            meta = {**fetched, **meta}
        self._meta = meta
        return meta

    def verifier(self) -> TokenVerifier:
        if self._verifier is None:
            meta = self.metadata()
            self._verifier = TokenVerifier(
                {
                    "issuer": meta.get("issuer") or self.issuer,
                    "audience": self.client_id,
                    "secret": self.client_secret,
                    "jwks_url": meta.get("jwks_uri", ""),
                    "algorithms": [],
                }
            )
        return self._verifier

    # -- flyten -----------------------------------------------------------
    def redirect_uri(self) -> str:
        base = str(self.cfg.get("server", "public_url", "")).strip().rstrip("/")
        if base:
            return f"{base}/auth/callback"
        return url_for("auth.callback", _external=True)

    def authorize_url(self) -> str:
        verifier = secrets.token_urlsafe(64)
        state = secrets.token_urlsafe(24)
        challenge = (
            base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
            .rstrip(b"=")
            .decode()
        )
        session["oidc_state"] = state
        session["oidc_verifier"] = verifier
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri(),
            "scope": " ".join(self.scopes),
            "state": state,
            "code_challenge": challenge,
            "code_challenge_method": "S256",
        }
        endpoint = self.metadata()["authorization_endpoint"]
        sep = "&" if "?" in endpoint else "?"
        return f"{endpoint}{sep}{urllib.parse.urlencode(params)}"

    def exchange(self, code: str, verifier: str) -> dict:
        data = {
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": self.redirect_uri(),
            "client_id": self.client_id,
            "code_verifier": verifier,
        }
        headers = {}
        if self.client_secret:
            basic = base64.b64encode(
                f"{self.client_id}:{self.client_secret}".encode()
            ).decode()
            headers["Authorization"] = f"Basic {basic}"
        try:
            return _post_json(self.metadata()["token_endpoint"], data, headers)
        except Exception as e:  # noqa: BLE001
            raise AuthError(f"Kodebyttet feilet ({e})") from e

    def principal(self, claims: dict) -> Principal:
        roles = _roles(_claim(claims, self.role_claim))
        if not roles and self.default_role:
            roles = (self.default_role,)
        sub = str(claims.get("sub") or claims.get("email") or "ukjent")
        return Principal(
            f"oidc:{sub}",
            _display_name(claims, self.name_claim),
            roles,
            "oidc",
            self.cfg.permissions(roles),
        )

    def describe(self) -> dict[str, Any]:
        return {"method": "oidc", "label": self.label, "login_url": "/auth/login"}


PROVIDERS = {"mesh": MeshProvider, "pin": PinProvider, "jwt": JwtProvider, "oidc": OidcProvider}


# --------------------------------------------------------------------------
# HTTP-hjelpere (stdlib, så vi slipper en ekstra avhengighet)
# --------------------------------------------------------------------------


def _get_json(url: str, timeout: int = 10) -> dict:
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:  # noqa: S310
        return json.loads(r.read().decode())


def _post_json(url: str, form: dict, headers: dict | None = None, timeout: int = 10) -> dict:
    body = urllib.parse.urlencode(form).encode()
    req = urllib.request.Request(
        url,
        data=body,
        headers={
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded",
            **(headers or {}),
        },
    )
    with urllib.request.urlopen(req, timeout=timeout) as r:  # noqa: S310
        return json.loads(r.read().decode())


# --------------------------------------------------------------------------
# Authenticator
# --------------------------------------------------------------------------


class Authenticator:
    def __init__(self, cfg: Config) -> None:
        self.cfg = cfg
        self.providers: list[Provider] = [
            PROVIDERS[name](cfg) for name in cfg.active_providers()
        ]
        # mesh er maskin-til-maskin og teller ikke som innlogging for folk:
        # en sone som bare har mesh-nøkkel skal fortsatt ha åpen side lokalt.
        self.human: list[Provider] = [p for p in self.providers if p.name != "mesh"]
        self.enabled = bool(self.human)

    def provider(self, name: str) -> Provider | None:
        return next((p for p in self.providers if p.name == name), None)

    # -- anonym -----------------------------------------------------------
    def anonymous(self) -> Principal:
        """Uten konfigurerte leverandører får anonyme alt – dagens oppførsel.

        Med leverandører får de det [roles].anonymous sier, som er ingenting
        som standard. Sett den til ["view"] om statusskjermen skal henge på
        veggen uten innlogging."""
        if not self.enabled:
            # Tomt navn: da står køen og loggen som før, uten en «bruker» som
            # i praksis bare betyr «hvem som helst». Og aldri «play»: den er
            # for en sentral med mesh-nøkkel, ikke for hvem som helst som
            # finner en åpen sone og vil spille lyd på den.
            return Principal("anonymous", "", ("anonymous",), "none",
                             frozenset(ACTIONS) - {"play", "register"})
        roles = ("anonymous",)
        return Principal("anonymous", "ikke innlogget", roles, "anonymous", self.cfg.permissions(roles))

    # -- sesjon -----------------------------------------------------------
    def from_session(self) -> Principal | None:
        raw = session.get(SESSION_KEY)
        if not isinstance(raw, dict):
            return None
        if raw.get("exp") and float(raw["exp"]) < time.time():
            session.pop(SESSION_KEY, None)
            return None
        roles = tuple(str(r) for r in raw.get("roles", ()))
        return Principal(
            str(raw.get("subject", "?")),
            str(raw.get("name", "?")),
            roles,
            str(raw.get("method", "jwt")),
            # Rettigheter slås opp på nytt hver gang, så en endring i
            # [roles] slår inn uten at folk må logge inn igjen.
            self.cfg.permissions(roles),
        )

    def remember(self, p: Principal, seconds: int) -> None:
        session.permanent = True
        session[SESSION_KEY] = {
            "subject": p.subject,
            "name": p.name,
            "roles": list(p.roles),
            "method": p.method,
            "exp": time.time() + seconds,
        }

    # -- oppslag ----------------------------------------------------------
    def resolve(self) -> tuple[Principal, str]:
        """Returnerer (principal, feilmelding). Kaster ikke – /api/config skal
        kunne spørre «hvem er jeg?» uten å få 401 i fleisen."""
        p = self.from_session()
        if p is not None:
            return p, ""
        for provider in self.providers:
            try:
                p = provider.authenticate(request)
            except AuthError as e:
                return self.anonymous(), str(e)
            if p is not None:
                return p, ""
        return self.anonymous(), ""

    def describe(self) -> dict[str, Any]:
        return {
            "enabled": self.enabled,
            "methods": [p.describe() for p in self.human],
            "logout_url": "/auth/logout" if self.enabled else None,
        }


_auth: Authenticator | None = None


def current() -> Principal:
    """Principalen for denne forespørselen, slått opp én gang."""
    if not hasattr(g, "principal"):
        assert _auth is not None, "auth.install() er ikke kalt"
        g.principal, g.auth_error = _auth.resolve()
    return g.principal


def error() -> str:
    current()
    return getattr(g, "auth_error", "")


def require(action: str) -> Principal:
    """Enten har denne brukeren lov, eller så stopper forespørselen her.

    401 når vi ikke vet hvem det er (klienten skal logge inn), 403 når vi vet
    hvem det er men rollen mangler handlingen.
    """
    p = current()
    if error():
        # Noen viste fram en PIN, et token eller en mesh-nøkkel som ikke
        # holdt. Det er en feil, ikke en anonym bruker – også på en åpen
        # sone, der anonym ellers hadde fått lov.
        abort(401, error())
    if p.can(action):
        return p
    if p.anonymous:
        abort(401, _login_hint())
    abort(403, f"Rollen din ({', '.join(p.roles) or 'ingen'}) har ikke lov til dette.")


def _login_hint() -> str:
    assert _auth is not None
    names = {p.name for p in _auth.human}
    if names == {"pin"}:
        return "Feil eller manglende PIN."
    if "oidc" in names:
        return "Du må logge inn."
    return "Innlogging kreves."


# --------------------------------------------------------------------------
# Oppsett og ruter
# --------------------------------------------------------------------------


def secret_key(cfg: Config) -> bytes:
    """Fast nøkkel, ellers logges alle ut ved hver restart av tjenesten."""
    configured = str(cfg.get("server", "secret_key", "")).strip()
    if configured:
        return configured.encode()
    path = cfg.path("state") / "secret_key"
    if path.is_file():
        return path.read_bytes()
    path.parent.mkdir(parents=True, exist_ok=True)
    key = secrets.token_bytes(32)
    path.write_bytes(key)
    path.chmod(0o600)
    log.info("lagde ny sesjonsnøkkel i %s", path)
    return key


bp = Blueprint("auth", __name__)


@bp.get("/auth/login")
def login():
    oidc = _auth.provider("oidc") if _auth else None
    if not isinstance(oidc, OidcProvider):
        abort(404, "OIDC er ikke satt opp")
    try:
        return redirect(oidc.authorize_url())
    except AuthError as e:
        abort(502, str(e))


@bp.get("/auth/callback")
def callback():
    oidc = _auth.provider("oidc") if _auth else None
    if not isinstance(oidc, OidcProvider):
        abort(404, "OIDC er ikke satt opp")
    if err := request.args.get("error"):
        abort(401, f"{err}: {request.args.get('error_description', '')}".strip(": "))
    state = session.pop("oidc_state", None)
    verifier = session.pop("oidc_verifier", None)
    if not state or not verifier or request.args.get("state") != state:
        abort(400, "Innloggingen kom ut av takt. Prøv på nytt.")
    code = request.args.get("code", "")
    if not code:
        abort(400, "Ingen kode fra identitetsleverandøren")
    try:
        tokens = oidc.exchange(code, verifier)
        id_token = tokens.get("id_token")
        if not id_token:
            abort(502, "Fikk ikke id_token fra identitetsleverandøren")
        claims = oidc.verifier().verify(id_token)
    except AuthError as e:
        abort(401, str(e))
    p = oidc.principal(claims)
    ttl = int(claims.get("exp", 0) - time.time()) if claims.get("exp") else 0
    _auth.remember(p, max(ttl, 3600))
    log.info("innlogget %s (%s) via oidc", p.name, p.subject)
    return redirect("/")


@bp.get("/auth/token")
def token_login():
    """Bro mens crew-panelet mangler OIDC: panelet lenker hit med et kortlevd
    JWT, vi verifiserer det og setter sesjonscookien. Ett klikk, ingen PIN."""
    provider = _auth.provider("jwt") if _auth else None
    if not isinstance(provider, JwtProvider):
        abort(404, "JWT er ikke satt opp")
    raw = request.args.get("t") or request.args.get("token") or ""
    if not raw.strip():
        abort(400, "Mangler token")
    try:
        claims = provider.verifier.verify(raw.strip())
    except AuthError as e:
        abort(401, str(e))
    p = provider.principal(claims)
    ttl = int(claims.get("exp", 0) - time.time()) if claims.get("exp") else 0
    _auth.remember(p, max(ttl, provider.session_seconds))
    log.info("innlogget %s (%s) via token-lenke", p.name, p.subject)
    return redirect("/")


@bp.get("/auth/logout")
def logout():
    session.clear()
    oidc = _auth.provider("oidc") if _auth else None
    if isinstance(oidc, OidcProvider) and oidc._meta:
        if end := oidc._meta.get("end_session_endpoint"):
            return redirect(end)
    return redirect("/")


def reconfigure(cfg: Config) -> Authenticator:
    """Bytter ut leverandørene uten å røre Flask-appen. For tester som vil
    prøve samme app med og uten innlogging."""
    global _auth
    _auth = Authenticator(cfg)
    return _auth


def install(app, cfg: Config) -> Authenticator:  # noqa: ANN001
    """Kobler autentisering på Flask-appen. Returnerer Authenticatoren."""
    global _auth
    _auth = Authenticator(cfg)
    app.secret_key = secret_key(cfg)
    app.config.update(
        SESSION_COOKIE_HTTPONLY=True,
        SESSION_COOKIE_SAMESITE="Lax",
        # Secure bare bak TLS: på rå HTTP over LAN/NetBird ville cookien
        # aldri blitt sendt, og ingen ville klart å logge inn.
        SESSION_COOKIE_SECURE=bool(cfg.get("server", "behind_tls", False)),
    )
    app.register_blueprint(bp)
    return _auth