(Feat) - Hashicorp secret manager, use TLS cert authentication (#7532)

* fix - don't print hcorp secrets in debug logs

* hcorp - tls auth fixes

* fix tls_ca_cert_path

* test_hashicorp_secret_manager_tls_cert_auth

* hcp secret docs

docs/my-website/docs/proxy/config_settings.md

@@ -391,6 +391,8 @@ router_settings:
 | GOOGLE_KMS_RESOURCE_NAME | Name of the resource in Google KMS
 | HF_API_BASE | Base URL for Hugging Face API
 | HCP_VAULT_ADDR | Address for [Hashicorp Vault Secret Manager](../secret.md#hashicorp-vault)
+| HCP_VAULT_CLIENT_CERT | Path to client certificate for [Hashicorp Vault Secret Manager](../secret.md#hashicorp-vault)
+| HCP_VAULT_CLIENT_KEY | Path to client key for [Hashicorp Vault Secret Manager](../secret.md#hashicorp-vault)
 | HCP_VAULT_NAMESPACE | Namespace for [Hashicorp Vault Secret Manager](../secret.md#hashicorp-vault)
 | HCP_VAULT_TOKEN | Token for [Hashicorp Vault Secret Manager](../secret.md#hashicorp-vault)
 | HELICONE_API_KEY | API key for Helicone service

docs/my-website/docs/secret.md

@@ -246,11 +246,23 @@ Read secrets from [Hashicorp Vault](https://developer.hashicorp.com/vault/docs/s
 
 Step 1. Add Hashicorp Vault details in your environment
 
+LiteLLM supports two methods of authentication:
+
+1. TLS cert authentication - `HCP_VAULT_CLIENT_CERT` and `HCP_VAULT_CLIENT_KEY`
+2. Token authentication - `HCP_VAULT_TOKEN`
+
 ```bash
 HCP_VAULT_ADDR="https://test-cluster-public-vault-0f98180c.e98296b2.z1.hashicorp.cloud:8200"
 HCP_VAULT_NAMESPACE="admin"
+
+# Authentication via TLS cert
+HCP_VAULT_CLIENT_CERT="path/to/client.pem"
+HCP_VAULT_CLIENT_KEY="path/to/client.key"
+
+# OR - Authentication via token
 HCP_VAULT_TOKEN="hvs.CAESIG52gL6ljBSdmq*****"
 
+
 # OPTIONAL
 HCP_VAULT_REFRESH_INTERVAL="86400" # defaults to 86400, frequency of cache refresh for Hashicorp Vault
 ```

litellm/secret_managers/hashicorp_secret_manager.py

@@ -1,6 +1,8 @@
 import os
 from typing import Optional
 
+import httpx
+
 import litellm
 from litellm._logging import verbose_logger
 from litellm.caching import InMemoryCache
@@ -22,6 +24,10 @@ class HashicorpSecretManager:
         # If your KV engine is mounted somewhere other than "secret", adjust here:
         self.vault_namespace = os.getenv("HCP_VAULT_NAMESPACE", None)
 
+        # Optional config for TLS cert auth
+        self.tls_cert_path = os.getenv("HCP_VAULT_CLIENT_CERT", "")
+        self.tls_key_path = os.getenv("HCP_VAULT_CLIENT_KEY", "")
+
         # Validate environment
         if not self.vault_token:
             raise ValueError(
@@ -41,6 +47,57 @@ class HashicorpSecretManager:
                 f"Hashicorp secret manager is only available for premium users. {CommonProxyErrors.not_premium_user.value}"
             )
 
+    def _auth_via_tls_cert(self) -> str:
+        """
+        Ref: https://developer.hashicorp.com/vault/api-docs/auth/cert
+
+        Request:
+        ```
+        curl \
+            --request POST \
+            --cacert vault-ca.pem \
+            --cert cert.pem \
+            --key key.pem \
+            --data @payload.json \
+            https://127.0.0.1:8200/v1/auth/cert/login
+
+        ```
+
+        Response:
+        ```
+        {
+            "auth": {
+                "client_token": "cf95f87d-f95b-47ff-b1f5-ba7bff850425",
+                "policies": ["web", "stage"],
+                "lease_duration": 3600,
+                "renewable": true
+            }
+        }
+
+        ```
+        """
+        verbose_logger.debug("Using TLS cert auth for Hashicorp Vault")
+
+        # Vault endpoint for cert-based login, e.g. '/v1/auth/cert/login'
+        login_url = f"{self.vault_addr}/v1/auth/cert/login"
+
+        try:
+            # We use the client cert and key for mutual TLS
+            resp = httpx.post(
+                login_url,
+                cert=(self.tls_cert_path, self.tls_key_path),
+            )
+            resp.raise_for_status()
+            token = resp.json()["auth"]["client_token"]
+            _lease_duration = resp.json()["auth"]["lease_duration"]
+            verbose_logger.info("Successfully obtained Vault token via TLS cert auth.")
+            self.cache.set_cache(
+                key="hcp_vault_token", value=token, ttl=_lease_duration
+            )
+            return token
+        except Exception as e:
+            raise RuntimeError(f"Could not authenticate to Vault via TLS cert: {e}")
+
     def get_url(self, secret_name: str) -> str:
         _url = f"{self.vault_addr}/v1/"
         if self.vault_namespace:
@@ -48,6 +105,11 @@ class HashicorpSecretManager:
         _url += f"secret/data/{secret_name}"
         return _url
 
+    def _get_request_headers(self) -> dict:
+        if self.tls_cert_path and self.tls_key_path:
+            return {"X-Vault-Token": self._auth_via_tls_cert()}
+        return {"X-Vault-Token": self.vault_token}
+
     async def async_read_secret(self, secret_name: str) -> Optional[str]:
         """
         Reads a secret from Vault KV v2 using an async HTTPX client.
@@ -65,9 +127,7 @@ class HashicorpSecretManager:
             _url = self.get_url(secret_name)
             url = _url
 
-            response = await async_client.get(
+            response = await async_client.get(url, headers=self._get_request_headers())
-                url, headers={"X-Vault-Token": self.vault_token}
-            )
             response.raise_for_status()
 
             # For KV v2, the secret is in response.json()["data"]["data"]
@@ -93,7 +153,7 @@ class HashicorpSecretManager:
             # For KV v2: /v1/<mount>/data/<path>
             url = self.get_url(secret_name)
 
-            response = sync_client.get(url, headers={"X-Vault-Token": self.vault_token})
+            response = sync_client.get(url, headers=self._get_request_headers())
             response.raise_for_status()
 
             # For KV v2, the secret is in response.json()["data"]["data"]

tests/secret_manager_tests/test_hashicorp.py

@@ -65,3 +65,38 @@ def test_hashicorp_secret_manager_get_secret():
             == "https://test-cluster-public-vault-0f98180c.e98296b2.z1.hashicorp.cloud:8200/v1/admin/secret/data/sample-secret-mock"
         )
         assert "X-Vault-Token" in mock_get.call_args.kwargs["headers"]
+
+
+def test_hashicorp_secret_manager_tls_cert_auth():
+    with patch("httpx.post") as mock_post:
+        # Configure the mock response for TLS auth
+        mock_auth_response = MagicMock()
+        mock_auth_response.json.return_value = {
+            "auth": {
+                "client_token": "test-client-token-12345",
+                "lease_duration": 3600,
+                "renewable": True,
+            }
+        }
+        mock_auth_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_auth_response
+
+        # Create a new instance with TLS cert config
+        test_manager = HashicorpSecretManager()
+        test_manager.tls_cert_path = "cert.pem"
+        test_manager.tls_key_path = "key.pem"
+
+        # Test the TLS auth method
+        token = test_manager._auth_via_tls_cert()
+
+        # Verify the token and request parameters
+        assert token == "test-client-token-12345"
+        mock_post.assert_called_once_with(
+            f"{test_manager.vault_addr}/v1/auth/cert/login",
+            cert=("cert.pem", "key.pem"),
+        )
+
+        # Verify the token was cached
+        assert (
+            test_manager.cache.get_cache("hcp_vault_token") == "test-client-token-12345"
+        )