"""Auth-gate middleware for the dashboard. Engaged when ``app.state.auth_required is True``. The gate's job: 1. Allow a small set of routes through unauthenticated (login page, ``/auth/*`` OAuth round trip, ``/api/auth/providers``, static assets). 2. For everything else, demand a valid session cookie and attach the verified :class:`Session` to ``request.state.session``. 3. On HTML routes, redirect missing/invalid cookies to ``/login``. On ``/api/*`` routes, return 401 JSON. The middleware is a no-op when ``auth_required`` is False (loopback mode); the legacy ``_SESSION_TOKEN`` ``auth_middleware`` handles those binds. """ from __future__ import annotations import logging from typing import Awaitable, Callable from fastapi import Request from fastapi.responses import JSONResponse, RedirectResponse, Response from hermes_cli.dashboard_auth import list_providers from hermes_cli.dashboard_auth.audit import AuditEvent, audit_log from hermes_cli.dashboard_auth.base import ProviderError from hermes_cli.dashboard_auth.cookies import read_session_cookies from hermes_cli.dashboard_auth.public_paths import PUBLIC_API_PATHS _log = logging.getLogger(__name__) # Prefixes that bypass the auth gate. Match via ``path == prefix`` or # ``path.startswith(prefix)`` — so ``/assets/`` (with trailing slash) # matches ``/assets/foo.css`` but not ``/assetsleak``. Auth-bootstrap # (login page, OAuth round trip, provider listing) and static asset # mounts go here. _GATE_PUBLIC_PREFIXES: tuple[str, ...] = ( "/auth/login", "/auth/callback", "/auth/logout", "/login", "/api/auth/providers", "/assets/", "/favicon.ico", "/ds-assets/", "/fonts/", "/fonts-terminal/", ) def _path_is_public(path: str) -> bool: """True if ``path`` bypasses the OAuth auth gate. Two sources of public-ness: * :data:`PUBLIC_API_PATHS` — the shared ``/api/*`` allowlist that the legacy ``_SESSION_TOKEN`` middleware also honours. Matched exactly (no prefix expansion) so adding ``/api/status`` doesn't accidentally expose ``/api/status/secret-extension``. * :data:`_GATE_PUBLIC_PREFIXES` — auth-bootstrap routes and static mounts. Prefix-matched so ``/assets/foo.css`` lights up via ``/assets/``. """ if path in PUBLIC_API_PATHS: return True return any( path == prefix or path.startswith(prefix) for prefix in _GATE_PUBLIC_PREFIXES ) def _client_ip(request: Request) -> str: fwd = request.headers.get("x-forwarded-for", "") if fwd: return fwd.split(",")[0].strip() return request.client.host if request.client else "" def _unauth_response(request: Request, *, reason: str) -> Response: """API routes → 401 JSON with ``login_url``; HTML routes → 302 → /login. The JSON envelope carries a ``login_url`` field with a ``next=`` query string so the SPA's global 401 handler can drop the user back where they were after re-auth. The contract is intentionally simple so any fetch-wrapper can implement the redirect without parsing details: if response.status === 401 && body.error in ("unauthenticated", "session_expired"): window.location.assign(body.login_url); HTML redirects also carry the ``next=`` query string so direct navigation to ``/sessions`` (etc.) without a cookie comes back to ``/sessions`` after login. Under a reverse proxy with ``X-Forwarded-Prefix: /hermes``, the ``login_url`` is prefixed (``/hermes/login?next=...``) so the browser's window.location.assign / Location: follow lands on the proxied login page rather than the bare ``/login`` (which the proxy doesn't route to the dashboard). """ from hermes_cli.dashboard_auth.prefix import prefix_from_request path = request.url.path next_param = _safe_next_target(request) prefix = prefix_from_request(request) login_url = ( f"{prefix}/login?next={next_param}" if next_param else f"{prefix}/login" ) if path.startswith("/api/"): # API routes never get redirects: the browser fetch() API would # follow a 302 into the cross-origin OAuth dance opaquely. Return # 401 with a structured envelope so the SPA can full-page-navigate # to login_url. error_code = ( "session_expired" if reason == "invalid_or_expired_session" else "unauthenticated" ) return JSONResponse( { "error": error_code, "detail": "Unauthorized", "reason": reason, "login_url": login_url, }, status_code=401, ) return RedirectResponse(url=login_url, status_code=302) def _safe_next_target(request: Request) -> str: """Build the URL-encoded ``next`` query value, or empty string. Only same-origin relative paths are accepted; absolute URLs or ``//evil.com`` open-redirect attempts are silently dropped. The empty string return means the caller produces a bare ``/login`` URL — fine, user lands at the dashboard root after re-auth. """ path = request.url.path # Reject anything that doesn't start with "/" or starts with "//" # (protocol-relative URL — would open-redirect to an attacker host). if not path or not path.startswith("/") or path.startswith("//"): return "" # Don't redirect back to the auth routes themselves — that loops. if any( path == p or path.startswith(p) for p in ("/login", "/auth/", "/api/auth/") ): return "" # Preserve query string if present (e.g. /sessions?page=2). query = request.url.query target = f"{path}?{query}" if query else path # urlencode the whole thing as a single value. from urllib.parse import quote return quote(target, safe="") async def gated_auth_middleware( request: Request, call_next: Callable[[Request], Awaitable[Response]], ) -> Response: """Engaged only when ``app.state.auth_required is True``. No-op pass-through in loopback mode so the legacy auth_middleware can handle those binds via ``_SESSION_TOKEN``. """ if not getattr(request.app.state, "auth_required", False): return await call_next(request) path = request.url.path if _path_is_public(path): return await call_next(request) at, _rt = read_session_cookies(request) if not at: return _unauth_response(request, reason="no_cookie") # Try every registered provider's verify_session in turn. Providers # MUST return None for tokens they don't recognise (not raise). This # lets multiple providers stack — the first one that recognises a # token wins. session = None for provider in list_providers(): try: session = provider.verify_session(access_token=at) except ProviderError as e: _log.warning( "dashboard-auth: provider %r unreachable during verify: %s", provider.name, e, ) audit_log( AuditEvent.SESSION_VERIFY_FAILURE, provider=provider.name, reason="provider_unreachable", ip=_client_ip(request), ) return JSONResponse( {"detail": f"Auth provider {provider.name!r} unreachable"}, status_code=503, ) if session is not None: break if session is None: audit_log( AuditEvent.SESSION_VERIFY_FAILURE, reason="no_provider_recognises", ip=_client_ip(request), ) response = _unauth_response(request, reason="invalid_or_expired_session") # Clear the dead cookie so the browser doesn't keep sending it. # Contract v1: no refresh token to retry with, so the only correct # next step is full re-auth via /login. Importing locally avoids a # cycle with cookies → middleware at module load. Pass the active # prefix so the deletion's Path matches the set-Path (otherwise # the browser ignores it). from hermes_cli.dashboard_auth.cookies import clear_session_cookies from hermes_cli.dashboard_auth.prefix import prefix_from_request clear_session_cookies(response, prefix=prefix_from_request(request)) return response request.state.session = session return await call_next(request)