From dcd949350dbcdaca89f233d641af44f2b6d00ee1 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 8 Oct 2026 03:03:55 -0700 Subject: [PATCH] fix(realtime): pick the backend websocket TLS context from the URL scheme The Azure and OpenAI realtime dials and the Responses WebSocket dial passed the shared TLS context to every backend, so a ws:// api_base failed with "ssl argument is incompatible with a ws:// URI" and ssl_verify false still verified a self-signed certificate. All three now call realtime_ssl_for_url, and the Azure realtime URL builder maps http:// to ws://. --- litellm/llms/azure/realtime/handler.py | 11 +- litellm/llms/custom_httpx/llm_http_handler.py | 8 +- litellm/llms/openai/realtime/handler.py | 17 +- tests/integration/_support/upstream.py | 24 +- .../test_realtime_websocket_dial_schemes.py | 340 ++++++++++++++++++ .../realtime/test_azure_realtime_handler.py | 52 +++ .../realtime/test_openai_realtime_handler.py | 40 +++ .../test_responses_websocket_all_providers.py | 69 ++++ 8 files changed, 533 insertions(+), 28 deletions(-) create mode 100644 tests/integration/providers/test_realtime_websocket_dial_schemes.py diff --git a/litellm/llms/azure/realtime/handler.py b/litellm/llms/azure/realtime/handler.py index 79dd7f098c8..2a93f6c7322 100644 --- a/litellm/llms/azure/realtime/handler.py +++ b/litellm/llms/azure/realtime/handler.py @@ -22,7 +22,7 @@ from ....litellm_core_utils.realtime_streaming import ( ScopedWebSocket, client_sent_openai_beta_realtime_header, ) -from ....llms.custom_httpx.http_handler import get_shared_realtime_ssl_context +from ....llms.custom_httpx.http_handler import realtime_ssl_for_url from ..azure import AzureChatCompletion # BACKEND_WS_URL = "ws://localhost:8080/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01" @@ -88,7 +88,7 @@ class AzureOpenAIRealtime(AzureChatCompletion): Construct Azure realtime WebSocket URL. Args: - api_base: Azure API base URL (will be converted from https:// to wss://) + api_base: Azure API base URL (https:// becomes wss://, http:// becomes ws://) model: Model deployment name api_version: Azure API version realtime_protocol: Protocol version to use: @@ -105,7 +105,7 @@ class AzureOpenAIRealtime(AzureChatCompletion): """ from urllib.parse import urlencode - api_base = api_base.replace("https://", "wss://") + ws_base: Final = api_base.replace("https://", "wss://").replace("http://", "ws://") # Determine path based on realtime_protocol (case-insensitive) _is_ga: Final = realtime_protocol is not None and realtime_protocol.upper() in ( @@ -128,7 +128,7 @@ class AzureOpenAIRealtime(AzureChatCompletion): query_parts.append(urlencode({"intent": intent})) qs: Final = "&".join(query_parts) - return f"{api_base}{path}?{qs}" if qs else f"{api_base}{path}" + return f"{ws_base}{path}?{qs}" if qs else f"{ws_base}{path}" def construct_url( self, @@ -176,12 +176,11 @@ class AzureOpenAIRealtime(AzureChatCompletion): auth_headers: Final = self.get_auth_headers(api_key=api_key, azure_ad_token=azure_ad_token) try: - ssl_context: Final = get_shared_realtime_ssl_context() async with websockets.connect( url, additional_headers=auth_headers, max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES, - ssl=ssl_context, + ssl=realtime_ssl_for_url(url), ) as backend_ws: realtime_streaming: Final = RealTimeStreaming( websocket, diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 124f45e9bf8..2567fbd27f8 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -199,7 +199,7 @@ def _rust_responses_websocket_enabled( return decision(context) is not Decision.PYTHON -from .http_handler import get_shared_realtime_ssl_context, realtime_ssl_for_url +from .http_handler import realtime_ssl_for_url if TYPE_CHECKING: from aiohttp import ClientSession @@ -6813,11 +6813,7 @@ class BaseLLMHTTPHandler: ws_url = urlunparse(_parsed._replace(query=urlencode({k: v[0] for k, v in _qs.items()}))) try: - ssl_context = get_shared_realtime_ssl_context() - if ws_url.startswith("wss://") and ssl_context is False: - ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE + ssl_context: Final = realtime_ssl_for_url(ws_url) logging_obj.pre_call( input=None, diff --git a/litellm/llms/openai/realtime/handler.py b/litellm/llms/openai/realtime/handler.py index 325169cfe15..cddf8a54be0 100644 --- a/litellm/llms/openai/realtime/handler.py +++ b/litellm/llms/openai/realtime/handler.py @@ -18,7 +18,7 @@ from ....litellm_core_utils.realtime_streaming import ( RealTimeStreaming, client_sent_openai_beta_realtime_header, ) -from ....llms.custom_httpx.http_handler import get_shared_realtime_ssl_context +from ....llms.custom_httpx.http_handler import realtime_ssl_for_url from ..common_utils import is_openai_backed_api_base from ..openai import OpenAIChatCompletion @@ -68,20 +68,9 @@ class OpenAIRealtime(OpenAIChatCompletion): url: WebSocket URL (ws:// or wss://) Returns: - SSL configuration (None, True, or SSLContext) + None for ws://, otherwise the shared SSL context, or an unverified one when ssl_verify is off """ - if url.startswith("ws://"): - return None - - # Use the shared SSL context which respects custom CA certs and SSL settings - ssl_config: Final = get_shared_realtime_ssl_context() - - # If ssl_config is False (ssl_verify=False), websockets library needs True instead - # to establish connection without verification (False would fail) - if ssl_config is False: - return True - - return ssl_config + return realtime_ssl_for_url(url) def _construct_url(self, api_base: str, query_params: RealtimeQueryParams) -> str: """ diff --git a/tests/integration/_support/upstream.py b/tests/integration/_support/upstream.py index 0c7dab90fcc..5bc18d607bf 100644 --- a/tests/integration/_support/upstream.py +++ b/tests/integration/_support/upstream.py @@ -383,10 +383,9 @@ class Provider: self.interactions[interaction_id] = cancelled return JSONResponse(_interaction_body(interaction_id, cancelled)) - async def realtime(self, websocket: WebSocket) -> None: + def _observed_upgrade(self, websocket: WebSocket) -> tuple[str, str, str]: authorization: Final = websocket.headers.get("authorization", "") api_key: Final = websocket.headers.get("api-key", "") - scenario_id: Final = authorization.removeprefix("Bearer ") if authorization else api_key self.observations.put( Observation( websocket.url.path, @@ -396,6 +395,10 @@ class Provider: api_key, ) ) + return authorization.removeprefix("Bearer ") if authorization else api_key, authorization, api_key + + async def realtime(self, websocket: WebSocket) -> None: + scenario_id, authorization, api_key = self._observed_upgrade(websocket) response: Final = self.scenario_store.get(scenario_id) if not isinstance(response, RealtimeResponse): await websocket.close(code=4404) @@ -420,6 +423,21 @@ class Provider: continue await websocket.send_json(_rendered_realtime_event(event, scenario_id)) + async def responses_websocket(self, websocket: WebSocket) -> None: + scenario_id, authorization, api_key = self._observed_upgrade(websocket) + response: Final = self.scenario_store.get(scenario_id) + if not isinstance(response, RealtimeResponse): + await websocket.close(code=4404) + return + await websocket.accept() + async for message in websocket.iter_json(): + payload: Final = JSON_OBJECT.validate_python(message) + self.observations.put(Observation(websocket.url.path, authorization, payload, "WEBSOCKET_FRAME", api_key)) + if payload.get("type") != "response.create": + continue + for event in response.events: + await websocket.send_json(_rendered_realtime_event(event, scenario_id)) + async def muse_realtime(self, websocket: WebSocket) -> None: await websocket.accept() handshake: Final = JSON_OBJECT.validate_python(await websocket.receive_json()) @@ -569,6 +587,8 @@ class Provider: WebSocketRoute("/v1/realtime", self.realtime), WebSocketRoute("/openai/v1/realtime", self.realtime), WebSocketRoute("/openai/realtime", self.realtime), + WebSocketRoute("/v1/responses", self.responses_websocket), + WebSocketRoute("/openai/v1/responses", self.responses_websocket), WebSocketRoute("/v1/asr/realtime", self.muse_realtime), WebSocketRoute(GEMINI_LIVE_PATH, self.gemini_live), ] diff --git a/tests/integration/providers/test_realtime_websocket_dial_schemes.py b/tests/integration/providers/test_realtime_websocket_dial_schemes.py new file mode 100644 index 00000000000..c32ae60b704 --- /dev/null +++ b/tests/integration/providers/test_realtime_websocket_dial_schemes.py @@ -0,0 +1,340 @@ +"""Backend websocket dials carry a TLS context only when the resolved URL is ``wss://``. + +An Azure or OpenAI realtime session and a Responses websocket session each open one backend socket. A +plaintext ``api_base`` (``http://`` or ``ws://``) resolves to ``ws://``, where the websockets client refuses +any ``ssl`` argument, and ``https://`` resolves to ``wss://``, where ``litellm_settings.ssl_verify: false`` +has to become a context that skips verification rather than ``False``, which the client rejects. Every row +runs against the scripted upstream on 127.0.0.1 and asserts the upgrade it saw, so a dial that never left +the proxy fails on the observation, not only on the client's close. +""" + +from __future__ import annotations + +import asyncio +import json +import uuid +from collections.abc import AsyncIterator, Iterator +from pathlib import Path +from typing import Final + +import httpx +import pytest +import websockets +import yaml +from pydantic import JsonValue +from websockets.asyncio.client import ClientConnection +from websockets.exceptions import ConnectionClosed + +from tests.integration._support.client import ( + JSON_OBJECT, + Gateway, + Scenario, + eventually, + gateway_from_environment, + object_value, + string_value, +) +from tests.integration._support.process import UpstreamSlot, owned_proxy, owned_upstream +from tests.integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario +from tests.integration.cost_calculation.cost_tracking_case import RealtimeResponse + +pytestmark: Final = pytest.mark.timeout(240) + +REALTIME_MODEL: Final = "gpt-realtime" +RESPONSES_MODEL: Final = "gpt-5.6" +API_VERSION: Final = "2025-04-01-preview" +PLAINTEXT_SCHEMES: Final = ("http", "ws") +AZURE_REALTIME_PATHS: Final = frozenset({"/openai/realtime", "/openai/v1/realtime"}) +OPENAI_REALTIME_PATH: Final = "/v1/realtime" +RESPONSES_DEPLOYMENTS: Final = (("openai", "/v1", "/v1/responses"), ("azure", "", "/openai/v1/responses")) +RESPONSES_TERMINAL: Final = frozenset({"response.completed", "response.failed", "error"}) +COMPLETED_REALTIME_TURN: Final = ("session.created", "response.done") +COMPLETED_RESPONSES_TURN: Final = ("response.created", "response.output_text.delta", "response.completed") +CONVERGENCE_SECONDS: Final = 30 +REALTIME_USAGE: Final[dict[str, JsonValue]] = { + "total_tokens": 12, + "input_tokens": 7, + "output_tokens": 5, + "input_token_details": {"text_tokens": 7, "audio_tokens": 0, "cached_tokens": 0}, + "output_token_details": {"text_tokens": 5, "audio_tokens": 0}, +} +RESPONSES_USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 5, "output_tokens": 2, "total_tokens": 7} + + +def _realtime_turn() -> RealtimeResponse: + return RealtimeResponse( + content_type="application/x-realtime", + session_model=REALTIME_MODEL, + events=( + { + "type": "response.done", + "event_id": "evt_$UNIQUE_ID", + "response": { + "id": "resp_$REQUEST_ID", + "object": "realtime.response", + "status": "completed", + "output": [], + "usage": REALTIME_USAGE, + }, + }, + ), + ) + + +def _responses_turn() -> RealtimeResponse: + response: Final[dict[str, JsonValue]] = { + "id": "resp_$REQUEST_ID", + "object": "response", + "created_at": 1700000000, + "model": RESPONSES_MODEL, + } + message: Final[dict[str, JsonValue]] = { + "type": "message", + "id": "msg_$REQUEST_ID", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "scripted $REQUEST_ID", "annotations": []}], + } + return RealtimeResponse( + content_type="application/x-realtime", + events=( + {"type": "response.created", "response": {**response, "status": "in_progress", "output": []}}, + { + "type": "response.output_text.delta", + "item_id": "msg_$REQUEST_ID", + "output_index": 0, + "content_index": 0, + "delta": "scripted $REQUEST_ID", + }, + { + "type": "response.completed", + "response": {**response, "status": "completed", "output": [message], "usage": RESPONSES_USAGE}, + }, + ), + ) + + +def _scripted(scenario: Scenario, response: RealtimeResponse, *, control_url: str) -> ScenarioHandle: + handle: Final = register_scenario(f"ws-dial-{uuid.uuid4().hex[:12]}", response, control_url=control_url) + scenario.cleanups.callback(delete_scenario, handle) + return handle + + +def _ws_base(http_url: str) -> str: + return http_url.replace("https://", "wss://").replace("http://", "ws://") + + +def _with_scheme(http_url: str, scheme: str) -> str: + return f"{scheme}://{http_url.removeprefix('http://')}" + + +def _proxy_ws(gateway: Gateway) -> str: + return _ws_base(str(gateway.client.base_url).rstrip("/")) + + +def _upstream_url(gateway: Gateway) -> str: + return gateway.upstream_url.rstrip("/") + + +def _realtime_deployment(scenario: Scenario, provider: str, scenario_id: str, api_base: str) -> str: + return scenario.model( + model=f"{provider}/{REALTIME_MODEL}", + api_key=scenario_id, + api_base=api_base, + api_version=API_VERSION, + model_info={"mode": "realtime"}, + ) + + +def _responses_deployment(scenario: Scenario, provider: str, scenario_id: str, api_base: str) -> str: + return scenario.model( + model=f"{provider}/{RESPONSES_MODEL}", + api_key=scenario_id, + api_base=api_base, + **({"api_version": API_VERSION} if provider == "azure" else {}), + ) + + +def _close_code(closed: ConnectionClosed) -> int: + return 1006 if closed.rcvd is None else closed.rcvd.code + + +async def _realtime_frames(socket: ClientConnection) -> AsyncIterator[dict[str, JsonValue]]: + try: + first: Final = JSON_OBJECT.validate_json(await socket.recv()) + yield first + if first.get("type") != "session.created": + async for message in socket: + yield JSON_OBJECT.validate_json(message) + return + await socket.send(json.dumps({"type": "response.create"})) + async for message in socket: + event: Final = JSON_OBJECT.validate_json(message) + yield event + if event.get("type") == "response.done": + break + except ConnectionClosed as closed: + yield {"type": "closed", "code": _close_code(closed)} + + +def _create_frame(model: str) -> str: + return json.dumps( + { + "type": "response.create", + "model": model, + "input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "say it"}]}], + } + ) + + +async def _responses_frames(socket: ClientConnection, model: str) -> AsyncIterator[dict[str, JsonValue]]: + try: + await socket.send(_create_frame(model)) + async for message in socket: + event: Final = JSON_OBJECT.validate_json(message) + yield event + if event.get("type") in RESPONSES_TERMINAL: + break + except ConnectionClosed as closed: + yield {"type": "closed", "code": _close_code(closed)} + + +async def _collect(frames: AsyncIterator[dict[str, JsonValue]]) -> tuple[dict[str, JsonValue], ...]: + return tuple([frame async for frame in frames]) + + +async def _realtime_session(ws_base: str, model: str, key: str) -> tuple[dict[str, JsonValue], ...]: + headers: Final = {"Authorization": f"Bearer {key}"} + async with websockets.connect(f"{ws_base}/v1/realtime?model={model}", additional_headers=headers) as socket: + return await asyncio.wait_for(_collect(_realtime_frames(socket)), 60) + + +async def _responses_session(ws_base: str, model: str, key: str) -> tuple[dict[str, JsonValue], ...]: + headers: Final = {"Authorization": f"Bearer {key}"} + async with websockets.connect(f"{ws_base}/v1/responses?model={model}", additional_headers=headers) as socket: + return await asyncio.wait_for(_collect(_responses_frames(socket, model)), 60) + + +def _types(events: tuple[dict[str, JsonValue], ...]) -> tuple[str, ...]: + return tuple(string_value(event["type"]) for event in events) + + +def _observations(control_url: str) -> tuple[dict[str, JsonValue], ...]: + verify: Final = not control_url.startswith("https://") + with httpx.Client(base_url=control_url, timeout=5, trust_env=False, verify=verify) as upstream: + return tuple(map(object_value, upstream.get("/__observations").json()["requests"])) + + +def _dialed_by(request: dict[str, JsonValue], scenario_id: str) -> bool: + bearer: Final = string_value(request["authorization"]).removeprefix("Bearer ") + return request["method"] == "WEBSOCKET" and scenario_id in (bearer, request["api_key"]) + + +def _upgrade_paths(control_url: str, scenario_id: str) -> tuple[str, ...]: + observed: Final = _observations(control_url) + return tuple(string_value(request["path"]) for request in observed if _dialed_by(request, scenario_id)) + + +def _assert_azure_realtime_upgrade(control_url: str, scenario_id: str) -> None: + paths: Final = _upgrade_paths(control_url, scenario_id) + assert len(paths) == 1 and paths[0] in AZURE_REALTIME_PATHS, paths + + +def _health(gateway: Gateway, model: str) -> dict[str, JsonValue]: + response: Final = gateway.request("GET", "/health", params={"model": model}) + return JSON_OBJECT.validate_json(response.content) + + +def _probed(body: dict[str, JsonValue]) -> int: + counts: Final = (body.get("healthy_count", 0), body.get("unhealthy_count", 0)) + return sum(count for count in counts if isinstance(count, int)) + + +def _converged_health(gateway: Gateway, model: str) -> dict[str, JsonValue]: + return eventually(lambda: _health(gateway, model), lambda body: _probed(body) == 1, seconds=CONVERGENCE_SECONDS) + + +@pytest.mark.parametrize("scheme", PLAINTEXT_SCHEMES) +def test_azure_realtime_session_over_a_plaintext_api_base(gateway: Gateway, scheme: str) -> None: + with gateway.scenario() as scenario: + handle: Final = _scripted(scenario, _realtime_turn(), control_url=_upstream_url(gateway)) + api_base: Final = _with_scheme(_upstream_url(gateway), scheme) + model: Final = _realtime_deployment(scenario, "azure", handle.scenario_id, api_base) + key: Final = scenario.key(models=[model]) + events: Final = asyncio.run(_realtime_session(_proxy_ws(gateway), model, key)) + assert _types(events) == COMPLETED_REALTIME_TURN, events + _assert_azure_realtime_upgrade(_upstream_url(gateway), handle.scenario_id) + + +@pytest.mark.parametrize("scheme", PLAINTEXT_SCHEMES) +def test_azure_realtime_health_check_over_a_plaintext_api_base(gateway: Gateway, scheme: str) -> None: + with gateway.scenario() as scenario: + handle: Final = _scripted(scenario, _realtime_turn(), control_url=_upstream_url(gateway)) + api_base: Final = _with_scheme(_upstream_url(gateway), scheme) + model: Final = _realtime_deployment(scenario, "azure", handle.scenario_id, api_base) + health: Final = _converged_health(gateway, model) + assert (health["healthy_count"], health["unhealthy_count"]) == (1, 0), health + _assert_azure_realtime_upgrade(_upstream_url(gateway), handle.scenario_id) + + +@pytest.mark.parametrize(("provider", "suffix", "path"), RESPONSES_DEPLOYMENTS, ids=["openai", "azure"]) +def test_responses_websocket_session_over_a_plaintext_api_base( + gateway: Gateway, provider: str, suffix: str, path: str +) -> None: + with gateway.scenario() as scenario: + handle: Final = _scripted(scenario, _responses_turn(), control_url=_upstream_url(gateway)) + model: Final = _responses_deployment(scenario, provider, handle.scenario_id, _upstream_url(gateway) + suffix) + key: Final = scenario.key(models=[model]) + events: Final = asyncio.run(_responses_session(_proxy_ws(gateway), model, key)) + assert _types(events) == COMPLETED_RESPONSES_TURN, events + assert _upgrade_paths(_upstream_url(gateway), handle.scenario_id) == (path,) + + +def _unverified_config(directory: Path) -> Path: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + unverified: Final = {**config, "litellm_settings": {**config["litellm_settings"], "ssl_verify": False}} + path: Final = directory / "ssl-verify-off.yaml" + path.write_text(yaml.safe_dump(unverified)) + return path + + +@pytest.fixture(scope="module") +def tls_upstream(tmp_path_factory: pytest.TempPathFactory) -> Iterator[UpstreamSlot]: + with owned_upstream(tmp_path_factory.mktemp("ws-dial-tls-upstream"), tls=True) as slot: + yield slot + + +@pytest.fixture(scope="module") +def unverified_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]: + directory: Final = tmp_path_factory.mktemp("ws-dial-unverified-proxy") + with ( + gateway_from_environment() as base, + owned_proxy(base, directory, {}, config=_unverified_config(directory)) as proxy, + ): + yield proxy + + +@pytest.mark.parametrize("provider", ["azure", "openai"]) +def test_realtime_session_with_verification_off_over_a_self_signed_upstream( + unverified_proxy: Gateway, tls_upstream: UpstreamSlot, provider: str +) -> None: + with unverified_proxy.scenario() as scenario: + handle: Final = _scripted(scenario, _realtime_turn(), control_url=tls_upstream.url) + model: Final = _realtime_deployment(scenario, provider, handle.scenario_id, tls_upstream.url) + key: Final = scenario.key(models=[model]) + events: Final = asyncio.run(_realtime_session(_proxy_ws(unverified_proxy), model, key)) + assert _types(events) == COMPLETED_REALTIME_TURN, events + paths: Final = _upgrade_paths(tls_upstream.url, handle.scenario_id) + assert len(paths) == 1 and paths[0] in (AZURE_REALTIME_PATHS | {OPENAI_REALTIME_PATH}), paths + + +def test_responses_websocket_session_with_verification_off_over_a_self_signed_upstream( + unverified_proxy: Gateway, tls_upstream: UpstreamSlot +) -> None: + with unverified_proxy.scenario() as scenario: + handle: Final = _scripted(scenario, _responses_turn(), control_url=tls_upstream.url) + model: Final = _responses_deployment(scenario, "openai", handle.scenario_id, f"{tls_upstream.url}/v1") + key: Final = scenario.key(models=[model]) + events: Final = asyncio.run(_responses_session(_proxy_ws(unverified_proxy), model, key)) + assert _types(events) == COMPLETED_RESPONSES_TURN, events + assert _upgrade_paths(tls_upstream.url, handle.scenario_id) == ("/v1/responses",) diff --git a/tests/unit/llms/azure/realtime/test_azure_realtime_handler.py b/tests/unit/llms/azure/realtime/test_azure_realtime_handler.py index 7a5ce8ab853..1b8be7be326 100644 --- a/tests/unit/llms/azure/realtime/test_azure_realtime_handler.py +++ b/tests/unit/llms/azure/realtime/test_azure_realtime_handler.py @@ -740,3 +740,55 @@ async def test_arealtime_forwards_deployment_azure_ad_token(monkeypatch, no_ambi ) assert mock_async_realtime.call_args.kwargs["azure_ad_token"] == "deployment-entra-token" + + +@pytest.mark.asyncio +async def test_construct_url_maps_http_to_ws(): + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + url = AzureOpenAIRealtime()._construct_url("http://127.0.0.1:8113", "gpt-realtime", "2024-10-01-preview") + + assert url.startswith("ws://127.0.0.1:8113/openai/realtime?") + + +async def _dial_through_mocked_connect(api_base: str) -> tuple[str, dict]: + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + with ( + patch("websockets.connect", return_value=_DummyAsyncContextManager(AsyncMock())) as mock_ws_connect, + patch("litellm.llms.azure.realtime.handler.RealTimeStreaming") as mock_realtime_streaming, + ): + mock_realtime_streaming.return_value.bidirectional_forward = AsyncMock() + await AzureOpenAIRealtime().async_realtime( + model="gpt-realtime", + websocket=AsyncMock(), + logging_obj=MagicMock(), + api_base=api_base, + api_key="test-key", + api_version="2024-10-01-preview", + ) + mock_ws_connect.assert_called_once() + return mock_ws_connect.call_args[0][0], mock_ws_connect.call_args[1] + + +@pytest.mark.parametrize("api_base", ["http://127.0.0.1:8113", "ws://127.0.0.1:8113"]) +@pytest.mark.asyncio +async def test_async_realtime_plaintext_api_base_dials_ws_without_ssl(api_base): + url, kwargs = await _dial_through_mocked_connect(api_base) + + assert url.startswith("ws://127.0.0.1:8113/openai/realtime?") + assert kwargs["ssl"] is None + + +@pytest.mark.asyncio +async def test_async_realtime_wss_with_ssl_verify_off_dials_with_an_unverified_context(monkeypatch): + import ssl + + monkeypatch.setattr("litellm.llms.custom_httpx.http_handler._shared_realtime_ssl_context", False) + + url, kwargs = await _dial_through_mocked_connect("https://127.0.0.1:8113") + + assert url.startswith("wss://127.0.0.1:8113/openai/realtime?") + assert isinstance(kwargs["ssl"], ssl.SSLContext) + assert kwargs["ssl"].verify_mode is ssl.CERT_NONE + assert kwargs["ssl"].check_hostname is False diff --git a/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py b/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py index cc42d84747a..d1f16b6fc72 100644 --- a/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py +++ b/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py @@ -526,3 +526,43 @@ async def test_realtime_query_params_preserve_missing_model(monkeypatch): called_kwargs = mock_async_realtime.call_args.kwargs assert called_kwargs["query_params"] == {"intent": "transcription"} + + +@pytest.mark.asyncio +async def test_async_realtime_wss_with_ssl_verify_off_dials_with_an_unverified_context(monkeypatch): + import ssl + + from litellm.llms.openai.realtime.handler import OpenAIRealtime + + monkeypatch.setattr("litellm.llms.custom_httpx.http_handler._shared_realtime_ssl_context", False) + + class DummyAsyncContextManager: + def __init__(self, value): + self.value = value + + async def __aenter__(self): + return self.value + + async def __aexit__(self, exc_type, exc, tb): + return None + + with ( + patch("websockets.connect", return_value=DummyAsyncContextManager(AsyncMock())) as mock_ws_connect, + patch("litellm.llms.openai.realtime.handler.RealTimeStreaming") as mock_realtime_streaming, + ): + mock_realtime_streaming.return_value.bidirectional_forward = AsyncMock() + await OpenAIRealtime().async_realtime( + model="gpt-realtime", + websocket=AsyncMock(), + logging_obj=MagicMock(), + api_base="https://127.0.0.1:8113", + api_key="test-key", + query_params={"model": "gpt-realtime"}, + ) + + mock_ws_connect.assert_called_once() + assert mock_ws_connect.call_args[0][0].startswith("wss://127.0.0.1:8113/v1/realtime?") + dialed_ssl = mock_ws_connect.call_args[1]["ssl"] + assert isinstance(dialed_ssl, ssl.SSLContext) + assert dialed_ssl.verify_mode is ssl.CERT_NONE + assert dialed_ssl.check_hostname is False diff --git a/tests/unit/responses/test_responses_websocket_all_providers.py b/tests/unit/responses/test_responses_websocket_all_providers.py index b29e9a6b7db..714a64f7f6e 100644 --- a/tests/unit/responses/test_responses_websocket_all_providers.py +++ b/tests/unit/responses/test_responses_websocket_all_providers.py @@ -2979,6 +2979,75 @@ def _wrapped_reasoning_item(): } +class TestNativeWebSocketDialSsl: + """The backend dial's ssl argument follows the resolved URL scheme: none for ws://, the shared + context for wss://, and an unverified context for wss:// when ssl_verify is off.""" + + @staticmethod + async def _dial(ws_url: str) -> dict: + from unittest.mock import AsyncMock, MagicMock, patch + + from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler + + captured = {} + + class FakeConnect: + def __init__(self, url, **kwargs): + captured.update(kwargs) + + async def __aenter__(self): + raise Exception("stop") + + async def __aexit__(self, *args): + pass + + mock_config = MagicMock(spec=OpenAIResponsesAPIConfig) + mock_config.supports_native_websocket.return_value = True + mock_config.get_websocket_url.return_value = ws_url + mock_config.validate_environment.return_value = {} + mock_logging = MagicMock() + mock_logging.pre_call = MagicMock() + mock_ws = MagicMock() + mock_ws.close = AsyncMock() + + with patch("websockets.connect", FakeConnect): + await BaseLLMHTTPHandler().async_responses_websocket( + model="gpt-5.6", + websocket=mock_ws, + logging_obj=mock_logging, + responses_api_provider_config=mock_config, + api_key="sk-test", + ) + assert "ssl" in captured, "the backend dial never happened" + return captured + + @pytest.mark.asyncio + async def test_ws_url_dials_without_ssl(self): + captured = await self._dial("ws://127.0.0.1:8113/v1/responses") + + assert captured["ssl"] is None + + @pytest.mark.asyncio + async def test_wss_url_dials_with_the_shared_context(self): + from litellm.llms.custom_httpx.http_handler import get_shared_realtime_ssl_context + + captured = await self._dial("wss://127.0.0.1:8113/v1/responses") + + assert captured["ssl"] is get_shared_realtime_ssl_context() + + @pytest.mark.asyncio + async def test_wss_url_with_ssl_verify_off_dials_with_an_unverified_context(self, monkeypatch): + import ssl + + monkeypatch.setattr("litellm.llms.custom_httpx.http_handler._shared_realtime_ssl_context", False) + + captured = await self._dial("wss://127.0.0.1:8113/v1/responses") + + assert isinstance(captured["ssl"], ssl.SSLContext) + assert captured["ssl"].verify_mode is ssl.CERT_NONE + assert captured["ssl"].check_hostname is False + + class TestNativeWebSocketEncryptedContentAffinity: @pytest.mark.asyncio