nebius.aio.token.federation_bearer.server.CallbackHandler

class documentation

class CallbackHandler: (source)

View In Hierarchy

Handler for OAuth callback server during federation authentication.

This class manages a local HTTP server that listens for OAuth authorization code callbacks. It validates the state using PKCE and extracts the authorization code for token exchange.

Method __init__ Undocumented
Async Method listen_and_serve Start the HTTP server on a free port.
Async Method shutdown Shutdown the HTTP server.
Async Method wait_for_code Wait for the authorization code to be received.
Property addr The full address of the callback server.
Property code The authorization code received from the OAuth callback.
Property port The port number the server is listening on.
Property state The PKCE state verifier used for callback validation.
Method _get_free_port Find and bind to a free port on localhost.
Async Method _handle_callback Handle the OAuth callback request.
Async Method _set_code Set the authorization code if the state is valid.
Instance Variable _addr Undocumented
Instance Variable _app Undocumented
Instance Variable _code Undocumented
Instance Variable _done Undocumented
Instance Variable _lock Undocumented
Instance Variable _port Undocumented
Instance Variable _runner Undocumented
Instance Variable _site Undocumented
Instance Variable _state Undocumented
def __init__(self): (source)

Undocumented

async def listen_and_serve(self): (source)

Start the HTTP server on a free port.

Binds to localhost on a free port and starts listening for callbacks.

async def shutdown(self): (source)

Shutdown the HTTP server.

Stops the site and cleans up the runner.

async def wait_for_code(self, timeout: float | None = None) -> str | None: (source)

Wait for the authorization code to be received.

Parameters
timeout:float or NoneOptional timeout in seconds.
Returns
str or NoneThe authorization code or None if timed out.
@property
addr: str = (source)

The full address of the callback server.

@property
code: str or None = (source)

The authorization code received from the OAuth callback.

@property
port: int or None = (source)

The port number the server is listening on.

@property
state: str = (source)

The PKCE state verifier used for callback validation.

def _get_free_port(self) -> tuple[int, socket.socket, str]: (source)

Find and bind to a free port on localhost.

Returns
tuple[int, socket.socket, str]A tuple of (port, socket, address).
async def _handle_callback(self, request: web.Request) -> web.Response: (source)

Handle the OAuth callback request.

Extracts the authorization code and state from the query parameters, validates the state, and sets the code if valid.

Parameters
request:web.RequestThe incoming HTTP request.
Returns
web.ResponseThe HTTP response to send back.
async def _set_code(self, code: str | None, state: str | None): (source)

Set the authorization code if the state is valid.

Parameters
code:str or NoneThe authorization code from the callback.
state:str or NoneThe state parameter from the callback.
_addr: str | None = (source)

Undocumented

_app = (source)

Undocumented

_code: str | None = (source)

Undocumented

_done = (source)

Undocumented

_lock = (source)

Undocumented

_port: int | None = (source)

Undocumented

_runner: web.AppRunner | None = (source)

Undocumented

_site: web.SockSite | None = (source)

Undocumented

_state: str = (source)

Undocumented