feat(proxy): accept TLS PEM from environment

litellm/proxy/enterprise_billing/billing_metrics.py

@@ -14,7 +14,6 @@ payload; the secret license key is never sent as an attribute or header.
 """
 
 import os
-import tempfile
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Final, Optional
 
@@ -28,6 +27,7 @@ from litellm._logging import verbose_proxy_logger
 from litellm.proxy.middleware.billable_request_metrics_middleware import (
     BillableCategory,
 )
+from litellm.proxy.tls_utils import materialize_pem_files
 
 if TYPE_CHECKING:
     from litellm.proxy._types import EnterpriseLicenseData
@@ -45,7 +45,6 @@ _METRICS_PATH: Final = "/v1/metrics"
 # values as env content cannot mount them as files, so inline PEM is written out.
 _PEM_PREFIX: Final = "-----BEGIN"
 _PEM_DIR_PREFIX: Final = "litellm-billing-mtls-"
-_PEM_FILE_MODE: Final = 0o600
 _CLIENT_CERT_FILENAME: Final = "client.crt"
 _CLIENT_KEY_FILENAME: Final = "client.key"
 _CA_CERT_FILENAME: Final = "ca.crt"
@@ -157,14 +156,6 @@ def _is_pem_content(value: str) -> bool:
     return value.lstrip().startswith(_PEM_PREFIX)
 
 
-def _write_pem(directory: str, filename: str, pem: str) -> str:
-    path: Final = os.path.join(directory, filename)
-    with open(path, "w", encoding="utf-8") as handle:
-        handle.write(pem if pem.endswith("\n") else f"{pem}\n")
-    os.chmod(path, _PEM_FILE_MODE)
-    return path
-
-
 def _resolve_credential_paths(*, client_cert: str, client_key: str, ca_cert: str | None) -> _CredentialPaths:
     """
     Accept either a filesystem path or inline PEM content for each credential.
@@ -176,22 +167,26 @@ def _resolve_credential_paths(*, client_cert: str, client_key: str, ca_cert: str
     recorder is built. Raises OSError if that write fails; the caller disables
     metering rather than propagating.
     """
-    inline: Final = tuple(value for value in (client_cert, client_key, ca_cert) if value and _is_pem_content(value))
-    if not inline:
+    inline_files: Final = {
+        filename: value
+        for filename, value in (
+            (_CLIENT_CERT_FILENAME, client_cert),
+            (_CLIENT_KEY_FILENAME, client_key),
+            (_CA_CERT_FILENAME, ca_cert),
+        )
+        if value and _is_pem_content(value)
+    }
+    if not inline_files:
         return _CredentialPaths(client_cert, client_key, ca_cert)
 
-    # mkdtemp is 0o700, so the 0o600 key file it holds is unreachable by other users.
-    directory: Final = tempfile.mkdtemp(prefix=_PEM_DIR_PREFIX)
+    paths: Final = materialize_pem_files(
+        inline_files,
+        directory_prefix=_PEM_DIR_PREFIX,
+    )
     return _CredentialPaths(
-        client_cert_path=(
-            _write_pem(directory, _CLIENT_CERT_FILENAME, client_cert) if _is_pem_content(client_cert) else client_cert
-        ),
-        client_key_path=(
-            _write_pem(directory, _CLIENT_KEY_FILENAME, client_key) if _is_pem_content(client_key) else client_key
-        ),
-        ca_cert_path=(
-            _write_pem(directory, _CA_CERT_FILENAME, ca_cert) if ca_cert and _is_pem_content(ca_cert) else ca_cert
-        ),
+        client_cert_path=paths.get(_CLIENT_CERT_FILENAME, client_cert),
+        client_key_path=paths.get(_CLIENT_KEY_FILENAME, client_key),
+        ca_cert_path=paths.get(_CA_CERT_FILENAME, ca_cert),
     )
 
 

litellm/proxy/proxy_cli.py

@@ -44,6 +44,8 @@ _deprioritize_script_dir_in_sys_path()
 sys.path.append(os.getcwd())
 
 config_filename: Final = "litellm.secrets"
+LITELLM_SSL_CERT_PEM: Final = "LITELLM_SSL_CERT_PEM"
+LITELLM_SSL_KEY_PEM: Final = "LITELLM_SSL_KEY_PEM"
 
 litellm_mode: Final = os.getenv("LITELLM_MODE", "DEV")  # "PRODUCTION", "DEV"
 if litellm_mode == "DEV":
@@ -178,6 +180,41 @@ def append_query_params(url: str | None, params: dict) -> str:
 
 
 class ProxyInitializationHelpers:
+    @staticmethod
+    def _resolve_ssl_file_paths(
+        ssl_certfile_path: str | None,
+        ssl_keyfile_path: str | None,
+    ) -> tuple[str | None, str | None]:
+        cert_pem_configured = LITELLM_SSL_CERT_PEM in os.environ
+        key_pem_configured = LITELLM_SSL_KEY_PEM in os.environ
+        path_configured = ssl_certfile_path is not None or ssl_keyfile_path is not None
+        pem_configured = cert_pem_configured or key_pem_configured
+
+        if path_configured and pem_configured:
+            raise click.ClickException(
+                "Configure inbound TLS with either the SSL file paths or "
+                "LITELLM_SSL_CERT_PEM and LITELLM_SSL_KEY_PEM, not both."
+            )
+        if path_configured:
+            if not ssl_certfile_path or not ssl_keyfile_path:
+                raise click.ClickException("Both --ssl_certfile_path and --ssl_keyfile_path are required for SSL.")
+            return ssl_certfile_path, ssl_keyfile_path
+        if not pem_configured:
+            return None, None
+
+        cert_pem = os.getenv(LITELLM_SSL_CERT_PEM)
+        key_pem = os.getenv(LITELLM_SSL_KEY_PEM)
+        if not cert_pem or not key_pem:
+            raise click.ClickException("Both LITELLM_SSL_CERT_PEM and LITELLM_SSL_KEY_PEM are required for SSL.")
+
+        from litellm.proxy.tls_utils import materialize_pem_files
+
+        paths = materialize_pem_files(
+            {"cert.pem": cert_pem, "key.pem": key_pem},
+            directory_prefix="litellm-proxy-tls-",
+        )
+        return paths["cert.pem"], paths["key.pem"]
+
     @staticmethod
     def _echo_litellm_version():
         pkg_version: Final = importlib.metadata.version("litellm")
@@ -1369,6 +1406,11 @@ def run_server(
             print("LiteLLM: Setup complete. Skipping server startup as requested.")
             return
 
+        ssl_certfile_path, ssl_keyfile_path = ProxyInitializationHelpers._resolve_ssl_file_paths(
+            ssl_certfile_path,
+            ssl_keyfile_path,
+        )
+
         running_uvicorn: Final = run_gunicorn is False and run_hypercorn is False
         uvicorn_args: Final = ProxyInitializationHelpers._get_default_unvicorn_init_args(
             host=host,

litellm/proxy/tls_utils.py

@@ -0,0 +1,17 @@
+import os
+import tempfile
+from collections.abc import Mapping
+
+PEM_FILE_MODE = 0o600
+
+
+def materialize_pem_files(files: Mapping[str, str], *, directory_prefix: str) -> dict[str, str]:
+    directory = tempfile.mkdtemp(prefix=directory_prefix)
+    paths: dict[str, str] = {}
+    for filename, pem in files.items():
+        path = os.path.join(directory, filename)
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write(pem if pem.endswith("\n") else f"{pem}\n")
+        os.chmod(path, PEM_FILE_MODE)
+        paths[filename] = path
+    return paths

tests/test_litellm/proxy/enterprise_billing/test_billing_metrics.py

@@ -265,10 +265,10 @@ def test_load_config_with_inline_pem_disabled_when_unwritable(monkeypatch):
     monkeypatch.setenv(bm.CLIENT_CERT_ENV, _CLIENT_CERT_PEM)
     monkeypatch.setenv(bm.CLIENT_KEY_ENV, _CLIENT_KEY_PEM)
 
-    def _explode(prefix=None):
+    def _explode(files, *, directory_prefix):
         raise OSError("read-only filesystem")
 
-    monkeypatch.setattr(bm.tempfile, "mkdtemp", _explode)
+    monkeypatch.setattr(bm, "materialize_pem_files", _explode)
 
     assert bm.load_billing_metrics_config(license_data=None, litellm_version="1.0") is None
 

tests/test_litellm/proxy/test_proxy_cli.py

@@ -21,6 +21,51 @@ from litellm.proxy.proxy_cli import ProxyInitializationHelpers, run_server
 
 @pytest.mark.xdist_group("proxy_cli")
 class TestProxyInitializationHelpers:
+    def test_resolve_ssl_file_paths_materializes_inline_pem(self, monkeypatch):
+        cert_pem = "-----BEGIN CERTIFICATE-----\ncert\n-----END CERTIFICATE-----"
+        key_pem = "-----BEGIN PRIVATE KEY-----\nkey\n-----END PRIVATE KEY-----"
+        monkeypatch.setenv("LITELLM_SSL_CERT_PEM", cert_pem)
+        monkeypatch.setenv("LITELLM_SSL_KEY_PEM", key_pem)
+
+        cert_path, key_path = ProxyInitializationHelpers._resolve_ssl_file_paths(None, None)
+
+        assert cert_path is not None
+        assert key_path is not None
+        assert Path(cert_path).read_text(encoding="utf-8") == f"{cert_pem}\n"
+        assert Path(key_path).read_text(encoding="utf-8") == f"{key_pem}\n"
+        assert os.stat(cert_path).st_mode & 0o777 == 0o600
+        assert os.stat(key_path).st_mode & 0o777 == 0o600
+
+    def test_resolve_ssl_file_paths_preserves_path_pair(self, monkeypatch):
+        monkeypatch.delenv("LITELLM_SSL_CERT_PEM", raising=False)
+        monkeypatch.delenv("LITELLM_SSL_KEY_PEM", raising=False)
+        assert ProxyInitializationHelpers._resolve_ssl_file_paths(
+            "/tls/cert.pem", "/tls/key.pem"
+        ) == ("/tls/cert.pem", "/tls/key.pem")
+
+    @pytest.mark.parametrize(
+        ("cert_path", "key_path", "cert_pem", "key_pem"),
+        [
+            ("/tls/cert.pem", None, None, None),
+            (None, "/tls/key.pem", None, None),
+            (None, None, "certificate", None),
+            (None, None, None, "private-key"),
+            ("/tls/cert.pem", "/tls/key.pem", "certificate", "private-key"),
+        ],
+    )
+    def test_resolve_ssl_file_paths_rejects_incomplete_or_mixed_sources(
+        self, monkeypatch, cert_path, key_path, cert_pem, key_pem
+    ):
+        monkeypatch.delenv("LITELLM_SSL_CERT_PEM", raising=False)
+        monkeypatch.delenv("LITELLM_SSL_KEY_PEM", raising=False)
+        if cert_pem is not None:
+            monkeypatch.setenv("LITELLM_SSL_CERT_PEM", cert_pem)
+        if key_pem is not None:
+            monkeypatch.setenv("LITELLM_SSL_KEY_PEM", key_pem)
+
+        with pytest.raises(click.ClickException):
+            ProxyInitializationHelpers._resolve_ssl_file_paths(cert_path, key_path)
+
     @patch("importlib.metadata.version")
     @patch("click.echo")
     def test_echo_litellm_version(self, mock_echo, mock_version):
@@ -507,6 +552,10 @@ class TestProxyInitializationHelpers:
         modified_url = append_query_params(None, {"connection_limit": 10})
         assert modified_url == ""
 
+    @patch(
+        "litellm.proxy.proxy_cli.ProxyInitializationHelpers._resolve_ssl_file_paths",
+        return_value=("/tls/resolved.crt", "/tls/resolved.key"),
+    )
     @patch("uvicorn.run")
     @patch("atexit.register")  # critical
     @patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
@@ -514,7 +563,12 @@ class TestProxyInitializationHelpers:
         "litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
     )
     def test_skip_server_startup(
-        self, mock_should_update, mock_setup_db, mock_atexit_register, mock_uvicorn_run
+        self,
+        mock_should_update,
+        mock_setup_db,
+        mock_atexit_register,
+        mock_uvicorn_run,
+        mock_resolve_ssl_paths,
     ):
         from click.testing import CliRunner
 
@@ -569,6 +623,7 @@ class TestProxyInitializationHelpers:
             ), f"exit_code={result.exit_code}, output={result.output}"
             assert "Skipping server startup" in result.output
             mock_uvicorn_run.assert_not_called()
+            mock_resolve_ssl_paths.assert_not_called()
 
             # --- normal startup ---
             mock_uvicorn_run.reset_mock()
@@ -579,6 +634,10 @@ class TestProxyInitializationHelpers:
                 result.exit_code == 0
             ), f"exit_code={result.exit_code}, output={result.output}"
             mock_uvicorn_run.assert_called_once()
+            mock_resolve_ssl_paths.assert_called_once_with(None, None)
+            call_kwargs = mock_uvicorn_run.call_args.kwargs
+            assert call_kwargs["ssl_certfile"] == "/tls/resolved.crt"
+            assert call_kwargs["ssl_keyfile"] == "/tls/resolved.key"
 
     @patch("uvicorn.run")
     @patch("atexit.register")