From a4bcd7cf6b90776986e468110cc1df51739bdcdd Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Wed, 21 Oct 2026 21:19:34 -0500 Subject: [PATCH] fix(realtime): enforce translation policies and normalize Azure transcription URLs --- .../litellm_core_utils/realtime_streaming.py | 20 ++- .../azure/realtime/http_transformation.py | 2 +- litellm/realtime_api/main.py | 4 +- .../test_realtime_streaming.py | 2 +- .../realtime/test_http_transformation.py | 23 ++- tests/unit/realtime_api/test_main.py | 141 ++++++++++++++---- 6 files changed, 160 insertions(+), 32 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 66195634efc..d391f17eba4 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -327,6 +327,24 @@ def _realtime_runner_guardrail( return callback if runs else None +def translation_guardrails_apply(request_data: Mapping[str, object]) -> bool: + from litellm.types.guardrails import GuardrailEventHooks + + if _guardrails_run_for_event_hooks(request_data, (GuardrailEventHooks.post_call,)): + return True + if any( + isinstance(bucket, Mapping) and "_guardrail_pipelines" in bucket + for bucket in (request_data.get("metadata"), request_data.get("litellm_metadata")) + ): + from litellm.proxy.utils import pipeline_managed_guardrail_names + + if pipeline_managed_guardrail_names(request_data, "pre_call") or pipeline_managed_guardrail_names( + request_data, "post_call" + ): + return True + return realtime_guardrails_apply(request_data) + + def realtime_guardrails_apply(request_data: Mapping[str, object]) -> bool: from litellm.types.guardrails import GuardrailEventHooks @@ -2039,7 +2057,7 @@ class RealTimeStreaming: return ClientLoopExit.CLIENT_DISCONNECTED async def bidirectional_forward(self) -> None: - if self._is_translation_session and self._has_realtime_guardrails(): + if self._is_translation_session and translation_guardrails_apply(self.request_data): await self.websocket.send_text( realtime_error_event(TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE, error_type="guardrail_error") ) diff --git a/litellm/llms/azure/realtime/http_transformation.py b/litellm/llms/azure/realtime/http_transformation.py index 732b0946c6d..a8136a8fe89 100644 --- a/litellm/llms/azure/realtime/http_transformation.py +++ b/litellm/llms/azure/realtime/http_transformation.py @@ -66,7 +66,7 @@ class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig): def get_transcription_session_url(self, api_base: str | None, model: str, api_version: str | None = None) -> str: base: Final = self.get_api_base(api_base) if self._uses_ga_api(model, api_version): - return f"{base.rstrip('/')}/openai/v1/realtime/transcription_sessions" + return _azure_realtime_v1_url(base, "transcription_sessions") version: Final = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17" return f"{join_api_base(base, 'openai/realtime/transcription_sessions')}?api-version={version}" diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index b7f99992593..ba41fa58635 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -48,7 +48,7 @@ from ..litellm_core_utils.realtime_errors import realtime_error_event, websocket from ..litellm_core_utils.realtime_streaming import ( TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE, client_websocket_scope, - realtime_guardrails_apply, + translation_guardrails_apply, ) from ..llms.azure.common_utils import get_azure_ad_token from ..llms.azure.realtime.handler import AzureOpenAIRealtime, azure_realtime_protocol_for_client @@ -164,7 +164,7 @@ def _translation_rejection( ) -> _TranslationRejection | None: if custom_llm_provider not in supported_providers: return _TranslationProviderUnsupported(provider=custom_llm_provider, supported_providers=supported_providers) - if realtime_guardrails_apply({"litellm_metadata": dict(litellm_metadata)}): + if translation_guardrails_apply({"litellm_metadata": dict(litellm_metadata)}): return _TranslationGuardrailsUnsupported() return None diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 729e5bb1576..a0f89772ca3 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -3397,7 +3397,7 @@ async def test_translation_rejects_applicable_guardrails_before_forwarding_audio await streaming.bidirectional_forward() messages: Final = tuple(json.loads(call.args[0]) for call in client.send_text.await_args_list) - if (default_on or requested) and event_hook is not GuardrailEventHooks.post_call: + if default_on or requested: assert messages[0]["error"]["type"] == "guardrail_error" client.close.assert_awaited_once_with( code=1008, reason="Translation sessions cannot enforce configured realtime guardrails" diff --git a/tests/unit/llms/azure/realtime/test_http_transformation.py b/tests/unit/llms/azure/realtime/test_http_transformation.py index 714ce9558cf..7c613b4c9a8 100644 --- a/tests/unit/llms/azure/realtime/test_http_transformation.py +++ b/tests/unit/llms/azure/realtime/test_http_transformation.py @@ -7,7 +7,11 @@ import litellm from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler _DATED_PREVIEW_API_VERSION: Final = "2025-04-01-preview" -_API_BASES: Final = ("https://my.openai.azure.com", "https://my.openai.azure.com/", "https://my.openai.azure.com/openai") +_API_BASES: Final = ( + "https://my.openai.azure.com", + "https://my.openai.azure.com/", + "https://my.openai.azure.com/openai", +) class _AzureThatRecordsTheRequest: @@ -55,3 +59,20 @@ async def test_azure_sdp_exchange_targets_the_ga_v1_surface_with_the_ephemeral_k assert str(request.url) == "https://my.openai.azure.com/openai/v1/realtime/calls" assert request.headers.get("authorization") == "Bearer ek_upstream" assert "api-key" not in request.headers + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api_base", (*_API_BASES, "https://my.openai.azure.com/openai/v1/")) +async def test_azure_ga_transcription_session_normalizes_the_api_base(api_base: str) -> None: + azure: Final = _AzureThatRecordsTheRequest() + await litellm.acreate_realtime_transcription_session( + model="azure/transcription-deployment", + transcription_session={"input_audio_transcription": {"model": "transcription-deployment"}}, + api_base=api_base, + api_key="azure-key", + api_version="v1", + client=azure.client(), + ) + (request,) = azure.requests + assert str(request.url) == "https://my.openai.azure.com/openai/v1/realtime/transcription_sessions" + assert request.headers.get("api-key") == "azure-key" diff --git a/tests/unit/realtime_api/test_main.py b/tests/unit/realtime_api/test_main.py index 071af4135ae..197c1e86917 100644 --- a/tests/unit/realtime_api/test_main.py +++ b/tests/unit/realtime_api/test_main.py @@ -10,7 +10,7 @@ from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from types import TracebackType -from typing import ClassVar, Final +from typing import ClassVar, Final, Literal from unittest.mock import AsyncMock, MagicMock, patch import httpx @@ -37,6 +37,9 @@ from litellm.realtime_api import main as realtime_main from litellm.realtime_api.main import with_resolved_session_model from litellm.types.guardrails import GuardrailEventHooks from litellm.types.realtime import RealtimeQueryParams +from litellm.proxy.litellm_pre_call_utils import _apply_resolved_guardrails_to_metadata +from litellm.types.proxy.policy_engine import Policy, PolicyCondition, PolicyMatchContext +from litellm.types.proxy.policy_engine.pipeline_types import GuardrailPipeline, PipelineStep @pytest.fixture @@ -600,7 +603,11 @@ async def test_bearer_health_check_sends_the_secret_behind_an_oidc_credential( monkeypatch.setattr( litellm, "credential_list", - [CredentialItem(credential_name="voice-sa", credential_values={"api_key": f"oidc/file/{token}"}, credential_info={})], + [ + CredentialItem( + credential_name="voice-sa", credential_values={"api_key": f"oidc/file/{token}"}, credential_info={} + ) + ], ) connect: Final = _CapturingConnect() with patch("websockets.connect", connect): @@ -1597,10 +1604,13 @@ async def test_openai_realtime_http_routes_send_the_federated_bearer_for_a_keyle @pytest.mark.asyncio @pytest.mark.parametrize("flow", ("client_secret", "calls")) async def test_translation_http_routes_preserve_explicit_azure_provider(flow: str) -> None: - expected_path: Final = "/openai/v1/realtime/translations/" + { - "client_secret": "client_secrets", - "calls": "calls", - }[flow] + expected_path: Final = ( + "/openai/v1/realtime/translations/" + + { + "client_secret": "client_secrets", + "calls": "calls", + }[flow] + ) def send_response(request: httpx.Request) -> httpx.Response: assert request.url.host == "azure-deployment.example" @@ -1923,24 +1933,25 @@ class _RecordingClientWebSocket: self.close_code = code -def _block_every_request_with_a_guardrail(monkeypatch: pytest.MonkeyPatch) -> None: +def _block_every_request_with_a_guardrail( + monkeypatch: pytest.MonkeyPatch, event_hook: GuardrailEventHooks = GuardrailEventHooks.pre_call +) -> None: monkeypatch.setattr( litellm, "callbacks", - [ - CustomGuardrail( - guardrail_name="translation-blocker", event_hook=GuardrailEventHooks.pre_call, default_on=True - ) - ], + [CustomGuardrail(guardrail_name="translation-blocker", event_hook=event_hook, default_on=True)], ) @pytest.mark.asyncio @pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls")) +@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call)) async def test_translation_http_routes_refuse_a_guarded_request_before_any_upstream_call( - route: str, monkeypatch: pytest.MonkeyPatch + route: str, + monkeypatch: pytest.MonkeyPatch, + event_hook: GuardrailEventHooks, ) -> None: - _block_every_request_with_a_guardrail(monkeypatch) + _block_every_request_with_a_guardrail(monkeypatch, event_hook) upstream: Final = _UpstreamThatMustNotBeCalled() with pytest.raises(litellm.BadRequestError, match=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE) as raised: @@ -1970,10 +1981,15 @@ _DEFAULT_ON_GUARDRAIL_OPT_OUTS: Final = ( (("translation_client_secrets", "client_secrets"), ("translation_calls", "calls")), ) @pytest.mark.parametrize("key_metadata", _DEFAULT_ON_GUARDRAIL_OPT_OUTS) +@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call)) async def test_translation_http_routes_serve_a_key_opted_out_of_the_default_on_guardrail( - route: str, path: str, key_metadata: Mapping[str, object], monkeypatch: pytest.MonkeyPatch + route: str, + path: str, + key_metadata: Mapping[str, object], + monkeypatch: pytest.MonkeyPatch, + event_hook: GuardrailEventHooks, ) -> None: - _block_every_request_with_a_guardrail(monkeypatch) + _block_every_request_with_a_guardrail(monkeypatch, event_hook) gateway: Final = _GatewayThatRecordsTheUrl() await _realtime_http_route_call( @@ -2009,10 +2025,12 @@ async def test_translation_http_routes_refuse_a_provider_other_than_openai_and_a @pytest.mark.asyncio +@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call)) async def test_translation_websocket_closes_1008_for_a_guarded_request_before_the_upstream_connect( monkeypatch: pytest.MonkeyPatch, + event_hook: GuardrailEventHooks, ) -> None: - _block_every_request_with_a_guardrail(monkeypatch) + _block_every_request_with_a_guardrail(monkeypatch, event_hook) websocket: Final = _RecordingClientWebSocket() connect: Final = _CapturingConnect() @@ -2034,10 +2052,13 @@ async def test_translation_websocket_closes_1008_for_a_guarded_request_before_th @pytest.mark.asyncio @pytest.mark.parametrize("key_metadata", _DEFAULT_ON_GUARDRAIL_OPT_OUTS) +@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call)) async def test_translation_websocket_opens_for_a_key_opted_out_of_the_default_on_guardrail( - key_metadata: Mapping[str, object], monkeypatch: pytest.MonkeyPatch + key_metadata: Mapping[str, object], + monkeypatch: pytest.MonkeyPatch, + event_hook: GuardrailEventHooks, ) -> None: - _block_every_request_with_a_guardrail(monkeypatch) + _block_every_request_with_a_guardrail(monkeypatch, event_hook) websocket: Final = _RecordingClientWebSocket() connect: Final = _CapturingConnect() @@ -2093,17 +2114,17 @@ class _NativeLifecycleGuardrail(CustomGuardrail): @pytest.mark.asyncio -@pytest.mark.parametrize( - ("guardrail_class", "event_hook"), - ((_NativeLifecycleGuardrail, GuardrailEventHooks.pre_call), (CustomGuardrail, GuardrailEventHooks.post_call)), -) -async def test_translation_websocket_opens_under_a_default_on_guardrail_the_realtime_runner_skips( - guardrail_class: type[CustomGuardrail], event_hook: GuardrailEventHooks, monkeypatch: pytest.MonkeyPatch +async def test_translation_websocket_opens_under_a_native_pre_call_guardrail( + monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( litellm, "callbacks", - [guardrail_class(guardrail_name="runner-skipped", event_hook=event_hook, default_on=True)], + [ + _NativeLifecycleGuardrail( + guardrail_name="runner-skipped", event_hook=GuardrailEventHooks.pre_call, default_on=True + ) + ], ) websocket: Final = _RecordingClientWebSocket() connect: Final = _CapturingConnect() @@ -2140,3 +2161,71 @@ async def test_translation_websocket_closes_1008_for_a_provider_other_than_opena assert [json.loads(event)["error"]["message"] for event in websocket.sent] == [ "Realtime translation supports only the azure, azure_ai, openai providers, got xai" ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls", "websocket")) +@pytest.mark.parametrize("mode", ("pre_call", "post_call")) +@pytest.mark.parametrize("metadata_bucket", ("metadata", "litellm_metadata")) +@pytest.mark.parametrize("selected,condition_matches", ((True, True), (True, False), (False, True))) +async def test_translation_rejects_only_applicable_policy_pipelines( + route: str, + mode: Literal["pre_call", "post_call"], + metadata_bucket: str, + selected: bool, + condition_matches: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(litellm, "callbacks", []) + request_data: Final[dict[str, object]] = {"metadata": {}, "litellm_metadata": {}} + _apply_resolved_guardrails_to_metadata( + data=request_data, + metadata_variable_name=metadata_bucket, + context=PolicyMatchContext(model="translation-deployment"), + policy_names=["translation-policy"] if selected else [], + policies={ + "translation-policy": Policy( + condition=PolicyCondition(model="translation-deployment" if condition_matches else "unrelated-model"), + pipeline=GuardrailPipeline(mode=mode, steps=[PipelineStep(guardrail="speech-policy", on_fail="block")]), + ) + }, + ) + metadata: Final = realtime_main._build_litellm_metadata(request_data) + blocked: Final = selected and condition_matches + if route == "websocket": + websocket: Final = _RecordingClientWebSocket() + connect: Final = _CapturingConnect() + with patch("websockets.connect", connect): + await realtime_main._arealtime.__wrapped__( + model="openai/gpt-realtime-translate", + websocket=websocket, + api_key="fake-key", + realtime_mode="translation", + litellm_logging_obj=_realtime_websocket_logging(), + litellm_metadata=metadata, + ) + if blocked: + assert connect.url is None + assert websocket.close_code == 1008 + assert TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE in websocket.sent[0] + else: + assert connect.url == "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate" + return + upstream: Final = _UpstreamThatMustNotBeCalled() + call: Final = _realtime_http_route_call( + route, + _realtime_http_logging(route), + custom_llm_provider="openai", + api_base="https://upstream.example", + client=AsyncHTTPHandler(transport=httpx.MockTransport(upstream)), + metadata=metadata, + ) + if blocked: + with pytest.raises(litellm.BadRequestError, match=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE) as raised: + await call + assert raised.value.status_code == 400 + assert upstream.called is False + else: + response: Final = await call + assert response.status_code == 200 + assert upstream.called is True