diff --git a/litellm/constants.py b/litellm/constants.py index d02114f56b1..2f93ff6e8ad 100644 --- a/litellm/constants.py +++ b/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" diff --git a/litellm/llms/opensearch/vector_stores/index_passthrough.py b/litellm/llms/opensearch/vector_stores/index_passthrough.py index 4283532335d..a5c2346253d 100644 --- a/litellm/llms/opensearch/vector_stores/index_passthrough.py +++ b/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: diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 292a147d200..f7bde0e9f12 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/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", ""), ): diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py index 56cd14d987d..0c41b6bc2ad 100644 --- a/litellm/proxy/common_utils/http_parsing_utils.py +++ b/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: diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index 2663f73fca1..19eae87880c 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/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) diff --git a/tests/unit/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py b/tests/unit/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py index ec305432672..9c9e418e615 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py +++ b/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(