feat(proxy): opt-in model_access_group_deny_by_default makes access-group models private

With general_settings.model_access_group_deny_by_default on, a model tagged with
model_info.access_groups is only reachable when the caller's grant names the group
or the model. Teams, users and ownerless keys on blanket grants (empty list, *,
all-proxy-models, all-team-models, wildcards) no longer reach tagged models, and
deployment ids, provider model names, routing groups and aliases are checked too.
Default off, so existing behavior is unchanged.
This commit is contained in:
ryan-crabbe-berri 2026-10-19 18:00:33 -07:00
parent 06d56eb6d7
commit 15f7dba7ca
9 changed files with 716 additions and 8 deletions

View file

@ -2765,6 +2765,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
default=False,
description="When True, a search tool must be explicitly listed in object_permission.search_tools: a virtual key needs its own grant plus its team's, a keyless team member needs the team's, and a user with neither needs their own. A missing permission record, an empty list, or an unresolved team grants nothing, and the unregistered search fallback is denied. The master key and dashboard sessions are exempt",
)
model_access_group_deny_by_default: bool = Field(
default=False,
description="When True, a model tagged with model_info.access_groups is only reachable by a caller whose grant names one of its groups or the model itself. The grant that counts is the team's for a team key, the user's for a personal key, and the key's own for a key with neither. An empty models list, '*', all-proxy-models and wildcard patterns no longer reach these models. Untagged models are unaffected, and proxy admins' personal access is exempt",
)
missing_session_id: Literal["generate", "reject", "omit"] | None = Field(
None,
description="What to do with LLM API requests that carry no session id (x-litellm-session-id header, metadata.session_id, etc.). 'generate' stamps one id into litellm_session_id, litellm_trace_id and metadata.session_id so SpendLogs and logging callbacks agree; 'reject' returns 400; 'omit' leaves SpendLogs.session_id null, matching callbacks such as Langfuse that only record a client-established metadata.session_id. Unset keeps the legacy behavior where SpendLogs falls back to the trace id while callbacks get no session id.",

View file

@ -94,6 +94,12 @@ from litellm.proxy.auth.model_access_denied import (
customer_model_access_denied_client_message,
model_access_denied_client_message,
)
from litellm.proxy.auth.model_access_group_grants import (
access_group_model_denied,
model_access_group_deny_by_default,
requested_model_names,
ungranted_access_group_models,
)
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import (
broadcast_evictions,
@ -1287,7 +1293,12 @@ async def common_checks(
llm_router=llm_router,
user_object=user_object,
key_model_aliases=key_model_aliases_for_auth_check(valid_token),
teams_grant_access_group_models=personal_key_team_model_access_applies(valid_token),
)
if _model and team_object is None and user_object is None and valid_token is not None:
await _require_key_access_group_grant(
model=_model, valid_token=valid_token, llm_router=llm_router, prisma_client=prisma_client
)
if end_user_object is not None and end_user_object.models:
with tracer.trace("litellm.proxy.auth.common_checks.can_customer_call_model"):
@ -5487,6 +5498,10 @@ async def can_key_call_model(
Raises:
- Exception: If token not allowed to call model
"""
if valid_token.user_id is None:
await _require_key_access_group_grant(
model=model, valid_token=valid_token, llm_router=llm_router, prisma_client=prisma_client
)
key_models: Final = _resolve_key_models_for_auth_check(valid_token=valid_token)
try:
return can_object_call_model(
@ -5519,6 +5534,39 @@ async def can_key_call_model(
raise
async def _require_key_access_group_grant(
model: str | Sequence[str],
valid_token: UserAPIKeyAuth,
llm_router: litellm.Router | None,
prisma_client: DatabaseClient | None,
) -> None:
"""A key with no team and no loaded owner owns its own grant, so deny-by-default reads the key's models"""
if (
valid_token.team_id is not None
or valid_token.user_role == LitellmUserRoles.PROXY_ADMIN
or not _is_strict_grant_identity(valid_token)
or personal_key_team_model_access_applies(valid_token)
or not _model_access_group_deny_by_default_enabled()
):
return
ungranted: Final = await _ungranted_after_access_group_ids(
ungranted=ungranted_access_group_models(
names=requested_model_names(
model, valid_token.team_model_aliases, key_model_aliases_for_auth_check(valid_token)
),
granted_models=_resolve_key_models_for_auth_check(valid_token=valid_token),
llm_router=llm_router,
team_id=None,
),
access_group_ids=valid_token.access_group_ids,
llm_router=llm_router,
team_id=None,
prisma_client=prisma_client,
)
if ungranted:
raise access_group_model_denied(model=model, ungranted=ungranted, object_type="key")
async def _personal_key_owner(
valid_token: UserAPIKeyAuth,
prisma_client: PrismaClient | None,
@ -5547,6 +5595,40 @@ async def _personal_key_owner(
return None
async def _require_teamless_owner_access_group_grant(
model: str,
valid_token: UserAPIKeyAuth,
llm_router: litellm.Router | None,
prisma_client: PrismaClient | None,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> None:
if (
not _is_strict_grant_identity(valid_token)
or personal_key_team_model_access_applies(valid_token)
or not _model_access_group_deny_by_default_enabled()
):
return
owner: Final = await _personal_key_owner(
valid_token=valid_token,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if owner is None:
await _require_key_access_group_grant(
model=model, valid_token=valid_token, llm_router=llm_router, prisma_client=prisma_client
)
return
_require_user_access_group_grant(
model=model,
user_role=owner.user_role,
user_models=_user_models(owner),
llm_router=llm_router,
key_model_aliases=key_model_aliases_for_auth_check(valid_token),
)
async def can_key_call_resolved_model(
model: str,
llm_model_list: Sequence[object] | None,
@ -5626,6 +5708,16 @@ async def can_key_call_resolved_model(
key_model_aliases=key_model_aliases_for_auth_check(valid_token),
)
if team_object is None:
await _require_teamless_owner_access_group_grant(
model=model,
valid_token=valid_token,
llm_router=llm_router,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if personal_key_team_model_access_applies(valid_token) and valid_token.user_id is not None:
owner: Final = await _personal_key_owner(
valid_token=valid_token,
@ -5709,7 +5801,18 @@ async def can_team_access_model(
1. First checks native team-level model permissions (current implementation)
2. If not allowed natively, falls back to access_group_ids on the team
3. Under general_settings.model_access_group_deny_by_default, a model in a model access group also needs
the team to name that group or model explicitly
"""
if team_object is not None:
await _require_team_access_group_grant(
model=model,
team_object=team_object,
llm_router=llm_router,
team_model_aliases=team_model_aliases,
key_model_aliases=key_model_aliases,
prisma_client=prisma_client,
)
try:
return can_object_call_model(
model=model,
@ -5741,6 +5844,68 @@ async def can_team_access_model(
raise
def _proxy_general_settings() -> Mapping[str, object]:
from litellm.proxy.proxy_server import general_settings
return typed_general_settings(general_settings)
def _model_access_group_deny_by_default_enabled() -> bool:
return model_access_group_deny_by_default(_proxy_general_settings())
_MODEL_NAMES: Final = TypeAdapter(tuple[str, ...])
def _user_models(user_object: LiteLLM_UserTable) -> tuple[str, ...]:
return _MODEL_NAMES.validate_python(user_object.models)
async def _ungranted_after_access_group_ids(
ungranted: tuple[str, ...],
access_group_ids: Sequence[str] | None,
llm_router: Router | None,
team_id: str | None,
prisma_client: DatabaseClient | None,
) -> tuple[str, ...]:
if not ungranted or not access_group_ids:
return ungranted
return ungranted_access_group_models(
names=ungranted,
granted_models=await get_models_from_access_groups(
access_group_ids=access_group_ids, prisma_client=prisma_client
),
llm_router=llm_router,
team_id=team_id,
)
async def _require_team_access_group_grant(
model: str | Sequence[str],
team_object: LiteLLM_TeamTable,
llm_router: Router | None,
team_model_aliases: Mapping[str, str] | None,
key_model_aliases: Mapping[str, str] | None,
prisma_client: DatabaseClient | None,
) -> None:
if team_object.team_id == UI_TEAM_ID or not _model_access_group_deny_by_default_enabled():
return
ungranted: Final = await _ungranted_after_access_group_ids(
ungranted=ungranted_access_group_models(
names=requested_model_names(model, team_model_aliases, key_model_aliases),
granted_models=team_object.models,
llm_router=llm_router,
team_id=team_object.team_id,
),
access_group_ids=team_object.access_group_ids,
llm_router=llm_router,
team_id=team_object.team_id,
prisma_client=prisma_client,
)
if ungranted:
raise access_group_model_denied(model=model, ungranted=ungranted, object_type="team")
async def get_authorized_resources_from_key_access_groups(
valid_token: UserAPIKeyAuth | None,
team_object: LiteLLM_TeamTable | None,
@ -5827,9 +5992,14 @@ async def key_access_group_grants_model(
key_model_aliases=key_model_aliases,
object_type="key",
)
return True
except ProxyException:
return False
return not _model_access_group_deny_by_default_enabled() or not ungranted_access_group_models(
names=requested_model_names(model, valid_token.team_model_aliases if valid_token else None, key_model_aliases),
granted_models=authorized_models,
llm_router=llm_router,
team_id=valid_token.team_id if valid_token else None,
)
_key_access_group_grants_model: Final = key_access_group_grants_model
@ -5890,6 +6060,7 @@ async def can_user_call_model(
llm_router: Router | None,
user_object: LiteLLM_UserTable | None,
key_model_aliases: Mapping[str, str] | None = None,
teams_grant_access_group_models: bool = False,
) -> Literal[True]:
if user_object is None:
return True
@ -5907,13 +6078,42 @@ async def can_user_call_model(
code=status.HTTP_403_FORBIDDEN,
)
return can_object_call_model(
user_models: Final = _user_models(user_object)
can_object_call_model(
model=model,
llm_router=llm_router,
models=user_object.models,
models=list(user_models),
key_model_aliases=key_model_aliases,
object_type="user",
)
if not teams_grant_access_group_models:
_require_user_access_group_grant(
model=model,
user_role=user_object.user_role,
user_models=user_models,
llm_router=llm_router,
key_model_aliases=key_model_aliases,
)
return True
def _require_user_access_group_grant(
model: str | Sequence[str],
user_role: str | None,
user_models: Sequence[str],
llm_router: Router | None,
key_model_aliases: Mapping[str, str] | None,
) -> None:
if user_role == LitellmUserRoles.PROXY_ADMIN or not _model_access_group_deny_by_default_enabled():
return
ungranted: Final = ungranted_access_group_models(
names=requested_model_names(model, key_model_aliases=key_model_aliases),
granted_models=user_models,
llm_router=llm_router,
team_id=None,
)
if ungranted:
raise access_group_model_denied(model=model, ungranted=ungranted, object_type="user")
def personal_key_team_model_access_applies(valid_token: UserAPIKeyAuth | None) -> bool:
@ -6558,7 +6758,7 @@ async def can_caller_call_search_tool(
async def can_token_call_search_tool(search_tool_name: str, valid_token: UserAPIKeyAuth) -> Literal[True]:
"""`can_caller_call_search_tool` against the proxy's own settings, team cache and database"""
from litellm.proxy.proxy_server import general_settings, prisma_client, proxy_logging_obj, user_api_key_cache
from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache
async def _load_team_object() -> LiteLLM_TeamTable | None:
if not valid_token.team_id:
@ -6574,7 +6774,7 @@ async def can_token_call_search_tool(search_tool_name: str, valid_token: UserAPI
return await can_caller_call_search_tool(
search_tool_name=search_tool_name,
valid_token=valid_token,
general_settings=typed_general_settings(general_settings),
general_settings=_proxy_general_settings(),
load_team_object=_load_team_object,
)

View file

@ -0,0 +1,141 @@
from collections.abc import Iterable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal
from fastapi import status
from pydantic import ConfigDict, ValidationError
import litellm
from litellm.proxy._types import (
ConfigGeneralSettings,
ModelAccessDeniedProxyException,
ProxyErrorTypes,
)
from litellm.proxy.auth.model_access_denied import model_access_denied_client_message
from litellm.types.llms.base_model import LiteLLMBaseModel
if TYPE_CHECKING:
from litellm.router import Router
_SETTING: Final = "model_access_group_deny_by_default"
class _DeploymentModelInfo(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
access_groups: tuple[str, ...] | None = None
class _DeploymentRow(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
model_name: str
model_info: _DeploymentModelInfo | None = None
@property
def groups(self) -> frozenset[str]:
return frozenset(self.model_info.access_groups or ()) if self.model_info is not None else frozenset()
def model_access_group_deny_by_default(general_settings: Mapping[str, object]) -> bool:
"""
Startup rejects a non-boolean value from the config file. A non-boolean value that reaches
general_settings another way enables the policy, so it fails closed
"""
value: Final = general_settings.get(_SETTING, False)
if isinstance(value, bool):
return value
try:
return ConfigGeneralSettings.model_validate(
MappingProxyType({_SETTING: value})
).model_access_group_deny_by_default
except ValidationError:
return True
def _alias_chain(
model: str, team_model_aliases: Mapping[str, str] | None, key_model_aliases: Mapping[str, str] | None
) -> Iterator[str]:
after_team: Final = team_model_aliases.get(model, model) if team_model_aliases else model
after_key: Final = key_model_aliases.get(after_team, after_team) if key_model_aliases else after_team
yield from (model, after_team, after_key, litellm.model_alias_map.get(after_key, after_key))
def _alias_chains(
model: str | Sequence[str],
team_model_aliases: Mapping[str, str] | None,
key_model_aliases: Mapping[str, str] | None,
) -> Iterator[str]:
for requested in (model,) if isinstance(model, str) else model:
yield from _alias_chain(requested, team_model_aliases, key_model_aliases)
def requested_model_names(
model: str | Sequence[str],
team_model_aliases: Mapping[str, str] | None = None,
key_model_aliases: Mapping[str, str] | None = None,
) -> tuple[str, ...]:
"""The requested models plus every model their team, key and proxy aliases point at"""
return tuple(dict.fromkeys(_alias_chains(model, team_model_aliases, key_model_aliases)))
def _rows(rows: Iterable[object]) -> Iterator[_DeploymentRow]:
for row in rows:
yield _DeploymentRow.model_validate(row)
def _routing_group_member_rows(name: str, llm_router: "Router", team_id: str | None) -> Iterator[_DeploymentRow]:
group: Final = llm_router.get_routing_group(llm_router.get_model_from_alias(model=name) or name)
for member in group.models if group is not None else ():
yield from _rows(llm_router.get_model_list(model_name=member, team_id=team_id) or ())
def _reachable_deployments(name: str, llm_router: "Router", team_id: str | None) -> Iterator[_DeploymentRow]:
"""Every deployment a request for `name` can be served by: its model group (aliases and wildcards
included), a routing group's members, a deployment id, or a deployment's litellm_params.model"""
yield from _rows(llm_router.get_model_list(model_name=name, team_id=team_id) or ())
yield from _routing_group_member_rows(name, llm_router, team_id)
deployment: Final = llm_router.get_deployment(model_id=name) if llm_router.has_model_id(name) else None
if deployment is not None:
yield _DeploymentRow.model_validate(deployment.model_dump(exclude_none=True))
if name in llm_router.deployment_names:
yield from _rows(row for row in llm_router.get_model_list() or () if row["litellm_params"].get("model") == name)
def ungranted_access_group_models(
names: Sequence[str],
granted_models: Sequence[str],
llm_router: "Router | None",
team_id: str | None,
) -> tuple[str, ...]:
"""
The names that can reach a deployment in a model access group which `granted_models` does not name,
by group name, the deployment's model name, or the requested name. Names match exactly, so a blanket
grant (empty list, "*", all-proxy-models, all-team-models) or a wildcard pattern names nothing
"""
if llm_router is None:
return ()
explicit: Final = frozenset(granted_models)
return tuple(name for name in names if not _is_explicitly_granted(name, explicit, llm_router, team_id))
def _is_explicitly_granted(name: str, explicit: frozenset[str], llm_router: "Router", team_id: str | None) -> bool:
return name in explicit or all(
not row.groups or row.model_name in explicit or not explicit.isdisjoint(row.groups)
for row in _reachable_deployments(name, llm_router, team_id)
)
def access_group_model_denied(
model: str | Sequence[str], ungranted: Sequence[str], object_type: Literal["key", "user", "team"]
) -> ModelAccessDeniedProxyException:
return ModelAccessDeniedProxyException(
message=model_access_denied_client_message(model=model if isinstance(model, str) else list(model)),
internal_message=(
f"{object_type} not allowed to access model. {', '.join(ungranted)} reaches a model access group and "
f"general_settings.{_SETTING} requires the {object_type} to list that group or model explicitly"
),
type=ProxyErrorTypes.get_model_access_error_type_for_object(object_type=object_type),
param="model",
code=status.HTTP_403_FORBIDDEN,
)

View file

@ -6922,6 +6922,7 @@ class ProxyConfig:
for name in (
"vector_store_deny_by_default",
"search_tool_deny_by_default",
"model_access_group_deny_by_default",
"vertex_ai_pass_through_drop_anthropic_betas",
"request_tag_mappings",
)

View file

@ -1327,7 +1327,7 @@ class Router:
self.assistants_config = assistants_config
self.search_tools = search_tools or []
self.guardrail_list = guardrail_list or []
self.deployment_names: list = [] # names of models under litellm_params. ex. azure/chatgpt-v-2
self.deployment_names: list[str] = [] # names of models under litellm_params. ex. azure/chatgpt-v-2
self.deployment_latency_map = {}
### CACHING ###
cache_type: Literal["local", "redis", "redis-semantic", "s3", "disk"] = "local" # default to an in-memory cache

View file

@ -1416,6 +1416,7 @@ async def test_find_team_with_model_access_model_group(monkeypatch):
proxy_server_module = types.ModuleType("proxy_server")
proxy_server_module.llm_router = router
proxy_server_module.general_settings = {}
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
team = LiteLLM_TeamTable(team_id="team-1", models=["test-group"])
@ -1467,6 +1468,7 @@ async def test_find_team_with_model_access_resolves_team_model_alias(monkeypatch
router = Router(model_list=[{"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o"}}])
proxy_server_module = types.ModuleType("proxy_server")
proxy_server_module.llm_router = router
proxy_server_module.general_settings = {}
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
team = LiteLLM_TeamTable(
@ -1526,6 +1528,7 @@ async def test_find_team_with_model_access_v1_messages_default_routes(monkeypatc
proxy_server_module = types.ModuleType("proxy_server")
proxy_server_module.llm_router = router
proxy_server_module.general_settings = {}
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
team = LiteLLM_TeamTable(team_id="coding-team", models=["coding_only_models"])
@ -1597,6 +1600,7 @@ async def test_find_team_with_model_access_reports_team_route_allowlist_denial()
prisma_team_table.find_unique = AsyncMock()
proxy_server_module: Final = ModuleType("proxy_server")
proxy_server_module.llm_router = None
proxy_server_module.general_settings = {}
with patch.dict(sys.modules, {"litellm.proxy.proxy_server": proxy_server_module}):
with pytest.raises(HTTPException) as exc_info:
@ -3707,6 +3711,7 @@ async def test_find_team_with_model_access_unresolved_group_claim_returns_none(
)
proxy_server_module = types.ModuleType("proxy_server")
proxy_server_module.llm_router = router
proxy_server_module.general_settings = {}
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
async def raise_404(*_args, **_kwargs):
@ -3844,6 +3849,7 @@ async def test_find_team_with_model_access_resolved_team_without_model_still_rai
)
proxy_server_module = types.ModuleType("proxy_server")
proxy_server_module.llm_router = router
proxy_server_module.general_settings = {}
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
team = LiteLLM_TeamTable(team_id="real-team", models=["gpt-3.5-turbo"])
@ -3925,6 +3931,7 @@ async def test_find_team_with_model_access_unresolved_group_claim_default_raises
)
proxy_server_module = types.ModuleType("proxy_server")
proxy_server_module.llm_router = router
proxy_server_module.general_settings = {}
monkeypatch.setitem(sys.modules, "litellm.proxy.proxy_server", proxy_server_module)
async def raise_404(*_args, **_kwargs):

View file

@ -0,0 +1,345 @@
from collections.abc import Iterator, Sequence
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import Request
import litellm
from litellm.proxy import proxy_server
from litellm.proxy._types import (
LiteLLM_AccessGroupTable,
LiteLLM_TeamTable,
LiteLLM_TeamTableCachedObj,
LiteLLM_UserTable,
LitellmUserRoles,
ModelAccessDeniedProxyException,
ProxyErrorTypes,
UserAPIKeyAuth,
)
from litellm.proxy.auth import auth_checks
from litellm.proxy.auth.auth_checks import (
can_key_call_model,
can_key_call_resolved_model,
can_user_call_model,
common_checks,
)
from litellm.proxy.auth.model_access_group_grants import ungranted_access_group_models
_ROUTER: Final = litellm.Router(
model_list=[
{
"model_name": "private-model",
"litellm_params": {"model": "openai/private-only-upstream", "api_key": "sk-fake"},
"model_info": {"id": "private-deployment-id", "access_groups": ["group-x"]},
},
{
"model_name": "openai/private-ft",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-fake"},
"model_info": {"access_groups": ["group-x"]},
},
{"model_name": "public-model", "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-fake"}},
],
model_group_alias={"alias-to-private": "private-model"},
routing_groups=[
{"group_name": "mixed-group", "models": ["private-model", "public-model"], "routing_strategy": "simple-shuffle"}
],
)
@pytest.fixture
def deny_by_default(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setitem(proxy_server.general_settings, "model_access_group_deny_by_default", True)
yield
@pytest.fixture
def no_team_membership(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
async def _no_membership(**_: object) -> None:
return None
monkeypatch.setattr(auth_checks, "get_team_membership", _no_membership)
yield
async def _common_checks(
model: str,
token: UserAPIKeyAuth,
team: LiteLLM_TeamTable | None = None,
user: LiteLLM_UserTable | None = None,
) -> bool:
return await common_checks(
request_body={"model": model},
team_object=team,
user_object=user,
end_user_object=None,
global_proxy_spend=None,
general_settings=proxy_server.general_settings,
route="/chat/completions",
llm_router=_ROUTER,
proxy_logging_obj=MagicMock(),
valid_token=token,
request=MagicMock(spec=Request),
)
def _team(models: Sequence[str], access_group_ids: Sequence[str] = ()) -> LiteLLM_TeamTable:
return LiteLLM_TeamTable(team_id="team-1", models=list(models), access_group_ids=list(access_group_ids))
_TEAM_KEY: Final = UserAPIKeyAuth(token="hashed-team-key", team_id="team-1")
@pytest.mark.asyncio
@pytest.mark.parametrize(
"team_models, model, allowed",
[
([], "private-model", False),
(["*"], "private-model", False),
(["all-proxy-models"], "private-model", False),
(["openai/*"], "openai/private-ft", False),
(["group-x"], "private-model", True),
(["private-model"], "private-model", True),
([], "public-model", True),
],
)
async def test_team_needs_an_explicit_grant_for_access_group_models(
deny_by_default: None, no_team_membership: None, team_models: list[str], model: str, allowed: bool
) -> None:
if allowed:
assert await _common_checks(model, _TEAM_KEY, team=_team(team_models)) is True
return
with pytest.raises(ModelAccessDeniedProxyException) as exc_info:
await _common_checks(model, _TEAM_KEY, team=_team(team_models))
assert exc_info.value.type == ProxyErrorTypes.team_model_access_denied
assert str(exc_info.value.code) == "403"
@pytest.mark.asyncio
@pytest.mark.parametrize("team_models, model", [([], "private-model"), (["openai/*"], "openai/private-ft")])
async def test_blanket_team_grant_is_unchanged_when_setting_is_off(
monkeypatch: pytest.MonkeyPatch, no_team_membership: None, team_models: list[str], model: str
) -> None:
monkeypatch.delitem(proxy_server.general_settings, "model_access_group_deny_by_default", raising=False)
assert await _common_checks(model, _TEAM_KEY, team=_team(team_models)) is True
@pytest.mark.asyncio
@pytest.mark.parametrize("model", ["private-deployment-id", "openai/private-only-upstream", "mixed-group"])
async def test_blanket_team_cannot_reach_an_access_group_deployment_by_another_name(
deny_by_default: None, no_team_membership: None, model: str
) -> None:
with pytest.raises(ModelAccessDeniedProxyException):
await _common_checks(model, _TEAM_KEY, team=_team([]))
@pytest.mark.parametrize(
"name, granted, ungranted",
[
("private-deployment-id", ["private-model"], ()),
("private-deployment-id", ["private-deployment-id"], ()),
("openai/private-only-upstream", ["group-x"], ()),
("mixed-group", ["private-model"], ()),
("mixed-group", ["public-model"], ("mixed-group",)),
("alias-to-private", ["public-model"], ("alias-to-private",)),
("alias-to-private", ["group-x"], ()),
("mixed-group", ["*", "all-proxy-models", "openai/*"], ("mixed-group",)),
],
)
def test_another_name_for_a_deployment_is_granted_by_its_model_or_group(
name: str, granted: list[str], ungranted: tuple[str, ...]
) -> None:
assert ungranted_access_group_models(names=[name], granted_models=granted, llm_router=_ROUTER, team_id=None) == (
ungranted
)
@pytest.mark.asyncio
@pytest.mark.parametrize("team_models, allowed", [([], False), (["group-x"], True)])
async def test_team_alias_cannot_reach_an_ungranted_access_group_model(
deny_by_default: None, no_team_membership: None, team_models: list[str], allowed: bool
) -> None:
token: Final = UserAPIKeyAuth(
token="hashed-team-key", team_id="team-1", team_model_aliases={"nice-name": "private-model"}
)
if allowed:
assert await _common_checks("nice-name", token, team=_team(team_models)) is True
return
with pytest.raises(ModelAccessDeniedProxyException):
await _common_checks("nice-name", token, team=_team(team_models))
@pytest.mark.asyncio
@pytest.mark.parametrize("granted_names, allowed", [(["private-model"], True), (["public-model"], False)])
async def test_team_unified_access_group_counts_as_an_explicit_grant(
deny_by_default: None, no_team_membership: None, granted_names: list[str], allowed: bool
) -> None:
group: Final = LiteLLM_AccessGroupTable(
access_group_id="ag-1", access_group_name="private", access_model_names=granted_names
)
with patch.object(auth_checks, "get_access_object", new_callable=AsyncMock, return_value=group):
team: Final = _team([], access_group_ids=["ag-1"])
if allowed:
assert await _common_checks("private-model", _TEAM_KEY, team=team) is True
return
with pytest.raises(ModelAccessDeniedProxyException):
await _common_checks("private-model", _TEAM_KEY, team=team)
@pytest.mark.asyncio
async def test_empty_key_list_on_a_team_key_defers_to_the_team_grant(deny_by_default: None) -> None:
token: Final = UserAPIKeyAuth(token="hashed-team-key", team_id="team-1", models=[])
assert await can_key_call_model(model="private-model", llm_model_list=None, valid_token=token, llm_router=_ROUTER)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"user, allowed",
[
(LiteLLM_UserTable(user_id="u1", models=[]), False),
(LiteLLM_UserTable(user_id="u1", models=["group-x"]), True),
(LiteLLM_UserTable(user_id="u1", models=[], user_role=LitellmUserRoles.PROXY_ADMIN.value), True),
],
)
async def test_personal_key_needs_an_explicit_user_grant(
deny_by_default: None, user: LiteLLM_UserTable, allowed: bool
) -> None:
token: Final = UserAPIKeyAuth(token="hashed-personal-key", user_id="u1")
if allowed:
assert await _common_checks("private-model", token, user=user) is True
return
with pytest.raises(ModelAccessDeniedProxyException) as exc_info:
await _common_checks("private-model", token, user=user)
assert exc_info.value.type == ProxyErrorTypes.user_model_access_denied
@pytest.mark.asyncio
async def test_user_grant_is_skipped_when_owner_teams_decide(deny_by_default: None) -> None:
assert await can_user_call_model(
model="private-model",
llm_router=_ROUTER,
user_object=LiteLLM_UserTable(user_id="u1", models=[]),
teams_grant_access_group_models=True,
)
@pytest.mark.asyncio
@pytest.mark.parametrize("owner_models, allowed", [([], False), (["private-model"], True)])
async def test_resolved_model_for_a_personal_key_needs_the_owner_grant(
monkeypatch: pytest.MonkeyPatch, deny_by_default: None, owner_models: list[str], allowed: bool
) -> None:
async def _get_user_object(**_: object) -> LiteLLM_UserTable:
return LiteLLM_UserTable(user_id="u1", models=owner_models)
monkeypatch.setattr(auth_checks, "get_user_object", _get_user_object)
token: Final = UserAPIKeyAuth(token="hashed-personal-key", user_id="u1")
if allowed:
await can_key_call_resolved_model(
model="private-model", llm_model_list=None, valid_token=token, llm_router=_ROUTER
)
return
with pytest.raises(ModelAccessDeniedProxyException):
await can_key_call_resolved_model(
model="private-model", llm_model_list=None, valid_token=token, llm_router=_ROUTER
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"token, allowed",
[
(UserAPIKeyAuth(token="hashed-service-key", models=[]), False),
(UserAPIKeyAuth(token="hashed-service-key", models=["group-x"]), True),
(UserAPIKeyAuth(token="hashed-service-key", models=[], user_role=LitellmUserRoles.PROXY_ADMIN), True),
],
)
async def test_key_with_no_team_or_user_owns_its_grant(
deny_by_default: None, token: UserAPIKeyAuth, allowed: bool
) -> None:
if allowed:
assert await can_key_call_model(
model="private-model", llm_model_list=None, valid_token=token, llm_router=_ROUTER
)
return
with pytest.raises(ModelAccessDeniedProxyException) as exc_info:
await can_key_call_model(model="private-model", llm_model_list=None, valid_token=token, llm_router=_ROUTER)
assert exc_info.value.type == ProxyErrorTypes.key_model_access_denied
@pytest.mark.asyncio
@pytest.mark.parametrize(
"token, allowed",
[
(UserAPIKeyAuth(token="hashed-service-key", models=["all-team-models"]), False),
(UserAPIKeyAuth(token="hashed-personal-key", user_id="missing-user", models=[]), False),
(UserAPIKeyAuth(token="hashed-personal-key", user_id="missing-user", models=["group-x"]), True),
],
)
async def test_teamless_key_without_a_loaded_owner_needs_its_own_grant(
deny_by_default: None, token: UserAPIKeyAuth, allowed: bool
) -> None:
if allowed:
assert await _common_checks("private-model", token) is True
return
with pytest.raises(ModelAccessDeniedProxyException) as exc_info:
await _common_checks("private-model", token)
assert exc_info.value.type == ProxyErrorTypes.key_model_access_denied
@pytest.mark.asyncio
async def test_resolved_model_for_an_all_team_models_key_with_no_team_needs_its_own_grant(
deny_by_default: None,
) -> None:
token: Final = UserAPIKeyAuth(token="hashed-service-key", models=["all-team-models"])
with pytest.raises(ModelAccessDeniedProxyException):
await can_key_call_resolved_model(
model="private-model", llm_model_list=None, valid_token=token, llm_router=_ROUTER
)
@pytest.mark.asyncio
@pytest.mark.parametrize("key_group_models, allowed", [(["*"], False), (["private-model"], True)])
async def test_key_access_group_overrides_the_team_only_with_an_explicit_grant(
monkeypatch: pytest.MonkeyPatch,
deny_by_default: None,
no_team_membership: None,
key_group_models: list[str],
allowed: bool,
) -> None:
monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
group: Final = LiteLLM_AccessGroupTable(
access_group_id="key-ag",
access_group_name="key-group",
access_model_names=key_group_models,
assigned_key_ids=["hashed-team-key"],
)
token: Final = UserAPIKeyAuth(token="hashed-team-key", team_id="team-1", access_group_ids=["key-ag"])
with patch.object(auth_checks, "get_access_object", new_callable=AsyncMock, return_value=group):
if allowed:
assert await _common_checks("private-model", token, team=_team([])) is True
return
with pytest.raises(ModelAccessDeniedProxyException):
await _common_checks("private-model", token, team=_team([]))
@pytest.mark.asyncio
@pytest.mark.parametrize("owner_team_models, allowed", [(["group-x"], True), ([], False)])
async def test_personal_key_reaches_an_access_group_model_through_the_owner_team_grant(
monkeypatch: pytest.MonkeyPatch,
deny_by_default: None,
no_team_membership: None,
owner_team_models: list[str],
allowed: bool,
) -> None:
async def _get_team_object(**_: object) -> LiteLLM_TeamTableCachedObj:
return LiteLLM_TeamTableCachedObj(team_id="team-1", models=owner_team_models)
monkeypatch.setattr(litellm, "personal_key_model_access_from_teams", True)
monkeypatch.setattr(auth_checks, "get_team_object", _get_team_object)
token: Final = UserAPIKeyAuth(token="hashed-personal-key", user_id="u1")
owner: Final = LiteLLM_UserTable(user_id="u1", models=[], teams=["team-1"])
if allowed:
assert await _common_checks("private-model", token, user=owner) is True
return
with pytest.raises(ModelAccessDeniedProxyException):
await _common_checks("private-model", token, user=owner)

View file

@ -2546,7 +2546,9 @@ async def test_load_config_logs_disabled_budget_reservation_once(tmp_path, monke
@pytest.mark.asyncio
@pytest.mark.parametrize("setting", ["vector_store_deny_by_default", "search_tool_deny_by_default"])
@pytest.mark.parametrize(
"setting", ["vector_store_deny_by_default", "search_tool_deny_by_default", "model_access_group_deny_by_default"]
)
@pytest.mark.parametrize(("yaml_value", "expected"), [("true", True), ("false", False)])
async def test_load_config_yaml_deny_by_default_is_boolean(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, setting: str, yaml_value: str, expected: bool
@ -2564,7 +2566,9 @@ async def test_load_config_yaml_deny_by_default_is_boolean(
@pytest.mark.asyncio
@pytest.mark.parametrize("setting", ["vector_store_deny_by_default", "search_tool_deny_by_default"])
@pytest.mark.parametrize(
"setting", ["vector_store_deny_by_default", "search_tool_deny_by_default", "model_access_group_deny_by_default"]
)
@pytest.mark.parametrize("yaml_value", ["", "enabled"], ids=["null", "string"])
async def test_load_config_rejects_non_boolean_deny_by_default(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, setting: str, yaml_value: str

View file

@ -30148,6 +30148,12 @@ export interface components {
* @description What to do with LLM API requests that carry no session id (x-litellm-session-id header, metadata.session_id, etc.). 'generate' stamps one id into litellm_session_id, litellm_trace_id and metadata.session_id so SpendLogs and logging callbacks agree; 'reject' returns 400; 'omit' leaves SpendLogs.session_id null, matching callbacks such as Langfuse that only record a client-established metadata.session_id. Unset keeps the legacy behavior where SpendLogs falls back to the trace id while callbacks get no session id.
*/
missing_session_id?: ("generate" | "reject" | "omit") | null;
/**
* Model Access Group Deny By Default
* @description When True, a model tagged with model_info.access_groups is only reachable by a caller whose grant names one of its groups or the model itself. The grant that counts is the team's for a team key, the user's for a personal key, and the key's own for a key with neither. An empty models list, '*', all-proxy-models and wildcard patterns no longer reach these models. Untagged models are unaffected, and proxy admins' personal access is exempt
* @default false
*/
model_access_group_deny_by_default: boolean;
/**
* Model List Healthy Only
* @description When true, `/models`, `/v1/models/{id}` and `/model/info` hide models whose backing deployments are all unhealthy, for every caller, without needing `healthy_only=true` per request. Requires `background_health_checks: true`, and keeps deployment health state cached without turning on `enable_health_check_routing`, so routing is unaffected. With no health state nothing is hidden. Hiding is presentation-only, a hidden model can still be called.