chore(proxy): strict media-type match for form bodies

``_read_request_body`` and ``get_request_body`` routed on
``"form" in content_type`` / ``"multipart/form-data" in content_type``,
which match any header containing the literal — ``application/form-json``,
``multiform/anything``, ``application/json; xform=1``. Starlette's
``request.form()`` returns an empty ``FormData`` for any non-canonical
type without consuming the body, so the auth-time pre-read saw ``{}``
and skipped the banned-param check while the handler's later
``request.body()`` saw the original JSON payload.

Parse the media type per RFC 7231 (substring before ``;``, trimmed,
lowercased) and accept only ``application/x-www-form-urlencoded`` and
``multipart/form-data``. Replace both substring sites with the shared
``_is_form_content_type`` helper.

Tests pin: case/whitespace/charset variants of the two real types
match; ``application/form-json`` and similar substring-match traps
fall through to the JSON parse path; real form POSTs continue to
route through ``request.form()``.

litellm/proxy/common_utils/http_parsing_utils.py

@@ -13,6 +13,25 @@ from litellm.proxy.common_utils.callback_utils import (
 from litellm.types.router import Deployment
 
 
+_FORM_CONTENT_TYPES: frozenset[str] = frozenset(
+    {"application/x-www-form-urlencoded", "multipart/form-data"}
+)
+
+
+def _is_form_content_type(content_type: str) -> bool:
+    """
+    True iff Starlette's ``request.form()`` will actually parse this body.
+
+    Substring matching ``"form"`` is unsafe: ``request.form()`` returns empty
+    ``FormData`` for non-canonical types without consuming the body, leaving
+    the auth-time pre-read and the handler's read seeing different payloads.
+    """
+    if not content_type:
+        return False
+    media_type = content_type.split(";", 1)[0].strip().lower()
+    return media_type in _FORM_CONTENT_TYPES
+
+
 async def _read_request_body(request: Optional[Request]) -> Dict:
     """
     Safely read the request body and parse it as JSON.
@@ -37,7 +56,7 @@ async def _read_request_body(request: Optional[Request]) -> Dict:
         _request_headers: dict = _safe_get_request_headers(request=request)
         content_type = _request_headers.get("content-type", "")
 
-        if "form" in content_type:
+        if _is_form_content_type(content_type):
             parsed_body = dict(await request.form())
             if "metadata" in parsed_body and isinstance(parsed_body["metadata"], str):
                 parsed_body["metadata"] = json.loads(parsed_body["metadata"])
@@ -306,18 +325,13 @@ async def get_request_body(request: Request) -> Dict[str, Any]:
     Read the request body and parse it as JSON.
     """
     if request.method == "POST":
-        if request.headers.get("content-type", "") == "application/json":
+        content_type = request.headers.get("content-type", "")
+        if content_type.split(";", 1)[0].strip().lower() == "application/json":
             return await _read_request_body(request)
-        elif "multipart/form-data" in request.headers.get(
-            "content-type", ""
-        ) or "application/x-www-form-urlencoded" in request.headers.get(
-            "content-type", ""
-        ):
+        elif _is_form_content_type(content_type):
             return await get_form_data(request)
         else:
-            raise ValueError(
-                f"Unsupported content type: {request.headers.get('content-type')}"
-            )
+            raise ValueError(f"Unsupported content type: {content_type}")
     return {}
 
 

tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py

@@ -16,6 +16,7 @@ sys.path.insert(
 import litellm
 from litellm.proxy._types import ProxyException
 from litellm.proxy.common_utils.http_parsing_utils import (
+    _is_form_content_type,
     _read_request_body,
     _safe_get_request_headers,
     _safe_get_request_parsed_body,
@@ -853,3 +854,76 @@ class TestGetTagsFromRequestBodyStringCoerce:
 
         tags = get_tags_from_request_body({"metadata": {"tags": ["x"]}})
         assert tags == ["x"]
+
+
+class TestIsFormContentType:
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "application/x-www-form-urlencoded",
+            "multipart/form-data",
+            "multipart/form-data; boundary=----WebKitFormBoundary",
+            "Application/X-WWW-Form-Urlencoded",
+            "  multipart/form-data  ",
+            "application/x-www-form-urlencoded; charset=utf-8",
+        ],
+    )
+    def test_form_types_match(self, content_type):
+        assert _is_form_content_type(content_type) is True
+
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "",
+            "application/json",
+            "application/json; charset=utf-8",
+            "application/form-json",
+            "multiform/anything",
+            "application/json; xform=1",
+            "application/xml-with-form-data-but-not-actually",
+            "text/plain",
+            "form",
+        ],
+    )
+    def test_non_form_types_rejected(self, content_type):
+        assert _is_form_content_type(content_type) is False
+
+
+class TestReadRequestBodyNonCanonicalContentType:
+    """A JSON body with a ``"form"``-substring Content-Type must parse as JSON."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "application/form-json",
+            "application/json; xform=1",
+            "multiform/anything",
+        ],
+    )
+    async def test_json_body_with_formlike_content_type_parses_as_json(
+        self, content_type
+    ):
+        payload = {"user_config": {"model_list": []}, "model": "x"}
+
+        mock_request = MagicMock()
+        mock_request.body = AsyncMock(return_value=orjson.dumps(payload))
+        mock_request.form = AsyncMock(return_value={})
+        mock_request.headers = {"content-type": content_type}
+        mock_request.scope = {}
+
+        result = await _read_request_body(mock_request)
+        assert result == payload
+        mock_request.form.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_real_form_post_still_parsed_as_form(self):
+        mock_request = MagicMock()
+        mock_request.form = AsyncMock(return_value={"k": "v"})
+        mock_request.body = AsyncMock(return_value=b"")
+        mock_request.headers = {"content-type": "application/x-www-form-urlencoded"}
+        mock_request.scope = {}
+
+        result = await _read_request_body(mock_request)
+        assert result == {"k": "v"}
+        mock_request.form.assert_awaited_once()