On this page
auth_middleware wraps an ASGI app to authenticate every HTTP and WebSocket request via the Authorization header, attaching AuthContext to scope state or rejecting with 401 / close code 4001.
#auth.src.orxtra.auth._middleware
#auth.src.orxtra.auth._middleware
#auth_middleware
def auth_middleware(app: ASGIApp, authenticator: Authenticator) -> ASGIAppPure ASGI middleware that authenticates requests via the Authorization header.
On success, attaches the AuthContext to scope["state"]["auth_context"]. On failure: - HTTP: returns 401 JSON error. - WebSocket: consumes websocket.connect, then sends websocket.close with code 4001. The inner app is never called.
Non-HTTP/non-WebSocket scopes (lifespan etc.) are passed through unchanged.
#_attach_auth_context
def _attach_auth_context(scope: Scope, auth_context: object) -> NoneStore the AuthContext in scope["state"]["auth_context"].
#_reject_websocket
async def _reject_websocket(receive: Receive, send: Send) -> NoneReject a WebSocket during the handshake.
Consumes the websocket.connect message, then sends websocket.close with code 4001 (unauthorized). This is the correct ASGI sequence for rejecting a WebSocket before acceptance.
#_extract_bearer_token
def _extract_bearer_token(headers: list[tuple[bytes, bytes]]) -> str | NoneExtract the bearer token from ASGI headers.
#_send_error
async def _send_error(send: Send, status: int, detail: str) -> NoneSend a JSON error response.