onepassword

onepassword

1Password Connect SDK utilities for retrieving credentials from 1Password vaults.

This module provides utilities for authenticating with Domo using credentials stored in 1Password vaults via the 1Password Connect API.

Required Environment Variables: OP_CONNECT_HOST: Your 1Password Connect server URL (e.g., https://your-connect-server.com) OP_CONNECT_TOKEN: Your 1Password Connect API token for authentication

Installation: pip install onepasswordconnectsdk

Usage:

Basic authentication from 1Password:

from crew_dcs.integrations.onepassword import get_auth_from_onepass

auth = await get_auth_from_onepass(
    vault_id="your-vault-id",
    item_title="sdk_my-instance"
)
# auth is a validated DomoTokenAuth instance

Working with 1Password items directly:

from crew_dcs.integrations.onepassword import (
    generate_client,
    get_item_by_title,
    display_item
)

client = await generate_client()
item = await get_item_by_title(client, vault_id, "sdk_instance")
display_item(item, show_fields=True)

display_item

display_item(
    item: Item,
    show_fields: bool = False,
    show_secrets: bool = False,
)

Display formatted output of a 1Password item.

Parameters:

Name	Type	Description	Default
item	Item	The 1Password item to display	required
show_fields	bool	Whether to display all fields in the item	False
show_secrets	bool	Whether to reveal password fields (default hidden for security)	False

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

def display_item(item: Item, show_fields: bool = False, show_secrets: bool = False):
    """
    Display formatted output of a 1Password item.

    Args:
        item: The 1Password item to display
        show_fields: Whether to display all fields in the item
        show_secrets: Whether to reveal password fields (default hidden for security)
    """
    print(f"Item: {item.title}")
    print(f"Category: {item.category}")
    if hasattr(item, "tags") and item.tags:
        print(f"Tags: {', '.join(item.tags)}")

    if not show_fields or not item.fields:
        return

    print("\nFields:")
    print("-" * 40)

    for field in item.fields:
        field_label = field.label if field.label else f"Field ({field.purpose})"
        field_type = field.type if field.type else "TEXT"

        if field.purpose == "PASSWORD" and not show_secrets:
            print(
                f"{field_label} ({field_type}): [HIDDEN - use show_secrets=True to reveal]"
            )
        else:
            print(f"{field_label} ({field_type}): {field.value}")

generate_client async

generate_client(
    env_path: str | None = None,
    op_connect_host: str | None = None,
    op_connect_token: str | None = None,
) -> Client

Generate a 1Password Connect client from environment variables.

This function loads environment variables from a .env file and creates a 1Password Connect client using the OnePassword SDK.

Required environment variables: - OP_CONNECT_HOST: Your 1Password Connect server URL - OP_CONNECT_TOKEN: Your 1Password Connect API token

Parameters:

Name	Type	Description	Default
env_path	str | None	Path to the .env file (optional)	None
op_connect_host	str | None	Override for OP_CONNECT_HOST env var	None
op_connect_token	str | None	Override for OP_CONNECT_TOKEN env var	None

Returns:

Type	Description
Client	OnePassword Connect SDK client instance

Raises:

Type	Description
ImportError	If onepasswordconnectsdk is not installed
ValueError	If required credentials are missing
AssertionError	If the .env file cannot be loaded

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

async def generate_client(
    env_path: str | None = None,
    op_connect_host: str | None = None,
    op_connect_token: str | None = None,
) -> Client:
    """
    Generate a 1Password Connect client from environment variables.

    This function loads environment variables from a .env file and creates a 1Password
    Connect client using the OnePassword SDK.

    Required environment variables:
    - OP_CONNECT_HOST: Your 1Password Connect server URL
    - OP_CONNECT_TOKEN: Your 1Password Connect API token

    Args:
        env_path: Path to the .env file (optional)
        op_connect_host: Override for OP_CONNECT_HOST env var
        op_connect_token: Override for OP_CONNECT_TOKEN env var

    Returns:
        OnePassword Connect SDK client instance

    Raises:
        ImportError: If onepasswordconnectsdk is not installed
        ValueError: If required credentials are missing
        AssertionError: If the .env file cannot be loaded
    """

    if env_path:
        from dotenv import load_dotenv

        assert load_dotenv(env_path, override=True), (
            f"Failed to load environment variables from {env_path}"
        )

    if not op_connect_host:
        op_connect_host = os.getenv("OP_CONNECT_HOST")

    if not op_connect_token:
        op_connect_token = os.getenv("OP_CONNECT_TOKEN")

    if not op_connect_host:
        await logger.error("OP_CONNECT_HOST is not set in environment")
        raise ValueError("OP_CONNECT_HOST is not set in environment")

    if not op_connect_token:
        await logger.error("OP_CONNECT_TOKEN is not set in environment")
        raise ValueError("OP_CONNECT_TOKEN is not set in environment")

    client: Client = Client(op_connect_host, op_connect_token)

    await test_op_connection(client)

    return client

get_all_vault_items async

get_all_vault_items(
    client: Client, vault_id: str, debug_prn: bool = False
) -> list[Item]

List all items in the specified vault.

Parameters:

Name	Type	Description	Default
client	Client	1Password Connect client instance	required
vault_id	str	UUID of the vault	required
debug_prn	bool	If True, print items to console	False

Returns:

Type	Description
list[Item]	List of items in the vault

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

async def get_all_vault_items(
    client: Client, vault_id: str, debug_prn: bool = False
) -> list[Item]:
    """
    List all items in the specified vault.

    Args:
        client: 1Password Connect client instance
        vault_id: UUID of the vault
        debug_prn: If True, print items to console

    Returns:
        List of items in the vault
    """
    items = client.get_items(vault_id=vault_id)

    if debug_prn:
        print(f"Vault {vault_id} contains {len(items)} items:")
        print("-" * 50)

        for item in items:
            display_item(item, show_secrets=False)
            print("-" * 30)

    return items

get_auth_from_onepass async

get_auth_from_onepass(
    vault_id: str,
    item_title: str,
    domo_instance: str | None = None,
    client: Client | None = None,
    debug_api: bool = False,
) -> DomoTokenAuth | None

Retrieve Domo auth from 1Password.

This function retrieves a Domo access token from a 1Password item and creates a validated DomoTokenAuth instance.

Parameters:

Name	Type	Description	Default
vault_id	str	1Password vault ID	required
item_title	str	Title of the 1Password item (e.g., "sdk_my-instance")	required
domo_instance	str | None	Optional override for instance name (defaults to item_title without "sdk_" prefix)	None
client	Client | None	Optional existing 1Password client (will create one if not provided)	None
debug_api	bool	Enable debug output for API calls	False

Returns:

Type	Description
DomoTokenAuth | None	DomoTokenAuth if successful, None if credentials not found or invalid

Example

auth = await get_auth_from_onepass( vault_id="your-vault-id", item_title="sdk_my-instance" )

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

async def get_auth_from_onepass(
    vault_id: str,
    item_title: str,
    domo_instance: str | None = None,
    client: Client | None = None,
    debug_api: bool = False,
) -> dmda.DomoTokenAuth | None:
    """
    Retrieve Domo auth from 1Password.

    This function retrieves a Domo access token from a 1Password item and creates
    a validated DomoTokenAuth instance.

    Args:
        vault_id: 1Password vault ID
        item_title: Title of the 1Password item (e.g., "sdk_my-instance")
        domo_instance: Optional override for instance name (defaults to item_title without "sdk_" prefix)
        client: Optional existing 1Password client (will create one if not provided)
        debug_api: Enable debug output for API calls

    Returns:
        DomoTokenAuth if successful, None if credentials not found or invalid

    Example:
        auth = await get_auth_from_onepass(
            vault_id="your-vault-id",
            item_title="sdk_my-instance"
        )
    """

    await logger.debug(f"Attempting 1Password auth for {item_title}")

    try:
        client = client or await generate_client()
        op_cred = await get_item_by_title(
            client, vault_id, item_title, is_suppress_error=True
        )

        if not op_cred:
            await logger.debug(f"1Password item '{item_title}' not found in vault")
            return None

        await logger.debug(f"Found 1Password item for {item_title}")
        domo_access_token = _get_field_value(op_cred, field_purpose="password")
        domo_instance = domo_instance or item_title.replace("sdk_", "")

        if not domo_access_token or not domo_instance:
            await logger.warning("Domo token or instance not found in 1Password item")
            return None

        auth = dmda.DomoTokenAuth(
            domo_instance=domo_instance, domo_access_token=domo_access_token
        )

        is_valid = await auth.print_is_token(debug_api=debug_api)

        if not is_valid:
            await logger.warning(
                f"1Password token validation failed for {domo_instance}"
            )
            return None

        await logger.info(f"Successfully authenticated via 1Password ({domo_instance})")
        return auth

    except (KeyError, AssertionError, LookupError) as e:
        # Expected failures: missing env vars, connection issues, item not found
        await logger.debug(f"1Password auth failed (expected): {e}")
        raise e

get_item_by_title async

get_item_by_title(
    client: Client,
    vault_id: str,
    title: str,
    return_full_item: bool = True,
    is_suppress_error: bool = False,
) -> Item | None

Retrieve an item from 1Password vault by its title.

Parameters:

Name	Type	Description	Default
client	Client	1Password Connect client instance	required
vault_id	str	UUID of the vault to search	required
title	str	Title of the item to retrieve	required
return_full_item	bool	If True, fetch full item with all decrypted fields	True
is_suppress_error	bool	If True, return None instead of raising exception	False

Returns:

Type	Description
Item | None	The 1Password item if found, or None if is_suppress_error=True

Raises:

Type	Description
ValueError	If item not found and is_suppress_error=False

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

async def get_item_by_title(
    client: Client,
    vault_id: str,
    title: str,
    return_full_item: bool = True,
    is_suppress_error: bool = False,
) -> Item | None:
    """
    Retrieve an item from 1Password vault by its title.

    Args:
        client: 1Password Connect client instance
        vault_id: UUID of the vault to search
        title: Title of the item to retrieve
        return_full_item: If True, fetch full item with all decrypted fields
        is_suppress_error: If True, return None instead of raising exception

    Returns:
        The 1Password item if found, or None if is_suppress_error=True

    Raises:
        ValueError: If item not found and is_suppress_error=False
    """
    items = await get_all_vault_items(client, vault_id)

    try:
        basic_item = next(item for item in items if item.title == title)

    except StopIteration as e:
        if is_suppress_error:
            return None

        await logger.error(f"Item with title '{title}' not found in vault '{vault_id}'")
        raise ValueError(
            f"Item with title '{title}' not found in vault '{vault_id}'"
        ) from e

    if not return_full_item:
        return basic_item

    # Get the full item with all fields and decrypted values
    return client.get_item(basic_item.id, vault_id)

test_op_connection async

test_op_connection(client: Client) -> bool

Test connectivity to 1Password Connect server.

Parameters:

Name	Type	Description	Default
client	Client	1Password Connect client instance	required

Returns:

Type	Description
bool	True if connection successful

Raises:

Type	Description
gaierror	If DNS resolution fails
ConnectError	If unable to connect to server

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

async def test_op_connection(client: Client) -> bool:
    """
    Test connectivity to 1Password Connect server.

    Args:
        client: 1Password Connect client instance

    Returns:
        True if connection successful

    Raises:
        gaierror: If DNS resolution fails
        ConnectError: If unable to connect to server
    """
    try:
        client.get_vaults()
        await logger.info("Successfully connected to 1Password Connect server.")
        return True

    except (gaierror, ConnectError) as e:
        await logger.error(
            f"Network error occurred: {e} unable to access 1Password, are you connected to the VPN?"
        )
        raise e