mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-23 05:34:41 +00:00
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://.
This commit is contained in:
mateo-berri
parent
0e26edfdb9
commit
dcd949350d
8 changed files with 533 additions and 28 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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",)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue