scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 207

fix(router): refuse a tagged auto-router entry to callers without its access group

Browse source
This commit is contained in:
ryan-crabbe-berri 2026-10-20 12:15:12 -07:00
parent a7f1d42459
commit acf069e93f
2 changed files with 32 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/router.py
View file

@ -15551,6 +15551,9 @@ class Router:
model=registered_model_name, strategy_tags=selected_strategy.tags, request_kwargs=request_kwargs
)
definition: Final = selected_strategy.deployment
strategy_source: Final = marker if marker is not None else definition.model_dump() if definition else None
if strategy_source is not None:
_ = require_granted_deployment(strategy_source, model, request_kwargs)
config_id: Final = definition.model_info.id if definition else None
updated_at: Final = definition.model_info.updated_at if definition else None
router_kind: Final = classify_strategy_router_model(definition.litellm_params.model) if definition else None

29
tests/unit/router_utils/test_model_access_group_grants.py
View file

@ -152,3 +152,32 @@ def test_resolution_finds_a_tagged_deployment_by_id_for_any_caller() -> None:
)
assert isinstance(deployment, dict)
assert (name, deployment["model_info"]["id"]) == ("openai/private-id-upstream", "private-id")
def _router_with_tagged_complexity_router() -> litellm.Router:
return litellm.Router(
model_list=[
_deployment("public-model", "public", "public-id"),
{
"model_name": "private-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "public-model", "MEDIUM": "public-model", "COMPLEX": "public-model"}
},
"complexity_router_default_model": "public-model",
},
"model_info": {"id": "private-router-id", "access_groups": ["group-x"]},
},
],
)
async def test_a_tagged_auto_router_refuses_ungranted_callers_even_when_it_picks_a_public_target() -> None:
with pytest.raises(litellm.PermissionDeniedError) as denied:
await _reply(_router_with_tagged_complexity_router(), "private-router", frozenset({"public-model"}))
assert denied.value.status_code == 403
async def test_a_tagged_auto_router_serves_callers_granted_its_group() -> None:
assert await _reply(_router_with_tagged_complexity_router(), "private-router", frozenset({"group-x"})) == "public"