diff --git a/litellm/proxy/realtime_endpoints/call_sessions.py b/litellm/proxy/realtime_endpoints/call_sessions.py index cf4937f3252..0827d4b5057 100644 --- a/litellm/proxy/realtime_endpoints/call_sessions.py +++ b/litellm/proxy/realtime_endpoints/call_sessions.py @@ -400,9 +400,7 @@ async def _create_codex_realtime_call(request: Request) -> Response: call: Final = parse_call_response( response, alias=model, - owner=hashlib.sha256( - f"Bearer {owner_key}".encode(), usedforsecurity=False - ).hexdigest(), + owner=hashlib.sha256(f"Bearer {owner_key}".encode(), usedforsecurity=False).hexdigest(), expires_at=time.time() + 3600, ) except ValueError as exc: diff --git a/litellm/proxy/realtime_endpoints/live.py b/litellm/proxy/realtime_endpoints/live.py index 8f68e870497..663b31cc5fb 100644 --- a/litellm/proxy/realtime_endpoints/live.py +++ b/litellm/proxy/realtime_endpoints/live.py @@ -344,6 +344,48 @@ def _pinned(handle: LiveHandle) -> LiveDeployment: return _DEPLOYMENT.validate_python(_mutable(handle.deployment)) +def _validate_pinned_deployment(handle: LiveHandle) -> LiveDeployment: + """Reject handles whose deployment was removed, blocked, or replaced.""" + from litellm.proxy import proxy_server as server + + deployment: Final = _pinned(handle) + router = server.llm_router + if router is None or deployment.model_id is None: + raise HTTPException(410, "Live session deployment is no longer available") + configured_raw = router.get_deployment(model_id=deployment.model_id) + if configured_raw is None: + raise HTTPException(410, "Live session deployment is no longer available") + configured_data: Final = ( + configured_raw.model_dump() + if hasattr(configured_raw, "model_dump") + else vars(configured_raw) + if not isinstance(configured_raw, Mapping) + else configured_raw + ) + configured: Final = _MAPPING.validate_python(configured_data) + model_info: Final = _MAPPING.validate_python(configured.get("model_info", _EMPTY)) + if model_info.get("blocked") is True: + raise HTTPException(410, "Live session deployment is no longer available") + params: Final = _object(configured["litellm_params"]) + qualified: Final = str(params.get("model", "")) + prefix, _, suffix = qualified.partition("/") + provider: Final = prefix if prefix in ("openai", "chatgpt") else "openai" + upstream: Final = suffix if prefix in ("openai", "chatgpt") else qualified + if any( + ( + deployment.model != upstream, + deployment.provider != provider, + str(model_info.get("id")) != deployment.model_id, + params.get("api_base") != deployment.api_base, + params.get("api_key") != deployment.api_key, + (params.get("extra_headers") or _EMPTY) != deployment.extra_headers, + (params.get("extra_query") or _EMPTY) != deployment.extra_query, + ) + ): + raise HTTPException(410, "Live session deployment is no longer available") + return deployment + + def _new_handle( session_id: str, alias: str, @@ -1170,7 +1212,9 @@ async def _create(request: Request, token: str | None = None) -> Response: _request(request, MappingProxyType({**body, "model": model})), ownership.auth, model, ownership=ownership ) as prepared: await _authorize_fork_policy(_processed_body(body, prepared.processed), source, ownership.auth) - deployment: Final = _pinned(source) if source else await _deployment(model, prepared.processed) + deployment: Final = ( + _validate_pinned_deployment(source) if source else await _deployment(model, prepared.processed) + ) transport: Final = LiveTransport(deployment, request.headers) path: Final = live_session_path(source.session_id, "fork") if source else "live/sessions" response: Final = await transport.request( @@ -1412,7 +1456,9 @@ async def websocket_live_session(websocket: WebSocket, session_id: str | None = ) else: await _authorize_fork_policy(_processed_body(first, prepared.processed), source, ownership.auth) - deployment: Final = _pinned(source) if source else await _deployment(model, prepared.processed) + deployment: Final = ( + _validate_pinned_deployment(source) if source else await _deployment(model, prepared.processed) + ) path: Final = ( live_session_path(source.session_id, "attach" if attached else "fork") if source diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_live.py b/tests/test_litellm/proxy/realtime_endpoints/test_live.py index d4df56345cd..950dd384054 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_live.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_live.py @@ -26,11 +26,14 @@ def encryption_key(monkeypatch): monkeypatch.setenv("LITELLM_SALT_KEY", "test-live-encryption-key") -def handle(owner="owner"): +def handle(owner="owner", model_id=None): + deployment = {"model": "gpt-live", "provider": "openai"} + if model_id is not None: + deployment["model_id"] = model_id return live._new_handle( "sess_upstream", "voice", - LiveDeployment(model="gpt-live"), + LiveDeployment(**deployment), UserAPIKeyAuth(api_key=owner), None, ) @@ -128,6 +131,8 @@ def test_only_protocol_session_ids_are_rewritten_and_application_values_survive( @pytest.fixture def route_client(monkeypatch): + from litellm.proxy import proxy_server + auth = UserAPIKeyAuth(api_key="owner") deployment = LiveDeployment(model="gpt-live", provider="openai", api_key="upstream-key", model_id="deployment-a") transport = SimpleNamespace( @@ -153,6 +158,21 @@ def route_client(monkeypatch): monkeypatch.setattr(live, "_precall", precall) monkeypatch.setattr(live, "_deployment", selected) monkeypatch.setattr(live, "_supervise", supervised) + monkeypatch.setattr( + proxy_server, + "llm_router", + SimpleNamespace( + get_deployment=lambda model_id: ( + { + "model_name": "voice", + "litellm_params": {"model": "openai/gpt-live"}, + "model_info": {"id": model_id}, + } + if model_id == "deployment-a" + else None + ) + ), + ) factory = Mock(return_value=transport) monkeypatch.setattr(live, "LiveTransport", factory) app = FastAPI() @@ -198,8 +218,7 @@ def test_create_preserves_configuration_and_returns_owned_json_session(route_cli def test_fork_preserves_empty_overrides_and_pins_source_deployment(route_client): - source = handle() - source = source.model_copy(update={"deployment": {**source.deployment, "model_id": "deployment-a"}}) + source = handle(model_id="deployment-a") token = live.encode_session(source) body = {"session": {}, "transport": {"type": "webrtc", "sdp": "offer"}} result = route_client.client.post(f"/v1/live/sessions/{token}/fork", json=body) @@ -210,6 +229,39 @@ def test_fork_preserves_empty_overrides_and_pins_source_deployment(route_client) route_client.selected.assert_not_awaited() +@pytest.mark.parametrize( + "configured", + [ + None, + { + "model_name": "voice", + "litellm_params": {"model": "openai/gpt-replaced"}, + "model_info": {"id": "deployment-a"}, + }, + { + "model_name": "voice", + "litellm_params": {"model": "openai/gpt-live"}, + "model_info": {"id": "deployment-a", "blocked": True}, + }, + ], + ids=["removed", "replaced", "blocked"], +) +def test_fork_rejects_removed_or_replaced_source_deployment(route_client, monkeypatch, configured): + from litellm.proxy import proxy_server + + monkeypatch.setattr(proxy_server.llm_router, "get_deployment", lambda model_id: configured) + token = live.encode_session(handle(model_id="deployment-a")) + + result = route_client.client.post( + f"/v1/live/sessions/{token}/fork", + json={"session": {}, "transport": {"type": "webrtc", "sdp": "offer"}}, + ) + + assert result.status_code == 410 + assert "no longer available" in result.json()["detail"] + route_client.transport.request.assert_not_awaited() + + def test_fork_cannot_change_model_even_to_same_alias(route_client): token = live.encode_session(handle()) response = route_client.client.post(f"/v1/live/sessions/{token}/fork", json={"session": {"model": "voice"}}) @@ -706,7 +758,7 @@ async def test_explicit_fork_backend_is_authorized_even_when_startup_policy_diff def test_restricted_client_fork_can_inherit_delegation(route_client): route_client.auth.models = ["voice"] body = {"session": {}} - token = live.encode_session(handle()) + token = live.encode_session(handle(model_id="deployment-a")) route_client.transport.request.return_value = httpx.Response( 200, json={"session": {"id": "sess_fork"}, "transport": {"type": "webrtc", "sdp": "answer"}} ) @@ -1942,9 +1994,7 @@ def test_websocket_requires_api_key_then_session_start(route_client): ws.receive_json() assert anonymous.value.code == 1008 - with route_client.client.websocket_connect( - "/v1/live/sessions", headers={"Authorization": "Bearer owner"} - ) as ws: + with route_client.client.websocket_connect("/v1/live/sessions", headers={"Authorization": "Bearer owner"}) as ws: ws.send_json({"type": "ping"}) with pytest.raises(WebSocketDisconnect) as wrong_first: ws.receive_json() @@ -1969,7 +2019,7 @@ async def test_attached_socket_authorizes_policy_and_reuses_source_without_start monkeypatch.setattr(live, "RealTimeStreaming", AttachedStream) authorize = AsyncMock() monkeypatch.setattr(live, "_authorize_delegation", authorize) - token = live.encode_session(handle()) + token = live.encode_session(handle(model_id="deployment-a")) inbound = iter([{"type": "websocket.connect"}, {"type": "websocket.disconnect", "code": 1000}]) sent: list = [] @@ -2012,9 +2062,7 @@ def test_websocket_connection_failure_closes_with_internal_error(route_client): from starlette.websockets import WebSocketDisconnect route_client.transport.connect = AsyncMock(return_value=None) - with route_client.client.websocket_connect( - "/v1/live/sessions", headers={"Authorization": "Bearer owner"} - ) as ws: + with route_client.client.websocket_connect("/v1/live/sessions", headers={"Authorization": "Bearer owner"}) as ws: ws.send_json({"type": "session.start", "session": {"model": "voice"}}) with pytest.raises(WebSocketDisconnect) as internal: ws.receive_json()