New lookup plug-in: Bitwarden Secrets Manager (#6389)

* add Bitwarden Secrets Manager lookup

* fix pep8 and yamllint complaints

* fix version_added, add maintainer and copyright notice

* document BWS_ACCESS_TOKEN env var and declare as required

* avoid returning nested list

* update 'value of a secret' example after f6c4492c

* update copyright notice in bitwarden_secrets_manager plugin

thx felixfontein

Co-authored-by: Felix Fontein <felix@fontein.de>

* rename classes to distinguish from existing bw plugin

* use AnsibleLookupError, formatting

* bump version_added to 7.0.0

Co-authored-by: Felix Fontein <felix@fontein.de>

* ci fix: python style guide calls for excessive blank lines

https://peps.python.org/pep-0008/#blank-lines

* first attempt at unit tests for bws lookup

* ci fix: remove trailing newline

* attempt to fix tests object not callable error

* address formatting, tests and pyright suggestions

* reduce scope of mocked code for more real test coverage

only the actual bws CLI call is mocked now, this should enable the
exception thrown test to succeed if I didn't add new problems

* fix undefined variable 'expected_rc'

* fix mocked _run method to return correct data types

* keep list of one element for test case comparison

* bump version_added to 7.2.0

Co-authored-by: Felix Fontein <felix@fontein.de>

---------

Co-authored-by: jantari <jantari@github.com>
Co-authored-by: Felix Fontein <felix@fontein.de>

.github/BOTMETA.yml

@@ -204,6 +204,8 @@ files:
     maintainers: ddelnano shinuza
   $lookups/:
     labels: lookups
+  $lookups/bitwarden_secrets_manager.py:
+    maintainers: jantari
   $lookups/bitwarden.py:
     maintainers: lungj
   $lookups/cartesian.py: {}

plugins/lookup/bitwarden_secrets_manager.py

@@ -0,0 +1,125 @@
+# -*- coding: utf-8 -*-
+# 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 (absolute_import, division, print_function)
+
+__metaclass__ = type
+
+DOCUMENTATION = """
+    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 = """
+- 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 = """
+  _raw:
+    description: List containing one or more secrets.
+    type: list
+    elements: dict
+"""
+
+from subprocess import Popen, PIPE
+
+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
+
+
+class BitwardenSecretsManagerException(AnsibleLookupError):
+    pass
+
+
+class BitwardenSecretsManager(object):
+    def __init__(self, path='bws'):
+        self._cli_path = path
+
+    @property
+    def cli_path(self):
+        return self._cli_path
+
+    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_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,
+            'get', 'secret', secret_id
+        ]
+
+        out, err, rc = self._run(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()

tests/unit/plugins/lookup/test_bitwarden_secrets_manager.py

@@ -0,0 +1,83 @@
+# -*- coding: utf-8 -*-
+# 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 (absolute_import, division, print_function)
+
+__metaclass__ = type
+
+import json
+from ansible_collections.community.general.tests.unit.compat import unittest
+from ansible_collections.community.general.tests.unit.compat.mock import patch
+
+from ansible.errors import AnsibleLookupError
+from ansible.plugins.loader import lookup_loader
+from ansible_collections.community.general.plugins.lookup.bitwarden_secrets_manager import BitwardenSecretsManager
+
+
+MOCK_SECRETS = [
+    {
+        "object": "secret",
+        "id": "ababc4a8-c242-4e54-bceb-77d17cdf2e07",
+        "organizationId": "3c33066c-a0bf-4e70-9a3c-24cda6aaddd5",
+        "projectId": "81869439-bfe5-442f-8b4e-b172e68b0ab2",
+        "key": "TEST_SECRET",
+        "value": "1234supersecret5678",
+        "note": "A test secret to use when developing the ansible bitwarden_secrets_manager lookup plugin",
+        "creationDate": "2023-04-23T13:13:37.7507017Z",
+        "revisionDate": "2023-04-23T13:13:37.7507017Z"
+    },
+    {
+        "object": "secret",
+        "id": "d4b7c8fa-fc95-40d7-a13c-6e186ee69d53",
+        "organizationId": "3c33066c-a0bf-4e70-9a3c-24cda6aaddd5",
+        "projectId": "81869439-bfe5-442f-8b4e-b172e68b0ab2",
+        "key": "TEST_SECRET_2",
+        "value": "abcd_such_secret_very_important_efgh",
+        "note": "notes go here",
+        "creationDate": "2023-04-23T13:26:44.0392906Z",
+        "revisionDate": "2023-04-23T13:26:44.0392906Z"
+    }
+]
+
+
+class MockBitwardenSecretsManager(BitwardenSecretsManager):
+
+    def _run(self, args, stdin=None):
+        # secret_id is the last argument passed to the bws CLI
+        secret_id = args[-1]
+        rc = 1
+        out = ""
+        err = ""
+        found_secrets = list(filter(lambda record: record["id"] == secret_id, MOCK_SECRETS))
+
+        if len(found_secrets) == 0:
+            err = "simulated bws CLI error: 404 no secret with such id"
+        elif len(found_secrets) == 1:
+            rc = 0
+            # The real bws CLI will only ever return one secret / json object for the "get secret <secret-id>" command
+            out = json.dumps(found_secrets[0])
+        else:
+            # This should never happen unless there's an error in the test MOCK_SECRETS.
+            # The real Bitwarden Secrets Manager assigns each secret a unique ID.
+            raise ValueError("More than 1 secret found with id: '{0}'. Impossible!".format(secret_id))
+
+        return out, err, rc
+
+
+class TestLookupModule(unittest.TestCase):
+
+    def setUp(self):
+        self.lookup = lookup_loader.get('community.general.bitwarden_secrets_manager')
+
+    @patch('ansible_collections.community.general.plugins.lookup.bitwarden_secrets_manager._bitwarden_secrets_manager', new=MockBitwardenSecretsManager())
+    def test_bitwarden_secrets_manager(self):
+        # Getting a secret by its id should return the full secret info
+        self.assertEqual([MOCK_SECRETS[0]], self.lookup.run(['ababc4a8-c242-4e54-bceb-77d17cdf2e07'], bws_access_token='123'))
+
+    @patch('ansible_collections.community.general.plugins.lookup.bitwarden_secrets_manager._bitwarden_secrets_manager', new=MockBitwardenSecretsManager())
+    def test_bitwarden_secrets_manager_no_match(self):
+        # Getting a nonexistant secret id throws exception
+        with self.assertRaises(AnsibleLookupError):
+            self.lookup.run(['nonexistant_id'], bws_access_token='123')