diff --git a/litellm/__init__.py b/litellm/__init__.py index 8e5dd68a186..76df10d1d28 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -367,6 +367,7 @@ enable_preview_features: bool = False return_response_headers: bool = False # get response headers from LLM Api providers - example x-remaining-requests, enable_json_schema_validation: bool = False enable_model_config_credential_overrides: bool = False +enable_model_prefix_enforcement: bool = False enable_key_alias_format_validation: bool = ( False # opt-in validation of key_alias format on /key/generate and /key/update ) diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py index af60a315c60..9c5c8291448 100644 --- a/litellm/proxy/auth/auth_checks.py +++ b/litellm/proxy/auth/auth_checks.py @@ -150,6 +150,7 @@ from litellm.proxy.guardrails.tool_name_extraction import ( ) from litellm.proxy.route_llm_request import route_request from litellm.proxy.spend_tracking.budget_reservation import get_budget_window_start +from litellm.proxy.litellm_pre_call_utils import resolve_wildcard_prefixed_model from litellm.proxy.spend_tracking.carried_budget_state import carry_organization_budget_state from litellm.proxy.spend_tracking.spend_counter_batch import SPEND_COUNTERS_TARGET from litellm.proxy.utils import PrismaClient, ProxyLogging, log_db_metrics @@ -5007,30 +5008,101 @@ def _can_object_call_model( from litellm.router_strategy.complexity_router.context_compaction import native_compaction_parent - compaction_parent: Final = native_compaction_parent(model) - global_or_router_alias_target: Final = ( - litellm.model_alias_map[model] - if model in litellm.model_alias_map + effective_model: str = model + if litellm.enable_model_prefix_enforcement and llm_router is not None: + _is_direct_alias = ( + (team_model_aliases and model in team_model_aliases) + or (key_model_aliases and model in key_model_aliases) + or (model in litellm.model_alias_map) + or (model in (llm_router.registry().aliases() if hasattr(llm_router, "registry") else (llm_router.model_group_alias or {}))) + ) + if not _is_direct_alias: + _resolved_model = resolve_wildcard_prefixed_model(model=model, llm_router=llm_router) + if _resolved_model is None: + raise ModelAccessDeniedProxyException( + message=( + customer_model_access_denied_client_message(model=model) + if object_type == "customer" + else model_access_denied_client_message(model=model) + ), + internal_message=f"Invalid or mismatched model prefix for model={model}", + type=ProxyErrorTypes.get_model_access_error_type_for_object(object_type=object_type), + param="model", + code=status.HTTP_403_FORBIDDEN, + ) + effective_model = _resolved_model + + compaction_parent: Final = native_compaction_parent(effective_model) + global_or_router_alias_target: str | None = ( + litellm.model_alias_map[effective_model] + if effective_model in litellm.model_alias_map else ( - llm_router.registry().resolve(model, kind="alias") - if llm_router is not None and model in llm_router.registry().aliases() + llm_router.registry().resolve(effective_model, kind="alias") + if llm_router is not None and hasattr(llm_router, "registry") and effective_model in llm_router.registry().aliases() else None ) ) - after_team_alias: Final = team_model_aliases.get(model, model) if team_model_aliases else model + if ( + litellm.enable_model_prefix_enforcement + and llm_router is not None + and global_or_router_alias_target is not None + and not global_or_router_alias_target.startswith("model_name_") + ): + _resolved_global_target = resolve_wildcard_prefixed_model( + model=global_or_router_alias_target, llm_router=llm_router + ) + if _resolved_global_target is None: + raise ModelAccessDeniedProxyException( + message=( + customer_model_access_denied_client_message(model=model) + if object_type == "customer" + else model_access_denied_client_message(model=model) + ), + internal_message=f"Invalid or mismatched model prefix for alias target={global_or_router_alias_target}", + type=ProxyErrorTypes.get_model_access_error_type_for_object(object_type=object_type), + param="model", + code=status.HTTP_403_FORBIDDEN, + ) + global_or_router_alias_target = _resolved_global_target + after_team_alias: Final = ( + team_model_aliases.get(effective_model, effective_model) if team_model_aliases else effective_model + ) after_key_alias: Final = ( key_model_aliases.get(after_team_alias, after_team_alias) if key_model_aliases else after_team_alias ) after_global_alias: Final = litellm.model_alias_map.get(after_key_alias, after_key_alias) - dispatched_model: Final = ( + dispatched_model: str = ( key_model_aliases.get(after_global_alias, after_global_alias) if key_model_aliases else after_global_alias ) key_alias_applied: Final = after_key_alias != after_team_alias or dispatched_model != after_global_alias + if ( + litellm.enable_model_prefix_enforcement + and llm_router is not None + and (key_alias_applied or after_team_alias != effective_model) + and not dispatched_model.startswith("model_name_") + ): + _resolved_dispatched = resolve_wildcard_prefixed_model(model=dispatched_model, llm_router=llm_router) + if _resolved_dispatched is None: + raise ModelAccessDeniedProxyException( + message=( + customer_model_access_denied_client_message(model=model) + if object_type == "customer" + else model_access_denied_client_message(model=model) + ), + internal_message=f"Invalid or mismatched model prefix for alias target={dispatched_model}", + type=ProxyErrorTypes.get_model_access_error_type_for_object(object_type=object_type), + param="model", + code=status.HTTP_403_FORBIDDEN, + ) + if key_alias_applied: + dispatched_model = _resolved_dispatched + else: + effective_model = _resolved_dispatched potential_models: Final = ( (dispatched_model,) if key_alias_applied else ( - *((model, compaction_parent) if compaction_parent is not None else (model,)), + *((effective_model, compaction_parent) if compaction_parent is not None else (effective_model,)), *((global_or_router_alias_target,) if global_or_router_alias_target else ()), ) ) @@ -5038,7 +5110,7 @@ def _can_object_call_model( () 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 + model=effective_model, llm_router=llm_router, team_id=team_id, team_model_aliases=team_model_aliases ) ) @@ -7783,6 +7855,9 @@ def is_model_allowed_by_pattern(model: str, allowed_model_pattern: str) -> bool: bool: True if model matches the pattern, False otherwise """ if "*" in allowed_model_pattern: + prefix: Final = allowed_model_pattern.removesuffix("*") + if allowed_model_pattern.endswith("*") and "*" not in prefix and "\n" not in model: + return model.startswith(prefix) pattern: Final = f"^{allowed_model_pattern.replace('*', '.*')}$" return bool(re.match(pattern, model)) diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 82135cb1b74..8e7b23fa461 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -339,14 +339,26 @@ async def _resolve_router_settings_model_group_alias( rewrite_request_model(request_data, request, target) +from litellm.proxy.litellm_pre_call_utils import resolve_wildcard_prefixed_model + + def _get_model_names_for_budget_checks( model: str | list[str] | None, ) -> list[str]: if model is None: return [] - if isinstance(model, str): - return [model] - return model + base_models: list[str] = [model] if isinstance(model, str) else list(model) + if litellm.enable_model_prefix_enforcement: + expanded: list[str] = [] + for m in base_models: + if m not in expanded: + expanded.append(m) + if isinstance(m, str): + resolved = resolve_wildcard_prefixed_model(model=m) + if resolved is not None and resolved not in expanded: + expanded.append(resolved) + return expanded + return base_models def _budget_fallback_target_model(target: object) -> str | None: diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index 4444d6d25f9..257cca829d2 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -1,3 +1,4 @@ +import sys import asyncio import copy import itertools @@ -3081,6 +3082,20 @@ async def add_litellm_data_to_request( user_api_key_dict=user_api_key_dict, ) + # Normalize non-alias model name to configured router Model Name before alias resolution + if litellm.enable_model_prefix_enforcement and llm_router is not None and isinstance((_cur_model := data.get("model")), str): + _team_aliases = user_api_key_dict.team_model_aliases + _key_aliases = user_api_key_dict.aliases + _is_direct_alias = ( + (isinstance(_team_aliases, dict) and _cur_model in _team_aliases) + or (isinstance(_key_aliases, dict) and _cur_model in _key_aliases) + or (_cur_model in litellm.model_alias_map) + or (_cur_model in (llm_router.registry().aliases() if hasattr(llm_router, "registry") else (llm_router.model_group_alias or {}))) + ) + if not _is_direct_alias: + if (_resolved_model := resolve_wildcard_prefixed_model(_cur_model, llm_router)) is not None and _resolved_model != _cur_model: + data["model"] = _resolved_model + # Save pre-alias model name for credential override lookup _pre_alias_model: Final = data.get("model") @@ -3096,6 +3111,17 @@ async def add_litellm_data_to_request( user_api_key_dict=user_api_key_dict, ) + # If an alias rewrote the model to a non-team-internal target, normalize that target as well + if ( + litellm.enable_model_prefix_enforcement + and llm_router is not None + and isinstance((_post_alias_model := data.get("model")), str) + and _post_alias_model != _pre_alias_model + and not _post_alias_model.startswith("model_name_") + ): + if (_resolved_post := resolve_wildcard_prefixed_model(_post_alias_model, llm_router)) is not None and _resolved_post != _post_alias_model: + data["model"] = _resolved_post + data[_metadata_variable_name]["litellm_roi_estimator"] = ( getattr(request.state, "litellm_roi_estimator", False) is True ) @@ -3666,6 +3692,206 @@ def _update_model_if_key_alias_exists( data["model"] = user_api_key_dict.aliases[_model] +from litellm.litellm_core_utils.get_llm_provider_logic import ( + resolve_model_and_provider_for_metadata, +) + +_MODEL_FAMILY_CACHE_LEN: int = -1 +_MODEL_FAMILY_PROVIDER_MAP: dict[str, str] = {} +_ROUTER_MODEL_LIST_REF: list | None = None +_ROUTER_MODEL_LIST_LEN: int = -1 +_ROUTER_EXACT_MODELS: frozenset[str] = frozenset() +_ROUTER_EXACT_LITELLM_MAP: dict[str, str] = {} +_ROUTER_PREFIX_TO_PUB: dict[str, str] = {} +_ROUTER_PROVIDER_TO_RULE: dict[str, tuple[str, str, str]] = {} +_ROUTER_RESOLVED_CACHE: dict[str, str] = {} + + +def _infer_provider_from_model_cost_family(bare_model: str) -> str | None: + global _MODEL_FAMILY_CACHE_LEN + model_cost = litellm.model_cost + if isinstance(model_cost, dict) and len(model_cost) != _MODEL_FAMILY_CACHE_LEN: + _MODEL_FAMILY_PROVIDER_MAP.clear() + for sample_m, sample_info in model_cost.items(): + if not isinstance(sample_m, str) or "/" in sample_m or not isinstance(sample_info, dict): + continue + fam, has_dash, _ = sample_m.lower().partition("-") + prov = sample_info.get("litellm_provider") + if has_dash and fam and isinstance(prov, str) and prov and fam not in _MODEL_FAMILY_PROVIDER_MAP: + _MODEL_FAMILY_PROVIDER_MAP[fam] = prov + _MODEL_FAMILY_CACHE_LEN = len(model_cost) + family, has_dash, _ = bare_model.lower().partition("-") + return _MODEL_FAMILY_PROVIDER_MAP.get(family) if has_dash else None + + +def _rebuild_router_prefix_index(model_list: list) -> None: + global _ROUTER_MODEL_LIST_REF, _ROUTER_MODEL_LIST_LEN + global _ROUTER_EXACT_MODELS, _ROUTER_EXACT_LITELLM_MAP + global _ROUTER_PREFIX_TO_PUB, _ROUTER_PROVIDER_TO_RULE + + exact_models: set[str] = set() + exact_litellm_map: dict[str, str] = {} + prefix_to_pub: dict[str, str] = {} + provider_to_rule: dict[str, tuple[str, str, str]] = {} + pending_exact: list[tuple[str, str, str]] = [] + + for dep in model_list: + if isinstance(dep, dict): + pub_name = dep.get("model_name", "") + litellm_params = dep.get("litellm_params") or {} + model_info = dep.get("model_info") or {} + else: + pub_name = getattr(dep, "model_name", "") + litellm_params = getattr(dep, "litellm_params", None) or {} + model_info = getattr(dep, "model_info", None) or {} + + if not isinstance(pub_name, str) or not pub_name: + continue + + owner_team_id = ( + model_info.get("team_id") + if isinstance(model_info, dict) + else getattr(model_info, "team_id", None) + ) + if owner_team_id is not None or pub_name.startswith("model_name_"): + continue + + if isinstance(litellm_params, dict): + litellm_model = litellm_params.get("model", "") + dep_custom_provider = litellm_params.get("custom_llm_provider", "") + else: + litellm_model = getattr(litellm_params, "model", "") + dep_custom_provider = getattr(litellm_params, "custom_llm_provider", "") + + if pub_name.endswith("/*"): + if isinstance(litellm_model, str) and litellm_model.endswith("/*"): + llm_wildcard = litellm_model + elif litellm_model == "*" and isinstance(dep_custom_provider, str) and dep_custom_provider: + llm_wildcard = f"{dep_custom_provider}/*" + else: + continue + pub_prefix = pub_name.removesuffix("*") + litellm_prefix = llm_wildcard.removesuffix("*") + base_llm = litellm_prefix.removesuffix("/") + prefix_to_pub[pub_prefix] = pub_prefix + prefix_to_pub[litellm_prefix] = pub_prefix + provider_to_rule[base_llm] = (pub_prefix, litellm_prefix, base_llm) + else: + exact_models.add(pub_name) + if isinstance(litellm_model, str) and litellm_model: + pending_exact.append( + ( + pub_name, + litellm_model, + dep_custom_provider if isinstance(dep_custom_provider, str) else "", + ) + ) + + for pub_name, litellm_model, dep_prov in pending_exact: + llm_prov, llm_sep, llm_bare = litellm_model.partition("/") + if not llm_sep: + llm_bare = litellm_model + llm_prov = dep_prov + if llm_bare == pub_name or ( + llm_prov and f"{llm_prov}/" not in prefix_to_pub and llm_prov not in provider_to_rule + ): + exact_litellm_map.setdefault(litellm_model, pub_name) + if not llm_sep and dep_prov: + exact_litellm_map.setdefault(f"{dep_prov}/{litellm_model}", pub_name) + + _ROUTER_EXACT_MODELS = frozenset(exact_models) + _ROUTER_EXACT_LITELLM_MAP = exact_litellm_map + _ROUTER_PREFIX_TO_PUB = prefix_to_pub + _ROUTER_PROVIDER_TO_RULE = provider_to_rule + _ROUTER_RESOLVED_CACHE.clear() + _ROUTER_MODEL_LIST_REF = model_list + _ROUTER_MODEL_LIST_LEN = len(model_list) + + +def _compute_wildcard_prefixed_model(model: str) -> str | None: + if model in _ROUTER_EXACT_MODELS: + return model + if model in _ROUTER_EXACT_LITELLM_MAP: + return _ROUTER_EXACT_LITELLM_MAP[model] + + prefix_part, sep, bare_model = model.partition("/") + if not sep: + bare_model = model + + if not bare_model or (sep and bare_model in _ROUTER_EXACT_MODELS): + return None + + given_prefix = f"{prefix_part}/" if sep else "" + if sep: + if given_prefix not in _ROUTER_PREFIX_TO_PUB: + return None + if "/" in bare_model: + inner_prefix, inner_sep, _ = bare_model.partition("/") + if inner_sep and f"{inner_prefix}/" in _ROUTER_PREFIX_TO_PUB: + return None + + stripped_model = bare_model + custom_llm_provider: str | None = None + try: + stripped_model, custom_llm_provider = resolve_model_and_provider_for_metadata(model=bare_model) + except Exception: + custom_llm_provider = _infer_provider_from_model_cost_family(bare_model) + + if not custom_llm_provider: + if sep: + pub_p = _ROUTER_PREFIX_TO_PUB.get(given_prefix) + return f"{pub_p}{bare_model}" if pub_p else None + return model + + rule = _ROUTER_PROVIDER_TO_RULE.get(custom_llm_provider) + if rule is None: + base_prov, _, _ = custom_llm_provider.partition("-") + if base_prov == custom_llm_provider: + base_prov, _, _ = custom_llm_provider.partition("_") + rule = _ROUTER_PROVIDER_TO_RULE.get(base_prov) + + if rule is not None: + pub_prefix, litellm_prefix, _ = rule + if given_prefix in ("", pub_prefix, litellm_prefix): + return f"{pub_prefix}{stripped_model}" + return None + + return model if not sep else None + + +def resolve_wildcard_prefixed_model( + model: str, + llm_router: Router | None = None, +) -> str | None: + if not litellm.enable_model_prefix_enforcement: + return model + if llm_router is None: + _proxy_srv = sys.modules.get("litellm.proxy.proxy_server") + llm_router = getattr(_proxy_srv, "llm_router", None) if _proxy_srv is not None else None + if llm_router is None: + return model + model_list = llm_router.model_list + if model_list is _ROUTER_MODEL_LIST_REF and len(model_list) == _ROUTER_MODEL_LIST_LEN: + try: + return _ROUTER_RESOLVED_CACHE[model] + except (KeyError, TypeError): + if not isinstance(model, str) or not model: + return model + else: + if not model_list or not isinstance(model, str) or not model: + return model + _rebuild_router_prefix_index(model_list) + + resolved = _compute_wildcard_prefixed_model(model) + if resolved is not None and len(model) <= 1024 and len(resolved) <= 1024: + if len(_ROUTER_RESOLVED_CACHE) >= 4096: + _ROUTER_RESOLVED_CACHE.pop(next(iter(_ROUTER_RESOLVED_CACHE)), None) + _ROUTER_RESOLVED_CACHE[model] = resolved + if resolved != model: + _ROUTER_RESOLVED_CACHE[resolved] = resolved + return resolved + + def _apply_credential_overrides_from_model_config( data: dict, user_api_key_dict: UserAPIKeyAuth, diff --git a/tests/test_litellm/proxy/auth/test_model_prefix_enforcement.py b/tests/test_litellm/proxy/auth/test_model_prefix_enforcement.py new file mode 100644 index 00000000000..e056f96d832 --- /dev/null +++ b/tests/test_litellm/proxy/auth/test_model_prefix_enforcement.py @@ -0,0 +1,211 @@ +import pytest + +import litellm +from litellm.proxy._types import ModelAccessDeniedProxyException +from litellm.proxy.auth.auth_checks import can_object_call_model +from litellm.proxy.auth.user_api_key_auth import _get_model_names_for_budget_checks +from litellm.proxy.litellm_pre_call_utils import resolve_wildcard_prefixed_model + + +class _DummyRouter: + def __init__(self, model_list): + self.model_list = model_list + self.model_group_alias = {} + self.model_name_to_deployment_indices = {} + + def get_model_access_groups(self, model_name: str, team_id: str | None = None): + return [] + + +@pytest.fixture +def sample_router(): + return _DummyRouter( + model_list=[ + {"model_name": "openai/*", "litellm_params": {"model": "openai/*"}}, + {"model_name": "google/*", "litellm_params": {"model": "vertex_ai/*"}}, + {"model_name": "anthropic/*", "litellm_params": {"model": "anthropic/*"}}, + { + "model_name": "codex-auto-review", + "litellm_params": {"model": "openai/codex-auto-review"}, + }, + { + "model_name": "DeepSeek-V4-Flash", + "litellm_params": { + "model": "DeepSeek-V4-Flash", + "custom_llm_provider": "hosted_vllm", + }, + }, + ] + ) + + +def test_prefix_enforcement_disabled_by_default(sample_router, monkeypatch): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", False) + assert resolve_wildcard_prefixed_model("gpt-6.1-sol", sample_router) == "gpt-6.1-sol" + assert resolve_wildcard_prefixed_model("vertex_ai/gemini-2.5-pro", sample_router) == "vertex_ai/gemini-2.5-pro" + assert resolve_wildcard_prefixed_model("openai/gemini-3.8-flash", sample_router) == "openai/gemini-3.8-flash" + + +@pytest.mark.parametrize( + "raw_model,expected_model", + [ + ("gpt-6.1-sol", "openai/gpt-6.1-sol"), + ("openai/gpt-6.1-sol", "openai/gpt-6.1-sol"), + ("gemini-2.5-pro", "google/gemini-2.5-pro"), + ("vertex_ai/gemini-2.5-pro", "google/gemini-2.5-pro"), + ("google/gemini-2.5-pro", "google/gemini-2.5-pro"), + ("gemini-3.8-flash", "google/gemini-3.8-flash"), + ("vertex_ai/gemini-3.8-flash", "google/gemini-3.8-flash"), + ("google/gemini-3.8-flash", "google/gemini-3.8-flash"), + ("claude-sonnet-4-5", "anthropic/claude-sonnet-4-5"), + ("anthropic/claude-sonnet-4-5", "anthropic/claude-sonnet-4-5"), + ("codex-auto-review", "codex-auto-review"), + ("openai/codex-auto-review", "codex-auto-review"), + ("DeepSeek-V4-Flash", "DeepSeek-V4-Flash"), + ("hosted_vllm/DeepSeek-V4-Flash", "DeepSeek-V4-Flash"), + ], +) +def test_prefix_enforcement_normalizes_valid_inputs(sample_router, monkeypatch, raw_model, expected_model): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + assert resolve_wildcard_prefixed_model(raw_model, sample_router) == expected_model + + +@pytest.mark.parametrize( + "invalid_model", + [ + "openai/gemini-3.8-flash", + "vertex/gemini-3.8-flash", + "vertext/gemini-3.8-flash", + "test/gemini-3.8-flash", + "google/gpt-6.1-sol", + "openai/DeepSeek-V4-Flash", + "google/codex-auto-review", + "openai/openai/gpt-6.1-sol", + "openai/chatgpt/gpt-4o", + ], +) +def test_prefix_enforcement_rejects_mismatched_or_unknown_prefixes(sample_router, monkeypatch, invalid_model): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + assert resolve_wildcard_prefixed_model(invalid_model, sample_router) is None + with pytest.raises(ModelAccessDeniedProxyException) as exc_info: + can_object_call_model( + model=invalid_model, + llm_router=sample_router, + models=["openai/*", "google/*", "anthropic/*", "codex-auto-review", "DeepSeek-V4-Flash"], + object_type="team", + ) + assert str(exc_info.value.code) == "403" + + +def test_can_object_call_model_allows_normalized_wildcard_access(sample_router, monkeypatch): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + for valid_google_call in ("gemini-2.5-pro", "vertex_ai/gemini-2.5-pro", "google/gemini-2.5-pro"): + assert ( + can_object_call_model( + model=valid_google_call, + llm_router=sample_router, + models=["google/*"], + object_type="team", + ) + is True + ) + + +def test_key_alias_chained_after_prefix_normalization_is_enforced(monkeypatch): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + router = _DummyRouter( + model_list=[ + {"model_name": "allowed-model", "litellm_params": {"model": "openai/gpt-4o"}}, + ] + ) + with pytest.raises(ModelAccessDeniedProxyException) as exc_info: + can_object_call_model( + model="openai/gpt-4o", + llm_router=router, + models=["allowed-model"], + object_type="key", + key_model_aliases={"allowed-model": "private-model"}, + ) + assert str(exc_info.value.code) == "403" + assert "openai/gpt-4o" in exc_info.value.message + assert "private-model" not in exc_info.value.message + + +def test_team_scoped_deployments_are_skipped_and_never_leaked(monkeypatch): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + router = _DummyRouter( + model_list=[ + {"model_name": "openai/*", "litellm_params": {"model": "openai/*"}}, + { + "model_name": "model_name_team-1_secret-uuid", + "litellm_params": {"model": "openai/gpt-4o"}, + "model_info": {"team_id": "team-1"}, + }, + ] + ) + assert resolve_wildcard_prefixed_model("openai/gpt-4o", router) == "openai/gpt-4o" + with pytest.raises(ModelAccessDeniedProxyException) as exc_info: + can_object_call_model( + model="openai/gpt-4o", + llm_router=router, + models=["google/*"], + object_type="team", + ) + assert "model_name_team-1_secret-uuid" not in exc_info.value.message + assert "openai/gpt-4o" in exc_info.value.message + + +def test_budget_check_model_names_include_normalized_model(monkeypatch): + import litellm.proxy.proxy_server as proxy_server + + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + router = _DummyRouter( + model_list=[ + {"model_name": "allowed-model", "litellm_params": {"model": "openai/gpt-4o"}}, + ] + ) + monkeypatch.setattr(proxy_server, "llm_router", router) + assert _get_model_names_for_budget_checks("openai/gpt-4o") == ["openai/gpt-4o", "allowed-model"] + + +def test_oversized_model_names_are_not_cached(sample_router, monkeypatch): + from litellm.proxy import litellm_pre_call_utils + + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + oversized = "openai/gpt-4o-" + ("a" * 2000) + resolved = resolve_wildcard_prefixed_model(oversized, sample_router) + assert resolved == oversized + assert oversized not in litellm_pre_call_utils._ROUTER_RESOLVED_CACHE + + +def test_custom_named_deployment_does_not_hijack_wildcard_model(monkeypatch): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + router = _DummyRouter( + model_list=[ + {"model_name": "hr-assistant", "litellm_params": {"model": "openai/gpt-4o"}}, + {"model_name": "openai/*", "litellm_params": {"model": "openai/*"}}, + {"model_name": "codex-auto-review", "litellm_params": {"model": "openai/codex-auto-review"}}, + ] + ) + assert resolve_wildcard_prefixed_model("openai/gpt-4o", router) == "openai/gpt-4o" + assert resolve_wildcard_prefixed_model("gpt-4o", router) == "openai/gpt-4o" + assert resolve_wildcard_prefixed_model("hr-assistant", router) == "hr-assistant" + assert resolve_wildcard_prefixed_model("openai/codex-auto-review", router) == "codex-auto-review" + + +def test_in_place_model_list_mutation_and_unprefixed_custom_deployment(monkeypatch): + monkeypatch.setattr(litellm, "enable_model_prefix_enforcement", True) + router = _DummyRouter( + model_list=[ + {"model_name": "hr-bot", "litellm_params": {"model": "gpt-4o"}}, + {"model_name": "openai/*", "litellm_params": {"model": "openai/*"}}, + ] + ) + assert resolve_wildcard_prefixed_model("gpt-4o", router) == "openai/gpt-4o" + assert resolve_wildcard_prefixed_model("vertex_ai/gemini-2.5-pro", router) is None + + # Mutate model_list in-place (simulating /model/new) + router.model_list.append( + {"model_name": "google/*", "litellm_params": {"model": "vertex_ai/*"}} + ) + assert resolve_wildcard_prefixed_model("vertex_ai/gemini-2.5-pro", router) == "google/gemini-2.5-pro"