scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-11 03:38:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 82

feat(proxy): default at-rest encryption to AES-256-GCM with HKDF v3 and make PyNaCl optional

Browse source 
New writes use v3:gcm: (AES-256-GCM under HKDF-SHA256). v2:gcm: and unprefixed
XSalsa20 values stay readable under the raw SHA-256 derivation they were written
with. xsalsa20-poly1305 remains an explicit opt-in and is refused under
LITELLM_FIPS_MODE. PyNaCl imports are lazy and a missing PyNaCl on legacy
ciphertext raises one error naming the legacy-encryption extra and the
re-encrypt path. pynacl moves from the proxy extra to legacy-encryption, which
the standard Dockerfiles install

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-24 08:32:05 +00:00
parent 44cb6c23d4
commit e8cee19bee
21 changed files with 619 additions and 246 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Dockerfile
View file

@ -78,6 +78,7 @@ COPY litellm-proxy-extras/pyproject.toml litellm-proxy-extras/
# Install third-party dependencies (cached unless pyproject.toml/uv.lock change)
RUN uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
@ -100,6 +101,7 @@ RUN sed -i 's/\r$//' docker/build_admin_ui.sh && chmod +x docker/build_admin_ui.
# Install project and workspace packages (fast - deps already cached)
RUN uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \

2
backend/Dockerfile
View file

@ -43,6 +43,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=litellm-proxy-extras/pyproject.toml,target=litellm-proxy-extras/pyproject.toml \
uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
@ -55,6 +56,7 @@ COPY . .
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \

2
docker/Dockerfile.database
View file

@ -76,6 +76,7 @@ COPY litellm-proxy-extras/pyproject.toml litellm-proxy-extras/
# Install third-party dependencies (cached unless pyproject.toml/uv.lock change)
RUN uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
@ -98,6 +99,7 @@ RUN sed -i 's/\r$//' docker/build_admin_ui.sh && chmod +x docker/build_admin_ui.
# Install project and workspace packages (fast - deps already cached)
RUN uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \

3
docker/Dockerfile.non_root
View file

@ -82,6 +82,7 @@ COPY litellm-proxy-extras/pyproject.toml litellm-proxy-extras/
RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \
uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
@ -110,6 +111,7 @@ RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \
if [ "$PROXY_EXTRAS_SOURCE" = "published" ]; then \
uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
@ -120,6 +122,7 @@ RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \
else \
uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \

2
gateway/Dockerfile
View file

@ -59,6 +59,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=litellm-proxy-extras/pyproject.toml,target=litellm-proxy-extras/pyproject.toml \
uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
@ -71,6 +72,7 @@ COPY . .
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \

6
litellm/proxy/_experimental/mcp_server/bridge_token_flow.py
View file

@ -14,9 +14,7 @@ from typing_extensions import assert_never
from litellm._logging import verbose_logger
from litellm.proxy._experimental.mcp_server.oauth_utils import TOKEN_NO_CACHE_HEADERS
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
_V2_GCM_PREFIX, # pyright: ignore[reportPrivateUsage] # reuse the encrypted credential's format discriminator
)
from litellm.proxy.common_utils.encrypt_decrypt_utils import is_versioned_gcm
from litellm.types.mcp_server.mcp_server_manager import MCPServer
if TYPE_CHECKING:
@ -99,7 +97,7 @@ async def _opaque_bearer_is_gateway_credential(token: str) -> bool:
user_api_key_cache,
)
if is_envelope(token) or is_refresh_envelope(token) or token.startswith(_V2_GCM_PREFIX):
if is_envelope(token) or is_refresh_envelope(token) or is_versioned_gcm(token):
return True
try:
if ExperimentalUIJWTToken.get_key_object_from_ui_hash_key(token) is not None:

2
litellm/proxy/_experimental/mcp_server/db.py
View file

@ -1418,7 +1418,7 @@ async def rotate_mcp_server_credentials_master_key(prisma_client: PrismaClient,
def _decode_user_credential(stored: str) -> str | None:
"""Read back a value persisted in ``LiteLLM_MCPUserCredentials.credential_b64``.
Tries nacl decryption first (current write format). Falls back to a
Tries the at-rest decryptor first (current write format). Falls back to a
plain ``urlsafe_b64decode`` for rows persisted by older code that wrote
the credential without encryption. Returns ``None`` when neither path
yields a valid string.

26
litellm/proxy/_experimental/mcp_server/outbound_credentials/envelope.py
View file

@ -33,14 +33,13 @@ defend against non-UTF-8 field content that cannot survive JSON parsing.
from __future__ import annotations
import base64
from datetime import datetime, timedelta
from typing import Final, Literal, TypeAlias
import jwt
from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value, encrypt_value
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_if_encrypted_with, encrypt_value_helper
ENVELOPE_PREFIX: Final = "llm_env_"
"""Marker prefix on every serialized ACCESS envelope so the edge can cheaply tell an envelope
@ -540,22 +539,15 @@ def _decode_claims(
def _encrypt_grant_blob(plaintext: str, encryption_key: SecretStr) -> str:
ciphertext: Final = bytes(encrypt_value(value=plaintext, signing_key=encryption_key.get_secret_value()))
return base64.urlsafe_b64encode(ciphertext).decode("ascii")
return encrypt_value_helper(value=plaintext, new_encryption_key=encryption_key.get_secret_value())
def _decrypt_grant(
blob: str,
encryption_key: SecretStr,
) -> UpstreamTokenGrant | DecryptFailed | MalformedPayload:
from nacl.exceptions import CryptoError
try:
plaintext: Final = decrypt_value(
value=base64.urlsafe_b64decode(blob),
signing_key=encryption_key.get_secret_value(),
)
except (CryptoError, ValueError):
plaintext: Final = decrypt_if_encrypted_with(value=blob, signing_key=encryption_key.get_secret_value())
if plaintext is None:
return DecryptFailed()
try:
return UpstreamTokenGrant.model_validate_json(plaintext)
@ -567,14 +559,8 @@ def _decrypt_refresh(
blob: str,
encryption_key: SecretStr,
) -> RefreshCredential | DecryptFailed | MalformedPayload:
from nacl.exceptions import CryptoError
try:
plaintext: Final = decrypt_value(
value=base64.urlsafe_b64decode(blob),
signing_key=encryption_key.get_secret_value(),
)
except (CryptoError, ValueError):
plaintext: Final = decrypt_if_encrypted_with(value=blob, signing_key=encryption_key.get_secret_value())
if plaintext is None:
return DecryptFailed()
try:
return RefreshCredential.model_validate_json(plaintext)

8
litellm/proxy/client/cli/commands/encryption.py
View file

@ -29,11 +29,11 @@ def encryption():
)
@click.pass_context
def migrate(ctx: click.Context, check_only: bool, dry_run: bool):
"""Re-encrypt at-rest credentials into the AES-256-GCM (v2:gcm:) format.
"""Re-encrypt at-rest credentials into the versioned AES-256-GCM (v3:gcm:) format.
Requires the proxy to be started with
``general_settings.encryption_algorithm: aes-256-gcm``. Idempotent and
resumable; safe to re-run after an interruption.
Requires the proxy to write ``aes-256-gcm`` (the default; refused while
``general_settings.encryption_algorithm: xsalsa20-poly1305`` is set). Idempotent
and resumable; safe to re-run after an interruption.
Examples:
lite encryption migrate --check # attestation scan, no writes

172
litellm/proxy/common_utils/encrypt_decrypt_utils.py
View file

@ -1,27 +1,49 @@
import base64
import hashlib
import os
from collections.abc import Mapping
from typing import Final, Literal, cast
from typing import TYPE_CHECKING, Final, Literal, cast
from pydantic import TypeAdapter, ValidationError
from litellm._logging import verbose_proxy_logger
from litellm.proxy.common_utils.fips import FipsModeError, is_fips_mode
# Versioned ciphertext marker for AES-256-GCM values.
# Format: "v2:gcm:" + base64url(nonce(12) || ciphertext || tag(16)).
if TYPE_CHECKING:
from nacl.secret import SecretBox
# Versioned ciphertext markers for AES-256-GCM values, each
# "<version>:gcm:" + base64url(nonce(12) || ciphertext || tag(16)).
# v3 derives the key with HKDF-SHA256, v2 with a raw SHA-256 of the salt key.
# Legacy XSalsa20-Poly1305 (nacl) values carry no marker; the colon in the
# prefix can never appear in base64url(nacl output), so the prefix check is an
# unambiguous discriminator between the two formats on read.
# unambiguous discriminator between the formats on read.
_V3_GCM_PREFIX: Final = "v3:gcm:"
_V2_GCM_PREFIX: Final = "v2:gcm:"
_GCM_PREFIXES: Final = (_V3_GCM_PREFIX, _V2_GCM_PREFIX)
_HKDF_INFO: Final = b"litellm-at-rest-v3"
# general_settings key selecting the at-rest encryption algorithm for new writes.
# Default preserves the legacy algorithm so existing deployments are byte-for-byte
# unchanged until they explicitly opt in. Decrypt is always format-detecting, so
# flipping this flag forward (or back) never strands previously-written data.
# Decrypt is always format-detecting, so the setting never strands stored data.
_ENCRYPTION_ALGORITHM_SETTING: Final = "encryption_algorithm"
_ALGO_AES_GCM: Final = "aes-256-gcm"
_ALGO_XSALSA20: Final = "xsalsa20-poly1305"
_NACL_MIN_CIPHERTEXT_BYTES: Final = 40
_LEGACY_ENCRYPTION_HELP: Final = (
"Install the legacy-encryption extra (pip install 'litellm[legacy-encryption]') on a non-FIPS image and "
"re-encrypt stored credentials with `litellm-proxy encryption migrate` (POST /credentials/migrate-encryption) "
"before running without PyNaCl"
)
class LegacyEncryptionUnavailableError(RuntimeError):
pass
def is_versioned_gcm(value: str) -> bool:
return value.startswith(_GCM_PREFIXES)
def _get_salt_key():
from litellm.proxy.proxy_server import master_key
@ -35,63 +57,66 @@ def _get_salt_key():
def _get_encryption_algorithm() -> str:
"""
Resolve the configured at-rest encryption algorithm for *new writes*.
"""Resolve the at-rest encryption algorithm for new writes from ``general_settings.encryption_algorithm``.
Read from ``general_settings.encryption_algorithm`` at write time. Defaults to
the legacy XSalsa20-Poly1305 algorithm so deployments that have not opted in
keep producing byte-for-byte identical ciphertext.
Defaults to AES-256-GCM. ``xsalsa20-poly1305`` stays available as an explicit opt-in for deployments that
still need byte-for-byte legacy output, except under ``LITELLM_FIPS_MODE`` where it is refused.
"""
try:
from litellm.proxy.proxy_server import general_settings
algo: Final = general_settings.get(_ENCRYPTION_ALGORITHM_SETTING, _ALGO_XSALSA20)
except Exception:
# general_settings may not be importable in some contexts (e.g. SDK-only
# use of these helpers). Fall back to the legacy algorithm.
return _ALGO_XSALSA20
if isinstance(algo, str) and algo.lower() == _ALGO_AES_GCM:
algo: Final = general_settings.get(_ENCRYPTION_ALGORITHM_SETTING, _ALGO_AES_GCM)
except Exception: # noqa: BLE001 # proxy_server is not importable in SDK-only use of these helpers
return _ALGO_AES_GCM
if not isinstance(algo, str) or algo.lower() != _ALGO_XSALSA20:
return _ALGO_AES_GCM
if is_fips_mode():
raise FipsModeError(
f"general_settings.{_ENCRYPTION_ALGORITHM_SETTING}={_ALGO_XSALSA20} is not allowed under "
f"LITELLM_FIPS_MODE: XSalsa20-Poly1305 is not a FIPS approved algorithm. Remove the setting to write "
f"{_ALGO_AES_GCM}"
)
return _ALGO_XSALSA20
def _derive_key(signing_key: str) -> bytes:
"""Derive a 32-byte key from the salt/master key (shared by both algorithms).
Known limitation: this is a single-pass, unsalted ``SHA-256`` of the key, not
a dedicated KDF (HKDF/PBKDF2). It is the *same* derivation the legacy nacl
path already uses, so the AES path introduces no new weakness and stays
interoperable with existing key sourcing; AES-256-GCM's per-value 12-byte
random nonce gives the unique (key, nonce) pairs GCM requires. Moving both
algorithms to HKDF-SHA256 would be more defensible in an audit but is a
separate, coordinated change (it must re-derive or re-encrypt existing data).
"""
import hashlib
def _derive_key_sha256(signing_key: str) -> bytes:
"""Historical derivation shared by legacy nacl values and ``v2:gcm:`` values: one unsalted SHA-256."""
return hashlib.sha256(signing_key.encode()).digest()
def _derive_key_hkdf(signing_key: str) -> bytes:
"""Derivation for ``v3:gcm:`` values: HKDF-SHA256 with a fixed info string."""
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=_HKDF_INFO).derive(signing_key.encode())
def _encrypt_aes_gcm(value: str, signing_key: str) -> str:
"""Encrypt under AES-256-GCM and return the versioned ``v2:gcm:`` string."""
"""Encrypt under AES-256-GCM and return the versioned ``v3:gcm:`` string."""
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
nonce: Final = os.urandom(12)
# AESGCM.encrypt returns ciphertext || tag(16); wire format is nonce || that.
blob: Final = AESGCM(_derive_key(signing_key)).encrypt(nonce, value.encode("utf-8"), None)
return _V2_GCM_PREFIX + base64.urlsafe_b64encode(nonce + blob).decode("utf-8")
blob: Final = AESGCM(_derive_key_hkdf(signing_key)).encrypt(nonce, value.encode("utf-8"), None)
return _V3_GCM_PREFIX + base64.urlsafe_b64encode(nonce + blob).decode("utf-8")
def _decrypt_aes_gcm(value: str, signing_key: str) -> str:
"""Decrypt a versioned ``v2:gcm:`` string produced by :func:`_encrypt_aes_gcm`."""
"""Decrypt a versioned ``v3:gcm:`` or ``v2:gcm:`` string, deriving the key the way its version was written."""
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
raw: Final = base64.urlsafe_b64decode(value[len(_V2_GCM_PREFIX) :])
if value.startswith(_V3_GCM_PREFIX):
key, prefix = _derive_key_hkdf(signing_key), _V3_GCM_PREFIX
else:
key, prefix = _derive_key_sha256(signing_key), _V2_GCM_PREFIX
raw: Final = base64.urlsafe_b64decode(value[len(prefix) :])
# An empty plaintext still serializes to nonce(12) || tag(16) = 28 bytes, so a
# short/empty buffer here is a corrupt value: let AESGCM.decrypt raise and be
# swallowed by decrypt_value_helper (returns None/original), same as legacy.
nonce, blob = raw[:12], raw[12:]
return AESGCM(_derive_key(signing_key)).decrypt(nonce, blob, None).decode("utf-8")
return AESGCM(key).decrypt(nonce, blob, None).decode("utf-8")
def encrypt_value_helper(value: str, new_encryption_key: str | None = None):
@ -100,7 +125,7 @@ def encrypt_value_helper(value: str, new_encryption_key: str | None = None):
try:
if isinstance(value, str):
if _get_encryption_algorithm() == _ALGO_AES_GCM:
# AES path: the v2:gcm: output is already a base64url string, so it
# AES path: the v3:gcm: output is already a base64url string, so it
# is returned directly with no extra base64 wrapper.
return _encrypt_aes_gcm(value=value, signing_key=cast(str, signing_key))
@ -131,19 +156,25 @@ def _legacy_ciphertext_bytes(value: str) -> bytes:
def _decrypt_with_signing_key(value: str, signing_key: str) -> str:
# Versioned AES-256-GCM values are detected before any base64 decode.
# The prefix is the algorithm tag the legacy nacl format never carried.
if value.startswith(_V2_GCM_PREFIX):
if is_versioned_gcm(value):
return _decrypt_aes_gcm(value=value, signing_key=signing_key)
return decrypt_value(value=_legacy_ciphertext_bytes(value), signing_key=signing_key)
def decrypt_if_encrypted_with(value: str, signing_key: str) -> str | None:
"""None unless value is a ciphertext under signing_key."""
"""None unless value is a ciphertext under signing_key.
A legacy ciphertext met without PyNaCl installed is not silently "not a ciphertext": it raises
:class:`LegacyEncryptionUnavailableError` naming the re-encrypt path.
"""
try:
# base64 decoding skips characters outside its alphabet, so "" and "*" decode to no bytes,
# which decrypt_value reads as an empty plaintext under any key.
decodes_to_nothing: Final = not value.startswith(_V2_GCM_PREFIX) and not _legacy_ciphertext_bytes(value)
decodes_to_nothing: Final = not is_versioned_gcm(value) and not _legacy_ciphertext_bytes(value)
return None if decodes_to_nothing else _decrypt_with_signing_key(value=value, signing_key=signing_key)
except LegacyEncryptionUnavailableError:
raise
except Exception: # noqa: BLE001 # base64, nacl and AES-GCM each raise their own "not a ciphertext" type
return None
@ -162,6 +193,8 @@ def decrypt_value_helper(
# if it's not str - do not decrypt it, return the value
return value
except LegacyEncryptionUnavailableError:
raise
except Exception as e:
error_message = f"Error decrypting value for key: {key}, Did your master_key/salt key change recently? \nError: {e}\nSet permanent salt key - https://docs.litellm.ai/docs/proxy/prod#5-set-litellm-salt-key"
if exception_type == "debug":
@ -177,50 +210,27 @@ def decrypt_value_helper(
return None
def encrypt_value(value: str, signing_key: str):
import hashlib
def _legacy_secret_box(signing_key: str, purpose: str) -> "SecretBox":
try:
import nacl.secret
except ImportError as error:
raise LegacyEncryptionUnavailableError(
f"Cannot {purpose} with the legacy {_ALGO_XSALSA20} algorithm: PyNaCl is not installed. "
f"{_LEGACY_ENCRYPTION_HELP}"
) from error
return nacl.secret.SecretBox(_derive_key_sha256(signing_key))
import nacl.secret
import nacl.utils
# get 32 byte master key #
hash_object: Final = hashlib.sha256(signing_key.encode())
hash_bytes: Final = hash_object.digest()
# initialize secret box #
box: Final = nacl.secret.SecretBox(hash_bytes)
# encode message #
value_bytes: Final = value.encode("utf-8")
encrypted: Final = box.encrypt(value_bytes)
return encrypted
def encrypt_value(value: str, signing_key: str) -> bytes:
return bytes(_legacy_secret_box(signing_key, "encrypt").encrypt(value.encode("utf-8")))
def decrypt_value(value: bytes, signing_key: str) -> str:
import hashlib
import nacl.secret
import nacl.utils
# get 32 byte master key #
hash_object: Final = hashlib.sha256(signing_key.encode())
hash_bytes: Final = hash_object.digest()
# initialize secret box #
box: Final = nacl.secret.SecretBox(hash_bytes)
# Convert the bytes object to a string
try:
if len(value) == 0:
return ""
plaintext = box.decrypt(value)
plaintext = plaintext.decode("utf-8")
return plaintext
except Exception as e:
raise e
if len(value) == 0:
return ""
if len(value) < _NACL_MIN_CIPHERTEXT_BYTES:
raise ValueError(f"Value of {len(value)} bytes is too short to be a {_ALGO_XSALSA20} ciphertext")
return _legacy_secret_box(signing_key, "decrypt a stored value").decrypt(value).decode("utf-8")
class SecretMapDecodeError(RuntimeError):

41
litellm/proxy/management_endpoints/credential_migration.py
View file

@ -2,9 +2,9 @@
At-rest credential re-encryption migration.
Switches every encrypted-at-rest value from the legacy XSalsa20-Poly1305 (nacl)
format to the versioned AES-256-GCM (``v2:gcm:``) format produced by
``encrypt_decrypt_utils`` when ``general_settings.encryption_algorithm`` is set to
``aes-256-gcm``.
format to the versioned AES-256-GCM (``v3:gcm:``) format produced by
``encrypt_decrypt_utils`` by default (``general_settings.encryption_algorithm``
not set to the legacy ``xsalsa20-poly1305``).
Design properties (see case 2026-06-24 fix plan):
@ -12,7 +12,7 @@ Design properties (see case 2026-06-24 fix plan):
it re-encrypts existing ciphertext under the same derived key but in the new
AES format. This is achieved by decrypting with the format-detecting reader and
re-encrypting through ``encrypt_value_helper`` with the AES gate enabled.
* **Idempotent.** A value already carrying the ``v2:gcm:`` prefix is recognised
* **Idempotent.** A value already carrying a ``v3:gcm:`` or ``v2:gcm:`` prefix is recognised
and left untouched, so re-running the migration is a no-op on migrated rows.
* **Resumable.** Walkers commit per row (or per small table), so an interrupted
run leaves a clean mixed state that a re-run completes.
@ -42,12 +42,13 @@ if TYPE_CHECKING:
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
_ALGO_AES_GCM,
_ENCRYPTION_ALGORITHM_SETTING,
_V2_GCM_PREFIX,
SecretMapDecodeError,
_get_encryption_algorithm,
_get_salt_key,
decode_secret_map,
decrypt_value_helper,
encrypt_value_helper,
is_versioned_gcm,
)
ValueClass = Literal["migrated", "legacy", "plaintext", "undecryptable", "not-a-string"]
@ -59,7 +60,7 @@ class LocationReport:
location: str
scanned: int = 0
migrated: int = 0 # values rewritten to v2 this run
migrated: int = 0 # values rewritten to versioned AES this run
already_v2: int = 0 # values already migrated (skipped)
plaintext: int = 0 # legacy-plaintext values (no ciphertext to migrate)
undecryptable: int = 0 # could not decrypt — preserved, not overwritten
@ -124,15 +125,15 @@ class MigrationReport:
def is_migrated(value: object) -> bool:
"""True if ``value`` is already an AES-256-GCM (``v2:gcm:``) ciphertext."""
return isinstance(value, str) and value.startswith(_V2_GCM_PREFIX)
"""True if ``value`` is already a versioned AES-256-GCM (``v3:gcm:`` or ``v2:gcm:``) ciphertext."""
return isinstance(value, str) and is_versioned_gcm(value)
def classify_value(value: object, key: str = "scan") -> ValueClass:
"""Classify a stored value for the residual scanner.
* ``not-a-string`` — not a string (numbers/bools/None left as-is on disk).
* ``migrated`` — carries the ``v2:gcm:`` prefix.
* ``migrated`` — carries a versioned ``gcm`` prefix.
* ``legacy`` — decrypts under the legacy nacl reader (still needs migrating).
* ``plaintext`` — a non-empty string that does not decrypt and is not v2;
treated as legacy plaintext (nothing to migrate).
@ -144,7 +145,7 @@ def classify_value(value: object, key: str = "scan") -> ValueClass:
return "not-a-string"
if value == "":
return "plaintext"
if value.startswith(_V2_GCM_PREFIX):
if is_versioned_gcm(value):
return "migrated"
decrypted: Final = decrypt_value_helper(value=value, key=key, exception_type="debug", return_original_value=False)
if decrypted is None:
@ -156,14 +157,14 @@ def classify_value(value: object, key: str = "scan") -> ValueClass:
def reencrypt_value(value: object, key: str = "migrate") -> object:
"""Re-encrypt a single stored string into the configured (AES) format.
Returns the value unchanged if it is not a string, is already ``v2:``, or
Returns the value unchanged if it is not a string, is already versioned AES, or
cannot be decrypted (skip-on-undecryptable). Otherwise decrypts under the
format-detecting reader and re-encrypts through ``encrypt_value_helper``
(which writes AES when the gate is on).
"""
if not isinstance(value, str) or value == "":
return value
if value.startswith(_V2_GCM_PREFIX):
if is_versioned_gcm(value):
return value # idempotent: already migrated
decrypted: Final = decrypt_value_helper(value=value, key=key, exception_type="debug", return_original_value=False)
if decrypted is None:
@ -194,13 +195,11 @@ def _assert_aes_gate_enabled() -> None:
Running the migration with the gate off would decrypt then re-encrypt right
back into the legacy format — a no-op that silently fails the migration.
"""
from litellm.proxy.proxy_server import general_settings
algo: Final = general_settings.get(_ENCRYPTION_ALGORITHM_SETTING)
if not (isinstance(algo, str) and algo.lower() == _ALGO_AES_GCM):
algo: Final = _get_encryption_algorithm()
if algo != _ALGO_AES_GCM:
raise RuntimeError(
"Encryption migration requires general_settings.encryption_algorithm: "
f"'{_ALGO_AES_GCM}'. Current value: {algo!r}. Set it before migrating "
f"Encryption migration requires general_settings.{_ENCRYPTION_ALGORITHM_SETTING}: "
f"'{_ALGO_AES_GCM}' (the default). Current value: {algo!r}. Remove the legacy opt-in before migrating "
"so re-encrypted values are written in the AES-256-GCM format."
)
@ -427,7 +426,7 @@ def _classify_callback_value(value: object) -> ValueClass:
Encrypted callback vars carry the ``litellm_enc::`` marker in front of the
ciphertext; strip it, then classify the inner value the same way the
covered-table scanner does (``v2:gcm:`` prefix -> migrated, nacl-decryptable
covered-table scanner does (versioned ``gcm`` prefix -> migrated, nacl-decryptable
-> legacy, otherwise plaintext). Detecting legacy by decrypt rather than by a
re-encrypt delta is what makes the ``check_encryption`` attestation correct
even when run with the AES write gate off.
@ -625,8 +624,8 @@ async def migrate_encryption(
) -> MigrationReport:
"""Run the full at-rest re-encryption migration.
Requires ``general_settings.encryption_algorithm == 'aes-256-gcm'`` so writes
are produced in the AES format. Idempotent and resumable: re-running skips
Requires the proxy to write ``aes-256-gcm`` (the default) so re-encrypted
values are produced in the AES format. Idempotent and resumable: re-running skips
already-migrated values and finishes any partial run.
A ``dry_run`` performs no writes: the covered tables are scanned read-only

4
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -5332,9 +5332,9 @@ async def migrate_encryption_endpoint(
),
):
"""
Re-encrypt all at-rest credentials into the AES-256-GCM (``v2:gcm:``) format.
Re-encrypt all at-rest credentials into the versioned AES-256-GCM (``v3:gcm:``) format.
Admin only. Requires ``general_settings.encryption_algorithm: aes-256-gcm``.
Admin only. Requires the proxy to write ``aes-256-gcm`` (the default).
Idempotent and resumable — re-running skips already-migrated values. Pass
``dry_run=true`` for a non-mutating scan (equivalent to ``--check``).
"""

2
migrations/Dockerfile
View file

@ -55,6 +55,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=litellm-proxy-extras/pyproject.toml,target=litellm-proxy-extras/pyproject.toml \
uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra extra_proxy \
--python python3.13
@ -64,6 +65,7 @@ COPY . .
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra legacy-encryption \
--extra extra_proxy \
--python python3.13

3
pyproject.toml
View file

@ -67,7 +67,6 @@ proxy = [
"PyJWT>=2.13.0,<3.0",
"python-multipart>=0.0.27,<1.0",
"cryptography>=49.0.0,<51.0",
"pynacl>=1.6.2,<2.0",
"websockets>=15.0.1,<16.0",
"boto3>=1.43.1,<2.0",
"azure-identity>=1.25.2,<2.0",
@ -121,6 +120,8 @@ utils = [
"numpydoc>=1.8.0,<2.0",
]
caching = ["diskcache>=5.6.3,<6.0"]
# Reads and writes the legacy XSalsa20-Poly1305 at-rest format; the FIPS image omits it
legacy-encryption = ["pynacl>=1.6.2,<2.0"]
mcp = ["mcp>=2.2.0,<3", "httpx2>=2.5.0,<3", "pydantic>=2.12.0,<3"]
# Driver for the MongoDB Atlas vector store; Atlas Vector Search has no HTTP query API.
# The floor is 4.9 because that is the release AsyncMongoClient landed in.

263
tests/integration/management/test_at_rest_encryption_format.py Normal file
View file

@ -0,0 +1,263 @@
"""At-rest encryption writes the versioned AES-256-GCM format and keeps reading every older format.
The proxy, Postgres and the router run for real. Older rows are built by hand from the wire formats litellm
owns (nacl SecretBox and AES-GCM under the raw SHA-256 derivation) and inserted straight into the model table,
the way a deployment upgraded in place would find them. Every read is proven at the scripted upstream: the
Authorization header it receives is the plaintext key the proxy decrypted.
"""
import base64
import hashlib
import json
import os
import signal
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Final
import httpx
import nacl.secret
import psycopg
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value
from tests.integration._support.database import read_rows
from tests.integration._support.process import owned_proxy_process
V3_PREFIX: Final = "v3:gcm:"
V2_PREFIX: Final = "v2:gcm:"
def _salt_key() -> str:
return os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt")
def _sha256_key() -> bytes:
return hashlib.sha256(_salt_key().encode()).digest()
def _legacy_nacl_ciphertext(plaintext: str) -> str:
sealed: Final = nacl.secret.SecretBox(_sha256_key()).encrypt(plaintext.encode())
return base64.urlsafe_b64encode(bytes(sealed)).decode()
def _legacy_v2_ciphertext(plaintext: str) -> str:
nonce: Final = os.urandom(12)
blob: Final = AESGCM(_sha256_key()).encrypt(nonce, plaintext.encode(), None)
return V2_PREFIX + base64.urlsafe_b64encode(nonce + blob).decode()
def _stored_api_key(model_id: str) -> str:
rows: Final = read_rows(
"SELECT litellm_params->>'api_key' AS api_key FROM \"LiteLLM_ProxyModelTable\" WHERE model_id = %s",
(model_id,),
)
assert len(rows) == 1, rows
return string_value(rows[0]["api_key"])
def _stored_credential_api_key(name: str) -> str:
rows: Final = read_rows(
"SELECT credential_values->>'api_key' AS api_key FROM \"LiteLLM_CredentialsTable\" WHERE credential_name = %s",
(name,),
)
assert len(rows) == 1, rows
return string_value(rows[0]["api_key"])
def _insert_model_row(gateway: Gateway, scenario: Scenario, ciphertext: str) -> tuple[str, str]:
model_id: Final = f"enc-{uuid.uuid4().hex}"
model_name: Final = f"integration-{uuid.uuid4().hex}"
params: Final = {
"model": "openai/gpt-4o-mini",
"api_key": ciphertext,
"api_base": f"{gateway.upstream_url}/v1",
}
with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection:
connection.execute(
'INSERT INTO "LiteLLM_ProxyModelTable" '
"(model_id, model_name, litellm_params, model_info, created_by, updated_by) "
"VALUES (%s, %s, %s::jsonb, %s::jsonb, %s, %s)",
(model_id, model_name, json.dumps(params), json.dumps({"id": model_id}), "integration", "integration"),
)
scenario.cleanups.callback(_delete_model_row_if_present, gateway, model_id)
return model_id, model_name
def _delete_model_row_if_present(gateway: Gateway, model_id: str) -> None:
response: Final = gateway.request("POST", "/model/delete", {"id": model_id})
assert response.status_code in (200, 400, 404), response.text
with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection:
connection.execute('DELETE FROM "LiteLLM_ProxyModelTable" WHERE model_id = %s', (model_id,))
def _new_model(gateway: Gateway, scenario: Scenario, api_key: str) -> tuple[str, str]:
model_name: Final = f"integration-{uuid.uuid4().hex}"
created: Final = gateway.post(
"/model/new",
{
"model_name": model_name,
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": api_key,
"api_base": f"{gateway.upstream_url}/v1",
},
"model_info": {},
},
)
model_id: Final = string_value(object_value(created["model_info"])["id"])
scenario.cleanups.callback(scenario.delete_model, model_id)
return model_id, model_name
def _drain_upstream(gateway: Gateway) -> tuple[tuple[str, str], ...]:
with httpx.Client(base_url=gateway.upstream_url, timeout=15, trust_env=False) as upstream:
observed: Final = upstream.get("/__observations")
observed.raise_for_status()
requests: Final = JSON_OBJECT.validate_json(observed.content)["requests"]
assert isinstance(requests, list), requests
entries: Final = tuple(object_value(value) for value in requests)
return tuple((json.dumps(entry["body"]), string_value(entry["authorization"])) for entry in entries)
def _authorizations_for(observed: tuple[tuple[str, str], ...], marker: str) -> tuple[str, ...]:
return tuple(authorization for body, authorization in observed if marker in body)
def _chat_reaches_upstream_with(gateway: Gateway, model_name: str, plaintext_key: str) -> None:
marker: Final = f"marker-{uuid.uuid4().hex}"
response: Final = eventually(
lambda: gateway.request(
"POST", "/v1/chat/completions", {"model": model_name, "messages": [{"role": "user", "content": marker}]}
),
lambda observed: observed.status_code == 200,
seconds=70,
)
assert response.status_code == 200, response.text
assert _authorizations_for(_drain_upstream(gateway), marker) == (f"Bearer {plaintext_key}",)
def _model_table_count(gateway: Gateway, counter: str) -> int:
report: Final = object_value(gateway.get("/credentials/migrate-encryption/check")["report"])
value: Final = object_value(object_value(report["locations"])["model_table"])[counter]
assert isinstance(value, int), report
return value
def test_new_model_api_key_is_stored_as_v3_gcm_and_decrypts_for_the_upstream_call(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
plaintext_key: Final = f"sk-upstream-{uuid.uuid4().hex}"
model_id, model_name = _new_model(gateway, scenario, plaintext_key)
stored: Final = _stored_api_key(model_id)
assert stored.startswith(V3_PREFIX), f"expected a {V3_PREFIX} ciphertext, stored {stored!r}"
assert plaintext_key not in stored
_chat_reaches_upstream_with(gateway, model_name, plaintext_key)
def test_new_credential_values_are_stored_as_v3_gcm(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
name: Final = f"credential-{uuid.uuid4().hex}"
plaintext_key: Final = f"sk-credential-{uuid.uuid4().hex}"
gateway.post(
"/credentials",
{"credential_name": name, "credential_values": {"api_key": plaintext_key}, "credential_info": {}},
)
scenario.cleanups.callback(gateway.request, "DELETE", f"/credentials/{name}")
stored: Final = _stored_credential_api_key(name)
assert stored.startswith(V3_PREFIX), f"expected a {V3_PREFIX} ciphertext, stored {stored!r}"
assert plaintext_key not in stored
def test_legacy_nacl_model_row_still_decrypts_for_the_upstream_call(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
plaintext_key: Final = f"sk-nacl-{uuid.uuid4().hex}"
_, model_name = _insert_model_row(gateway, scenario, _legacy_nacl_ciphertext(plaintext_key))
_chat_reaches_upstream_with(gateway, model_name, plaintext_key)
def test_v2_gcm_model_row_still_decrypts_for_the_upstream_call(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
plaintext_key: Final = f"sk-v2-{uuid.uuid4().hex}"
_, model_name = _insert_model_row(gateway, scenario, _legacy_v2_ciphertext(plaintext_key))
_chat_reaches_upstream_with(gateway, model_name, plaintext_key)
def test_encryption_check_counts_v3_as_migrated_and_nacl_as_legacy(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
migrated_before: Final = _model_table_count(gateway, "already_v2")
legacy_before: Final = _model_table_count(gateway, "legacy")
_new_model(gateway, scenario, f"sk-upstream-{uuid.uuid4().hex}")
_insert_model_row(gateway, scenario, _legacy_nacl_ciphertext(f"sk-nacl-{uuid.uuid4().hex}"))
_insert_model_row(gateway, scenario, _legacy_v2_ciphertext(f"sk-v2-{uuid.uuid4().hex}"))
migrated_after: Final = _model_table_count(gateway, "already_v2")
assert migrated_after == migrated_before + 3 + 1, (
"expected every litellm_params value of the new model (model, api_key, api_base) plus the v2 row "
f"counted as migrated, before {migrated_before} after {migrated_after}"
)
assert _model_table_count(gateway, "legacy") == legacy_before + 1
def test_migrate_encryption_runs_under_default_settings_and_rewrites_legacy_rows(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
plaintext_key: Final = f"sk-nacl-{uuid.uuid4().hex}"
model_id, model_name = _insert_model_row(gateway, scenario, _legacy_nacl_ciphertext(plaintext_key))
migrated: Final = gateway.request("POST", "/credentials/migrate-encryption", {})
assert migrated.status_code == 200, migrated.text
rewritten: Final = _stored_api_key(model_id)
assert rewritten.startswith(V3_PREFIX), f"expected a {V3_PREFIX} ciphertext, stored {rewritten!r}"
_chat_reaches_upstream_with(gateway, model_name, plaintext_key)
def test_undecryptable_model_row_does_not_stop_other_models_from_serving(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
_insert_model_row(gateway, scenario, base64.b64encode(os.urandom(48)).decode())
healthy_key: Final = f"sk-upstream-{uuid.uuid4().hex}"
_, healthy_model = _new_model(gateway, scenario, healthy_key)
_chat_reaches_upstream_with(gateway, healthy_model, healthy_key)
assert gateway.request("GET", "/health/liveliness").status_code == 200
def _burst_request(gateway: Gateway, model: str, marker: str) -> tuple[str, int]:
try:
response: Final = gateway.request(
"POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": marker}]}
)
return marker, response.status_code
except httpx.TransportError:
return marker, 0
def test_proxy_restart_mid_burst_reloads_every_encryption_format_from_the_database(
gateway: Gateway, tmp_path: Path
) -> None:
with gateway.scenario() as scenario:
keys: Final = (f"sk-nacl-{uuid.uuid4().hex}", f"sk-v2-{uuid.uuid4().hex}", f"sk-upstream-{uuid.uuid4().hex}")
_, nacl_model = _insert_model_row(gateway, scenario, _legacy_nacl_ciphertext(keys[0]))
_, v2_model = _insert_model_row(gateway, scenario, _legacy_v2_ciphertext(keys[1]))
_, v3_model = _new_model(gateway, scenario, keys[2])
models: Final = (nacl_model, v2_model, v3_model)
expected: Final = dict(zip(models, (f"Bearer {key}" for key in keys), strict=True))
markers: Final = tuple(f"burst-{uuid.uuid4().hex}" for _ in range(24))
with owned_proxy_process(gateway, tmp_path, {}, workers=2) as first:
for model, key in zip(models, keys, strict=True):
_chat_reaches_upstream_with(first.gateway, model, key)
with ThreadPoolExecutor(max_workers=24) as pool:
futures: Final = tuple(
pool.submit(_burst_request, first.gateway, models[index % 3], marker)
for index, marker in enumerate(markers)
)
eventually(lambda: sum(future.done() for future in futures), lambda done: done >= 6, seconds=60)
os.killpg(first.process.pid, signal.SIGTERM)
outcomes: Final = tuple(future.result(timeout=120) for future in futures)
served: Final = frozenset(marker for marker, status in outcomes if status == 200)
assert served, outcomes
observed: Final = _drain_upstream(gateway)
for index, marker in enumerate(markers):
seen = _authorizations_for(observed, marker)
assert len(seen) <= 1, (marker, seen)
if marker in served:
assert seen == (expected[models[index % 3]],), (marker, seen)
with owned_proxy_process(gateway, tmp_path, {}, workers=2) as restarted:
for model, key in zip(models, keys, strict=True):
_chat_reaches_upstream_with(restarted.gateway, model, key)

5
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_envelope.py
View file

@ -50,7 +50,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import
open_refresh_envelope,
user_identity,
)
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value, encrypt_value
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_if_encrypted_with, encrypt_value
_NOW = datetime(2026, 7, 9, 12, 0, 0, tzinfo=timezone.utc)
_SIGNING_KEY = "unit-test-signing-key-0123456789abcdef0123456789abcdef"
@ -136,7 +136,8 @@ def test_minimal_grant_round_trips_without_none_leakage_into_claims():
claims = _unverified_claims(token)
blob = claims["grant"]
assert isinstance(blob, str)
plaintext = decrypt_value(value=base64.urlsafe_b64decode(blob), signing_key=_ENCRYPTION_KEY)
plaintext = decrypt_if_encrypted_with(blob, _ENCRYPTION_KEY)
assert plaintext is not None
assert set(json.loads(plaintext)) == {"access_token", "token_type"}
opened = open_envelope(token, _KEYS, _NOW)
assert isinstance(opened, OpenedEnvelope)

2
tests/test_litellm/proxy/_experimental/mcp_server/test_db_credentials.py
View file

@ -472,7 +472,7 @@ async def test_secret_maps_create_update_round_trip(map_algorithm: str, field: s
created: Final = await create_mcp_server(prisma, create, touched_by="test")
first: Final = table.rows["srv-map"][field]
assert isinstance(first, str) and isinstance(json.loads(first), str)
assert json.loads(first).startswith("v2:gcm:") is (map_algorithm == "aes-256-gcm")
assert json.loads(first).startswith("v3:gcm:") is (map_algorithm == "aes-256-gcm")
assert "sensitive-secret" not in first and "TEMPLATE" not in first
assert getattr(created, field) == original == getattr(create, field)
assert decode_secret_map(first, key=field) == original

276
tests/test_litellm/proxy/common_utils/test_encrypt_decrypt_utils.py
View file

@ -1,148 +1,247 @@
"""
Tests for the at-rest credential encryption chokepoint.
Covers the AES-256-GCM (``v2:gcm:``) path, the ``encryption_algorithm`` config
gate, and the backward-compatibility guarantees that let legacy XSalsa20-Poly1305
(nacl) ciphertext and new AES values coexist and decrypt correctly.
Covers the AES-256-GCM default (``v3:gcm:`` written under HKDF-SHA256), the
``encryption_algorithm`` legacy opt-in and its FIPS refusal, and the dual-read
guarantees that keep ``v2:gcm:`` (raw SHA-256 AES) and unprefixed
XSalsa20-Poly1305 (nacl) ciphertext decrypting after the default flip.
"""
import base64
import hashlib
import os
import sys
import pytest
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from litellm.proxy import proxy_server
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
_V2_GCM_PREFIX,
_V3_GCM_PREFIX,
LegacyEncryptionUnavailableError,
_get_encryption_algorithm,
decrypt_if_encrypted_with,
decrypt_value_helper,
encrypt_value,
encrypt_value_helper,
is_versioned_gcm,
)
from litellm.proxy.common_utils.fips import FipsModeError
_SALT_KEY = "sk-salt-aes-1234"
def _use_aes(monkeypatch):
"""Flip the write-time algorithm to AES-256-GCM for the duration of a test."""
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "aes-256-gcm"})
def _use_legacy(monkeypatch):
"""Opt the write-time algorithm back into XSalsa20-Poly1305 for the duration of a test."""
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"})
def _sha256_key(signing_key: str = _SALT_KEY) -> bytes:
return hashlib.sha256(signing_key.encode()).digest()
def _hkdf_key(signing_key: str = _SALT_KEY) -> bytes:
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=b"litellm-at-rest-v3").derive(
signing_key.encode()
)
def _v2_ciphertext(plaintext: str, key: bytes) -> str:
"""Build a ``v2:gcm:`` value the way the previous release wrote it: AES-GCM under raw SHA-256."""
nonce = os.urandom(12)
return (
_V2_GCM_PREFIX + base64.urlsafe_b64encode(nonce + AESGCM(key).encrypt(nonce, plaintext.encode(), None)).decode()
)
def _legacy_nacl_ciphertext(plaintext: str, key: bytes) -> str:
"""Build an unprefixed value the way pre-AES releases wrote it: XSalsa20-Poly1305 under raw SHA-256."""
import nacl.secret
return base64.urlsafe_b64encode(bytes(nacl.secret.SecretBox(key).encrypt(plaintext.encode()))).decode()
@pytest.fixture(autouse=True)
def _salt_key(monkeypatch):
# Dominant convention in the test_litellm/ tree: set the key via env.
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-salt-aes-1234")
# Ensure the legacy default is in force unless a test opts into AES.
monkeypatch.setenv("LITELLM_SALT_KEY", _SALT_KEY)
monkeypatch.delenv("LITELLM_FIPS_MODE", raising=False)
monkeypatch.setattr(proxy_server, "general_settings", {})
yield
def test_aes_gcm_round_trip(monkeypatch):
"""A value written under AES-256-GCM is tagged v2:gcm: and decrypts back."""
_use_aes(monkeypatch)
def test_default_write_is_v3_gcm_and_round_trips():
ct = encrypt_value_helper("super-secret")
assert ct.startswith(_V2_GCM_PREFIX)
assert ct.startswith(_V3_GCM_PREFIX), ct
assert decrypt_value_helper(ct, key="t") == "super-secret"
def test_default_is_legacy_algorithm(monkeypatch):
"""With no config, writes stay on the legacy algorithm (no v2: marker)."""
ct = encrypt_value_helper("legacy-secret")
def test_explicit_aes_setting_writes_v3_gcm(monkeypatch):
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "AES-256-GCM"})
assert not ct.startswith(_V2_GCM_PREFIX)
assert decrypt_value_helper(ct, key="t") == "legacy-secret"
assert encrypt_value_helper("secret").startswith(_V3_GCM_PREFIX)
def test_legacy_nacl_value_still_decrypts_after_flag_flip(monkeypatch):
"""A value written under the old algorithm decrypts unchanged once AES is on.
This is the mixed-format readback guarantee: decrypt is format-detecting, so
flipping the flag forward never strands previously-written data.
"""
legacy = encrypt_value_helper("legacy-secret") # default = xsalsa20
assert not legacy.startswith(_V2_GCM_PREFIX)
_use_aes(monkeypatch)
# New writes are now AES, but the old value must still come back.
assert decrypt_value_helper(legacy, key="t") == "legacy-secret"
assert encrypt_value_helper("fresh").startswith(_V2_GCM_PREFIX)
def test_v2_prefix_is_idempotent_marker(monkeypatch):
"""The migration's skip-check: an already-v2 value is recognized by its prefix.
Re-encrypting an AES value yields a fresh (different nonce) AES value, but the
prefix is what lets a migration skip already-migrated rows without decrypting.
"""
_use_aes(monkeypatch)
def test_unknown_algorithm_falls_back_to_the_aes_default(monkeypatch):
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "rot13"})
ct = encrypt_value_helper("secret")
assert ct.startswith(_V2_GCM_PREFIX)
assert ct.startswith(_V3_GCM_PREFIX), ct
assert decrypt_value_helper(ct, key="t") == "secret"
def test_v3_value_is_aes_gcm_under_hkdf_sha256_not_raw_sha256():
ct = encrypt_value_helper("hkdf-secret")
raw = base64.urlsafe_b64decode(ct[len(_V3_GCM_PREFIX) :])
nonce, blob = raw[:12], raw[12:]
assert AESGCM(_hkdf_key()).decrypt(nonce, blob, None) == b"hkdf-secret"
with pytest.raises(InvalidTag):
AESGCM(_sha256_key()).decrypt(nonce, blob, None)
def test_each_v3_write_uses_a_fresh_nonce():
first, second = encrypt_value_helper("same-secret"), encrypt_value_helper("same-secret")
assert first != second
assert (
base64.urlsafe_b64decode(first[len(_V3_GCM_PREFIX) :])[:12]
!= base64.urlsafe_b64decode(second[len(_V3_GCM_PREFIX) :])[:12]
)
def test_v2_value_written_under_raw_sha256_still_decrypts():
legacy_v2 = _v2_ciphertext("v2-secret", _sha256_key())
assert decrypt_value_helper(legacy_v2, key="t") == "v2-secret"
assert decrypt_if_encrypted_with(legacy_v2, _SALT_KEY) == "v2-secret"
def test_v2_value_is_not_read_with_the_v3_derivation():
"""A v2 body relabelled as v3 must fail: the prefix, not the caller, picks the derivation."""
v2_body = _v2_ciphertext("v2-secret", _sha256_key())[len(_V2_GCM_PREFIX) :]
assert decrypt_if_encrypted_with(_V3_GCM_PREFIX + v2_body, _SALT_KEY) is None
assert decrypt_value_helper(_V3_GCM_PREFIX + v2_body, key="t", exception_type="debug") is None
def test_legacy_nacl_value_still_decrypts_under_the_aes_default():
legacy = _legacy_nacl_ciphertext("legacy-secret", _sha256_key())
assert not is_versioned_gcm(legacy)
assert decrypt_value_helper(legacy, key="t") == "legacy-secret"
assert decrypt_if_encrypted_with(legacy, _SALT_KEY) == "legacy-secret"
assert encrypt_value_helper("fresh").startswith(_V3_GCM_PREFIX)
def test_legacy_opt_in_writes_unprefixed_nacl_that_still_reads_back(monkeypatch):
_use_legacy(monkeypatch)
ct = encrypt_value_helper("legacy-secret")
assert not is_versioned_gcm(ct), ct
assert decrypt_value_helper(ct, key="t") == "legacy-secret"
assert _get_encryption_algorithm() == "xsalsa20-poly1305"
def test_legacy_opt_in_is_refused_under_fips_mode(monkeypatch):
_use_legacy(monkeypatch)
monkeypatch.setenv("LITELLM_FIPS_MODE", "true")
with pytest.raises(FipsModeError, match="xsalsa20-poly1305") as refused:
encrypt_value_helper("secret")
assert "aes-256-gcm" in str(refused.value)
def test_fips_mode_with_the_default_setting_writes_v3_gcm(monkeypatch):
monkeypatch.setenv("LITELLM_FIPS_MODE", "true")
ct = encrypt_value_helper("fips-secret")
assert ct.startswith(_V3_GCM_PREFIX), ct
assert decrypt_value_helper(ct, key="t") == "fips-secret"
def test_versioned_gcm_values_never_import_nacl(monkeypatch):
legacy_v2 = _v2_ciphertext("v2-secret", _sha256_key())
monkeypatch.setitem(sys.modules, "nacl", None)
monkeypatch.setitem(sys.modules, "nacl.secret", None)
assert decrypt_value_helper(encrypt_value_helper("v3-secret"), key="t") == "v3-secret"
assert decrypt_value_helper(legacy_v2, key="t") == "v2-secret"
def test_legacy_ciphertext_without_pynacl_names_the_reencrypt_path(monkeypatch):
legacy = _legacy_nacl_ciphertext("legacy-secret", _sha256_key())
monkeypatch.setitem(sys.modules, "nacl", None)
monkeypatch.setitem(sys.modules, "nacl.secret", None)
with pytest.raises(LegacyEncryptionUnavailableError) as missing:
decrypt_value_helper(legacy, key="t", exception_type="debug", return_original_value=True)
message = str(missing.value)
assert "PyNaCl is not installed" in message
assert "legacy-encryption" in message
assert "/credentials/migrate-encryption" in message
with pytest.raises(LegacyEncryptionUnavailableError):
decrypt_if_encrypted_with(legacy, _SALT_KEY)
def test_legacy_opt_in_without_pynacl_fails_the_write_not_silently(monkeypatch):
_use_legacy(monkeypatch)
monkeypatch.setitem(sys.modules, "nacl", None)
monkeypatch.setitem(sys.modules, "nacl.secret", None)
with pytest.raises(LegacyEncryptionUnavailableError, match="encrypt"):
encrypt_value_helper("secret")
def test_v3_prefix_is_the_idempotent_migration_marker():
ct = encrypt_value_helper("secret")
assert is_versioned_gcm(ct)
# Round-tripping does not change the plaintext, and the marker is stable.
again = encrypt_value_helper(decrypt_value_helper(ct, key="t"))
assert again.startswith(_V2_GCM_PREFIX)
assert again.startswith(_V3_GCM_PREFIX)
assert decrypt_value_helper(again, key="t") == "secret"
def test_aes_decrypt_failure_returns_none_not_raise(monkeypatch):
"""Decrypt contract preserved: a garbled v2 value returns None, never raises."""
_use_aes(monkeypatch)
garbled = _V2_GCM_PREFIX + "not-valid-base64-or-ciphertext!!!"
# exception_type="debug" exercises the swallow path; must not raise.
@pytest.mark.parametrize("prefix", [_V3_GCM_PREFIX, _V2_GCM_PREFIX])
def test_aes_decrypt_failure_returns_none_not_raise(prefix: str):
garbled = prefix + "not-valid-base64-or-ciphertext!!!"
assert decrypt_value_helper(garbled, key="t", exception_type="debug") is None
def test_aes_decrypt_failure_returns_original_when_requested(monkeypatch):
"""With return_original_value=True a bad v2 value comes back as-is, not None."""
_use_aes(monkeypatch)
garbled = _V2_GCM_PREFIX + "###"
def test_aes_decrypt_failure_returns_original_when_requested():
garbled = _V3_GCM_PREFIX + "###"
assert decrypt_value_helper(garbled, key="t", exception_type="debug", return_original_value=True) == garbled
def test_empty_string_round_trips_under_aes(monkeypatch):
"""Empty string is preserved through the AES path (parity with legacy)."""
_use_aes(monkeypatch)
def test_empty_string_round_trips_under_aes():
ct = encrypt_value_helper("")
assert ct.startswith(_V2_GCM_PREFIX)
assert ct.startswith(_V3_GCM_PREFIX)
assert decrypt_value_helper(ct, key="t") == ""
def test_callback_prefix_composes_with_v2(monkeypatch):
"""litellm_enc:: + v2:gcm:... round-trips through the callback read path.
Callback vars are stored as ``litellm_enc::<helper output>``; the read path
strips ``litellm_enc::`` then calls the helper, so the value handed to the
helper is ``v2:gcm:...``. Ordering must work end to end.
"""
def test_callback_prefix_composes_with_v3():
"""litellm_enc:: + v3:gcm:... round-trips through the callback read path."""
from litellm.proxy.common_utils.callback_utils import (
_CALLBACK_VAR_ENCRYPTED_PREFIX,
_decrypt_or_passthrough,
_encrypt_if_plaintext,
)
_use_aes(monkeypatch)
# "gcs_path_service_account" is a known-sensitive callback key.
stored = _encrypt_if_plaintext("gcs_path_service_account", "my-sa-secret")
assert stored.startswith(_CALLBACK_VAR_ENCRYPTED_PREFIX)
inner = stored[len(_CALLBACK_VAR_ENCRYPTED_PREFIX) :]
assert inner.startswith(_V2_GCM_PREFIX)
assert inner.startswith(_V3_GCM_PREFIX)
assert _decrypt_or_passthrough("gcs_path_service_account", stored) == "my-sa-secret"
def test_unknown_algorithm_falls_back_to_legacy(monkeypatch):
"""An unrecognized encryption_algorithm value does not produce v2 writes."""
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "rot13"})
ct = encrypt_value_helper("secret")
assert not ct.startswith(_V2_GCM_PREFIX)
assert decrypt_value_helper(ct, key="t") == "secret"
def test_decrypt_failure_debug_log_omits_raw_value(monkeypatch):
"""Regression for LIT-4152: the decrypt-failure debug breadcrumb must not
embed the raw value.
@ -191,10 +290,10 @@ def test_decrypt_failure_debug_log_omits_raw_value(monkeypatch):
assert result == secret
@pytest.mark.parametrize("use_aes", [False, True])
def test_explicit_key_decrypt_reads_only_values_written_under_that_key(monkeypatch, use_aes: bool):
if use_aes:
_use_aes(monkeypatch)
@pytest.mark.parametrize("use_legacy", [False, True])
def test_explicit_key_decrypt_reads_only_values_written_under_that_key(monkeypatch, use_legacy: bool):
if use_legacy:
_use_legacy(monkeypatch)
written_with_previous_key = encrypt_value_helper("stored-secret", new_encryption_key="sk-1234")
assert decrypt_if_encrypted_with(written_with_previous_key, "sk-1234") == "stored-secret"
@ -209,6 +308,7 @@ def test_explicit_key_decrypt_reads_only_values_written_under_that_key(monkeypat
"gpt-5.4-mini",
"https://example.invalid/v1",
"v2:gcm:",
"v3:gcm:",
"aGVsbG8=",
"*",
"-",
@ -224,10 +324,10 @@ def test_explicit_key_decrypt_rejects_values_that_are_not_ciphertexts(not_a_ciph
assert decrypt_if_encrypted_with(not_a_ciphertext, "sk-1234") is None
@pytest.mark.parametrize("use_aes", [False, True])
def test_explicit_key_decrypt_tells_an_encrypted_empty_string_from_no_ciphertext(monkeypatch, use_aes: bool):
if use_aes:
_use_aes(monkeypatch)
@pytest.mark.parametrize("use_legacy", [False, True])
def test_explicit_key_decrypt_tells_an_encrypted_empty_string_from_no_ciphertext(monkeypatch, use_legacy: bool):
if use_legacy:
_use_legacy(monkeypatch)
assert decrypt_if_encrypted_with(encrypt_value_helper("", new_encryption_key="sk-1234"), "sk-1234") == ""

32
tests/test_litellm/proxy/management_endpoints/test_credential_migration.py
View file

@ -15,7 +15,7 @@ import pytest
from litellm.proxy import proxy_server
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
_V2_GCM_PREFIX,
_V3_GCM_PREFIX,
encrypt_value_helper,
)
from litellm.proxy.management_endpoints import credential_migration as cm
@ -29,8 +29,8 @@ def salt_key(monkeypatch):
def _legacy_ct(value: str, monkeypatch) -> str:
"""Produce a legacy (nacl) ciphertext with the AES gate off."""
monkeypatch.setattr(proxy_server, "general_settings", {})
"""Produce a legacy (nacl) ciphertext through the explicit xsalsa20-poly1305 opt-in."""
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"})
return encrypt_value_helper(value)
@ -79,7 +79,7 @@ def test_reencrypt_value_legacy_to_v2(salt_key, monkeypatch):
out = cm.reencrypt_value(legacy)
assert out != legacy
assert out.startswith(_V2_GCM_PREFIX)
assert out.startswith(_V3_GCM_PREFIX)
def test_reencrypt_value_is_idempotent(salt_key, monkeypatch):
@ -110,7 +110,7 @@ def test_reencrypt_selective_dict(salt_key, monkeypatch):
data = {"api_key": legacy_key, "base_url": "https://x", "integration_token": None}
out = cm.reencrypt_selective_dict(data, ["api_key", "integration_token"])
assert out["api_key"].startswith(_V2_GCM_PREFIX)
assert out["api_key"].startswith(_V3_GCM_PREFIX)
assert out["base_url"] == "https://x" # untouched non-sensitive
assert out["integration_token"] is None # null skipped
@ -120,7 +120,7 @@ def test_reencrypt_selective_dict(salt_key, monkeypatch):
@pytest.mark.asyncio
async def test_migrate_requires_aes_gate(salt_key, monkeypatch):
monkeypatch.setattr(proxy_server, "general_settings", {}) # gate off
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"}) # legacy opt-in
with pytest.raises(RuntimeError, match="encryption_algorithm"):
await cm.migrate_encryption(
prisma_client=MagicMock(), user_api_key_dict=MagicMock()
@ -161,7 +161,7 @@ async def test_vantage_walker_migrates_legacy_field(salt_key, monkeypatch):
written = json.loads(
client.db.litellm_config.update.call_args.kwargs["data"]["param_value"]
)
assert written["api_key"].startswith(_V2_GCM_PREFIX)
assert written["api_key"].startswith(_V3_GCM_PREFIX)
assert written["base_url"] == "https://api.vantage.sh" # non-sensitive untouched
@ -305,8 +305,8 @@ async def test_callback_vars_walker_migrates_team_metadata(salt_key, monkeypatch
"""A team row with a legacy-encrypted callback var is rewritten to v2."""
from litellm.proxy.common_utils.callback_utils import encrypt_callback_vars
# Legacy-encrypt a callback var via the real callback path (gate off).
monkeypatch.setattr(proxy_server, "general_settings", {})
# Legacy-encrypt a callback var via the real callback path (legacy opt-in).
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"})
legacy_meta = encrypt_callback_vars(
{"logging": [{"callback_vars": {"gcs_path_service_account": "sa-secret"}}]}
)
@ -326,7 +326,7 @@ async def test_callback_vars_walker_migrates_team_metadata(salt_key, monkeypatch
client.db.litellm_teamtable.update.call_args.kwargs["data"]["metadata"]
)
inner = written["logging"][0]["callback_vars"]["gcs_path_service_account"]
assert "v2:gcm:" in inner
assert "v3:gcm:" in inner
@pytest.mark.asyncio
@ -334,7 +334,7 @@ async def test_callback_vars_walker_dry_run_reports_legacy(salt_key, monkeypatch
"""In --check (dry-run) mode, a legacy callback var counts as residual legacy."""
from litellm.proxy.common_utils.callback_utils import encrypt_callback_vars
monkeypatch.setattr(proxy_server, "general_settings", {})
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"})
legacy_meta = encrypt_callback_vars(
{"logging": [{"callback_vars": {"gcs_path_service_account": "sa-secret"}}]}
)
@ -366,7 +366,7 @@ async def test_callback_vars_walker_migrates_callback_settings_shape(
"""
from litellm.proxy.common_utils.callback_utils import encrypt_callback_vars
monkeypatch.setattr(proxy_server, "general_settings", {})
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"})
legacy_meta = encrypt_callback_vars(
{
"callback_settings": {
@ -391,7 +391,7 @@ async def test_callback_vars_walker_migrates_callback_settings_shape(
client.db.litellm_teamtable.update.call_args.kwargs["data"]["metadata"]
)
inner = written["callback_settings"]["callback_vars"]["gcs_path_service_account"]
assert "v2:gcm:" in inner
assert "v3:gcm:" in inner
@pytest.mark.asyncio
@ -401,13 +401,13 @@ async def test_check_reports_callback_var_legacy_with_gate_off(salt_key, monkeyp
Detection is decrypt-based, not a re-encrypt delta, so it does not depend on
the write gate. A heuristic that re-encrypts and counts new v2 values would
read zero here (gate off -> no v2 produced) and emit a false-clean
read zero here (legacy opt-in -> no versioned AES produced) and emit a false-clean
attestation -- exactly the compliance trap this guards against.
"""
from litellm.proxy.common_utils.callback_utils import encrypt_callback_vars
# Legacy-encrypt a callback var, and leave the gate OFF for the check itself.
monkeypatch.setattr(proxy_server, "general_settings", {})
# Legacy-encrypt a callback var, and keep the legacy opt-in for the check itself.
monkeypatch.setattr(proxy_server, "general_settings", {"encryption_algorithm": "xsalsa20-poly1305"})
legacy_meta = encrypt_callback_vars(
{"logging": [{"callback_vars": {"gcs_path_service_account": "sa-secret"}}]}
)

4
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -3646,9 +3646,9 @@ export interface paths {
put?: never;
/**
* Migrate Encryption Endpoint
* @description Re-encrypt all at-rest credentials into the AES-256-GCM (``v2:gcm:``) format.
* @description Re-encrypt all at-rest credentials into the versioned AES-256-GCM (``v3:gcm:``) format.
*
* Admin only. Requires ``general_settings.encryption_algorithm: aes-256-gcm``.
* Admin only. Requires the proxy to write ``aes-256-gcm`` (the default).
* Idempotent and resumable — re-running skips already-migrated values. Pass
* ``dry_run=true`` for a non-mutating scan (equivalent to ``--check``).
*/

8
uv.lock generated
View file

@ -4555,6 +4555,9 @@ google = [
grpc = [
{ name = "grpcio" },
]
legacy-encryption = [
{ name = "pynacl" },
]
mcp = [
{ name = "httpx2" },
{ name = "mcp" },
@ -4585,7 +4588,6 @@ proxy = [
{ name = "polars" },
{ name = "pydantic" },
{ name = "pyjwt" },
{ name = "pynacl" },
{ name = "pyroscope-io", marker = "sys_platform != 'win32'" },
{ name = "python-multipart" },
{ name = "pyyaml" },
@ -4823,7 +4825,7 @@ requires-dist = [
{ name = "pydantic", marker = "extra == 'proxy'", specifier = ">=2.12.0,<3" },
{ name = "pydantic-settings", specifier = ">=2.14.1,<3.0" },
{ name = "pyjwt", marker = "extra == 'proxy'", specifier = ">=2.13.0,<3.0" },
{ name = "pynacl", marker = "extra == 'proxy'", specifier = ">=1.6.2,<2.0" },
{ name = "pynacl", marker = "extra == 'legacy-encryption'", specifier = ">=1.6.2,<2.0" },
{ name = "pypdf", marker = "extra == 'proxy-runtime'", specifier = ">=6.16.1,<7.0" },
{ name = "pyroscope-io", marker = "sys_platform != 'win32' and extra == 'proxy'", specifier = ">=0.8.16,<1.0" },
{ name = "python-dotenv", specifier = ">=1.0.0,<2.0" },
@ -4853,7 +4855,7 @@ requires-dist = [
{ name = "uvloop", marker = "sys_platform != 'win32' and extra == 'proxy'", specifier = ">=0.22.1,<1.0" },
{ name = "websockets", marker = "extra == 'proxy'", specifier = ">=15.0.1,<16.0" },
]
provides-extras = ["proxy", "cli", "extra-proxy", "utils", "caching", "mcp", "saml", "semantic-router", "mlflow", "grpc", "stt-vertex-chirp", "stt-nvidia-riva", "google", "bedrock-realtime", "proxy-runtime"]
provides-extras = ["proxy", "cli", "extra-proxy", "utils", "caching", "legacy-encryption", "mcp", "saml", "semantic-router", "mlflow", "grpc", "stt-vertex-chirp", "stt-nvidia-riva", "google", "bedrock-realtime", "proxy-runtime"]
[package.metadata.requires-dev]
benchmarks = [