scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-12 03:48:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 57

fix(realtime): trust confirmed formats and authorize default models

Browse source
This commit is contained in:
Emerson Gomes 2026-10-06 10:02:36 -05:00
parent 2f824e35aa
commit c7950e6f41
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
7 changed files with 215 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -7,6 +7,7 @@ from dataclasses import dataclass
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast
from openai.types.realtime.realtime_audio_formats import AudioPCM, AudioPCMA, AudioPCMU, RealtimeAudioFormats
from pydantic import TypeAdapter, ValidationError
from typing_extensions import ReadOnly
@ -113,23 +114,17 @@ def _decode_json_object(payload: str) -> Mapping[str, object]:
return json.loads(payload)
_TRANSLATION_AUDIO_FORMAT: Final = TypeAdapter(str | Mapping[str, object] | None)
_TRANSLATION_AUDIO_FORMAT: Final = TypeAdapter(str | RealtimeAudioFormats | None)
def _translation_audio_bytes_per_second(audio_format: str | Mapping[str, object] | None) -> float | None:
if isinstance(audio_format, str):
return 8000.0 if audio_format in ("g711_ulaw", "g711_alaw") else None
if audio_format is None:
return None
format_type: Final = audio_format.get("type")
rate: Final = audio_format.get("rate")
if format_type in ("audio/pcmu", "audio/pcma") and rate is None:
return 8000.0
if not isinstance(rate, (int, float)) or rate <= 0:
return None
if format_type == "audio/pcm":
return float(rate) * 2
return float(rate) if format_type in ("audio/pcmu", "audio/pcma") else None
def _translation_audio_bytes_per_second(audio_format: str | RealtimeAudioFormats | None) -> float | None:
match audio_format:
case "pcm16" | AudioPCM():
return 48000.0
case "g711_ulaw" | "g711_alaw" | AudioPCMU() | AudioPCMA():
return 8000.0
case _:
return None
class RealtimeEventNormalizer(Protocol):
@ -176,7 +171,7 @@ class RealTimeStreaming:
self._is_translation_session = translation_session
self._translation_input_seconds = 0.0
self._translation_input_bytes_per_second = 48000.0
self._translation_output_audio_bytes = 0
self._translation_output_seconds = 0.0
self._translation_output_bytes_per_second = 48000.0
self._translation_usage_finalized = False
@ -473,11 +468,7 @@ class RealTimeStreaming:
if reported_input_seconds is not None
else self._translation_input_seconds or None
)
synthetic_output_seconds: Final = (
self._translation_output_audio_bytes / self._translation_output_bytes_per_second
if self._translation_output_audio_bytes > 0
else None
)
synthetic_output_seconds: Final = self._translation_output_seconds or None
resolved_output_seconds: Final = output_seconds if output_seconds is not None else synthetic_output_seconds
if input_seconds is not None or resolved_output_seconds is not None:
if self._should_store_message(event_obj):
@ -527,7 +518,7 @@ class RealTimeStreaming:
decoded: Final = base64.b64decode(delta, validate=True)
except (ValueError, TypeError):
return
self._translation_output_audio_bytes += len(decoded)
self._translation_output_seconds += len(decoded) / self._translation_output_bytes_per_second
def _capture_translation_input_audio(self, message: str) -> None:
if not self._is_translation_session:
@ -536,7 +527,6 @@ class RealTimeStreaming:
event: Final = _decode_json_object(message)
except (json.JSONDecodeError, TypeError):
return
self._capture_translation_audio_formats(event)
if event.get("type") != "session.input_audio_buffer.append" or not isinstance(audio := event.get("audio"), str):
return
try:
@ -546,6 +536,8 @@ class RealTimeStreaming:
self._translation_input_seconds += len(decoded) / self._translation_input_bytes_per_second
def _capture_translation_audio_formats(self, event_obj: Mapping[str, object]) -> None:
if event_obj.get("type") not in ("session.created", "session.updated"):
return
session: Final = event_obj.get("session")
if not isinstance(session, dict):
return
@ -581,15 +573,18 @@ class RealTimeStreaming:
):
self._translation_usage_finalized = True
return
if self._translation_output_audio_bytes == 0 and self._translation_input_seconds == 0:
if self._translation_output_seconds == 0 and self._translation_input_seconds == 0:
return
output_seconds: Final = self._translation_output_audio_bytes / self._translation_output_bytes_per_second
synthetic_usage: Final = (
OpenAIRealtimeTranslationDurationUsage(
type="duration", input_seconds=self._translation_input_seconds, output_seconds=output_seconds
type="duration",
input_seconds=self._translation_input_seconds,
output_seconds=self._translation_output_seconds,
)
if self._translation_input_seconds > 0
else OpenAIRealtimeTranslationDurationUsage(type="duration", output_seconds=output_seconds)
else OpenAIRealtimeTranslationDurationUsage(
type="duration", output_seconds=self._translation_output_seconds
)
)
self.messages.append(OpenAIRealtimeTranslationClosedEvent(type="session.closed", usage=synthetic_usage))
self._translation_usage_finalized = True

10
litellm/proxy/agent_endpoints/auth/managed_authorization.py
View file

@ -110,11 +110,19 @@ def managed_inference_request(
cli_model: str | None,
path_model: object = None,
query_model: object = None,
intent: str | None = None,
) -> dict[str, object]:
from litellm.proxy.auth.route_checks import RouteChecks
if route in _MANAGED_REALTIME_ROUTES:
model: Final = query_model or body.get("model")
from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model
requested_model: Final = query_model if query_model is not None else body.get("model")
model: Final = (
resolve_realtime_route_model(None, intent, route.endswith("/realtime/translations"))
if requested_model is None
else requested_model
)
if not isinstance(model, str) or not model:
raise_identity_failure(
AgentIdentityFailure(message="Managed inference requires an explicit or configured model")

1
litellm/proxy/auth/user_api_key_auth.py
View file

@ -3347,6 +3347,7 @@ async def _authorize_authenticated_request(
user_model,
request.path_params.get("model") or request.path_params.get("model_name"),
request.query_params.get("model"),
intent=request.query_params.get("intent"),
)
if user_api_key_auth_obj.managed_agent_policy is not None
else request_data

10
litellm/proxy/common_utils/http_parsing_utils.py
View file

@ -30,6 +30,16 @@ _BINARY_CONTENT_TYPES: Final[frozenset[str]] = frozenset({"application/x-protobu
_ANNOTATION_QUALIFIERS: Final[frozenset[object]] = frozenset({Annotated, NotRequired, ReadOnly, Required})
def resolve_realtime_route_model(model: str | None, intent: str | None, is_translation: bool) -> str | None:
if model is not None:
return model
if is_translation:
return "gpt-realtime-translate"
if intent == "transcription":
return "gpt-realtime-whisper"
return None
def resolve_inference_model(
body_model: object,
settings: Mapping[str, object],

17
litellm/proxy/proxy_server.py
View file

@ -445,6 +445,9 @@ from litellm.proxy.common_utils.http_parsing_utils import (
get_form_data,
resolve_inference_model,
)
from litellm.proxy.common_utils.http_parsing_utils import (
resolve_realtime_route_model as _resolve_realtime_route_model,
)
from litellm.proxy.common_utils.load_config_utils import get_config_from_bucket
from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations
from litellm.proxy.common_utils.model_listing_utils import (
@ -13042,20 +13045,6 @@ async def _reject_realtime_session(
await _release_realtime_max_parallel_slot(user_api_key_dict)
def _resolve_realtime_route_model(
model: str | None,
intent: str | None,
is_translation: bool,
) -> str | None:
if model is not None:
return model
if is_translation:
return "gpt-realtime-translate"
if intent == "transcription":
return "gpt-realtime-whisper"
return None
def _resolve_realtime_upstream_query_model(
model: str | None,
intent: str | None,

121
tests/unit/litellm_core_utils/test_realtime_streaming.py
View file

@ -1,4 +1,5 @@
import asyncio
import base64
import json
from collections.abc import Coroutine, Mapping
from dataclasses import dataclass
@ -2943,14 +2944,14 @@ def test_translation_audio_duration_is_finalized_once(event_type: str):
@pytest.mark.asyncio
@pytest.mark.parametrize("output_bytes", (0, 48000))
@pytest.mark.parametrize("format_event_type", ("session.created", "session.update"))
@pytest.mark.parametrize("format_event_type", ("session.created", "session.updated"))
@pytest.mark.parametrize(
"audio_format,bytes_per_second",
(
("pcm16", 48000),
("g711_ulaw", 8000),
("g711_alaw", 8000),
({"type": "audio/pcm", "rate": 16000}, 32000),
({"type": "audio/pcm", "rate": 24000}, 48000),
({"type": "audio/pcmu", "rate": 8000}, 8000),
({"type": "audio/pcma", "rate": 8000}, 8000),
({"type": "audio/pcmu"}, 8000),
@ -2975,10 +2976,7 @@ async def test_translation_disconnect_bills_sent_input_audio(
"type": format_event_type,
"session": {"audio": {"input": {"format": audio_format}}},
}
if format_event_type == "session.update":
await streaming._send_to_backend(json.dumps(format_event))
else:
streaming._capture_translation_output_audio(format_event)
streaming._capture_translation_output_audio(format_event)
await streaming._send_to_backend(
json.dumps(
{
@ -3001,6 +2999,96 @@ async def test_translation_disconnect_bills_sent_input_audio(
]
@pytest.mark.asyncio
@pytest.mark.parametrize("rate", (24000, 1, 0, -1, float("nan"), float("inf")))
async def test_translation_client_format_updates_do_not_change_billed_duration(rate: int | float) -> None:
backend: Final = MagicMock()
backend.send = AsyncMock()
streaming: Final = RealTimeStreaming(
websocket=_ga_client_ws(), backend_ws=backend, logging_obj=MagicMock(), translation_session=True
)
streaming._capture_translation_output_audio(
{
"type": "session.created",
"session": {"audio": {"input": {"format": "g711_ulaw"}, "output": {"format": "g711_ulaw"}}},
}
)
input_event: Final = json.dumps(
{"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(8000)).decode()}
)
output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(8000)).decode()}
await streaming._send_to_backend(input_event)
streaming._capture_translation_output_audio(output_event)
await streaming._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"audio": {
"input": {"format": {"type": "audio/pcm", "rate": rate}},
"output": {"format": {"type": "audio/pcm", "rate": rate}},
}
},
}
)
)
await streaming._send_to_backend(input_event)
streaming._capture_translation_output_audio(output_event)
streaming._finalize_translation_usage()
assert streaming.messages == [
{"type": "session.closed", "usage": {"type": "duration", "input_seconds": 2.0, "output_seconds": 2.0}}
]
@pytest.mark.asyncio
async def test_translation_confirmed_format_changes_preserve_previous_audio_duration() -> None:
backend: Final = MagicMock()
backend.send = AsyncMock()
streaming: Final = RealTimeStreaming(
websocket=_ga_client_ws(), backend_ws=backend, logging_obj=MagicMock(), translation_session=True
)
await streaming._send_to_backend(
json.dumps({"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(48000)).decode()})
)
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()}
)
streaming._capture_translation_output_audio(
{
"type": "session.updated",
"session": {"audio": {"input": {"format": "g711_alaw"}, "output": {"format": "g711_alaw"}}},
}
)
await streaming._send_to_backend(
json.dumps({"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(8000)).decode()})
)
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(8000)).decode()}
)
streaming._finalize_translation_usage()
assert streaming.messages == [
{"type": "session.closed", "usage": {"type": "duration", "input_seconds": 2.0, "output_seconds": 2.0}}
]
@pytest.mark.parametrize("rate", (1, 0, -1, float("nan"), float("inf"), 10**400))
def test_translation_invalid_provider_format_preserves_audio_duration(rate: int | float) -> None:
streaming: Final = RealTimeStreaming(
websocket=_ga_client_ws(), backend_ws=MagicMock(), logging_obj=MagicMock(), translation_session=True
)
output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()}
streaming._capture_translation_output_audio(output_event)
streaming._capture_translation_output_audio(
{"type": "session.updated", "session": {"audio": {"output": {"format": {"type": "audio/pcm", "rate": rate}}}}}
)
streaming._capture_translation_output_audio(output_event)
streaming._finalize_translation_usage()
assert streaming.messages == [{"type": "session.closed", "usage": {"type": "duration", "output_seconds": 2.0}}]
@pytest.mark.asyncio
async def test_translation_failed_audio_send_is_not_billed() -> None:
backend: Final = MagicMock()
@ -3085,7 +3173,9 @@ def test_translation_does_not_duplicate_provider_duration_usage():
model="gpt-realtime-translate",
translation_session=True,
)
streaming._translation_output_audio_bytes = 48000
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()}
)
streaming.messages.append({"type": "session.closed", "usage": {"type": "duration", "output_seconds": 0.5}})
streaming._finalize_translation_usage()
@ -3133,7 +3223,9 @@ def test_translation_preserves_input_only_provider_usage(
model="gpt-realtime-translate",
translation_session=True,
)
streaming._translation_output_audio_bytes = output_audio_bytes
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(output_audio_bytes)).decode()}
)
streaming._capture_translation_output_audio(
{"type": "session.closed", "usage": {"type": "duration", "input_seconds": 0.25}}
)
@ -3153,7 +3245,9 @@ def test_translation_retained_input_only_close_event_bills_captured_output(monke
model="gpt-realtime-translate",
translation_session=True,
)
streaming._translation_output_audio_bytes = 48000
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()}
)
close_event: Final = {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 0.25}}
streaming._capture_translation_output_audio(close_event)
@ -3175,10 +3269,15 @@ def test_translation_malformed_provider_duration_uses_captured_audio(malformed_o
model="gpt-realtime-translate",
translation_session=True,
)
streaming._translation_output_audio_bytes = 48000
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(48000)).decode()}
)
streaming._capture_translation_output_audio(
{"type": "session.closed", "usage": {"type": "duration", "input_seconds": 0.25, "output_seconds": malformed_output}}
{
"type": "session.closed",
"usage": {"type": "duration", "input_seconds": 0.25, "output_seconds": malformed_output},
}
)
streaming._finalize_translation_usage()

75
tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py
View file

@ -447,9 +447,6 @@ def test_managed_inference_ignores_unsupported_query_model():
"/realtime",
"/v1/realtime",
"/openai/v1/realtime",
"/realtime/translations",
"/v1/realtime/translations",
"/openai/v1/realtime/translations",
),
)
def test_managed_realtime_requires_a_model_and_ignores_completion_defaults(route: str) -> None:
@ -465,6 +462,35 @@ def test_managed_realtime_requires_a_model_and_ignores_completion_defaults(route
)
@pytest.mark.parametrize(
"route", ("/realtime/translations", "/v1/realtime/translations", "/openai/v1/realtime/translations")
)
@pytest.mark.parametrize("query_model", (None, "requested"))
def test_managed_translation_authorizes_the_same_model_as_the_endpoint(route: str, query_model: str | None) -> None:
from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request
from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model
admitted: Final = managed_inference_request(
route, {}, {"completion_model": "unrelated-default"}, "cli", query_model=query_model
)
assert admitted["model"] == resolve_realtime_route_model(query_model, None, is_translation=True)
assert isinstance(admitted["model"], str) and admitted["model"]
@pytest.mark.parametrize("route", ("/realtime", "/v1/realtime", "/openai/v1/realtime"))
def test_managed_transcription_authorizes_the_same_model_as_the_endpoint(route: str) -> None:
from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request
from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model
admitted: Final = managed_inference_request(
route, {}, {"completion_model": "unrelated-default"}, "cli", intent="transcription"
)
assert admitted["model"] == resolve_realtime_route_model(None, "transcription", is_translation=False)
assert isinstance(admitted["model"], str) and admitted["model"]
@pytest.mark.parametrize("mode,user", [("autonomous", None), ("delegated", "verified-human")])
def test_matching_identity_revision_and_execution_mode_pass_admission(mode: str, user: str | None) -> None:
context: Final = ManagedAgentContext.model_validate(
@ -573,22 +599,41 @@ async def test_ordinary_agent_admission_preserves_legacy_authentication(
@pytest.mark.parametrize(
"route",
tuple(dict.fromkeys(
LiteLLMRoutes.openai_routes.value
+ LiteLLMRoutes.anthropic_routes.value
+ LiteLLMRoutes.google_routes.value
)),
tuple(
dict.fromkeys(
LiteLLMRoutes.openai_routes.value + LiteLLMRoutes.anthropic_routes.value + LiteLLMRoutes.google_routes.value
)
),
)
def test_registered_inference_routes_have_an_explicit_managed_access_decision(route: str) -> None:
from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_agent_route_allowed
normalized: Final = route.removeprefix("/openai").removeprefix("/v1beta").removeprefix("/v1")
unsupported: Final = normalized.startswith((
"/videos", "/batches", "/files", "/fine_tuning", "/assistants", "/threads", "/utils/",
"/vector_stores", "/vector_store/", "/search", "/containers", "/skills", "/claude-code/",
"/interactions", "/agents", "/responses/{", "/responses/input_tokens",
"/realtime/client_secrets", "/realtime/calls", "/realtime/transcription_sessions",
"/realtime/translations/client_secrets", "/realtime/translations/calls",
)) or normalized in ("/models", "/cursor/models", "/cursor/v1/models")
unsupported: Final = normalized.startswith(
(
"/videos",
"/batches",
"/files",
"/fine_tuning",
"/assistants",
"/threads",
"/utils/",
"/vector_stores",
"/vector_store/",
"/search",
"/containers",
"/skills",
"/claude-code/",
"/interactions",
"/agents",
"/responses/{",
"/responses/input_tokens",
"/realtime/client_secrets",
"/realtime/calls",
"/realtime/transcription_sessions",
"/realtime/translations/client_secrets",
"/realtime/translations/calls",
)
) or normalized in ("/models", "/cursor/models", "/cursor/v1/models")
concrete: Final = route.split("?")[0].replace("{model}", "model").replace("{model_name:path}", "model")
assert managed_agent_route_allowed(concrete, None) is not unsupported, route