fix(sglang): validate diffusion input_reference and bound media fetches

components/src/dynamo/common/http/__init__.py

@@ -71,9 +71,11 @@ def get_default_client() -> HttpClient:
     return _default
 
 
-async def fetch_bytes(url, timeout, *, policy=None) -> bytes:
+async def fetch_bytes(url, timeout, *, policy=None, max_bytes=None) -> bytes:
     """Singleton-backed convenience wrapper over :meth:`HttpClient.fetch_bytes`."""
-    return await get_default_client().fetch_bytes(url, timeout, policy=policy)
+    return await get_default_client().fetch_bytes(
+        url, timeout, policy=policy, max_bytes=max_bytes
+    )
 
 
 async def close_http_client() -> None:

components/src/dynamo/common/http/aiohttp_client.py

@@ -17,13 +17,24 @@
 import aiohttp
 from yarl import URL
 
-from .base import HttpClient, HttpConnectionError, HttpStatusError, HttpTimeoutError
+from .base import (
+    HttpClient,
+    HttpConnectionError,
+    HttpStatusError,
+    HttpTimeoutError,
+    collect_capped,
+)
+from .url_validator import describe_error_detail, describe_media_source
 
 logger = logging.getLogger(__name__)
 
 
 _REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308})
 
+# Read granularity for the capped reader. Chunks are joined, so this only
+# bounds how far past the limit a single read can carry.
+_READ_CHUNK = 64 * 1024
+
 
 class AiohttpClient(HttpClient):
     """aiohttp-backed concrete client."""
@@ -72,30 +83,40 @@ async def _get_session(self) -> aiohttp.ClientSession:
                 )
         return self._session
 
-    async def _fetch_simple(self, url: str, timeout: float) -> bytes:
+    async def _fetch_simple(
+        self, url: str, timeout: float, *, max_bytes: Optional[int] = None
+    ) -> bytes:
         session = await self._get_session()
         client_timeout = self._effective_timeout(timeout)
         try:
             async with session.get(
                 url, timeout=client_timeout, allow_redirects=True
             ) as response:
                 response.raise_for_status()
-                return await response.read()
+                return await collect_capped(
+                    response.content.iter_chunked(_READ_CHUNK), url, max_bytes
+                )
         except aiohttp.ClientResponseError as e:
             raise HttpStatusError(e.status, e.message or "", url) from e
         except (asyncio.TimeoutError, aiohttp.ServerTimeoutError) as e:
-            raise HttpTimeoutError(f"Timeout loading {url}") from e
+            raise HttpTimeoutError(
+                f"Timeout loading {describe_media_source(url)}"
+            ) from e
         except (
             aiohttp.ClientConnectionError,
             aiohttp.ClientConnectorError,
             aiohttp.ServerDisconnectedError,
         ) as e:
-            raise HttpConnectionError(f"Connection error loading {url}: {e}") from e
+            raise HttpConnectionError(
+                f"Connection error loading {describe_media_source(url)}: {describe_error_detail(str(e))}"
+            ) from e
         except aiohttp.ClientError as e:
-            raise HttpConnectionError(f"HTTP error loading {url}: {e}") from e
+            raise HttpConnectionError(
+                f"HTTP error loading {describe_media_source(url)}: {describe_error_detail(str(e))}"
+            ) from e
 
     async def _fetch_body_or_redirect(
-        self, url: str, timeout: float
+        self, url: str, timeout: float, *, max_bytes: Optional[int] = None
     ) -> tuple[bytes | None, str | None]:
         session = await self._get_session()
         client_timeout = self._effective_timeout(timeout)
@@ -108,23 +129,39 @@ async def _fetch_body_or_redirect(
                     if location:
                         next_url = str(response.url.join(URL(location)))
                         return None, next_url
-                    return await response.read(), None
+                    return (
+                        await collect_capped(
+                            response.content.iter_chunked(_READ_CHUNK), url, max_bytes
+                        ),
+                        None,
+                    )
 
                 try:
                     response.raise_for_status()
                 except aiohttp.ClientResponseError as e:
                     raise HttpStatusError(e.status, e.message or "", url) from e
-                return await response.read(), None
+                return (
+                    await collect_capped(
+                        response.content.iter_chunked(_READ_CHUNK), url, max_bytes
+                    ),
+                    None,
+                )
         except (asyncio.TimeoutError, aiohttp.ServerTimeoutError) as e:
-            raise HttpTimeoutError(f"Timeout loading {url}") from e
+            raise HttpTimeoutError(
+                f"Timeout loading {describe_media_source(url)}"
+            ) from e
         except (
             aiohttp.ClientConnectionError,
             aiohttp.ClientConnectorError,
             aiohttp.ServerDisconnectedError,
         ) as e:
-            raise HttpConnectionError(f"Connection error loading {url}: {e}") from e
+            raise HttpConnectionError(
+                f"Connection error loading {describe_media_source(url)}: {describe_error_detail(str(e))}"
+            ) from e
         except aiohttp.ClientError as e:
-            raise HttpConnectionError(f"HTTP error loading {url}: {e}") from e
+            raise HttpConnectionError(
+                f"HTTP error loading {describe_media_source(url)}: {describe_error_detail(str(e))}"
+            ) from e
 
     async def close(self) -> None:
         async with self._lock:

components/src/dynamo/common/http/base.py

@@ -13,17 +13,27 @@
 
 import abc
 import asyncio
-from typing import Optional
+from typing import AsyncIterator, Optional
 
 from dynamo.common.configuration.groups.http_args import HttpConfigBase, from_env
 
 from .url_validator import (
     _MAX_REDIRECTS,
     UrlValidationError,
     UrlValidationPolicy,
+    describe_error_detail,
+    describe_media_source,
     validate_url,
 )
 
+# Longest a client-supplied reason may render as inside an HttpStatusError.
+# Matches ``dynamo.llm.exceptions.HttpError._MAX_MESSAGE_LENGTH``, which the
+# binding applies to the other class it forwards on a 4xx. Generous on purpose:
+# callers build real guidance here -- the trtllm decoder hint is ~480 characters
+# and its test allows 2000 -- so a tight bound silently deletes the actionable
+# part and leaves only a truncated prefix.
+_MAX_MESSAGE_LENGTH = 8192
+
 
 class HttpError(Exception):
     """Base class for all HTTP fetch failures."""
@@ -41,12 +51,46 @@ class HttpStatusError(HttpError):
     """Server responded with a non-2xx status."""
 
     def __init__(self, status: int, message: str, url: str) -> None:
-        super().__init__(f"HTTP {status} for {url}: {message}")
+        # Both halves are client-supplied: ``url`` directly, and ``message``
+        # because aiohttp's own ClientResponseError text repeats the URL. The video
+        # diffusion handler puts str(exc) in its response body.
+        #
+        # ``message`` is bounded in the *attribute*, not just in the rendered
+        # string: errors.rs::extract_http_like_error reads ``.status`` and
+        # ``.message`` off this class by name and, per the SECURITY note there,
+        # forwards ``.message`` verbatim to the client on a 4xx. Nothing in the
+        # rendered text reaches that path. ``url`` is not part of that protocol,
+        # so it keeps its full value for debugging.
+        message = describe_error_detail(message, _MAX_MESSAGE_LENGTH)
+        super().__init__(f"HTTP {status} for {describe_media_source(url)}: {message}")
         self.status = status
         self.message = message
         self.url = url
 
 
+async def collect_capped(
+    chunks: AsyncIterator[bytes], url: str, max_bytes: Optional[int]
+) -> bytes:
+    """Join ``chunks`` into one body, refusing to buffer past ``max_bytes``.
+
+    The check is per chunk rather than on the finished body: a declared
+    Content-Length is attacker-controlled and absent on a chunked response, and
+    a single capped read is not enough either — aiohttp's ``read(n)`` returns
+    *at most* n bytes, so a short read would look like a body under the limit.
+    """
+    out: list[bytes] = []
+    total = 0
+    async for chunk in chunks:
+        total += len(chunk)
+        if max_bytes is not None and total > max_bytes:
+            raise UrlValidationError(
+                f"Media exceeds the {max_bytes} byte download limit: "
+                f"{describe_media_source(url)}"
+            )
+        out.append(chunk)
+    return b"".join(out)
+
+
 class HttpClient(abc.ABC):
     """Backend-neutral HTTP client.
 
@@ -66,6 +110,7 @@ async def fetch_bytes(
         timeout: float,
         *,
         policy: Optional[UrlValidationPolicy] = None,
+        max_bytes: Optional[int] = None,
     ) -> bytes:
         """Fetch ``url`` and return the response body.
 
@@ -74,20 +119,29 @@ async def fetch_bytes(
 
         ``policy=None``: use the backend's built-in redirect handling.
 
+        ``max_bytes`` set: refuse a body larger than that while it is being
+        read, so an attacker-chosen URL cannot buffer an unbounded response.
+        Raises :class:`UrlValidationError` — it is a verdict on a
+        client-supplied source, like the redirect cap below.
+
         ``policy`` set: follow redirects manually and revalidate each
         hop against the policy via :func:`url_validator.validate_url`.
         This is the SSRF-safe path; raises :class:`UrlValidationError`
         if any hop fails or the chain exceeds ``_MAX_REDIRECTS``.
         """
         if policy is None:
-            return await self._fetch_simple(url, timeout)
-        return await self._fetch_with_revalidation(url, timeout, policy)
+            return await self._fetch_simple(url, timeout, max_bytes=max_bytes)
+        return await self._fetch_with_revalidation(
+            url, timeout, policy, max_bytes=max_bytes
+        )
 
     async def _fetch_with_revalidation(
         self,
         url: str,
         timeout: float,
         policy: UrlValidationPolicy,
+        *,
+        max_bytes: Optional[int] = None,
     ) -> bytes:
         """Manual redirect loop with per-hop SSRF validation (backend-neutral)."""
         current = url
@@ -97,30 +151,36 @@ async def _fetch_with_revalidation(
             await validate_url(current, policy)
             visited.append(current)
 
-            body, redirect_to = await self._fetch_body_or_redirect(current, timeout)
+            body, redirect_to = await self._fetch_body_or_redirect(
+                current, timeout, max_bytes=max_bytes
+            )
 
             if redirect_to is None:
                 if body is None:
                     raise HttpError(
-                        f"Backend returned (None, None) for {current}; "
+                        f"Backend returned (None, None) for {describe_media_source(current)}; "
                         "expected bytes on a terminal (2xx or 3xx-without-Location) response"
                     )
                 return body
 
             if hops_remaining <= 0:
+                # ``visited`` holds attacker-chosen URLs; describe each one.
+                chain = [describe_media_source(hop) for hop in visited]
                 raise UrlValidationError(
-                    f"Too many redirects (max={_MAX_REDIRECTS}); chain={visited}"
+                    f"Too many redirects (max={_MAX_REDIRECTS}); chain={chain}"
                 )
             hops_remaining -= 1
             current = redirect_to
 
     @abc.abstractmethod
-    async def _fetch_simple(self, url: str, timeout: float) -> bytes:
+    async def _fetch_simple(
+        self, url: str, timeout: float, *, max_bytes: Optional[int] = None
+    ) -> bytes:
         """Backend's native redirect-following GET (no SSRF policy applied)."""
 
     @abc.abstractmethod
     async def _fetch_body_or_redirect(
-        self, url: str, timeout: float
+        self, url: str, timeout: float, *, max_bytes: Optional[int] = None
     ) -> tuple[bytes | None, str | None]:
         """Single hop with redirects disabled.