bitwarden.py

import json
import logging
import os
from functools import cache
from typing import Any

from _pytest.main import Session
from pyhelper_utils.shell import run_command
from timeout_sampler import retry

from utilities.exceptions import MissingEnvironmentVariableError

LOGGER = logging.getLogger(__name__)


def _run_bws_command(args: list[str]) -> Any:
    """Run bws CLI command and return parsed JSON output.

    Validates ACCESS_TOKEN then delegates to _run_bws_cli which retries
    on transient failures for up to 60 seconds.

    Args:
        args: Command arguments to pass to bws (e.g., ['secret', 'list'])

    Returns:
        Any: Parsed JSON response from bws CLI (can be list or dict depending on command)

    Raises:
        MissingEnvironmentVariableError: If ACCESS_TOKEN not set
        TimeoutExpiredError: If bws CLI repeatedly fails within the retry window
    """
    access_token = os.getenv("ACCESS_TOKEN")

    if not access_token:
        raise MissingEnvironmentVariableError("Bitwarden client needs ACCESS_TOKEN environment variable set up")

    return json.loads(_run_bws_cli(access_token=access_token, args=args))


@retry(wait_timeout=60, sleep=10)
def _run_bws_cli(access_token: str, args: list[str]) -> str:
    """Execute bws CLI command with retry on transient failures.

    Retries on any exception for up to 60 seconds with 10-second intervals.
    Returns raw JSON string (always truthy) so the retry decorator does not
    discard valid falsy parsed values like [] or {}.

    Args:
        access_token: Bitwarden access token
        args: Command arguments to pass to bws

    Returns:
        str: Raw JSON response string from bws CLI

    Raises:
        TimeoutExpiredError: If bws CLI repeatedly fails within the retry window
    """
    # hide_log_command=True disables check (CalledProcessError not raised),
    # so success must be checked explicitly below.
    success, stdout, stderr = run_command(
        command=["bws", "--access-token", access_token, *args],
        hide_log_command=True,
    )

    if not success:
        raise RuntimeError(f"bws command failed: {stderr}")

    json.loads(stdout)  # Validate JSON before returning
    return stdout


@cache
def get_all_cnv_tests_secrets() -> dict[str, str]:
    """Gets a list of all cnv-secrets saved in Bitwarden Secret Manager.

    Uses bws CLI to list all secrets associated with the organization.
    ACCESS_TOKEN environment variable must be set.

    Returns:
        dict[str, str]: Dictionary mapping secret name to secret UUID
    """
    data = _run_bws_command(args=["secret", "list"])

    LOGGER.info(f"Cache info stats for pulling secrets: {get_all_cnv_tests_secrets.cache_info()}")

    return {secret["key"]: secret["id"] for secret in data}


@cache
def get_cnv_tests_secret_by_name(secret_name: str, session: Session | None = None) -> dict[str, Any]:
    """Pull a specific secret from Bitwarden Secret Manager by name.

    Args:
        secret_name: Bitwarden Secret Manager secret name
        session: Pytest session object

    Returns:
        dict[str, Any]: Value of the saved secret (parsed from JSON) or empty dict if user passed `--disabled-bitwarden`

    Raises:
        ValueError: If secret is not found
    """
    if session and session.config.getoption("--disabled-bitwarden"):
        LOGGER.info("`--disabled-bitwarden` is set; skipping Bitwarden access.")
        return {}

    secrets = get_all_cnv_tests_secrets()

    secret_id = secrets.get(secret_name)
    if not secret_id:
        raise ValueError(f"Secret '{secret_name}' not found in Bitwarden")

    secret_data = _run_bws_command(args=["secret", "get", secret_id])
    secret_value = secret_data.get("value", "")

    secret_dict = json.loads(secret_value)
    LOGGER.info(f"Cache info stats for getting specific secret: {get_cnv_tests_secret_by_name.cache_info()}")
    return secret_dict