scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

fix(realtime): preserve wss ssl semantics and move live guardrail test

Browse source 
Keep TLS enabled for wss realtime sessions while honoring SSL_VERIFY=False via a no-verify SSLContext, move the OpenAI live guardrail test into llm_translation, and dedupe duplicated guardrail-detection helpers to prevent drift.

Made-with: Cursor
This commit is contained in:
Ishaan Jaffer 2026-02-26 16:45:55 -08:00
parent 7807e36b40
commit 0961900839
3 changed files with 15 additions and 31 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

@ -269,25 +269,7 @@ class RealTimeStreaming:
any guardrail that would actually check the transcript also disables
auto-response before the transcript arrives.
"""
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
_realtime_event_types = [
GuardrailEventHooks.realtime_input_transcription,
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
]
return any(
isinstance(cb, CustomGuardrail)
and any(
cb.should_run_guardrail(
data=self.request_data,
event_type=et,
)
for et in _realtime_event_types
)
for cb in litellm.callbacks
)
return self._has_realtime_guardrails()
async def run_realtime_guardrails(
self,

6
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -1,4 +1,5 @@
import json
import ssl
from typing import (
TYPE_CHECKING,
Any,
@ -4675,7 +4676,10 @@ class BaseLLMHTTPHandler:
try:
ssl_context = get_shared_realtime_ssl_context()
if url.startswith("wss://") and ssl_context is False:
ssl_context = True
# Keep TLS for wss:// while honoring SSL_VERIFY=False semantics.
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
async with websockets.connect( # type: ignore
url,
additional_headers=headers,

20
tests/test_litellm/litellm_core_utils/test_realtime_guardrails_openai.py → tests/llm_translation/realtime/test_realtime_guardrails_openai.py
View file

@ -4,12 +4,12 @@ Integration tests for RealTimeStreaming guardrails against a live OpenAI backend
These tests require OPENAI_API_KEY and are skipped if not set.
They verify end-to-end that:
1. A text message blocked by a guardrail → error event sent to client, NO AI response.
2. A voice transcript blocked by a guardrail → error event sent, response.create NOT sent.
1. A text message blocked by a guardrail -> error event sent to client, NO AI response.
2. A voice transcript blocked by a guardrail -> error event sent, response.create NOT sent.
3. A clean text message passes through and triggers a real OpenAI response.
Run with:
poetry run pytest tests/test_litellm/litellm_core_utils/test_realtime_guardrails_openai.py -v -s
poetry run pytest tests/llm_translation/realtime/test_realtime_guardrails_openai.py -v -s
"""
import asyncio
@ -32,7 +32,7 @@ OPENAI_REALTIME_URL = (
pytestmark = pytest.mark.skipif(
not OPENAI_API_KEY,
reason="OPENAI_API_KEY not set — skipping OpenAI realtime integration tests",
reason="OPENAI_API_KEY not set - skipping OpenAI realtime integration tests",
)
# A unique phrase guaranteed NOT to appear in normal assistant output.
@ -48,7 +48,7 @@ class PhraseBlockingGuardrail(CustomGuardrail):
for text in inputs.get("texts", []):
if BLOCKED_PHRASE in text:
raise ValueError(
f"Content blocked: contains forbidden test phrase."
"Content blocked: contains forbidden test phrase."
)
return inputs
@ -128,11 +128,11 @@ async def test_text_message_blocked_by_guardrail_no_ai_response():
) as backend_ws:
streaming, input_queue = await _build_streaming(client_events, backend_ws)
# Start backend → client forwarding
# Start backend -> client forwarding
backend_task = asyncio.create_task(
streaming.backend_to_client_send_messages()
)
# Start client → backend forwarding (reads from input_queue)
# Start client -> backend forwarding (reads from input_queue)
client_task = asyncio.create_task(streaming.client_ack_messages())
try:
@ -169,7 +169,6 @@ async def test_text_message_blocked_by_guardrail_no_ai_response():
# --- Assertions ---
event_types = [e.get("type") for e in client_events]
print(f"\n[test] client events received: {event_types}")
# 1. Must have received guardrail error
error_events = [e for e in client_events if e.get("type") == "error"]
@ -190,7 +189,7 @@ async def test_text_message_blocked_by_guardrail_no_ai_response():
f"Expected guardrail message in transcript delta, got: {event_types}"
)
# 3. No real AI response should have been generated — response.done would only
# 3. No real AI response should have been generated - response.done would only
# appear if we sent a response.create and OpenAI replied. We allow it in the
# synthetic form (empty output=[]) but NOT with actual AI content.
done_events = [e for e in client_events if e.get("type") == "response.done"]
@ -214,7 +213,7 @@ async def test_text_message_blocked_by_guardrail_no_ai_response():
async def test_voice_transcript_blocked_by_guardrail():
"""
Simulate a backend-side voice transcription event containing the blocked phrase.
Guardrail must block it — no response.create sent to OpenAI.
Guardrail must block it - no response.create sent to OpenAI.
"""
from websockets.exceptions import ConnectionClosed
@ -247,7 +246,6 @@ async def test_voice_transcript_blocked_by_guardrail():
await streaming.backend_to_client_send_messages()
event_types = [e.get("type") for e in client_events]
print(f"\n[test] client events received: {event_types}")
# 1. Error event must be sent to client
error_events = [e for e in client_events if e.get("type") == "error"]