fix(proxy): list videos with the deployment's credentials

GET /v1/videos now reaches the provider account of a deployment the caller may use. A model passed as ?model= or the x-litellm-model header is checked against the key's allowed models and routed through the router, so the deployment's api_key, api_base and other litellm_params apply

A provider-only list uses the caller's accessible deployment of that provider, team BYOK first, the same rule files and batches list follow. With no such deployment the env-key path is unchanged

A model the key cannot use answers 403 and an unknown model answers 400, both naming the model instead of reaching the provider

litellm/proxy/openai_files_endpoints/common_utils.py

@@ -3,7 +3,7 @@ import itertools
 import logging
 import mimetypes
 import re
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from types import MappingProxyType
@@ -559,7 +559,39 @@ def get_team_provider_credentials(
     """
     if llm_router is None:
         return None
+    match: Final = next(
+        _authorized_provider_credentials(
+            llm_router, user_api_key_dict, custom_llm_provider, include_accessible_deployments
+        ),
+        None,
+    )
+    return None if match is None else dict(match[1])
 
+
+def get_team_provider_deployment_id(
+    llm_router: Optional["Router"],
+    user_api_key_dict: "UserAPIKeyAuth",
+    custom_llm_provider: str,
+) -> str | None:
+    if llm_router is None:
+        return None
+    match: Final = next(
+        _authorized_provider_credentials(llm_router, user_api_key_dict, custom_llm_provider, True), None
+    )
+    deployment: Final = (
+        None
+        if match is None
+        else llm_router.registry().credentials(match[0], form="deployment", team_id=user_api_key_dict.team_id)
+    )
+    return None if deployment is None else deployment.model_info.id
+
+
+def _authorized_provider_credentials(
+    llm_router: "Router",
+    user_api_key_dict: "UserAPIKeyAuth",
+    custom_llm_provider: str,
+    include_accessible_deployments: bool,
+) -> Iterator[tuple[str, Mapping[str, object]]]:
     from litellm.proxy._types import SpecialModelNames
     from litellm.proxy.auth.auth_checks import is_model_allowed_by_pattern
     from litellm.proxy.auth.model_checks import get_complete_model_list, get_key_models
@@ -596,10 +628,10 @@ def get_team_provider_credentials(
             return True
         return isinstance(public_model_name, str) and _key_allowlist_covers(public_model_name, key_model_allowlist)
 
-    def _provider_credentials(model_id: str) -> dict[str, object] | None:
+    def _provider_credentials(model_id: str) -> tuple[str, Mapping[str, object]] | None:
         credentials: Final = registry.credentials(model_id, form="with_provider", team_id=team_id)
         if credentials is not None and credentials.get("custom_llm_provider") == custom_llm_provider:
-            return {key: value for key, value in credentials.items() if key != "model"}
+            return model_id, {key: value for key, value in credentials.items() if key != "model"}
         return None
 
     def _usable_team_deployment_order(deployment: _RouterDeployment) -> tuple[tuple[datetime, str], str] | None:
@@ -627,15 +659,10 @@ def get_team_provider_credentials(
             key=_by_team_deployment_rank,
         )
         team_deployment_ids: Final = tuple(deployment_id for _, deployment_id in team_deployment_orders)
-        team_credentials: Final = next(
-            (credentials for credentials in map(_provider_credentials, team_deployment_ids) if credentials is not None),
-            None,
-        )
-        if team_credentials is not None:
-            return team_credentials
+        yield from (found for found in map(_provider_credentials, team_deployment_ids) if found is not None)
 
     if not include_accessible_deployments:
-        return None
+        return
 
     # 2. Fall back to deployments the caller is allowed to access. The key's
     #    effective allowlist (sentinels and access groups already expanded by
@@ -668,15 +695,12 @@ def get_team_provider_credentials(
             is_model_allowed_by_pattern(model=model_name, allowed_model_pattern=granted) for granted in granted_models
         )
     )
-    return next(
-        (
-            credentials
-            for credentials in map(
-                _provider_credentials, dict.fromkeys(itertools.chain(granted_models, deployments_granted_by_wildcard))
-            )
-            if credentials is not None
-        ),
-        None,
+    yield from (
+        found
+        for found in map(
+            _provider_credentials, dict.fromkeys(itertools.chain(granted_models, deployments_granted_by_wildcard))
+        )
+        if found is not None
     )
 
 

litellm/proxy/video_endpoints/endpoints.py

@@ -8,6 +8,7 @@ from pydantic import TypeAdapter
 from starlette.datastructures import UploadFile as StarletteUploadFile
 
 from litellm.proxy._types import *
+from litellm.proxy.auth.auth_utils import MODEL_ROUTING_HEADER_NAME
 from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth, user_api_key_auth
 from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
 from litellm.proxy.common_utils.http_parsing_utils import (  # noqa: F401  # legacy module exports
@@ -28,6 +29,7 @@ from litellm.proxy.video_endpoints.utils import (
     fallback_requested_model,
     get_custom_provider_from_data,
     routable_model_name,
+    video_list_model,
     video_model_group,
     video_reference_to_id,
 )
@@ -193,6 +195,14 @@ async def video_list(
     # Process request using ProxyBaseLLMRequestProcessing
     processor: Final = ProxyBaseLLMRequestProcessing(data=data)
     try:
+        listing_model: Final = await video_list_model(
+            requested_model=request.query_params.get("model") or request.headers.get(MODEL_ROUTING_HEADER_NAME),
+            custom_llm_provider=custom_llm_provider,
+            llm_router=llm_router,
+            user_api_key_dict=user_api_key_dict,
+        )
+        if listing_model:
+            data["model"] = listing_model
         listed: Final[object] = await processor.base_process_llm_request(
             request=request,
             fastapi_response=fastapi_response,

litellm/proxy/video_endpoints/utils.py

@@ -3,6 +3,10 @@ from typing import TYPE_CHECKING, Any, Final
 import orjson
 from pydantic import BaseModel, ConfigDict
 
+from litellm.proxy.openai_files_endpoints.common_utils import (
+    authorize_model_for_key,
+    get_team_provider_deployment_id,
+)
 from litellm.types.videos.main import DecodedVideoId, VideoObject
 from litellm.types.videos.utils import (
     decode_video_id_with_provider,
@@ -12,6 +16,7 @@ from litellm.types.videos.utils import (
 )
 
 if TYPE_CHECKING:
+    from litellm.proxy._types import UserAPIKeyAuth
     from litellm.router import Router
 
 
@@ -92,6 +97,24 @@ def routable_model_name(model: object, llm_router: "Router | None", custom_llm_p
     return name if llm_router.registry().resolve(name, kind="from_id", custom_llm_provider=provider) else None
 
 
+async def video_list_model(
+    requested_model: str | None,
+    custom_llm_provider: str | None,
+    llm_router: "Router | None",
+    user_api_key_dict: "UserAPIKeyAuth",
+) -> str | None:
+    if requested_model:
+        await authorize_model_for_key(
+            model_id=requested_model, llm_router=llm_router, user_api_key_dict=user_api_key_dict
+        )
+        return requested_model
+    if not custom_llm_provider:
+        return None
+    return get_team_provider_deployment_id(
+        llm_router=llm_router, user_api_key_dict=user_api_key_dict, custom_llm_provider=custom_llm_provider
+    )
+
+
 def video_model_group(decoded: DecodedVideoId, llm_router: "Router | None") -> str | None:
     """
     The group a video id names. A provider model that several groups share resolves to the group of the

tests/unit/proxy/video_endpoints/test_endpoints.py

@@ -28,17 +28,22 @@ mock's call args), and a brand-new kwarg added to this layer surfaces as a failu
 
 from contextlib import ExitStack
 from dataclasses import dataclass
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterator
 from typing import Any, Dict, Final, Optional
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import orjson
 import pytest
+import respx
+from fastapi.testclient import TestClient
 
 
+import litellm
 import litellm.proxy.proxy_server as proxy_server
 import litellm.proxy.video_endpoints.endpoints as endpoints
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
 from litellm.proxy.utils import ProxyLogging
 from litellm.router import Router
@@ -662,6 +667,129 @@ async def test_list__provider_from_header(harness):
     }
 
 
+_DEPLOYMENT_ARK_HOST: Final = "ark.deployment.example"
+_DEPLOYMENT_ARK_KEY: Final = "deployment-ark-key"
+_ARK_TASKS_PATH: Final = "/api/v3/contents/generations/tasks"
+_EMPTY_ARK_TASK_LIST: Final = {"items": [], "total": 0}
+
+
+def _router_with_a_volcengine_and_an_openai_deployment() -> Router:
+    return Router(
+        model_list=[
+            {
+                "model_name": "seedance",
+                "litellm_params": {
+                    "model": "volcengine/doubao-seedance-2-5-260628",
+                    "api_key": _DEPLOYMENT_ARK_KEY,
+                    "api_base": f"https://{_DEPLOYMENT_ARK_HOST}/api/v3",
+                },
+            },
+            {
+                "model_name": "sora",
+                "litellm_params": {"model": "openai/sora-2", "api_key": "deployment-openai-key"},
+            },
+        ]
+    )
+
+
+@pytest.fixture
+def proxy_client_for_key_models(monkeypatch: pytest.MonkeyPatch) -> Iterator[Callable[[list[str]], TestClient]]:
+    for env_name in ("ARK_API_KEY", "VOLCENGINE_API_KEY", "ARK_API_BASE", "VOLCENGINE_API_BASE"):
+        monkeypatch.delenv(env_name, raising=False)
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(proxy_server, "llm_router", _router_with_a_volcengine_and_an_openai_deployment())
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    def proxy_client(key_models: list[str]) -> TestClient:
+        key: Final = UserAPIKeyAuth(api_key="sk-video-list", models=key_models)
+        monkeypatch.setitem(proxy_server.app.dependency_overrides, user_api_key_auth, lambda: key)
+        return TestClient(proxy_server.app)
+
+    yield proxy_client
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+
+@pytest.mark.parametrize(
+    "query, headers",
+    [
+        pytest.param({"model": "seedance"}, {}, id="model-query"),
+        pytest.param({}, {"x-litellm-model": "seedance"}, id="model-header"),
+        pytest.param({"custom_llm_provider": "volcengine"}, {}, id="provider-only"),
+    ],
+)
+def test_list__reaches_the_deployments_provider_account_with_its_credentials(
+    proxy_client_for_key_models: Callable[[list[str]], TestClient],
+    query: dict[str, str],
+    headers: dict[str, str],
+) -> None:
+    client: Final = proxy_client_for_key_models(["seedance"])
+
+    with respx.mock(assert_all_called=False) as respx_mock:
+        upstream: Final = respx_mock.get(f"https://{_DEPLOYMENT_ARK_HOST}{_ARK_TASKS_PATH}").mock(
+            return_value=httpx.Response(200, json=_EMPTY_ARK_TASK_LIST)
+        )
+        response: Final = client.get("/v1/videos", params=query, headers=headers)
+
+    assert response.status_code == 200, response.text
+    assert response.json()["data"] == []
+    assert upstream.call_count == 1
+    assert upstream.calls.last.request.headers["authorization"] == f"Bearer {_DEPLOYMENT_ARK_KEY}"
+
+
+def test_list__provider_only_never_borrows_a_deployment_the_key_cannot_use(
+    proxy_client_for_key_models: Callable[[list[str]], TestClient], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("ARK_API_KEY", "env-ark-key")
+    client: Final = proxy_client_for_key_models(["sora"])
+
+    with respx.mock(assert_all_called=False) as respx_mock:
+        upstream: Final = respx_mock.get(url__regex=f".*{_ARK_TASKS_PATH}$").mock(
+            return_value=httpx.Response(200, json=_EMPTY_ARK_TASK_LIST)
+        )
+        response: Final = client.get("/v1/videos", params={"custom_llm_provider": "volcengine"})
+
+    assert response.status_code == 200, response.text
+    assert upstream.calls.last.request.headers["authorization"] == "Bearer env-ark-key"
+    assert upstream.calls.last.request.url.host != _DEPLOYMENT_ARK_HOST
+
+
+def test_list__provider_only_on_a_proxy_without_a_router_is_a_clean_client_error(
+    proxy_client_for_key_models: Callable[[list[str]], TestClient], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(proxy_server, "llm_router", None)
+    client: Final = proxy_client_for_key_models([])
+
+    with respx.mock(assert_all_called=False) as respx_mock:
+        upstream: Final = respx_mock.route().mock(return_value=httpx.Response(200, json=_EMPTY_ARK_TASK_LIST))
+        response: Final = client.get("/v1/videos", params={"custom_llm_provider": "volcengine"})
+
+    assert response.status_code == 400, response.text
+    assert "Invalid model name passed in model=None" in response.json()["error"]["message"]
+    assert upstream.call_count == 0
+
+
+@pytest.mark.parametrize(
+    "key_models, requested_model",
+    [
+        pytest.param(["sora"], "seedance", id="model-the-key-cannot-use"),
+        pytest.param([], "no-such-model", id="unknown-model"),
+    ],
+)
+def test_list__a_model_that_cannot_be_listed_is_a_client_error_naming_it(
+    proxy_client_for_key_models: Callable[[list[str]], TestClient], key_models: list[str], requested_model: str
+) -> None:
+    client: Final = proxy_client_for_key_models(key_models)
+
+    with respx.mock(assert_all_called=False) as respx_mock:
+        upstream: Final = respx_mock.route().mock(return_value=httpx.Response(200, json=_EMPTY_ARK_TASK_LIST))
+        response: Final = client.get("/v1/videos", params={"model": requested_model})
+
+    assert 400 <= response.status_code < 500, response.text
+    assert requested_model in response.json()["error"]["message"]
+    assert "Traceback" not in response.text
+    assert upstream.call_count == 0
+
+
 # =========================================================================== #
 #   POST /v1/videos/{video_id}/remix  -  video_remix                           #
 # =========================================================================== #