From dfc78bac787d33a80380d1e70d0cc5ea6be82650 Mon Sep 17 00:00:00 2001 From: JackDanger Date: Sat, 17 Oct 2026 23:13:59 +0000 Subject: [PATCH] fix(proxy): team admins cannot repoint deployment api_base while the server-owned credential stays Generated-by: termag --- .../model_management_endpoints.py | 28 ++++- .../test_r2_9_api_base_team_gate.py | 103 ++++++++++++++++++ 2 files changed, 130 insertions(+), 1 deletion(-) create mode 100644 tests/unit/proxy/management_endpoints/test_r2_9_api_base_team_gate.py diff --git a/litellm/proxy/management_endpoints/model_management_endpoints.py b/litellm/proxy/management_endpoints/model_management_endpoints.py index 7460522f53e..9422ea177a7 100644 --- a/litellm/proxy/management_endpoints/model_management_endpoints.py +++ b/litellm/proxy/management_endpoints/model_management_endpoints.py @@ -2204,9 +2204,35 @@ class ModelManagementAuthChecks: ) -> None: if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN: return + stored: Final = _decrypted_litellm_params(model_params.litellm_params) + + # Static-key endpoint harvest (LD fork R2-9): a non-admin write that + # moves api_base/base_url while the stored server-owned api_key stays + # in place sends that key to the endpoint the caller names. The team + # must bring its own key (set api_key in the same write) or route + # through an admin-attached credential. + repoints_endpoint: Final = incoming_params is not None and not incoming_params.keys().isdisjoint( + ("api_base", "base_url") + ) + brings_own_key: Final = ( + incoming_params is not None and incoming_params.get("api_key") is not None + ) + server_key_stays: Final = ( + stored.get("api_key") not in (None, "") and not brings_own_key + ) + if repoints_endpoint and server_key_stays: + raise ProxyException( + message=( + "Only proxy admins can repoint a deployment's endpoint (api_base/base_url) " + "while its api_key stays server-owned; supply your own api_key with the " + "new endpoint." + ), + type=ProxyErrorTypes.auth_error.value, + code=status.HTTP_403_FORBIDDEN, + param="api_base", + ) if not write_touches_federation_surface(incoming_params): return - stored: Final = _decrypted_litellm_params(model_params.litellm_params) wif_fields: Final = await effective_server_owned_wif_fields(stored, incoming_params, prisma_client) if wif_fields: # ProxyException rather than HTTPException so the offending field stays a structured diff --git a/tests/unit/proxy/management_endpoints/test_r2_9_api_base_team_gate.py b/tests/unit/proxy/management_endpoints/test_r2_9_api_base_team_gate.py new file mode 100644 index 00000000000..8e114b75091 --- /dev/null +++ b/tests/unit/proxy/management_endpoints/test_r2_9_api_base_team_gate.py @@ -0,0 +1,103 @@ +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from litellm.types.router import GenericLiteLLMParams + + +@pytest.mark.asyncio +async def test_team_admin_cannot_move_api_base_keeping_server_key(): + from fastapi import HTTPException + from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth + from litellm.types.router import GenericLiteLLMParams + + from litellm.proxy.management_endpoints.model_management_endpoints import ( + ModelManagementAuthChecks, + ) + + stored = GenericLiteLLMParams(api_key="server-owned-key") + incoming = GenericLiteLLMParams(api_base="https://attacker.example") + model_params = MagicMock() + model_params.litellm_params = stored + + with pytest.raises(Exception) as exc: + await ModelManagementAuthChecks._reject_non_admin_wif_write( + model_params=model_params, + incoming_params=incoming.model_dump(exclude_none=True), + user_api_key_dict=UserAPIKeyAuth( + api_key="sk-team-admin", + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="team-admin-1", + ), + prisma_client=AsyncMock(), + ) + assert int(getattr(exc.value, "code", 0)) == 403 + assert "api_key stays server-owned" in str(getattr(exc.value, "message", getattr(exc.value, "detail", ""))) + + +@pytest.mark.asyncio +async def test_team_admin_bring_your_own_key_move_allowed(): + import litellm.proxy.management_endpoints.model_management_endpoints as mme + from litellm.proxy.management_endpoints.model_management_endpoints import ( + ModelManagementAuthChecks, + ) + from litellm.types.router import GenericLiteLLMParams + from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth + + stored = GenericLiteLLMParams(api_key="server-owned-key") + incoming = GenericLiteLLMParams( + api_base="https://my-team-proxy.example", api_key="my-team-key" + ) + model_params = MagicMock() + model_params.litellm_params = stored + + with ( + patch.object(mme, "effective_server_owned_wif_fields", AsyncMock(return_value=())), + ): + await ModelManagementAuthChecks._reject_non_admin_wif_write( + model_params=model_params, + incoming_params=incoming.model_dump(exclude_none=True), + user_api_key_dict=UserAPIKeyAuth( + api_key="sk-team", + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="team-admin-1", + ), + prisma_client=AsyncMock(), + ) + + +@pytest.mark.asyncio +async def test_team_admin_wif_association_still_guarded(): + """Under WIF the original server-owned-token rule keeps firing.""" + import litellm.proxy.management_endpoints.model_management_endpoints as mme + from litellm.proxy.management_endpoints.model_management_endpoints import ( + ModelManagementAuthChecks, + ) + from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth + + stored = GenericLiteLLMParams(api_key="server-key") + incoming = GenericLiteLLMParams(api_base="https://moved.example") + model_params = MagicMock() + model_params.litellm_params = stored + + with ( + patch.object( + mme, "effective_server_owned_wif_fields", AsyncMock(return_value=("oidc_token_url",)) + ), + ): + try: + await ModelManagementAuthChecks._reject_non_admin_wif_write( + model_params=model_params, + incoming_params=incoming.model_dump(exclude_none=True), + user_api_key_dict=UserAPIKeyAuth( + api_key="sk-team", user_role=LitellmUserRoles.INTERNAL_USER, user_id="t" + ), + prisma_client=AsyncMock(), + ) + except Exception as guard_error: + # On this stack both the static-key harvest gate and the WIF + # rule can fire for the same shape; accept either interdiction. + assert ("api_key stays server-owned" in str(guard_error)) or ( + "workload identity" in str(guard_error) + ) + return + raise AssertionError("expected a server-owned-key interdiction to fire")