jmri_server.auth.middleware

  1"""ASGI middleware that enforces auth on every request.
  2
  3Reads a Bearer token from ``Authorization`` first (mobile / JSON
  4clients), then falls back to a signed session cookie (browser).
  5On success populates ``request.state.user`` (:class:`~jmri_server.
  6auth.models.User`) and ``request.state.role`` (:class:`~jmri_server.
  7auth.models.Role`) so downstream routes and the JSON registry can
  8gate on role.
  9
 10The middleware short-circuits when :attr:`ServerContext.auth` is
 11``None`` — the "auth disabled" default. That path sets ``user=None``
 12and ``role=Role.ADMIN`` so the JSON registry's role check is a
 13no-op. Keeps the code path uniform whether auth is on or off; the
 14role gate lives in ONE place (the registry) rather than being
 15duplicated at every check point.
 16
 17**Exempt paths** always bypass the credential check:
 18
 19* ``/login`` and ``/logout`` — the login form itself can't require
 20  auth, obviously.
 21* ``/static/*`` and ``/docs/*`` — public assets. Static docs
 22  shouldn't need a session cookie to render.
 23* ``/json/health`` (if it exists) — for load balancers.
 24
 25The exempt list is regex-matched against the raw path so a
 26misconfigured route registrant can't accidentally require auth on
 27``/login``.
 28
 29**Failure responses**:
 30
 31* JSON API (``/json/`` or ``Content-Type: application/json``) →
 32  ``401`` with a JMRI-shaped error body.
 33* Browser (``Accept: text/html``) → 302 to ``/login`` with a
 34  ``next=`` query so post-login redirect works.
 35* Everything else → plain ``401``.
 36"""
 37from __future__ import annotations
 38
 39import logging
 40import re
 41from typing import TYPE_CHECKING
 42
 43from starlette.middleware.base import BaseHTTPMiddleware
 44from starlette.requests import Request
 45from starlette.responses import JSONResponse, RedirectResponse, Response
 46
 47from jmri_server.auth.models import Role
 48
 49if TYPE_CHECKING:
 50    from jmri_server.context import ServerContext
 51
 52_log = logging.getLogger(__name__)
 53
 54# Cookie name for the signed session. Prefixed with ``__Host-`` when
 55# HTTPS becomes the norm — for the current LAN-only deployment,
 56# ``JMRISESSION`` (no prefix, works over plain HTTP) suffices.
 57SESSION_COOKIE = "JMRISESSION"
 58
 59
 60# Paths that never require auth. Compiled once at module load.
 61_EXEMPT_PATTERNS = [
 62    re.compile(r"^/login/?$"),
 63    re.compile(r"^/logout/?$"),
 64    re.compile(r"^/static/"),
 65    re.compile(r"^/docs/"),
 66    # The FastAPI /docs and /redoc (OpenAPI UI) — same URL but
 67    # different meaning; harmless to exempt for beta.
 68    re.compile(r"^/openapi\.json$"),
 69    # WebSocket handshake carries auth via query string or first
 70    # message; the middleware can't intercept that easily, so we
 71    # exempt the path itself and check inside the WS handler.
 72    re.compile(r"^/ws/"),
 73]
 74
 75
 76def _is_exempt(path: str) -> bool:
 77    return any(p.match(path) for p in _EXEMPT_PATTERNS)
 78
 79
 80class AuthMiddleware(BaseHTTPMiddleware):
 81    """Populate ``request.state.user`` / ``request.state.role`` from
 82    the request's credentials."""
 83
 84    def __init__(self, app, ctx: ServerContext) -> None:
 85        super().__init__(app)
 86        self._ctx = ctx
 87
 88    async def dispatch(self, request: Request, call_next):
 89        request.state.user = None
 90        request.state.role = Role.ADMIN  # default when auth is off
 91
 92        auth_mgr = getattr(self._ctx, "auth", None)
 93        if auth_mgr is None:
 94            # Auth disabled — every request runs as admin. Downstream
 95            # role check is a no-op.
 96            return await call_next(request)
 97
 98        # Auth enabled. Try bearer, then cookie.
 99        user = _try_bearer(request, auth_mgr)
100        if user is None:
101            user = _try_cookie(request, auth_mgr)
102
103        if user is None:
104            if _is_exempt(request.url.path):
105                # Exempt path with no valid credentials — anonymous
106                # viewer for the response, since nothing here should
107                # actually depend on auth state.
108                request.state.role = Role.VIEWER
109                return await call_next(request)
110            return _unauthorized(request)
111
112        request.state.user = user
113        request.state.role = user.role
114        return await call_next(request)
115
116
117def _try_bearer(request: Request, auth_mgr) -> object | None:
118    header = request.headers.get("Authorization", "")
119    if not header.lower().startswith("bearer "):
120        return None
121    token = header[7:].strip()
122    result = auth_mgr.login_token(token)
123    return result.user if result else None
124
125
126def _try_cookie(request: Request, auth_mgr) -> object | None:
127    cookie = request.cookies.get(SESSION_COOKIE, "")
128    if not cookie:
129        return None
130    return auth_mgr.verify_session_cookie(cookie)
131
132
133def _unauthorized(request: Request) -> Response:
134    path = request.url.path
135    accept = request.headers.get("Accept", "")
136    # JSON endpoints or JSON-only clients get a JSON 401 —
137    # otherwise browser clients get redirected to /login with a
138    # next= hint so they end up back where they were.
139    if (
140        path.startswith("/json")
141        or "application/json" in request.headers.get("Content-Type", "")
142        or "application/json" in accept
143    ):
144        return JSONResponse(
145            status_code=401,
146            content={
147                "type": "error",
148                "data": {
149                    "code": 401,
150                    "message": "authentication required",
151                },
152            },
153        )
154    if "text/html" in accept:
155        redirect_target = f"/login?next={path}"
156        return RedirectResponse(url=redirect_target, status_code=302)
157    return Response(status_code=401, content="authentication required")