fix(chatgpt): validate sideband ownership and normalize image files

docs/chatgpt-codex-routes.md

@@ -1,52 +0,0 @@
-# ChatGPT OAuth routes used by Codex
-
-The ChatGPT provider supports image generation and editing, structured Responses output, Realtime WebSockets, and direct WebRTC offer exchange using the existing ChatGPT OAuth credentials configured with `CHATGPT_TOKEN_DIR` and `CHATGPT_AUTH_FILE`
-
-## Models and transports
-
-| Model | Proxy route | Upstream route |
-| --- | --- | --- |
-| `gpt-image-2` | `/v1/images/generations` | ChatGPT `/backend-api/codex/images/generations` |
-| `gpt-image-2` | `/v1/images/edits` | ChatGPT `/backend-api/codex/images/edits` |
-| `codex-auto-review` | `/v1/responses` | ChatGPT `/backend-api/codex/responses` |
-| `gpt-realtime-1.5` | `/v1/realtime` WebSocket | OpenAI `/v1/realtime` with ChatGPT OAuth |
-| `gpt-live-1-codex` | `/v1/realtime/calls`, then `/v1/live/{call_id}` | ChatGPT call signaling, OpenAI live sideband |
-| `gpt-4o-mini-transcribe` | Nested in Realtime `session.audio.input.transcription.model` | Same Realtime session |
-
-Codex uses `gpt-image-2` for both image routes. There is no separate `gpt-image-2-edit` model in that contract. Memory extraction and consolidation use ordinary Responses models and need no additional transport
-
-Register deployments with `litellm_params.model: chatgpt/<model>`. These auxiliary models do not need to appear in the conversation model selector. This contribution uses the existing single-account ChatGPT authentication contract
-
-## Images
-
-Both synchronous and asynchronous SDK calls support image generation and editing. Edits accept the usual `image` file input or Codex's JSON `images` array, containing one to five `{ "image_url": "data:image/png;base64,..." }` references. PNG, JPEG, WEBP, and HTTPS references are accepted. Masks and mixing `image` with `images` are rejected
-
-```python
-await litellm.aimage_edit(
-    model="chatgpt/gpt-image-2",
-    prompt="Change the blue circle to red",
-    images=[{"image_url": image_data_url}],
-)
-```
-
-## Guardian
-
-The Responses transformation preserves `text.format`, including strict JSON schemas. Custom Codex providers use `/responses` for `codex-auto-review`. The separate native `/guardian` route can be disabled by the upstream service; registering an alias does not enable that route
-
-## Realtime and GPT-Live
-
-Standard Realtime WebSockets use the OpenAI host with the selected ChatGPT access token and account header. Client protocol headers are forwarded without forwarding the client's proxy authorization header
-
-Codex WebRTC offers can use JSON or multipart `sdp` and `session` fields with an ordinary LiteLLM key. Signaling preserves the session shape and the `intent` and `architecture` query parameters. Existing raw SDP requests authenticated with an encrypted ephemeral key retain their existing route
-
-For GPT-Live, send `OpenAI-Alpha: quicksilver=v2`, `intent=quicksilver&architecture=avas`, and a Codex voice such as `sol`. Do not add the GA `session.type: realtime` field to a Frameless Bidi session. A generic `Voice session access denied` error can mean an unsupported voice; it is not sufficient evidence of missing account entitlement
-
-The returned `Location` contains an encrypted call identifier. The sideband connection must use the same LiteLLM bearer key and retain access to the requested alias. The identifier expires after one hour and remains valid across proxy workers sharing the same salt key. OAuth tokens are never returned to clients
-
-The direct `/v1/live?model=gpt-live-1-codex` WebSocket is forwarded, but upstream access can differ from WebRTC. A successful WebRTC call does not establish permission for direct live sessions. Upstream errors remain visible; the proxy does not replace the model, voice, or transport silently
-
-## Verification
-
-The integration was exercised with real image generation and JSON edits, strict Guardian JSON output, Realtime text and audio output, audio transcription, and a GPT-Live WebRTC session with a successful sideband context acknowledgment. Expired, malformed, tampered, and wrong-owner call identifiers have regression coverage
-
-References: [Codex source](https://github.com/openai/codex), [OpenClaw voice authentication](https://docs.openclaw.ai/providers/openai/voice-and-speech), and [Pi Codex signaling implementation](https://github.com/monotykamary/pi-better-openai/blob/main/src/live/transport.ts). These upstream capabilities may change independently of LiteLLM

gateway/routes/allowlist.py

@@ -104,6 +104,8 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
     "/{provider}/",
     "/toolset/",
     # Realtime / streaming
+    "/v1/live",
+    "/live",
     "/v1/realtime",
     "/realtime",
     # Health & ops

litellm/images/main.py

@@ -102,9 +102,7 @@ async def aimage_generation(*args, **kwargs) -> ImageResponse:
         ctx: Final = contextvars.copy_context()
         func_with_context: Final = partial(ctx.run, func)
 
-        _, custom_llm_provider, _, _ = get_llm_provider(
-            model=model, api_base=kwargs.get("api_base", None), litellm_params=GenericLiteLLMParams(**kwargs)
-        )
+        _, custom_llm_provider, _, _ = get_llm_provider(model=model, api_base=kwargs.get("api_base", None))
 
         # Await normally
         init_response: Final = await loop.run_in_executor(None, func_with_context)
@@ -229,7 +227,6 @@ def image_generation(
                 model=model,
                 custom_llm_provider=custom_llm_provider,
                 api_base=api_base,
-                litellm_params=GenericLiteLLMParams(**kwargs),
             )
         else:
             model = "dall-e-2"
@@ -789,7 +786,6 @@ def image_edit(
         model, custom_llm_provider, _, _ = get_llm_provider(
             model=model or DEFAULT_IMAGE_ENDPOINT_MODEL,
             custom_llm_provider=custom_llm_provider,
-            litellm_params=litellm_params,
         )
 
         # Check for custom provider
@@ -1007,7 +1003,7 @@ async def aimage_edit(
         # get custom llm provider so we can use this for mapping exceptions
         if custom_llm_provider is None:
             _, custom_llm_provider, _, _ = litellm.get_llm_provider(
-                model=model, api_base=local_vars.get("base_url", None), litellm_params=GenericLiteLLMParams(**kwargs)
+                model=model, api_base=local_vars.get("base_url", None)
             )
 
         func: Final = partial(

litellm/llms/chatgpt/images.py

@@ -1,5 +1,7 @@
 import base64
+import os
 from collections.abc import Mapping, Sequence
+from pathlib import Path
 from types import MappingProxyType
 from typing import Final
 
@@ -7,6 +9,7 @@ from httpx._types import FileTypes as HTTPFileTypes
 from httpx._types import RequestFiles
 from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
 
+from litellm.images.utils import ImageEditRequestUtils
 from litellm.llms.openai.image_edit.transformation import OpenAIImageEditConfig
 from litellm.llms.openai.image_generation.gpt_transformation import GPTImageGenerationConfig
 from litellm.types.llms.openai import AllMessageValues, FileTypes
@@ -21,14 +24,26 @@ class ReferenceImage(BaseModel):
     image_url: str = Field(pattern=r"^(data:image/(png|jpeg|webp);base64,|https://)")
 
 
-def encode_reference(file: HTTPFileTypes) -> dict[str, str]:  # mutable-ok: image handler requires dictionaries
+def encode_reference(
+    file: HTTPFileTypes | FileTypes,
+) -> dict[str, str]:  # mutable-ok: image handler requires dictionaries
     content: Final = file[1] if isinstance(file, tuple) else file
-    content_type: Final = file[2] if isinstance(file, tuple) and len(file) >= 3 else "image/png"
+    raw: Final = (
+        Path(os.fsdecode(content)).read_bytes()
+        if isinstance(content, os.PathLike)
+        else content.encode()
+        if isinstance(content, str)
+        else content
+        if isinstance(content, bytes)
+        else content.read()
+    )
+    content_type: Final = (
+        file[2]
+        if isinstance(file, tuple) and len(file) >= 3 and file[2]
+        else ImageEditRequestUtils.get_image_content_type(raw)
+    )
     if content_type not in ("image/png", "image/jpeg", "image/webp"):
         raise ValueError("Reference images must be PNG, JPEG, or WEBP")
-    raw: Final = (
-        content.encode() if isinstance(content, str) else content if isinstance(content, bytes) else content.read()
-    )
     return {  # mutable-ok: JSON request serialization
         "image_url": f"data:{content_type};base64," + base64.b64encode(raw).decode("ascii")
     }
@@ -101,7 +116,7 @@ class ChatGPTImageEditConfig(OpenAIImageEditConfig):
         self,
         model: str,
         prompt: str | None,
-        image: FileTypes | None,
+        image: FileTypes | Sequence[FileTypes] | None,
         image_edit_optional_request_params: Mapping[str, object],
         litellm_params: GenericLiteLLMParams,
         headers: Mapping[str, object],
@@ -122,16 +137,13 @@ class ChatGPTImageEditConfig(OpenAIImageEditConfig):
                 "images": tuple(item.model_dump() for item in validated),
             }, ()
 
-        data, files = super().transform_image_edit_request(
-            model,
-            prompt,
-            image,
-            dict(image_edit_optional_request_params),  # mutable-ok: parent edit adapter requires dictionaries
-            litellm_params,
-            dict(headers),  # mutable-ok: parent edit adapter requires dictionaries
-        )
-        parts: Final = files.items() if isinstance(files, Mapping) else files
-        encoded: Final = tuple(encode_reference(file) for field, file in parts if field == "image[]")
+        inputs: Final = tuple(image) if isinstance(image, list) else (image,) if image is not None else ()
+        encoded: Final = tuple(encode_reference(file) for file in inputs)
         if not 1 <= len(encoded) <= 5:
             raise ValueError("images must contain between 1 and 5 reference images")
-        return {**data, "images": encoded}, ()  # mutable-ok: JSON request serialization
+        return {  # mutable-ok: JSON request serialization
+            "model": model,
+            "prompt": prompt,
+            **image_edit_optional_request_params,
+            "images": encoded,
+        }, ()

litellm/proxy/proxy_server.py

@@ -11598,7 +11598,7 @@ async def codex_live_sideband_endpoint(
     call_id: str,
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket),
 ) -> None:
-    from litellm.proxy.realtime_endpoints.codex import codex_realtime_sideband
+    from litellm.llms.chatgpt.codex import codex_realtime_sideband
 
     await codex_realtime_sideband(websocket, call_id, user_api_key_dict)
 
@@ -11619,8 +11619,8 @@ async def realtime_websocket_endpoint(
     ),
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket),
 ):
-    if call_id is not None and call_id.startswith("rtc_litellm_"):
-        from litellm.proxy.realtime_endpoints.codex import codex_realtime_sideband
+    if call_id is not None:
+        from litellm.llms.chatgpt.codex import codex_realtime_sideband
 
         await codex_realtime_sideband(websocket, call_id, user_api_key_dict)
         return

litellm/proxy/realtime_endpoints/endpoints.py

@@ -376,7 +376,7 @@ async def proxy_realtime_calls(
     fastapi_response: Response,
 ) -> Response:
     if request.headers.get("content-type", "").split(";", 1)[0] in ("application/json", "multipart/form-data"):
-        from litellm.proxy.realtime_endpoints.codex import create_codex_realtime_call
+        from litellm.llms.chatgpt.codex import create_codex_realtime_call
 
         return await create_codex_realtime_call(request)
 

litellm/realtime_api/main.py

@@ -141,7 +141,6 @@ async def acreate_realtime_client_secret(
         model=model_name,
         api_base=litellm_params.api_base,
         api_key=litellm_params.api_key,
-        litellm_params=litellm_params,
     )
     (
         provider_config,
@@ -210,7 +209,6 @@ async def acreate_realtime_transcription_session(
         model=model_name,
         api_base=litellm_params.api_base,
         api_key=litellm_params.api_key,
-        litellm_params=litellm_params,
     )
     (
         provider_config,
@@ -272,7 +270,6 @@ async def arealtime_calls(
         model=model_name,
         api_base=litellm_params.api_base,
         api_key=litellm_params.api_key,
-        litellm_params=litellm_params,
     )
     provider_config, resolved_api_base, _ = _get_realtime_http_provider_config(
         custom_llm_provider=custom_llm_provider,
@@ -380,7 +377,6 @@ async def _arealtime(
         model=model,
         api_base=api_base,
         api_key=api_key,
-        litellm_params=litellm_params,
     )
 
     # If the client supplied `model` in the URL, ensure it uses the normalized

tests/test_litellm/llms/chatgpt/test_codex.py

@@ -0,0 +1,212 @@
+import hashlib
+import time
+
+import pytest
+from fastapi import HTTPException, WebSocket
+
+from litellm.proxy._types import ProxyException, UserAPIKeyAuth
+from litellm.llms.chatgpt import codex
+from litellm.llms.chatgpt.codex import CodexRealtimeCall, decode_call, encode_call
+
+
+def test_sideband_token_binds_owner_and_model(monkeypatch):
+    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
+    call = CodexRealtimeCall(
+        call_id="rtc_test",
+        model="gpt-live-1-codex",
+        alias="gpt-live-1-codex",
+        owner=hashlib.sha256(b"Bearer test-owner").hexdigest(),
+        expires_at=time.time() + 300,
+    )
+    token = encode_call(call)
+    assert "/" not in token
+    assert decode_call(token, "Bearer test-owner") == call
+    with pytest.raises(HTTPException) as error:
+        decode_call(token, "Bearer different-owner")
+    assert error.value.status_code == 403
+    with pytest.raises(HTTPException):
+        decode_call(token[:30] + "tampered" + token[30:], "Bearer test-owner")
+
+
+def test_sideband_rejects_expired_token(monkeypatch):
+    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
+    call = CodexRealtimeCall(
+        call_id="rtc_test",
+        model="gpt-realtime-1.5",
+        alias="gpt-realtime-1.5",
+        owner=hashlib.sha256(b"Bearer test-owner").hexdigest(),
+        expires_at=time.time() - 1,
+    )
+    with pytest.raises(HTTPException):
+        decode_call(encode_call(call), "Bearer test-owner")
+
+
+@pytest.mark.parametrize("token", ["", "rtc_other", "rtc_litellm_%%%%", "rtc_litellm_a"])
+def test_sideband_rejects_malformed_tokens(token):
+    with pytest.raises(HTTPException):
+        decode_call(token, "Bearer test-owner")
+
+
+@pytest.mark.asyncio
+async def test_sideband_rejects_revoked_model_access(monkeypatch):
+    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
+    call = CodexRealtimeCall(
+        call_id="rtc_test",
+        model="gpt-live-1-codex",
+        alias="voice",
+        owner=hashlib.sha256(b"Bearer test-owner").hexdigest(),
+        expires_at=time.time() + 300,
+    )
+    sent = []
+
+    async def receive():
+        return {"type": "websocket.connect"}
+
+    async def send(message):
+        sent.append(message)
+
+    async def deny_model(**kwargs):
+        raise ProxyException("Model access revoked", "auth_error", "model", 403)
+
+    monkeypatch.setattr(codex, "can_key_call_resolved_model", deny_model)
+    websocket = WebSocket(
+        {"type": "websocket", "headers": [(b"authorization", b"Bearer test-owner")]}, receive, send
+    )
+    await codex.codex_realtime_sideband(websocket, encode_call(call), UserAPIKeyAuth())
+    assert sent == [{"type": "websocket.close", "code": 1008, "reason": "Invalid realtime call"}]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("call_id", ["rtc_raw", "", "rtc_litellm_invalid"])
+async def test_realtime_endpoint_rejects_untrusted_call_ids(monkeypatch, call_id):
+    from unittest.mock import AsyncMock
+    from fastapi import WebSocket
+    from litellm.proxy import proxy_server as server
+    from litellm.proxy._types import UserAPIKeyAuth
+
+    sent = []
+
+    async def receive():
+        return {"type": "websocket.connect"}
+
+    async def send(message):
+        sent.append(message)
+
+    route = AsyncMock()
+    monkeypatch.setattr(server, "route_request", route)
+    websocket = WebSocket({"type": "websocket", "headers": [], "query_string": b""}, receive, send)
+    await server.realtime_websocket_endpoint(
+        websocket, model="gpt-realtime-1.5", call_id=call_id,
+        intent=None, guardrails=None, user_api_key_dict=UserAPIKeyAuth()
+    )
+    assert sent == [{"type": "websocket.close", "code": 1008, "reason": "Invalid realtime call"}]
+    route.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("multipart", [False, True])
+async def test_offer_exchange_wraps_call_and_filters_client_headers(monkeypatch, multipart):
+    import json
+    from unittest.mock import AsyncMock
+
+    import httpx
+    from fastapi import Request, WebSocket
+    from litellm.proxy import common_request_processing, proxy_server
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.llms.chatgpt import codex
+    import litellm
+
+    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
+    session = {"model": "voice-alias", "audio": {"output": {"voice": "sol"}}}
+    if multipart:
+        body_request = httpx.Request("POST", "http://test/v1/realtime/calls", files={
+            "sdp": (None, "v=0\r\n"), "session": (None, json.dumps(session))
+        })
+    else:
+        body_request = httpx.Request("POST", "http://test/v1/realtime/calls", json={"sdp": "v=0\r\n", "session": session})
+    body = body_request.read()
+
+    async def receive():
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    request = Request({"type": "http", "method": "POST", "path": "/v1/realtime/calls",
+        "query_string": b"intent=quicksilver&architecture=avas&untrusted=bad",
+        "headers": [(b"content-type", body_request.headers["content-type"].encode()),
+                    (b"authorization", b"Bearer owner"), (b"openai-alpha", b"quicksilver=v2"),
+                    (b"x-untrusted", b"bad")]}, receive)
+    auth = UserAPIKeyAuth()
+    authenticate = AsyncMock(return_value=auth)
+    authorize = AsyncMock()
+    monkeypatch.setattr(codex, "user_api_key_auth", authenticate)
+    monkeypatch.setattr(codex, "can_key_call_resolved_model", authorize)
+
+    class Processor:
+        def __init__(self, data):
+            self.data = data
+
+        async def common_processing_pre_call_logic(self, **kwargs):
+            assert kwargs["user_api_key_dict"] is auth
+            return self.data, None
+
+    monkeypatch.setattr(common_request_processing, "ProxyBaseLLMRequestProcessing", Processor)
+
+    async def route(**kwargs):
+        data = kwargs["data"]
+        assert data["sdp_body"] == b"v=0\r\n"
+        assert data["session"] == session
+        assert data["extra_headers"] == {"openai-alpha": "quicksilver=v2"}
+        assert data["extra_query"] == {"intent": "quicksilver", "architecture": "avas"}
+
+        async def respond():
+            return httpx.Response(201, content=b"v=0\r\nanswer", headers={"Location": "/v1/realtime/calls/rtc_private"},
+                extensions={"chatgpt_realtime": {"model": "gpt-live-1-codex"}})
+        return respond()
+
+    monkeypatch.setattr(proxy_server, "route_request", route)
+    response = await codex.create_codex_realtime_call(request)
+    assert response.status_code == 201
+    assert response.body == b"v=0\r\nanswer"
+    token = response.headers["location"].rsplit("/", 1)[-1]
+    call = codex.decode_call(token, "Bearer owner")
+    assert call.call_id == "rtc_private"
+    assert call.alias == "voice-alias"
+    assert call.model == "gpt-live-1-codex"
+    assert "rtc_private" not in token
+    assert time.time() < call.expires_at < time.time() + 3601
+    authorize.assert_awaited_once()
+
+    sent = []
+
+    async def send(message):
+        sent.append(message)
+
+    async def receive_ws():
+        return {"type": "websocket.connect"}
+
+    websocket = WebSocket({"type": "websocket", "headers": [(b"authorization", b"Bearer owner")]}, receive_ws, send)
+    forward = AsyncMock()
+    monkeypatch.setattr(litellm, "_arealtime", forward)
+    await codex.codex_realtime_sideband(websocket, token, auth)
+    assert sent[0]["type"] == "websocket.accept"
+    assert forward.await_args.kwargs["chatgpt_realtime_call_id"] == "rtc_private"
+    assert forward.await_args.kwargs["model"] == "chatgpt/gpt-live-1-codex"
+    assert authorize.await_count == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("body", [b"not json", b'{}', b'{"sdp":"v=0","session":{}}'])
+async def test_invalid_offers_fail_before_authentication(monkeypatch, body):
+    from unittest.mock import AsyncMock
+    from fastapi import Request
+    from litellm.llms.chatgpt import codex
+
+    async def receive():
+        return {"type": "http.request", "body": body}
+
+    request = Request({"type": "http", "headers": [(b"content-type", b"application/json")]}, receive)
+    authenticate = AsyncMock()
+    monkeypatch.setattr(codex, "user_api_key_auth", authenticate)
+    with pytest.raises(HTTPException) as error:
+        await codex.create_codex_realtime_call(request)
+    assert error.value.status_code == 400
+    authenticate.assert_not_called()

tests/test_litellm/llms/chatgpt/test_images.py

@@ -104,3 +104,15 @@ async def test_async_codex_edit_without_multipart_image(chatgpt_tokens):
     assert str(requests[0].url).endswith("/codex/images/edits")
     assert requests[0].headers["content-type"] == "application/json"
     await client.client.aclose()
+
+
+@pytest.mark.parametrize("as_tuple", [False, True])
+def test_edit_accepts_filesystem_path(tmp_path, as_tuple):
+    image = tmp_path / "reference.png"
+    image.write_bytes(b"reference image bytes")
+    data, files = ChatGPTImageEditConfig().transform_image_edit_request(
+        "gpt-image-2", "edit", ("reference.png", image, "image/png") if as_tuple else image,
+        {}, GenericLiteLLMParams(), {}
+    )
+    assert not files
+    assert data["images"] == ({"image_url": "data:image/png;base64," + base64.b64encode(image.read_bytes()).decode()},)

tests/test_litellm/proxy/realtime_endpoints/test_codex.py

@@ -1,76 +0,0 @@
-import hashlib
-import time
-
-import pytest
-from fastapi import HTTPException, WebSocket
-
-from litellm.proxy._types import ProxyException, UserAPIKeyAuth
-from litellm.proxy.realtime_endpoints import codex
-from litellm.proxy.realtime_endpoints.codex import CodexRealtimeCall, decode_call, encode_call
-
-
-def test_sideband_token_binds_owner_and_model(monkeypatch):
-    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
-    call = CodexRealtimeCall(
-        call_id="rtc_test",
-        model="gpt-live-1-codex",
-        alias="gpt-live-1-codex",
-        owner=hashlib.sha256(b"Bearer test-owner").hexdigest(),
-        expires_at=time.time() + 300,
-    )
-    token = encode_call(call)
-    assert "/" not in token
-    assert decode_call(token, "Bearer test-owner") == call
-    with pytest.raises(HTTPException) as error:
-        decode_call(token, "Bearer different-owner")
-    assert error.value.status_code == 403
-    with pytest.raises(HTTPException):
-        decode_call(token[:30] + "tampered" + token[30:], "Bearer test-owner")
-
-
-def test_sideband_rejects_expired_token(monkeypatch):
-    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
-    call = CodexRealtimeCall(
-        call_id="rtc_test",
-        model="gpt-realtime-1.5",
-        alias="gpt-realtime-1.5",
-        owner=hashlib.sha256(b"Bearer test-owner").hexdigest(),
-        expires_at=time.time() - 1,
-    )
-    with pytest.raises(HTTPException):
-        decode_call(encode_call(call), "Bearer test-owner")
-
-
-@pytest.mark.parametrize("token", ["", "rtc_other", "rtc_litellm_%%%%", "rtc_litellm_a"])
-def test_sideband_rejects_malformed_tokens(token):
-    with pytest.raises(HTTPException):
-        decode_call(token, "Bearer test-owner")
-
-
-@pytest.mark.asyncio
-async def test_sideband_rejects_revoked_model_access(monkeypatch):
-    monkeypatch.setenv("LITELLM_SALT_KEY", "test-only-salt-for-codex-realtime")
-    call = CodexRealtimeCall(
-        call_id="rtc_test",
-        model="gpt-live-1-codex",
-        alias="voice",
-        owner=hashlib.sha256(b"Bearer test-owner").hexdigest(),
-        expires_at=time.time() + 300,
-    )
-    sent = []
-
-    async def receive():
-        return {"type": "websocket.connect"}
-
-    async def send(message):
-        sent.append(message)
-
-    async def deny_model(**kwargs):
-        raise ProxyException("Model access revoked", "auth_error", "model", 403)
-
-    monkeypatch.setattr(codex, "can_key_call_resolved_model", deny_model)
-    websocket = WebSocket(
-        {"type": "websocket", "headers": [(b"authorization", b"Bearer test-owner")]}, receive, send
-    )
-    await codex.codex_realtime_sideband(websocket, encode_call(call), UserAPIKeyAuth())
-    assert sent == [{"type": "websocket.close", "code": 1008, "reason": "Invalid realtime call"}]

tests/test_litellm/realtime_api/test_main.py

@@ -61,7 +61,7 @@ def _run_client_secret(session, model, monkeypatch):
         captured.update(kwargs)
         return object()
 
-    def mock_get_llm_provider(model, api_base, api_key, litellm_params=None):
+    def mock_get_llm_provider(model, api_base, api_key):
         return model, "openai", None, api_base
 
     monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
@@ -161,7 +161,7 @@ async def test_arealtime_vertex_branch_resolves_credentials_under_a_bound(monkey
     async def hanging_token_refresh(**kwargs):
         await asyncio.sleep(30)
 
-    def mock_get_llm_provider(model, api_base, api_key, litellm_params=None):
+    def mock_get_llm_provider(model, api_base, api_key):
         return model, "vertex_ai", None, api_base
 
     monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)