diff --git a/README.md b/README.md index 5f2c066cdb..2c40817469 100644 --- a/README.md +++ b/README.md @@ -390,6 +390,10 @@ The Realtime API enables you to build low-latency, multi-modal conversational ex Under the hood the SDK uses the [`websockets`](https://websockets.readthedocs.io/en/stable/) library to manage connections. +The retained `client.beta.realtime` interface does not exchange workload-identity tokens or sign Bedrock requests. A client using either method must supply its own `Authorization` settings through `default_headers` or `connect(extra_headers=...)`. Otherwise, the SDK rejects the WebSocket connection locally. + +Connection headers override default headers case-insensitively, and `omit` removes a header. These settings preserve the caller's header choices without enabling provider authentication. X.509 workload identity supports HTTP only. Beta WebSockets always reject X.509 clients, even with explicit headers. + The Realtime API works through a combination of client-sent events and server-sent events. Clients can send events to do things like update session configuration or send text and audio inputs. Server events confirm when audio responses have completed, or when a text response from the model has been received. A full event reference can be found [here](https://platform.openai.com/docs/api-reference/realtime-client-events) and a guide can be found [here](https://platform.openai.com/docs/guides/realtime). Basic text based example: diff --git a/src/openai/lib/_client_auth.py b/src/openai/lib/_client_auth.py index 81e0c071b1..982d82ccac 100644 --- a/src/openai/lib/_client_auth.py +++ b/src/openai/lib/_client_auth.py @@ -51,6 +51,17 @@ def _has_omitted_header(headers: Headers, header: str) -> bool: ) +def _validate_beta_realtime_auth(self: _sdk_client.OpenAI | _sdk_client.AsyncOpenAI, extra_headers: Headers) -> None: + if _sdk_client.is_x509_workload_identity(self.workload_identity): + raise _sdk_client.OpenAIError("This authentication method is not supported by WebSocket connections") + if (self.workload_identity is not None or self._provider_runtime is not None) and not ( + _has_header(self.default_headers, "Authorization") or _has_header(extra_headers, "Authorization") + ): + # Retained beta connections allow caller-owned authentication, including + # an explicit Omit. They do not exchange or sign provider credentials. + raise _sdk_client.OpenAIError("This authentication method is not supported by WebSocket connections") + + def _set_request_api_key( self: _sdk_client.OpenAI | _sdk_client.AsyncOpenAI, options: FinalRequestOptions, api_key: str ) -> None: diff --git a/src/openai/resources/beta/realtime/realtime.py b/src/openai/resources/beta/realtime/realtime.py index f28086c468..76b837a53c 100644 --- a/src/openai/resources/beta/realtime/realtime.py +++ b/src/openai/resources/beta/realtime/realtime.py @@ -351,6 +351,9 @@ async def __aenter__(self) -> AsyncRealtimeConnection: await connection.close() ``` """ + from ....lib._client_auth import _validate_beta_realtime_auth + + _validate_beta_realtime_auth(self.__client, self.__extra_headers) try: from ....lib._websocket import _WebSocketConnect as connect except ImportError as exc: @@ -538,6 +541,9 @@ def __enter__(self) -> RealtimeConnection: connection.close() ``` """ + from ....lib._client_auth import _validate_beta_realtime_auth + + _validate_beta_realtime_auth(self.__client, self.__extra_headers) try: from websockets.sync.client import connect except ImportError as exc: diff --git a/tests/lib/test_beta_realtime_auth_policy.py b/tests/lib/test_beta_realtime_auth_policy.py new file mode 100644 index 0000000000..a9a194294b --- /dev/null +++ b/tests/lib/test_beta_realtime_auth_policy.py @@ -0,0 +1,151 @@ +"""Offline beta Realtime provider preflight and explicit-header compatibility.""" + +from __future__ import annotations + +import threading +from typing import Any + +import httpx2 +import pytest +from websockets.sync.server import serve + +import openai +from openai.auth import WorkloadIdentity, x509_workload_identity +from openai.providers import bedrock + + +@pytest.fixture(autouse=True) +def synthetic_environment(monkeypatch: pytest.MonkeyPatch) -> None: + for name in ( + "OPENAI_API_KEY", + "OPENAI_ADMIN_KEY", + "OPENAI_BASE_URL", + "OPENAI_ORG_ID", + "OPENAI_PROJECT_ID", + "OPENAI_CUSTOM_HEADERS", + "OPENAI_WEBHOOK_SECRET", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + + +def identity(kind: str, callbacks: list[str]) -> dict[str, Any]: + def token() -> str: + callbacks.append("token") + raise AssertionError("implicit provider credential callback must not run") + + if kind == "bedrock": + return {"provider": bedrock(region="us-east-1", token_provider=token)} + if kind == "workload": + return { + "workload_identity": WorkloadIdentity( + identity_provider_id="idp_synthetic", + service_account_id="sa_synthetic", + provider={"token_type": "jwt", "get_token": token}, + ) + } + return { + "workload_identity": x509_workload_identity( + identity_provider_id="idp_synthetic", service_account_id="sa_synthetic" + ) + } + + +async def connect( + mode: str, kind: str, defaults: dict[str, Any], extra: dict[str, Any] +) -> tuple[BaseException | None, list[str], list[list[str]]]: + callbacks: list[str] = [] + upgrades: list[list[str]] = [] + headers_captured = threading.Event() + + def issuer(_request: httpx2.Request) -> httpx2.Response: + callbacks.append("issuer") + raise AssertionError("implicit HTTP exchange must not run") + + def peer(socket: Any) -> None: + assert socket.request.path.startswith("/v1/realtime?") + assert socket.request.headers["X-Beta-Client"] == "synthetic-default" + assert socket.request.headers["X-Beta-Connection"] == "synthetic-extra" + upgrades.append(socket.request.headers.get_all("Authorization")) + headers_captured.set() + socket.close(1000) + + auth = identity(kind, callbacks) + defaults = {"X-Beta-Client": "synthetic-default", **defaults} + extra = {"X-Beta-Connection": "synthetic-extra", **extra} + refused: BaseException | None = None + with serve(peer, "127.0.0.1", 0, close_timeout=1) as server: + worker = threading.Thread(target=server.serve_forever) + worker.start() + try: + url = f"ws://127.0.0.1:{server.socket.getsockname()[1]}/v1" + if mode == "sync": + with openai.OpenAI( + **auth, + default_headers=defaults, + websocket_base_url=url, + http_client=httpx2.Client(transport=httpx2.MockTransport(issuer), trust_env=False), + max_retries=0, + ) as client: + try: + with client.beta.realtime.connect(model="gpt-realtime", extra_headers=extra): + assert headers_captured.wait(3), "Peer did not record upgrade headers" + except openai.OpenAIError as failure: + refused = failure + else: + async with openai.AsyncOpenAI( + **auth, + default_headers=defaults, + websocket_base_url=url, + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(issuer), trust_env=False), + max_retries=0, + ) as client: + try: + async with client.beta.realtime.connect(model="gpt-realtime", extra_headers=extra): + assert headers_captured.wait(3), "Peer did not record upgrade headers" + except openai.OpenAIError as failure: + refused = failure + finally: + server.shutdown() + worker.join(3) + assert not worker.is_alive() + return refused, callbacks, upgrades + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("kind", ["workload", "x509", "bedrock"]) +async def test_bare_beta_provider_is_rejected_before_credentials_or_upgrade(mode: str, kind: str) -> None: + refused, callbacks, upgrades = await connect(mode, kind, {}, {}) + assert refused is not None and "not supported by WebSocket" in str(refused) + assert callbacks == [] and upgrades == [] + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_x509_explicit_header_does_not_enable_unsupported_websocket_profile(mode: str) -> None: + refused, callbacks, upgrades = await connect(mode, "x509", {}, {"Authorization": "Bearer synthetic-manual"}) + assert refused is not None and "not supported by WebSocket" in str(refused) + assert callbacks == [] and upgrades == [] + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("kind", ["workload", "bedrock"]) +@pytest.mark.parametrize("header_profile", ["default", "extra", "override", "omit"]) +async def test_beta_explicit_authorization_policy_is_preserved(mode: str, kind: str, header_profile: str) -> None: + defaults: dict[str, Any] = {} + extra: dict[str, Any] = {} + expected = ["Bearer synthetic-manual"] + if header_profile == "default": + defaults = {"authorization": "Bearer synthetic-manual"} + elif header_profile == "extra": + extra = {"authorization": "Bearer synthetic-manual"} + elif header_profile == "override": + defaults = {"Authorization": "Bearer synthetic-overridden"} + extra = {"AUTHORIZATION": "Bearer synthetic-manual"} + else: + defaults = {"Authorization": "Bearer synthetic-overridden"} + extra = {"authorization": openai.omit} + expected = [] + refused, callbacks, upgrades = await connect(mode, kind, defaults, extra) + assert refused is None and callbacks == [] + assert upgrades == [expected]