docker secret manager

litellm/proxy/proxy_server.py

@@ -3571,6 +3571,23 @@ class ProxyConfig:
                 )
 
                 CyberArkSecretManager()
+            elif (
+                key_management_system
+                == KeyManagementSystem.DOCKER_SECRET_MANAGER.value
+            ):
+                from litellm.secret_managers.docker_secret_manager import (
+                    DockerSecretsManager,
+                )
+
+                _docker_kwargs = {}
+                if (
+                    litellm._key_management_settings is not None
+                    and litellm._key_management_settings.secrets_dir is not None
+                ):
+                    _docker_kwargs["secrets_dir"] = (
+                        litellm._key_management_settings.secrets_dir
+                    )
+                DockerSecretsManager(**_docker_kwargs)
             elif key_management_system == KeyManagementSystem.CUSTOM.value:
                 ### LOAD CUSTOM SECRET MANAGER ###
                 from litellm.secret_managers.custom_secret_manager_loader import (

litellm/secret_managers/docker_secret_manager.py

@@ -0,0 +1,144 @@
+"""
+Docker Secrets Manager
+
+Reads secrets from files mounted at `/run/secrets/<secret_name>` by Docker
+Swarm or `docker run --secret`. The file contents are returned as-is after
+stripping surrounding whitespace (like the trailing newline Docker adds)
+"""
+
+import os
+from typing import Any, Dict, Optional, Union
+import httpx
+import litellm
+from litellm._logging import verbose_logger
+from litellm.proxy._types import KeyManagementSystem
+from .base_secret_manager import BaseSecretManager
+
+
+class DockerSecretsManager(BaseSecretManager):
+    """
+    secrets are mounted by under these default directories
+    linux: '/run/secrets/<SECRET_NAME>'
+    windows: 'C:\\ProgramData\\Docker\\secrets'
+    
+    `docker run --secret` mount each secret as a file whose
+    name equals the secret name and whose content is the secret value. This
+    manager simply reads those files, so no external SDK or credentials are
+    required
+    """
+
+    @staticmethod
+    def _default_secrets_dir() -> str:
+        if os.name == "nt":
+            return r"C:\ProgramData\Docker\secrets"
+        return "/run/secrets"
+
+    def __init__(self, secrets_dir: Optional[str] = None) -> None:
+        self.secrets_dir = (
+            secrets_dir
+            if secrets_dir is not None
+            else DockerSecretsManager._default_secrets_dir()
+        )
+
+        litellm.secret_manager_client = self
+        litellm._key_management_system = KeyManagementSystem.DOCKER_SECRET_MANAGER
+
+    def _secret_path(self, secret_name: str) -> str:
+        """Return the absolute filesystem path for *secret_name*."""
+        return os.path.join(self.secrets_dir, secret_name)
+
+    def _read_file(self, secret_name: str) -> Optional[str]:
+        """
+        Read the content of the secret file, stripping surrounding whitespace.
+
+        Returns None when the file does not exist.  Logs a warning and returns
+        None on permission errors so the caller can fall back to env vars
+        """
+        path = self._secret_path(secret_name)
+        try:
+            with open(path, "r") as fh:
+                return fh.read().strip()
+        except FileNotFoundError:
+            # base case: secret simply isn't present as a Docker secret
+            return None
+        except PermissionError:
+            verbose_logger.warning(
+                "DockerSecretsManager: permission denied reading '%s'. "
+                "Ensure the process has read access to the secrets directory.",
+                path,
+            )
+            return None
+
+    # ------------------------------------------------------------------
+    # BaseSecretManager interface
+    # ------------------------------------------------------------------
+
+    async def async_read_secret(
+        self,
+        secret_name: str,
+        optional_params: Optional[dict] = None,
+        timeout: Optional[Union[float, httpx.Timeout]] = None,
+    ) -> Optional[str]:
+        """
+        async read a docker secret
+        Docker secrets are plain files this method delegates directly to the synchronous file read.
+        """
+        return self._read_file(secret_name)
+
+    def sync_read_secret(
+        self,
+        secret_name: str,
+        optional_params: Optional[dict] = None,
+        timeout: Optional[Union[float, httpx.Timeout]] = None,
+    ) -> Optional[str]:
+        """Synchronously read a Docker secret."""
+        return self._read_file(secret_name)
+
+    async def async_write_secret(
+        self,
+        secret_name: str,
+        secret_value: str,
+        description: Optional[str] = None,
+        optional_params: Optional[dict] = None,
+        timeout: Optional[Union[float, httpx.Timeout]] = None,
+        tags: Optional[Union[dict, list]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Docker secrets are managed by the Docker daemon and cannot be written
+        at runtime via the filesystem.  This operation is intentionally
+        unsupported.
+        """
+        verbose_logger.warning(
+            "DockerSecretsManager: write operations are not supported. "
+            "Docker secrets are managed by the Docker daemon."
+        )
+        return {
+            "status": "not_supported",
+            "message": (
+                "DockerSecretsManager does not support write operations. "
+                "Manage secrets through the Docker CLI or Swarm."
+            ),
+        }
+
+    async def async_delete_secret(
+        self,
+        secret_name: str,
+        recovery_window_in_days: Optional[int] = 7,
+        optional_params: Optional[dict] = None,
+        timeout: Optional[Union[float, httpx.Timeout]] = None,
+    ) -> dict:
+        """
+        Docker secrets cannot be deleted at runtime via the filesystem.  This
+        operation is intentionally unsupported.
+        """
+        verbose_logger.warning(
+            "DockerSecretsManager: delete operations are not supported. "
+            "Docker secrets are managed by the Docker daemon."
+        )
+        return {
+            "status": "not_supported",
+            "message": (
+                "DockerSecretsManager does not support delete operations. "
+                "Manage secrets through the Docker CLI or Swarm."
+            ),
+        }

litellm/secret_managers/secret_manager_handler.py

@@ -153,6 +153,17 @@ def get_secret_from_manager(  # noqa: PLR0915
             print_verbose(f"An error occurred - {str(e)}")
             raise e
 
+    elif key_manager == KeyManagementSystem.DOCKER_SECRET_MANAGER.value:
+        try:
+            secret = client.sync_read_secret(secret_name=secret_name)
+            if secret is None:
+                raise ValueError(
+                    f"No secret found in Docker 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

@@ -14,7 +14,7 @@ class KeyManagementSystem(enum.Enum):
     LOCAL = "local"
     AWS_KMS = "aws_kms"
     CUSTOM = "custom"
-
+    DOCKER_SECRET_MANAGER = "docker"
 
 class KeyManagementSettings(LiteLLMPydanticObjectBase):
     hosted_keys: Optional[List] = None
@@ -72,3 +72,9 @@ class KeyManagementSettings(LiteLLMPydanticObjectBase):
 
     aws_sts_endpoint: Optional[str] = None
     """Custom STS endpoint URL (useful for VPC endpoints or testing)"""
+
+    secrets_dir: Optional[str] = None
+    """
+    Directory to read secrets from. Only used by DockerSecretsManager.
+    Defaults to ``/run/secrets`` (the standard Docker secrets mount point).
+    """

tests/test_litellm/test_docker_secret_manager.py

@@ -0,0 +1,197 @@
+"""
+Unit tests for DockerSecretsManager.
+
+Uses pytest's `tmp_path` fixture to create real temporary files so no file-I/O
+mocking is required.
+"""
+
+import asyncio
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../.."))
+
+import litellm
+from litellm.secret_managers import docker_secret_manager
+from litellm.secret_managers.docker_secret_manager import DockerSecretsManager
+from litellm.types.secret_managers.main import KeyManagementSettings, KeyManagementSystem
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def make_manager(tmp_path) -> DockerSecretsManager:
+    """Return a DockerSecretsManager pointed at *tmp_path*."""
+    return DockerSecretsManager(secrets_dir=str(tmp_path))
+
+
+def write_secret(tmp_path, name: str, value: str) -> None:
+    """Write *value* to a file named *name* inside *tmp_path*."""
+    secret_file = tmp_path / name
+    secret_file.write_text(value)
+
+
+# ---------------------------------------------------------------------------
+# sync_read_secret
+# ---------------------------------------------------------------------------
+
+
+def test_sync_read_secret_existing_file(tmp_path):
+    """sync_read_secret returns the file content when the secret exists."""
+    write_secret(tmp_path, "MY_API_KEY", "sk-abc123")
+    manager = make_manager(tmp_path)
+
+    result = manager.sync_read_secret("MY_API_KEY")
+
+    assert result == "sk-abc123"
+
+
+def test_sync_read_secret_missing_file(tmp_path):
+    """sync_read_secret returns None when the secret file does not exist."""
+    manager = make_manager(tmp_path)
+
+    result = manager.sync_read_secret("NONEXISTENT_SECRET")
+
+    assert result is None
+
+
+def test_sync_read_secret_strips_trailing_newline(tmp_path):
+    """sync_read_secret strips the trailing newline Docker appends to secret files."""
+    # Docker always adds a trailing newline; verify we strip it.
+    write_secret(tmp_path, "DB_PASSWORD", "p@ssw0rd\n")
+    manager = make_manager(tmp_path)
+
+    result = manager.sync_read_secret("DB_PASSWORD")
+
+    assert result == "p@ssw0rd"
+
+
+def test_sync_read_secret_strips_surrounding_whitespace(tmp_path):
+    """sync_read_secret strips all surrounding whitespace, not just newlines."""
+    write_secret(tmp_path, "TOKEN", "  mytoken  \n")
+    manager = make_manager(tmp_path)
+
+    result = manager.sync_read_secret("TOKEN")
+
+    assert result == "mytoken"
+
+
+# ---------------------------------------------------------------------------
+# async_read_secret
+# ---------------------------------------------------------------------------
+
+
+def test_async_read_secret_existing_file(tmp_path):
+    """async_read_secret returns the file content when the secret exists."""
+    write_secret(tmp_path, "ASYNC_KEY", "async-value-xyz")
+    manager = make_manager(tmp_path)
+
+    result = asyncio.run(manager.async_read_secret("ASYNC_KEY"))
+
+    assert result == "async-value-xyz"
+
+
+def test_async_read_secret_missing_file(tmp_path):
+    """async_read_secret returns None when the secret file does not exist."""
+    manager = make_manager(tmp_path)
+
+    result = asyncio.run(manager.async_read_secret("NO_SUCH_SECRET"))
+
+    assert result is None
+
+
+def test_async_read_secret_strips_whitespace(tmp_path):
+    """async_read_secret strips trailing whitespace just like the sync variant."""
+    write_secret(tmp_path, "SECRET_WITH_NEWLINE", "value\n")
+    manager = make_manager(tmp_path)
+
+    result = asyncio.run(manager.async_read_secret("SECRET_WITH_NEWLINE"))
+
+    assert result == "value"
+
+
+# ---------------------------------------------------------------------------
+# Custom secrets_dir
+# ---------------------------------------------------------------------------
+
+
+def test_custom_secrets_dir(tmp_path):
+    """DockerSecretsManager respects a custom secrets_dir."""
+    custom_dir = tmp_path / "custom_secrets"
+    custom_dir.mkdir()
+    (custom_dir / "CUSTOM_SECRET").write_text("custom-value")
+
+    manager = DockerSecretsManager(secrets_dir=str(custom_dir))
+
+    assert manager.sync_read_secret("CUSTOM_SECRET") == "custom-value"
+    # A file in the default (non-custom) directory should NOT be found.
+    assert manager.sync_read_secret("MISSING") is None
+
+
+def test_secrets_dir_from_key_management_settings(tmp_path):
+    """secrets_dir in KeyManagementSettings is honoured when constructing DockerSecretsManager."""
+    (tmp_path / "MY_KEY").write_text("from-settings")
+    settings = KeyManagementSettings(secrets_dir=str(tmp_path))
+
+    manager = DockerSecretsManager(secrets_dir=settings.secrets_dir)
+
+    assert manager.sync_read_secret("MY_KEY") == "from-settings"
+    assert manager.secrets_dir == str(tmp_path)
+
+
+def test_default_secrets_dir_non_windows(monkeypatch):
+    """Without an explicit directory, non-Windows defaults to /run/secrets."""
+    monkeypatch.setattr(docker_secret_manager.os, "name", "posix")
+
+    manager = DockerSecretsManager()
+
+    assert manager.secrets_dir == "/run/secrets"
+
+
+def test_default_secrets_dir_windows(monkeypatch):
+    """Without an explicit directory, Windows defaults to ProgramData Docker secrets."""
+    monkeypatch.setattr(docker_secret_manager.os, "name", "nt")
+
+    manager = DockerSecretsManager()
+
+    assert manager.secrets_dir == r"C:\ProgramData\Docker\secrets"
+
+
+# ---------------------------------------------------------------------------
+# Registration: __init__ wires up litellm globals
+# ---------------------------------------------------------------------------
+
+
+def test_init_registers_as_secret_manager_client(tmp_path):
+    """Constructing DockerSecretsManager sets litellm.secret_manager_client."""
+    manager = make_manager(tmp_path)
+
+    assert litellm.secret_manager_client is manager
+    assert litellm._key_management_system == KeyManagementSystem.DOCKER_SECRET_MANAGER
+
+
+# ---------------------------------------------------------------------------
+# Unsupported operations return "not_supported" status
+# ---------------------------------------------------------------------------
+
+
+def test_async_write_secret_not_supported(tmp_path):
+    """async_write_secret returns a not_supported status dict."""
+    manager = make_manager(tmp_path)
+
+    result = asyncio.run(manager.async_write_secret("KEY", "value"))
+
+    assert result["status"] == "not_supported"
+
+
+def test_async_delete_secret_not_supported(tmp_path):
+    """async_delete_secret returns a not_supported status dict."""
+    manager = make_manager(tmp_path)
+
+    result = asyncio.run(manager.async_delete_secret("KEY"))
+
+    assert result["status"] == "not_supported"