refactor(lookup): update lookup plugins to use DVLSLookupHelper

Author: dion-gionet

plugins/lookup/secret.py

@@ -1,3 +1,6 @@
+# GNU General Public License v3.0+ (see LICENSE-GPL-3 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
@@ -22,7 +25,8 @@
     description:
       - Field name to extract from credential.
       - Supported fields depend on credential type.
-      - Common fields include username, password, domain, connectionString, apiId, apiKey, tenantId, clientId, clientSecret, privateKeyData, publicKeyData, privateKeyPassPhrase.
+      - Common fields include username, password, domain, connectionString, apiId, apiKey, tenantId, clientId, clientSecret, privateKeyData,
+        publicKeyData, privateKeyPassPhrase.
     type: str
     default: password
   server_base_url:
@@ -43,7 +47,6 @@
       - Falls back to DVLS_APP_SECRET environment variable if not provided.
     type: str
     required: false
-    no_log: true
   vault_id:
     description:
       - Vault UUID containing the credential.
@@ -53,7 +56,8 @@
 notes:
   - Requires network access to DVLS server.
   - Authentication token is cached for the duration of the playbook run.
-  - Supported fields include username, password, domain, connectionString, apiId, apiKey, tenantId, clientId, clientSecret, privateKeyData, publicKeyData, privateKeyPassPhrase.
+  - Supported fields include username, password, domain, connectionString, apiId, apiKey, tenantId, clientId, clientSecret, privateKeyData,
+    publicKeyData, privateKeyPassPhrase.
 """
 
 EXAMPLES = r"""
@@ -105,11 +109,12 @@
 """
 
 from ansible.errors import AnsibleError
+from ansible.plugins.lookup import LookupBase
 from ansible.utils.display import Display
 
 try:
     from ansible_collections.devolutions.dvls.plugins.module_utils.lookup_base import (
-        BaseDVLSLookup,
+        DVLSLookupHelper,
         SUPPORTED_CREDENTIAL_FIELDS,
     )
 except ImportError as e:
@@ -118,10 +123,45 @@
 display = Display()
 
 
-class LookupModule(BaseDVLSLookup):
+class LookupModule(LookupBase):
     """Lookup plugin to retrieve a specific field from a DVLS credential."""
 
-    def transform_result(self, credential, term):
+    def __init__(self, *args, **kwargs):
+        super(LookupModule, self).__init__(*args, **kwargs)
+        self._helper = DVLSLookupHelper(display, AnsibleError)
+
+    def run(self, terms, variables=None, **kwargs):
+        """
+        Main lookup execution method.
+
+        Handles authentication, credential retrieval, and result transformation.
+        """
+        self.set_options(var_options=variables, direct=kwargs)
+
+        config = self._helper.get_config(self.get_option, variables)
+        self._helper.authenticate(
+            config["server_base_url"], config["app_key"], config["app_secret"]
+        )
+
+        results = []
+        for term in terms:
+            try:
+                credential = self._helper.get_credential(
+                    config["server_base_url"], config["vault_id"], term
+                )
+                result = self._transform_result(credential, term)
+                results.append(result)
+
+            except AnsibleError:
+                raise
+            except Exception as e:
+                raise AnsibleError(
+                    f"Failed to retrieve credential '{term}': {e}"
+                ) from e
+
+        return results
+
+    def _transform_result(self, credential, term):
         """
         Extract a specific field from the credential.
 
@@ -136,7 +176,8 @@ def transform_result(self, credential, term):
 
         if field not in SUPPORTED_CREDENTIAL_FIELDS:
             raise AnsibleError(
-                f"Invalid field '{field}'. Supported fields: {', '.join(sorted(SUPPORTED_CREDENTIAL_FIELDS))}"
+                f"Invalid field '{field}'. "
+                f"Supported fields: {', '.join(sorted(SUPPORTED_CREDENTIAL_FIELDS))}"
             )
 
         field_value = credential.get(field)

plugins/lookup/secret_full.py

@@ -1,3 +1,6 @@
+# GNU General Public License v3.0+ (see LICENSE-GPL-3 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
@@ -37,7 +40,6 @@
       - Falls back to DVLS_APP_SECRET environment variable if not provided.
     type: str
     required: false
-    no_log: true
   vault_id:
     description:
       - Vault UUID containing the credential.
@@ -109,31 +111,53 @@
 """
 
 from ansible.errors import AnsibleError
+from ansible.plugins.lookup import LookupBase
 from ansible.utils.display import Display
 
 try:
     from ansible_collections.devolutions.dvls.plugins.module_utils.lookup_base import (
-        BaseDVLSLookup,
+        DVLSLookupHelper,
     )
 except ImportError as e:
     raise AnsibleError(f"Failed to import DVLS module_utils: {e}")
 
 display = Display()
 
 
-class LookupModule(BaseDVLSLookup):
+class LookupModule(LookupBase):
     """Lookup plugin to retrieve complete DVLS credential objects."""
 
-    def transform_result(self, credential, term):
-        """
-        Return the complete credential object.
+    def __init__(self, *args, **kwargs):
+        super(LookupModule, self).__init__(*args, **kwargs)
+        self._helper = DVLSLookupHelper(display, AnsibleError)
 
-        Args:
-            credential: Complete credential object from DVLS
-            term: Original lookup term
+    def run(self, terms, variables=None, **kwargs):
+        """
+        Main lookup execution method.
 
-        Returns:
-            dict: The complete credential object
+        Handles authentication, credential retrieval, and result transformation.
         """
-        display.vvv(f"Successfully retrieved credential '{term}'")
-        return credential
+        self.set_options(var_options=variables, direct=kwargs)
+
+        config = self._helper.get_config(self.get_option, variables)
+        self._helper.authenticate(
+            config["server_base_url"], config["app_key"], config["app_secret"]
+        )
+
+        results = []
+        for term in terms:
+            try:
+                credential = self._helper.get_credential(
+                    config["server_base_url"], config["vault_id"], term
+                )
+                display.vvv(f"Successfully retrieved credential '{term}'")
+                results.append(credential)
+
+            except AnsibleError:
+                raise
+            except Exception as e:
+                raise AnsibleError(
+                    f"Failed to retrieve credential '{term}': {e}"
+                ) from e
+
+        return results

plugins/module_utils/lookup_base.py

@@ -6,10 +6,6 @@
 import os
 import re
 
-from ansible.errors import AnsibleError
-from ansible.plugins.lookup import LookupBase
-from ansible.utils.display import Display
-
 try:
     from ansible_collections.devolutions.dvls.plugins.module_utils.auth import (
         login,
@@ -19,10 +15,9 @@
         get_vault_entry,
         get_vault_entry_from_name,
     )
-except ImportError as e:
-    raise AnsibleError(f"Failed to import DVLS module_utils: {e}")
-
-display = Display()
+except ImportError:
+    # Will be caught by lookup plugins
+    pass
 
 UUID_PATTERN = re.compile(
     r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
@@ -45,39 +40,47 @@
 }
 
 
-class BaseDVLSLookup(LookupBase):
-    """Base class for DVLS lookup plugins with shared authentication and retrieval logic."""
+class DVLSLookupHelper:
+    """Helper class for DVLS lookup plugins with shared authentication and retrieval logic."""
+
+    def __init__(self, display_instance, ansible_error_class):
+        """
+        Initialize the helper.
 
-    def __init__(self, *args, **kwargs):
-        super(BaseDVLSLookup, self).__init__(*args, **kwargs)
+        Args:
+            display_instance: Display instance for logging
+            ansible_error_class: AnsibleError class for raising exceptions
+        """
         self._token = None
         self._server_base_url = None
         self._cleanup_registered = False
+        self._display = display_instance
+        self._ansible_error = ansible_error_class
 
     def _cleanup(self):
         """Cleanup method called at interpreter exit"""
         if self._token and self._server_base_url:
             try:
-                display.vvv("Logging out from DVLS")
+                self._display.vvv("Logging out from DVLS")
                 logout(self._server_base_url, self._token)
             except Exception as e:
-                display.warning(f"Failed to logout from DVLS during cleanup: {e}")
+                self._display.warning(f"Failed to logout from DVLS during cleanup: {e}")
 
     def _is_uuid(self, value):
         """Check if a string matches UUID format"""
         return UUID_PATTERN.match(value) is not None
 
-    def _get_config(self, variables):
+    def get_config(self, get_option, variables):
         """Extract configuration from options and environment variables"""
-        server_base_url = self.get_option("server_base_url") or os.environ.get(
+        server_base_url = get_option("server_base_url") or os.environ.get(
             "DVLS_SERVER_BASE_URL"
         )
-        app_key = self.get_option("app_key") or os.environ.get("DVLS_APP_KEY")
-        app_secret = self.get_option("app_secret") or os.environ.get("DVLS_APP_SECRET")
-        vault_id = self.get_option("vault_id") or os.environ.get("DVLS_VAULT_ID")
+        app_key = get_option("app_key") or os.environ.get("DVLS_APP_KEY")
+        app_secret = get_option("app_secret") or os.environ.get("DVLS_APP_SECRET")
+        vault_id = get_option("vault_id") or os.environ.get("DVLS_VAULT_ID")
 
         if not all([server_base_url, app_key, app_secret, vault_id]):
-            raise AnsibleError(
+            raise self._ansible_error(
                 "Missing required configuration. Set DVLS_SERVER_BASE_URL, DVLS_APP_KEY, "
                 "DVLS_APP_SECRET, and DVLS_VAULT_ID environment variables or pass as parameters."
             )
@@ -89,21 +92,21 @@ def _get_config(self, variables):
             "vault_id": vault_id,
         }
 
-    def _authenticate(self, server_base_url, app_key, app_secret):
+    def authenticate(self, server_base_url, app_key, app_secret):
         """Authenticate to DVLS and cache the token"""
         if not self._token:
             try:
-                display.vvv(f"Authenticating to DVLS at {server_base_url}")
+                self._display.vvv(f"Authenticating to DVLS at {server_base_url}")
                 self._token = login(server_base_url, app_key, app_secret)
                 self._server_base_url = server_base_url
 
                 if not self._cleanup_registered:
                     atexit.register(self._cleanup)
                     self._cleanup_registered = True
             except Exception as e:
-                raise AnsibleError(f"DVLS authentication failed: {e}") from e
+                raise self._ansible_error(f"DVLS authentication failed: {e}") from e
 
-    def _get_credential(self, server_base_url, vault_id, term):
+    def get_credential(self, server_base_url, vault_id, term):
         """
         Retrieve a credential from DVLS by name or UUID.
 
@@ -115,70 +118,26 @@ def _get_credential(self, server_base_url, vault_id, term):
         Returns:
             dict: Complete credential object
         """
-        display.vvv(f"Looking up credential: {term}")
+        self._display.vvv(f"Looking up credential: {term}")
 
         if self._is_uuid(term):
-            display.vvv(f"Using ID lookup for {term}")
+            self._display.vvv(f"Using ID lookup for {term}")
             response = get_vault_entry(server_base_url, self._token, vault_id, term)
             credential = response.get("data", {})
         else:
-            display.vvv(f"Using name lookup for {term}")
+            self._display.vvv(f"Using name lookup for {term}")
             response = get_vault_entry_from_name(
                 server_base_url, self._token, vault_id, term
             )
             entries = response.get("data", [])
             if not entries:
-                raise AnsibleError(f"Credential '{term}' not found in vault {vault_id}")
+                raise self._ansible_error(
+                    f"Credential '{term}' not found in vault {vault_id}"
+                )
             entry_id = entries[0].get("id")
             full_response = get_vault_entry(
                 server_base_url, self._token, vault_id, entry_id
             )
             credential = full_response.get("data", {})
 
         return credential
-
-    def transform_result(self, credential, term):
-        """
-        Transform the credential into the desired output format.
-
-        This method must be implemented by subclasses.
-
-        Args:
-            credential: Complete credential object from DVLS
-            term: Original lookup term
-
-        Returns:
-            The transformed result (field value, full object, etc.)
-        """
-        raise NotImplementedError("Subclasses must implement transform_result()")
-
-    def run(self, terms, variables=None, **kwargs):
-        """
-        Main lookup execution method.
-
-        Handles authentication, credential retrieval, and result transformation.
-        """
-        self.set_options(var_options=variables, direct=kwargs)
-
-        config = self._get_config(variables)
-        self._authenticate(
-            config["server_base_url"], config["app_key"], config["app_secret"]
-        )
-
-        results = []
-        for term in terms:
-            try:
-                credential = self._get_credential(
-                    config["server_base_url"], config["vault_id"], term
-                )
-                result = self.transform_result(credential, term)
-                results.append(result)
-
-            except AnsibleError:
-                raise
-            except Exception as e:
-                raise AnsibleError(
-                    f"Failed to retrieve credential '{term}': {e}"
-                ) from e
-
-        return results