From c510bd1aad2e17d52ee72d979aeaf24a6692910e Mon Sep 17 00:00:00 2001 From: hx <1367557521@qq.com> Date: Sun, 27 Sep 2026 18:50:13 +0800 Subject: [PATCH] fix(proxy): enforce per-caller ownership on proxy-created videos Record the creating caller's owner scope in LiteLLM_ManagedObjectTable (file_purpose=video, keyed by the provider-native video id) when videos are created, remixed, edited, or extended, and reject status, content, remix, edit, and extension calls on a video the caller does not own with 403. Proxy admins bypass the check, untracked videos are admin-only, and list results are filtered to the caller's videos. Without a database the checks are skipped, matching previous behavior. Co-authored-by: Cursor --- litellm/proxy/video_endpoints/endpoints.py | 19 +- litellm/proxy/video_endpoints/ownership.py | 167 ++++++++++++++ .../proxy/video_endpoints/test_ownership.py | 218 ++++++++++++++++++ 3 files changed, 403 insertions(+), 1 deletion(-) create mode 100644 litellm/proxy/video_endpoints/ownership.py create mode 100644 tests/test_litellm/proxy/video_endpoints/test_ownership.py diff --git a/litellm/proxy/video_endpoints/endpoints.py b/litellm/proxy/video_endpoints/endpoints.py index 540860c1a5b..9c31876d298 100644 --- a/litellm/proxy/video_endpoints/endpoints.py +++ b/litellm/proxy/video_endpoints/endpoints.py @@ -16,6 +16,11 @@ from litellm.proxy.common_utils.openai_endpoint_utils import ( get_custom_llm_provider_from_request_query, ) from litellm.proxy.image_endpoints.endpoints import batch_to_bytesio +from litellm.proxy.video_endpoints.ownership import ( + assert_user_can_access_video, + filter_video_list_for_caller, + record_video_owner, +) from litellm.proxy.video_endpoints.utils import ( encode_character_id_in_response, extract_model_from_target_model_names, @@ -116,6 +121,7 @@ async def video_generation( version=version, ) else: + await record_video_owner(generated, user_api_key_dict) return generated @@ -203,7 +209,7 @@ async def video_list( version=version, ) else: - return listed + return await filter_video_list_for_caller(listed, user_api_key_dict) @router.get( @@ -250,6 +256,8 @@ async def video_status( version, ) + await assert_user_can_access_video(video_id, user_api_key_dict) + # Create data with video_id data: Final[dict[str, object]] = {"video_id": video_id} @@ -353,6 +361,8 @@ async def video_content( version, ) + await assert_user_can_access_video(video_id, user_api_key_dict) + # Create data with video_id data: Final[dict[str, object]] = {"video_id": video_id} @@ -462,6 +472,8 @@ async def video_remix( version, ) + await assert_user_can_access_video(video_id, user_api_key_dict) + data: Final = await _read_request_body(request=request) data["video_id"] = video_id @@ -514,6 +526,7 @@ async def video_remix( version=version, ) else: + await record_video_owner(remixed, user_api_key_dict) return remixed @@ -780,6 +793,7 @@ async def video_edit( data["video_id"] = "" else: data["video_id"] = video_reference_to_id(uploaded_video) + await assert_user_can_access_video(data["video_id"], user_api_key_dict) decoded: Final = decode_video_id_with_provider(data["video_id"]) provider_from_id: Final = decoded.get("custom_llm_provider") @@ -827,6 +841,7 @@ async def video_edit( version=version, ) else: + await record_video_owner(edited, user_api_key_dict) return edited @@ -877,6 +892,7 @@ async def video_extension( data: Final = await _read_request_body(request=request) data["video_id"] = video_reference_to_id(data.pop("video", None)) + await assert_user_can_access_video(data["video_id"], user_api_key_dict) decoded: Final = decode_video_id_with_provider(data["video_id"]) provider_from_id: Final = decoded.get("custom_llm_provider") @@ -924,4 +940,5 @@ async def video_extension( version=version, ) else: + await record_video_owner(extended, user_api_key_dict) return extended diff --git a/litellm/proxy/video_endpoints/ownership.py b/litellm/proxy/video_endpoints/ownership.py new file mode 100644 index 00000000000..49d8fe0eb13 --- /dev/null +++ b/litellm/proxy/video_endpoints/ownership.py @@ -0,0 +1,167 @@ +"""Per-caller ownership of videos created through the proxy. + +Ownership rows live in ``LiteLLM_ManagedObjectTable`` with ``file_purpose="video"`` +(the same table and owner scopes container ownership uses). A row is keyed by the +provider-native video id only, so re-wrapping an id with a different provider or +model_id cannot dodge the lookup. Proxy admins bypass the check; for everyone else, +a video without a row is treated as admin-only. + +Without a connected database there is nowhere to record ownership: recording and +checks are skipped and videos remain reachable by any caller allowed on the route. +""" + +import json +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING, Final + +from fastapi import HTTPException +from typing_extensions import TypeIs # noqa: TID251 # narrows untyped wire payloads without a runtime conversion + +from litellm._logging import verbose_proxy_logger +from litellm.caching.in_memory_cache import InMemoryCache +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.common_utils.resource_ownership import ( + get_primary_resource_owner_scope, + get_resource_owner_scopes, + is_proxy_admin, + user_can_access_resource_owner, +) +from litellm.repositories.table_repositories import ManagedObjectRepository +from litellm.types.videos.utils import extract_original_video_id + +if TYPE_CHECKING: + from litellm.proxy.utils import PrismaClient + +VIDEO_OBJECT_PURPOSE: Final = "video" + +# Status is polled; a short-lived cache of known owners keeps polling off the DB. +# Only positive answers are cached, so a missing row is always re-checked. +_VIDEO_OWNER_CACHE: Final = InMemoryCache(max_size_in_memory=10000, default_ttl=60) + + +def _video_model_object_id(video_id: str) -> str: + return f"{VIDEO_OBJECT_PURPOSE}:{extract_original_video_id(video_id)}" + + +def _video_id_of(item: object) -> str | None: + match item: + case {"id": str(video_id)} if video_id: + return video_id + case object(id=str(video_id)) if video_id: + return video_id + case _: + return None + + +def _is_object_mapping(value: object) -> TypeIs[Mapping[str, object]]: # guard-ok: provider JSON objects have str keys + return isinstance(value, Mapping) + + +def _is_object_sequence(value: object) -> TypeIs[Sequence[object]]: # guard-ok: a list is a Sequence of anything + return isinstance(value, list) + + +def _get_prisma_client() -> "PrismaClient | None": + from litellm.proxy.proxy_server import prisma_client + + return prisma_client + + +async def record_video_owner(response: object, user_api_key_dict: UserAPIKeyAuth) -> None: + """Stamp the caller as owner of a video the provider just created. + + Failures are logged rather than raised: the provider job already exists and is + billed, so the caller still gets its id; the untracked video is admin-only. + """ + prisma_client: Final = _get_prisma_client() + if prisma_client is None: + return + video_id: Final = _video_id_of(response) + owner: Final = get_primary_resource_owner_scope(user_api_key_dict) + if video_id is None or owner is None: + verbose_proxy_logger.warning( + "Skipping video ownership tracking: response has no id or caller has no identity scope" + ) + return + model_object_id: Final = _video_model_object_id(video_id) + row: Final = { # mutable-ok: prisma rows are plain dicts + "unified_object_id": video_id, + "model_object_id": model_object_id, + "file_object": json.dumps({"id": video_id, "object": "video"}), # mutable-ok: serialized immediately + "file_purpose": VIDEO_OBJECT_PURPOSE, + "created_by": owner, + "updated_by": owner, + } + try: + await ManagedObjectRepository(prisma_client).table.upsert( + where={"model_object_id": model_object_id}, # mutable-ok: prisma filters are plain dicts + data={"create": row, "update": {"updated_by": owner}}, # mutable-ok: prisma payloads are plain dicts + ) + except Exception as e: + verbose_proxy_logger.exception( + "Video ownership recording failed; video_id=%s is untracked and admin-only: %s", video_id, e + ) + + +async def _get_video_owner(prisma_client: "PrismaClient", model_object_id: str) -> str | None: + cached: Final = _VIDEO_OWNER_CACHE.get_cache(model_object_id) + if isinstance(cached, str): + return cached + row: Final = await ManagedObjectRepository(prisma_client).table.find_first( + where={ # mutable-ok: prisma filters are plain dicts + "model_object_id": model_object_id, + "file_purpose": VIDEO_OBJECT_PURPOSE, + } + ) + owner: Final = getattr(row, "created_by", None) if row is not None else None + if not isinstance(owner, str): + return None + _VIDEO_OWNER_CACHE.set_cache(model_object_id, owner) + return owner + + +async def assert_user_can_access_video(video_id: str, user_api_key_dict: UserAPIKeyAuth) -> None: + """Raise 403 unless the caller owns ``video_id`` (or is a proxy admin).""" + if not video_id or is_proxy_admin(user_api_key_dict): + return + prisma_client: Final = _get_prisma_client() + if prisma_client is None: + return + owner: Final = await _get_video_owner(prisma_client, _video_model_object_id(video_id)) + if not user_can_access_resource_owner(owner, user_api_key_dict): + raise HTTPException(status_code=403, detail="Forbidden") + + +async def filter_video_list_for_caller(listed: object, user_api_key_dict: UserAPIKeyAuth) -> object: + """Drop videos the caller does not own from a provider list page. + + Pagination cursors are left as the provider returned them so ``after`` keeps + walking the provider's list even when a page has no videos owned by the caller. + """ + if is_proxy_admin(user_api_key_dict) or not _is_object_mapping(listed): + return listed + prisma_client: Final = _get_prisma_client() + items: Final = listed.get("data") + if prisma_client is None or not _is_object_sequence(items): + return listed + ids_by_item: Final = tuple((item, _video_id_of(item)) for item in items) + candidate_ids: Final = tuple( + _video_model_object_id(video_id) for _, video_id in ids_by_item if video_id is not None + ) + owner_scopes: Final = get_resource_owner_scopes(user_api_key_dict) + rows: Final = ( + await ManagedObjectRepository(prisma_client).table.find_many( + where={ # mutable-ok: prisma filters are plain dicts + "model_object_id": {"in": list(candidate_ids)}, # mutable-ok: prisma filters are plain dicts + "file_purpose": VIDEO_OBJECT_PURPOSE, + "created_by": {"in": owner_scopes}, # mutable-ok: prisma filters are plain dicts + } + ) + if candidate_ids and owner_scopes + else () + ) + owned: Final = frozenset(row.model_object_id for row in rows) + kept: Final = tuple( + item for item, video_id in ids_by_item if video_id is not None and _video_model_object_id(video_id) in owned + ) + return {**listed, "data": kept} # mutable-ok: provider list JSON body diff --git a/tests/test_litellm/proxy/video_endpoints/test_ownership.py b/tests/test_litellm/proxy/video_endpoints/test_ownership.py new file mode 100644 index 00000000000..64dd83149d7 --- /dev/null +++ b/tests/test_litellm/proxy/video_endpoints/test_ownership.py @@ -0,0 +1,218 @@ +""" +Ownership tests for the proxy video endpoints. + +Requests go through the real FastAPI routes and the real ownership module. Only two +I/O boundaries are replaced: the provider call (``base_process_llm_request``) and the +Prisma ``litellm_managedobjecttable`` delegate, which is an in-memory table honoring +the subset of the query API the ownership module uses. +""" + +import uuid +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import litellm.proxy.proxy_server as proxy_server +from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth, hash_token +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing +from litellm.proxy.video_endpoints import endpoints +from litellm.types.videos.utils import encode_video_id_with_provider + +ALICE = UserAPIKeyAuth(user_id="alice", api_key=hash_token("sk-alice")) +BOB = UserAPIKeyAuth(user_id="bob", api_key=hash_token("sk-bob")) +KEY_ONLY_A = UserAPIKeyAuth(api_key=hash_token("sk-service-a")) +KEY_ONLY_B = UserAPIKeyAuth(api_key=hash_token("sk-service-b")) +ADMIN = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN) + + +def _matches(row: SimpleNamespace, where: dict) -> bool: + for field, condition in where.items(): + value = getattr(row, field, None) + if isinstance(condition, dict): + if value not in condition["in"]: + return False + elif value != condition: + return False + return True + + +class InMemoryManagedObjectTable: + def __init__(self, fail_writes: bool = False): + self.rows: dict[str, SimpleNamespace] = {} + self.fail_writes = fail_writes + + async def upsert(self, where: dict, data: dict) -> SimpleNamespace: + if self.fail_writes: + raise RuntimeError("database unavailable") + key = where["model_object_id"] + if key in self.rows: + self.rows[key] = SimpleNamespace(**{**vars(self.rows[key]), **data["update"]}) + else: + self.rows[key] = SimpleNamespace(**data["create"]) + return self.rows[key] + + async def find_first(self, where: dict) -> SimpleNamespace | None: + return next((row for row in self.rows.values() if _matches(row, where)), None) + + async def find_many(self, where: dict) -> list[SimpleNamespace]: + return [row for row in self.rows.values() if _matches(row, where)] + + +class Provider: + """The provider behind ``base_process_llm_request``: creates and serves videos.""" + + def __init__(self): + self.calls: list[str] = [] + + async def respond(self, processor, *, route_type: str, **kwargs): + self.calls.append(route_type) + if route_type in ("avideo_generation", "avideo_remix", "avideo_edit", "avideo_extension"): + return {"id": f"video_{uuid.uuid4().hex}", "object": "video", "status": "queued"} + if route_type == "avideo_content": + return b"mp4-bytes" + if route_type == "avideo_list": + return {"object": "list", "data": processor.data["listed"], "has_more": False} + return {"id": processor.data["video_id"], "object": "video", "status": "completed"} + + +@pytest.fixture +def provider(monkeypatch) -> Provider: + fake = Provider() + + async def base_process_llm_request(processor, **kwargs): + return await fake.respond(processor, **kwargs) + + monkeypatch.setattr(ProxyBaseLLMRequestProcessing, "base_process_llm_request", base_process_llm_request) + return fake + + +@pytest.fixture +def table(monkeypatch) -> InMemoryManagedObjectTable: + rows = InMemoryManagedObjectTable() + monkeypatch.setattr( + proxy_server, "prisma_client", SimpleNamespace(db=SimpleNamespace(litellm_managedobjecttable=rows)) + ) + return rows + + +def _client(auth: UserAPIKeyAuth) -> TestClient: + app = FastAPI() + app.include_router(endpoints.router) + app.dependency_overrides[user_api_key_auth] = lambda: auth + return TestClient(app) + + +def _create_video(auth: UserAPIKeyAuth) -> str: + response = _client(auth).post("/v1/videos", json={"model": "sora-2", "prompt": "a sunset"}) + assert response.status_code == 200 + return response.json()["id"] + + +def _access(auth: UserAPIKeyAuth, video_id: str) -> dict[str, int]: + client = _client(auth) + return { + "status": client.get(f"/v1/videos/{video_id}").status_code, + "content": client.get(f"/v1/videos/{video_id}/content").status_code, + "remix": client.post(f"/v1/videos/{video_id}/remix", json={"prompt": "again"}).status_code, + "edit": client.post("/v1/videos/edits", json={"prompt": "brighter", "video": {"id": video_id}}).status_code, + "extension": client.post( + "/v1/videos/extensions", json={"prompt": "continue", "video": {"id": video_id}} + ).status_code, + } + + +ALL_OK = {"status": 200, "content": 200, "remix": 200, "edit": 200, "extension": 200} +ALL_FORBIDDEN = {"status": 403, "content": 403, "remix": 403, "edit": 403, "extension": 403} + + +def test_owner_can_use_every_video_route(provider, table): + video_id = _create_video(ALICE) + + assert table.rows[f"video:{video_id}"].created_by == "alice" + assert _access(ALICE, video_id) == ALL_OK + + +def test_other_user_is_forbidden_and_the_provider_is_never_called(provider, table): + video_id = _create_video(ALICE) + provider.calls.clear() + + assert _access(BOB, video_id) == ALL_FORBIDDEN + assert provider.calls == [] + + +def test_key_scoped_owner_blocks_a_different_key(provider, table): + video_id = _create_video(KEY_ONLY_A) + + assert _access(KEY_ONLY_B, video_id) == ALL_FORBIDDEN + assert _access(KEY_ONLY_A, video_id) == ALL_OK + + +def test_proxy_admin_can_access_any_video(provider, table): + video_id = _create_video(ALICE) + + assert _access(ADMIN, video_id) == ALL_OK + + +def test_untracked_video_is_admin_only_when_a_database_is_connected(provider, table): + video_id = f"video_{uuid.uuid4().hex}" + + assert _access(ALICE, video_id) == ALL_FORBIDDEN + assert _access(ADMIN, video_id) == ALL_OK + + +def test_rewrapping_the_provider_id_does_not_bypass_the_owner_check(provider, table): + video_id = _create_video(ALICE) + rewrapped = encode_video_id_with_provider(video_id, "azure", "attacker-deployment") + + assert rewrapped != video_id + assert _access(BOB, rewrapped) == ALL_FORBIDDEN + + +def test_videos_derived_from_an_owned_video_belong_to_the_caller(provider, table): + video_id = _create_video(ALICE) + remixed = _client(ALICE).post(f"/v1/videos/{video_id}/remix", json={"prompt": "again"}).json()["id"] + + assert _access(ALICE, remixed)["status"] == 200 + assert _access(BOB, remixed)["status"] == 403 + + +def test_list_only_returns_the_callers_videos(provider, table): + alice_video = _create_video(ALICE) + bob_video = _create_video(BOB) + listed = [{"id": alice_video, "object": "video"}, {"id": bob_video, "object": "video"}] + provider_listing = {"listed": listed} + + def list_as(auth: UserAPIKeyAuth) -> list[str]: + async def with_listing(processor, **kwargs): + processor.data.update(provider_listing) + return await provider.respond(processor, **kwargs) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(ProxyBaseLLMRequestProcessing, "base_process_llm_request", with_listing) + response = _client(auth).get("/v1/videos") + assert response.status_code == 200 + return [item["id"] for item in response.json()["data"]] + + assert list_as(ALICE) == [alice_video] + assert list_as(BOB) == [bob_video] + assert list_as(ADMIN) == [alice_video, bob_video] + + +def test_without_a_database_ownership_is_not_enforced(provider, monkeypatch): + monkeypatch.setattr(proxy_server, "prisma_client", None) + video_id = _create_video(ALICE) + + assert _access(BOB, video_id) == ALL_OK + + +def test_failed_ownership_write_still_returns_the_created_video(provider, table): + table.fail_writes = True + + video_id = _create_video(ALICE) + + assert video_id.startswith("video_") + assert table.rows == {} + assert _access(ALICE, video_id)["status"] == 403