[PR #5851/7b8b73f1 backport][stable-6] Add support to Bitwarden Lookup for filtering results by collection (#5849) (#5904)

Add support to Bitwarden Lookup for filtering results by collection (#5849) (#5851)

* Add support to Bitwarden Lookup for filtering results by collection id (#5849)

* Debug

* Add support to Bitwarden Lookup for filtering results by collection id (#5849)

* Update comments

* Fix blank line issue

* Fix unit tests for bitwarden lookup plugin. Add changelog fragment file.

* Change collectionId to collection_id parameter on bitwarden plugin

* Fix collection id parameter name when used in bw cli

(cherry picked from commit 7b8b73f17f)

Co-authored-by: Piotr <skc.peter@gmail.com>

changelogs/fragments/5851-lookup-bitwarden-add-filter-by-collection-id-parameter.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - bitwarden lookup plugin - implement filtering results by ``collection_id`` parameter (https://github.com/ansible-collections/community.general/issues/5849).

plugins/lookup/bitwarden.py

@@ -28,8 +28,12 @@ DOCUMENTATION = """
         default: name
         version_added: 5.7.0
       field:
-        description: Field to fetch; leave unset to fetch whole response.
+        description: Field to fetch. Leave unset to fetch whole response.
         type: str
+      collection_id:
+        description: Collection ID to filter results by collection. Leave unset to skip filtering.
+        type: str
+        version_added: 6.3.0
 """
 
 EXAMPLES = """
@@ -43,6 +47,11 @@ EXAMPLES = """
     msg: >-
       {{ lookup('community.general.bitwarden', 'bafba515-af11-47e6-abe3-af1200cd18b2', search='id', field='password') }}
 
+- name: "Get 'password' from Bitwarden record named 'a_test' from collection"
+  ansible.builtin.debug:
+    msg: >-
+      {{ lookup('community.general.bitwarden', 'a_test', field='password', collection_id='bafba515-af11-47e6-abe3-af1200cd18b2') }}
+
 - name: "Get full Bitwarden record named 'a_test'"
   ansible.builtin.debug:
     msg: >-
@@ -96,10 +105,17 @@ class Bitwarden(object):
             raise BitwardenException(err)
         return to_text(out, errors='surrogate_or_strict'), to_text(err, errors='surrogate_or_strict')
 
-    def _get_matches(self, search_value, search_field):
+    def _get_matches(self, search_value, search_field, collection_id):
         """Return matching records whose search_field is equal to key.
         """
-        out, err = self._run(['list', 'items', '--search', search_value])
+
+        # Prepare set of params for Bitwarden CLI
+        params = ['list', 'items', '--search', search_value]
+
+        if collection_id:
+            params.extend(['--collectionid', collection_id])
+
+        out, err = self._run(params)
 
         # This includes things that matched in different fields.
         initial_matches = AnsibleJSONDecoder().raw_decode(out)[0]
@@ -107,12 +123,13 @@ class Bitwarden(object):
         # Filter to only include results from the right field.
         return [item for item in initial_matches if item[search_field] == search_value]
 
-    def get_field(self, field, search_value, search_field="name"):
-        """Return a list of the specified field for records whose search_field match search_value.
+    def get_field(self, field, search_value, search_field="name", collection_id=None):
+        """Return a list of the specified field for records whose search_field match search_value
+        and filtered by collection if collection has been provided.
 
         If field is None, return the whole record for each match.
         """
-        matches = self._get_matches(search_value, search_field)
+        matches = self._get_matches(search_value, search_field, collection_id)
 
         if field in ['autofillOnPageLoad', 'password', 'passwordRevisionDate', 'totp', 'uris', 'username']:
             return [match['login'][field] for match in matches]
@@ -135,10 +152,11 @@ class LookupModule(LookupBase):
         self.set_options(var_options=variables, direct=kwargs)
         field = self.get_option('field')
         search_field = self.get_option('search')
+        collection_id = self.get_option('collection_id')
         if not _bitwarden.unlocked:
             raise AnsibleError("Bitwarden Vault locked. Run 'bw unlock'.")
 
-        return [_bitwarden.get_field(field, term, search_field) for term in terms]
+        return [_bitwarden.get_field(field, term, search_field, collection_id) for term in terms]
 
 
 _bitwarden = Bitwarden()

tests/unit/plugins/lookup/test_bitwarden.py

@@ -113,7 +113,7 @@ class MockBitwarden(Bitwarden):
 
     unlocked = True
 
-    def _get_matches(self, search_value, search_field="name"):
+    def _get_matches(self, search_value, search_field="name", collection_id=None):
         return list(filter(lambda record: record[search_field] == search_value, MOCK_RECORDS))