nat.front_ends.fastapi.auth_flow_handlers.websocket_flow_handler

Attributes

logger

Classes

FlowState
WebSocketAuthenticationFlowHandler	Handles front-end specific flows for authentication clients.

Module Contents

logger

class FlowState

future: asyncio.Future

challenge: str | None = None

verifier: str | None = None

client: authlib.integrations.httpx_client.AsyncOAuth2Client | None = None

config: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig | None = None

class WebSocketAuthenticationFlowHandler(

add_flow_cb: collections.abc.Callable[[str, FlowState], collections.abc.Awaitable[None]],

remove_flow_cb: collections.abc.Callable[[str], collections.abc.Awaitable[None]],

web_socket_message_handler: nat.front_ends.fastapi.message_handler.WebSocketMessageHandler,

auth_timeout_seconds: float = 300.0,

)

Bases: nat.authentication.interfaces.FlowHandlerBase

Handles front-end specific flows for authentication clients.

Each front end will define a FlowHandler that will implement the authenticate method.

The authenticate method will be stored as the callback in the ContextState.user_auth_callback

_add_flow_cb: collections.abc.Callable[[str, FlowState], collections.abc.Awaitable[None]]

_remove_flow_cb: collections.abc.Callable[[str], collections.abc.Awaitable[None]]

_web_socket_message_handler: nat.front_ends.fastapi.message_handler.WebSocketMessageHandler

_auth_timeout_seconds: float = 300.0

async authenticate(

config: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig,

method: nat.data_models.authentication.AuthFlowType,

) → nat.data_models.authentication.AuthenticatedContext

Perform the authentication process for the client.

This method handles the necessary steps to authenticate the client with the target API service, which may include obtaining tokens, refreshing credentials, or completing multistep authentication flows.

Raises:

NotImplementedError: Must be implemented by subclasses.

create_oauth_client(

config: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig,

) → authlib.integrations.httpx_client.AsyncOAuth2Client

_create_authorization_url(

client: authlib.integrations.httpx_client.AsyncOAuth2Client,

config: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig,

state: str,

verifier: str | None = None,

challenge: str | None = None,

) → str

Create OAuth authorization URL with proper error handling.

Args:

client: The OAuth2 client instance config: OAuth2 configuration state: OAuth state parameter verifier: PKCE verifier (if using PKCE) challenge: PKCE challenge (if using PKCE)

Returns:

The authorization URL

async _handle_oauth2_auth_code_flow(

config: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig,

) → nat.data_models.authentication.AuthenticatedContext