auth

auth

Authentication Provider System

This module provides a flexible, extensible authentication system with multiple provider strategies.

Key Components: - AuthMethod: Enum of authentication strategies - AuthResult: Dataclass tracking which method succeeded and metadata - AuthProvider: Protocol for implementing custom providers - Built-in providers: Environment, CodeEngine, OnePassword - get_auth_with_providers: Main orchestration function

Usage:

Basic usage with default providers:

from crew_dcs.integrations.auth import get_auth_with_providers

# Try all available methods in order
result = await get_auth_with_providers(
    target_instance="my-instance",
    vault_id="op-vault-id"  # Optional: enables OnePassword
)

print(f"Authenticated via {result.method.value}")
auth = result.auth  # Use the DomoAuth instance

Custom provider chain:

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

providers = [
    OnePasswordAuthProvider(),
    EnvironmentAuthProvider(),
]

result = await get_auth_with_providers(
    providers=providers,
    target_instance="my-instance",
    vault_id="op-vault-id"
)

Custom provider implementation:

from crew_dcs.integrations.auth import AuthProvider, AuthMethod
import crew_dcs.auth as dmda

class VaultAuthProvider:
    def __init__(self):
        self.method = AuthMethod.CUSTOM

    async def try_auth(self, **kwargs) -> dmda.DomoAuth | None:
        # Your custom authentication logic
        try:
            token = retrieve_from_vault(kwargs.get('vault_key'))
            return dmda.DomoTokenAuth(
                domo_instance=kwargs['target_instance'],
                domo_access_token=token
            )
        except Exception:
            return None

OnePassword Utilities:

Direct access to OnePassword utilities:

from crew_dcs.integrations.auth.onepassword import (
    generate_client,
    get_item_by_title,
    get_auth_from_onepass
)

# Create a 1Password client
client = await generate_client()

# Get an item by title
item = await get_item_by_title(client, vault_id="...", title="sdk_instance")

# Get auth directly (without provider system)
auth = await get_auth_from_onepass(
    vault_id="...",
    item_title="sdk_instance"
)

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)

CodeEngineAuthProvider

CodeEngineAuthProvider()

Provider for CodeEngine sudo authentication.

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

def __init__(self):
    self.method = AuthMethod.CODEENGINE

try_auth async

try_auth(
    sudo_function_instance_auth: DomoAuth | None = None,
    target_instance: str | None = None,
    account_name: str | None = None,
    sudo_package_id: str | None = None,
    sudo_package_version: str | None = None,
    function_name: str = "get_account",
    debug_api: bool = False,
    **kwargs
) -> DomoAuth | None

Try to authenticate via CodeEngine sudo.

Parameters:

Name	Type	Description	Default
sudo_function_instance_auth	DomoAuth | None	Authenticated auth for CodeEngine calls	None
target_instance	str | None	Target Domo instance	None
account_name	str | None	Account identifier (defaults to sdk_{target_instance})	None
sudo_package_id	str | None	CodeEngine package ID	None
sudo_package_version	str | None	CodeEngine package version	None
function_name	str	CodeEngine function name	'get_account'
debug_api	bool	Enable debug output	False
**kwargs		Additional arguments (ignored)	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth if successful, None if prerequisites missing or auth fails

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

async def try_auth(
    self,
    sudo_function_instance_auth: dmda.DomoAuth | None = None,
    target_instance: str | None = None,
    account_name: str | None = None,
    sudo_package_id: str | None = None,
    sudo_package_version: str | None = None,
    function_name: str = "get_account",
    debug_api: bool = False,
    **kwargs,
) -> dmda.DomoAuth | None:
    """
    Try to authenticate via CodeEngine sudo.

    Args:
        sudo_function_instance_auth: Authenticated auth for CodeEngine calls
        target_instance: Target Domo instance
        account_name: Account identifier (defaults to sdk_{target_instance})
        sudo_package_id: CodeEngine package ID
        sudo_package_version: CodeEngine package version
        function_name: CodeEngine function name
        debug_api: Enable debug output
        **kwargs: Additional arguments (ignored)

    Returns:
        DomoAuth if successful, None if prerequisites missing or auth fails
    """
    from crew_dcs.integrations.auth_utils import get_auth_from_codeengine

    if (
        not sudo_function_instance_auth
        or not sudo_package_id
        or not sudo_package_version
    ):
        await logger.error("CodeEngine auth prerequisites not met")
        raise ValueError("Missing prerequisites for CodeEngine auth")

    if not target_instance:
        await logger.error("CodeEngine auth requires target_instance")
        raise ValueError("Missing target_instance for CodeEngine auth")

    await logger.info(
        f"Attempting CodeEngine auth for {target_instance} "
        f"(package: {sudo_package_id})"
    )

    try:
        return await get_auth_from_codeengine(
            config_auth=sudo_function_instance_auth,
            target_instance=target_instance,
            sudo_package_id=sudo_package_id,
            sudo_package_version=sudo_package_version,
            retrieval_account_name=account_name or f"sdk_{target_instance}",
            function_name=function_name,
            debug_api=debug_api,
        )
    except (KeyError, AssertionError) as e:
        # Expected failures: missing account, invalid token
        await logger.error(f"CodeEngine auth failed: {e}")
        raise e

EnvironmentAuthProvider

EnvironmentAuthProvider()

Provider for environment variable authentication.

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

def __init__(self):
    self.method = AuthMethod.ENVIRONMENT

try_auth async

try_auth(
    target_instance: str | None = None,
    domo_instance_env_var: str = "DOMO_INSTANCE",
    domo_token_env_var: str = "DOMO_ACCESS_TOKEN",
    **kwargs
) -> DomoAuth | None

Try to authenticate from environment variables.

Parameters:

Name	Type	Description	Default
target_instance	str | None	Override instance name (sets env var temporarily)	None
domo_instance_env_var	str	Environment variable name for instance	'DOMO_INSTANCE'
domo_token_env_var	str	Environment variable name for token	'DOMO_ACCESS_TOKEN'
**kwargs		Additional arguments (ignored)	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth if successful, None if env vars not available

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

async def try_auth(
    self,
    target_instance: str | None = None,
    domo_instance_env_var: str = "DOMO_INSTANCE",
    domo_token_env_var: str = "DOMO_ACCESS_TOKEN",
    **kwargs,
) -> dmda.DomoAuth | None:
    """
    Try to authenticate from environment variables.

    Args:
        target_instance: Override instance name (sets env var temporarily)
        domo_instance_env_var: Environment variable name for instance
        domo_token_env_var: Environment variable name for token
        **kwargs: Additional arguments (ignored)

    Returns:
        DomoAuth if successful, None if env vars not available
    """
    await logger.debug(
        f"Attempting environment auth for {target_instance or 'default instance'}"
    )

    try:
        from crew_dcs.integrations.auth_utils import get_auth_from_env

        # Temporarily override env var if target_instance provided
        original = None
        if target_instance:
            original = os.environ.get(domo_instance_env_var)
            os.environ[domo_instance_env_var] = target_instance

        auth = await get_auth_from_env(
            domo_instance_env_var=domo_instance_env_var,
            domo_token_env_var=domo_token_env_var,
        )

        # Restore original env var
        if target_instance:
            if original:
                os.environ[domo_instance_env_var] = original
            else:
                os.environ.pop(domo_instance_env_var, None)

        return auth

    except (KeyError, AssertionError) as e:
        await logger.debug(f"Environment auth failed: {e}")
        raise e

OnePasswordAuthProvider

OnePasswordAuthProvider()

Provider for 1Password authentication.

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

def __init__(self):
    self.method = AuthMethod.ONEPASSWORD

try_auth async

try_auth(
    vault_id: str | None = None,
    item_title: str | None = None,
    target_instance: str | None = None,
    domo_instance: str | None = None,
    client=None,
    debug_api: bool = False,
    **kwargs
) -> DomoAuth | None

Try to authenticate via 1Password.

Parameters:

Name	Type	Description	Default
vault_id	str | None	1Password vault ID	None
item_title	str | None	1Password item title (defaults to sdk_{target_instance})	None
target_instance	str | None	Target Domo instance name	None
domo_instance	str | None	Override for final instance name	None
client		Optional existing 1Password client	None
debug_api	bool	Enable debug output	False
**kwargs		Additional arguments (ignored)	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth if successful, None if prerequisites missing or auth fails

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

async def try_auth(
    self,
    vault_id: str | None = None,
    item_title: str | None = None,
    target_instance: str | None = None,
    domo_instance: str | None = None,
    client=None,
    debug_api: bool = False,
    **kwargs,
) -> dmda.DomoAuth | None:
    """
    Try to authenticate via 1Password.

    Args:
        vault_id: 1Password vault ID
        item_title: 1Password item title (defaults to sdk_{target_instance})
        target_instance: Target Domo instance name
        domo_instance: Override for final instance name
        client: Optional existing 1Password client
        debug_api: Enable debug output
        **kwargs: Additional arguments (ignored)

    Returns:
        DomoAuth if successful, None if prerequisites missing or auth fails
    """

    from .onepassword import get_auth_from_onepass

    if not vault_id:
        await logger.error("OnePassword auth requires vault_id")
        raise ValueError("vault_id is required for OnePassword auth")

    # Determine item title and instance
    resolved_item_title = item_title or (
        target_instance and f"sdk_{target_instance}"
    )
    resolved_instance = domo_instance or target_instance

    if not resolved_item_title:
        await logger.error(
            "OnePassword auth requires item_title or target_instance"
        )
        raise ValueError(
            "item_title or target_instance is required for OnePassword auth"
        )

    await logger.info(f"Attempting OnePassword auth for {resolved_item_title}")

    return await get_auth_from_onepass(
        vault_id=vault_id,
        item_title=resolved_item_title,
        domo_instance=resolved_instance,
        client=client,
        debug_api=debug_api,
    )

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}"
    )

Modules

• core
• onepassword
• providers