fix(realtime): reject stale live deployment forks

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:

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

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()