nebius.aio.token.federation

module documentation

(source)

Federation bearer authentication utilities.

This module provides functions to perform OAuth 2.0 authorization code flow with PKCE for federation bearer token authentication. It handles browser-based authentication, callback server management, and token exchange.

Class	`GetTokenResult`	Result of token retrieval.
Async Function	`authorize`	Perform full OAuth authorization flow and return access token.
Async Function	`get_code`	Obtain authorization code via OAuth callback.
Async Function	`get_token`	Exchange authorization code for access token.
Function	`https_url`	Ensure the URL uses HTTPS scheme.
Async Function	`open_browser`	Open the given URL in the default web browser.
Variable	`log`	Undocumented

async def authorize(client_id: str, federation_endpoint: str, federation_id: str, writer: TextIO | None = None, no_browser_open: bool = False, timeout: float | None = 300, ssl_ctx: ssl.SSLContext | None = None) -> GetTokenResult: (source) ¶

Perform full OAuth authorization flow and return access token.

Combines getting the authorization code and exchanging it for a token.

Parameters
client_id:str	The OAuth client ID.
federation_endpoint:str	The base federation endpoint URL.
federation_id:str	The federation identifier.
writer:`TextIO` or None	Optional text stream to write messages to.
no_browser_open:bool	If True, do not open browser automatically.
timeout:float or None	Timeout in seconds for the entire flow.
ssl_ctx:`ssl.SSLContext` or None	Optional SSL context for token request.
Returns
`GetTokenResult`	The token result containing access token and expiration.

async def get_code(client_id: str, auth_endpoint: str, federation_id: str, pkce_code: PKCE, writer: TextIO | None = None, no_browser_open: bool = False, timeout: float | None = 300) -> tuple[str, str]: (source) ¶

Obtain authorization code via OAuth callback.

Starts a local callback server, constructs the authorization URL with PKCE, opens the browser (or prints the URL), and waits for the callback.

Parameters
client_id:str	The OAuth client ID.
auth_endpoint:str	The authorization endpoint URL.
federation_id:str	The federation identifier.
pkce_code:`PKCE`	The PKCE code challenge and method.
writer:`TextIO` or None	Optional text stream to write messages to.
no_browser_open:bool	If True, do not open browser automatically.
timeout:float or None	Timeout in seconds for waiting for the code.
Returns
tuple[str, str]	A tuple of (authorization_code, redirect_uri).
Raises
`RuntimeError`	If browser fails to open or no code received.
`TimeoutError`	If timeout waiting for code.

async def get_token(client_id: str, token_url: str, code: str, redirect_uri: str, verifier: str, ssl_ctx: ssl.SSLContext | None = None) -> GetTokenResult: (source) ¶

Exchange authorization code for access token.

Sends a POST request to the token endpoint with the code and PKCE verifier.

Parameters
client_id:str	The OAuth client ID.
token_url:str	The token endpoint URL.
code:str	The authorization code received from callback.
redirect_uri:str	The redirect URI used in the request.
verifier:str	The PKCE code verifier.
ssl_ctx:`ssl.SSLContext` or None	Optional SSL context for the request.
Returns
`GetTokenResult`	The token result containing access token and expiration.
Raises
`RuntimeError`	If token request fails or response is invalid.

def https_url(raw_url: str) -> str: (source) ¶

Ensure the URL uses HTTPS scheme.

Parameters
raw_url:str	The input URL string.
Returns
str	The URL with HTTPS scheme.

async def open_browser(url: str): (source) ¶

Open the given URL in the default web browser.

Handles special case for WSL environments by using cmd.exe.

Parameters
url:str	The URL to open.

log = (source) ¶

Undocumented

nebius.aio.token.federation_bearer.auth

`nebius.aio.token.federation_bearer.auth`