From 0722d6efdc0dfd52360d840b06751a2e1e321728 Mon Sep 17 00:00:00 2001 From: hx <1367557521@qq.com> Date: Sun, 27 Sep 2026 18:15:56 +0800 Subject: [PATCH] fix(proxy): stop embedding the caller's key in video ids The owner stamp added to every provider's video id was base64, not encrypted, and fell back to the raw bearer credential whenever the auth object had no hashed token (no master key, custom auth). It also did not bind anything: the provider request id can be decoded out of a stamped id and replayed as a plain id. Video ids go back to being the same bearer capability they are upstream Co-authored-by: Cursor --- litellm/proxy/video_endpoints/endpoints.py | 21 +----------- litellm/proxy/video_endpoints/utils.py | 33 +------------------ litellm/types/videos/main.py | 1 - litellm/types/videos/utils.py | 10 +----- .../proxy/video_endpoints/test_utils.py | 14 -------- 5 files changed, 3 insertions(+), 76 deletions(-) diff --git a/litellm/proxy/video_endpoints/endpoints.py b/litellm/proxy/video_endpoints/endpoints.py index 4bddb528e0c..125bb30f86c 100644 --- a/litellm/proxy/video_endpoints/endpoints.py +++ b/litellm/proxy/video_endpoints/endpoints.py @@ -17,14 +17,11 @@ from litellm.proxy.common_utils.openai_endpoint_utils import ( ) from litellm.proxy.image_endpoints.endpoints import batch_to_bytesio from litellm.proxy.video_endpoints.utils import ( - assert_video_owner, encode_character_id_in_response, extract_model_from_target_model_names, get_custom_provider_from_data, infer_video_provider_from_model, resolve_video_request_model, - stamp_video_owner, - video_owner_from_key, video_reference_to_id, ) from litellm.types.videos.utils import ( @@ -120,19 +117,7 @@ async def video_generation( version=version, ) else: - return _stamp_generated_video_owner( - generated, video_owner_from_key(user_api_key_dict.token, user_api_key_dict.api_key) - ) - - -def _stamp_generated_video_owner(result: object, owner: str | None) -> object: - video_id: Final = getattr(result, "id", None) - if isinstance(video_id, str): - result.id = stamp_video_owner(video_id, owner) # mutable-ok: response object id is rewritten before return - return result - if isinstance(result, dict) and isinstance(result.get("id"), str): - result["id"] = stamp_video_owner(result["id"], owner) # mutable-ok: JSON response body - return result + return generated @router.get( @@ -266,8 +251,6 @@ async def video_status( version, ) - assert_video_owner(video_id, video_owner_from_key(user_api_key_dict.token, user_api_key_dict.api_key)) - # Create data with video_id data: Final[dict[str, object]] = {"video_id": video_id} @@ -371,8 +354,6 @@ async def video_content( version, ) - assert_video_owner(video_id, video_owner_from_key(user_api_key_dict.token, user_api_key_dict.api_key)) - # Create data with video_id data: Final[dict[str, object]] = {"video_id": video_id} diff --git a/litellm/proxy/video_endpoints/utils.py b/litellm/proxy/video_endpoints/utils.py index baa2aea81c3..9795e925863 100644 --- a/litellm/proxy/video_endpoints/utils.py +++ b/litellm/proxy/video_endpoints/utils.py @@ -3,44 +3,13 @@ from typing import Final, Protocol import orjson -from litellm.proxy._types import ProxyException -from litellm.types.videos.utils import ( - decode_video_id_with_provider, - encode_character_id_with_provider, - encode_video_id_with_provider, -) +from litellm.types.videos.utils import encode_character_id_with_provider class VideoModelIdResolver(Protocol): def resolve_model_name_from_model_id(self, model_id: str | None) -> str | None: ... -def video_owner_from_key(token: str | None, api_key: str | None) -> str | None: - return token or api_key - - -def assert_video_owner(video_id: str, owner: str | None) -> None: - recorded: Final = decode_video_id_with_provider(video_id).get("owner") - if recorded and recorded != owner: - raise ProxyException( - message="Video does not belong to this API key", - type="permission_error", - param="video_id", - code=403, - ) - - -def stamp_video_owner(video_id: str, owner: str | None) -> str: - if not owner: - return video_id - decoded: Final = decode_video_id_with_provider(video_id) - provider: Final = decoded.get("custom_llm_provider") - raw_id: Final = decoded.get("video_id") - if not provider or not raw_id or decoded.get("owner"): - return video_id - return encode_video_id_with_provider(raw_id, provider, decoded.get("model_id"), owner) - - def infer_video_provider_from_model(model: str | None) -> str | None: if not isinstance(model, str) or not model: return None diff --git a/litellm/types/videos/main.py b/litellm/types/videos/main.py index f876df2d765..f4369fd95af 100644 --- a/litellm/types/videos/main.py +++ b/litellm/types/videos/main.py @@ -102,7 +102,6 @@ class DecodedVideoId(TypedDict, total=False): custom_llm_provider: str | None model_id: str | None video_id: str - owner: ReadOnly[str | None] class CharacterObject(BaseModel): diff --git a/litellm/types/videos/utils.py b/litellm/types/videos/utils.py index e07f71b3240..b23b2269543 100644 --- a/litellm/types/videos/utils.py +++ b/litellm/types/videos/utils.py @@ -35,9 +35,7 @@ def _add_base64_padding(value: str) -> str: return value -def encode_video_id_with_provider( - video_id: str, provider: str, model_id: str | None = None, owner: str | None = None -) -> str: +def encode_video_id_with_provider(video_id: str, provider: str, model_id: str | None = None) -> str: """Encode provider and model_id into video_id using base64.""" if not provider or not video_id: return video_id @@ -52,8 +50,6 @@ def encode_video_id_with_provider( # ID is not encoded (even if it starts with video_), so encode it assembled_id = str(SpecialEnums.LITELLM_MANAGED_VIDEO_COMPLETE_STR.value).format(provider, model_id or "", video_id) - if owner: - assembled_id = f"{assembled_id};owner:{owner}" base64_encoded_id: Final[str] = base64.b64encode(assembled_id.encode("utf-8")).decode("utf-8") @@ -93,24 +89,20 @@ def decode_video_id_with_provider(encoded_video_id: str) -> DecodedVideoId: custom_llm_provider = None model_id = None decoded_video_id = encoded_video_id - owner = None if len(parts) >= 3: custom_llm_provider_part: Final = parts[0] model_id_part: Final = parts[1] video_id_part: Final = parts[2] - owner_part: Final = next((part for part in parts[3:] if part.startswith("owner:")), None) custom_llm_provider = custom_llm_provider_part.replace("litellm:custom_llm_provider:", "") model_id = model_id_part.replace("model_id:", "") decoded_video_id = video_id_part.replace("video_id:", "") - owner = owner_part.removeprefix("owner:") if owner_part else None return DecodedVideoId( custom_llm_provider=custom_llm_provider, model_id=model_id, video_id=decoded_video_id, - owner=owner, ) except Exception as e: verbose_logger.debug("Error decoding video_id '%s': %s", encoded_video_id, e) diff --git a/tests/test_litellm/proxy/video_endpoints/test_utils.py b/tests/test_litellm/proxy/video_endpoints/test_utils.py index 3fb523944aa..4f7e4d5523e 100644 --- a/tests/test_litellm/proxy/video_endpoints/test_utils.py +++ b/tests/test_litellm/proxy/video_endpoints/test_utils.py @@ -15,21 +15,17 @@ are checked by the genuine decode round-trip. import pytest -from litellm.proxy._types import ProxyException from litellm.proxy.video_endpoints.utils import ( - assert_video_owner, encode_character_id_in_response, extract_model_from_target_model_names, get_custom_provider_from_data, infer_video_provider_from_model, resolve_video_request_model, - stamp_video_owner, video_reference_to_id, ) from litellm.types.videos.utils import ( decode_character_id_with_provider, encode_character_id_with_provider, - encode_video_id_with_provider, ) # =========================================================================== # @@ -256,16 +252,6 @@ def test_encode__object_non_str_or_empty_id_unchanged(bad_id): assert resp.id == bad_id # untouched -def test_stamp_and_assert_video_owner_round_trip(): - encoded = encode_video_id_with_provider("req_1", "xai", "grok-imagine-video") - stamped = stamp_video_owner(encoded, "key-hash") - assert stamped != encoded - assert_video_owner(stamped, "key-hash") - with pytest.raises(ProxyException): - assert_video_owner(stamped, "other-key") - assert_video_owner(encoded, "other-key") - - def test_encode__object_without_id_attr_returned_unchanged(): resp = _Resp() out = encode_character_id_in_response(resp, "azure", "model-1")