Authentication Methods

Authentication in flyte-sdk is managed through a flexible system of authenticators, all deriving from the Authenticator base class. This design allows for various authentication flows to be plugged in, catering to different deployment scenarios and security requirements.

The Base Authenticator

The remote._client.auth._authenticators.base.Authenticator class provides the foundational structure and common functionalities for all authentication flows. It handles aspects like endpoint configuration, client configuration resolution, credential storage and retrieval, and secure HTTP communication.

When you initialize an authenticator, you provide core parameters such as the authentication endpoint, and optionally, a ClientConfigStore to fetch remote configurations or a ClientConfig object for static settings. It also supports HTTP proxy settings, SSL verification, and custom CA certificates.

class Authenticator(object):
    def __init__(
        self,
        endpoint: str,
        *,
        cfg_store: typing.Optional[ClientConfigStore] = None,
        client_config: typing.Optional[ClientConfig] = None,
        credentials: typing.Optional[Credentials] = None,
        http_session: typing.Optional[httpx.AsyncClient] = None,
        http_proxy_url: typing.Optional[str] = None,
        verify: bool = True,
        ca_cert_path: typing.Optional[str] = None,
        default_header_key: str = "authorization",
        **kwargs,
    ):
        pass # Placeholder for initialization logic

All concrete authenticator implementations must provide an asynchronous _do_refresh_credentials() method. This method is responsible for performing the actual authentication flow to obtain or refresh access tokens. The base class handles the thread-safe and coroutine-safe refreshing of credentials via refresh_credentials().

PKCE Authenticator for Browser-Based Login

When your application requires user interaction through a web browser for authentication, such as in a typical desktop or mobile application, the remote._client.auth._authenticators.pkce.PKCEAuthenticator implements the Proof Key for Code Exchange (PKCE) OAuth 2.0 flow. This authenticator automatically opens a browser window for the user to log in.

You typically don't instantiate PKCEAuthenticator directly. Instead, you specify "Pkce" as the auth_type when initializing the Flyte client, and the get_async_authenticator factory function handles its creation:

# Example from remote/_client/auth/_authenticators/factory.py
match auth_type:
    case "Pkce":
        from flyte.remote._client.auth._authenticators.pkce import PKCEAuthenticator

        return PKCEAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)

Internally, PKCEAuthenticator manages the generation of code_verifier and code_challenge and orchestrates the interaction with the authorization and token endpoints. The _do_refresh_credentials() method first attempts to refresh credentials using a refresh token if available. If not, it initiates the full PKCE flow, which involves user interaction.

class PKCEAuthenticator(Authenticator):
    async def _do_refresh_credentials(self) -> Credentials:
        await self._initialize_auth_client()
        if self._creds:
            pass # Attempt to refresh token
        return await self._auth_client.get_creds_from_remote()

Auth0 Specific Configuration for PKCE

If you are using Auth0 with PKCEAuthenticator, you must manually configure your config.yaml to include specific scopes to enable refresh tokens and caching. For instance, admin.scopes should include ["offline_access", "offline", "all", "openid"]. However, for your FlyteCTL Helm configuration, only ["offline", "all"] should be used, as OIDC scopes are not grantable in Auth0 customer APIs.

Device Code Authenticator for Headless Environments

For applications running in headless environments or on input-constrained devices where a browser-based login is not feasible, the remote._client.auth._authenticators.device_code.DeviceCodeAuthenticator implements the OAuth 2.0 Device Authorization Grant flow.

Similar to PKCEAuthenticator, this authenticator is typically created by the factory function when "DeviceFlow" is specified as the auth_type:

# Example from remote/_client/auth/_authenticators/factory.py
match auth_type:
    # ...
    case "DeviceFlow":
        from flyte.remote._client.auth._authenticators.device_code import DeviceCodeAuthenticator

        return DeviceCodeAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)

The _do_refresh_credentials() method in DeviceCodeAuthenticator first attempts to refresh credentials using an existing refresh token. If that fails or no refresh token is present, it initiates the device code flow. This involves displaying a URL and a user code to the console, which the user must enter in a browser on another device to complete authentication.

class DeviceCodeAuthenticator(Authenticator):
    async def _do_refresh_credentials(self) -> Credentials:
        cfg = await self._resolve_config()

        if cfg.device_authorization_endpoint is None:
            raise AuthenticationError(
                "Device Authentication is not available on the Flyte backend / authentication server"
            )

        if self._creds and self._creds.refresh_token:
            pass # Attempt to refresh token

        # Fall back to device flow
        resp = await token_client.get_device_code(
            cfg.device_authorization_endpoint,
            cfg.client_id,
            scopes=cfg.scopes,
            http_session=self._http_session,
        )

        full_uri = f"{resp.verification_uri}?user_code={resp.user_code}"
        click.secho(f"To Authenticate, navigate in a browser to the following URL: {click.style(full_uri, fg='blue', underline=True)}")
        pass # Poll token endpoint

Client Credentials Authenticator for Service Accounts

For machine-to-machine authentication, such as for service accounts or backend services, the remote._client.auth._authenticators.client_credentials.ClientCredentialsAuthenticator uses the OAuth 2.0 Client Credentials Grant flow. This flow relies on a client_id and client_credentials_secret to obtain an access token.

To use this authenticator, you would specify "ClientSecret" as the auth_type:

# Example from remote/_client/auth/_authenticators/factory.py
match auth_type:
    # ...
    case "ClientSecret":
        from flyte.remote._client.auth._authenticators.client_credentials import ClientCredentialsAuthenticator

        return ClientCredentialsAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)

When initializing ClientCredentialsAuthenticator, you must provide both the client_id and client_credentials_secret:

class ClientCredentialsAuthenticator(Authenticator):
    def __init__(
        self,
        client_id: str,
        client_credentials_secret: str,
        **kwargs,
    ):
        if not client_id or not client_credentials_secret:
            raise ValueError("both client_id and client_credentials_secret are required.")
        self._client_id = client_id
        self._client_credentials_secret = client_credentials_secret
        super().__init__(**kwargs)

The _do_refresh_credentials() method constructs a basic authorization header using the provided client ID and secret, then requests a token from the configured token_endpoint.

Async Command Authenticator for Custom Authentication

When standard authentication flows don't fit your needs, or you need to integrate with an existing external authentication system, the remote._client.auth._authenticators.external_command.AsyncCommandAuthenticator allows you to retrieve an access token by executing an arbitrary external command.

This authenticator is selected by specifying "ExternalCommand" as the auth_type:

# Example from remote/_client/auth/_authenticators/factory.py
match auth_type:
    # ...
    case "ExternalCommand":
        from flyte.remote._client.auth._authenticators.external_command import AsyncCommandAuthenticator

        return AsyncCommandAuthenticator(endpoint=endpoint, command=command, verify=verify, **kwargs)

You must provide a command as a list of strings during initialization. This command will be executed, and its standard output will be used as the access token.

class AsyncCommandAuthenticator(Authenticator):
    def __init__(self, command: typing.Optional[typing.List[str]], **kwargs):
        self._cmd = command
        if not self._cmd:
            raise AuthenticationError("Command cannot be empty for command authenticator")
        super().__init__(**kwargs)

The _do_refresh_credentials() method uses asyncio.create_subprocess_exec to run the specified command. It captures stdout and stderr, and if the command exits successfully, the stdout is decoded and used as the access_token.

class AsyncCommandAuthenticator(Authenticator):
    async def _do_refresh_credentials(self) -> Credentials:
        cmd_joined = " ".join(typing.cast(str, self._cmd))
        try:
            process = await asyncio.create_subprocess_exec(
                *typing.cast(typing.List[str], 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 `{cmd_joined}`."
                    f" Please execute this command in your terminal to debug."
                )

            return Credentials(for_endpoint=self._endpoint, access_token=stdout.decode().strip())
        except Exception as e:
            raise AuthenticationError(
                f"Failed to refresh token with command `{cmd_joined}`."
                f" Please execute this command in your terminal to debug."
            )

It's crucial that the command provided is valid and returns the access token on standard output. An AuthenticationError will be raised if the command fails to execute or returns a non-zero exit code.

AsyncCommandAuthenticator is also used for proxy authentication, where it's instantiated with a proxy_command and a custom header_key of "proxy-authorization".

Client Configuration Management

Authentication flows often require specific client configuration details, such as token endpoints, client IDs, and redirect URIs. flyte-sdk manages these configurations using ClientConfig and ClientConfigStore.

ClientConfig

The remote._client.auth._client_config.ClientConfig is a Pydantic BaseModel that defines the structure for all necessary client-side authentication settings. It includes fields like token_endpoint, authorization_endpoint, redirect_uri, client_id, scopes, and audience.

class ClientConfig(pydantic.BaseModel):
    token_endpoint: str
    authorization_endpoint: str
    redirect_uri: str
    client_id: str
    device_authorization_endpoint: typing.Optional[str] = None
    scopes: typing.Optional[typing.List[str]] = None
    header_key: str = "authorization"
    audience: typing.Optional[str] = None

    def with_override(self, other: "ClientConfig") -> "ClientConfig":
        """
        Returns a new ClientConfig instance with the values from the other instance overriding the current instance.
        """
        pass # Simplified for documentation example

ClientConfig also provides a with_override() method, which allows you to merge two ClientConfig instances, with the values from the other instance taking precedence. This is used internally by authenticators to combine local and remote configurations.

You can provide a ClientConfig object directly when initializing the Flyte client, allowing for custom or pre-configured authentication settings:

# Example from _initialize.py
async def _initialize_client(
    api_key: str | None = None,
    auth_type: AuthType = "Pkce",
    endpoint: str | None = None,
    client_config: ClientConfig | None = None,
    # ...
  ):
    pass # Function body

ClientConfigStore

The remote._client.auth._client_config.ClientConfigStore is an abstract base class that defines an interface for retrieving ClientConfig objects. Its primary method is get_client_config().

class ClientConfigStore(object):
    @abstractmethod
    async def get_client_config(self) -> ClientConfig: ...

Concrete implementations, such as RemoteClientConfigStore, are responsible for fetching these configurations from various sources, including remote Flyte servers. The Authenticator's _resolve_config() method uses a ClientConfigStore to fetch remote configurations and merge them with any locally provided ClientConfig.

If an authenticator is initialized without a cfg_store and attempts to resolve configuration, it will raise a ValueError.

Centralized Authenticator Creation

The remote._client.auth._authenticators.factory.get_async_authenticator function serves as the central entry point for creating authenticator instances. It abstracts away the complexity of choosing and initializing the correct authenticator based on the auth_type parameter.

# Excerpt from remote/_client/auth/_authenticators/factory.py
def get_async_authenticator(
    endpoint: str,
    auth_type: AuthType,
    cfg_store: typing.Optional[ClientConfigStore] = None,
    command: typing.Optional[typing.List[str]] = None,
    verify: bool = True,
    **kwargs,
) -> Authenticator:
    match auth_type:
        case "Pkce":
            from flyte.remote._client.auth._authenticators.pkce import PKCEAuthenticator

            return PKCEAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)
        case "ClientSecret":
            from flyte.remote._client.auth._authenticators.client_credentials import ClientCredentialsAuthenticator

            return ClientCredentialsAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)
        case "ExternalCommand":
            from flyte.remote._client.auth._authenticators.external_command import AsyncCommandAuthenticator

            return AsyncCommandAuthenticator(endpoint=endpoint, command=command, verify=verify, **kwargs)
        case "DeviceFlow":
            from flyte.remote._client.auth._authenticators.device_code import DeviceCodeAuthenticator

            return DeviceCodeAuthenticator(endpoint=endpoint, cfg_store=cfg_store, verify=verify, **kwargs)
        case _:
            raise ValueError(f"Unknown auth type {auth_type}")

This factory function ensures that the appropriate authenticator is created with the necessary parameters, simplifying the authentication setup for flyte-sdk users.