orxtra v0.13.0 /auth.src.orxtra.auth._middleware
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

python
def auth_middleware(app: ASGIApp, authenticator: Authenticator) -> ASGIApp

Pure 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

python
def _attach_auth_context(scope: Scope, auth_context: object) -> None

Store the AuthContext in scope["state"]["auth_context"].

#_reject_websocket

python
async def _reject_websocket(receive: Receive, send: Send) -> None

Reject 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

python
def _extract_bearer_token(headers: list[tuple[bytes, bytes]]) -> str | None

Extract the bearer token from ASGI headers.

#_send_error

python
async def _send_error(send: Send, status: int, detail: str) -> None

Send a JSON error response.

Search