fix(proxy): keep advertised models inside the caller's listing authorization

Review of #43310 surfaced two ways a catalog entry escaped the filtering the
routed rows go through, plus two smaller issues.

Reserve every name the router knows, not just the ids left in this caller's
listing. The collision check compared against the post-filter listing, so a
deployment removed by team scoping, a pause or the health filter was treated as
a free id: an entry naming it put it back, under whatever owner the config
declared. A key restricted to one model could be shown two deployments it had
no access to. Access group names are reserved too, so an entry cannot shadow a
group.

Run catalog ids through the same listing callbacks as routed rows, so a
callback that hides one is honored instead of bypassed.

Mark every catalog row `catalog_only`, since /models feeds the team and key
model pickers and an unroutable id was indistinguishable from a routable one
there.

Deduplicate in one pass rather than rescanning earlier entries, and drop two
comments that restated the code beside them.

litellm/proxy/common_utils/advertised_models.py

@@ -11,9 +11,14 @@ Discovery only, in both directions. Nothing here registers a route, so a request
 naming a catalog id still fails as an unknown model, and `GET /v1/models/{id}`
 still answers 404: the catalog says what exists, not what this proxy serves.
 
-A catalog entry never displaces a routed one. An id already in the listing wins,
-so a typo here cannot mask a working model or impersonate it, and a repeated id
-is listed once.
+A catalog entry never displaces a routed one, and never resurrects one. The
+reserved set is every name the router knows plus every id already listed, not
+just what this caller can see, so an entry cannot re-expose a model that team
+scoping, a pause or a health filter had hidden from them, nor impersonate it
+under a different owner. A repeated id is listed once.
+
+Every row is marked `catalog_only`, so a client picking models out of the
+listing can tell an advertised id from one the proxy will actually route.
 
 A row carries exactly what the operator declared. Nothing is inferred from the
 cost map, so an entry that happens to share a name with a known model does not
@@ -28,7 +33,8 @@ is logged and skipped, leaving the listing exactly as it would have been.
 from __future__ import annotations
 
 from collections.abc import Collection, Mapping
-from typing import Final
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final
 
 from pydantic import TypeAdapter, ValidationError
 
@@ -37,6 +43,9 @@ from litellm.constants import DEFAULT_MODEL_CREATED_AT_TIME
 from litellm.proxy._types import AdvertisedModel
 from litellm.types.proxy.model_listing import ModelInfoResponse
 
+if TYPE_CHECKING:
+    from litellm.router import Router
+
 ADVERTISED_MODELS_SETTING: Final = "advertised_models"
 
 _ADVERTISED_MODELS_ADAPTER: Final = TypeAdapter(tuple[AdvertisedModel, ...])
@@ -64,25 +73,39 @@ def _listing_row(entry: AdvertisedModel) -> ModelInfoResponse:
         "object": "model",
         "created": DEFAULT_MODEL_CREATED_AT_TIME,
         "owned_by": entry.owned_by,
+        "catalog_only": True,
     }
     return row
 
 
+def _reserved_ids(listed_ids: Collection[str], llm_router: Router | None) -> frozenset[str]:
+    """Ids a catalog entry may not claim.
+
+    Every name the router knows counts, not just the ids this caller can see, so
+    an entry cannot re-expose a deployment that team scoping, a pause or a health
+    filter had already removed from their listing.
+    """
+    if llm_router is None:
+        return frozenset(listed_ids)
+    return frozenset(listed_ids).union(llm_router.get_model_names(), llm_router.get_model_access_groups())
+
+
+def _first_per_id(entries: tuple[AdvertisedModel, ...]) -> tuple[AdvertisedModel, ...]:
+    """The first entry declared for each id, in configured order."""
+    latest_wins: Final = MappingProxyType({entry.id: entry for entry in reversed(entries)})
+    ordered_ids: Final = tuple(dict.fromkeys(entry.id for entry in entries))
+    return tuple(latest_wins[entry_id] for entry_id in ordered_ids)
+
+
 def advertised_model_rows(
     general_settings: Mapping[str, object],
     listed_ids: Collection[str],
+    llm_router: Router | None = None,
 ) -> tuple[ModelInfoResponse, ...]:
-    """Listing rows for the configured catalog entries, in configured order.
-
-    `listed_ids` are the ids the listing already carries; entries naming one of
-    them are dropped so a routed model is never displaced.
-    """
-    already_listed: Final = frozenset(listed_ids)
-    candidates: Final = tuple(
-        entry for entry in configured_advertised_models(general_settings) if entry.id not in already_listed
-    )
-    return tuple(
-        _listing_row(entry)
-        for index, entry in enumerate(candidates)
-        if all(earlier.id != entry.id for earlier in candidates[:index])
-    )
+    """Listing rows for the configured catalog entries, in configured order."""
+    configured: Final = configured_advertised_models(general_settings)
+    if not configured:
+        return ()
+    reserved: Final = _reserved_ids(listed_ids, llm_router)
+    candidates: Final = tuple(entry for entry in configured if entry.id not in reserved)
+    return tuple(_listing_row(entry) for entry in _first_per_id(candidates))

litellm/proxy/proxy_server.py

@@ -177,6 +177,7 @@ if TYPE_CHECKING:
 
     from litellm.integrations.opentelemetry import OpenTelemetry
     from litellm.proxy.health_check_utils.shared_health_check_manager import SharedHealthCheckManager
+    from litellm.types.proxy.model_listing import ModelInfoResponse
 
     Span = _Span | Any
 else:
@@ -11213,6 +11214,20 @@ async def _entries_kept_by_listing_callbacks(
     return tuple(entry for entry in entries if entry[0] not in hidden)
 
 
+async def _catalog_listing_rows(
+    settings: Mapping[str, object],
+    listed_ids: Sequence[str],
+    user_api_key_dict: UserAPIKeyAuth,
+) -> tuple["ModelInfoResponse", ...]:
+    """Catalog-only rows for this caller, filtered by the same listing callbacks
+    the routed rows went through, so an operator's callback can hide them too."""
+    rows: Final = advertised_model_rows(settings, listed_ids, llm_router)
+    if not rows:
+        return ()
+    hidden: Final = await _names_hidden_by_listing_callbacks(user_api_key_dict, tuple(row["id"] for row in rows))
+    return tuple(row for row in rows if row["id"] not in hidden)
+
+
 async def _deployment_hidden_by_listing_callbacks(deployment: Deployment, user_api_key_dict: UserAPIKeyAuth) -> bool:
     listed_name: Final = _translate_model_name_for_response(deployment.model_dump(exclude_none=True)).get("model_name")
     if not isinstance(listed_name, str):
@@ -11391,11 +11406,10 @@ async def model_list(
             model_info["id"] = response_id
             model_data.append(model_info)
 
-        # Catalog-only entries are advertised to every caller: they name no deployment,
-        # so the per-caller filters above have nothing to scope them by. A listing asked
-        # for access groups alone gets none of them, since a catalog entry is not a group.
         if not only_model_access_groups:
-            model_data.extend(advertised_model_rows(settings, tuple(row["id"] for row in model_data)))
+            model_data.extend(
+                await _catalog_listing_rows(settings, [row["id"] for row in model_data], user_api_key_dict)
+            )
 
         if wants_anthropic_format:
             admin_listing: Final = cast(Sequence[ModelInfoResponse], model_data)  # cast-ok: rows built above
@@ -11457,9 +11471,8 @@ async def model_list(
         model_info["id"] = response_id
         model_data.append(model_info)
 
-    # Same catalog merge as the scope=expand branch above.
     if not only_model_access_groups:
-        model_data.extend(advertised_model_rows(settings, tuple(row["id"] for row in model_data)))
+        model_data.extend(await _catalog_listing_rows(settings, [row["id"] for row in model_data], user_api_key_dict))
 
     if wants_anthropic_format:
         listing: Final = cast(Sequence[ModelInfoResponse], model_data)  # cast-ok: rows built above

litellm/types/proxy/model_listing.py

@@ -2,7 +2,7 @@
 
 from typing import Literal
 
-from typing_extensions import NotRequired, TypedDict
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 
 class ModelInfoMetadata(TypedDict):
@@ -13,6 +13,10 @@ class ModelInfoResponse(TypedDict):
     """OpenAI-compatible model object. `mode`, `max_input_tokens`, and
     `max_output_tokens` are attached when the cost map or deployment config
     knows them; `metadata` is present only with include_metadata=true.
+
+    `catalog_only` marks a row the proxy advertises but does not route, so a
+    client choosing a model from this listing can skip the ones a request would
+    be rejected for. It is absent on every routable model.
     """
 
     id: str
@@ -23,3 +27,4 @@ class ModelInfoResponse(TypedDict):
     max_input_tokens: NotRequired[int]
     max_output_tokens: NotRequired[int]
     metadata: NotRequired[ModelInfoMetadata]
+    catalog_only: NotRequired[ReadOnly[Literal[True]]]

tests/test_litellm/proxy/common_utils/test_advertised_models.py

@@ -27,6 +27,7 @@ def test_entry_becomes_a_row_carrying_its_configured_id_and_owner():
             "object": "model",
             "created": DEFAULT_MODEL_CREATED_AT_TIME,
             "owned_by": "example-provider",
+            "catalog_only": True,
         },
     ), f"expected one row carrying exactly the configured id and owner, got {rows}"
 
@@ -78,6 +79,7 @@ def test_row_carries_only_declared_fields_even_for_a_model_the_cost_map_knows():
             "object": "model",
             "created": DEFAULT_MODEL_CREATED_AT_TIME,
             "owned_by": "example-provider",
+            "catalog_only": True,
         },
     ), f"a catalog row must not pick up cost map details for {known_model}, got {rows}"
 
@@ -91,3 +93,64 @@ def test_configured_entries_are_typed_and_keep_their_order():
         ("catalog-only-model", "example-provider"),
         ("second", "other-provider"),
     ], f"entries should be parsed in configured order, got {entries}"
+
+
+class _RouterStub:
+    """Minimal stand-in for the bits of Router this module reads."""
+
+    def __init__(self, names: tuple[str, ...], groups: tuple[str, ...] = ()) -> None:
+        self._names: Final = names
+        self._groups: Final = groups
+
+    def get_model_names(self) -> list[str]:
+        return list(self._names)
+
+    def get_model_access_groups(self) -> dict[str, list[str]]:
+        return {group: [] for group in self._groups}
+
+
+def test_entry_naming_a_routed_model_hidden_from_this_caller_is_dropped():
+    """A catalog entry must not re-expose a deployment the listing filtered out.
+
+    `listed_ids` carries only what this caller may see, so an id that was scoped
+    away, paused or health-filtered would otherwise reappear under any owner the
+    config names.
+    """
+    rows: Final = advertised_model_rows(
+        _settings({"id": "hidden-deployment", "owned_by": "impostor"}),
+        [],
+        _RouterStub(("hidden-deployment",)),
+    )
+
+    assert rows == (), f"a routed model absent from this caller's listing must stay absent, got {rows}"
+
+
+def test_entry_naming_an_access_group_is_dropped():
+    rows: Final = advertised_model_rows(
+        _settings({"id": "beta-models", "owned_by": "impostor"}),
+        [],
+        _RouterStub((), ("beta-models",)),
+    )
+
+    assert rows == (), f"a catalog entry must not shadow an access group name, got {rows}"
+
+
+def test_entry_is_listed_when_the_router_knows_nothing_about_it():
+    rows: Final = advertised_model_rows(
+        _settings(CATALOG_ENTRY),
+        ["routed-model"],
+        _RouterStub(("routed-model", "another-model")),
+    )
+
+    assert [row["id"] for row in rows] == ["catalog-only-model"], (
+        f"an id no deployment claims should still be listed, got {rows}"
+    )
+
+
+def test_every_row_is_marked_catalog_only():
+    rows: Final = advertised_model_rows(_settings(CATALOG_ENTRY, {"id": "second", "owned_by": "other"}), [])
+
+    assert [row.get("catalog_only") for row in rows] == [
+        True,
+        True,
+    ], f"every catalog row must be marked so clients can tell it from a routable model, got {rows}"

tests/test_litellm/proxy/test_model_list_advertised_models.py

@@ -10,8 +10,11 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.proxy import proxy_server
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.utils import ProxyLogging
 
 CATALOG_SETTINGS: Final = {"advertised_models": [{"id": "catalog-only-model", "owned_by": "example-provider"}]}
 
@@ -130,3 +133,69 @@ async def test_catalog_entry_is_listed_for_scope_expand(patched_model_list, monk
         "routed-model",
         "catalog-only-model",
     ], f"the admin listing should carry catalog entries too, got {rows}"
+
+
+@pytest.mark.asyncio
+async def test_catalog_entry_cannot_resurrect_a_model_this_caller_may_not_see(patched_model_list, monkeypatch):
+    """A deployment filtered out of this caller's listing stays out.
+
+    The router knows `restricted-model`, but this key cannot see it, so it never
+    reaches `model_data`. Advertising that id must not put it back.
+    """
+    patched_model_list.get_model_names = MagicMock(return_value=["routed-model", "restricted-model"])
+    monkeypatch.setattr(
+        proxy_server,
+        "general_settings",
+        {"advertised_models": [{"id": "restricted-model", "owned_by": "impostor"}]},
+    )
+
+    rows: Final = await _listing()
+
+    assert [row["id"] for row in rows] == ["routed-model"], (
+        f"a deployment hidden from this caller must not reappear as a catalog entry, got {rows}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_catalog_rows_are_marked_and_routed_rows_are_not(patched_model_list, monkeypatch):
+    monkeypatch.setattr(proxy_server, "general_settings", CATALOG_SETTINGS)
+
+    rows: Final = await _listing()
+
+    assert [(row["id"], row.get("catalog_only")) for row in rows] == [
+        ("routed-model", None),
+        ("catalog-only-model", True),
+    ], f"only catalog rows carry the marker, so a client can tell them apart, got {rows}"
+
+
+class _HidingGate(CustomLogger):
+    """A listing callback that hides whichever names it was given."""
+
+    def __init__(self, hidden: frozenset[str]) -> None:
+        super().__init__()
+        self.hidden = hidden
+        self.seen: list[tuple[str, ...]] = []
+
+    async def async_filter_listed_models(self, user_api_key_dict, model_names):
+        self.seen.append(tuple(model_names))
+        return [name for name in model_names if name not in self.hidden]
+
+
+@pytest.mark.asyncio
+async def test_listing_callbacks_can_hide_a_catalog_entry(patched_model_list, monkeypatch):
+    """Catalog rows go through the same per-caller callbacks as routed rows.
+
+    Appending them afterwards would let an entry past a callback that was meant
+    to hide it.
+    """
+    gate: Final = _HidingGate(frozenset({"catalog-only-model"}))
+    monkeypatch.setattr(litellm, "callbacks", [gate])
+    ProxyLogging._callback_capabilities_cache.clear()
+    monkeypatch.setattr(proxy_server, "general_settings", CATALOG_SETTINGS)
+
+    rows: Final = await _listing()
+
+    assert [row["id"] for row in rows] == ["routed-model"], (
+        f"a callback that hides a catalog id must keep it out of the listing, got {rows}"
+    )
+    assert ("catalog-only-model",) in gate.seen, f"the catalog id must be offered to the callback, saw {gate.seen}"