core

core

Core authentication types and orchestration logic.

This module defines the core types and main orchestration function for the authentication provider system.

AuthMethod

Bases: Enum

Enum identifying authentication methods.

AuthProvider

Bases: Protocol

Protocol for implementing custom auth providers.

Any class implementing this protocol can be used as an auth provider. Must have a method attribute and implement try_auth method.

Example

class CustomAuthProvider: def init(self): self.method = AuthMethod.CUSTOM

async def try_auth(self, **kwargs) -> dmda.DomoAuth | None:
    # Your custom logic
    return auth or None

try_auth async

try_auth(**kwargs) -> DomoAuth | None

Attempt authentication. Return None if this method unavailable or fails.

Parameters:

Name	Type	Description	Default
**kwargs		Provider-specific arguments	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth instance if successful, None otherwise

Source code in src/crew_dcs/integrations/auth/core.py

async def try_auth(self, **kwargs) -> dmda.DomoAuth | None:
    """
    Attempt authentication. Return None if this method unavailable or fails.

    Args:
        **kwargs: Provider-specific arguments

    Returns:
        DomoAuth instance if successful, None otherwise
    """
    ...

AuthResult dataclass

AuthResult(
    auth: DomoAuth,
    method: AuthMethod,
    details: dict[str, Any],
)

Result of authentication with metadata about how it was obtained.

Attributes:

Name	Type	Description
auth	DomoAuth	The authenticated DomoAuth instance
method	AuthMethod	Which authentication method succeeded
details	dict[str, Any]	Additional metadata (e.g., instance, account name, provider name)

get_auth_with_providers async

get_auth_with_providers(
    providers: list[AuthProvider] | None = None,
    **auth_kwargs
) -> AuthResult

Try authentication providers in order until one succeeds.

Parameters:

Name	Type	Description	Default
providers	list[AuthProvider] | None	Ordered list of auth providers to try. If None, uses default set.	None
**auth_kwargs		Arguments passed to each provider's try_auth method	{}

Returns:

Type	Description
AuthResult	AuthResult with successful auth and metadata

Raises:

Type	Description
ValueError	If all providers fail

Example

Use default providers

result = await get_auth_with_providers(target_instance="my-instance") print(f"Authenticated via {result.method.value}")

Custom provider order

from crew_dcs.integrations.auth import ( EnvironmentAuthProvider, OnePasswordAuthProvider )

providers = [MyCustomProvider(), EnvironmentAuthProvider()] result = await get_auth_with_providers( providers=providers, target_instance="my-instance" )

Source code in src/crew_dcs/integrations/auth/core.py

async def get_auth_with_providers(
    providers: list[AuthProvider] | None = None, **auth_kwargs
) -> AuthResult:
    """
    Try authentication providers in order until one succeeds.

    Args:
        providers: Ordered list of auth providers to try. If None, uses default set.
        **auth_kwargs: Arguments passed to each provider's try_auth method

    Returns:
        AuthResult with successful auth and metadata

    Raises:
        ValueError: If all providers fail

    Example:
        # Use default providers
        result = await get_auth_with_providers(target_instance="my-instance")
        print(f"Authenticated via {result.method.value}")

        # Custom provider order
        from crew_dcs.integrations.auth import (
            EnvironmentAuthProvider,
            OnePasswordAuthProvider
        )

        providers = [MyCustomProvider(), EnvironmentAuthProvider()]
        result = await get_auth_with_providers(
            providers=providers,
            target_instance="my-instance"
        )
    """
    # Resolve environment defaults
    auth_kwargs.setdefault("target_instance", os.environ.get("DOMO_INSTANCE"))
    auth_kwargs.setdefault("vault_id", os.environ.get("OP_VAULT_ID"))
    auth_kwargs.setdefault("sudo_package_id", os.environ.get("SUDO_PACKAGE_ID"))
    auth_kwargs.setdefault(
        "sudo_package_version", os.environ.get("SUDO_PACKAGE_VERSION")
    )

    # Default provider order - import here to avoid circular imports
    if providers is None:
        from .providers import (
            CodeEngineAuthProvider,
            EnvironmentAuthProvider,
            OnePasswordAuthProvider,
        )

        # If config_auth is provided, prioritize CodeEngine and skip Environment
        # (since we already have auth for sudo operations)
        if auth_kwargs.get("config_auth"):
            providers = [
                CodeEngineAuthProvider(),
                OnePasswordAuthProvider(),
            ]
        else:
            providers = [
                OnePasswordAuthProvider(),
                EnvironmentAuthProvider(),
            ]

    await logger.info(
        f"Attempting authentication for {auth_kwargs.get('target_instance', 'unspecified instance')} "
        f"with {len(providers)} provider(s)"
    )

    for provider in providers:
        await logger.debug(f"Trying {provider.method.value} auth provider")
        auth = await provider.try_auth(**auth_kwargs)

        if auth:
            result = AuthResult(
                auth=auth,
                method=provider.method,
                details={
                    "instance": auth_kwargs.get("target_instance"),
                    "provider": provider.__class__.__name__,
                    "account_name": auth_kwargs.get("account_name"),
                },
            )
            await logger.info(
                f"Successfully authenticated via {provider.method.value} "
                f"for {auth_kwargs.get('target_instance')}"
            )
            return result

    tried_methods = [p.method.value for p in providers]
    await logger.error(f"All authentication providers failed. Tried: {tried_methods}")
    raise ValueError(
        f"All {len(providers)} authentication providers failed. Tried: {tried_methods}"
    )