feat(secret_managers): add read-only Keeper Secrets Manager backend

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/proxy_server.py

@@ -5217,6 +5217,12 @@ class ProxyConfig:
                 )
 
                 CyberArkSecretManager()
+            elif key_management_system == KeyManagementSystem.KEEPER.value:
+                from litellm.secret_managers.keeper_secret_manager import (
+                    KeeperSecretManager,
+                )
+
+                KeeperSecretManager()
             elif key_management_system == KeyManagementSystem.CUSTOM.value:
                 ### LOAD CUSTOM SECRET MANAGER ###
                 from litellm.secret_managers.custom_secret_manager_loader import (

litellm/secret_managers/keeper_secret_manager.py

@@ -0,0 +1,124 @@
+"""
+Keeper Secrets Manager (KSM) integration.
+
+Read-only support for resolving provider credentials (e.g. an OpenAI or Gemini
+api_key) that live in a Keeper vault. Secrets are addressed with Keeper Notation
+(https://docs.keeper.io/en/keeperpam/secrets-manager/about/keeper-notation),
+for example ``<record_uid>/field/password``.
+
+Configuration (environment variables):
+* ``KSM_CONFIG``   base64 KSM device configuration (preferred, persistent)
+* ``KSM_TOKEN``    one-time access token used to bind a new device
+* ``KSM_HOSTNAME`` optional region hostname (e.g. ``keepersecurity.eu``)
+"""
+
+import asyncio
+import os
+from typing import Protocol, cast
+
+import httpx
+
+import litellm
+from litellm._logging import verbose_logger
+from litellm.caching import InMemoryCache
+from litellm.constants import SECRET_MANAGER_REFRESH_INTERVAL
+from litellm.proxy._types import KeyManagementSystem
+
+from .base_secret_manager import BaseSecretManager
+
+_READ_ONLY_RESPONSE: dict[str, str] = {
+    "status": "not_supported",
+    "message": "Keeper Secrets Manager integration is read-only.",
+}
+
+
+class _KeeperClient(Protocol):
+    def get_notation_results(self, notation: str) -> list[str]: ...
+
+
+class KeeperSecretManager(BaseSecretManager):
+    def __init__(self, client: _KeeperClient | None = None):
+        self.config_b64 = os.getenv("KSM_CONFIG")
+        self.token = os.getenv("KSM_TOKEN")
+        self.hostname = os.getenv("KSM_HOSTNAME")
+
+        self._client: _KeeperClient = client if client is not None else self._build_client()
+
+        litellm.secret_manager_client = self
+        litellm._key_management_system = KeyManagementSystem.KEEPER
+
+        _refresh_interval = os.environ.get("KSM_REFRESH_INTERVAL", SECRET_MANAGER_REFRESH_INTERVAL)
+        _refresh_interval = int(_refresh_interval) if _refresh_interval else SECRET_MANAGER_REFRESH_INTERVAL
+        self.cache = InMemoryCache(default_ttl=_refresh_interval)
+
+    def _build_client(self) -> _KeeperClient:
+        if not self.config_b64 and not self.token:
+            raise ValueError(
+                "Missing Keeper Secrets Manager credentials. Set either:\n"
+                "  - KSM_CONFIG (base64 device configuration), or\n"
+                "  - KSM_TOKEN (one-time access token)"
+            )
+
+        try:
+            from keeper_secrets_manager_core import SecretsManager
+            from keeper_secrets_manager_core.storage import InMemoryKeyValueStorage
+        except ImportError as e:
+            raise ImportError(
+                "keeper-secrets-manager-core is not installed. Run `pip install keeper-secrets-manager-core`."
+            ) from e
+
+        storage = InMemoryKeyValueStorage(self.config_b64) if self.config_b64 else InMemoryKeyValueStorage()
+        client = SecretsManager(token=self.token, hostname=self.hostname, config=storage)
+        return cast(_KeeperClient, client)  # cast-ok: KSM SDK has no stubs; _KeeperClient pins the one method used
+
+    def _read_notation(self, secret_name: str) -> str | None:
+        cached: str | None = self.cache.get_cache(secret_name)
+        if cached is not None:
+            return cached
+        try:
+            results = self._client.get_notation_results(secret_name)
+        except Exception as e:  # noqa: BLE001  # KSM SDK raises varied errors; a failed lookup is a miss
+            verbose_logger.exception(f"Error reading secret from Keeper Secrets Manager: {e}")
+            return None
+        if not results:
+            verbose_logger.debug(f"Secret {secret_name} not found in Keeper Secrets Manager")
+            return None
+        value = results[0]
+        self.cache.set_cache(secret_name, value)
+        return value
+
+    async def async_read_secret(
+        self,
+        secret_name: str,
+        optional_params: dict[str, object] | None = None,
+        timeout: float | httpx.Timeout | None = None,
+    ) -> str | None:
+        return await asyncio.to_thread(self._read_notation, secret_name)
+
+    def sync_read_secret(
+        self,
+        secret_name: str,
+        optional_params: dict[str, object] | None = None,
+        timeout: float | httpx.Timeout | None = None,
+    ) -> str | None:
+        return self._read_notation(secret_name)
+
+    async def async_write_secret(
+        self,
+        secret_name: str,
+        secret_value: str,
+        description: str | None = None,
+        optional_params: dict[str, object] | None = None,
+        timeout: float | httpx.Timeout | None = None,
+        tags: dict[str, object] | list[object] | None = None,
+    ) -> dict[str, str]:
+        return dict(_READ_ONLY_RESPONSE)
+
+    async def async_delete_secret(
+        self,
+        secret_name: str,
+        recovery_window_in_days: int | None = 7,
+        optional_params: dict[str, object] | None = None,
+        timeout: float | httpx.Timeout | None = None,
+    ) -> dict[str, str]:
+        return dict(_READ_ONLY_RESPONSE)

litellm/secret_managers/secret_manager_handler.py

@@ -140,6 +140,15 @@ def get_secret_from_manager(
             print_verbose(f"An error occurred - {str(e)}")
             raise e
 
+    elif key_manager == KeyManagementSystem.KEEPER.value:
+        try:
+            secret = client.sync_read_secret(secret_name=secret_name)
+            if secret is None:
+                raise ValueError(f"No secret found in Keeper Secrets Manager for {secret_name}")
+        except Exception as e:
+            print_verbose(f"An error occurred - {str(e)}")
+            raise e
+
     elif key_manager == KeyManagementSystem.CUSTOM.value:
         # Check if client is a CustomSecretManager instance
         from litellm.integrations.custom_secret_manager import CustomSecretManager

litellm/types/secret_managers/main.py

@@ -11,6 +11,7 @@ class KeyManagementSystem(enum.Enum):
     GOOGLE_SECRET_MANAGER = "google_secret_manager"
     HASHICORP_VAULT = "hashicorp_vault"
     CYBERARK = "cyberark"
+    KEEPER = "keeper"
     LOCAL = "local"
     AWS_KMS = "aws_kms"
     CUSTOM = "custom"

pyproject.toml

@@ -93,6 +93,7 @@ extra_proxy = [
     "resend>=2.23.0,<3.0",
     "redisvl>=0.4.1,<1.0",
     "a2a-sdk>=1.1.0,<2.0",
+    "keeper-secrets-manager-core>=17.3.0,<18.0",
 ]
 utils = [
     # Not in Docker or PyPI proxy extra.

tests/litellm_utils_tests/test_keeper.py

@@ -0,0 +1,111 @@
+"""
+Unit tests for the Keeper Secrets Manager (KSM) read-only integration.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../.."))
+
+from litellm.secret_managers.keeper_secret_manager import KeeperSecretManager
+from litellm.secret_managers.secret_manager_handler import get_secret_from_manager
+from litellm.types.secret_managers.main import KeyManagementSystem
+
+
+def _make_manager(results):
+    client = MagicMock()
+    client.get_notation_results.return_value = results
+    return KeeperSecretManager(client=client), client
+
+
+def test_keeper_enum_value():
+    assert KeyManagementSystem.KEEPER.value == "keeper"
+
+
+def test_sync_read_returns_first_notation_value():
+    manager, client = _make_manager(["sk-openai-123"])
+
+    value = manager.sync_read_secret("RECORD_UID/field/password")
+
+    assert value == "sk-openai-123"
+    client.get_notation_results.assert_called_once_with("RECORD_UID/field/password")
+
+
+def test_sync_read_caches_and_avoids_second_call():
+    manager, client = _make_manager(["sk-openai-123"])
+
+    first = manager.sync_read_secret("RECORD_UID/field/password")
+    second = manager.sync_read_secret("RECORD_UID/field/password")
+
+    assert first == second == "sk-openai-123"
+    client.get_notation_results.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_async_read_returns_value():
+    manager, client = _make_manager(["sk-gemini-abc"])
+
+    value = await manager.async_read_secret("RECORD_UID/field/password")
+
+    assert value == "sk-gemini-abc"
+    client.get_notation_results.assert_called_once_with("RECORD_UID/field/password")
+
+
+def test_sync_read_missing_secret_returns_none():
+    manager, client = _make_manager([])
+
+    assert manager.sync_read_secret("RECORD_UID/field/password") is None
+
+
+def test_sync_read_swallows_client_errors():
+    client = MagicMock()
+    client.get_notation_results.side_effect = RuntimeError("boom")
+    manager = KeeperSecretManager(client=client)
+
+    assert manager.sync_read_secret("RECORD_UID/field/password") is None
+
+
+@pytest.mark.asyncio
+async def test_write_and_delete_are_read_only():
+    manager, _ = _make_manager(["v"])
+
+    write = await manager.async_write_secret(secret_name="x", secret_value="y")
+    delete = await manager.async_delete_secret(secret_name="x")
+
+    assert write["status"] == "not_supported"
+    assert delete["status"] == "not_supported"
+
+
+def test_missing_credentials_raise():
+    for var in ("KSM_CONFIG", "KSM_TOKEN"):
+        os.environ.pop(var, None)
+
+    with pytest.raises(ValueError, match="Missing Keeper Secrets Manager credentials"):
+        KeeperSecretManager()
+
+
+def test_handler_dispatches_keeper_read():
+    manager, client = _make_manager(["sk-from-keeper"])
+
+    secret = get_secret_from_manager(
+        client=manager,
+        key_manager=KeyManagementSystem.KEEPER.value,
+        secret_name="RECORD_UID/field/password",
+    )
+
+    assert secret == "sk-from-keeper"
+    client.get_notation_results.assert_called_once_with("RECORD_UID/field/password")
+
+
+def test_handler_raises_when_keeper_secret_missing():
+    manager, _ = _make_manager([])
+
+    with pytest.raises(ValueError, match="No secret found in Keeper Secrets Manager"):
+        get_secret_from_manager(
+            client=manager,
+            key_manager=KeyManagementSystem.KEEPER.value,
+            secret_name="RECORD_UID/field/password",
+        )

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -24616,7 +24616,7 @@ export interface components {
          * KeyManagementSystem
          * @enum {string}
          */
-        KeyManagementSystem: "google_kms" | "azure_key_vault" | "aws_secret_manager" | "google_secret_manager" | "hashicorp_vault" | "cyberark" | "local" | "aws_kms" | "custom";
+        KeyManagementSystem: "google_kms" | "azure_key_vault" | "aws_secret_manager" | "google_secret_manager" | "hashicorp_vault" | "cyberark" | "keeper" | "local" | "aws_kms" | "custom";
         /**
          * KeyMetadata
          * @description Metadata for a key

uv.lock

@@ -10,7 +10,7 @@ resolution-markers = [
 ]
 
 [options]
-exclude-newer = "2026-07-19T00:00:06.091071Z"
+exclude-newer = "2026-07-19T18:53:57.627392561Z"
 exclude-newer-span = "P3D"
 
 [manifest]
@@ -3686,6 +3686,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/41/45/1a4ed80516f02155c51f51e8cedb3c1902296743db0bbc66608a0db2814f/jsonschema_specifications-2025.9.1-py3-none-any.whl", hash = "sha256:98802fee3a11ee76ecaca44429fda8a41bff98b00a0f2838151b113f210cc6fe", size = 18437, upload-time = "2025-09-08T01:34:57.871Z" },
 ]
 
+[[package]]
+name = "keeper-secrets-manager-core"
+version = "17.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "cryptography" },
+    { name = "requests" },
+    { name = "urllib3" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/d6/55/fbf9b0952cc53ae473697302b71f00fbc60ee7486a28aaf91c14dcd8bf58/keeper_secrets_manager_core-17.3.0.tar.gz", hash = "sha256:dbd3c2b2160a92aed46c2f5355cbd793492f43ba260379d78957e7ad75a96acf", size = 59643, upload-time = "2026-06-15T19:57:30.961Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b6/e8/b82a950ef31c3d53d7a0377de1f9819ae79470d3d766992fba6fe752fc93/keeper_secrets_manager_core-17.3.0-py3-none-any.whl", hash = "sha256:80327afaa06a04925acfc2d61f405a0831ef00e6c6fb2fff841a0704d9617221", size = 59813, upload-time = "2026-06-15T19:57:29.76Z" },
+]
+
 [[package]]
 name = "kiwisolver"
 version = "1.5.0"
@@ -4152,6 +4166,7 @@ extra-proxy = [
     { name = "azure-keyvault-secrets" },
     { name = "google-cloud-iam" },
     { name = "google-cloud-kms" },
+    { name = "keeper-secrets-manager-core" },
     { name = "prisma" },
     { name = "redisvl" },
     { name = "resend" },
@@ -4361,6 +4376,7 @@ requires-dist = [
     { name = "inquirerpy", marker = "extra == 'proxy'", specifier = ">=0.3.4,<1.0" },
     { name = "jinja2", specifier = ">=3.1.6,<4.0" },
     { name = "jsonschema", specifier = ">=4.0.0,<5.0" },
+    { name = "keeper-secrets-manager-core", marker = "extra == 'extra-proxy'", specifier = ">=17.3.0,<18.0" },
     { name = "langfuse", marker = "extra == 'proxy-runtime'", specifier = ">=2.59.7,<3.0" },
     { name = "litellm-enterprise", marker = "extra == 'proxy'", editable = "enterprise" },
     { name = "litellm-proxy-extras", marker = "extra == 'proxy'", editable = "litellm-proxy-extras" },