nat.front_ends.console.authentication_flow_handler#

Attributes#

logger

Classes#

_FlowState

ConsoleAuthenticationFlowHandler

Authentication helper for CLI / console environments. Supports:

Module Contents#

logger#
class _FlowState#
future: asyncio.Future#
challenge: str | None = None#
verifier: str | None = None#
token_url: str | None = None#
use_pkce: bool | None = None#
class ConsoleAuthenticationFlowHandler#

Bases: nat.authentication.interfaces.FlowHandlerBase

Authentication helper for CLI / console environments. Supports:

  • HTTP Basic (username/password)

  • OAuth 2 Authorization‑Code with optional PKCE

_server_controller: nat.front_ends.fastapi.fastapi_front_end_controller._FastApiFrontEndController | None = None#
_redirect_app: fastapi.FastAPI | None = None#
_flows: dict[str, _FlowState]#
_active_flows = 0#
_server_lock#
_oauth_client: authlib.integrations.httpx_client.AsyncOAuth2Client | None = None#
async authenticate(
config: nat.data_models.authentication.AuthProviderBaseConfig,
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.

construct_oauth_client(
cfg: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig,
) → authlib.integrations.httpx_client.AsyncOAuth2Client#

Separated for easy overriding in tests (to inject ASGITransport).

_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

static _handle_http_basic() → nat.data_models.authentication.AuthenticatedContext#
async _handle_oauth2_auth_code_flow(
cfg: nat.authentication.oauth2.oauth2_auth_code_flow_provider_config.OAuth2AuthCodeFlowProviderConfig,
) → nat.data_models.authentication.AuthenticatedContext#
async _build_redirect_app() → fastapi.FastAPI#

  • If cfg.run_redirect_local_server == True → start a local server.

  • Else → only build the redirect app and save it to self._redirect_app

    for in‑process testing.

async _start_redirect_server() → None#
async _stop_redirect_server() → None#
property redirect_app: fastapi.FastAPI | None#

In test mode (run_redirect_local_server=False) the in‑memory redirect app is exposed for testing purposes.