Merge 999d23dfce into 1df25e26cf

litellm/litellm_core_utils/llm_request_utils.py

@@ -2,6 +2,7 @@ from collections.abc import Mapping
 from typing import Final
 
 import litellm
+from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
 
 
 def _form_field_value(value: object) -> str:
@@ -13,18 +14,31 @@ def _form_field_value(value: object) -> str:
 
 
 def _flatten_form_field(key: str, value: object) -> tuple[tuple[str, str], ...]:
-    if isinstance(value, Mapping):
-        return tuple(
-            item for subkey, subvalue in value.items() for item in _flatten_form_field(f"{key}[{subkey}]", subvalue)
-        )
-    if isinstance(value, (list, tuple)):
-        return tuple(item for entry in value for item in _flatten_form_field(f"{key}[]", entry))
-    if value is None:
-        return ()
-    serialized: Final = _form_field_value(value)
-    if not serialized:
-        return ()
-    return ((key, serialized),)
+    pending_fields: Final[  # mutable-ok: depth-capped stack walks nested JSON into multipart names
+        list[tuple[str, object, int]]
+    ] = [  # mutable-ok: depth-capped stack walks nested JSON into multipart names
+        (key, value, 0)
+    ]
+    flat_fields: Final[list[tuple[str, str]]] = []  # mutable-ok: local accumulator
+    while pending_fields:
+        current_key, current_value, depth = pending_fields.pop()
+        if depth > DEFAULT_MAX_RECURSE_DEPTH:
+            raise ValueError("form field nesting exceeds max depth")
+        if isinstance(current_value, Mapping):
+            pending_fields.extend(
+                (f"{current_key}[{subkey}]", subvalue, depth + 1)
+                for subkey, subvalue in reversed(tuple(current_value.items()))
+            )
+            continue
+        if isinstance(current_value, (list, tuple)):
+            pending_fields.extend((f"{current_key}[]", entry, depth + 1) for entry in reversed(tuple(current_value)))
+            continue
+        if current_value is None:
+            continue
+        serialized = _form_field_value(current_value)
+        if serialized:
+            flat_fields.append((current_key, serialized))
+    return tuple(flat_fields)
 
 
 def _is_form_scalar(value: object) -> bool:
@@ -32,23 +46,36 @@ def _is_form_scalar(value: object) -> bool:
 
 
 def _flatten_form_data_field(key: str, value: object) -> tuple[tuple[str, str | tuple[str, ...]], ...]:
-    if isinstance(value, Mapping):
-        return tuple(
-            item
-            for subkey, subvalue in value.items()
-            for item in _flatten_form_data_field(f"{key}[{subkey}]", subvalue)
-        )
-    if isinstance(value, (list, tuple)):
-        if all(_is_form_scalar(entry) for entry in value):
-            serialized_fields: Final = tuple(field for entry in value if (field := _form_field_value(entry)))
-            return ((key, serialized_fields),) if serialized_fields else ()
-        return tuple(item for entry in value for item in _flatten_form_data_field(f"{key}[]", entry))
-    if value is None:
-        return ()
-    serialized: Final = _form_field_value(value)
-    if not serialized:
-        return ()
-    return ((key, serialized),)
+    pending_fields: Final[  # mutable-ok: depth-capped stack walks nested JSON into multipart names
+        list[tuple[str, object, int]]
+    ] = [  # mutable-ok: depth-capped stack walks nested JSON into multipart names
+        (key, value, 0)
+    ]
+    flat_fields: Final[list[tuple[str, str | tuple[str, ...]]]] = []  # mutable-ok: local accumulator
+    while pending_fields:
+        current_key, current_value, depth = pending_fields.pop()
+        if depth > DEFAULT_MAX_RECURSE_DEPTH:
+            raise ValueError("form field nesting exceeds max depth")
+        if isinstance(current_value, Mapping):
+            pending_fields.extend(
+                (f"{current_key}[{subkey}]", subvalue, depth + 1)
+                for subkey, subvalue in reversed(tuple(current_value.items()))
+            )
+            continue
+        if isinstance(current_value, (list, tuple)):
+            if all(_is_form_scalar(entry) for entry in current_value):
+                serialized_fields = tuple(field for entry in current_value if (field := _form_field_value(entry)))
+                if serialized_fields:
+                    flat_fields.append((current_key, serialized_fields))
+                continue
+            pending_fields.extend((f"{current_key}[]", entry, depth + 1) for entry in reversed(tuple(current_value)))
+            continue
+        if current_value is None:
+            continue
+        serialized = _form_field_value(current_value)
+        if serialized:
+            flat_fields.append((current_key, serialized))
+    return tuple(flat_fields)
 
 
 def flatten_form_field_values(*sources: Mapping[str, object] | None) -> tuple[tuple[str, str | tuple[str, ...]], ...]:

litellm/llms/hosted_vllm/videos/__init__.py

@@ -0,0 +1,9 @@
+from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
+
+from .transformation import HostedVLLMVideoConfig
+
+__all__ = ("HostedVLLMVideoConfig",)
+
+
+def get_hosted_vllm_video_config(model: str | None) -> BaseVideoConfig:
+    return HostedVLLMVideoConfig()

litellm/llms/hosted_vllm/videos/transformation.py

@@ -0,0 +1,206 @@
+"""Video generation for Hosted VLLM (vLLM-Omni OpenAI-compatible /v1/videos)."""
+
+import json
+from collections.abc import Mapping
+from io import BufferedReader
+from types import MappingProxyType
+from typing import Final
+from urllib.parse import urlparse
+
+from httpx._types import FileTypes, RequestFiles
+
+from litellm.images.utils import ImageEditRequestUtils
+from litellm.litellm_core_utils.url_utils import SSRFError, validate_url
+from litellm.llms.openai.videos.transformation import OpenAIVideoConfig
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.videos.main import VideoCreateOptionalRequestParams
+
+_EXCLUDED_FORM_KEYS: Final = frozenset(
+    {
+        "model",
+        "prompt",
+        "extra_headers",
+        "extra_query",
+        "extra_body",
+        "timeout",
+        "custom_llm_provider",
+        "input_reference",
+        "characters",
+    }
+)
+
+_VLLM_OMNI_VIDEO_PARAMS: Final = (
+    "image_reference",
+    "video_reference",
+    "audio_reference",
+    "width",
+    "height",
+    "num_frames",
+    "fps",
+    "num_inference_steps",
+    "guidance_scale",
+    "guidance_scale_2",
+    "boundary_ratio",
+    "flow_shift",
+    "true_cfg_scale",
+    "seed",
+    "generate_sound",
+    "sound_duration",
+    "negative_prompt",
+    "enable_frame_interpolation",
+    "frame_interpolation_exp",
+    "frame_interpolation_scale",
+    "frame_interpolation_model_path",
+    "lora",
+    "extra_params",
+    "aspect_ratio",
+)
+
+_REFERENCE_URL_KEYS: Final = MappingProxyType(
+    {
+        "image_reference": "image_url",
+        "video_reference": "video_url",
+        "audio_reference": "audio_url",
+    }
+)
+
+
+def _serialize_form_value(value: object) -> str:
+    if isinstance(value, str):
+        return value
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (Mapping, list, tuple)):
+        return json.dumps(value)
+    return str(value)
+
+
+def _maybe_json(value: object) -> object:
+    if not isinstance(value, str):
+        return value
+    stripped: Final = value.strip()
+    if not stripped or stripped[0] not in "{[":
+        return value
+    return json.loads(stripped)
+
+
+def _reject_unsafe_media_url(url: str) -> None:
+    scheme: Final = urlparse(url).scheme.lower()
+    if scheme in ("", "data"):
+        return
+    if scheme not in ("http", "https"):
+        raise SSRFError(f"URL scheme '{scheme}' is not allowed")
+    validate_url(url)
+
+
+def _reject_unsafe_urls_in_item(url_key: str, item: object) -> None:
+    if not isinstance(item, Mapping):
+        return
+    url: Final = item.get(url_key)
+    if isinstance(url, str):
+        _reject_unsafe_media_url(url)
+
+
+def _reject_unsafe_media_urls(field_name: str, value: object) -> None:
+    url_key: Final = _REFERENCE_URL_KEYS.get(field_name)
+    if url_key is None:
+        return
+    parsed: Final = _maybe_json(value)
+    if isinstance(parsed, list):
+        for item in parsed:
+            _reject_unsafe_urls_in_item(url_key, item)
+        return
+    if isinstance(parsed, Mapping):
+        _reject_unsafe_urls_in_item(url_key, parsed)
+
+
+def _form_value(key: str, value: object) -> str:
+    _reject_unsafe_media_urls(key, value)
+    return _serialize_form_value(value)
+
+
+def _input_reference_file(reference: object) -> tuple[str, FileTypes]:
+    content_type: Final = ImageEditRequestUtils.get_image_content_type(reference)
+    if isinstance(reference, BufferedReader):
+        return ("input_reference", (reference.name, reference, content_type))
+    return ("input_reference", ("input_reference.png", reference, content_type))
+
+
+class HostedVLLMVideoConfig(OpenAIVideoConfig):
+    """
+    vLLM-Omni videos API is OpenAI-compatible but requires multipart/form-data.
+
+    https://docs.vllm.ai/projects/vllm-omni/en/latest/serving/videos_api/
+    """
+
+    def get_supported_openai_params(self, model: str) -> list:  # mutable-ok: BaseVideoConfig contract
+        return [  # mutable-ok: BaseVideoConfig returns list
+            *super().get_supported_openai_params(model),
+            *_VLLM_OMNI_VIDEO_PARAMS,
+        ]
+
+    def map_openai_params(
+        self,
+        video_create_optional_params: VideoCreateOptionalRequestParams,
+        model: str,
+        drop_params: bool,
+    ) -> dict:  # mutable-ok: BaseVideoConfig contract; extra_body merge mutates this dict
+        return {  # mutable-ok: VideoGenerationRequestUtils.update/pop extra_body onto this mapping
+            key: value for key, value in video_create_optional_params.items() if value is not None
+        }
+
+    def validate_environment(
+        self,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+        model: str,
+        api_key: str | None = None,
+        litellm_params: GenericLiteLLMParams | None = None,
+    ) -> dict:  # mutable-ok: BaseVideoConfig contract
+        resolved_key: Final = (
+            (litellm_params.api_key if litellm_params is not None else None)
+            or api_key
+            or get_secret_str("HOSTED_VLLM_API_KEY")
+            or "fake-api-key"
+        )
+        return {**headers, "Authorization": f"Bearer {resolved_key}"}  # mutable-ok: httpx headers are a dict
+
+    def get_complete_url(
+        self,
+        model: str,
+        api_base: str | None,
+        litellm_params: dict,  # mutable-ok: BaseVideoConfig contract
+    ) -> str:
+        resolved_api_base: Final = api_base or get_secret_str("HOSTED_VLLM_API_BASE")
+        if resolved_api_base is None:
+            raise ValueError(
+                "api_base not set for Hosted VLLM videos API. "
+                "Set via api_base parameter or HOSTED_VLLM_API_BASE environment variable"
+            )
+        trimmed: Final = resolved_api_base.rstrip("/")
+        if trimmed.endswith("/v1"):
+            return f"{trimmed}/videos"
+        return f"{trimmed}/v1/videos"
+
+    def transform_video_create_request(
+        self,
+        model: str,
+        prompt: str,
+        api_base: str,
+        video_create_optional_request_params: dict,  # mutable-ok: BaseVideoConfig contract
+        litellm_params: GenericLiteLLMParams,
+        headers: dict,  # mutable-ok: BaseVideoConfig contract
+    ) -> tuple[dict, RequestFiles, str]:  # mutable-ok: BaseVideoConfig contract
+        data: Final = {  # mutable-ok: BaseVideoConfig contract returns a data dict
+            "model": model,
+            "prompt": prompt,
+            **{  # mutable-ok: spread remaining Omni form fields into that data dict
+                key: _form_value(key, value)
+                for key, value in video_create_optional_request_params.items()
+                if key not in _EXCLUDED_FORM_KEYS and value is not None
+            },
+        }
+        input_reference: Final = video_create_optional_request_params.get("input_reference")
+        if input_reference is None:
+            return data, (), api_base
+        return data, (_input_reference_file(input_reference),), api_base

litellm/provider_endpoints_support_backup.json

@@ -1180,7 +1180,8 @@
         "files": true,
         "rerank": true,
         "a2a": true,
-        "interactions": true
+        "interactions": true,
+        "video_generations": true
       }
     },
     "huggingface": {

litellm/utils.py

@@ -9038,6 +9038,10 @@ class ProviderConfigManager:
             from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig
 
             return RunwayMLVideoConfig()
+        elif LlmProviders.HOSTED_VLLM == provider:
+            from litellm.llms.hosted_vllm.videos import get_hosted_vllm_video_config
+
+            return get_hosted_vllm_video_config(model)
         return None
 
     @staticmethod

provider_endpoints_support.json

@@ -1277,7 +1277,8 @@
         "files": true,
         "rerank": true,
         "a2a": true,
-        "interactions": true
+        "interactions": true,
+        "video_generations": true
       }
     },
     "huggingface": {

tests/test_litellm/litellm_core_utils/test_llm_request_utils.py

@@ -1,4 +1,5 @@
 import httpx
+import pytest
 
 from litellm.litellm_core_utils.llm_request_utils import (
     flatten_form_field_values,
@@ -80,9 +81,7 @@ def test_flatten_form_field_values_later_source_wins_on_collision():
 
 
 def test_flatten_form_field_values_keeps_scalar_lists_as_repeated_fields():
-    assert flatten_form_field_values(
-        {"loras": ["a", "b", "c"], "generation_config": {"tags": [1, 2]}, "seed": 42}
-    ) == (
+    assert flatten_form_field_values({"loras": ["a", "b", "c"], "generation_config": {"tags": [1, 2]}, "seed": 42}) == (
         ("loras", ("a", "b", "c")),
         ("generation_config[tags]", ("1", "2")),
         ("seed", "42"),
@@ -97,3 +96,12 @@ def test_flatten_form_field_values_scalar_list_survives_update_into_multipart():
 
     assert names.count("loras") == 2
     assert names.count("model") == 1
+
+
+def test_flatten_form_field_values_rejects_over_deep_nesting():
+    nested: object = "leaf"
+    for _ in range(102):
+        nested = {"k": nested}
+    assert isinstance(nested, dict)
+    with pytest.raises(ValueError, match="max depth"):
+        flatten_form_field_values(nested)

tests/test_litellm/llms/hosted_vllm/videos/test_hosted_vllm_video_transformation.py

@@ -0,0 +1,312 @@
+"""Tests for hosted_vllm video generation (vLLM-Omni /v1/videos)."""
+
+import json
+from io import BytesIO
+
+import httpx
+import pytest
+
+import litellm
+from litellm.litellm_core_utils.url_utils import SSRFError
+from litellm.llms.custom_httpx.http_handler import HTTPHandler
+from litellm.llms.hosted_vllm.videos import get_hosted_vllm_video_config
+from litellm.llms.hosted_vllm.videos.transformation import (
+    HostedVLLMVideoConfig,
+    _serialize_form_value,
+)
+from litellm.types.router import GenericLiteLLMParams
+from litellm.types.utils import LlmProviders
+from litellm.types.videos.main import VideoObject
+from litellm.utils import ProviderConfigManager
+
+
+def test_provider_config_registration():
+    config = ProviderConfigManager.get_provider_video_config(
+        model="hosted_vllm/MiniMax-H3",
+        provider=LlmProviders.HOSTED_VLLM,
+    )
+
+    assert config is not None
+    assert isinstance(config, HostedVLLMVideoConfig)
+    assert isinstance(get_hosted_vllm_video_config("MiniMax-H3"), HostedVLLMVideoConfig)
+
+
+def test_get_complete_url_appends_videos():
+    config = HostedVLLMVideoConfig()
+
+    assert (
+        config.get_complete_url(model="MiniMax-H3", api_base="http://localhost:8091", litellm_params={})
+        == "http://localhost:8091/v1/videos"
+    )
+    assert (
+        config.get_complete_url(model="MiniMax-H3", api_base="http://localhost:8091/v1", litellm_params={})
+        == "http://localhost:8091/v1/videos"
+    )
+    assert (
+        config.get_complete_url(model="MiniMax-H3", api_base="http://localhost:8091/v1/", litellm_params={})
+        == "http://localhost:8091/v1/videos"
+    )
+
+
+def test_get_complete_url_requires_api_base():
+    config = HostedVLLMVideoConfig()
+
+    with pytest.raises(ValueError, match="api_base not set"):
+        config.get_complete_url(model="MiniMax-H3", api_base=None, litellm_params={})
+
+
+def test_validate_environment_defaults_to_fake_api_key():
+    config = HostedVLLMVideoConfig()
+
+    headers = config.validate_environment(
+        headers={},
+        model="MiniMax-H3",
+        litellm_params=GenericLiteLLMParams(),
+    )
+
+    assert headers.get("Authorization") == "Bearer fake-api-key"
+
+
+def test_validate_environment_uses_provided_api_key():
+    config = HostedVLLMVideoConfig()
+
+    headers = config.validate_environment(
+        headers={"X-Test": "1"},
+        model="MiniMax-H3",
+        litellm_params=GenericLiteLLMParams(api_key="my-custom-key"),
+    )
+
+    assert headers.get("Authorization") == "Bearer my-custom-key"
+    assert headers.get("X-Test") == "1"
+
+
+def test_transform_video_create_request_uses_multipart_form_fields():
+    """vLLM-Omni rejects JSON create bodies. Extra Omni fields must be form parts."""
+    config = HostedVLLMVideoConfig()
+    extra_params = {"task": "t2va", "duration": 10.0, "audio_flow_shift": 3.0}
+
+    data, files, url = config.transform_video_create_request(
+        model="MiniMax-H3",
+        prompt="three cats march into a bedroom playing tiny brass instruments",
+        api_base="http://localhost:8091/v1/videos",
+        video_create_optional_request_params={
+            "width": 1280,
+            "height": 720,
+            "fps": 24,
+            "num_inference_steps": 20,
+            "flow_shift": 12,
+            "seed": 1101,
+            "aspect_ratio": "16:9",
+            "extra_params": extra_params,
+            "extra_headers": {"X-Ignored": "yes"},
+        },
+        litellm_params=GenericLiteLLMParams(),
+        headers={},
+    )
+
+    assert url == "http://localhost:8091/v1/videos"
+    assert files == ()
+    assert data["model"] == "MiniMax-H3"
+    assert data["prompt"] == "three cats march into a bedroom playing tiny brass instruments"
+    assert data["width"] == "1280"
+    assert data["height"] == "720"
+    assert data["fps"] == "24"
+    assert data["num_inference_steps"] == "20"
+    assert data["flow_shift"] == "12"
+    assert data["seed"] == "1101"
+    assert data["aspect_ratio"] == "16:9"
+    assert json.loads(data["extra_params"]) == extra_params
+    assert "extra_headers" not in data
+
+
+def test_transform_video_create_request_keeps_openai_size_and_seconds():
+    config = HostedVLLMVideoConfig()
+
+    data, files, _ = config.transform_video_create_request(
+        model="Wan2.2",
+        prompt="a mountain lake at sunrise",
+        api_base="http://localhost:8091/v1/videos",
+        video_create_optional_request_params={"seconds": "8", "size": "1280x720"},
+        litellm_params=GenericLiteLLMParams(),
+        headers={},
+    )
+
+    assert files == ()
+    assert data["seconds"] == "8"
+    assert data["size"] == "1280x720"
+
+
+def test_transform_video_create_request_attaches_input_reference_file():
+    config = HostedVLLMVideoConfig()
+    reference = BytesIO(b"fake-png")
+    reference.name = "input.png"
+
+    data, files, _ = config.transform_video_create_request(
+        model="Wan2.2",
+        prompt="animate this image",
+        api_base="http://localhost:8091/v1/videos",
+        video_create_optional_request_params={"input_reference": reference, "width": 832},
+        litellm_params=GenericLiteLLMParams(),
+        headers={},
+    )
+
+    assert data["width"] == "832"
+    assert "input_reference" not in data
+    reference_parts = [value for name, value in files if name == "input_reference"]
+    assert len(reference_parts) == 1
+    filename, content, content_type = reference_parts[0]
+    assert filename == "input_reference.png"
+    assert content is reference
+    assert content_type == "image/png"
+
+
+def test_serialize_form_value_does_not_quote_plain_strings():
+    assert _serialize_form_value("16:9") == "16:9"
+    assert _serialize_form_value(True) == "true"
+    assert _serialize_form_value({"task": "t2va"}) == json.dumps({"task": "t2va"})
+
+
+def test_map_openai_params_passes_through_omni_fields():
+    config = HostedVLLMVideoConfig()
+
+    mapped = config.map_openai_params(
+        video_create_optional_params={
+            "width": 1280,
+            "extra_params": {"task": "t2va"},
+            "aspect_ratio": "16:9",
+            "extra_body": None,
+        },
+        model="MiniMax-H3",
+        drop_params=False,
+    )
+
+    assert mapped["width"] == 1280
+    assert mapped["extra_params"] == {"task": "t2va"}
+    assert mapped["aspect_ratio"] == "16:9"
+    assert "extra_body" not in mapped
+
+
+def test_get_supported_openai_params_includes_omni_extensions():
+    config = HostedVLLMVideoConfig()
+    supported = config.get_supported_openai_params("MiniMax-H3")
+
+    assert "prompt" in supported
+    assert "input_reference" in supported
+    assert "width" in supported
+    assert "extra_params" in supported
+    assert "aspect_ratio" in supported
+    assert "image_reference" in supported
+    assert "audio_reference" in supported
+
+
+def _http_handler_for(handler) -> HTTPHandler:
+    return HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(handler)))
+
+
+def test_video_generation_posts_multipart_not_json():
+    captured: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured.append(request)
+        return httpx.Response(
+            200,
+            json={
+                "id": "video-123",
+                "object": "video",
+                "status": "queued",
+                "created_at": 1701234567,
+            },
+        )
+
+    response = litellm.video_generation(
+        model="hosted_vllm/MiniMax-H3",
+        prompt="three cats march into a bedroom playing tiny brass instruments",
+        api_base="http://localhost:8091",
+        api_key="test-key",
+        client=_http_handler_for(handler),
+        extra_body={
+            "width": 1280,
+            "height": 720,
+            "fps": 24,
+            "extra_params": {"task": "t2va", "duration": 10.0},
+        },
+    )
+
+    assert isinstance(response, VideoObject)
+    assert response.status == "queued"
+    assert len(captured) == 1
+    request = captured[0]
+    assert str(request.url) == "http://localhost:8091/v1/videos"
+    assert request.headers["authorization"] == "Bearer test-key"
+    body = request.content
+    assert b'name="prompt"' in body
+    assert b"three cats march into a bedroom playing tiny brass instruments" in body
+    assert b'name="width"' in body
+    assert b"1280" in body
+    assert b'name="extra_params"' in body
+    assert b"t2va" in body
+    assert request.headers.get("content-type", "").startswith("multipart/form-data")
+
+
+def test_http_image_reference_is_forwarded_not_downloaded():
+    config = HostedVLLMVideoConfig()
+    data, files, _ = config.transform_video_create_request(
+        model="MiniMax-H3",
+        prompt="a person singing",
+        api_base="http://localhost:8091/v1/videos",
+        video_create_optional_request_params={
+            "image_reference": {"image_url": "http://1.1.1.1/face.png"},
+        },
+        litellm_params=GenericLiteLLMParams(),
+        headers={},
+    )
+
+    assert files == ()
+    payload = json.loads(data["image_reference"])
+    assert payload["image_url"] == "http://1.1.1.1/face.png"
+
+
+def test_data_url_image_reference_is_forwarded():
+    data_url = "data:image/png;base64,AAAA"
+    config = HostedVLLMVideoConfig()
+    data, files, _ = config.transform_video_create_request(
+        model="MiniMax-H3",
+        prompt="a person singing",
+        api_base="http://localhost:8091/v1/videos",
+        video_create_optional_request_params={"image_reference": {"image_url": data_url}},
+        litellm_params=GenericLiteLLMParams(),
+        headers={},
+    )
+
+    assert files == ()
+    assert json.loads(data["image_reference"])["image_url"] == data_url
+
+
+def test_metadata_url_in_image_reference_is_rejected():
+    config = HostedVLLMVideoConfig()
+    with pytest.raises(SSRFError, match="blocked address"):
+        config.transform_video_create_request(
+            model="MiniMax-H3",
+            prompt="a person singing",
+            api_base="http://localhost:8091/v1/videos",
+            video_create_optional_request_params={
+                "image_reference": {"image_url": "http://169.254.169.254/latest/meta-data/"},
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )
+
+
+def test_file_scheme_media_reference_is_rejected():
+    config = HostedVLLMVideoConfig()
+    with pytest.raises(SSRFError, match="scheme"):
+        config.transform_video_create_request(
+            model="MiniMax-H3",
+            prompt="a person singing",
+            api_base="http://localhost:8091/v1/videos",
+            video_create_optional_request_params={
+                "video_reference": {"video_url": "file:///etc/passwd"},
+            },
+            litellm_params=GenericLiteLLMParams(),
+            headers={},
+        )