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 <cursoragent@cursor.com>

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}
 

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

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):

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)

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")