feat(vault): add separate login and secret namespaces for HashiCorp Vault

litellm/proxy/_lazy_openapi_snapshot.json

@@ -7235,6 +7235,18 @@
               "description": "Certificate role name for TLS cert authentication",
               "title": "Vault Cert Role"
             },
+            "vault_login_namespace": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Namespace for AppRole and TLS cert login (X-Vault-Namespace header); falls back to vault_namespace",
+              "title": "Vault Login Namespace"
+            },
             "vault_mount_name": {
               "anyOf": [
                 {
@@ -7256,7 +7268,7 @@
                   "type": "null"
                 }
               ],
-              "description": "Vault namespace (for multi-tenant Vault, sent as X-Vault-Namespace header)",
+              "description": "Vault namespace used for both login and secret operations unless overridden below",
               "title": "Vault Namespace"
             },
             "vault_path_prefix": {
@@ -7271,6 +7283,18 @@
               "description": "Optional path prefix for secrets (e.g., myapp -> secret/data/myapp/{secret_name})",
               "title": "Vault Path Prefix"
             },
+            "vault_secret_namespace": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Namespace for secret reads and writes (URL path segment); falls back to vault_namespace",
+              "title": "Vault Secret Namespace"
+            },
             "vault_token": {
               "anyOf": [
                 {

litellm/proxy/management_endpoints/config_override_endpoints.py

@@ -3,6 +3,7 @@
 import os
 from collections.abc import Mapping, Sequence
 from datetime import datetime, timezone
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Final, Protocol
 
 from fastapi import APIRouter, Depends, Header, HTTPException
@@ -143,6 +144,8 @@ async def _emit_config_override_audit_log(
     "client_key": "HCP_VAULT_CLIENT_KEY",
     "vault_cert_role": "HCP_VAULT_CERT_ROLE",
     "vault_namespace": "HCP_VAULT_NAMESPACE",
+    "vault_login_namespace": "HCP_VAULT_LOGIN_NAMESPACE",
+    "vault_secret_namespace": "HCP_VAULT_SECRET_NAMESPACE",
     "vault_mount_name": "HCP_VAULT_MOUNT_NAME",
     "vault_path_prefix": "HCP_VAULT_PATH_PREFIX",
 }
@@ -627,9 +630,8 @@ async def test_hashicorp_vault_connection(
     try:
         async_client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.SecretManager)
         lookup_url: Final = f"{client.vault_addr}/v1/auth/token/lookup-self"
-        if client.vault_namespace:
-            headers["X-Vault-Namespace"] = client.vault_namespace
-        response: Final = await async_client.get(lookup_url, headers=headers)
+        lookup_headers: Final[Mapping[str, str]] = MappingProxyType({**headers, **client._get_login_headers()})
+        response: Final = await async_client.get(lookup_url, headers=lookup_headers)
         response.raise_for_status()
     except Exception as e:
         raise HTTPException(

litellm/secret_managers/hashicorp_secret_manager.py

@@ -1,5 +1,6 @@
 import os
 from collections.abc import Mapping
+from types import MappingProxyType
 from typing import Final, Protocol
 
 import httpx
@@ -85,15 +86,20 @@ def _json_object_body(response: _JsonObjectSource) -> dict[str, object]:
     return response.json()
 
 
+def _as_json_object(value: object) -> Mapping[str, object] | None:
+    return value if isinstance(value, Mapping) else None
+
+
 class HashicorpSecretManager(BaseSecretManager):
     def __init__(self):
         from litellm.proxy.proxy_server import CommonProxyErrors, premium_user
 
         # Vault-specific config
         self.vault_addr = os.getenv("HCP_VAULT_ADDR", "http://127.0.0.1:8200")
         self.vault_token = os.getenv("HCP_VAULT_TOKEN", "")
-        # Vault namespace (for X-Vault-Namespace header)
         self.vault_namespace = os.getenv("HCP_VAULT_NAMESPACE", None)
+        self.login_namespace_override = os.getenv("HCP_VAULT_LOGIN_NAMESPACE", None)
+        self.secret_namespace_override = os.getenv("HCP_VAULT_SECRET_NAMESPACE", None)
         # KV engine mount name (default: "secret")
         # If your KV engine is mounted somewhere other than "secret", set HCP_VAULT_MOUNT_NAME
         self.vault_mount_name = os.getenv("HCP_VAULT_MOUNT_NAME", "secret")
@@ -182,9 +188,7 @@ def _auth_via_approle(self) -> str:
         # Vault endpoint for AppRole login
         login_url: Final = f"{self.vault_addr}/v1/auth/{self.approle_mount_path}/login"
 
-        headers: Final = {}
-        if hasattr(self, "vault_namespace") and self.vault_namespace:
-            headers["X-Vault-Namespace"] = self.vault_namespace
+        headers: Final = self._get_login_headers()
 
         try:
             client: Final = _get_httpx_client()
@@ -245,12 +249,7 @@ def _auth_via_tls_cert(self) -> str:
         # Vault endpoint for cert-based login, e.g. '/v1/auth/cert/login'
         login_url: Final = f"{self.vault_addr}/v1/auth/cert/login"
 
-        # Include your Vault namespace in the header if you're using namespaces.
-        # E.g. self.vault_namespace = 'mynamespace/'
-        # If you only have root namespace, you can omit this header entirely.
-        headers: Final = {}
-        if hasattr(self, "vault_namespace") and self.vault_namespace:
-            headers["X-Vault-Namespace"] = self.vault_namespace
+        headers: Final = self._get_login_headers()
         try:
             # We use the client cert and key for mutual TLS
             client: Final = httpx.Client(cert=(self.tls_cert_path, self.tls_key_path))
@@ -273,6 +272,23 @@ def _auth_via_tls_cert(self) -> str:
     def _get_tls_cert_auth_body(self) -> dict:
         return {"name": self.vault_cert_role}
 
+    @property
+    def vault_login_namespace(self) -> str | None:
+        if self.login_namespace_override is not None:
+            return self.login_namespace_override
+        return self.vault_namespace
+
+    @property
+    def vault_secret_namespace(self) -> str | None:
+        if self.secret_namespace_override is not None:
+            return self.secret_namespace_override
+        return self.vault_namespace
+
+    def _get_login_headers(self) -> Mapping[str, str]:
+        if self.vault_login_namespace:
+            return MappingProxyType({"X-Vault-Namespace": self.vault_login_namespace})
+        return MappingProxyType({})
+
     def get_url(
         self,
         secret_name: str,
@@ -292,7 +308,9 @@ def get_url(
         - With path prefix: http://127.0.0.1:8200/v1/secret/data/myapp/mykey
         """
         raise_if_unsafe_secret_name(secret_name)
-        resolved_namespace = self._sanitize_path_component(namespace if namespace is not None else self.vault_namespace)
+        resolved_namespace = self._sanitize_path_component(
+            namespace if namespace is not None else self.vault_secret_namespace
+        )
         resolved_mount = self._sanitize_path_component(mount_name if mount_name is not None else self.vault_mount_name)
         if resolved_mount is None:
             resolved_mount = "secret"
@@ -336,7 +354,7 @@ def _extract_secret_manager_settings(self, optional_params: dict | None) -> dict
     def _build_secret_target(self, secret_name: str, optional_params: dict | None) -> _VaultSecretTarget:
         settings: Final = self._extract_secret_manager_settings(optional_params)
 
-        namespace: Final = settings.get("namespace", self.vault_namespace)
+        namespace: Final = settings.get("namespace", self.vault_secret_namespace)
         mount: Final = settings.get("mount", self.vault_mount_name)
         path_prefix: Final = settings.get("path_prefix", self.vault_path_prefix)
         data_key_override: Final = settings.get("data")
@@ -387,25 +405,21 @@ async def async_read_secret(
         secret_name is just the path inside the KV mount (e.g., 'myapp/config').
         Returns the entire data dict from data.data, or None on failure.
         """
-        if self.cache.get_cache(secret_name) is not None:
-            return self.cache.get_cache(secret_name)
         async_client: Final = get_async_httpx_client(
             llm_provider=httpxSpecialProvider.SecretManager,
         )
         try:
-            # For KV v2: /v1/<mount>/data/<path>
-            # Example: http://127.0.0.1:8200/v1/secret/data/myapp/config
-            _url: Final = self.get_url(secret_name)
-            url: Final = _url
+            target: Final = self._build_secret_target(secret_name, optional_params)
+            cached_body: Final = self.cache.get_cache(target["url"])
+            if cached_body is not None:
+                return self._get_secret_value_from_json_response(cached_body, target["data_key"])
 
-            response: Final = await async_client.get(url, headers=self._get_request_headers())
+            response: Final = await async_client.get(target["url"], headers=self._get_request_headers())
             response.raise_for_status()
 
-            # For KV v2, the secret is in response.json()["data"]["data"]
             json_resp: Final = _json_object_body(response)
-            _value: Final = self._get_secret_value_from_json_response(json_resp)
-            self.cache.set_cache(secret_name, _value)
-            return _value
+            self.cache.set_cache(target["url"], json_resp)
+            return self._get_secret_value_from_json_response(json_resp, target["data_key"])
 
         except Exception as e:
             verbose_logger.exception("Error reading secret from Hashicorp Vault: %s", e)
@@ -422,21 +436,19 @@ def sync_read_secret(
         secret_name is just the path inside the KV mount (e.g., 'myapp/config').
         Returns the entire data dict from data.data, or None on failure.
         """
-        if self.cache.get_cache(secret_name) is not None:
-            return self.cache.get_cache(secret_name)
         sync_client: Final = _get_httpx_client()
         try:
-            # For KV v2: /v1/<mount>/data/<path>
-            url: Final = self.get_url(secret_name)
+            target: Final = self._build_secret_target(secret_name, optional_params)
+            cached_body: Final = self.cache.get_cache(target["url"])
+            if cached_body is not None:
+                return self._get_secret_value_from_json_response(cached_body, target["data_key"])
 
-            response: Final = sync_client.get(url, headers=self._get_request_headers())
+            response: Final = sync_client.get(target["url"], headers=self._get_request_headers())
             response.raise_for_status()
 
-            # For KV v2, the secret is in response.json()["data"]["data"]
             json_resp: Final = _json_object_body(response)
-            _value: Final = self._get_secret_value_from_json_response(json_resp)
-            self.cache.set_cache(secret_name, _value)
-            return _value
+            self.cache.set_cache(target["url"], json_resp)
+            return self._get_secret_value_from_json_response(json_resp, target["data_key"])
 
         except Exception as e:
             verbose_logger.exception("Error reading secret from Hashicorp Vault: %s", e)
@@ -625,10 +637,10 @@ async def async_rotate_secret(
                     )
                 else:
                     # Clear cache for the old secret only if deletion was successful
-                    self.cache.delete_cache(current_secret_name)
+                    self.cache.delete_cache(current_target["url"])
 
             # Clear cache for the new secret (or updated secret if names are the same)
-            self.cache.delete_cache(new_secret_name)
+            self.cache.delete_cache(new_target["url"])
 
             return create_response
 
@@ -669,10 +681,7 @@ async def async_delete_secret(
             response: Final = await async_client.delete(url=target["url"], headers=self._get_request_headers())
             response.raise_for_status()
 
-            # Clear the cache for this secret
-            self.cache.delete_cache(secret_name)
-            if target["secret_name"] != secret_name:
-                self.cache.delete_cache(target["secret_name"])
+            self.cache.delete_cache(target["url"])
 
             return {
                 "status": "success",
@@ -682,7 +691,9 @@ async def async_delete_secret(
             verbose_logger.exception("Error deleting secret from Hashicorp Vault: %s", e)
             return {"status": "error", "message": str(e)}
 
-    def _get_secret_value_from_json_response(self, json_resp: dict | None) -> str | None:
+    def _get_secret_value_from_json_response(
+        self, json_resp: Mapping[str, object] | None, data_key: str = "key"
+    ) -> str | None:
         """
         Get the secret value from the JSON response
 
@@ -708,4 +719,11 @@ def _get_secret_value_from_json_response(self, json_resp: dict | None) -> str |
         """
         if json_resp is None:
             return None
-        return json_resp.get("data", {}).get("data", {}).get("key", None)
+        outer: Final = _as_json_object(json_resp.get("data"))
+        if outer is None:
+            return None
+        inner: Final = _as_json_object(outer.get("data"))
+        if inner is None:
+            return None
+        value: Final = inner.get(data_key)
+        return value if isinstance(value, str) else None

litellm/types/proxy/management_endpoints/config_overrides.py

@@ -40,7 +40,15 @@ class HashicorpVaultConfig(BaseModel):
     )
     vault_namespace: str | None = Field(
         default=None,
-        description="Vault namespace (for multi-tenant Vault, sent as X-Vault-Namespace header)",
+        description="Vault namespace used for both login and secret operations unless overridden below",
+    )
+    vault_login_namespace: str | None = Field(
+        default=None,
+        description="Namespace for AppRole and TLS cert login (X-Vault-Namespace header); falls back to vault_namespace",
+    )
+    vault_secret_namespace: str | None = Field(
+        default=None,
+        description="Namespace for secret reads and writes (URL path segment); falls back to vault_namespace",
     )
     vault_mount_name: str | None = Field(
         default=None,

tests/test_litellm/proxy/management_endpoints/test_config_override_endpoints.py

@@ -220,6 +220,65 @@ async def test_hashicorp_vault_crud_lifecycle(client, monkeypatch):
         _cleanup()
 
 
+@pytest.mark.asyncio
+async def test_hashicorp_vault_login_and_secret_namespaces(client, monkeypatch):
+    """POST maps the two namespace fields to their env vars; test_connection
+    validates the token in the login namespace, not the secret namespace."""
+    from litellm.secret_managers.hashicorp_secret_manager import HashicorpSecretManager
+
+    mock_prisma, mock_db = _make_mock_db()
+    mock_cfg = _make_mock_proxy_config()
+    monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+    monkeypatch.setattr(ps, "proxy_config", mock_cfg)
+    old_client, old_kms = litellm.secret_manager_client, litellm._key_management_system
+    _set_admin()
+
+    try:
+        r = client.post(
+            VAULT_URL,
+            json={
+                "vault_addr": "https://vault.example.com",
+                "vault_token": "tok",
+                "vault_login_namespace": "root",
+                "vault_secret_namespace": "teams/team-a",
+            },
+        )
+        assert r.status_code == 200
+        assert os.environ["HCP_VAULT_LOGIN_NAMESPACE"] == "root"
+        assert os.environ["HCP_VAULT_SECRET_NAMESPACE"] == "teams/team-a"
+        assert os.environ.get("HCP_VAULT_NAMESPACE") is None
+        data = _upserted_data(mock_db)
+        assert data["vault_login_namespace"] == "enc_root"
+        assert data["vault_secret_namespace"] == "enc_teams/team-a"
+
+        mock_manager = MagicMock(spec=HashicorpSecretManager)
+        mock_manager.vault_addr = "https://vault.example.com"
+        mock_manager.vault_login_namespace = "root"
+        mock_manager.vault_secret_namespace = "teams/team-a"
+        auth_headers = {"X-Vault-Token": "tok"}
+        mock_manager._get_request_headers = MagicMock(return_value=auth_headers)
+        mock_manager._get_login_headers = MagicMock(return_value={"X-Vault-Namespace": "root"})
+        litellm.secret_manager_client = mock_manager  # test-quality-ok: endpoint hot-reloads litellm globals; test must set and restore them
+
+        mock_response = MagicMock()
+        mock_response.raise_for_status = MagicMock()
+        mock_http = MagicMock()
+        mock_http.get = AsyncMock(return_value=mock_response)
+        with patch(  # test-quality-ok: patching proxy-internal collaborator to isolate the endpoint
+            "litellm.proxy.management_endpoints.config_override_endpoints.get_async_httpx_client",
+            return_value=mock_http,
+        ):
+            r = client.post(VAULT_URL + "/test_connection")
+        assert r.status_code == 200
+        assert mock_http.get.call_args.args[0] == "https://vault.example.com/v1/auth/token/lookup-self"
+        assert mock_http.get.call_args.kwargs["headers"] == {"X-Vault-Token": "tok", "X-Vault-Namespace": "root"}
+        assert auth_headers == {"X-Vault-Token": "tok"}
+    finally:
+        litellm.secret_manager_client = old_client  # test-quality-ok: endpoint hot-reloads litellm globals; test must set and restore them
+        litellm._key_management_system = old_kms  # test-quality-ok: endpoint hot-reloads litellm globals; test must set and restore them
+        _cleanup()
+
+
 @pytest.mark.asyncio
 async def test_hashicorp_vault_validation_errors_and_access_control(
     client, monkeypatch