From 277ee2e94374f53283d93e4815e78d4970569caf Mon Sep 17 00:00:00 2001 From: Kerry Lu Date: Wed, 4 Nov 2026 23:51:33 +0000 Subject: [PATCH] fix(proxy): abort the request with a 499 when the client disconnects during the body upload (#48640) Backport of #48640 to lts/1.108.x. Cherry-picked from 59b8c73930b7 (main). Dropped the pick's edit to the ValidationError branch the line does not have. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/constants.py | 1 + litellm/proxy/common_request_processing.py | 3 +- .../proxy/common_utils/http_parsing_utils.py | 65 ++++++++++++------- .../observability/test_guardrail_effects.py | 60 +++++++++++++++++ .../common_utils/test_http_parsing_utils.py | 42 ++++++++++++ 5 files changed, 147 insertions(+), 24 deletions(-) diff --git a/litellm/constants.py b/litellm/constants.py index 57d5452385d..9fcfb3f3e5c 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -702,6 +702,7 @@ DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE: Final = os.getenv( ) LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED: Final = 499 +LITELLM_CLIENT_DISCONNECT_DETAIL: Final = "Client disconnected the request" EMAIL_BUDGET_ALERT_TTL: Final = get_env_int("EMAIL_BUDGET_ALERT_TTL", 24 * 60 * 60) # 24 hours in seconds EMAIL_BUDGET_ALERT_MAX_SPEND_ALERT_PERCENTAGE: Final = get_env_float( diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index d04a84e73a4..d31a35df35f 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -38,6 +38,7 @@ from litellm.anthropic_interface.exceptions import AnthropicErrorSseFrame, anthr from litellm.constants import ( DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE, DEFAULT_MAX_RECURSE_DEPTH, + LITELLM_CLIENT_DISCONNECT_DETAIL, LITELLM_DETAILED_TIMING, LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED, MAX_LITELLM_CALL_ID_LENGTH, @@ -1784,7 +1785,7 @@ def _has_attribute_error_in_chain(exc: Exception) -> bool: return False -_CLIENT_DISCONNECT_DETAIL: Final = "Client disconnected the request" +_CLIENT_DISCONNECT_DETAIL: Final = LITELLM_CLIENT_DISCONNECT_DETAIL @runtime_checkable diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py index f1e409781ef..01795affed2 100644 --- a/litellm/proxy/common_utils/http_parsing_utils.py +++ b/litellm/proxy/common_utils/http_parsing_utils.py @@ -5,16 +5,19 @@ from types import MappingProxyType, UnionType from typing import Annotated, Any, Final, Literal, Union, get_args, get_origin import orjson -from fastapi import Request, UploadFile, status +from fastapi import HTTPException, Request, UploadFile, status from pydantic import TypeAdapter from starlette._utils import get_route_path -from starlette.requests import HTTPConnection +from starlette.datastructures import FormData +from starlette.requests import ClientDisconnect, HTTPConnection from typing_extensions import NotRequired, ReadOnly, Required, assert_never from litellm._logging import verbose_proxy_logger from litellm.constants import ( AZURE_SPEECH_PASS_THROUGH_ROUTE_PREFIX, CLIENT_REQUESTED_MODEL_SCOPE_KEY, + LITELLM_CLIENT_DISCONNECT_DETAIL, + LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED, MAX_REQUEST_BODY_SIZE_TO_REPAIR_MB, ) from litellm.integrations.otel.runtime import phase_event @@ -208,6 +211,29 @@ def _parse_form_metadata(raw_metadata: str) -> object: return parsed +async def _read_form_data(request: Request) -> FormData: + try: + form_data: Final = await request.form() + _mark_body_received(_declared_content_length(request.headers)) + except ClientDisconnect: + raise + except Exception as e: + # ``request.form()`` raises on malformed multipart (missing + # boundary, malformed chunk encoding, …). Surface as 400 so + # the auth-time pre-read does not silently cache ``{}`` while + # a later raw-body re-read sees the original payload — + # banned-param checks must see the same body the handler + # acts on. + verbose_proxy_logger.error("Invalid form payload: %s", e) + raise ProxyException( + message=f"Invalid form payload: {e}", + type="invalid_request_error", + param="request_body", + code=status.HTTP_400_BAD_REQUEST, + ) + return form_data + + async def read_request_body(request: Request | None) -> dict: """ Safely read the request body and parse it as JSON. @@ -218,10 +244,9 @@ async def read_request_body(request: Request | None) -> dict: Returns: - dict: Parsed request data as a dictionary or an empty dictionary if parsing fails """ + if request is None: + return {} try: - if request is None: - return {} - if is_otlp_trace_request(request): return {} @@ -238,24 +263,7 @@ async def read_request_body(request: Request | None) -> dict: _mark_body_received(len(binary_body)) parsed_body = _parse_binary_body(binary_body) elif _is_form_content_type(content_type): - try: - form_data: Final = await request.form() - _mark_body_received(_declared_content_length(request.headers)) - except Exception as e: - # ``request.form()`` raises on malformed multipart (missing - # boundary, malformed chunk encoding, …). Surface as 400 so - # the auth-time pre-read does not silently cache ``{}`` while - # a later raw-body re-read sees the original payload — - # banned-param checks must see the same body the handler - # acts on. - verbose_proxy_logger.error("Invalid form payload: %s", e) - raise ProxyException( - message=f"Invalid form payload: {e}", - type="invalid_request_error", - param="request_body", - code=status.HTTP_400_BAD_REQUEST, - ) - form_body: Final = dict(form_data) + form_body: Final = dict(await _read_form_data(request)) raw_metadata: Final = form_body.get("metadata") parsed_body = ( {**form_body, "metadata": _parse_form_metadata(raw_metadata)} @@ -318,6 +326,17 @@ async def read_request_body(request: Request | None) -> dict: # Re-raise ProxyException as-is verbose_proxy_logger.error("Invalid JSON payload received: %s", e) raise + except ClientDisconnect as e: + from litellm.proxy.proxy_server import ( + _close_dangling_otel_server_span, # pyright: ignore[reportPrivateUsage] # proxy_server keeps the span-close helper private; the Starlette HTTPException handler does not stamp the OTel server span opened at auth time, so the body reader must before it aborts + ) + + verbose_proxy_logger.info("Client disconnected before the request body was fully received") + disconnected: Final = HTTPException( + status_code=LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED, detail=LITELLM_CLIENT_DISCONNECT_DETAIL + ) + _close_dangling_otel_server_span(request, LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED, exc=disconnected) + raise disconnected from e except Exception as e: # Catch unexpected errors to avoid crashes verbose_proxy_logger.exception("Unexpected error reading request body - %s", e) diff --git a/tests/integration/observability/test_guardrail_effects.py b/tests/integration/observability/test_guardrail_effects.py index 462480592ef..8409648bec8 100644 --- a/tests/integration/observability/test_guardrail_effects.py +++ b/tests/integration/observability/test_guardrail_effects.py @@ -1,7 +1,9 @@ import asyncio import json import re +import socket import threading +import time import uuid from collections.abc import Callable, Iterator, Mapping from concurrent.futures import ThreadPoolExecutor @@ -11,6 +13,7 @@ from dataclasses import dataclass from pathlib import Path from types import MappingProxyType from typing import Final +from urllib.parse import urlsplit import anthropic import httpx @@ -2928,3 +2931,60 @@ def test_disconnect_scans_keep_recording_after_a_worker_is_killed(gateway: Gatew assert all(rig.secret(index) in line for index, line in enumerate(received)), received assert _post_call_statuses(rig, rig.model, rows=8) == (("success",),) * 8 assert rig.owned.process.poll() is None + + +def _openai_upstream(request: Request) -> Reply: + if request.target.endswith("/vector_stores"): + return Reply(body=json.dumps({"id": "vs_" + uuid.uuid4().hex, "object": "vector_store", "name": None}).encode()) + return Reply(body=_chat_completion_body("control-reply")) + + +def _post_half_of_the_body_then_hang_up(gateway: Gateway, path: str, body: bytes) -> None: + address: Final = urlsplit(str(gateway.client.base_url)) + assert address.hostname is not None and address.port is not None, address + head: Final = ( + f"POST {path} HTTP/1.1\r\nHost: {address.netloc}\r\nAuthorization: Bearer {gateway.key}\r\n" + f"Content-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n" + ).encode() + with socket.create_connection((address.hostname, address.port)) as connection: + connection.sendall(head + body[: len(body) // 2]) + time.sleep(0.5) + + +@pytest.mark.parametrize( + ("path", "body"), + ( + pytest.param( + "/v1/chat/completions", + b'{"model": "MODEL", "messages": [{"role": "user", "content": "Say OK"}], "max_tokens": 16}', + id="chat-completions-where-an-empty-body-is-invalid", + ), + pytest.param( + "/v1/vector_stores", + b'{"name": "integration-store", "metadata": {"owner": "integration", "purpose": "half upload"}}', + id="vector-stores-where-an-empty-body-is-valid", + ), + ), +) +def test_client_disconnect_during_the_body_upload_runs_no_endpoint_and_writes_no_spend_row( + gateway: Gateway, tmp_path: Path, path: str, body: bytes +) -> None: + with ( + wire_server(_openai_upstream) as upstream, + owned_proxy_process( + gateway, tmp_path, {"OPENAI_BASE_URL": upstream.url + "/v1", "OPENAI_API_KEY": "synthetic-openai-key"} + ) as owned, + owned.gateway.scenario() as scenario, + ): + model: Final = scenario.model(api_base=upstream.url + "/v1", api_key="synthetic-openai-key") + _post_half_of_the_body_then_hang_up(owned.gateway, path, body.replace(b"MODEL", model.encode())) + control: Final = owned.gateway.chat(model) + rows: Final = eventually( + lambda: owned.gateway.rows('SELECT request_id, call_type, status FROM "LiteLLM_SpendLogs"', ()), + lambda values: any(row["request_id"] == control["id"] for row in values), + seconds=70, + ) + assert [(row["call_type"], row["status"]) for row in rows] == [("acompletion", "success")], rows + assert [request.target for request in upstream.drain()] == ["/v1/chat/completions"], ( + "the half-uploaded request must never reach the provider" + ) diff --git a/tests/unit/proxy/common_utils/test_http_parsing_utils.py b/tests/unit/proxy/common_utils/test_http_parsing_utils.py index cd5820fa2fd..beb2e35f823 100644 --- a/tests/unit/proxy/common_utils/test_http_parsing_utils.py +++ b/tests/unit/proxy/common_utils/test_http_parsing_utils.py @@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import orjson import pytest +from fastapi import HTTPException from fastapi.testclient import TestClient from starlette.datastructures import FormData from starlette.requests import Request @@ -1592,3 +1593,44 @@ async def test_form_data_metadata_keeps_non_ascii_and_valid_surrogate_pairs(): result = await read_request_body(mock_request) assert result["metadata"]["tags"] == ["café", "😀"] + + +def _request_whose_client_disconnects_mid_upload(content_type: str) -> Request: + scope: Final = { + "type": "http", + "method": "POST", + "path": "/v1/chat/completions", + "headers": [(b"content-type", content_type.encode()), (b"content-length", b"96")], + "query_string": b"", + } + + async def receive() -> Message: + return {"type": "http.disconnect"} + + return Request(scope, receive) + + +@pytest.mark.parametrize( + "content_type", + ("application/json", "multipart/form-data; boundary=synthetic-boundary", "application/x-protobuf"), +) +async def test_read_request_body_aborts_with_499_when_the_client_disconnects_mid_upload(content_type: str) -> None: + request: Final = _request_whose_client_disconnects_mid_upload(content_type) + with pytest.raises(HTTPException) as raised: + await read_request_body(request) + assert (raised.value.status_code, raised.value.detail) == (499, "Client disconnected the request") + assert _safe_get_request_parsed_body(request) is None, "a half-received body must never be cached as {}" + + +async def test_read_request_body_closes_the_otel_server_span_when_the_client_disconnects_mid_upload( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from litellm.proxy import proxy_server + + span: Final = MagicMock() + monkeypatch.setattr(proxy_server, "open_telemetry_logger", MagicMock(), raising=False) + request: Final = _request_whose_client_disconnects_mid_upload("application/json") + request.state.parent_otel_span = span + with pytest.raises(HTTPException): + await read_request_body(request) + assert (span.end.called, request.state.parent_otel_span) == (True, None), "the auth-time span must not dangle"