fix(realtime): reject unsupported image URL schemes; document fetch/file-read footguns

decart/process/request.py

@@ -14,6 +14,16 @@ async def file_input_to_bytes(
 ) -> tuple[bytes, str]:
     """Convert various file input types to bytes asynchronously.
 
+    A ``str`` input is read from disk when it names an existing file, otherwise
+    fetched as an ``http(s)`` URL.
+
+    Security note: the local-file read and the URL fetch are conveniences meant
+    for *trusted* inputs. Do not pass an untrusted / user-supplied string here
+    from a server — a filesystem path is read from your server's disk (local
+    file disclosure) and an ``http(s)`` value is fetched from your server's
+    network position (SSRF — internal services, cloud metadata). For untrusted
+    input, resolve it yourself and pass ``bytes``.
+
     Args:
         input_data: The file input (bytes, Path, str, or file-like object)
         session: Reusable aiohttp session for URL fetching

decart/realtime/client.py

@@ -40,6 +40,24 @@ async def _image_to_base64(
     image: Union[bytes, str, Path],
     http_session: aiohttp.ClientSession,
 ) -> str:
+    """Resolve an image input to a raw base64 string (no ``data:`` prefix).
+
+    Accepted inputs:
+
+    - ``bytes`` — encoded directly.
+    - ``Path`` — read from the local filesystem, then encoded.
+    - ``str`` — interpreted by shape: a ``data:`` URL is decoded locally; an
+      ``http(s)`` URL is fetched and encoded; an existing local file path is
+      read from disk; anything else is assumed to already be raw base64.
+
+    Security note: the ``http(s)`` fetch and the local-file read are
+    conveniences meant for *trusted* inputs — your own code, local files, a CLI
+    or notebook. Do **not** pass an untrusted / user-supplied string here from a
+    server. An ``http(s)`` value is fetched from your server's network position
+    (SSRF — it can reach internal services and cloud-metadata endpoints), and a
+    filesystem path is read from your server's disk (local file disclosure).
+    For untrusted input, resolve it yourself and pass ``bytes``.
+    """
     if isinstance(image, Path):
         image_bytes, _ = await file_input_to_bytes(image, http_session)
         return base64.b64encode(image_bytes).decode("utf-8")
@@ -63,6 +81,14 @@ async def _image_to_base64(
             image_bytes, _ = await file_input_to_bytes(image, http_session)
             return base64.b64encode(image_bytes).decode("utf-8")
 
+        # A URL-shaped string with a scheme we don't handle (e.g. file://,
+        # ftp://, s3://, blob:) would otherwise be returned verbatim and sent
+        # onward as if it were base64, failing opaquely at the API. Reject it
+        # here instead. A single-character scheme is a Windows drive letter
+        # (e.g. "C:\\img.png"), not a URL; raw base64 has no scheme at all.
+        if len(parsed.scheme) > 1:
+            raise InvalidInputError(f"Unsupported image URL scheme: {parsed.scheme!r}")
+
         # Non-URL, non-file string — treat as raw base64 (matches TS SDK behavior)
         return image
 

tests/test_realtime_unit.py

@@ -628,3 +628,28 @@ def post(self, url, headers):
     ]
     assert room_info.livekit_url == "wss://livekit.example"
     assert room_info.session_id == "room-123"
+
+
+@pytest.mark.asyncio
+async def test_image_to_base64_passes_through_raw_base64():
+    from decart.realtime.client import _image_to_base64
+
+    b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAAAAAA6fptVAAAACklEQVR4nGNgAAAAAgAB"
+    assert await _image_to_base64(b64, AsyncMock()) == b64
+
+
+@pytest.mark.asyncio
+async def test_image_to_base64_decodes_data_url():
+    from decart.realtime.client import _image_to_base64
+
+    assert await _image_to_base64("data:image/png;base64,AAAA", AsyncMock()) == "AAAA"
+
+
+@pytest.mark.asyncio
+async def test_image_to_base64_rejects_unsupported_url_scheme():
+    from decart.realtime.client import _image_to_base64
+    from decart.errors import InvalidInputError
+
+    for value in ("file:///etc/passwd", "ftp://example.com/x.png", "blob:https://x/uuid"):
+        with pytest.raises(InvalidInputError):
+            await _image_to_base64(value, AsyncMock())