scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-28 06:23:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 94

fix(realtime): enforce translation policies and normalize Azure transcription URLs

Browse source
This commit is contained in:
Emerson Gomes 2026-10-21 21:19:34 -05:00
parent 787b620059
commit a4bcd7cf6b
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
6 changed files with 160 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -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")
)

2
litellm/llms/azure/realtime/http_transformation.py
View file

@ -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}"

4
litellm/realtime_api/main.py
View file

@ -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

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

@ -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"

23
tests/unit/llms/azure/realtime/test_http_transformation.py
View file

@ -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"

141
tests/unit/realtime_api/test_main.py
View file

@ -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