fix(cartesia): redact API keys from websocket handshake errors

livekit-agents/livekit/agents/utils/connection_pool.py

@@ -177,10 +177,19 @@ def prewarm(self) -> None:
             return
 
         async def _prewarm_impl() -> None:
-            async with self._connect_lock:
-                if not self._connections:
-                    conn = await self._connect(timeout=self._connect_timeout)
-                    self._available.add(conn)
+            try:
+                async with self._connect_lock:
+                    if not self._connections:
+                        conn = await self._connect(timeout=self._connect_timeout)
+                        self._available.add(conn)
+            except Exception as e:
+                # Swallow the error so asyncio does not log an unretrieved task
+                # exception. Log only the exception type: str(e) / repr(e) can
+                # embed request headers or URL credentials (?api_key=, &jwt_token=).
+                logger.warning(
+                    "failed to prewarm connection pool",
+                    extra={"exception_type": type(e).__name__},
+                )
 
         task = asyncio.create_task(_prewarm_impl())
         self._prewarm_task = weakref.ref(task)

livekit-plugins/livekit-plugins-cartesia/livekit/plugins/cartesia/_recognize_streams/auto_finalize_recognize_stream.py

@@ -24,6 +24,7 @@
 from livekit import rtc
 from livekit.agents import (
     APIConnectionError,
+    APIStatusError,
     LanguageCode,
     stt,
     utils,
@@ -405,8 +406,21 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
                 "Established new Cartesia STT WebSocket connection",
                 extra={"cartesia_request_id": c_request_id},
             )
-        except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e:
-            raise APIConnectionError("failed to connect to cartesia", retryable=True) from e
+        except asyncio.TimeoutError:
+            raise APIConnectionError("failed to connect to cartesia", retryable=True) from None
+        except aiohttp.ClientResponseError as e:
+            # Do not chain the aiohttp error: RequestInfo embeds auth headers and
+            # can leak API keys via Task/exception repr in logs (see #6739).
+            raise APIStatusError(
+                message=e.message, status_code=e.status, request_id=None, body=None
+            ) from None
+        except Exception as e:
+            # Do not chain the cause: some transport errors embed auth headers or
+            # URL credentials that would leak via __cause__ / traceback (see #6739).
+            raise APIConnectionError(
+                f"failed to connect to cartesia ({type(e).__name__})",
+                retryable=True,
+            ) from None
         return ws
 
     def _send_transcript_event(self, event_type: stt.SpeechEventType, transcript: str) -> None:

livekit-plugins/livekit-plugins-cartesia/livekit/plugins/cartesia/_recognize_streams/legacy_recognize_stream.py

@@ -24,6 +24,7 @@
 from livekit import rtc
 from livekit.agents import (
     APIConnectionError,
+    APIStatusError,
     LanguageCode,
     stt,
     utils,
@@ -329,8 +330,20 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
                 "Established new Cartesia STT WebSocket connection",
                 extra={"cartesia_request_id": self._request_id},
             )
-        except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e:
-            raise APIConnectionError("failed to connect to cartesia") from e
+        except asyncio.TimeoutError:
+            raise APIConnectionError("failed to connect to cartesia") from None
+        except aiohttp.ClientResponseError as e:
+            # Do not chain the aiohttp error: RequestInfo embeds auth headers and
+            # can leak API keys via Task/exception repr in logs (see #6739).
+            raise APIStatusError(
+                message=e.message, status_code=e.status, request_id=None, body=None
+            ) from None
+        except Exception as e:
+            # Do not chain the cause: some transport errors embed auth headers or
+            # URL credentials that would leak via __cause__ / traceback (see #6739).
+            raise APIConnectionError(
+                f"failed to connect to cartesia ({type(e).__name__})",
+            ) from None
         return ws
 
     def _process_stream_event(self, data: STTEventMessage) -> None:

livekit-plugins/livekit-plugins-cartesia/livekit/plugins/cartesia/tts.py

@@ -214,16 +214,30 @@ def provider(self) -> str:
     async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse:
         session = self._ensure_session()
         url = self._opts.get_ws_url(f"/tts/websocket?cartesia_version={self._opts.api_version}")
-        ws = await asyncio.wait_for(
-            session.ws_connect(
-                url,
-                headers={
-                    "User-Agent": USER_AGENT,
-                    API_AUTH_HEADER: self._opts.api_key,
-                },
-            ),
-            timeout,
-        )
+        try:
+            ws = await asyncio.wait_for(
+                session.ws_connect(
+                    url,
+                    headers={
+                        "User-Agent": USER_AGENT,
+                        API_AUTH_HEADER: self._opts.api_key,
+                    },
+                ),
+                timeout,
+            )
+        except asyncio.TimeoutError:
+            raise APITimeoutError() from None
+        except aiohttp.ClientResponseError as e:
+            # Do not chain the aiohttp error: RequestInfo embeds auth headers and
+            # can leak API keys via Task/exception repr in logs (see #6739).
+            raise APIStatusError(
+                message=e.message, status_code=e.status, request_id=None, body=None
+            ) from None
+        except Exception as e:
+            # Do not chain the cause: some transport errors embed auth headers or
+            # URL credentials that would leak via __cause__ / traceback (see #6739).
+            raise APIConnectionError(type(e).__name__) from None
+
         c_request_id = ws._response.headers.get(REQUEST_ID_HEADER)
         logger.debug(
             "Established new Cartesia TTS WebSocket connection",
@@ -547,6 +561,8 @@ async def _recv_task(ws: aiohttp.ClientWebSocketResponse, cartesia_context_id: s
             raise APIStatusError(
                 message=e.message, status_code=e.status, request_id=None, body=None
             ) from None
+        except APIError:
+            raise
         except Exception as e:
             logger.exception(
                 "Cartesia connection error. Include the cartesia_context_id to support@cartesia.ai for help debugging.",

tests/test_connection_pool.py

@@ -1,6 +1,9 @@
 import time
 
 import pytest
+from aiohttp import RequestInfo, WSServerHandshakeError
+from multidict import CIMultiDict, CIMultiDictProxy
+from yarl import URL
 
 from livekit.agents.utils import ConnectionPool
 
@@ -26,6 +29,15 @@ async def dummy_connect(timeout: float):
     return dummy_connect
 
 
+def _handshake_error_with_api_key(api_key: str) -> WSServerHandshakeError:
+    url = URL("wss://api.cartesia.ai/tts/websocket")
+    headers = CIMultiDict({"Host": "api.cartesia.ai", "X-API-Key": api_key})
+    request_info = RequestInfo(
+        url=url, method="GET", headers=CIMultiDictProxy(headers), real_url=url
+    )
+    return WSServerHandshakeError(request_info, (), status=401, message="Unauthorized")
+
+
 @pytest.mark.asyncio
 async def test_get_reuses_connection():
     dummy_connect = dummy_connect_factory()
@@ -83,3 +95,53 @@ async def test_get_expired():
 
     conn2 = await pool.get(timeout=10.0)
     assert conn2 is not conn, "Expected a new connection to be returned."
+
+
+@pytest.mark.asyncio
+async def test_prewarm_failure_does_not_leak_api_key_in_logs(caplog):
+    """Prewarm must swallow connect failures so asyncio never logs an unretrieved
+    task whose exception repr embeds auth headers (livekit/agents#6739)."""
+    secret = "cartesia-secret-api-key-do-not-log"
+
+    async def failing_connect(timeout: float):
+        raise _handshake_error_with_api_key(secret)
+
+    pool = ConnectionPool(connect_cb=failing_connect)
+    with caplog.at_level("WARNING"):
+        pool.prewarm()
+        task = pool._prewarm_task()
+        assert task is not None
+        await task
+
+    assert secret not in repr(task)
+    assert all(secret not in record.getMessage() for record in caplog.records)
+    warning_records = [
+        r for r in caplog.records if "failed to prewarm connection pool" in r.getMessage()
+    ]
+    assert warning_records
+    assert warning_records[0].exception_type == "WSServerHandshakeError"
+
+
+@pytest.mark.asyncio
+async def test_prewarm_failure_does_not_leak_url_credentials_in_logs(caplog):
+    """str(exception) can embed ?api_key= / &jwt_token=; logs must not include them."""
+    secret_key = "url-secret-api-key-do-not-log"
+    secret_jwt = "url-secret-jwt-token-do-not-log"
+
+    async def failing_connect(timeout: float):
+        raise ConnectionError(f"wss://example.com/ws?api_key={secret_key}&jwt_token={secret_jwt}")
+
+    pool = ConnectionPool(connect_cb=failing_connect)
+    with caplog.at_level("WARNING"):
+        pool.prewarm()
+        task = pool._prewarm_task()
+        assert task is not None
+        await task
+
+    assert all(secret_key not in record.getMessage() for record in caplog.records)
+    assert all(secret_jwt not in record.getMessage() for record in caplog.records)
+    warning_records = [
+        r for r in caplog.records if "failed to prewarm connection pool" in r.getMessage()
+    ]
+    assert warning_records
+    assert warning_records[0].exception_type == "ConnectionError"

tests/test_plugin_cartesia_tts.py

@@ -0,0 +1,78 @@
+"""Cartesia TTS websocket connect must not leak API keys via exception repr."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from aiohttp import RequestInfo, WSServerHandshakeError
+from multidict import CIMultiDict, CIMultiDictProxy
+from yarl import URL
+
+from livekit.agents import APIConnectionError, APIStatusError
+
+pytestmark = pytest.mark.plugin("cartesia")
+
+SECRET_API_KEY = "cartesia-secret-api-key-do-not-log"
+
+
+def _handshake_error_with_api_key(api_key: str) -> WSServerHandshakeError:
+    url = URL("wss://api.cartesia.ai/tts/websocket")
+    headers = CIMultiDict(
+        {
+            "Host": "api.cartesia.ai",
+            "User-Agent": "LiveKit Agents Cartesia Plugin/test",
+            "X-API-Key": api_key,
+        }
+    )
+    request_info = RequestInfo(
+        url=url, method="GET", headers=CIMultiDictProxy(headers), real_url=url
+    )
+    return WSServerHandshakeError(request_info, (), status=401, message="Unauthorized")
+
+
+@pytest.mark.asyncio
+async def test_connect_ws_redacts_api_key_from_handshake_error():
+    from livekit.plugins.cartesia import TTS
+
+    tts = TTS(api_key=SECRET_API_KEY)
+
+    async def _raise(*_args, **_kwargs):
+        raise _handshake_error_with_api_key(SECRET_API_KEY)
+
+    session = MagicMock()
+    session.ws_connect = MagicMock(side_effect=_raise)
+    tts._session = session
+
+    with pytest.raises(APIStatusError) as exc_info:
+        await tts._connect_ws(timeout=5.0)
+
+    err = exc_info.value
+    assert err.status_code == 401
+    assert SECRET_API_KEY not in repr(err)
+    assert err.__cause__ is None
+
+
+@pytest.mark.asyncio
+async def test_connect_ws_generic_error_does_not_chain_cause():
+    """Generic connect failures must not chain __cause__ (may embed auth headers)."""
+    from livekit.plugins.cartesia import TTS
+
+    tts = TTS(api_key=SECRET_API_KEY)
+    leaky = ConnectionError(f"wss://api.cartesia.ai/tts/websocket?api_key={SECRET_API_KEY}")
+
+    async def _raise(*_args, **_kwargs):
+        raise leaky
+
+    session = MagicMock()
+    session.ws_connect = MagicMock(side_effect=_raise)
+    tts._session = session
+
+    with pytest.raises(APIConnectionError) as exc_info:
+        await tts._connect_ws(timeout=5.0)
+
+    err = exc_info.value
+    assert err.__cause__ is None
+    assert str(err) == "ConnectionError"
+    assert SECRET_API_KEY not in repr(err)
+    assert SECRET_API_KEY not in str(err)