fix(proxy): let OpenSearch _bulk and _msearch NDJSON bodies through auth

opensearch-py sends _bulk and _msearch NDJSON with content-type application/json, and the
auth-time body read parsed it as one JSON document and answered 400 before the route ran.
Auth now skips parsing OpenSearch NDJSON bodies the same way it skips Azure Speech audio.

litellm/constants.py

@@ -1748,6 +1748,7 @@ AZURE_SPEECH_PASS_THROUGH_ROUTE_PREFIX: Final = "/azure_speech"
 AZURE_SPEECH_SHORT_AUDIO_PATH_PREFIX: Final = "/speech/"
 AZURE_SPEECH_BATCH_PATH_PREFIX: Final = "/speechtotext/"
 AZURE_SPEECH_FAST_TRANSCRIPTION_PATH: Final = "/speechtotext/transcriptions:transcribe"
+OPENSEARCH_PASS_THROUGH_ROUTE_PREFIX: Final = "/opensearch"
 AZURE_SPEECH_STT_DOMAIN: Final = "stt.speech.microsoft.com"
 AZURE_SPEECH_COGNITIVE_SERVICES_DOMAIN: Final = "api.cognitive.microsoft.com"
 AZURE_SPEECH_SUBSCRIPTION_KEY_HEADER: Final = "Ocp-Apim-Subscription-Key"

litellm/llms/opensearch/vector_stores/index_passthrough.py

@@ -87,6 +87,15 @@ INDEX_OPERATIONS: Final = (
 )
 
 
+_NDJSON_APIS: Final = frozenset(
+    operation.api for operation in INDEX_OPERATIONS if operation.body in ("bulk", "msearch")
+)
+
+
+def carries_ndjson_body(route: str, media_type: str) -> bool:
+    return media_type == NDJSON_CONTENT_TYPE or route.rpartition("/")[2] in _NDJSON_APIS
+
+
 def _endpoint_patterns(permission: IndexPermission) -> Iterator[tuple[IndexHttpMethod, str]]:
     for operation in INDEX_OPERATIONS:
         if operation.permission == permission:

litellm/proxy/auth/user_api_key_auth.py

@@ -143,7 +143,7 @@ from litellm.proxy.common_utils.http_parsing_utils import (  # noqa: F401  # leg
     _safe_get_request_headers,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
     _safe_get_request_query_params,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
     _safe_set_request_parsed_body,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
-    is_opaque_audio_pass_through_request,
+    is_opaque_pass_through_request,
     populate_request_with_path_params,
     read_raw_json_body,
     read_request_body,
@@ -2045,7 +2045,7 @@ async def _read_request_body_deferring_parse_failure(
     must run (resolving identity onto the request's trace) before the 400 goes
     out; the caller re-raises the returned exception once identity is seeded.
     """
-    if is_opaque_audio_pass_through_request(
+    if is_opaque_pass_through_request(
         route=get_request_route(request=request),
         content_type=safe_get_request_headers(request=request).get("content-type", ""),
     ):

litellm/proxy/common_utils/http_parsing_utils.py

@@ -19,6 +19,7 @@ from litellm.constants import (
     LITELLM_CLIENT_DISCONNECT_DETAIL,
     LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED,
     MAX_REQUEST_BODY_SIZE_TO_REPAIR_MB,
+    OPENSEARCH_PASS_THROUGH_ROUTE_PREFIX,
 )
 from litellm.integrations.otel.runtime import phase_event
 from litellm.proxy._types import ProxyException
@@ -374,12 +375,18 @@ async def read_request_body(request: Request | None) -> dict:
 _read_request_body: Final = read_request_body
 
 
-def is_opaque_audio_pass_through_request(route: str, content_type: str) -> bool:
-    """Azure Speech bodies (raw audio, multipart uploads) are forwarded byte for byte, so auth must not consume them."""
+def is_opaque_pass_through_request(route: str, content_type: str) -> bool:
+    """Azure Speech audio and OpenSearch NDJSON bodies are forwarded byte for byte, so auth must not parse them."""
     media_type: Final = _normalize_media_type(content_type)
-    return route.startswith(f"{AZURE_SPEECH_PASS_THROUGH_ROUTE_PREFIX}/") and (
-        media_type.startswith("audio/") or media_type == "multipart/form-data"
-    )
+    if route.startswith(f"{AZURE_SPEECH_PASS_THROUGH_ROUTE_PREFIX}/"):
+        return media_type.startswith("audio/") or media_type == "multipart/form-data"
+    if route.startswith(f"{OPENSEARCH_PASS_THROUGH_ROUTE_PREFIX}/"):
+        from litellm.llms.opensearch.vector_stores.index_passthrough import (  # noqa: PLC0415  # loaded only for OpenSearch routes, like the route itself
+            carries_ndjson_body,
+        )
+
+        return carries_ndjson_body(route, media_type)
+    return False
 
 
 async def read_raw_json_body(request: Request | None) -> bytes | None:

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -1015,8 +1015,8 @@ async def opensearch_proxy_route(  # noqa: ANN201  # returns create_pass_through
     `write` on it in `allowed_vector_store_indexes`, and LiteLLM signs the call with the vector store's AWS identity.
     """
     from litellm.llms.opensearch.vector_stores.index_passthrough import (
-        NDJSON_CONTENT_TYPE,
         IndexRefusal,
+        carries_ndjson_body,
         index_operation,
         parse_index_path,
         prepare_index_request,
@@ -1060,7 +1060,7 @@ async def opensearch_proxy_route(  # noqa: ANN201  # returns create_pass_through
     )
     setattr(request.state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, prepared.body)
     setattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY, {})
-    if prepared.media_type == NDJSON_CONTENT_TYPE:
+    if carries_ndjson_body(endpoint, prepared.media_type):
         safe_set_request_parsed_body(request, {})
     return await endpoint_func(request, fastapi_response, user_api_key_dict)
 

tests/unit/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -9729,21 +9729,24 @@ class TestOpenSearchProxyRoute:
             pytest.param("_msearch", b'{}\n{"query":{"match_all":{}}}\n{"preference":"a"}\n{}\n', id="msearch"),
         ],
     )
+    @pytest.mark.parametrize("content_type", [NDJSON_TYPE, "application/json"])
     def test_an_ndjson_body_without_index_references_is_sent_byte_for_byte(
-        self, opensearch_client: TestClient, monkeypatch: pytest.MonkeyPatch, api: str, body: bytes
+        self, opensearch_client: TestClient, monkeypatch: pytest.MonkeyPatch, api: str, body: bytes, content_type: str
     ) -> None:
-        _act_as(monkeypatch, _opensearch_caller("read", "write"))
+        monkeypatch.setattr("litellm.proxy.proxy_server.master_key", SHARED_MASTER_KEY)
         with respx.mock(assert_all_called=True) as upstream:
             route = upstream.post(f"{DOCS_UPSTREAM}/{api}").mock(
                 return_value=httpx.Response(200, json={"errors": False, "items": []})
             )
             response = opensearch_client.post(
-                f"/opensearch/docs/{api}", content=body, headers={"Content-Type": NDJSON_TYPE}
+                f"/opensearch/docs/{api}",
+                content=body,
+                headers={"Content-Type": content_type, "Authorization": f"Bearer {SHARED_MASTER_KEY}"},
             )
 
         assert response.status_code == 200
         sent: Final = route.calls.last.request
-        assert (sent.content, sent.headers["content-type"]) == (body, NDJSON_TYPE)
+        assert (sent.content, sent.headers["content-type"]) == (body, content_type)
         assert _verified_sigv4_scope(sent) == ("aoss", "us-east-1")
 
     @pytest.mark.parametrize(