corvix.web.middleware

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 <token> or X-Corvix-Token: <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

logger
_SESSION_COOKIE_NAME
SESSION_MAX_AGE_SECONDS
_PUBLIC_EXACT
_PUBLIC_PREFIXES
_MISCONFIGURED
_SECRET_CACHE
_SECRET_CACHE_TTL

Classes

TokenAuthMiddleware

Optional token-based authentication middleware.

Functions

_make_session_cookie(→ str)	Return a signed, time-limited session cookie value.
_verify_session_cookie(→ bool)	Return True when value is a valid, unexpired session token.
_parse_cookies(→ dict[str, str])	Parse a raw Cookie header value into a name→value dict.
_get_secret(→ str)	Return the configured secret token, or an empty string if not set.
_is_public(→ bool)	Return True when path should never require authentication.
_send_json_401(→ None)	Send a minimal JSON 401 Unauthorized response via ASGI.
_send_redirect(→ None)	Send a 302 redirect response via ASGI.
_parse_request_headers(→ dict[bytes, bytes])	Extract and normalise HTTP headers from an ASGI scope.
_check_api_auth(→ bool)	Return True when the request carries valid API credentials.
_check_ui_auth(→ bool)	Return True when the request carries a valid corvix_session cookie.

Module Contents

corvix.web.middleware.logger

corvix.web.middleware._SESSION_COOKIE_NAME = 'corvix_session'

corvix.web.middleware.SESSION_MAX_AGE_SECONDS: int = 86400

corvix.web.middleware._PUBLIC_EXACT: frozenset[str]

corvix.web.middleware._PUBLIC_PREFIXES: tuple[str, Ellipsis] = ('/assets/',)

corvix.web.middleware._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.

corvix.web.middleware._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 hmac.compare_digest() to prevent timing attacks.

corvix.web.middleware._parse_cookies(cookie_header: str) → dict[str, str]

Parse a raw Cookie header value into a name→value dict.

Uses http.cookies.SimpleCookie from the standard library to handle quoted values and other edge cases correctly.

corvix.web.middleware._MISCONFIGURED: bool = False

corvix.web.middleware._SECRET_CACHE: tuple[float, str] | None = None

corvix.web.middleware._SECRET_CACHE_TTL: float = 60.0

corvix.web.middleware._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.

corvix.web.middleware._is_public(path: str) → bool

Return True when path should never require authentication.

async corvix.web.middleware._send_json_401(send: litestar.types.asgi_types.Send) → None

Send a minimal JSON 401 Unauthorized response via ASGI.

async corvix.web.middleware._send_redirect(send: litestar.types.asgi_types.Send, location: bytes) → None

Send a 302 redirect response via ASGI.

corvix.web.middleware._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 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.

corvix.web.middleware._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.

corvix.web.middleware._check_ui_auth(raw_headers: dict[bytes, bytes], secret: str) → bool

Return True when the request carries a valid corvix_session cookie.

class corvix.web.middleware.TokenAuthMiddleware

Bases: 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.

scopes

async 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

Authenticate the request or pass it through.