diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 094b6d45a9a..15060004945 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -2601,8 +2601,6 @@ def handle_live_session_duration_cost( custom_llm_provider: str, litellm_model_name: str, ) -> float: - if any(event.get("type") == "response.done" for event in results): - return 0.0 terminal: Final = next((event for event in reversed(results) if event.get("type") == "session.closed"), None) if terminal is None: return 0.0 diff --git a/litellm/proxy/realtime_endpoints/call_sessions.py b/litellm/proxy/realtime_endpoints/call_sessions.py index bac2abd6415..8989fadfead 100644 --- a/litellm/proxy/realtime_endpoints/call_sessions.py +++ b/litellm/proxy/realtime_endpoints/call_sessions.py @@ -223,7 +223,31 @@ def decode_call(token: str, authorization: str) -> CodexRealtimeCall: return call +MAX_REALTIME_OFFER_BYTES: Final = 8 * 1024 * 1024 + + +async def _cache_bounded_offer_body(request: Request) -> None: + try: + if int(request.headers.get("content-length", "")) > MAX_REALTIME_OFFER_BYTES: + raise HTTPException(413, "Realtime offer exceeds the 8 MiB limit") + except ValueError: + pass + if hasattr(request, "_body"): + if len(request._body) > MAX_REALTIME_OFFER_BYTES: # pyright: ignore[reportPrivateUsage] # validate Starlette's cached body without consuming it again + raise HTTPException(413, "Realtime offer exceeds the 8 MiB limit") + return + if request._form is not None and request._stream_consumed: # pyright: ignore[reportPrivateUsage] # a mixed-case empty form cache may leave the stream unread + return + body: Final = bytearray() + async for chunk in request.stream(): + if len(body) + len(chunk) > MAX_REALTIME_OFFER_BYTES: + raise HTTPException(413, "Realtime offer exceeds the 8 MiB limit") + body.extend(chunk) + request._body = bytes(body) # pyright: ignore[reportPrivateUsage] # Starlette has no public setter for its shared body cache + + async def read_codex_offer(request: Request) -> CodexRealtimeOffer: + await _cache_bounded_offer_body(request) content_type: Final = request.headers.get("content-type", "") if _normalize_media_type(content_type) == "multipart/form-data": if content_type.split(";", 1)[0] != "multipart/form-data" and not await request.form(): diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_call_sessions.py b/tests/test_litellm/proxy/realtime_endpoints/test_call_sessions.py index d730add23b2..dffd4500ed7 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_call_sessions.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_call_sessions.py @@ -11,6 +11,110 @@ from litellm.llms.chatgpt.codex import CodexRealtimeCall from litellm.proxy.realtime_endpoints.call_sessions import decode_call, encode_call +@pytest.mark.asyncio +@pytest.mark.parametrize("multipart", [False, True]) +@pytest.mark.parametrize("content_length", [None, "1", "999999999"]) +async def test_oversized_offer_stops_before_auth_or_multipart_files(monkeypatch, multipart, content_length): + import json + from unittest.mock import AsyncMock, Mock + + from fastapi import Request + + monkeypatch.setattr(codex, "MAX_REALTIME_OFFER_BYTES", 1024) + if multipart: + body = ( + b'--Boundary\r\nContent-Disposition: form-data; name="extra"; filename="large.bin"\r\n\r\n' + + b"x" * 2048 + + b"\r\n--Boundary--\r\n" + ) + media_type = b"Multipart/Form-Data; boundary=Boundary" + else: + body = json.dumps({"sdp": "x" * 2048, "session": {"model": "voice"}}).encode() + media_type = b"application/json" + chunks = [body[offset : offset + 256] for offset in range(0, len(body), 256)] + received = [] + + async def receive(): + chunk = chunks.pop(0) + received.append(len(chunk)) + return {"type": "http.request", "body": chunk, "more_body": bool(chunks)} + + headers = [(b"content-type", media_type)] + if content_length is not None: + headers.append((b"content-length", content_length.encode())) + request = Request({"type": "http", "headers": headers}, receive) + if multipart: + from litellm.proxy.common_utils.http_parsing_utils import _read_request_body + + assert await _read_request_body(request) == {} + authenticate = AsyncMock() + create_file = Mock(side_effect=AssertionError("Oversized offers must not create temporary files")) + monkeypatch.setattr(codex, "user_api_key_auth", authenticate) + monkeypatch.setattr("starlette.formparsers.SpooledTemporaryFile", create_file) + with pytest.raises(HTTPException) as rejected: + await codex.create_codex_realtime_call(request) + assert rejected.value.status_code == 413 + assert sum(received) <= 1280 + assert chunks + authenticate.assert_not_awaited() + create_file.assert_not_called() + + +@pytest.mark.asyncio +async def test_offer_at_size_limit_keeps_body_available_for_custom_auth(monkeypatch): + import json + + from fastapi import Request + + monkeypatch.setattr(codex, "MAX_REALTIME_OFFER_BYTES", 1024) + empty = {"sdp": "", "session": {"model": "voice"}} + sdp = "x" * (1024 - len(json.dumps(empty).encode())) + body = json.dumps({"sdp": sdp, "session": {"model": "voice"}}).encode() + chunks = [body[:512], body[512:]] + + async def receive(): + return {"type": "http.request", "body": chunks.pop(0), "more_body": bool(chunks)} + + request = Request({"type": "http", "headers": [(b"content-type", b"application/json")]}, receive) + offer = await codex.read_codex_offer(request) + assert offer.sdp == sdp + assert offer.session.model == "voice" + assert await request.body() == body + assert not chunks + + +@pytest.mark.asyncio +async def test_empty_pre_read_multipart_offer_returns_invalid_offer(): + from fastapi import Request + + async def receive(): + return {"type": "http.request", "body": b"--Boundary--\r\n", "more_body": False} + + request = Request( + {"type": "http", "headers": [(b"content-type", b"multipart/form-data; boundary=Boundary")]}, receive + ) + assert not await request.form() + with pytest.raises(HTTPException) as rejected: + await codex.create_codex_realtime_call(request) + assert rejected.value.status_code == 400 + + +@pytest.mark.asyncio +async def test_oversized_pre_read_offer_is_rejected_before_decoding(monkeypatch): + from fastapi import Request + + monkeypatch.setattr(codex, "MAX_REALTIME_OFFER_BYTES", 1024) + + async def receive(): + return {"type": "http.request", "body": b"x" * 2048, "more_body": False} + + request = Request({"type": "http", "headers": [(b"content-type", b"application/json")]}, receive) + await request.body() + with pytest.raises(HTTPException) as rejected: + await codex.read_codex_offer(request) + assert rejected.value.status_code == 413 + + @pytest.mark.asyncio @pytest.mark.parametrize("malformed", [False, True]) async def test_mixed_case_offer_preserves_boundary_metadata_and_closes_extra_files(monkeypatch, malformed): diff --git a/tests/test_litellm/test_cost_calculator.py b/tests/test_litellm/test_cost_calculator.py index 5886ba8b85a..9f9db7588c6 100644 --- a/tests/test_litellm/test_cost_calculator.py +++ b/tests/test_litellm/test_cost_calculator.py @@ -4827,12 +4827,43 @@ def test_live_terminal_is_not_counted_twice(monkeypatch): assert handle_realtime_stream_cost_calculation( [_live_terminal_event(), _live_terminal_event()], Usage(), "chatgpt", "live-priced-test" ) == pytest.approx(0.1) - assert ( - handle_realtime_stream_cost_calculation( - [{"type": "response.done", "response": {"usage": {}}}, _live_terminal_event()], - Usage(), - "chatgpt", - "live-priced-test", - ) - == 0 + + +@pytest.mark.parametrize("with_tokens", [False, True]) +@pytest.mark.parametrize("terminal_count", [1, 2]) +@pytest.mark.parametrize("duration_priced", [False, True]) +def test_live_terminal_with_response_done_preserves_configured_billing( + monkeypatch, with_tokens, terminal_count, duration_priced +): + monkeypatch.setitem( + litellm.model_cost, + "realtime-deployment-test", + { + "litellm_provider": "chatgpt", + "mode": "realtime", + **( + {"input_cost_per_second": 0.025} + if duration_priced + else {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002} + ), + }, ) + events = [ + { + "type": "response.done", + "response": { + "usage": ({"input_tokens": 10, "output_tokens": 5, "total_tokens": 15} if with_tokens else {}) + }, + }, + *(_live_terminal_event() for _ in range(terminal_count)), + ] + usage = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(events) + result = RealtimeAPITokenUsageProcessor.create_logging_realtime_object(usage, events) + assert completion_cost( + completion_response=result, + model="gpt-live-1-codex" if duration_priced else "gpt-realtime-1.5", + custom_llm_provider="chatgpt", + call_type="_arealtime", + custom_pricing=True, + router_model_id="realtime-deployment-test", + ) == pytest.approx(0.1 if duration_priced else (0.02 if with_tokens else 0))