fix(proxy): re-encrypt callback_vars, vantage/cloudzero and SSO secrets on master key rotation

litellm/proxy/common_utils/callback_utils.py

@@ -1,4 +1,5 @@
 import copy
+from functools import partial
 from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Literal, Optional
 
 import litellm
@@ -579,13 +580,17 @@ def normalize_callback_names(callbacks: Iterable[Any]) -> List[Any]:
     return [c.lower() if isinstance(c, str) else c for c in callbacks]
 
 
-def encrypt_callback_vars(metadata: Any) -> Any:
+def encrypt_callback_vars(metadata: Any, new_encryption_key: Optional[str] = None) -> Any:
     """Return a deep copy of metadata with callback_vars values encrypted at rest.
 
     Idempotent: a value that already decrypts cleanly is left unchanged so
     round-trips through edit forms don't double-encrypt.
+
+    ``new_encryption_key`` re-encrypts under a different key than the one
+    currently in memory; master-key rotation passes the new key here so the
+    decrypt-then-encrypt round trip lands ciphertext readable under the new key.
     """
-    return _transform_callback_vars(metadata, _encrypt_if_plaintext)
+    return _transform_callback_vars(metadata, partial(_encrypt_if_plaintext, new_encryption_key=new_encryption_key))
 
 
 def decrypt_callback_vars(metadata: Any) -> Any:
@@ -626,7 +631,7 @@ def is_sensitive_callback_key(
     return _CALLBACK_VAR_MASKER.is_sensitive_key(key)
 
 
-def _encrypt_if_plaintext(key: str, value: Any) -> Any:
+def _encrypt_if_plaintext(key: str, value: Any, new_encryption_key: Optional[str] = None) -> Any:
     if not isinstance(value, str) or not value:
         return value
     if not is_sensitive_callback_key(key):
@@ -639,7 +644,7 @@ def _encrypt_if_plaintext(key: str, value: Any) -> Any:
         # plaintext under K2 and wrap them a second time.
         return value
     try:
-        return _CALLBACK_VAR_ENCRYPTED_PREFIX + encrypt_value_helper(value)
+        return _CALLBACK_VAR_ENCRYPTED_PREFIX + encrypt_value_helper(value, new_encryption_key=new_encryption_key)
     except Exception:
         # No salt key / master key configured — leave the value as-is rather
         # than crash the write. Dev environments without LITELLM_SALT_KEY hit

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -56,6 +56,10 @@ from litellm.proxy.common_utils.callback_utils import (
     decrypt_callback_vars,
     encrypt_callback_vars,
 )
+from litellm.proxy.common_utils.encrypt_decrypt_utils import (
+    decrypt_value_helper,
+    encrypt_value_helper,
+)
 from litellm.proxy.common_utils.rbac_utils import check_org_admin_can_generate_keys
 from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
 from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
@@ -104,6 +108,7 @@ from litellm.repositories.model_repository import ModelRepository
 from litellm.repositories.table_repositories import (
     DeletedVerificationTokenRepository,
     DeprecatedVerificationTokenRepository,
+    SSOConfigRepository,
 )
 from litellm.repositories.team_repository import TeamRepository
 from litellm.repositories.user_repository import UserRepository
@@ -4110,6 +4115,116 @@ async def delete_key_aliases(
     )
 
 
+def _reencrypt_secret_field(field_name: str, value: Any, new_master_key: str) -> Any:
+    """Decrypt one at-rest field with the current key and re-encrypt it with the
+    new key.
+
+    Non-strings, empty strings, and values that do not decrypt pass through
+    unchanged so a rotation never corrupts a value or double-encrypts one that
+    was already ciphertext under an unknown key.
+    """
+    if not isinstance(value, str) or not value:
+        return value
+    decrypted = decrypt_value_helper(value=value, key=field_name, exception_type="debug", return_original_value=False)
+    if decrypted is None:
+        return value
+    return encrypt_value_helper(decrypted, new_encryption_key=new_master_key)
+
+
+async def _rotate_callback_vars(
+    prisma_client: PrismaClient,
+    table_name: Literal["team", "verification_token"],
+    new_master_key: str,
+) -> None:
+    """Re-encrypt ``callback_vars`` credentials stored on team / verification-token
+    metadata under the new master key.
+
+    Both metadata shapes are covered: ``metadata.logging[*].callback_vars`` and
+    the top-level ``metadata.callback_settings.callback_vars`` (e.g.
+    ``LANGFUSE_SECRET_KEY``). Values decrypt with the current in-memory key and
+    re-encrypt with ``new_master_key``, so they stay readable once the master key
+    is swapped. A row whose transform fails is preserved, never dropped.
+    """
+    if table_name == "team":
+        table = TeamRepository(prisma_client).table
+        pk = "team_id"
+    else:
+        table = VerificationTokenRepository(prisma_client).table
+        pk = "token"
+
+    rows = await table.find_many()
+    for row in rows or []:
+        metadata = getattr(row, "metadata", None)
+        if isinstance(metadata, str):
+            metadata = json.loads(metadata)
+        if not isinstance(metadata, dict) or ("logging" not in metadata and "callback_settings" not in metadata):
+            continue
+        re_encrypted = encrypt_callback_vars(decrypt_callback_vars(metadata), new_encryption_key=new_master_key)
+        if re_encrypted == metadata:
+            continue
+        await table.update(
+            where={pk: getattr(row, pk)},
+            data={"metadata": safe_dumps(re_encrypted)},
+        )
+
+
+async def _rotate_config_settings(
+    prisma_client: PrismaClient,
+    param_name: str,
+    sensitive_fields: Tuple[str, ...],
+    new_master_key: str,
+) -> None:
+    """Re-encrypt the sensitive fields of a JSON ``LiteLLM_Config`` row (e.g.
+    ``vantage_settings`` / ``cloudzero_settings``) under the new master key.
+    """
+    record = await ConfigRepository(prisma_client).table.find_unique(where={"param_name": param_name})
+    if record is None or record.param_value is None:
+        return
+    settings = record.param_value
+    if isinstance(settings, str):
+        settings = json.loads(settings)
+    if not isinstance(settings, dict):
+        return
+    new_settings = {
+        **settings,
+        **{
+            field_name: _reencrypt_secret_field(field_name, settings.get(field_name), new_master_key)
+            for field_name in sensitive_fields
+            if field_name in settings
+        },
+    }
+    if new_settings == settings:
+        return
+    await ConfigRepository(prisma_client).table.update(
+        where={"param_name": param_name},
+        data={"param_value": safe_dumps(new_settings)},
+    )
+
+
+async def _rotate_sso_config(prisma_client: PrismaClient, new_master_key: str) -> None:
+    """Re-encrypt every stored SSO field under the new master key. SSO settings
+    are all encrypted on save, so every string field is re-encrypted; non-string
+    fields (role/team mappings) pass through unchanged.
+    """
+    record = await SSOConfigRepository(prisma_client).table.find_unique(where={"id": "sso_config"})
+    if record is None or record.sso_settings is None:
+        return
+    settings = record.sso_settings
+    if isinstance(settings, str):
+        settings = json.loads(settings)
+    if not isinstance(settings, dict):
+        return
+    new_settings = {
+        field_name: _reencrypt_secret_field(field_name, value, new_master_key) for field_name, value in settings.items()
+    }
+    if new_settings == settings:
+        return
+    await SSOConfigRepository(prisma_client).table.update(
+        where={"id": "sso_config"},
+        data={"sso_settings": safe_dumps(new_settings)},
+    )
+
+
 async def _rotate_master_key(
     prisma_client: PrismaClient,
     user_api_key_dict: UserAPIKeyAuth,
@@ -4256,6 +4371,30 @@ async def _rotate_master_key(
                 continue
         verbose_proxy_logger.debug(f"Successfully re-encrypted {len(credentials)} credentials with new master key")
 
+    best_effort_steps: Tuple[Tuple[str, Callable[[], Any]], ...] = (
+        (
+            "verification_token callback_vars",
+            lambda: _rotate_callback_vars(prisma_client, "verification_token", new_master_key),
+        ),
+        ("team callback_vars", lambda: _rotate_callback_vars(prisma_client, "team", new_master_key)),
+        (
+            "vantage_settings",
+            lambda: _rotate_config_settings(
+                prisma_client, "vantage_settings", ("api_key", "integration_token"), new_master_key
+            ),
+        ),
+        (
+            "cloudzero_settings",
+            lambda: _rotate_config_settings(prisma_client, "cloudzero_settings", ("api_key",), new_master_key),
+        ),
+        ("sso_config", lambda: _rotate_sso_config(prisma_client, new_master_key)),
+    )
+    for label, step in best_effort_steps:
+        try:
+            await step()
+        except Exception as e:  # noqa: BLE001  # best-effort tail: one table's failure must not abort an in-progress rotation
+            verbose_proxy_logger.warning("Failed to rotate %s during master key rotation: %s", label, str(e))
+
 
 def _require_proxy_admin(user_api_key_dict: UserAPIKeyAuth) -> None:
     from litellm.proxy._types import CommonProxyErrors

tests/test_litellm/proxy/common_utils/test_callback_utils.py

@@ -226,6 +226,32 @@ def test_encrypt_callback_vars_round_trip(monkeypatch):
     )
 
 
+def test_encrypt_callback_vars_new_encryption_key(monkeypatch):
+    """Passing new_encryption_key encrypts under that key (not the current salt
+    key), so the values only decrypt once the salt/master key is swapped to it.
+    This is what master-key rotation relies on.
+    """
+    monkeypatch.setenv("LITELLM_SALT_KEY", "old-salt-32-bytes-aaaaaaaaaaaaaa")
+    new_key = "new-salt-32-bytes-bbbbbbbbbbbbbb"
+    encrypted = encrypt_callback_vars(_sample_metadata(), new_encryption_key=new_key)
+
+    enc_secret = encrypted["logging"][0]["callback_vars"]["langfuse_secret_key"]
+    assert enc_secret.startswith("litellm_enc::")
+
+    # Still under the OLD salt key: decrypt fails, value not recovered.
+    under_old = decrypt_callback_vars(encrypted)
+    assert (
+        under_old["logging"][0]["callback_vars"]["langfuse_secret_key"] != "sk-lf-secret"
+    )
+
+    # Swap the salt key to the new key: now it decrypts cleanly.
+    monkeypatch.setenv("LITELLM_SALT_KEY", new_key)
+    under_new = decrypt_callback_vars(encrypted)
+    assert (
+        under_new["logging"][0]["callback_vars"]["langfuse_secret_key"] == "sk-lf-secret"
+    )
+
+
 def test_encrypt_callback_vars_is_idempotent(monkeypatch):
     _set_salt_key(monkeypatch)
     once = encrypt_callback_vars(_sample_metadata())

tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py

@@ -14480,3 +14480,241 @@ async def test_regenerate_key_non_admin_permissions_rejected_before_enterprise_g
     assert int(exc.value.code) == 403
     assert "permissions" in str(exc.value.message)
     assert "Enterprise" not in str(exc.value.message)
+
+
+class _FakeTable:
+    """Minimal async table stub recording the last update payload."""
+
+    def __init__(self, rows):
+        self._rows = rows
+        self.updates = []
+
+    async def find_many(self):
+        return self._rows
+
+    async def find_unique(self, where):
+        return self._rows[0] if self._rows else None
+
+    async def update(self, where, data):
+        self.updates.append({"where": where, "data": data})
+
+
+def _fake_prisma_with_table(db_attr, rows):
+    from unittest.mock import MagicMock
+
+    table = _FakeTable(rows)
+    prisma_client = MagicMock()
+    prisma_client.db = MagicMock()
+    setattr(prisma_client.db, db_attr, table)
+    return prisma_client, table
+
+
+@pytest.mark.asyncio
+async def test_rotate_callback_vars_reencrypts_under_new_master_key(monkeypatch):
+    """callback_vars on verification-token metadata (both logging[*] and
+    callback_settings shapes) must decrypt under the OLD master key and be
+    re-encrypted under the NEW one, so they stay readable after rotation and
+    stop producing the recurring 'Error decrypting value' log.
+    """
+    import copy
+    from types import SimpleNamespace
+
+    from litellm.proxy.common_utils.callback_utils import (
+        decrypt_callback_vars,
+        encrypt_callback_vars,
+    )
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _rotate_callback_vars,
+    )
+
+    old_key = "sk-old-master-key"
+    new_key = "sk-new-master-key"
+    monkeypatch.delenv("LITELLM_SALT_KEY", raising=False)
+
+    plaintext_metadata = {
+        "logging": [
+            {
+                "callback_name": "langfuse",
+                "callback_vars": {
+                    "LANGFUSE_SECRET_KEY": "sk-langfuse-secret",
+                    "LANGFUSE_HOST": "https://cloud.langfuse.com",
+                },
+            }
+        ],
+        "callback_settings": {
+            "callback_vars": {"LANGSMITH_API_KEY": "ls-super-secret"}
+        },
+    }
+
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", old_key)
+    encrypted_under_old = encrypt_callback_vars(copy.deepcopy(plaintext_metadata))
+    # sanity: the secret was actually encrypted (prefix present, value changed)
+    old_secret = encrypted_under_old["logging"][0]["callback_vars"]["LANGFUSE_SECRET_KEY"]
+    assert old_secret.startswith("litellm_enc::")
+    assert old_secret != plaintext_metadata["logging"][0]["callback_vars"]["LANGFUSE_SECRET_KEY"]
+
+    row = SimpleNamespace(token="sk-hash-123", metadata=copy.deepcopy(encrypted_under_old))
+    prisma_client, table = _fake_prisma_with_table("litellm_verificationtoken", [row])
+
+    await _rotate_callback_vars(
+        prisma_client=prisma_client,
+        table_name="verification_token",
+        new_master_key=new_key,
+    )
+
+    assert len(table.updates) == 1
+    assert table.updates[0]["where"] == {"token": "sk-hash-123"}
+    rotated_metadata = json.loads(table.updates[0]["data"]["metadata"])
+
+    # Decrypting with the NEW master key recovers the original plaintext.
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", new_key)
+    decrypted_new = decrypt_callback_vars(rotated_metadata)
+    assert (
+        decrypted_new["logging"][0]["callback_vars"]["LANGFUSE_SECRET_KEY"]
+        == "sk-langfuse-secret"
+    )
+    assert (
+        decrypted_new["callback_settings"]["callback_vars"]["LANGSMITH_API_KEY"]
+        == "ls-super-secret"
+    )
+    # Non-sensitive var is untouched (never encrypted, never lost).
+    assert (
+        decrypted_new["logging"][0]["callback_vars"]["LANGFUSE_HOST"]
+        == "https://cloud.langfuse.com"
+    )
+
+    # Decrypting with the OLD master key must now FAIL (value not recovered),
+    # proving the ciphertext was genuinely re-keyed and not left under old key.
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", old_key)
+    decrypted_old = decrypt_callback_vars(rotated_metadata)
+    assert (
+        decrypted_old["logging"][0]["callback_vars"]["LANGFUSE_SECRET_KEY"]
+        != "sk-langfuse-secret"
+    )
+
+
+@pytest.mark.asyncio
+async def test_rotate_callback_vars_skips_rows_without_callback_metadata(monkeypatch):
+    from types import SimpleNamespace
+
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _rotate_callback_vars,
+    )
+
+    monkeypatch.delenv("LITELLM_SALT_KEY", raising=False)
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", "sk-old")
+
+    rows = [
+        SimpleNamespace(team_id="t1", metadata={"foo": "bar"}),
+        SimpleNamespace(team_id="t2", metadata=None),
+    ]
+    prisma_client, table = _fake_prisma_with_table("litellm_teamtable", rows)
+
+    await _rotate_callback_vars(
+        prisma_client=prisma_client,
+        table_name="team",
+        new_master_key="sk-new",
+    )
+
+    assert table.updates == []
+
+
+@pytest.mark.asyncio
+async def test_rotate_config_settings_reencrypts_sensitive_fields(monkeypatch):
+    """vantage / cloudzero config secrets must be re-encrypted under the new key
+    while non-secret fields (base_url) are left untouched.
+    """
+    import json as _json
+    from types import SimpleNamespace
+
+    from litellm.proxy.common_utils.encrypt_decrypt_utils import (
+        decrypt_value_helper,
+        encrypt_value_helper,
+    )
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _rotate_config_settings,
+    )
+
+    old_key = "sk-old-master-key"
+    new_key = "sk-new-master-key"
+    monkeypatch.delenv("LITELLM_SALT_KEY", raising=False)
+
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", old_key)
+    settings = {
+        "api_key": encrypt_value_helper("vantage-api-key"),
+        "integration_token": encrypt_value_helper("vantage-int-token"),
+        "base_url": "https://api.vantage.sh",
+    }
+    row = SimpleNamespace(param_name="vantage_settings", param_value=_json.dumps(settings))
+    prisma_client, table = _fake_prisma_with_table("litellm_config", [row])
+
+    await _rotate_config_settings(
+        prisma_client=prisma_client,
+        param_name="vantage_settings",
+        sensitive_fields=("api_key", "integration_token"),
+        new_master_key=new_key,
+    )
+
+    assert len(table.updates) == 1
+    rotated = _json.loads(table.updates[0]["data"]["param_value"])
+    assert rotated["base_url"] == "https://api.vantage.sh"
+
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", new_key)
+    assert (
+        decrypt_value_helper(rotated["api_key"], key="api_key", exception_type="debug")
+        == "vantage-api-key"
+    )
+    assert (
+        decrypt_value_helper(
+            rotated["integration_token"], key="integration_token", exception_type="debug"
+        )
+        == "vantage-int-token"
+    )
+
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", old_key)
+    assert (
+        decrypt_value_helper(rotated["api_key"], key="api_key", exception_type="debug")
+        != "vantage-api-key"
+    )
+
+
+@pytest.mark.asyncio
+async def test_rotate_sso_config_reencrypts_all_string_fields(monkeypatch):
+    from types import SimpleNamespace
+
+    from litellm.proxy.common_utils.encrypt_decrypt_utils import (
+        decrypt_value_helper,
+        encrypt_value_helper,
+    )
+    from litellm.proxy.management_endpoints.key_management_endpoints import (
+        _rotate_sso_config,
+    )
+
+    old_key = "sk-old-master-key"
+    new_key = "sk-new-master-key"
+    monkeypatch.delenv("LITELLM_SALT_KEY", raising=False)
+
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", old_key)
+    sso_settings = {
+        "google_client_secret": encrypt_value_helper("google-secret"),
+        "role_mappings": {"admin": ["proxy_admin"]},
+    }
+    row = SimpleNamespace(id="sso_config", sso_settings=sso_settings)
+    prisma_client, table = _fake_prisma_with_table("litellm_ssoconfig", [row])
+
+    await _rotate_sso_config(prisma_client=prisma_client, new_master_key=new_key)
+
+    assert len(table.updates) == 1
+    rotated = json.loads(table.updates[0]["data"]["sso_settings"])
+    # non-string field passes through unchanged
+    assert rotated["role_mappings"] == {"admin": ["proxy_admin"]}
+
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", new_key)
+    assert (
+        decrypt_value_helper(
+            rotated["google_client_secret"],
+            key="google_client_secret",
+            exception_type="debug",
+        )
+        == "google-secret"
+    )