From fc379199523cbbea8fcd1cef7f10182a17591aec Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 2 Jul 2026 15:10:01 -0400 Subject: [PATCH] fix(#5430): harden ElevenLabs TTS transport guards --- api/routes.py | 11 ++- tests/test_issue3510_elevenlabs_tts.py | 123 +++++++++++++++++++++++-- 2 files changed, 122 insertions(+), 12 deletions(-) diff --git a/api/routes.py b/api/routes.py index 87d202b8..4037fafa 100644 --- a/api/routes.py +++ b/api/routes.py @@ -16670,21 +16670,26 @@ def _handle_tts(handler, parsed): "voice_settings": {"stability": 0.5, "similarity_boost": 0.75}, }).encode("utf-8") - from urllib.request import Request, urlopen as _urlopen - + from urllib.request import Request, build_opener req = Request(url, data=req_body, headers={ "xi-api-key": api_key, "Content-Type": "application/json", "Accept": "audio/mpeg", }) + class _NoRedirectElevenlabsHandler(HTTPRedirectHandler): + """Refuse to follow redirects on the ElevenLabs call.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + raise ValueError("ElevenLabs TTS upstream attempted a redirect") + # Buffer the full response before sending first byte. # The streaming endpoint is designed for chunked delivery, but urllib's # chunked-read path adds per-chunk overhead that dominates short TTS # payloads. A hard cap keeps the buffered path bounded even if the # upstream misbehaves. try: - with _urlopen(req, timeout=30) as resp: + with _tts_open(req, timeout=30, opener_factory=lambda: build_opener(ProxyHandler({}), _NoRedirectElevenlabsHandler())) as resp: audio_data = _buffer_tts_audio_response(resp) except ValueError: logger.warning("ElevenLabs TTS rejected an invalid upstream response", exc_info=True) diff --git a/tests/test_issue3510_elevenlabs_tts.py b/tests/test_issue3510_elevenlabs_tts.py index 6ee67f2d..82746a2d 100644 --- a/tests/test_issue3510_elevenlabs_tts.py +++ b/tests/test_issue3510_elevenlabs_tts.py @@ -1,7 +1,7 @@ """Coverage for the ElevenLabs TTS engine on the /api/ttt endpoint (#3510). Exercises the engine routing + guard rails of _handle_tts's elevenlabs branch -in-process via a fake handler. The happy path mocks urllib.urlopen so no real +in-process via a fake handler. The happy path mocks routes._tts_open so no real ElevenLabs network call is made; the rejection paths (missing key, bad config voice_id) bail before any network call. """ @@ -105,16 +105,14 @@ def test_elevenlabs_happy_path_streams_mp3(monkeypatch): def __exit__(self, *a): return False - def _fake_urlopen(req, timeout=30): + def _fake_tts_open(req, timeout=30, opener_factory=None, **_kw): captured["url"] = req.full_url captured["xi_api_key"] = req.get_header("Xi-api-key") captured["body"] = json.loads(req.data.decode("utf-8")) return _Resp() - # The branch does `from urllib.request import ... urlopen as _urlopen` at - # call time, so patching the source module is what takes effect. - import urllib.request as _ur - monkeypatch.setattr(_ur, "urlopen", _fake_urlopen) + # The ElevenLabs branch now routes through `_tts_open`. + monkeypatch.setattr(routes, "_tts_open", _fake_tts_open) h = _post({"text": "hello world", "engine": "elevenlabs"}, client="9.9.9.3") routes._handle_tts(h, None) @@ -131,6 +129,11 @@ def test_elevenlabs_happy_path_streams_mp3(monkeypatch): def test_elevenlabs_overlong_text_rejected_before_engine(monkeypatch): """The shared 5000-char cap applies to the elevenlabs engine too (no call).""" monkeypatch.setenv("ELEVENLABS_API_KEY", "sk-test") + + def _fail_if_called(*_args, **_kwargs): + raise AssertionError("ElevenLabs upstream called") + + monkeypatch.setattr(routes, "_tts_open", _fail_if_called) h = _post({"text": "x" * 5001, "engine": "elevenlabs"}, client="9.9.9.4") routes._handle_tts(h, None) assert h.status == 400 @@ -158,15 +161,117 @@ def test_elevenlabs_rejects_oversized_upstream_audio(monkeypatch): def __exit__(self, *a): return False - def _fake_urlopen(req, timeout=30): + def _fake_tts_open(req, timeout=30, opener_factory=None, **_kw): return _Resp() - import urllib.request as _ur monkeypatch.setattr(routes, "_TTS_PROXY_MAX_BYTES", 4) - monkeypatch.setattr(_ur, "urlopen", _fake_urlopen) + monkeypatch.setattr(routes, "_tts_open", _fake_tts_open) h = _post({"text": "hello world", "engine": "elevenlabs"}, client="9.9.9.5") routes._handle_tts(h, None) assert h.status == 502 assert "ElevenLabs TTS generation failed" in (h.payload() or {}).get("error", "") + + +def test_elevenlabs_tts_does_not_follow_upstream_redirect(monkeypatch): + # Regression proof: a fake redirect response should fail at _tts_open, not be followed. + import urllib.request as _ur + from urllib.request import HTTPRedirectHandler + + class _Resp: + def __init__(self): + self._chunks = [b"ID3fakeaudio", b""] + self._i = 0 + def info(self): + return {} + def read(self, n=-1): + c = self._chunks[self._i] if self._i < len(self._chunks) else b"" + self._i += 1 + return c + def __enter__(self): + return self + def __exit__(self, *a): + return False + + def _fake_urlopen(req, timeout=30): + return _Resp() + + def _fake_tts_open(req, timeout=30, opener_factory=None, **_kw): + assert opener_factory is not None + opener = opener_factory() + redirect_handler = next( + (handler for handler in opener.handlers if isinstance(handler, HTTPRedirectHandler)), + None, + ) + assert redirect_handler is not None + with pytest.raises(ValueError): + redirect_handler.redirect_request( + req, + None, + 302, + "Found", + {"Location": "http://169.254.169.254/latest/meta-data"}, + "http://169.254.169.254/latest/meta-data", + ) + raise ValueError("redirect blocked 302") + + # Keep the legacy path patched as fake audio to keep base and head checks + # behaviorally distinct. + monkeypatch.setenv("ELEVENLABS_API_KEY", "sk-test") + import api.config as _cfg + monkeypatch.setattr(_cfg, "get_config", lambda: {"tts": {"elevenlabs": {"voice_id": "pNInz6obpgDQGcFmaJgB", "model": "eleven_multilingual_v2"}}}) + monkeypatch.setattr(_ur, "urlopen", _fake_urlopen) + monkeypatch.setattr(routes, "_tts_open", _fake_tts_open) + h = _post({"text": "hello world", "engine": "elevenlabs"}, client="9.9.9.6") + routes._handle_tts(h, None) + + assert h.status == 502 + assert "ElevenLabs TTS generation failed" in (h.payload() or {}).get("error", "") + + +def test_elevenlabs_tts_uses_no_proxy_opener(monkeypatch): + captured = {} + + class _Resp: + def __init__(self): + self.headers = {"Content-Type": "audio/mpeg"} + self._chunks = [b"ID3fakeaudio", b""] + self._i = 0 + def read(self, n=-1): + c = self._chunks[self._i] if self._i < len(self._chunks) else b"" + self._i += 1 + return c + def __enter__(self): + return self + def __exit__(self, *a): + return False + + def _fake_tts_open(req, timeout=30, opener_factory=None, **_kw): + captured["opener_factory"] = opener_factory + opener = opener_factory() + captured["opener_handlers"] = opener.handlers + return _Resp() + + from urllib.request import HTTPRedirectHandler + + original_proxy_handler = routes.ProxyHandler + + def _spy_proxy_handler(proxies=None, **_kw): + captured["proxy_init_kwargs"] = {"proxies": proxies} + return original_proxy_handler(proxies) + + monkeypatch.setattr(routes, "ProxyHandler", _spy_proxy_handler) + + monkeypatch.setenv("ELEVENLABS_API_KEY", "sk-test") + import api.config as _cfg + monkeypatch.setattr(_cfg, "get_config", lambda: {"tts": {"elevenlabs": {"voice_id": "pNInz6obpgDQGcFmaJgB", "model": "eleven_multilingual_v2"}}}) + monkeypatch.setattr(routes, "_tts_open", _fake_tts_open) + + h = _post({"text": "hello world", "engine": "elevenlabs"}, client="9.9.9.7") + routes._handle_tts(h, None) + + assert h.status == 200 + assert h.wfile.getvalue() == b"ID3fakeaudio" + assert captured["proxy_init_kwargs"] == {"proxies": {}} + assert any(isinstance(handler, HTTPRedirectHandler) for handler in captured["opener_handlers"])