Add HashiCorp Vault Manager

alberefe · alberefe · commit ebc99aa583b1 · 2025-11-17T22:38:46.000+01:00
Signed-off-by: Alberto Ferrer Sánchez &lt;alberefe@gmail.com&gt;
diff --git a/README.md b/README.md
@@ -174,6 +174,80 @@ _NOTE: the parameter "item_name" corresponds with the field "name" of the json.
 
 The module uses the [Bitwarden CLI](https://bitwarden.com/help/cli/) to interact with Bitwarden.
 
+### HashiCorp Vault
+
+```python
+from grimoirelab_toolkit.credential_manager.hc_manager import HashicorpManager
+
+
+# Instantiate the HashiCorp Vault manager using the vault URL and token
+# certificate can be a boolean (True/False) or a path to a CA bundle file
+hc_manager = HashicorpManager("https://vault.example.com", "your_token", certificate=True)
+
+# Retrieve a secret from HashiCorp Vault
+github_secret = hc_manager.get_secret("github")
+elasticsearch_secret = hc_manager.get_secret("elasticsearch")
+```
+
+#### Response format
+
+When calling `get_secret(item_name)`, the method returns a JSON object with the following structure:
+
+_NOTE: the parameter "item_name" corresponds to the secret path in HashiCorp Vault._
+
+##### Example Response
+
+```json
+{
+  "request_id": "d09e2bb5-00ee-576b-6078-5d291d35ccc3",
+  "lease_id": "",
+  "renewable": false,
+  "lease_duration": 0,
+  "data": {
+    "data": {
+      "username": "test_user",
+      "password": "test_pass",
+      "api_key": "test_key"
+    },
+    "metadata": {
+      "created_time": "2024-11-23T12:20:59.985132927Z",
+      "custom_metadata": null,
+      "deletion_time": "",
+      "destroyed": false,
+      "version": 1
+    }
+  },
+  "wrap_info": null,
+  "warnings": null,
+  "auth": null,
+  "mount_type": "kv"
+}
+```
+
+Field Descriptions
+
+- request_id: Unique identifier for this Vault request
+- lease_id: Lease identifier for renewable secrets (empty for KV secrets)
+- renewable: Boolean indicating if the secret is renewable
+- lease_duration: Lease duration in seconds (0 for KV secrets)
+- data: Main data object containing the secret
+  - data: The actual secret key-value pairs
+    - username: Username credential
+    - password: Password credential
+    - api_key: API key or other custom fields
+  - metadata: Vault metadata for this secret
+    - created_time: Secret creation timestamp (ISO 8601 format)
+    - custom_metadata: Custom metadata if configured
+    - deletion_time: Soft deletion timestamp (empty if not deleted)
+    - destroyed: Boolean indicating if secret version is destroyed
+    - version: Secret version number
+- wrap_info: Response wrapping information (null if not wrapped)
+- warnings: Array of warning messages (null if none)
+- auth: Authentication information (null for read operations)
+- mount_type: Type of secrets engine (typically "kv" for key-value)
+
+The module uses the [hvac](https://hvac.readthedocs.io/) Python library to interact with HashiCorp Vault.
+
 ## License
 
 Licensed under GNU General Public License (GPL), version 3 or later.
diff --git a/grimoirelab_toolkit/credential_manager/exceptions.py b/grimoirelab_toolkit/credential_manager/exceptions.py
@@ -26,6 +26,7 @@
     "InvalidCredentialsError",
     "CredentialNotFoundError",
     "BitwardenCLIError",
+    "HashicorpVaultError",
 ]
 
 
@@ -51,3 +52,9 @@ class BitwardenCLIError(CredentialManagerError):
     """Raised for Bitwarden CLI specific errors."""
 
     pass
+
+
+class HashicorpVaultError(CredentialManagerError):
+    """Raised for HashiCorp Vault-specific operation errors."""
+
+    pass
diff --git a/grimoirelab_toolkit/credential_manager/hc_manager.py b/grimoirelab_toolkit/credential_manager/hc_manager.py
@@ -0,0 +1,102 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) Grimoirelab Contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Author:
+#       Alberto Ferrer Sánchez (alberefe@gmail.com)
+#
+
+import logging
+import hvac
+import hvac.exceptions
+
+from .exceptions import (HashicorpVaultError,
+                         CredentialNotFoundError)
+
+logger = logging.getLogger(__name__)
+
+
+class HashicorpManager:
+    """Retrieve credentials from HashiCorp Vault.
+
+    This class defines functions to initialize a client and retrieve
+    secrets from HashiCorp Vault. The workflow is:
+
+    manager = HashicorpManager(vault_url, token, certificate)
+    manager.get_secret("github")
+    manager.get_secret("elasticsearch")
+
+    The manager initializes the client using the vault_url, token,
+    and certificate given as arguments when creating the instance,
+    so the object is reusable along the program.
+
+    The get_secret function returns the whole item object, with metadata
+    included, so the user can choose to store it and retrieve desired data.
+    """
+
+    def __init__(self, vault_url: str, token: str, certificate: str | bool = None):
+        """
+        Creates HashicorpManager object using token authentication
+
+        :param str vault_url: The URL of the vault
+        :param str token: The access token for authentication
+        :param Union[str, bool, None] certificate: TLS verification setting. Either a boolean to indicate whether TLS
+            verification should be performed, a string pointing at the CA bundle to use for
+            verification
+
+        :raises ConnectionError: If connection issues occur
+        """
+        try:
+            logger.debug("Creating Vault client")
+            # Initialize client with URL, token, and certificate verification setting
+            self.client = hvac.Client(url=vault_url, token=token, verify=certificate)
+            logger.debug("Vault client initialized successfully")
+        except Exception as e:
+            logger.error("An error occurred initializing the client: %s", str(e))
+            raise e
+
+    def get_secret(self, item_name: str) -> dict:
+        """Retrieve an item from the HashiCorp Vault.
+
+        Retrieves all the fields stored for an item with the name
+        provided as an argument and returns them as a dictionary.
+
+        The returned dictionary includes fields such as:
+        - data: The actual secret data and metadata
+        - request_id, lease_id, renewable, lease_duration
+        - Other vault metadata
+
+        :param str item_name: The name of the item to retrieve
+
+        :returns: Dictionary containing the secret data and metadata
+        :rtype: dict
+
+        :raises CredentialNotFoundError: If the secret path is not found
+        :raises HashicorpVaultError: If Vault operations fail
+        """
+        try:
+            logger.info("Retrieving credentials from vault: %s", item_name)
+            # Read secret from KV secrets engine
+            secret = self.client.secrets.kv.read_secret(path=item_name)
+            return secret
+        except hvac.exceptions.InvalidPath:
+            logger.error("The path %s does not exist in the vault", item_name)
+            raise CredentialNotFoundError(
+                f"Secret path '{item_name}' not found in Vault"
+            )
+        except Exception as e:
+            logger.error("Error retrieving the secret: %s", str(e))
+            raise HashicorpVaultError(f"Vault operation failed: {e}")
diff --git a/tests/test_hc_manager.py b/tests/test_hc_manager.py
@@ -0,0 +1,150 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) Grimoirelab Contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Author:
+#     Alberto Ferrer Sánchez (alberefe@gmail.com)
+#
+
+import unittest
+from unittest.mock import patch
+import hvac.exceptions
+
+from grimoirelab_toolkit.credential_manager.hc_manager import HashicorpManager
+from grimoirelab_toolkit.credential_manager.exceptions import (
+    CredentialNotFoundError,
+    HashicorpVaultError,
+)
+
+
+class TestHashicorpManager(unittest.TestCase):
+    """Tests for HashicorpManager class."""
+
+    def setUp(self):
+        """Set up common test fixtures."""
+        self.vault_url = "http://vault-url"
+        self.token = "test-token"
+        self.certificate = "test-certificate"
+
+        self.mock_secret_response = {
+            "auth": None,
+            "data": {
+                "data": {
+                    "password": "test_pass",
+                    "username": "test_user",
+                    "api_key": "test_key",
+                },
+                "metadata": {
+                    "created_time": "2024-11-23T12:20:59.985132927Z",
+                    "custom_metadata": None,
+                    "deletion_time": "",
+                    "destroyed": False,
+                    "version": 1,
+                },
+            },
+            "lease_duration": 0,
+            "lease_id": "",
+            "mount_type": "kv",
+            "renewable": False,
+            "request_id": "d09e2bb5-00ee-576b-6078-5d291d35ccc3",
+            "warnings": None,
+            "wrap_info": None,
+        }
+
+    @patch("hvac.Client")
+    def test_initialization_success(self, mock_hvac_client):
+        """Test successful initialization with valid credentials."""
+        mock_instance = mock_hvac_client.return_value
+
+        manager = HashicorpManager(self.vault_url, self.token, self.certificate)
+
+        self.assertIsNotNone(manager.client)
+        mock_hvac_client.assert_called_once_with(
+            url=self.vault_url, token=self.token, verify=self.certificate
+        )
+
+    @patch("hvac.Client")
+    def test_initialization_failure(self, mock_hvac_client):
+        """Test initialization fails when connection to Vault fails."""
+        mock_hvac_client.side_effect = hvac.exceptions.VaultError("Connection failed")
+
+        with self.assertRaises(hvac.exceptions.VaultError) as context:
+            HashicorpManager(self.vault_url, self.token, self.certificate)
+
+        self.assertIn("Connection failed", str(context.exception))
+
+    @patch("hvac.Client")
+    def test_get_secret_success(self, mock_hvac_client):
+        """Test successful secret retrieval."""
+        mock_instance = mock_hvac_client.return_value
+        mock_instance.secrets.kv.read_secret.return_value = self.mock_secret_response
+
+        manager = HashicorpManager(self.vault_url, self.token, self.certificate)
+        result = manager.get_secret("test_service")
+
+        # Verify it returns the full secret object
+        self.assertIsInstance(result, dict)
+        self.assertEqual(result, self.mock_secret_response)
+        self.assertEqual(result["data"]["data"]["api_key"], "test_key")
+        mock_instance.secrets.kv.read_secret.assert_called_once_with(
+            path="test_service"
+        )
+
+    @patch("hvac.Client")
+    def test_get_secret_not_found(self, mock_hvac_client):
+        """Test get_secret raises error when secret path not found."""
+        mock_instance = mock_hvac_client.return_value
+        mock_instance.secrets.kv.read_secret.side_effect = hvac.exceptions.InvalidPath()
+
+        manager = HashicorpManager(self.vault_url, self.token, self.certificate)
+
+        with self.assertRaises(CredentialNotFoundError) as context:
+            manager.get_secret("nonexistent_service")
+
+        self.assertIn("nonexistent_service", str(context.exception))
+        self.assertIn("not found", str(context.exception))
+
+    @patch("hvac.Client")
+    def test_get_secret_permission_denied(self, mock_hvac_client):
+        """Test get_secret raises error when access is forbidden."""
+        mock_instance = mock_hvac_client.return_value
+        mock_instance.secrets.kv.read_secret.side_effect = hvac.exceptions.Forbidden()
+
+        manager = HashicorpManager(self.vault_url, self.token, self.certificate)
+
+        with self.assertRaises(HashicorpVaultError) as context:
+            manager.get_secret("test_service")
+
+        self.assertIn("Vault operation failed", str(context.exception))
+
+    @patch("hvac.Client")
+    def test_vault_connection_error(self, mock_hvac_client):
+        """Test get_secret raises error when Vault is down or sealed."""
+        mock_instance = mock_hvac_client.return_value
+        mock_instance.secrets.kv.read_secret.side_effect = hvac.exceptions.VaultDown(
+            "Vault is sealed"
+        )
+
+        manager = HashicorpManager(self.vault_url, self.token, self.certificate)
+
+        with self.assertRaises(HashicorpVaultError) as context:
+            manager.get_secret("test_service")
+
+        self.assertIn("Vault operation failed", str(context.exception))
+
+
+if __name__ == "__main__":
+    unittest.main(warnings="ignore")
