Middleware

Module contents

This package provides custom middleware components that intercept and process requests or responses in the application lifecycle.

class back_chat.middleware.AuthMiddleware(*args, **kwargs)[source]

Bases: BaseHTTPMiddleware

Middleware to handle authentication for incoming HTTP requests.

This middleware intercepts requests and determines if they need authentication based on a predefined list of paths that can bypass authentication (e.g., docs, health check).

If authentication is required, it checks for the presence of a valid API key or a valid authorization token (JWT), decoding it via Keycloak OpenID service.

__jump_paths__

List of URL paths that do not require

Type:

list

authentication.
__auth__

Header key name for authorization token.

Type:

str

decode_token(token: str)[source]

Decode a JWT token after stripping ‘Bearer ‘ prefix.

Parameters:

token – Raw token string from the authorization header.

Returns:

Decoded token payload (usually a dict).

async dispatch(request: Request, call_next: Callable[[Request], Awaitable[Response]]) Response[source]

Process incoming HTTP request, enforcing authentication if required.

If the request path is in the bypass list, it proceeds without checks. Otherwise, it verifies authentication and returns an unauthorized response if authentication fails.

Parameters:
  • request – The incoming HTTP request.

  • call_next – The next middleware or request handler callable.

Returns:

Response from next handler or unauthorized response.

get_header_token(request: Request) str[source]

Get the authorization token from the request headers.

Parameters:

request – The incoming HTTP request.

Returns:

Authorization header value or empty string if missing.

get_user_config(request: Request) dict | None[source]

Extract user configuration by decoding the JWT token from the request.

Parameters:

request – The incoming HTTP request.

Returns:

Decoded token payload dict if valid, else None.

is_auth(request: Request) dict | None[source]

Check whether the request is authenticated.

Currently implemented by trying to get user config from token.

Parameters:

request – The incoming HTTP request.

Returns:

User configuration dict if authenticated, else None.

static unauthorised(code: int = 401, msg: str = 'Unauthorised') JSONResponse[source]

Return a JSON response indicating an unauthorized access attempt.

Parameters:
  • code – HTTP status code to return (default 401).

  • msg – Message to include in the response body

(default ‘Unauthorised’). :return: JSONResponse with status code and message.

class back_chat.middleware.WebSocketAuthMiddleware[source]

Bases: object

Custom middleware for WebSocket authentication.

This class does not inherit from BaseHTTPMiddleware and is specifically designed to handle authentication for WebSocket connections by verifying a JWT token provided in the headers.

__auth__

The name of the header expected to contain the

Type:

str

authorization token.
decode_token(token: str)[source]

Decodes a JWT token by stripping the ‘Bearer ‘ prefix and using the decoding method configured in KEYCLOAK_OPENID.

Parameters:

token – The full JWT token, possibly with a ‘Bearer ‘ prefix.

Returns:

The decoded payload from the JWT token.

Raises:

Exceptions raised by KEYCLOAK_OPENID.decode_token if the

token is invalid.

is_auth(token: str) str[source]

Verifies the authenticity of the JWT token.

Attempts to decode the token and, if valid, returns the payload. If an error occurs, logs an authentication failure message and returns an empty string.

Parameters:

token – The JWT token to verify.

Returns:

The decoded payload if valid, otherwise an empty string.

async unauthorised(websocket: WebSocket, code: int = 1008, msg: str = 'Unauthorised')[source]

Closes the WebSocket connection with a specific close code and reason.

Close code 1008 is used to indicate a policy violation or failed authentication.

Parameters:
  • websocket – The WebSocket instance to close.

  • code – The WebSocket close code (default is 1008).

  • msg – The reason message for closing the connection (default is

‘Unauthorised’).

Submodules

Auth module

class back_chat.middleware.auth.AuthMiddleware(*args, **kwargs)[source]

Bases: BaseHTTPMiddleware

Middleware to handle authentication for incoming HTTP requests.

This middleware intercepts requests and determines if they need authentication based on a predefined list of paths that can bypass authentication (e.g., docs, health check).

If authentication is required, it checks for the presence of a valid API key or a valid authorization token (JWT), decoding it via Keycloak OpenID service.

__jump_paths__

List of URL paths that do not require

Type:

list

authentication.
__auth__

Header key name for authorization token.

Type:

str

decode_token(token: str)[source]

Decode a JWT token after stripping ‘Bearer ‘ prefix.

Parameters:

token – Raw token string from the authorization header.

Returns:

Decoded token payload (usually a dict).

async dispatch(request: Request, call_next: Callable[[Request], Awaitable[Response]]) Response[source]

Process incoming HTTP request, enforcing authentication if required.

If the request path is in the bypass list, it proceeds without checks. Otherwise, it verifies authentication and returns an unauthorized response if authentication fails.

Parameters:
  • request – The incoming HTTP request.

  • call_next – The next middleware or request handler callable.

Returns:

Response from next handler or unauthorized response.

get_header_token(request: Request) str[source]

Get the authorization token from the request headers.

Parameters:

request – The incoming HTTP request.

Returns:

Authorization header value or empty string if missing.

get_user_config(request: Request) dict | None[source]

Extract user configuration by decoding the JWT token from the request.

Parameters:

request – The incoming HTTP request.

Returns:

Decoded token payload dict if valid, else None.

is_auth(request: Request) dict | None[source]

Check whether the request is authenticated.

Currently implemented by trying to get user config from token.

Parameters:

request – The incoming HTTP request.

Returns:

User configuration dict if authenticated, else None.

static unauthorised(code: int = 401, msg: str = 'Unauthorised') JSONResponse[source]

Return a JSON response indicating an unauthorized access attempt.

Parameters:
  • code – HTTP status code to return (default 401).

  • msg – Message to include in the response body

(default ‘Unauthorised’). :return: JSONResponse with status code and message.

Auth websocket module

class back_chat.middleware.auth_websocket.WebSocketAuthMiddleware[source]

Bases: object

Custom middleware for WebSocket authentication.

This class does not inherit from BaseHTTPMiddleware and is specifically designed to handle authentication for WebSocket connections by verifying a JWT token provided in the headers.

__auth__

The name of the header expected to contain the

Type:

str

authorization token.
decode_token(token: str)[source]

Decodes a JWT token by stripping the ‘Bearer ‘ prefix and using the decoding method configured in KEYCLOAK_OPENID.

Parameters:

token – The full JWT token, possibly with a ‘Bearer ‘ prefix.

Returns:

The decoded payload from the JWT token.

Raises:

Exceptions raised by KEYCLOAK_OPENID.decode_token if the

token is invalid.

is_auth(token: str) str[source]

Verifies the authenticity of the JWT token.

Attempts to decode the token and, if valid, returns the payload. If an error occurs, logs an authentication failure message and returns an empty string.

Parameters:

token – The JWT token to verify.

Returns:

The decoded payload if valid, otherwise an empty string.

async unauthorised(websocket: WebSocket, code: int = 1008, msg: str = 'Unauthorised')[source]

Closes the WebSocket connection with a specific close code and reason.

Close code 1008 is used to indicate a policy violation or failed authentication.

Parameters:
  • websocket – The WebSocket instance to close.

  • code – The WebSocket close code (default is 1008).

  • msg – The reason message for closing the connection (default is

‘Unauthorised’).