|
1 | 1 | import json |
2 | 2 | import logging |
3 | 3 | import os |
4 | | -from functools import lru_cache |
| 4 | +from functools import cache |
| 5 | +from typing import Any |
5 | 6 |
|
6 | | -from bitwarden_sdk import BitwardenClient |
| 7 | +from pyhelper_utils.shell import run_command |
7 | 8 |
|
8 | 9 | from utilities.exceptions import MissingEnvironmentVariableError |
9 | 10 |
|
10 | 11 | LOGGER = logging.getLogger(__name__) |
11 | | -# Bitwarden SDK: https://github.com/bitwarden/sdk/blob/main/languages/python/bitwarden_sdk/bitwarden_client.py |
12 | 12 |
|
13 | 13 |
|
14 | | -def get_bitwarden_secrets_client(): |
15 | | - """ |
16 | | - Creates a BitwardenClient instance, logs in using ACCESS_TOKEN environment variable and bitwarden AuthClient |
17 | | - instance, and returns SecretsClient. To use bitwarden secret manager ACCESS_TOKEN and ORGANIZATION_ID environment |
18 | | - variables must be set |
| 14 | +def _run_bws_command(args: list[str]) -> Any: |
| 15 | + """Run bws CLI command and return parsed JSON output. |
| 16 | +
|
| 17 | + Args: |
| 18 | + args: Command arguments to pass to bws (e.g., ['secret', 'list']) |
19 | 19 |
|
20 | 20 | Returns: |
21 | | - SecretsClient: Returns SecretsClient instance to be used for secret manager calls |
| 21 | + Any: Parsed JSON response from bws CLI (can be list or dict depending on command) |
| 22 | +
|
| 23 | + Raises: |
| 24 | + MissingEnvironmentVariableError: If ACCESS_TOKEN not set |
22 | 25 | """ |
23 | | - if not (os.getenv("ACCESS_TOKEN") and os.getenv("ORGANIZATION_ID")): |
24 | | - raise MissingEnvironmentVariableError( |
25 | | - "Bitwarden client needs ORGANIZATION_ID and ACCESS_TOKEN environment variable set up" |
26 | | - ) |
27 | | - bitwarden_client = BitwardenClient() |
28 | | - bitwarden_client.auth().login_access_token(access_token=os.getenv("ACCESS_TOKEN")) |
29 | | - return bitwarden_client.secrets() |
| 26 | + access_token = os.getenv("ACCESS_TOKEN") |
30 | 27 |
|
| 28 | + if not access_token: |
| 29 | + raise MissingEnvironmentVariableError("Bitwarden client needs ACCESS_TOKEN environment variable set up") |
31 | 30 |
|
32 | | -@lru_cache |
33 | | -def get_all_cnv_tests_secrets(bitwarden_secrets_client): |
34 | | - """ |
35 | | - Using Bitwarden SecretsClient, gets a list of all cnv-secrets saved in bitwarden secret manager (associated with |
36 | | - a specific organization id). ORGANIZATION_ID is expected to set via environment variable. |
| 31 | + _, stdout, _ = run_command( |
| 32 | + command=["bws", "--access-token", access_token] + args, |
| 33 | + capture_output=True, |
| 34 | + check=True, |
| 35 | + hide_log_command=True, |
| 36 | + ) |
37 | 37 |
|
38 | | - Args: |
39 | | - bitwarden_secrets_client (SecretsClient): Bitwarden SecretsClient instance |
| 38 | + return json.loads(stdout) |
| 39 | + |
| 40 | + |
| 41 | +@cache |
| 42 | +def get_all_cnv_tests_secrets() -> dict[str, str]: |
| 43 | + """Gets a list of all cnv-secrets saved in Bitwarden Secret Manager. |
| 44 | +
|
| 45 | + Uses bws CLI to list all secrets associated with the organization. |
| 46 | + ACCESS_TOKEN environment variable must be set. |
40 | 47 |
|
41 | 48 | Returns: |
42 | | - dict: dictionary of secret name and secret uuid associated with the organization |
| 49 | + dict[str, str]: Dictionary mapping secret name to secret UUID |
43 | 50 | """ |
44 | | - secrets = bitwarden_secrets_client.list(organization_id=os.getenv("ORGANIZATION_ID")).data.data |
| 51 | + data = _run_bws_command(args=["secret", "list"]) |
| 52 | + |
45 | 53 | LOGGER.info(f"Cache info stats for pulling secrets: {get_all_cnv_tests_secrets.cache_info()}") |
46 | | - return {secret.key: secret.id for secret in secrets} |
47 | 54 |
|
| 55 | + return {secret["key"]: secret["id"] for secret in data} |
48 | 56 |
|
49 | | -@lru_cache |
50 | | -def get_cnv_tests_secret_by_name(secret_name): |
51 | | - """ |
52 | | - Pull a specific secret from bitwarden secret manager by name |
| 57 | + |
| 58 | +@cache |
| 59 | +def get_cnv_tests_secret_by_name(secret_name: str) -> dict[str, Any]: |
| 60 | + """Pull a specific secret from Bitwarden Secret Manager by name. |
53 | 61 |
|
54 | 62 | Args: |
55 | | - secret_name (str): Bitwarden secret manager secret name |
| 63 | + secret_name: Bitwarden Secret Manager secret name |
56 | 64 |
|
57 | 65 | Returns: |
58 | | - dict: value of the saved secret |
| 66 | + dict[str, Any]: Value of the saved secret (parsed from JSON) |
| 67 | +
|
| 68 | + Raises: |
| 69 | + ValueError: If secret is not found |
59 | 70 | """ |
60 | | - bitwarden_secrets_client = get_bitwarden_secrets_client() |
61 | | - secrets = get_all_cnv_tests_secrets(bitwarden_secrets_client=bitwarden_secrets_client) |
62 | | - secret_dict = None |
63 | | - for secret_key, secret_value in secrets.items(): |
64 | | - if secret_key == secret_name: |
65 | | - secret_dict = json.loads(bitwarden_secrets_client.get(id=secret_value).data.value) |
66 | | - break |
| 71 | + secrets = get_all_cnv_tests_secrets() |
| 72 | + |
| 73 | + secret_id = secrets.get(secret_name) |
| 74 | + if not secret_id: |
| 75 | + raise ValueError(f"Secret '{secret_name}' not found in Bitwarden") |
| 76 | + |
| 77 | + secret_data = _run_bws_command(args=["secret", "get", secret_id]) |
| 78 | + secret_value = secret_data.get("value", "") |
| 79 | + |
| 80 | + secret_dict = json.loads(secret_value) |
67 | 81 | LOGGER.info(f"Cache info stats for getting specific secret: {get_cnv_tests_secret_by_name.cache_info()}") |
68 | | - assert secret_dict, f"secret {secret_name} is either not found or does not have valid values." |
69 | 82 | return secret_dict |
0 commit comments