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

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yassin 2026-09-17 01:41:48 +00:00
parent 351a54e849
commit 15bfe8f28a
10 changed files with 398 additions and 47 deletions

View file

@ -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": [
{

View file

@ -3,6 +3,7 @@ import json
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 @@ HASHICORP_ENV_VAR_MAPPING: Final[dict[str, str]] = {
"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(

View file

@ -1,5 +1,6 @@
import os
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final, Protocol
import httpx
@ -92,8 +93,9 @@ class HashicorpSecretManager(BaseSecretManager):
# 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 +184,7 @@ class HashicorpSecretManager(BaseSecretManager):
# 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 +245,7 @@ class HashicorpSecretManager(BaseSecretManager):
# 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 +268,23 @@ class HashicorpSecretManager(BaseSecretManager):
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 +304,9 @@ class HashicorpSecretManager(BaseSecretManager):
- 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 +350,7 @@ class HashicorpSecretManager(BaseSecretManager):
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,24 +401,21 @@ class HashicorpSecretManager(BaseSecretManager):
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_value: Final = self.cache.get_cache(target["url"])
if cached_value is not None:
return cached_value
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)
_value: Final = self._get_secret_value_from_json_response(json_resp, target["data_key"])
self.cache.set_cache(target["url"], _value)
return _value
except Exception as e:
@ -422,20 +433,19 @@ class HashicorpSecretManager(BaseSecretManager):
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_value: Final = self.cache.get_cache(target["url"])
if cached_value is not None:
return cached_value
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)
_value: Final = self._get_secret_value_from_json_response(json_resp, target["data_key"])
self.cache.set_cache(target["url"], _value)
return _value
except Exception as e:
@ -625,10 +635,10 @@ class HashicorpSecretManager(BaseSecretManager):
)
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 +679,7 @@ class HashicorpSecretManager(BaseSecretManager):
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 +689,7 @@ class HashicorpSecretManager(BaseSecretManager):
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: dict | None, data_key: str = "key") -> str | None:
"""
Get the secret value from the JSON response
@ -708,4 +715,4 @@ class HashicorpSecretManager(BaseSecretManager):
"""
if json_resp is None:
return None
return json_resp.get("data", {}).get("data", {}).get("key", None)
return json_resp.get("data", {}).get("data", {}).get(data_key, None)

View file

@ -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,

View file

@ -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

View file

@ -0,0 +1,208 @@
import datetime
from collections.abc import Mapping
from pathlib import Path
from typing import Final
import pytest
import respx
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
import litellm.proxy.proxy_server
from litellm.secret_managers.hashicorp_secret_manager import HashicorpSecretManager
VAULT_ADDR: Final = "http://vault.test:8200"
LOGIN_RESPONSE: Final = {"auth": {"client_token": "hvs.login-token", "lease_duration": 3600}}
SECRET_RESPONSE: Final = {"data": {"data": {"key": "sk-from-vault", "password": "pw-from-vault"}}}
NAMESPACE_ENV_VARS: Final = ("HCP_VAULT_NAMESPACE", "HCP_VAULT_LOGIN_NAMESPACE", "HCP_VAULT_SECRET_NAMESPACE")
def _build_manager(monkeypatch: pytest.MonkeyPatch, env: Mapping[str, str]) -> HashicorpSecretManager:
monkeypatch.setattr(litellm.proxy.proxy_server, "premium_user", True)
for name in NAMESPACE_ENV_VARS:
monkeypatch.delenv(name, raising=False)
monkeypatch.setenv("HCP_VAULT_ADDR", VAULT_ADDR)
monkeypatch.setenv("HCP_VAULT_APPROLE_ROLE_ID", "role-id")
monkeypatch.setenv("HCP_VAULT_APPROLE_SECRET_ID", "secret-id")
for name, value in env.items():
monkeypatch.setenv(name, value)
return HashicorpSecretManager()
@pytest.mark.parametrize(
("env", "expected_login_namespace", "expected_secret_namespace"),
[
({"HCP_VAULT_LOGIN_NAMESPACE": "root", "HCP_VAULT_SECRET_NAMESPACE": "teams/team-a"}, "root", "teams/team-a"),
({"HCP_VAULT_NAMESPACE": "admin"}, "admin", "admin"),
({"HCP_VAULT_NAMESPACE": "admin", "HCP_VAULT_LOGIN_NAMESPACE": "root"}, "root", "admin"),
({"HCP_VAULT_NAMESPACE": "admin", "HCP_VAULT_SECRET_NAMESPACE": "teams/team-a"}, "admin", "teams/team-a"),
],
)
@respx.mock
def test_sync_read_uses_login_namespace_for_approle_and_secret_namespace_for_url(
monkeypatch: pytest.MonkeyPatch,
env: Mapping[str, str],
expected_login_namespace: str,
expected_secret_namespace: str,
) -> None:
manager: Final = _build_manager(monkeypatch, env)
login_route: Final = respx.post(f"{VAULT_ADDR}/v1/auth/approle/login").respond(json=LOGIN_RESPONSE)
read_route: Final = respx.get(f"{VAULT_ADDR}/v1/{expected_secret_namespace}/secret/data/OPENAI_API_KEY").respond(
json=SECRET_RESPONSE
)
assert manager.sync_read_secret("OPENAI_API_KEY") == "sk-from-vault"
assert login_route.call_count == 1
assert login_route.calls.last.request.headers["X-Vault-Namespace"] == expected_login_namespace
assert read_route.call_count == 1
read_request: Final = read_route.calls.last.request
assert read_request.headers["X-Vault-Token"] == "hvs.login-token"
assert "X-Vault-Namespace" not in read_request.headers
@respx.mock
def test_login_header_is_omitted_when_no_namespace_is_configured(monkeypatch: pytest.MonkeyPatch) -> None:
manager: Final = _build_manager(monkeypatch, {})
login_route: Final = respx.post(f"{VAULT_ADDR}/v1/auth/approle/login").respond(json=LOGIN_RESPONSE)
read_route: Final = respx.get(f"{VAULT_ADDR}/v1/secret/data/OPENAI_API_KEY").respond(json=SECRET_RESPONSE)
assert manager.sync_read_secret("OPENAI_API_KEY") == "sk-from-vault"
assert "X-Vault-Namespace" not in login_route.calls.last.request.headers
assert read_route.call_count == 1
@respx.mock
def test_sync_read_per_secret_namespace_overrides_secret_namespace(monkeypatch: pytest.MonkeyPatch) -> None:
manager: Final = _build_manager(
monkeypatch, {"HCP_VAULT_LOGIN_NAMESPACE": "root", "HCP_VAULT_SECRET_NAMESPACE": "teams/team-a"}
)
login_route: Final = respx.post(f"{VAULT_ADDR}/v1/auth/approle/login").respond(json=LOGIN_RESPONSE)
read_route: Final = respx.get(f"{VAULT_ADDR}/v1/teams/team-b/kv-prod/data/virtual-keys/DB_PASSWORD").respond(
json=SECRET_RESPONSE
)
optional_params: Final = {
"secret_manager_settings": {
"namespace": "teams/team-b",
"mount": "kv-prod",
"path_prefix": "virtual-keys",
"data": "password",
}
}
assert manager.sync_read_secret("DB_PASSWORD", optional_params=optional_params) == "pw-from-vault"
assert login_route.calls.last.request.headers["X-Vault-Namespace"] == "root"
assert read_route.call_count == 1
@respx.mock
def test_sync_read_caches_per_resolved_target(monkeypatch: pytest.MonkeyPatch) -> None:
manager: Final = _build_manager(monkeypatch, {"HCP_VAULT_SECRET_NAMESPACE": "teams/team-a"})
respx.post(f"{VAULT_ADDR}/v1/auth/approle/login").respond(json=LOGIN_RESPONSE)
team_a_route: Final = respx.get(f"{VAULT_ADDR}/v1/teams/team-a/secret/data/SHARED").respond(
json={"data": {"data": {"key": "team-a-value"}}}
)
team_b_route: Final = respx.get(f"{VAULT_ADDR}/v1/teams/team-b/secret/data/SHARED").respond(
json={"data": {"data": {"key": "team-b-value"}}}
)
team_b_params: Final = {"secret_manager_settings": {"namespace": "teams/team-b"}}
assert manager.sync_read_secret("SHARED") == "team-a-value"
assert manager.sync_read_secret("SHARED", optional_params=team_b_params) == "team-b-value"
assert manager.sync_read_secret("SHARED") == "team-a-value"
assert team_a_route.call_count == 1
assert team_b_route.call_count == 1
@pytest.mark.asyncio
@respx.mock
async def test_async_read_uses_secret_namespace_and_login_namespace(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
manager: Final = _build_manager(
monkeypatch, {"HCP_VAULT_LOGIN_NAMESPACE": "root", "HCP_VAULT_SECRET_NAMESPACE": "teams/team-a"}
)
login_route: Final = respx.post(f"{VAULT_ADDR}/v1/auth/approle/login").respond(json=LOGIN_RESPONSE)
read_route: Final = respx.get(f"{VAULT_ADDR}/v1/teams/team-a/secret/data/OPENAI_API_KEY").respond(
json=SECRET_RESPONSE
)
assert await manager.async_read_secret("OPENAI_API_KEY") == "sk-from-vault"
assert login_route.calls.last.request.headers["X-Vault-Namespace"] == "root"
assert read_route.call_count == 1
assert "X-Vault-Namespace" not in read_route.calls.last.request.headers
@pytest.mark.asyncio
@respx.mock
async def test_async_write_and_read_share_the_secret_namespace_target(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
manager: Final = _build_manager(
monkeypatch, {"HCP_VAULT_LOGIN_NAMESPACE": "root", "HCP_VAULT_SECRET_NAMESPACE": "teams/team-a"}
)
respx.post(f"{VAULT_ADDR}/v1/auth/approle/login").respond(json=LOGIN_RESPONSE)
write_route: Final = respx.post(f"{VAULT_ADDR}/v1/teams/team-a/secret/data/VIRTUAL_KEY").respond(
json={"data": {"version": 1}}
)
read_route: Final = respx.get(f"{VAULT_ADDR}/v1/teams/team-a/secret/data/VIRTUAL_KEY").respond(
json={"data": {"data": {"key": "sk-virtual"}}}
)
await manager.async_write_secret("VIRTUAL_KEY", "sk-virtual")
assert await manager.async_read_secret("VIRTUAL_KEY") == "sk-virtual"
assert write_route.call_count == 1
assert read_route.call_count == 1
def _write_self_signed_cert(directory: Path) -> tuple[Path, Path]:
private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
name: Final = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "litellm-test")])
now: Final = datetime.datetime.now(datetime.timezone.utc)
certificate: Final = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.public_key(private_key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now)
.not_valid_after(now + datetime.timedelta(days=1))
.sign(private_key, hashes.SHA256())
)
cert_path: Final = directory / "client.crt"
key_path: Final = directory / "client.key"
cert_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM))
key_path.write_bytes(
private_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
)
return cert_path, key_path
@respx.mock
def test_tls_login_uses_login_namespace(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
cert, key = _write_self_signed_cert(tmp_path)
monkeypatch.setattr(litellm.proxy.proxy_server, "premium_user", True)
for name in NAMESPACE_ENV_VARS:
monkeypatch.delenv(name, raising=False)
monkeypatch.delenv("HCP_VAULT_APPROLE_ROLE_ID", raising=False)
monkeypatch.delenv("HCP_VAULT_APPROLE_SECRET_ID", raising=False)
monkeypatch.setenv("HCP_VAULT_ADDR", VAULT_ADDR)
monkeypatch.setenv("HCP_VAULT_CLIENT_CERT", str(cert))
monkeypatch.setenv("HCP_VAULT_CLIENT_KEY", str(key))
monkeypatch.setenv("HCP_VAULT_NAMESPACE", "admin")
monkeypatch.setenv("HCP_VAULT_LOGIN_NAMESPACE", "root")
manager: Final = HashicorpSecretManager()
login_route: Final = respx.post(f"{VAULT_ADDR}/v1/auth/cert/login").respond(json=LOGIN_RESPONSE)
assert manager._auth_via_tls_cert() == "hvs.login-token"
assert login_route.calls.last.request.headers["X-Vault-Namespace"] == "root"

View file

@ -26,6 +26,8 @@ vi.mock("@/lib/toast", () => ({
const ALL_FIELDS = [
"vault_addr",
"vault_namespace",
"vault_login_namespace",
"vault_secret_namespace",
"vault_mount_name",
"vault_path_prefix",
"vault_token",
@ -84,16 +86,19 @@ describe("EditHashicorpVaultModal", () => {
await waitFor(() => {
expect(mutate).toHaveBeenCalledTimes(1);
});
expect(mutate.mock.calls[0][0]).toEqual({
const expectedPayload = {
vault_addr: "https://vault.example.com",
vault_namespace: "team-ns",
vault_login_namespace: "",
vault_secret_namespace: "",
vault_mount_name: "",
vault_path_prefix: "",
approle_role_id: "",
approle_mount_path: "",
client_cert: "",
vault_cert_role: "",
});
};
expect(mutate.mock.calls[0][0]).toEqual(expectedPayload);
});
it("sends a sensitive field only once it is typed into", async () => {
@ -110,6 +115,25 @@ describe("EditHashicorpVaultModal", () => {
expect(mutate.mock.calls[0][0]).toMatchObject({ vault_token: "rotated-token" });
});
it("sends the login and secret namespaces the admin types in", async () => {
setup({ values: { vault_addr: "https://vault.example.com", vault_namespace: "root" } });
const user = userEvent.setup();
renderModal();
fireEvent.change(screen.getByLabelText("Login Namespace"), { target: { value: "root" } });
fireEvent.change(screen.getByLabelText("Secret Namespace"), { target: { value: "teams/team-a" } });
await save(user);
await waitFor(() => {
expect(mutate).toHaveBeenCalledTimes(1);
});
expect(mutate.mock.calls[0][0]).toMatchObject({
vault_namespace: "root",
vault_login_namespace: "root",
vault_secret_namespace: "teams/team-a",
});
});
it("never seeds a stored secret into its input", () => {
setup({ values: { vault_token: "super-secret-token", approle_secret_id: "super-secret-id" } });
renderModal();

View file

@ -26,7 +26,14 @@ interface VaultFieldGroup {
const FIELD_GROUPS: VaultFieldGroup[] = [
{
title: "Connection",
fields: ["vault_addr", "vault_namespace", "vault_mount_name", "vault_path_prefix"],
fields: [
"vault_addr",
"vault_namespace",
"vault_login_namespace",
"vault_secret_namespace",
"vault_mount_name",
"vault_path_prefix",
],
},
{
title: "Token Authentication",

View file

@ -3,6 +3,8 @@ export const SENSITIVE_FIELDS = new Set(["vault_token", "approle_secret_id", "cl
export const FIELD_LABELS: Record<string, string> = {
vault_addr: "Vault Address",
vault_namespace: "Namespace",
vault_login_namespace: "Login Namespace",
vault_secret_namespace: "Secret Namespace",
vault_mount_name: "KV Mount Name",
vault_path_prefix: "Path Prefix",
vault_token: "Token",

View file

@ -28799,6 +28799,11 @@ export interface components {
* @description Certificate role name for TLS cert authentication
*/
vault_cert_role?: string | null;
/**
* Vault Login Namespace
* @description Namespace for AppRole and TLS cert login (X-Vault-Namespace header); falls back to vault_namespace
*/
vault_login_namespace?: string | null;
/**
* Vault Mount Name
* @description KV engine mount name (default: secret)
@ -28806,7 +28811,7 @@ export interface components {
vault_mount_name?: string | null;
/**
* Vault Namespace
* @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_namespace?: string | null;
/**
@ -28814,6 +28819,11 @@ export interface components {
* @description Optional path prefix for secrets (e.g., myapp -> secret/data/myapp/{secret_name})
*/
vault_path_prefix?: string | null;
/**
* Vault Secret Namespace
* @description Namespace for secret reads and writes (URL path segment); falls back to vault_namespace
*/
vault_secret_namespace?: string | null;
/**
* Vault Token
* @description Token for Vault token-based authentication