Authentication Architecture

The flyte-sdk authentication architecture is built around a modular set of authenticators that manage the lifecycle of security credentials for communicating with Flyte Admin. By abstracting the specific OAuth2 flows and token retrieval methods into dedicated classes, the SDK provides a consistent interface for RPC interceptors to obtain and refresh authentication headers.

The Authenticator Base Class

The Authenticator class in flyte.remote._client.auth._authenticators.base serves as the foundation for all authentication flows. It is responsible for three primary tasks:

1. Credential Management: It retrieves and stores credentials using the KeyringStore (unless disabled via disable_keyring).
2. Configuration Resolution: It merges local client settings with remote configuration fetched from the Flyte server.
3. Thread-Safe Refreshing: It implements a "double-check" pattern using an asyncio.Lock and a _creds_id to ensure that concurrent requests do not trigger redundant network calls for token refreshes.

When an RPC call fails due to an expired token, the interceptor calls refresh_credentials(creds_id). The Authenticator only proceeds with a refresh if the provided creds_id matches its current internal ID, indicating that no other thread has already updated the credentials.

# From flyte.remote._client.auth._authenticators.base
async def refresh_credentials(self, creds_id: str | None = None):
    if creds_id and creds_id != self._creds_id:
        # Credentials have been refreshed by another thread/coroutine
        return

    async with self._async_lock:
        if creds_id and creds_id != self._creds_id:
            return

        try:
            self._creds = await self._do_refresh_credentials()
            KeyringStore.store(self._creds, disable=self._disable_keyring)
        except Exception:
            KeyringStore.delete(self._endpoint, disable=self._disable_keyring)
            raise
        self._creds_id = self._creds.id

Configuration Resolution

Authenticators rely on ClientConfig (defined in flyte.remote._client.auth._client_config) to determine OAuth2 endpoints, client IDs, and scopes. The _resolve_config method in the base class fetches remote configuration from a ClientConfigStore and applies local overrides. This allows the flyte-sdk to automatically discover the correct authentication endpoints from the Flyte backend while still allowing users to specify custom scopes or audiences in their local configuration.

Supported Authentication Flows

The flyte-sdk implements several standard and custom authentication flows to accommodate different environments.

PKCE (Proof Key for Code Exchange)

The PKCEAuthenticator is the default flow for interactive user sessions. It initiates a browser-based login by opening the system's default web browser and starting a local callback server to capture the authorization code. It uses an AuthorizationClient to manage the exchange of the authorization code for access and refresh tokens.

Client Credentials

For service-to-service authentication or when using API keys, the ClientCredentialsAuthenticator uses a client_id and client_credentials_secret. It performs a standard OAuth2 client credentials grant to obtain a token.

# From flyte.remote._client.auth._authenticators.client_credentials
async def _do_refresh_credentials(self) -> Credentials:
    cfg = await self._resolve_config()
    authorization_header = token_client.get_basic_authorization_header(
        self._client_id, self._client_credentials_secret
    )
    token, refresh_token, expires_in = await token_client.get_token(
        token_endpoint=cfg.token_endpoint,
        authorization_header=authorization_header,
        # ... other parameters
    )
    return Credentials(access_token=token, refresh_token=refresh_token, expires_in=expires_in)

Device Code Flow

The DeviceCodeAuthenticator is designed for headless environments (like SSH sessions) where a browser cannot be opened automatically. It requests a device code from the authorization server and prints a URL and user code to the terminal. The SDK then polls the token endpoint until the user completes the login on another device.

External Command

The AsyncCommandAuthenticator allows the flyte-sdk to integrate with external credential helpers. It executes a specified shell command (e.g., gcloud auth print-access-token) and uses the command's standard output as the access token. This is useful for environments where tokens are managed by external CLI tools.

# From flyte.remote._client.auth._authenticators.external_command
async def _do_refresh_credentials(self) -> Credentials:
    process = await asyncio.create_subprocess_exec(
        *self._cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await process.communicate()
    if process.returncode != 0:
        raise AuthenticationError(f"Failed to refresh token with command `{self._cmd}`")
    return Credentials(for_endpoint=self._endpoint, access_token=stdout.decode().strip())

Passthrough

The PassthroughAuthenticator is used when authentication metadata is provided directly via a context manager using flyte.remote.auth_metadata(). It does not perform any OAuth2 flow but instead extracts headers from the current execution context.

Integration and Factory

Authenticators are instantiated via the get_async_authenticator factory in flyte.remote._client.auth._authenticators.factory. This factory maps the auth_type configuration string to the appropriate class:

auth_type	Class
Pkce	PKCEAuthenticator
ClientSecret	ClientCredentialsAuthenticator
ExternalCommand	AsyncCommandAuthenticator
DeviceFlow	DeviceCodeAuthenticator
Passthrough	PassthroughAuthenticator

Once created, the authenticator is wrapped in a set of ConnectRPC interceptors (e.g., AuthUnaryInterceptor) that automatically inject the Authorization header into outgoing requests and handle transparent token refreshing on 401 Unauthorized errors.