This commit is contained in:
Abhimanyu Kapur 2026-09-13 00:00:58 -07:00 committed by GitHub
commit a72d70a5f1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 1110 additions and 6 deletions

View file

@ -3999,6 +3999,10 @@ def _check_model_access_helper(
return True
# Read-only stand-in for an absent model_group_alias map.
_EMPTY_ALIAS_MAP: Final[Mapping[str, object]] = MappingProxyType({})
def _can_object_call_model(
model: str | list[str],
llm_router: Router | None,
@ -4058,6 +4062,34 @@ def _can_object_call_model(
):
return True
# Canonical resolution: if the requested model has no manual alias but
# *provably* serves the same model as something in the request's
# allowed-models list, that's a match. Crucially, this is NOT OR-semantics
# (as with manual aliases, where the admin blessed the alias itself). It's
# AND-on-target: the target must be explicitly allowed, and the raw
# requested name being in the allowlist grants nothing. This prevents a key
# allowed ["stale-deployment-name"] from gaining access to a different
# deployment via a canonical rewrite.
if llm_router and model not in (llm_router.model_group_alias or _EMPTY_ALIAS_MAP):
canonical_target: Final[object] = llm_router.resolve_canonical_model_name(
model=model,
request_team_id=team_id,
)
# Require a real model-group name; a non-string from a router stub must
# never be treated as a grant.
if (
isinstance(canonical_target, str)
and canonical_target
and _check_model_access_helper(
model=canonical_target,
llm_router=llm_router,
models=models,
team_model_aliases=team_model_aliases,
team_id=team_id,
)
):
return True
raise ProxyException(
message=f"{object_type} not allowed to access model. This {object_type} can only access models={models}. Tried to access {model}",
type=ProxyErrorTypes.get_model_access_error_type_for_object(object_type=object_type),

View file

@ -4727,6 +4727,30 @@ def should_load_db_object(object_type: str | SupportedDBObjectType) -> bool:
return any(str(obj) == object_type_str for obj in supported_db_objects)
def _proxy_router_general_settings(
configured: RouterGeneralSettings | Mapping[str, Any] | None,
) -> RouterGeneralSettings:
"""The proxy's RouterGeneralSettings, preserving operator config.
``async_only_mode`` is a proxy-runtime requirement -- only async clients are
initialised on this path -- so it is always forced on. Every other field the
operator set under ``router_settings.router_general_settings`` is kept.
Before this existed the proxy passed a hardcoded
``RouterGeneralSettings(async_only_mode=True)`` as an explicit keyword
alongside ``**router_params``, so an operator who set the key in config hit
"got multiple values for keyword argument" at startup and had no way to set
proxy-side router general settings at all (e.g. ``model_name_resolution``).
"""
if configured is None:
return RouterGeneralSettings(async_only_mode=True)
settings: Final = (
RouterGeneralSettings(**configured) if isinstance(configured, Mapping) else configured.model_copy()
)
settings.async_only_mode = True
return settings
class ProxyConfig:
"""
Abstraction class on top of config loading/updating logic. Gives us one place to control all config updating logic.
@ -6144,13 +6168,18 @@ class ProxyConfig:
verbose_proxy_logger.warning(
"Key '%s' is not a valid argument for Router.__init__(). Ignoring this key.", k
)
# `async_only_mode` is a proxy-runtime requirement (only async clients are
# initialised here), so it is forced on. Everything else the operator set
# under `router_settings.router_general_settings` -- e.g.
# `model_name_resolution: strict` -- is preserved; passing the key in
# config used to collide with this keyword and raise TypeError at startup.
router_params["router_general_settings"] = _proxy_router_general_settings(
router_params.get("router_general_settings")
)
router = litellm.Router(
**router_params,
assistants_config=assistants_config,
search_tools=search_tools,
router_general_settings=RouterGeneralSettings(
async_only_mode=True # only init async clients
),
ignore_invalid_deployments=True, # don't raise an error if a deployment is invalid
fallback_access_check=router_fallback_access_check,
auto_router_capability_limit=_license_check.auto_router_capability_limit,
@ -6608,9 +6637,10 @@ class ProxyConfig:
verbose_proxy_logger.debug("_model_list: %s", _model_list)
llm_router = litellm.Router(
model_list=_model_list,
router_general_settings=RouterGeneralSettings(
async_only_mode=True # only init async clients
),
# DB-sourced model list: no config router_settings in scope
# here, so this is the proxy default (async_only_mode on,
# everything else at its RouterGeneralSettings default).
router_general_settings=_proxy_router_general_settings(None),
search_tools=search_tools,
ignore_invalid_deployments=True,
fallback_access_check=router_fallback_access_check,

View file

@ -236,6 +236,73 @@ def get_team_id_from_data(data: dict) -> str | None:
return None
def _requested_model_metadata(
requested_model: str,
) -> dict: # mutable-ok: request metadata is consumed downstream as a plain dict
"""A fresh metadata dict recording the client's original model spelling.
Downstream logging consumes request metadata as a plain mutable dict, so
this is deliberately not frozen.
"""
return {"requested_model": requested_model} # mutable-ok: request metadata is consumed downstream as a plain dict
async def _canonical_target_is_allowed(
canonical_target: str,
llm_router: LitellmRouter,
user_api_key_dict: UserAPIKeyAuth | None,
) -> bool:
"""Whether the caller may call a canonically-resolved target model group.
AND-on-target: the caller must be permitted to call the *resolved* group.
The requested spelling having passed the earlier auth check is not enough --
without this, a key whose allowlist holds only a stale, unserved name would
ride the rewrite onto a deployment it was never granted. (Auth ran on the
requested string before routing, when the target group was not yet known.)
Uses ``can_key_call_resolved_model`` -- the same helper every other
post-resolution auth site uses (model_group_alias rewrites, realtime
endpoints, auto-router) -- so the key, team, team-member, and project
allowlists are all re-checked against the target. Checking only the key
(``can_key_call_model``) would leave a team whose allowlist holds a stale
unserved name able to ride the rewrite onto a deployment it was never
granted, since an unrestricted *key* on that team passes the key-level
check on its own.
A denial returns False rather than raising, so the request falls through to
the same 400 an unresolvable model gets today and the response reveals
nothing about the target's existence.
Absent key context fails CLOSED. Most ``route_request`` callers (image
generation, rerank, moderation, speech, transcription, realtime, Responses
WebSocket) are authenticated but do not currently forward
``user_api_key_dict``, so treating "no key context" as "allowed" would run
the rewrite with no target authorization at all on exactly those paths.
Declining instead costs those endpoints only the convenience rewrite --
they behave as they do today, resolution simply never engages -- while
keeping the AND-on-target guarantee unconditional. Threading the key
through those call sites is the follow-up that re-enables resolution for
them; until then this must not be the hole through which the check is
skipped.
"""
if user_api_key_dict is None:
return False
from litellm.proxy.auth.auth_checks import (
can_key_call_resolved_model, # pyright: ignore[reportUnknownVariableType] - auth_checks is partially typed
)
try:
await can_key_call_resolved_model(
model=canonical_target,
llm_model_list=llm_router.model_list,
valid_token=user_api_key_dict,
llm_router=llm_router,
)
except Exception: # noqa: BLE001 # any auth failure declines the rewrite; never widens access
return False
return True
_shared_session_lock: asyncio.Lock | None = None
@ -711,6 +778,50 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr
elif user_model is not None or route_type == "allm_passthrough_route":
return getattr(litellm, f"{route_type}")(**data)
# Last resort before failing: the requested name may be a different spelling
# of a model this router already serves (e.g. a harness sending the dated
# 'claude-haiku-4-5-20251001' at a gateway that deploys it as
# 'anthropic/claude-haiku-4-5'). Resolution is same-provider and
# identity-attested only, and runs here -- after every configured route,
# including wildcards and default_deployment, has declined -- so it can only
# turn a hard failure into a success, never re-point working traffic.
requested_model: Final[object] = data.get("model") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] - data is an untyped request dict
if llm_router is not None and isinstance(requested_model, str):
canonical_target: Final[object] = llm_router.resolve_canonical_model_name(
model=requested_model,
request_team_id=team_id,
)
# Require a real model-group name: a router stub that returns a
# non-string (e.g. a test double) must not be read as "resolved".
if isinstance(canonical_target, str) and canonical_target:
# AND-on-target: the caller must be allowed to call the *resolved*
# group. The requested spelling passing the earlier auth check is
# not enough -- without this, a key whose allowlist holds only a
# stale unserved name would ride the rewrite onto a deployment it
# was never granted. (Auth ran on the requested string before
# routing; the target group was not visible to it then.) On denial
# the rewrite is simply declined -- the request falls through to
# the same 400 it gets today, revealing nothing about the target.
target_allowed: Final = await _canonical_target_is_allowed(
canonical_target=canonical_target,
llm_router=llm_router,
user_api_key_dict=user_api_key_dict,
)
if target_allowed:
# Preserve the client's spelling for spend logs / debugging --
# after the rewrite it is otherwise invisible downstream.
metadata_field: Final = "litellm_metadata" if "litellm_metadata" in data else "metadata"
existing_metadata: Final[object] = data.get(metadata_field) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] - data is an untyped request dict
# `data` is rewritten in place here because every other route in
# this function does the same; see the rebind-ok notes below.
if isinstance(existing_metadata, dict):
existing_metadata.setdefault("requested_model", requested_model) # pyright: ignore[reportUnknownMemberType] - metadata dict is untyped
else:
stamp: Final = _requested_model_metadata(requested_model)
data[metadata_field] = stamp # rebind-ok: in-place rewrite, as everywhere in route_request
data["model"] = canonical_target # rebind-ok: in-place data rewrite, as everywhere in route_request
return getattr(llm_router, f"{route_type}")(**data)
# if no route found then it's a bad request
route_name: Final = ROUTE_ENDPOINT_MAPPING.get(route_type, route_type)
raise ProxyModelNotFoundError(

View file

@ -144,6 +144,12 @@ from litellm.router_utils.batch_utils import (
replace_model_in_jsonl,
should_replace_model_in_jsonl,
)
from litellm.router_utils.canonical_model_resolution import (
build_canonical_index,
)
from litellm.router_utils.canonical_model_resolution import (
lookup as canonical_lookup,
)
from litellm.router_utils.client_initalization_utils import InitalizeCachedClient
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
@ -280,6 +286,7 @@ from litellm.utils import (
Rules,
function_setup,
get_llm_provider,
get_model_cost_mutation_generation,
get_non_default_completion_params,
get_secret,
get_utc_datetime,
@ -955,6 +962,20 @@ class Router:
self.get_deployment_model_info
)
self._routing_group_rows: tuple[DeploymentTypedDict, ...] | None = None
# Lazily-built (provider, canonical_name) -> model_group index for
# ``resolve_canonical_model_name``. Invalidated alongside the model
# group info cache and on cost-map mutation (generation counter).
self._canonical_model_index: Mapping[tuple[str, str], str | None] | None = None
self._canonical_model_index_cost_generation: int = -1
# (requested, target) pairs already announced at INFO, so a hot path
# logs once per distinct pair rather than once per request. Keyed on
# the pair, not just the target, so each new requested spelling that
# resolves to an already-seen target is still observable -- this is
# the signal used to size demand for possible follow-up resolution
# rules, so a second spelling silently sharing the first's log line
# would undercount it.
# mutable-ok: grow-only log-dedup ledger, intentionally accumulates across requests
self._canonical_resolution_logged: set[tuple[str, str]] = set() # mutable-ok: grow-only log-dedup ledger
self._init_routing_groups(None)
self._provider_unresolved_deployments: tuple[Callable[[], Deployment | None], ...] = ()
@ -11864,6 +11885,7 @@ class Router:
self.cached_deployment_model_info.cache_clear()
self._zero_cost_cache.clear()
self._routing_group_rows = None
self._canonical_model_index = None
def _invalidate_access_groups_cache(self) -> None:
"""Invalidate the cached access groups.
@ -12383,6 +12405,62 @@ class Router:
"""
return resolve_model_group_alias(self.model_group_alias, model)
def _get_canonical_model_index(self) -> Mapping[tuple[str, str], str | None]:
"""The ``(provider, canonical_name) -> model group`` index, built on demand.
Rebuilt when the model list changes (the index is dropped by
``_invalidate_model_group_info_cache``) or when ``litellm.model_cost``
mutates, since identity attestation reads it.
"""
cost_generation: Final = get_model_cost_mutation_generation()
if self._canonical_model_index is None or self._canonical_model_index_cost_generation != cost_generation:
try:
self._canonical_model_index = build_canonical_index(self.model_list)
except Exception as exc: # noqa: BLE001 # index construction must never brick a router; degrade to 'strict'
verbose_router_logger.error("canonical-resolution: index build failed, disabling feature: %s", exc)
self._canonical_model_index = MappingProxyType({})
self._canonical_model_index_cost_generation = cost_generation
return self._canonical_model_index
def resolve_canonical_model_name(self, model: str, request_team_id: str | None = None) -> str | None:
"""The model group that provably serves ``model`` under another spelling.
Returns None unless every one of these holds:
- ``router_general_settings.model_name_resolution`` is ``"canonical"``
- ``model`` is not already served (``is_recognized_model``)
- no team route, pattern/wildcard route, or ``default_deployment`` would
take the request -- those are operator-configured catch-alls and must
keep winning
- a single model group matches ``model``'s canonical identity *on the
same provider*
Callers must treat a non-None result as authorization-relevant: the
target model group is what the key/team must be permitted to call.
"""
if self.router_general_settings.model_name_resolution != "canonical":
return None
if not model or self.is_recognized_model(model):
return None
# Operator-configured catch-alls outrank inference.
if self.default_deployment is not None or len(self.pattern_router.patterns) > 0:
return None
if request_team_id is not None and request_team_id in self.team_pattern_routers:
return None
if model in self.deployment_names:
return None
target: Final = canonical_lookup(self._get_canonical_model_index(), model)
if target is None:
return None
# A target whose deployments have all been removed is not a live route.
if not self.model_name_to_deployment_indices.get(target):
return None
log_key: Final = (model, target)
if log_key not in self._canonical_resolution_logged:
self._canonical_resolution_logged.add(log_key)
verbose_router_logger.info("canonical-resolution: '%s' -> '%s'", model, target)
return target
def _get_deployment_by_litellm_model(self, model: str) -> list:
"""
Get the deployment by litellm model.

View file

@ -0,0 +1,275 @@
# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportAny=false
# litellm.model_cost is loaded dynamically from a JSON price map (see
# litellm.litellm_core_utils.get_model_cost_map) and is untyped at that
# boundary, same as every other reader of it in litellm/utils.py. Every value
# pulled from it here is re-validated with isinstance before use.
"""Same-provider canonical model-name resolution.
Harnesses hardcode concrete model IDs. A client that asks for
``claude-haiku-4-5-20251001`` against a gateway that serves the very same model
under the deployment name ``anthropic/claude-haiku-4-5`` gets a 403/400 today,
even though the gateway *is* serving what was asked for. Only the spelling
differs.
This module builds a ``canonical name -> model group`` index from the router's
own deployments so that spelling difference can be bridged. Two hard rules keep
the bridge from becoming a guess:
1. **Identity, not similarity.** Two names are equivalent only when the model
cost map attests they are the same model: both present, same
``litellm_provider``, same ``mode``, and identical pricing. Family/version
hops (``claude-sonnet-4-5`` -> ``claude-sonnet-5``) are never equivalences,
because that changes which model answers.
2. **Never across providers.** The requested name's inferred provider must equal
the target deployment's provider. The same weights on Bedrock, Vertex, and
the first-party API differ in credentials, data residency, quota pool, and
billing; picking between them is an operator decision, not something a
gateway should infer. Cross-provider mapping stays available through an
explicit ``model_group_alias``.
Resolution is a last resort: callers consult it only after every existing route
(exact name, deployment id, ``model_group_alias``, routing group, team route,
wildcard/pattern route, ``default_deployment``) has declined, so a request that
succeeds today can never be re-pointed by this module.
See also ``Router.resolve_canonical_model_name``.
"""
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Final
import litellm
from litellm._logging import verbose_router_logger
# Cost-map fields that must match exactly for two names to be called the same
# model. Pricing equality is a tripwire against false identity, not the
# definition of it -- the provider/mode checks below carry that weight.
_IDENTITY_ATTESTING_FIELDS: Final[tuple[str, ...]] = (
"litellm_provider",
"mode",
"input_cost_per_token",
"output_cost_per_token",
"max_input_tokens",
"max_output_tokens",
)
# Sentinel stored in the index when two distinct model groups claim the same
# canonical identity. Resolution then declines rather than silently picking a
# billing path the operator never sanctioned.
_AMBIGUOUS: Final = None
# Read-only stand-in for a missing sub-mapping on a deployment row.
_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
def _cost_map_entry(model: str) -> Mapping[str, object] | None:
"""The cost-map entry for ``model``, or None when absent."""
entry: Final = litellm.model_cost.get(model)
return entry if isinstance(entry, Mapping) else None
def _same_model_per_cost_map(name_a: str, name_b: str) -> bool:
"""Whether the cost map attests ``name_a`` and ``name_b`` are one model.
Both names must be present with identical provider, mode, and pricing. A
customer fine-tune or an unknown vanity name is absent from the map and so
can never be equated with anything -- which is the point.
"""
entry_a: Final = _cost_map_entry(name_a)
entry_b: Final = _cost_map_entry(name_b)
if entry_a is None or entry_b is None:
return False
return all(entry_a.get(field) == entry_b.get(field) for field in _IDENTITY_ATTESTING_FIELDS)
def _infer_provider(model: str) -> str | None:
"""The provider LiteLLM would route ``model`` to, or None if undecidable.
Wraps ``get_llm_provider``, which raises ``BadRequestError`` for names it
cannot place. An undecidable name simply never participates in resolution.
"""
try:
_, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
except Exception: # noqa: BLE001 # an unplaceable name simply never resolves; never fail the request
return None
return custom_llm_provider or None
def canonicalize(model: str, custom_llm_provider: str | None = None) -> tuple[str, str] | None:
"""Reduce ``model`` to a ``(provider, canonical_name)`` identity.
The canonical name is the model string with any LiteLLM provider-route
prefix removed (``anthropic/claude-opus-5`` -> ``claude-opus-5``), which is
LiteLLM's own routing syntax rather than part of the model's identity. The
provider is carried alongside so equality checks are always provider-scoped.
``custom_llm_provider`` -- the deployment's explicit provider override --
wins over whatever the model string implies. A deployment can carry a
first-party-looking id (``claude-haiku-4-5``) while actually being served
through Bedrock, Vertex, or OpenRouter; inferring the provider from the
string alone would index it as Anthropic and let an Anthropic-form request
be rewritten onto that other provider's credentials, quota, and bill --
exactly the cross-provider hop rule 2 in the module docstring forbids.
Returns None when the provider cannot be inferred.
"""
if not model:
return None
provider: Final = custom_llm_provider or _infer_provider(model)
if provider is None:
return None
# get_llm_provider returns the model with its routing prefix stripped, which
# is exactly the normalization wanted here.
try:
stripped, _, _, _ = litellm.get_llm_provider(model=model)
except Exception: # noqa: BLE001 # an unplaceable name simply never resolves; never fail the request
return None
return (provider, stripped or model)
def _undated_variants(model: str) -> tuple[str, ...]:
"""Plausible dated<->undated spellings of ``model``, unvalidated.
Purely syntactic candidate generation: every candidate is still gated by
``_same_model_per_cost_map`` before it is treated as an equivalence, so a
coincidental date-like suffix on an unrelated model cannot create a false
match (it will not be in the cost map, or will not match on pricing).
Only an 8-digit ``-YYYYMMDD`` suffix is considered. Deliberately narrower
than the cost-lookup heuristics in ``litellm.utils`` (which strip any
trailing ``-\\d+`` and would conflate ``gemini-1.5-pro-001`` with ``-002``):
a wrong match there mis-prices a log line, a wrong match here serves the
wrong model.
"""
parts: Final = model.rsplit("-", 1)
if len(parts) == 2 and len(parts[1]) == 8 and parts[1].isdigit():
return (parts[0],)
return ()
def build_canonical_index(
deployments: Sequence[Mapping[str, object]],
) -> Mapping[tuple[str, str], str | None]:
"""Map ``(provider, canonical_name) -> model group`` for ``deployments``.
A model group is indexed only when every one of its deployments agrees on
the same canonical identity; mixed groups are skipped. When two groups claim
one identity the entry is set to ``_AMBIGUOUS`` (None) so lookups decline.
Team-owned deployments (``model_info.team_id`` set) are never indexed. A
team boundary is an operator-drawn access/billing boundary exactly like a
provider boundary (see the module docstring's rule 2): auto-resolution must
not cross it. Concretely, without this exclusion a global (no-team) key
could request a team's deployment under an unclaimed spelling -- e.g. the
dated Anthropic ID -- and land on that team's credentials and quota, since
``is_recognized_model``/target-authorization checks pass for unrestricted
keys and don't themselves re-derive team ownership. A team-scoped model
remains reachable exactly as it is today: by its team_public_model_name,
through the existing team-route machinery, which this module never touches.
Never raises: a malformed deployment or cost-map entry degrades to a smaller
index, never to a router that fails to boot.
"""
# Both accumulate across the deployment scan, then the result is frozen into
# a MappingProxyType before it leaves this function.
index: Final[dict[tuple[str, str], str | None]] = {} # mutable-ok: local accumulator, frozen on return
group_identity: Final[dict[str, tuple[str, str] | None]] = {} # mutable-ok: local accumulator, never escapes
for deployment in deployments:
try:
model_info = deployment.get("model_info") or _EMPTY # rebind-ok: per-deployment loop variable
if isinstance(model_info, Mapping) and model_info.get("team_id"):
continue
# Deployments come from operator config and DB rows, so the declared
# str types are validated at runtime rather than trusted.
model_group = deployment.get("model_name") # rebind-ok: per-deployment loop variable
litellm_params = deployment.get("litellm_params") or _EMPTY # rebind-ok: per-deployment loop variable
underlying = (
litellm_params.get("model") if isinstance(litellm_params, Mapping) else None
) # rebind-ok: per-deployment loop variable
if not isinstance(model_group, str) or not isinstance(underlying, str):
continue
# An explicit provider override decides the provider; see canonicalize().
provider_override = (
litellm_params.get("custom_llm_provider") if isinstance(litellm_params, Mapping) else None
) # rebind-ok: per-deployment loop variable
identity = canonicalize(
underlying,
custom_llm_provider=provider_override if isinstance(provider_override, str) else None,
)
if model_group in group_identity and group_identity[model_group] != identity:
# Deployments in this group disagree about what they serve; the
# group cannot stand for a single canonical identity.
group_identity[model_group] = None
continue
group_identity.setdefault(model_group, identity)
except Exception as exc: # noqa: BLE001 # pragma: no cover - a malformed deployment must not abort the index build
verbose_router_logger.debug("canonical-resolution: skipping deployment: %s", exc)
continue
for model_group, identity in group_identity.items():
if identity is None:
continue
provider, canonical_name = identity
# Index the canonical spelling plus any dated<->undated sibling the cost
# map attests is the same model: the undated form of a dated canonical
# name, and every dated cost-map entry whose undated form is this one.
undated_siblings = tuple( # rebind-ok: per-group loop variable
candidate
for candidate in _undated_variants(canonical_name)
if _same_model_per_cost_map(canonical_name, candidate)
)
dated_siblings = tuple( # rebind-ok: per-group loop variable
dated
for dated, entry in litellm.model_cost.items()
if isinstance(dated, str)
and isinstance(entry, Mapping)
and dated != canonical_name
and dated not in undated_siblings
and _undated_variants(dated) == (canonical_name,)
and _same_model_per_cost_map(canonical_name, dated)
)
spellings = (canonical_name, *undated_siblings, *dated_siblings) # rebind-ok: per-group loop variable
for spelling in spellings:
key = (provider, spelling)
if key not in index:
index[key] = model_group
continue
existing = index[key]
if existing != model_group:
# Two groups, same identity: decline rather than choose.
index[key] = _AMBIGUOUS
verbose_router_logger.info(
"canonical-resolution: '%s' is served by more than one model group "
"(%s, %s); auto-resolution disabled for it. Add an explicit "
"model_group_alias to pick one.",
spelling,
existing,
model_group,
)
return MappingProxyType(index)
def lookup(
index: Mapping[tuple[str, str], str | None],
requested_model: str,
) -> str | None:
"""The model group serving ``requested_model``, or None.
None covers every decline: unknown provider, no identity match, or an
ambiguous identity. Pure dict lookup after canonicalization -- no I/O.
"""
identity: Final = canonicalize(requested_model)
if identity is None:
return None
target: Final = index.get(identity)
if target is None:
return None
# A request already naming its serving group is not a rewrite.
if target == requested_model:
return None
return target

View file

@ -852,6 +852,11 @@ class RouterGeneralSettings(BaseModel):
pass_through_all_models: bool = Field(
default=False
) # if passed a model not llm_router model list, pass through the request to litellm.acompletion/embedding
model_name_resolution: Literal["canonical", "strict"] = Field(
default="canonical"
) # "canonical": an unknown requested model name that provably names an already-served model
# (same provider, cost-map-attested identity, e.g. dated vs undated spelling) is routed to the
# serving model group instead of failing. "strict": unknown names fail exactly as before.
class RouterRateLimitErrorBasic(ValueError):

View file

@ -0,0 +1,573 @@
"""Tests for same-provider canonical model-name resolution.
Invariants under test (see litellm/router_utils/canonical_model_resolution.py):
- I1: resolution never re-points a request that any existing route accepts
(exact name, alias, wildcard/pattern, default_deployment, team routes).
- I2: identity only -- same provider, cost-map-attested; never a family/version
hop, never cross-provider (Bedrock/Vertex/first-party are distinct).
- I3: auth is AND-on-target -- the resolved group must itself be allowed.
- Ambiguity fails closed.
"""
import pytest
from litellm import Router
from litellm.proxy.auth.auth_checks import _can_object_call_model
from litellm.proxy._types import ProxyException
from litellm.router_utils.canonical_model_resolution import (
build_canonical_index,
canonicalize,
lookup,
)
ANTHROPIC_GROUP = "anthropic/claude-haiku-4-5"
DATED = "claude-haiku-4-5-20251001"
UNDATED = "claude-haiku-4-5"
BEDROCK_FORM = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
@pytest.fixture
def anthropic_router() -> Router:
return Router(
model_list=[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "sk-test"},
}
]
)
class TestCanonicalize:
def test_strips_provider_route_prefix(self):
assert canonicalize("anthropic/claude-haiku-4-5") == ("anthropic", UNDATED)
def test_bare_dated_name_infers_provider(self):
assert canonicalize(DATED) == ("anthropic", DATED)
def test_bedrock_form_is_bedrock_not_anthropic(self):
identity = canonicalize(BEDROCK_FORM)
assert identity is not None
assert identity[0] == "bedrock"
def test_unknown_model_returns_none(self):
assert canonicalize("totally-made-up-model-xyz") is None
def test_empty_returns_none(self):
assert canonicalize("") is None
class TestBuildIndexAndLookup:
def test_dated_and_undated_spellings_resolve(self):
index = build_canonical_index(
[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
}
]
)
assert lookup(index, DATED) == ANTHROPIC_GROUP
assert lookup(index, UNDATED) == ANTHROPIC_GROUP
def test_no_family_version_hop(self):
"""claude-sonnet-5 must never resolve to a haiku group (different model)."""
index = build_canonical_index(
[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
}
]
)
assert lookup(index, "claude-sonnet-5") is None
def test_cross_provider_never_matches(self):
"""An Anthropic-form request must not land on a Bedrock-only deployment,
and a Bedrock-form request must not land on a first-party deployment."""
bedrock_only = build_canonical_index(
[
{
"model_name": "claude-haiku-bedrock",
"litellm_params": {"model": f"bedrock/{BEDROCK_FORM}"},
}
]
)
assert lookup(bedrock_only, DATED) is None
assert lookup(bedrock_only, UNDATED) is None
# Same-provider spelling still works for the Bedrock group.
assert lookup(bedrock_only, BEDROCK_FORM) == "claude-haiku-bedrock"
anthropic_only = build_canonical_index(
[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
}
]
)
assert lookup(anthropic_only, BEDROCK_FORM) is None
def test_vertex_deployment_does_not_capture_anthropic_request(self):
index = build_canonical_index(
[
{
"model_name": "claude-haiku-vertex",
"litellm_params": {"model": "vertex_ai/claude-haiku-4-5"},
}
]
)
assert lookup(index, DATED) is None
def test_ambiguous_identity_fails_closed(self):
"""Two same-provider groups serving one model: resolution must decline."""
index = build_canonical_index(
[
{
"model_name": "haiku-prod",
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
},
{
"model_name": "haiku-experiments",
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
},
]
)
assert lookup(index, DATED) is None
assert lookup(index, UNDATED) is None
def test_ambiguity_guard_not_defeated_by_group_named_like_sentinel(self):
"""Regression: the ambiguity check used to compare against the string
'__absent__' as an "is this key missing" sentinel. A model group
literally named '__absent__' collided with that sentinel and could
silently overwrite a same-identity entry instead of triggering the
ambiguity decline."""
index = build_canonical_index(
[
{
"model_name": "__absent__",
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
},
{
"model_name": "haiku-2",
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
},
]
)
assert lookup(index, DATED) is None
assert lookup(index, UNDATED) is None
def test_custom_llm_provider_override_decides_provider(self):
"""Regression (I2): a deployment's explicit custom_llm_provider wins over
whatever the model string implies. A first-party-looking id served via
OpenRouter/Bedrock/Vertex must not be indexed as Anthropic -- otherwise
an Anthropic-form request rides the rewrite onto that other provider's
credentials, quota, and bill, which is exactly the cross-provider hop
rule 2 forbids."""
index = build_canonical_index(
[
{
"model_name": "haiku-via-openrouter",
"litellm_params": {
"model": "claude-haiku-4-5",
"custom_llm_provider": "openrouter",
},
},
]
)
# Indexed under the real provider, not the one the string implies.
assert ("openrouter", UNDATED) in index
assert ("anthropic", UNDATED) not in index
# An Anthropic-form request must not reach the OpenRouter deployment.
assert lookup(index, DATED) is None
assert lookup(index, UNDATED) is None
def test_custom_llm_provider_override_still_resolves_within_provider(self):
"""The override narrows the provider, it does not disable resolution:
a request that infers to the same overridden provider still resolves."""
index = build_canonical_index(
[
{
"model_name": "haiku-via-bedrock",
"litellm_params": {
"model": "claude-haiku-4-5",
"custom_llm_provider": "bedrock",
},
},
]
)
assert index[("bedrock", UNDATED)] == "haiku-via-bedrock"
def test_team_owned_deployment_never_indexed(self):
"""Regression: a team-owned deployment (model_info.team_id set) must
never enter the global canonical index. Without this, a no-team key
could request an unclaimed spelling of a model whose only server is a
team's private deployment and land on that team's credentials/quota --
a team boundary is an access/billing boundary exactly like the
cross-provider boundary and must not be crossed by inference."""
index = build_canonical_index(
[
{
"model_name": "internal-team-model-xyz",
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
"model_info": {
"team_id": "team-A",
"team_public_model_name": "claude-haiku-4-5",
},
},
]
)
assert lookup(index, DATED) is None
assert lookup(index, UNDATED) is None
def test_team_owned_deployment_does_not_block_global_sibling(self):
"""A team-owned deployment coexisting with a global deployment of the
same identity must not suppress resolution to the global one."""
index = build_canonical_index(
[
{
"model_name": "internal-team-model-xyz",
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
"model_info": {
"team_id": "team-A",
"team_public_model_name": "claude-haiku-4-5",
},
},
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
},
]
)
assert lookup(index, DATED) == ANTHROPIC_GROUP
def test_request_for_own_group_name_is_not_a_rewrite(self):
index = build_canonical_index(
[
{
"model_name": UNDATED, # group named exactly the canonical name
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
}
]
)
assert lookup(index, UNDATED) is None
def test_malformed_deployments_are_skipped_not_fatal(self):
index = build_canonical_index(
[
{"model_name": None, "litellm_params": {"model": "anthropic/claude-haiku-4-5"}},
{"model_name": "ok-group", "litellm_params": None},
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5"},
},
]
)
assert lookup(index, DATED) == ANTHROPIC_GROUP
class TestRouterResolveCanonicalModelName:
def test_claude_code_case(self, anthropic_router: Router):
assert anthropic_router.resolve_canonical_model_name(DATED) == ANTHROPIC_GROUP
def test_recognized_model_short_circuits(self, anthropic_router: Router):
"""I1: a name the router already serves is never rewritten."""
assert anthropic_router.resolve_canonical_model_name(ANTHROPIC_GROUP) is None
def test_no_team_caller_never_resolves_onto_team_owned_deployment(self):
"""Regression: end-to-end version of the team-leak fix. A no-team
caller asking for an unclaimed spelling must not resolve onto a
deployment that is the sole server of that identity but is owned by a
team -- that would leak the team's credentials/quota to a global key."""
router = Router(
model_list=[
{
"model_name": "internal-team-model-xyz",
"litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "team-secret"},
"model_info": {"team_id": "team-A", "team_public_model_name": "claude-haiku-4-5"},
},
]
)
assert router.resolve_canonical_model_name(DATED, request_team_id=None) is None
# Even the requesting team's own id must not resolve through this path --
# team-scoped models are reached via the existing team-route machinery,
# not via canonical inference.
assert router.resolve_canonical_model_name(DATED, request_team_id="team-A") is None
def test_log_dedup_keyed_on_pair_not_target_alone(self, anthropic_router: Router, caplog: pytest.LogCaptureFixture):
"""Regression: a second distinct requested spelling resolving to an
already-logged target must still get its own log line -- the
(requested, target) cardinality is the signal used to size demand for
follow-up resolution rules, so deduping on target alone would
undercount it."""
import logging
with caplog.at_level(logging.INFO, logger="LiteLLM Router"):
assert anthropic_router.resolve_canonical_model_name(DATED) == ANTHROPIC_GROUP
assert anthropic_router.resolve_canonical_model_name(UNDATED) == ANTHROPIC_GROUP
# Re-requesting the same spelling must not double-log.
assert anthropic_router.resolve_canonical_model_name(DATED) == ANTHROPIC_GROUP
messages = [r.message for r in caplog.records if "canonical-resolution" in r.message]
assert any(DATED in m for m in messages)
assert any(UNDATED in m for m in messages)
assert sum(1 for m in messages if DATED in m) == 1
def test_strict_mode_disables_resolution(self):
from litellm.types.router import RouterGeneralSettings
router = Router(
model_list=[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "sk-test"},
}
],
router_general_settings=RouterGeneralSettings(model_name_resolution="strict"),
)
assert router.resolve_canonical_model_name(DATED) is None
def test_wildcard_route_outranks_resolution(self):
"""I1: an operator catch-all keeps winning; resolution declines entirely."""
router = Router(
model_list=[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "sk-test"},
},
{
"model_name": "openai/*",
"litellm_params": {"model": "openai/*", "api_key": "sk-test"},
},
]
)
assert router.resolve_canonical_model_name(DATED) is None
def test_default_deployment_outranks_resolution(self):
router = Router(
model_list=[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "sk-test"},
},
{
"model_name": "*",
"litellm_params": {"model": "*", "api_key": "sk-test"},
},
]
)
assert router.resolve_canonical_model_name(DATED) is None
def test_manual_alias_outranks_resolution(self):
"""A model_group_alias for the same spelling wins (is_recognized_model)."""
router = Router(
model_list=[
{
"model_name": ANTHROPIC_GROUP,
"litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "sk-test"},
},
{
"model_name": "other-group",
"litellm_params": {"model": "anthropic/claude-opus-5", "api_key": "sk-test"},
},
],
model_group_alias={DATED: "other-group"},
)
assert router.resolve_canonical_model_name(DATED) is None
def test_get_canonical_model_index_builds_and_caches(self, anthropic_router: Router):
"""The index is built on demand, memoized, and keyed by identity."""
index = anthropic_router._get_canonical_model_index()
assert index[("anthropic", UNDATED)] == ANTHROPIC_GROUP
# Second call returns the same memoized object (no rebuild).
assert anthropic_router._get_canonical_model_index() is index
def test_get_canonical_model_index_survives_build_failure(
self, anthropic_router: Router, monkeypatch: pytest.MonkeyPatch
):
"""A failing index build degrades to 'strict', never raises."""
import litellm.router as router_module
def boom(*_args: object, **_kwargs: object) -> dict:
raise RuntimeError("cost map exploded")
monkeypatch.setattr(router_module, "build_canonical_index", boom)
anthropic_router._canonical_model_index = None
assert anthropic_router._get_canonical_model_index() == {}
assert anthropic_router.resolve_canonical_model_name(DATED) is None
def test_index_rebuilds_after_model_list_change(self, anthropic_router: Router):
assert anthropic_router.resolve_canonical_model_name(DATED) == ANTHROPIC_GROUP
anthropic_router.set_model_list(
[
{
"model_name": "gpt-group",
"litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-test"},
}
]
)
assert anthropic_router.resolve_canonical_model_name(DATED) is None
class TestAuthAndOnTarget:
"""I3: an auto-resolved request is allowed iff the TARGET is allowed."""
def test_target_allowed_grants_requested_spelling(self, anthropic_router: Router):
assert (
_can_object_call_model(
model=DATED,
llm_router=anthropic_router,
models=[ANTHROPIC_GROUP],
object_type="key",
)
is True
)
def test_target_not_allowed_denies(self, anthropic_router: Router):
with pytest.raises(ProxyException):
_can_object_call_model(
model=DATED,
llm_router=anthropic_router,
models=["some-other-model"],
object_type="key",
)
def test_unrelated_model_still_denied(self, anthropic_router: Router):
with pytest.raises(ProxyException):
_can_object_call_model(
model="claude-sonnet-5",
llm_router=anthropic_router,
models=[ANTHROPIC_GROUP],
object_type="key",
)
def test_unrestricted_key_unchanged(self, anthropic_router: Router):
# Empty allowlist = unrestricted; behavior must not change.
assert (
_can_object_call_model(
model=DATED,
llm_router=anthropic_router,
models=[],
object_type="key",
)
is True
)
class TestCanonicalTargetReAuth:
"""The rewrite's re-auth must use the *resolved-model* helper.
Regression: the hook originally called ``can_key_call_model``, which checks
only the key's own allowlist. Team, team-member, and project allowlists were
therefore never re-checked against the resolved target, so an unrestricted
key on a team whose allowlist held only a stale unserved name could ride the
rewrite onto a deployment that team was never granted.
``can_key_call_resolved_model`` is the helper every other post-resolution
auth site uses (model_group_alias rewrites, realtime endpoints, auto-router).
"""
@pytest.mark.asyncio
async def test_uses_resolved_model_helper_so_team_scope_is_rechecked(self, anthropic_router: Router):
from unittest.mock import AsyncMock, patch
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.route_llm_request import _canonical_target_is_allowed
with patch(
"litellm.proxy.auth.auth_checks.can_key_call_resolved_model",
new=AsyncMock(return_value=None),
) as resolved_check:
allowed = await _canonical_target_is_allowed(
canonical_target=ANTHROPIC_GROUP,
llm_router=anthropic_router,
user_api_key_dict=UserAPIKeyAuth(token="t", team_id="team-A"),
)
assert allowed is True
resolved_check.assert_awaited_once()
assert resolved_check.await_args.kwargs["model"] == ANTHROPIC_GROUP
@pytest.mark.asyncio
async def test_denial_declines_rewrite_rather_than_raising(self, anthropic_router: Router):
"""A denial must return False (request falls through to the usual 400),
never propagate an exception that would leak the target's existence."""
from unittest.mock import AsyncMock, patch
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.route_llm_request import _canonical_target_is_allowed
with patch(
"litellm.proxy.auth.auth_checks.can_key_call_resolved_model",
new=AsyncMock(side_effect=ProxyException(message="denied", type="auth_error", param=None, code=401)),
):
allowed = await _canonical_target_is_allowed(
canonical_target=ANTHROPIC_GROUP,
llm_router=anthropic_router,
user_api_key_dict=UserAPIKeyAuth(token="t", team_id="team-A"),
)
assert allowed is False
@pytest.mark.asyncio
async def test_absent_auth_context_fails_closed(self, anthropic_router: Router):
"""Regression: absent key context must DECLINE the rewrite, not allow it.
Most route_request callers (image generation, rerank, moderation,
speech, transcription, realtime, Responses WebSocket) are authenticated
but don't currently forward user_api_key_dict. Returning True here
would run the rewrite with no target authorization at all on exactly
those paths -- a key whose allowlist holds only the stale requested
spelling could reach a target it was never granted. Declining costs
those endpoints only the convenience rewrite; the AND-on-target
guarantee stays unconditional."""
from litellm.proxy.route_llm_request import _canonical_target_is_allowed
assert (
await _canonical_target_is_allowed(
canonical_target=ANTHROPIC_GROUP,
llm_router=anthropic_router,
user_api_key_dict=None,
)
is False
)
class TestProxyRouterGeneralSettings:
"""Operators must be able to set model_name_resolution on the proxy.
Regression: the proxy passed a hardcoded RouterGeneralSettings(async_only_mode=True)
as an explicit keyword alongside **router_params, so setting
router_general_settings in config raised "got multiple values for keyword
argument" at startup -- leaving no way to select 'strict'.
"""
def test_config_settings_preserved_and_async_only_forced(self):
from litellm.proxy.proxy_server import _proxy_router_general_settings
settings = _proxy_router_general_settings({"model_name_resolution": "strict"})
assert settings.model_name_resolution == "strict"
assert settings.async_only_mode is True
def test_async_only_mode_cannot_be_disabled_by_config(self):
from litellm.proxy.proxy_server import _proxy_router_general_settings
settings = _proxy_router_general_settings({"async_only_mode": False, "model_name_resolution": "strict"})
assert settings.async_only_mode is True
assert settings.model_name_resolution == "strict"
def test_none_yields_proxy_default(self):
from litellm.proxy.proxy_server import _proxy_router_general_settings
settings = _proxy_router_general_settings(None)
assert settings.async_only_mode is True
assert settings.model_name_resolution == "canonical"
def test_model_instance_is_not_mutated(self):
from litellm.proxy.proxy_server import _proxy_router_general_settings
from litellm.types.router import RouterGeneralSettings
original = RouterGeneralSettings(async_only_mode=False, model_name_resolution="strict")
settings = _proxy_router_general_settings(original)
assert settings.async_only_mode is True
# The caller's object must not be rewritten at a distance.
assert original.async_only_mode is False