corvix.web.middleware ===================== .. py:module:: corvix.web.middleware .. autoapi-nested-parse:: Token-based authentication middleware for the Corvix web app. When ``CORVIX_SECRET_TOKEN`` (or ``CORVIX_SECRET_TOKEN_FILE``) is set: * ``/api/*`` routes (except ``/api/health``) require an ``Authorization: Bearer `` or ``X-Corvix-Token: `` header, **or** a valid ``corvix_session`` cookie (so the browser SPA works after logging in via the web UI without needing to inject headers). * UI routes (``/``, ``/dashboards/*``) require a ``corvix_session`` cookie; requests without a valid cookie are redirected to ``/login``. * ``/api/health``, ``/assets/*``, ``/login``, and ``/logout`` are always public. When the environment variable is *not* set, the middleware is a no-op and the app behaves exactly as before (backward compatible). HTTPS detection --------------- The ``corvix_session`` cookie is marked ``Secure`` only when the request arrives over HTTPS. The scheme is read from ``request.url.scheme``, which reflects the real protocol when uvicorn is started with ``--proxy-headers`` (trusting ``X-Forwarded-Proto`` from a reverse proxy). Without that flag, direct HTTPS connections are still detected correctly via the connection scheme. Do **not** rely on raw ``X-Forwarded-Proto`` header inspection from application code: untrusted clients can spoof it. Attributes ---------- .. autoapisummary:: corvix.web.middleware.logger corvix.web.middleware._SESSION_COOKIE_NAME corvix.web.middleware.SESSION_MAX_AGE_SECONDS corvix.web.middleware._PUBLIC_EXACT corvix.web.middleware._PUBLIC_PREFIXES corvix.web.middleware._MISCONFIGURED corvix.web.middleware._SECRET_CACHE corvix.web.middleware._SECRET_CACHE_TTL Classes ------- .. autoapisummary:: corvix.web.middleware.TokenAuthMiddleware Functions --------- .. autoapisummary:: corvix.web.middleware._make_session_cookie corvix.web.middleware._verify_session_cookie corvix.web.middleware._parse_cookies corvix.web.middleware._get_secret corvix.web.middleware._is_public corvix.web.middleware._send_json_401 corvix.web.middleware._send_redirect corvix.web.middleware._parse_request_headers corvix.web.middleware._check_api_auth corvix.web.middleware._check_ui_auth Module Contents --------------- .. py:data:: logger .. py:data:: _SESSION_COOKIE_NAME :value: 'corvix_session' .. py:data:: SESSION_MAX_AGE_SECONDS :type: int :value: 86400 .. py:data:: _PUBLIC_EXACT :type: frozenset[str] .. py:data:: _PUBLIC_PREFIXES :type: tuple[str, Ellipsis] :value: ('/assets/',) .. py:function:: _make_session_cookie(secret: str) -> str Return a signed, time-limited session cookie value. Format: ``{expiry_unix_timestamp}:{hmac_sha256_hex}`` The HMAC covers both the fixed context string and the expiry timestamp so that the expiry cannot be extended without knowing the secret. .. py:function:: _verify_session_cookie(secret: str, value: str) -> bool Return True when *value* is a valid, unexpired session token. Validates the HMAC signature and rejects tokens whose expiry timestamp is in the past. Uses :func:`hmac.compare_digest` to prevent timing attacks. .. py:function:: _parse_cookies(cookie_header: str) -> dict[str, str] Parse a raw ``Cookie`` header value into a name→value dict. Uses :class:`http.cookies.SimpleCookie` from the standard library to handle quoted values and other edge cases correctly. .. py:data:: _MISCONFIGURED :type: bool :value: False .. py:data:: _SECRET_CACHE :type: tuple[float, str] | None :value: None .. py:data:: _SECRET_CACHE_TTL :type: float :value: 60.0 .. py:function:: _get_secret() -> str Return the configured secret token, or an empty string if not set. Results are cached for ``_SECRET_CACHE_TTL`` seconds so that deployments using ``CORVIX_SECRET_TOKEN_FILE`` do not incur a synchronous disk read on every request. A configuration change takes effect within one TTL window. When both ``CORVIX_SECRET_TOKEN`` and ``CORVIX_SECRET_TOKEN_FILE`` are set simultaneously, logs a warning *once* per process and returns an empty string (auth disabled) to avoid flooding logs under load. .. py:function:: _is_public(path: str) -> bool Return True when *path* should never require authentication. .. py:function:: _send_json_401(send: litestar.types.asgi_types.Send) -> None :async: Send a minimal JSON 401 Unauthorized response via ASGI. .. py:function:: _send_redirect(send: litestar.types.asgi_types.Send, location: bytes) -> None :async: Send a 302 redirect response via ASGI. .. py:function:: _parse_request_headers(scope: litestar.types.asgi_types.Scope) -> dict[bytes, bytes] Extract and normalise HTTP headers from an ASGI scope. RFC 7230 §3.2.4: header field values are ISO-8859-1 (latin-1); using latin-1 decoding (rather than strict UTF-8) means malformed byte sequences never raise :exc:`UnicodeDecodeError` and produce a clean 401/redirect instead of a 500. RFC 6265 §5.4: a request may carry multiple ``Cookie`` headers; they must be treated as if joined by ``"; "``. A plain dict comprehension would silently drop all but the last occurrence, so Cookie header bytes are accumulated separately and joined before being stored. .. py:function:: _check_api_auth(raw_headers: dict[bytes, bytes], secret: str) -> bool Return True when the request carries valid API credentials. Checks ``Authorization: Bearer`` and ``X-Corvix-Token`` headers first (programmatic clients, curl, etc.). Falls back to the ``corvix_session`` cookie so browser SPAs work after logging in via the web UI without needing to inject custom headers into every ``fetch()`` call. .. py:function:: _check_ui_auth(raw_headers: dict[bytes, bytes], secret: str) -> bool Return True when the request carries a valid ``corvix_session`` cookie. .. py:class:: TokenAuthMiddleware Bases: :py:obj:`litestar.middleware.base.ASGIMiddleware` Optional token-based authentication middleware. Reads the secret from ``CORVIX_SECRET_TOKEN`` (or ``CORVIX_SECRET_TOKEN_FILE``) with a 60-second TTL cache so that config changes take effect without a restart while avoiding per-request file I/O. When the variable is absent the middleware is a transparent pass-through. .. py:attribute:: scopes .. py:method:: handle(scope: litestar.types.asgi_types.Scope, receive: litestar.types.asgi_types.Receive, send: litestar.types.asgi_types.Send, next_app: litestar.types.asgi_types.ASGIApp) -> None :async: Authenticate the request or pass it through.