Reisacher-Forderung/app/auth.py

import logging
import secrets
from urllib.parse import parse_qs, quote

from fastapi import HTTPException, Request, status
from fastapi.responses import JSONResponse, RedirectResponse, Response
from starlette.middleware.base import BaseHTTPMiddleware

from .config import settings
from .db import SessionLocal
from .ldap import DirectoryUnavailable, lookup_directory_user
from .navigation import normalize_page_url
from .oauth_token import (
    InvalidSessionToken,
    create_access_token,
    decode_access_token,
    renew_access_token,
    token_needs_renewal,
)
from .permissions import Principal, highest_role, load_entitlements

logger = logging.getLogger(__name__)

PUBLIC_PREFIXES = ("/static/",)
PUBLIC_PATHS = {"/login", "/health"}
UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}


def set_session_cookie(response, token: str) -> None:
    response.set_cookie(
        settings.SESSION_COOKIE_NAME,
        token,
        secure=settings.COOKIE_SECURE,
        httponly=True,
        samesite="lax",
        path="/",
    )


def clear_session_cookie(response) -> None:
    response.delete_cookie(
        settings.SESSION_COOKIE_NAME,
        secure=settings.COOKIE_SECURE,
        httponly=True,
        samesite="lax",
        path="/",
    )


def safe_next_url(value: str | None, default: str = "/select") -> str:
    return normalize_page_url(value, default)


def _principal(payload: dict, entitlements) -> Principal:
    return Principal(
        username=payload["sub"],
        name=payload["name"],
        email=payload["email"],
        role=highest_role(entitlements),
        auth_source=payload["auth_source"],
        csrf=payload["csrf"],
        entitlements=entitlements,
    )


async def _submitted_csrf(request: Request) -> str:
    header = request.headers.get("X-CSRF-Token", "")
    if header:
        return header
    content_type = request.headers.get("content-type", "")
    if content_type.startswith("application/x-www-form-urlencoded"):
        body = await request.body()
        return parse_qs(body.decode("utf-8", errors="replace")).get("_csrf", [""])[0]
    return ""


class AuthenticationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        is_public = request.url.path in PUBLIC_PATHS or request.url.path.startswith(PUBLIC_PREFIXES)
        token = request.cookies.get(settings.SESSION_COOKIE_NAME)
        payload = None
        new_token = None

        if token:
            try:
                payload = decode_access_token(token)
            except InvalidSessionToken:
                payload = None

        proxy_user = request.headers.get(settings.AUTH_PROXY_HEADER)
        proxy_host = request.client.host if request.client else None
        trusted_proxy_identity = (
            proxy_user if not settings.debug_auth_enabled and settings.is_trusted_proxy(proxy_host) else None
        )
        if proxy_user and not trusted_proxy_identity and not settings.debug_auth_enabled:
            logger.warning("Nicht vertrauenswürdiger Proxy-Identitätsheader von %s ignoriert", proxy_host)

        try:
            with SessionLocal() as db:
                if payload:
                    entitlements = load_entitlements(db, payload["sub"])
                    principal = _principal(payload, entitlements)
                    request.state.principal = principal
                    if token_needs_renewal(payload, principal.role):
                        new_token = renew_access_token(payload, principal.role)
                elif trusted_proxy_identity:
                    directory_user = lookup_directory_user(trusted_proxy_identity)
                    if not directory_user or directory_user.disabled:
                        return JSONResponse({"detail": "Windows-Benutzer ist nicht gültig"}, status_code=401)
                    entitlements = load_entitlements(db, directory_user.username)
                    role = highest_role(entitlements)
                    new_token = create_access_token(
                        username=directory_user.username,
                        name=directory_user.name,
                        email=directory_user.email,
                        role=role,
                        auth_source="iis",
                    )
                    payload = decode_access_token(new_token)
                    request.state.principal = _principal(payload, entitlements)
        except HTTPException as exc:
            return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)
        except ValueError:
            return JSONResponse({"detail": "Windows-Benutzer ist nicht gültig"}, status_code=401)
        except DirectoryUnavailable:
            logger.exception("IIS-Benutzer konnte nicht im AD aufgelöst werden")
            return JSONResponse({"detail": "Anmeldedienst ist derzeit nicht verfügbar"}, status_code=503)
        except Exception:
            logger.exception("Berechtigungen konnten nicht geladen werden")
            return JSONResponse({"detail": "Berechtigungssystem ist derzeit nicht verfügbar"}, status_code=503)

        principal = getattr(request.state, "principal", None)
        if not principal and not is_public:
            logger.info(
                "Anmeldung erforderlich",
                extra={
                    "event": "auth_login_required",
                    "reason": "invalid_or_expired_session" if token else "missing_session",
                },
            )
            next_url = request.url.path
            if request.url.query:
                next_url += "?" + request.url.query
            next_url = normalize_page_url(next_url, current_url=request.headers.get("HX-Current-URL"))
            login_url = f"/login?next={quote(next_url, safe='')}"
            if request.headers.get("HX-Request", "").lower() == "true":
                response = Response(status_code=200, headers={"HX-Redirect": login_url})
            else:
                response = RedirectResponse(login_url, status_code=303)
            if token:
                clear_session_cookie(response)
            return response

        if principal and request.method in UNSAFE_METHODS and request.url.path != "/login":
            submitted = await _submitted_csrf(request)
            if not submitted or not secrets.compare_digest(submitted, principal.csrf):
                return JSONResponse({"detail": "Ungültiger CSRF-Token"}, status_code=status.HTTP_403_FORBIDDEN)

        response = await call_next(request)
        if new_token and not getattr(request.state, "suppress_session_refresh", False):
            set_session_cookie(response, new_token)
        return response