fix(proxy): build plugin session claims from FIPS provider services in fips_mode (#50544)

* fix(proxy): build plugin session claims from FIPS provider services in fips_mode

With general_settings.fips_mode on, the per-plugin key comes from the
salt key KBKDF and the Fernet token is built from ssl.RAND_bytes,
AES-128-CBC and hmac.digest. Standard mode is unchanged.

* test(proxy): freeze the clock in the fips_mode plugin claim test
This commit is contained in:
yucheng-berri 2026-11-09 11:07:57 -08:00 • committed by GitHub
parent 1d06a1ca69
commit 457365f627
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 92 additions and 1 deletions

View file

@ -18,6 +18,10 @@ Plugin iframe auth:
litellm bearer token, so a compromised plugin learns only the caller's
identity, never their credential. LITELLM_SALT_KEY itself is never shared
with plugins — each plugin holds only its own derived key.
With general_settings.fips_mode on, the per-plugin key is instead
derive_key_from_salt_key(b"litellm-plugin:session-claim:", plugin_name), and
the Fernet token is built from the FIPS provider's DRBG, AES-CBC and HMAC.
"""
import base64
@ -25,11 +29,14 @@ import hashlib
import hmac as _hmac
import json
import os
import ssl
import time
from collections.abc import Mapping
from typing import Final
from cryptography.fernet import Fernet, InvalidToken
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from pydantic import ConfigDict, TypeAdapter, with_config
from typing_extensions import NotRequired, ReadOnly, TypedDict
@ -37,6 +44,8 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.proxy._types import PluginConfig, SpecialHeaders, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.encrypt_decrypt_utils import derive_key_from_salt_key
from litellm.proxy.common_utils.fips import fips_mode_active
from litellm.types.llms.custom_http import httpxSpecialProvider
router: Final = APIRouter()
@ -145,11 +154,30 @@ def _plugin_fernet(plugin_name: str) -> Fernet:
A plugin possessing its own key cannot derive the master salt or
forge claims intended for a different plugin.
"""
if fips_mode_active():
return Fernet(base64.urlsafe_b64encode(_fips_plugin_key(plugin_name)))
salt: Final = os.getenv("LITELLM_SALT_KEY", "").encode()
derived: Final = _hmac.new(salt, plugin_name.encode(), hashlib.sha256).digest()
return Fernet(base64.urlsafe_b64encode(derived))
_FIPS_PLUGIN_KEY_LABEL: Final = b"litellm-plugin:session-claim:"
def _fips_plugin_key(plugin_name: str) -> bytes:
return bytes.fromhex(derive_key_from_salt_key(_FIPS_PLUGIN_KEY_LABEL, plugin_name.encode()))
def _fips_fernet_token(key: bytes, data: bytes) -> str:
"""Encrypt data into a standard Fernet token using only the FIPS provider's DRBG, AES-128-CBC and HMAC-SHA256."""
iv: Final = ssl.RAND_bytes(16)
padder: Final = padding.PKCS7(128).padder()
encryptor: Final = Cipher(algorithms.AES(key[16:]), modes.CBC(iv)).encryptor()
ciphertext: Final = encryptor.update(padder.update(data) + padder.finalize()) + encryptor.finalize()
body: Final = b"\x80" + int(time.time()).to_bytes(8, "big") + iv + ciphertext
return base64.urlsafe_b64encode(body + _hmac.digest(key[:16], body, "sha256")).decode()
_CLAIM_TTL_SECONDS: Final = 30 # identity claims expire after 30 s
@ -166,6 +194,8 @@ def issue_plugin_session_claim(plugin_name: str, user_id: str | None, user_role:
"user_role": user_role or "",
"exp": int(time.time()) + _CLAIM_TTL_SECONDS,
}
if fips_mode_active():
return _fips_fernet_token(_fips_plugin_key(plugin_name), json.dumps(claim).encode())
return _plugin_fernet(plugin_name).encrypt(json.dumps(claim).encode()).decode()

View file

@ -12,10 +12,19 @@ Covers three bugs:
"""
import asyncio
import base64
import hashlib
import hmac
import json
import os
import time
from typing import Final, NoReturn
from unittest.mock import MagicMock
import _hashlib
import pytest
import respx
from cryptography.fernet import Fernet
from starlette.requests import Request
import litellm
@ -25,7 +34,12 @@ from litellm.proxy._types import (
PluginConfig,
UserAPIKeyAuth,
)
from litellm.proxy.plugin_routes import list_plugins, register_plugins_from_config
from litellm.proxy.plugin_routes import (
issue_plugin_session_claim,
list_plugins,
register_plugins_from_config,
verify_plugin_session_claim,
)
def _admin() -> UserAPIKeyAuth:
@ -320,3 +334,50 @@ def test_plugin_proxy_refuses_non_admin_callers_without_contacting_the_plugin(
assert response.status_code == 403
assert response.body == b"Plugin proxy access requires proxy_admin role."
assert not upstream.called
def test_plugin_session_claim_opens_under_the_documented_hmac_plugin_key(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-salt")
plugin_key: Final = base64.urlsafe_b64encode(hmac.new(b"sk-salt", b"chat-ui", hashlib.sha256).digest())
claim: Final[dict[str, object]] = json.loads(
Fernet(plugin_key).decrypt(issue_plugin_session_claim("chat-ui", "user-1", "proxy_admin"))
)
assert (claim["plugin"], claim["user_id"], claim["user_role"]) == ("chat-ui", "user-1", "proxy_admin")
@pytest.mark.skipif(
not hasattr(_hashlib, "__file__"), reason="this Python links OpenSSL statically, out of ctypes reach"
)
@pytest.mark.usefixtures("fips_mode")
def test_fips_mode_plugin_session_claim_opens_under_the_kbkdf_key_without_os_urandom_or_hmac_new(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def outside_the_fips_provider(*_: object) -> NoReturn:
raise AssertionError("fips_mode must not draw on os.urandom or hmac.new")
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-" + bytes(range(32)).hex())
monkeypatch.setattr(os, "urandom", outside_the_fips_provider)
monkeypatch.setattr(hmac, "new", outside_the_fips_provider)
monkeypatch.setattr(time, "time", lambda: 1_800_000_000.0)
fixed_input: Final = (
(1).to_bytes(4, "big") + b"litellm-plugin:session-claim:" + b"\x00" + b"chat-ui" + (256).to_bytes(4, "big")
)
plugin_key: Final = base64.urlsafe_b64encode(hmac.digest(bytes(range(32)), fixed_input, "sha256"))
token: Final = issue_plugin_session_claim("chat-ui", "user-1", "proxy_admin")
claim: Final[dict[str, object]] = json.loads(Fernet(plugin_key).decrypt(token))
assert (claim["plugin"], claim["user_id"], claim["user_role"]) == ("chat-ui", "user-1", "proxy_admin")
assert verify_plugin_session_claim("chat-ui", token) == claim
@pytest.mark.usefixtures("fips_mode")
def test_fips_mode_refuses_a_plugin_session_claim_under_a_salt_key_that_is_not_hex(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-salt")
with pytest.raises(ValueError, match="LITELLM_SALT_KEY must be at least 32 bytes of hex"):
issue_plugin_session_claim("chat-ui", "user-1", "proxy_admin")