community.general/plugins/lookup/bitwarden_secrets_manager.py

# Copyright (c) 2023, jantari (https://github.com/jantari)
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations


DOCUMENTATION = r"""
name: bitwarden_secrets_manager
author:
  - jantari (@jantari)
requirements:
  - bws (command line utility)
short_description: Retrieve secrets from Bitwarden Secrets Manager
version_added: 7.2.0
description:
  - Retrieve secrets from Bitwarden Secrets Manager.
options:
  _terms:
    description: Secret ID(s) to fetch values for.
    required: true
    type: list
    elements: str
  bws_access_token:
    description: The BWS access token to use for this lookup.
    env:
      - name: BWS_ACCESS_TOKEN
    required: true
    type: str
"""

EXAMPLES = r"""
- name: Get a secret relying on the BWS_ACCESS_TOKEN environment variable for authentication
  ansible.builtin.debug:
    msg: >-
      {{ lookup("community.general.bitwarden_secrets_manager", "2bc23e48-4932-40de-a047-5524b7ddc972") }}

- name: Get a secret passing an explicit access token for authentication
  ansible.builtin.debug:
    msg: >-
      {{
        lookup(
          "community.general.bitwarden_secrets_manager",
          "2bc23e48-4932-40de-a047-5524b7ddc972",
          bws_access_token="9.4f570d14-4b54-42f5-bc07-60f4450b1db5.YmluYXJ5LXNvbWV0aGluZy0xMjMK:d2h5IGhlbGxvIHRoZXJlCg=="
        )
      }}

- name: Get two different secrets each using a different access token for authentication
  ansible.builtin.debug:
    msg:
      - '{{ lookup("community.general.bitwarden_secrets_manager", "2bc23e48-4932-40de-a047-5524b7ddc972", bws_access_token=token1) }}'
      - '{{ lookup("community.general.bitwarden_secrets_manager", "9d89af4c-eb5d-41f5-bb0f-4ae81215c768", bws_access_token=token2) }}'
  vars:
    token1: "9.4f570d14-4b54-42f5-bc07-60f4450b1db5.YmluYXJ5LXNvbWV0aGluZy0xMjMK:d2h5IGhlbGxvIHRoZXJlCg=="
    token2: "1.69b72797-6ea9-4687-a11e-848e41a30ae6.YW5zaWJsZSBpcyBncmVhdD8K:YW5zaWJsZSBpcyBncmVhdAo="

- name: Get just the value of a secret
  ansible.builtin.debug:
    msg: >-
      {{ lookup("community.general.bitwarden_secrets_manager", "2bc23e48-4932-40de-a047-5524b7ddc972").value }}
"""

RETURN = r"""
_raw:
  description: List containing one or more secrets.
  type: list
  elements: dict
"""

from subprocess import Popen, PIPE
from time import sleep

from ansible.errors import AnsibleLookupError
from ansible.module_utils.common.text.converters import to_text
from ansible.parsing.ajson import AnsibleJSONDecoder
from ansible.plugins.lookup import LookupBase

from ansible_collections.community.general.plugins.module_utils.version import LooseVersion


class BitwardenSecretsManagerException(AnsibleLookupError):
    pass


class BitwardenSecretsManager(object):
    def __init__(self, path='bws'):
        self._cli_path = path
        self._max_retries = 3
        self._retry_delay = 1

    @property
    def cli_path(self):
        return self._cli_path

    def _run_with_retry(self, args, stdin=None, retries=0):
        out, err, rc = self._run(args, stdin)

        if rc != 0:
            if retries >= self._max_retries:
                raise BitwardenSecretsManagerException("Max retries exceeded. Unable to retrieve secret.")

            if "Too many requests" in err:
                delay = self._retry_delay * (2 ** retries)
                sleep(delay)
                return self._run_with_retry(args, stdin, retries + 1)
            else:
                raise BitwardenSecretsManagerException(f"Command failed with return code {rc}: {err}")

        return out, err, rc

    def _run(self, args, stdin=None):
        p = Popen([self.cli_path] + args, stdout=PIPE, stderr=PIPE, stdin=PIPE)
        out, err = p.communicate(stdin)
        rc = p.wait()
        return to_text(out, errors='surrogate_or_strict'), to_text(err, errors='surrogate_or_strict'), rc

    def get_bws_version(self):
        """Get the version of the Bitwarden Secrets Manager CLI.
        """
        out, err, rc = self._run(['--version'])
        if rc != 0:
            raise BitwardenSecretsManagerException(to_text(err))
        # strip the prefix and grab the last segment, the version number
        return out.split()[-1]

    def get_secret(self, secret_id, bws_access_token):
        """Get and return the secret with the given secret_id.
        """

        # Prepare set of params for Bitwarden Secrets Manager CLI
        # Color output was not always disabled correctly with the default 'auto' setting so explicitly disable it.
        params = [
            '--color', 'no',
            '--access-token', bws_access_token
        ]

        # bws version 0.3.0 introduced a breaking change in the command line syntax:
        # pre-0.3.0: verb noun
        # 0.3.0 and later: noun verb
        bws_version = self.get_bws_version()
        if LooseVersion(bws_version) < LooseVersion('0.3.0'):
            params.extend(['get', 'secret', secret_id])
        else:
            params.extend(['secret', 'get', secret_id])

        out, err, rc = self._run_with_retry(params)
        if rc != 0:
            raise BitwardenSecretsManagerException(to_text(err))

        return AnsibleJSONDecoder().raw_decode(out)[0]


class LookupModule(LookupBase):
    def run(self, terms, variables=None, **kwargs):
        self.set_options(var_options=variables, direct=kwargs)
        bws_access_token = self.get_option('bws_access_token')

        return [_bitwarden_secrets_manager.get_secret(term, bws_access_token) for term in terms]


_bitwarden_secrets_manager = BitwardenSecretsManager()