From 56eb5ef55097f7e04c53cafc5d94d7ccd9fceef0 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 14 May 2026 14:52:04 +0000
Subject: [PATCH] chore(proxy): extract _is_json_content_type symmetric helper

Mirror ``_is_form_content_type`` for the JSON branch of
``get_request_body`` so both classifications share the same media-type
normalisation (strip params, trim, lowercase) and any future change
to the parsing rules has one place to update.

Adds tests for ``_is_json_content_type`` and for ``get_request_body``
covering the canonical JSON / form / unsupported / non-POST paths.
---
 .../proxy/common_utils/http_parsing_utils.py  | 19 +++--
 .../common_utils/test_http_parsing_utils.py   | 71 +++++++++++++++++++
 2 files changed, 85 insertions(+), 5 deletions(-)

diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py
index d9bfc7659b1..60c3ee43dd4 100644
--- a/litellm/proxy/common_utils/http_parsing_utils.py
+++ b/litellm/proxy/common_utils/http_parsing_utils.py
@@ -18,6 +18,13 @@ _FORM_CONTENT_TYPES: frozenset[str] = frozenset(
 )
 
 
+def _normalize_media_type(content_type: str) -> str:
+    """Return the bare media type per RFC 7231: strip params, trim, lowercase."""
+    if not content_type:
+        return ""
+    return content_type.split(";", 1)[0].strip().lower()
+
+
 def _is_form_content_type(content_type: str) -> bool:
     """
     True iff Starlette's ``request.form()`` will actually parse this body.
@@ -26,10 +33,12 @@ def _is_form_content_type(content_type: str) -> bool:
     ``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
+    return _normalize_media_type(content_type) in _FORM_CONTENT_TYPES
+
+
+def _is_json_content_type(content_type: str) -> bool:
+    """True iff the body should be parsed as JSON."""
+    return _normalize_media_type(content_type) == "application/json"
 
 
 async def _read_request_body(request: Optional[Request]) -> Dict:
@@ -326,7 +335,7 @@ async def get_request_body(request: Request) -> Dict[str, Any]:
     """
     if request.method == "POST":
         content_type = request.headers.get("content-type", "")
-        if content_type.split(";", 1)[0].strip().lower() == "application/json":
+        if _is_json_content_type(content_type):
             return await _read_request_body(request)
         elif _is_form_content_type(content_type):
             return await get_form_data(request)
diff --git a/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py b/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py
index 3c6ef56461d..8be125dcf11 100644
--- a/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py
+++ b/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py
@@ -17,6 +17,7 @@ import litellm
 from litellm.proxy._types import ProxyException
 from litellm.proxy.common_utils.http_parsing_utils import (
     _is_form_content_type,
+    _is_json_content_type,
     _read_request_body,
     _safe_get_request_headers,
     _safe_get_request_parsed_body,
@@ -927,3 +928,73 @@ class TestReadRequestBodyNonCanonicalContentType:
         result = await _read_request_body(mock_request)
         assert result == {"k": "v"}
         mock_request.form.assert_awaited_once()
+
+
+class TestIsJsonContentType:
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "application/json",
+            "application/json; charset=utf-8",
+            "Application/JSON",
+            "  application/json  ",
+        ],
+    )
+    def test_json_types_match(self, content_type):
+        assert _is_json_content_type(content_type) is True
+
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "",
+            "application/x-www-form-urlencoded",
+            "multipart/form-data",
+            "application/jsonl",
+            "application/jsonschema",
+            "text/json",
+            "application/json-patch+json",
+        ],
+    )
+    def test_non_json_types_rejected(self, content_type):
+        assert _is_json_content_type(content_type) is False
+
+
+class TestGetRequestBody:
+    @pytest.mark.asyncio
+    async def test_json_with_charset_param_parses_as_json(self):
+        payload = {"k": "v"}
+        mock_request = MagicMock()
+        mock_request.method = "POST"
+        mock_request.body = AsyncMock(return_value=orjson.dumps(payload))
+        mock_request.headers = {"content-type": "application/json; charset=utf-8"}
+        mock_request.scope = {}
+
+        result = await get_request_body(mock_request)
+        assert result == payload
+
+    @pytest.mark.asyncio
+    async def test_form_post_routes_to_form_data(self):
+        mock_request = MagicMock()
+        mock_request.method = "POST"
+        mock_request.headers = {"content-type": "multipart/form-data; boundary=x"}
+        mock_request.form = AsyncMock(return_value={"k": "v"})
+        mock_request.scope = {}
+
+        result = await get_request_body(mock_request)
+        assert result == {"k": "v"}
+
+    @pytest.mark.asyncio
+    async def test_substring_match_no_longer_accepted(self):
+        mock_request = MagicMock()
+        mock_request.method = "POST"
+        mock_request.headers = {"content-type": "application/form-json"}
+        mock_request.scope = {}
+
+        with pytest.raises(ValueError, match="Unsupported content type"):
+            await get_request_body(mock_request)
+
+    @pytest.mark.asyncio
+    async def test_non_post_returns_empty(self):
+        mock_request = MagicMock()
+        mock_request.method = "GET"
+        assert await get_request_body(mock_request) == {}