scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

fix(proxy): authorize the public name of a team-scoped model on batch retrieve/cancel (#47067)

Browse source 
* fix(proxy): authorize the public name of a team-scoped model on batch retrieve/cancel

A unified batch id resolves to the deployment's internal model_name
(model_name_{team_id}_{uuid}), which is in no key or team allowlist, so
authorize_model_for_key always denied and team keys got 403 retrieving or
cancelling their own batches. Authorize against model_info
team_public_model_name when the resolved deployment is team-scoped, falling
back to the resolved name otherwise.

Fixes #46958

* fix(proxy): authorize team-scoped batch models by their public name on both auth paths

* fix(proxy): accept the owning team's public name for its own team deployment on every auth path

Managed batch, file, fine-tuning, vector store, character and video ids resolve to the
team deployment's internal name again, so teams that granted the internal name keep access.
The model access check now also accepts the owning team's public name and its legacy team
aliases for that internal name, and only when the caller's team owns the deployment.

* fix(proxy): match a team deployment's public and alias names against the caller's own grants

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
(cherry picked from commit e979bd0ecf)
This commit is contained in:
Mrunmayee Daware 2026-10-18 12:48:11 +05:30 • committed by Yuneng Jiang
parent 05de300ed6
commit 3ed027b5a6
No known key found for this signature in database
3 changed files with 273 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
litellm/proxy/auth/auth_checks.py
View file

@ -4420,6 +4420,24 @@ def _check_model_access_helper(
return True
def _owning_team_names_for_team_deployment(
model: str,
llm_router: Router | None,
team_id: str | None,
team_model_aliases: Mapping[str, str] | None,
) -> tuple[str, ...]:
if llm_router is None or team_id is None or not model.startswith(f"model_name_{team_id}_"):
return ()
deployment: Final = llm_router.get_deployment_by_model_group_name(model_group_name=model)
public_name: Final = (
deployment.model_info.team_public_model_name
if deployment is not None and deployment.model_info.team_id == team_id
else None
)
legacy_aliases: Final = tuple(alias for alias, target in (team_model_aliases or {}).items() if target == model)
return (*((public_name,) if public_name else ()), *legacy_aliases)
def _can_object_call_model(
model: str | list[str],
llm_router: Router | None,
@ -4491,6 +4509,13 @@ def _can_object_call_model(
*((global_or_router_alias_target,) if global_or_router_alias_target else ()),
)
)
owning_team_names: Final = (
()
if key_alias_applied
else _owning_team_names_for_team_deployment(
model=model, llm_router=llm_router, team_id=team_id, team_model_aliases=team_model_aliases
)
)
## check model access for alias + underlying model - allow if either is in allowed models
for m in potential_models:
@ -4502,6 +4527,11 @@ def _can_object_call_model(
team_id=team_id,
):
return True
if any(
_check_model_access_helper(model=name, llm_router=llm_router, models=models, team_id=team_id)
for name in owning_team_names
):
return True
internal_message: Final = (
f"{object_type} not allowed to access model. This {object_type} can only access models={models}. "

119
tests/unit/proxy/auth/test_auth_utils.py
View file

@ -30,6 +30,7 @@ from litellm.proxy.auth.auth_utils import (
log_once_if_budget_reservation_disabled,
warn_once_if_custom_auth_skips_common_checks,
)
from litellm.router import Router
class TestCustomAuthCommonChecksWarning:
@ -1167,6 +1168,124 @@ async def test_managed_batch_routes_pass_team_model_access_check(route, request_
)
_TEAM_DEPLOYMENT_ID: Final = "team-a-deployment-id"
_TEAM_INTERNAL_MODEL_NAME: Final = "model_name_team-a_6d1f3c2e-0000-4000-8000-000000000000"
_TEAM_PUBLIC_MODEL_NAME: Final = "team-batch-model"
_TEAM_MANAGED_BATCH_ID: Final = _encode_managed_id(
f"litellm_proxy;model_id:{_TEAM_DEPLOYMENT_ID};llm_batch_id:provider-batch-789"
)
def _team_scoped_router(public_model_name: str | None = _TEAM_PUBLIC_MODEL_NAME) -> Router:
return Router(
model_list=[
{
"model_name": _TEAM_INTERNAL_MODEL_NAME,
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "test-key"},
"model_info": {
"id": _TEAM_DEPLOYMENT_ID,
"team_id": "team-a",
**({"team_public_model_name": public_model_name} if public_model_name else {}),
},
},
]
)
def _team_scoped_character_id() -> str:
from litellm.types.videos.utils import encode_character_id_with_provider
return encode_character_id_with_provider(character_id="char_team", provider="openai", model_id=_TEAM_DEPLOYMENT_ID)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"route, request_data",
[
pytest.param("/v1/batches/{batch_id}", {"batch_id": _TEAM_MANAGED_BATCH_ID}, id="batch-retrieve"),
pytest.param("/v1/batches/{batch_id}/cancel", {"batch_id": _TEAM_MANAGED_BATCH_ID}, id="batch-cancel"),
pytest.param(
"/v1/fine_tuning/jobs/{fine_tuning_job_id}",
{
"fine_tuning_job_id": _encode_managed_id(
f"litellm_proxy;model_id:{_TEAM_DEPLOYMENT_ID};generic_response_id:ftjob-1"
)
},
id="fine-tuning-job",
),
pytest.param(
"/v1/vector_stores/{vector_store_id}",
{
"vector_store_id": _encode_managed_id(
"litellm_proxy:vector_store;unified_id,vs-unified;"
f"target_model_names,{_TEAM_PUBLIC_MODEL_NAME};resource_id,vs_1;model_id,{_TEAM_DEPLOYMENT_ID}"
)
},
id="vector-store",
),
pytest.param(
"/v1/videos/characters/{character_id}",
{"character_id": _team_scoped_character_id()},
id="character",
),
],
)
async def test_team_scoped_managed_ids_authorize_only_the_owning_team_by_public_name(
route: str, request_data: dict[str, str]
) -> None:
from litellm.proxy._types import ProxyException
from litellm.proxy.auth.auth_checks import can_key_call_resolved_model
llm_router: Final = _team_scoped_router()
async def check(team_id: str) -> None:
await can_key_call_resolved_model(
model=get_model_from_request(
request_data=request_data, route=route, llm_router=llm_router, team_id=team_id
),
llm_model_list=None,
valid_token=UserAPIKeyAuth(api_key="sk-team", team_id=team_id, team_models=[_TEAM_PUBLIC_MODEL_NAME]),
llm_router=llm_router,
)
await check("team-a")
with pytest.raises(ProxyException) as exc_info:
await check("team-b")
assert exc_info.value.code == "403"
@pytest.mark.asyncio
async def test_legacy_team_alias_authorizes_a_team_batch_only_for_keys_granted_the_alias() -> None:
from litellm.proxy._types import ProxyException
from litellm.proxy.auth.auth_checks import can_key_call_resolved_model
llm_router: Final = _team_scoped_router(public_model_name=None)
async def check(key_models: list[str]) -> None:
await can_key_call_resolved_model(
model=get_model_from_request(
request_data={"batch_id": _TEAM_MANAGED_BATCH_ID},
route="/v1/batches/{batch_id}",
llm_router=llm_router,
team_id="team-a",
),
llm_model_list=None,
valid_token=UserAPIKeyAuth(
api_key="sk-team",
team_id="team-a",
models=key_models,
team_models=["legacy-team-model"],
team_model_aliases={"legacy-team-model": _TEAM_INTERNAL_MODEL_NAME},
),
llm_router=llm_router,
)
await check(["legacy-team-model"])
with pytest.raises(ProxyException) as exc_info:
await check(["some-other-model"])
assert exc_info.value.code == "403"
def test_get_model_from_request_resolves_character_id_model_with_router():
from litellm.types.videos.utils import encode_character_id_with_provider

125
tests/unit/proxy/batches_endpoints/test_endpoints.py
View file

@ -35,7 +35,7 @@ import logging
from contextlib import ExitStack
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
from typing import Any, Dict, Final, Optional
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
@ -3259,6 +3259,129 @@ async def test_cancel__unified_batch_id_rejects_key_without_model_grant(cancel_h
cancel_harness.router_acancel.assert_not_called()
TEAM_INTERNAL_MODEL_NAME: Final = "model_name_team-a_00000000-0000-0000-0000-000000000000"
TEAM_PUBLIC_MODEL_NAME: Final = "team-gpt4o-mini"
def _team_scoped_deployment() -> Deployment:
return Deployment(
model_name=TEAM_INTERNAL_MODEL_NAME,
litellm_params=LiteLLM_Params(model="gpt-4o-mini"),
model_info=ModelInfo(id="dep-team-a-1", team_id="team-a", team_public_model_name=TEAM_PUBLIC_MODEL_NAME),
)
def _route_unified_batch_to_team_deployment(
harness: RetrieveHarness | CancelHarness, deployment: Deployment | None = None
) -> None:
target: Final = deployment or _team_scoped_deployment()
harness.router.resolve_model_name_from_model_id = MagicMock(return_value=target.model_name)
harness.router.get_deployment_by_model_group_name = MagicMock(return_value=target)
@pytest.mark.asyncio
async def test_retrieve__team_scoped_deployment_authorizes_owning_team_by_public_name(
retrieve_harness: RetrieveHarness,
) -> None:
_route_unified_batch_to_team_deployment(retrieve_harness)
resp: Final = await call_retrieve(
retrieve_harness, UNIFIED_BATCH_ID_FOR_GPT4O_MINI, user=_key_restricted_to(TEAM_PUBLIC_MODEL_NAME)
)
assert retrieve_harness.router_aretrieve.call_count == 1
assert resp._hidden_params["model_id"] == "gpt-4o-mini"
@pytest.mark.asyncio
async def test_cancel__team_scoped_deployment_authorizes_owning_team_by_public_name(
cancel_harness: CancelHarness,
) -> None:
_route_unified_batch_to_team_deployment(cancel_harness)
await call_cancel(cancel_harness, UNIFIED_BATCH_ID_FOR_GPT4O_MINI, user=_key_restricted_to(TEAM_PUBLIC_MODEL_NAME))
assert cancel_harness.router_acancel.call_count == 1
@pytest.mark.asyncio
@pytest.mark.parametrize(
"deployment, user",
[
pytest.param(
_team_scoped_deployment(), _key_restricted_to("some-other-model"), id="owning-team-key-without-the-model"
),
pytest.param(
_team_scoped_deployment(),
UserAPIKeyAuth(
api_key="sk-team-b",
team_id="team-b",
team_models=[TEAM_PUBLIC_MODEL_NAME],
models=[TEAM_PUBLIC_MODEL_NAME],
),
id="other-team-key-granted-the-same-public-name",
),
pytest.param(
Deployment(
model_name="model_name_team-a_ml_00000000-0000-0000-0000-000000000000",
litellm_params=LiteLLM_Params(model="gpt-4o-mini"),
model_info=ModelInfo(
id="dep-team-a-ml-1", team_id="team-a_ml", team_public_model_name=TEAM_PUBLIC_MODEL_NAME
),
),
_key_restricted_to(TEAM_PUBLIC_MODEL_NAME),
id="team-whose-id-prefixes-the-owning-team-id",
),
],
)
async def test_retrieve__team_scoped_deployment_rejects_keys_outside_its_grant(
retrieve_harness: RetrieveHarness, deployment: Deployment, user: UserAPIKeyAuth
) -> None:
_route_unified_batch_to_team_deployment(retrieve_harness, deployment)
with pytest.raises(ProxyException) as exc_info:
await call_retrieve(retrieve_harness, UNIFIED_BATCH_ID_FOR_GPT4O_MINI, user=user)
assert exc_info.value.code == "403"
retrieve_harness.router_aretrieve.assert_not_called()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"deployment, user",
[
pytest.param(
_team_scoped_deployment(),
_key_restricted_to(TEAM_INTERNAL_MODEL_NAME),
id="owning-team-key-granted-the-internal-name",
),
pytest.param(
Deployment(
model_name=TEAM_INTERNAL_MODEL_NAME,
litellm_params=LiteLLM_Params(model="gpt-4o-mini"),
model_info=ModelInfo(id="dep-team-a-1", team_id="team-a"),
),
UserAPIKeyAuth(
api_key="sk-legacy",
team_id="team-a",
team_models=["legacy-team-model"],
models=["legacy-team-model"],
team_model_aliases={"legacy-team-model": TEAM_INTERNAL_MODEL_NAME},
),
id="legacy-team-model-reached-through-a-team-alias",
),
],
)
async def test_retrieve__team_scoped_deployment_authorizes_internal_name_and_legacy_alias_grants(
retrieve_harness: RetrieveHarness, deployment: Deployment, user: UserAPIKeyAuth
) -> None:
_route_unified_batch_to_team_deployment(retrieve_harness, deployment)
await call_retrieve(retrieve_harness, UNIFIED_BATCH_ID_FOR_GPT4O_MINI, user=user)
assert retrieve_harness.router_aretrieve.call_count == 1
@pytest.mark.asyncio
async def test_cancel__executed_batch_rejects_key_without_model_grant(cancel_harness, executed_runner):
runner, factory = executed_runner