diff --git a/litellm/proxy/plugin_routes.py b/litellm/proxy/plugin_routes.py index cfbcfefe9e8..966136f6fa7 100644 --- a/litellm/proxy/plugin_routes.py +++ b/litellm/proxy/plugin_routes.py @@ -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() diff --git a/tests/unit/proxy/test_plugin_routes.py b/tests/unit/proxy/test_plugin_routes.py index 7d3fbda14fd..23145f13156 100644 --- a/tests/unit/proxy/test_plugin_routes.py +++ b/tests/unit/proxy/test_plugin_routes.py @@ -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")