Merge #5079 (OpenAI TTS + SSRF hardening) into stage

api/routes.py

@@ -15066,8 +15066,139 @@ def _normalize_tts_prosody(value, *, unit: str) -> str | None:
     return None
 
 
+_TTS_PROXY_MAX_BYTES = 16 * 1024 * 1024
+_TTS_LOCALHOST_HOSTS = {"127.0.0.1", "::1", "localhost"}
+
+
+def _tts_host_is_blocked_target(hostname: str) -> bool:
+    """True if the hostname resolves to (or literally is) a private / loopback /
+    link-local / reserved / multicast address — the SSRF-risk targets that an
+    OpenAI-compatible TTS base_url must not be allowed to reach. Public hosts
+    (a user's own hosted OpenAI-compatible server) are allowed; the explicit
+    localhost-over-http dev case is handled separately by the caller."""
+    import ipaddress
+    import socket
+
+    host = (hostname or "").strip().lower()
+    if not host:
+        return True
+
+    def _addr_blocked(ip_str: str) -> bool:
+        try:
+            ip = ipaddress.ip_address(ip_str)
+        except ValueError:
+            return False
+        return (
+            ip.is_private
+            or ip.is_loopback
+            or ip.is_link_local
+            or ip.is_reserved
+            or ip.is_multicast
+            or ip.is_unspecified
+        )
+
+    # Literal IP host?
+    try:
+        ipaddress.ip_address(host)
+        return _addr_blocked(host)
+    except ValueError:
+        pass
+
+    # DNS host: resolve and block if ANY resolved address is a blocked target
+    # (defends against a hostname pointing at an internal/link-local address).
+    # A DNS-resolution failure is NOT treated as an SSRF block — an unresolvable
+    # host simply can't be reached (the outbound request fails naturally), and
+    # failing closed here would wrongly reject legitimate public hosts that don't
+    # resolve in a sandboxed/offline environment. Only a host that resolves to a
+    # blocked address is rejected.
+    try:
+        infos = socket.getaddrinfo(host, None)
+    except Exception:
+        return False
+    for info in infos:
+        sockaddr = info[4]
+        if sockaddr and _addr_blocked(str(sockaddr[0])):
+            return True
+    return False
+
+
+def _normalized_openai_tts_base_url(base_url: str) -> str:
+    from urllib.parse import urlsplit, urlunsplit
+
+    raw = str(base_url or "").strip()
+    parsed = urlsplit(raw)
+    hostname = (parsed.hostname or "").strip().lower()
+    if parsed.username or parsed.password:
+        raise ValueError("invalid OpenAI base_url in config")
+    if not parsed.scheme or not parsed.netloc or parsed.query or parsed.fragment:
+        raise ValueError("invalid OpenAI base_url in config")
+    if parsed.scheme == "https":
+        # Public https hosts are allowed (a user's own OpenAI-compatible server),
+        # but reject private/loopback/link-local/reserved targets to close the
+        # SSRF surface (e.g. https://169.254.169.254, https://10.x internal).
+        if _tts_host_is_blocked_target(hostname):
+            raise ValueError("invalid OpenAI base_url in config")
+    elif parsed.scheme == "http" and hostname in _TTS_LOCALHOST_HOSTS:
+        # Explicit localhost-over-http dev/self-hosted case only.
+        pass
+    else:
+        raise ValueError("invalid OpenAI base_url in config")
+    path = parsed.path.rstrip("/") or ""
+    return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
+
+
+def _buffer_tts_audio_response(resp, *, max_bytes: int | None = None) -> bytes:
+    if max_bytes is None:
+        max_bytes = _TTS_PROXY_MAX_BYTES
+    headers = getattr(resp, "headers", None)
+    content_type = ""
+    if headers is not None:
+        try:
+            content_type = str(headers.get("Content-Type") or "")
+        except Exception:
+            content_type = ""
+    if not content_type:
+        try:
+            info = resp.info()
+            content_type = str(info.get("Content-Type") or "")
+        except Exception:
+            content_type = ""
+    # A present Content-Type that isn't audio/* is rejected. A MISSING
+    # Content-Type is tolerated: some OpenAI-compatible servers stream audio
+    # bytes without setting Content-Type, and the success path defaults the
+    # browser-facing type to audio/mpeg (encoded in the tests). The SSRF
+    # base-url guard is the primary defense against reaching a non-audio
+    # internal endpoint.
+    if content_type and not content_type.lower().startswith("audio/"):
+        raise ValueError("upstream returned non-audio content")
+    audio_data = bytearray()
+    while True:
+        chunk = resp.read(65536)
+        if not chunk:
+            break
+        audio_data.extend(chunk)
+        if len(audio_data) > max_bytes:
+            raise ValueError("upstream audio exceeded byte limit")
+    return bytes(audio_data)
+
+
+def _tts_open(req, *, timeout=30, opener_factory=None):
+    """Thin network seam for the TTS upstream fetch so tests can intercept it.
+
+    Defaults to a no-redirect opener (built by opener_factory) so an upstream
+    redirect can't carry the Authorization bearer to — or SSRF-bounce the
+    request into — a different/private target after base-url validation passed.
+    Tests monkeypatch this function (or urllib.request.urlopen) to inject a
+    stub response."""
+    if opener_factory is not None:
+        opener = opener_factory()
+        return opener.open(req, timeout=timeout)
+    from urllib.request import urlopen as _urlopen
+    return _urlopen(req, timeout=timeout)
+
+
 def _handle_tts(handler, parsed):
-    """Generate TTS audio via Edge TTS. POST JSON body only.
+    """Generate TTS audio via supported server TTS engines. POST JSON body only.
 
     Design note addressing deep review blocker #4 (synchronous I/O):
     The server uses ThreadingHTTPServer (see server.py:173), so each request
@@ -15087,7 +15218,7 @@ def _handle_tts(handler, parsed):
     voice = "zh-CN-XiaoxiaoNeural"
     rate_str = ""
     pitch_str = ""
-    engine = "edge"  # "edge" | "elevenlabs" | "browser" (browser is client-side only)
+    engine = "edge"  # "edge" | "elevenlabs" | "openai" | "browser" (browser is client-side only)
 
     if handler.command != "POST":
         from api.helpers import bad as _bad
@@ -15228,17 +15359,15 @@ def _handle_tts(handler, parsed):
         # 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. With the 5000-char cap enforced above the buffer is bounded
-        # (a few MB of mp3 worst case), so full-buffer-then-send is faster in
-        # practice and simpler to reason about.
-        audio_data = b""
+        # payloads. A hard cap keeps the buffered path bounded even if the
+        # upstream misbehaves.
         try:
             with _urlopen(req, timeout=30) as resp:
-                while True:
-                    chunk = resp.read(65536)
-                    if not chunk:
-                        break
-                    audio_data += chunk
+                audio_data = _buffer_tts_audio_response(resp)
+        except ValueError:
+            logger.warning("ElevenLabs TTS rejected an invalid upstream response", exc_info=True)
+            from api.helpers import bad as _bad
+            return _bad(handler, "ElevenLabs TTS generation failed", 502)
         except Exception:
             logger.exception("ElevenLabs TTS generation failed")
             from api.helpers import bad as _bad
@@ -15255,6 +15384,98 @@ def _handle_tts(handler, parsed):
             pass
         return True
 
+    # ── OpenAI-compatible TTS ──────────────────────────────────────────
+    if engine == "openai":
+        api_key = os.getenv("VOICE_TOOLS_OPENAI_KEY", "").strip()
+        if not api_key:
+            api_key = os.getenv("OPENAI_API_KEY", "").strip()
+        if not api_key:
+            try:
+                from api.onboarding import _load_env_file
+                from api.profiles import get_active_hermes_home
+                env_cfg = _load_env_file(get_active_hermes_home() / ".env")
+                api_key = env_cfg.get("VOICE_TOOLS_OPENAI_KEY", "") or env_cfg.get("OPENAI_API_KEY", "")
+            except Exception:
+                pass
+        if not api_key:
+            from api.helpers import bad as _bad
+            return _bad(handler, "OpenAI API key not configured", 503)
+
+        from urllib.parse import urlunsplit as _urlunsplit
+
+        base_url = _urlunsplit(("https", "api.openai.com", "/v1", "", ""))
+        model = "gpt-4o-mini-tts"
+        oai_voice = "alloy"
+        try:
+            from api.config import get_config
+            tts_cfg = (get_config() or {}).get("tts", {})
+            if isinstance(tts_cfg, dict):
+                oai_cfg = tts_cfg.get("openai", {})
+                if isinstance(oai_cfg, dict):
+                    base_url = _normalized_openai_tts_base_url(oai_cfg.get("base_url") or base_url)
+                    model = oai_cfg.get("model") or model
+                    oai_voice = oai_cfg.get("voice") or oai_voice
+                else:
+                    base_url = _normalized_openai_tts_base_url(base_url)
+            else:
+                base_url = _normalized_openai_tts_base_url(base_url)
+        except ValueError:
+            from api.helpers import bad as _bad
+            return _bad(handler, "invalid OpenAI base_url in config", 400)
+        except Exception:
+            pass
+
+        url = f"{base_url}/audio/speech"
+        req_body = json.dumps({
+            "model": model,
+            "input": text,
+            "voice": oai_voice,
+        }).encode("utf-8")
+
+        from urllib.request import Request, build_opener, HTTPRedirectHandler, urlopen as _urlopen
+
+        class _NoRedirectTtsHandler(HTTPRedirectHandler):
+            """Refuse to follow redirects on the TTS call. A redirect is never a
+            legitimate response to a POST /audio/speech, and following one would
+            (a) carry the Authorization bearer to the redirect target and
+            (b) let a public host bounce the request to a private/link-local
+            SSRF target after the base-url validation already passed."""
+
+            def redirect_request(self, req, fp, code, msg, headers, newurl):
+                raise ValueError("OpenAI TTS upstream attempted a redirect")
+
+        req = Request(url, data=req_body, headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+            "Accept": "audio/mpeg",
+        })
+
+        # Use a no-redirect opener so an upstream redirect can't carry the bearer
+        # to (or SSRF-bounce into) a different/private target. _tts_open is a thin
+        # module seam so tests can still intercept the network call.
+        try:
+            with _tts_open(req, timeout=30, opener_factory=lambda: build_opener(_NoRedirectTtsHandler())) as resp:
+                audio_data = _buffer_tts_audio_response(resp)
+        except ValueError:
+            logger.warning("OpenAI TTS rejected an invalid upstream response", exc_info=True)
+            from api.helpers import bad as _bad
+            return _bad(handler, "OpenAI TTS generation failed", 502)
+        except Exception:
+            logger.exception("OpenAI TTS generation failed")
+            from api.helpers import bad as _bad
+            return _bad(handler, "OpenAI TTS generation failed", 500)
+
+        handler.send_response(200)
+        handler.send_header("Content-Type", "audio/mpeg")
+        handler.send_header("Cache-Control", "no-store")
+        handler.send_header("Content-Length", str(len(audio_data)))
+        handler.end_headers()
+        try:
+            handler.wfile.write(audio_data)
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+        return True
+
     # ── Edge TTS ────────────────────────────────────────────────────────
     allowed = {
         "zh-CN-XiaoxiaoNeural", "zh-CN-XiaoyiNeural", "zh-CN-YunxiNeural",

static/boot.js

@@ -926,12 +926,12 @@ window._micPendingSend=window._micPendingSend||false;
 // handles selection, the dropdown option, and playback. Mirrors registerHermesSkin.
 //
 //   window.registerHermesTtsEngine({
-//     id: 'voicevox',            // [a-z0-9_-], not a built-in (browser/edge/elevenlabs)
+//     id: 'voicevox',            // [a-z0-9_-], not a built-in (browser/edge/elevenlabs/openai)
 //     label: 'VOICEVOX (local)',
 //     synthesize(text, opts) { return Promise<ArrayBuffer|Blob>; }
 //   }) -> true on success, false if rejected
 var _HERMES_TTS_ENGINES = Object.create(null);
-var _HERMES_TTS_RESERVED = { browser:1, edge:1, elevenlabs:1 };
+var _HERMES_TTS_RESERVED = { browser:1, edge:1, elevenlabs:1, openai:1 };
 function _hermesTtsValidId(id){ return typeof id==='string' && /^[a-z0-9][a-z0-9_-]{0,31}$/.test(id); }
 function _hermesAddTtsOption(id, label){
   var sel=document.getElementById('settingsTtsEngine');
@@ -1296,6 +1296,46 @@ window._hermesTtsSynth=function(id, text, opts){
       });
       return;
     }
+    if(engine==="openai"){
+      _ttsSpeaking=true;
+      fetch(new URL('api/tts', document.baseURI || location.href).href, {
+        method: 'POST',
+        headers: {'Content-Type': 'application/json'},
+        body: JSON.stringify({text: clean, engine: 'openai'})
+      })
+      .then(r => {
+        if(!r.ok) throw new Error('TTS request failed: ' + r.status);
+        return r.blob();
+      })
+      .then(blob => {
+        const url = URL.createObjectURL(blob);
+        const audio = new Audio(url);
+        _playingEdgeAudio=audio;
+        audio.onended = () => {
+          _ttsSpeaking=false;
+          if(_playingEdgeAudio===audio) _playingEdgeAudio=null;
+          URL.revokeObjectURL(url);
+          if(_voiceModeActive) setTimeout(()=>_startListening(),500);
+        };
+        audio.onerror = () => {
+          _ttsSpeaking=false;
+          if(_playingEdgeAudio===audio) _playingEdgeAudio=null;
+          URL.revokeObjectURL(url);
+          if(_voiceModeActive) setTimeout(()=>_startListening(),1000);
+        };
+        audio.play().catch(() => {
+          _ttsSpeaking=false;
+          if(_playingEdgeAudio===audio) _playingEdgeAudio=null;
+          URL.revokeObjectURL(url);
+          if(_voiceModeActive) setTimeout(()=>_startListening(),1000);
+        });
+      })
+      .catch(() => {
+        _ttsSpeaking=false;
+        if(_voiceModeActive) setTimeout(()=>_startListening(),1000);
+      });
+      return;
+    }
     if(engine==="edge"){
       const voice=localStorage.getItem("hermes-tts-voice")||"zh-CN-XiaoxiaoNeural";
       const savedRate=parseFloat(localStorage.getItem("hermes-tts-rate"));

static/index.html

@@ -1195,7 +1195,7 @@
               <div style="font-size:11px;color:var(--muted);margin-top:4px" data-i18n="settings_desc_raw_audio">Record and send the original audio file to the agent instead of converting it to text first. The agent can then transcribe it or process the raw audio (emotion, background noise, custom STT). Like Telegram's voice message behavior.</div>
             </div>
             <div class="settings-field">
-              <label for="settingsTtsEngine" data-i18n="settings_label_tts_engine">TTS Engine</label><select id="settingsTtsEngine" style="width:100%;padding:8px;background:var(--code-bg);color:var(--text);border:1px solid var(--border2);border-radius:6px"><option value="browser">Browser speech synthesis</option><option value="edge">Edge TTS (server)</option><option value="elevenlabs">ElevenLabs (server)</option></select><div style="font-size:11px;color:var(--muted);margin-top:4px" data-i18n="settings_desc_tts_engine">Choose speech engine. Edge TTS uses Microsoft neural voices via the server.</div></div><div class="settings-field"><label for="settingsTtsVoice" data-i18n="settings_label_tts_voice">Voice</label>
+              <label for="settingsTtsEngine" data-i18n="settings_label_tts_engine">TTS Engine</label><select id="settingsTtsEngine" style="width:100%;padding:8px;background:var(--code-bg);color:var(--text);border:1px solid var(--border2);border-radius:6px"><option value="browser">Browser speech synthesis</option><option value="edge">Edge TTS (server)</option><option value="elevenlabs">ElevenLabs (server)</option><option value="openai">OpenAI TTS (server)</option></select><div style="font-size:11px;color:var(--muted);margin-top:4px" data-i18n="settings_desc_tts_engine">Choose speech engine. Edge TTS uses Microsoft neural voices via the server.</div></div><div class="settings-field"><label for="settingsTtsVoice" data-i18n="settings_label_tts_voice">Voice</label>
               <select id="settingsTtsVoice" style="width:100%;padding:8px;background:var(--code-bg);color:var(--text);border:1px solid var(--border2);border-radius:6px">
                 <option value="">Default system voice</option>
               </select>

static/panels.js

@@ -8293,6 +8293,8 @@ async function loadSettingsPanel(){
       const current=localStorage.getItem('hermes-tts-voice')||'';
       if(engine==='elevenlabs'){
         ttsVoiceSel.innerHTML='<option value="">Hermy — ElevenLabs (server-configured)</option>';
+      } else if(engine==='openai'){
+        ttsVoiceSel.innerHTML='<option value="">OpenAI voice (server-configured)</option>';
       } else if(engine==='edge'){
         const edgeVoices=[
           {value:'zh-CN-XiaoxiaoNeural',label:'Xiaoxiao (Chinese, Female)'},

static/ui.js

@@ -7193,6 +7193,10 @@ function speakMessage(btn){
   if(!clean) return;
 
   const engine=localStorage.getItem('hermes-tts-engine')||'browser';
+  if(engine==='openai'){
+    _playOpenaiTts(clean, btn);
+    return;
+  }
   if(engine==='elevenlabs'){
     _playElevenLabsTts(clean, btn);
     return;
@@ -7265,6 +7269,33 @@ function _playElevenLabsTts(text, btn){
   .catch(function(e){ _fail((e&&e.message)||'ElevenLabs TTS failed'); });
 }
 
+function _playOpenaiTts(text, btn){
+  if(btn) btn.dataset.speaking='1';
+  _ttsSpeaking=true;
+  const _fail=function(msg){
+    _ttsSpeaking=false;_playingEdgeAudio=null;
+    if(btn)btn.dataset.speaking='0';
+    if(msg&&typeof showToast==='function') showToast(msg,4000,'error');
+  };
+  fetch(new URL('api/tts', document.baseURI || location.href).href, {
+    method:'POST',
+    headers:{'Content-Type':'application/json'},
+    body:JSON.stringify({text:text, engine:'openai'})
+  })
+  .then(function(r){
+    if(!r.ok){
+      return r.json().catch(function(){return {};}).then(function(j){
+        throw new Error((j&&j.error)||('TTS request failed: '+r.status));
+      });
+    }
+    return r.arrayBuffer();
+  })
+  .then(function(buf){
+    return _playAudioBuf(buf, btn, 'OpenAI TTS');
+  })
+  .catch(function(e){ _fail((e&&e.message)||'OpenAI TTS failed'); });
+}
+
 // ── Shared AudioContext for TTS playback (no blob URLs needed) ──
 let _ttsAudioCtx=null;
 function _getTtsAudioCtx(){
@@ -7344,6 +7375,10 @@ function autoReadLastAssistant(){
   if(!text.trim()) return;
   const clean=_stripForTTS(text);
   if(!clean) return;
+  if(engine==='openai'){
+    _playOpenaiTts(clean, null);
+    return;
+  }
   if(engine==='elevenlabs'){
     _playElevenLabsTts(clean, null);
     return;

tests/test_499_tts_playback.py

@@ -57,6 +57,18 @@ class TestTtsUtilityFunctions:
         assert 'speechSynthesis.speak' in src, \
             "speakMessage must call speechSynthesis.speak"
 
+    def test_speak_message_routes_openai(self):
+        src = _read('ui.js')
+        assert "if(engine==='openai')" in src, \
+            "speakMessage must branch for the OpenAI TTS engine"
+        assert '_playOpenaiTts(clean, btn);' in src, \
+            "speakMessage must route OpenAI TTS through _playOpenaiTts"
+
+    def test_auto_read_routes_openai(self):
+        src = _read('ui.js')
+        assert '_playOpenaiTts(clean, null);' in src, \
+            "autoReadLastAssistant must route OpenAI TTS through _playOpenaiTts"
+
 
 class TestTtsSpeakerButton:
     """Speaker button is rendered on assistant messages."""
@@ -109,6 +121,11 @@ class TestTtsSettings:
         assert 'settingsTtsVoice' in src, \
             "TTS voice selector not found in index.html"
 
+    def test_tts_engine_includes_openai_option(self):
+        src = _read('index.html')
+        assert '<option value="openai">OpenAI TTS (server)</option>' in src, \
+            "settingsTtsEngine must expose the OpenAI server TTS option"
+
     def test_tts_rate_slider(self):
         src = _read('index.html')
         assert 'settingsTtsRate' in src, \
@@ -133,6 +150,13 @@ class TestTtsSettings:
         assert 'function _applyTtsEnabled(' in src, \
             "_applyTtsEnabled function not found in panels.js"
 
+    def test_openai_voice_placeholder_in_panels(self):
+        src = _read('panels.js')
+        assert "engine==='openai'" in src, \
+            "panels.js must branch for the OpenAI TTS engine"
+        assert 'OpenAI voice (server-configured)' in src, \
+            "OpenAI TTS must present a server-configured voice placeholder"
+
 
 class TestTtsI18n:
     """TTS i18n keys exist in the English locale."""

tests/test_extension_tts_engine_registration.py

@@ -31,8 +31,11 @@ def test_public_api_present():
 
 def test_reserved_builtins_guarded():
     assert "_HERMES_TTS_RESERVED" in BOOT_JS
-    # browser/edge/elevenlabs must be reserved so an extension can't shadow them
-    assert "browser:1" in BOOT_JS and "edge:1" in BOOT_JS and "elevenlabs:1" in BOOT_JS
+    # browser/edge/elevenlabs/openai must be reserved so an extension can't shadow them
+    assert "browser:1" in BOOT_JS
+    assert "edge:1" in BOOT_JS
+    assert "elevenlabs:1" in BOOT_JS
+    assert "openai:1" in BOOT_JS
 
 
 def test_both_playback_paths_check_registry():
@@ -75,6 +78,8 @@ def test_registration_behavior():
         // reserved key rejected
         results.reservedRejected = (window.registerHermesTtsEngine({
           id: 'edge', label: 'x', synthesize: () => new ArrayBuffer(1) }) === false);
+        results.openaiReservedRejected = (window.registerHermesTtsEngine({
+          id: 'openai', label: 'x', synthesize: () => new ArrayBuffer(1) }) === false);
         // bad id rejected
         results.badIdRejected = (window.registerHermesTtsEngine({
           id: 'Bad Id!', label: 'x', synthesize: () => new ArrayBuffer(1) }) === false);
@@ -101,6 +106,7 @@ def test_registration_behavior():
     assert r["validOk"] is True
     assert r["isRegistered"] is True
     assert r["reservedRejected"] is True
+    assert r["openaiReservedRejected"] is True
     assert r["badIdRejected"] is True
     assert r["noSynthRejected"] is True
     assert r["optionListed"] is True

tests/test_issue3510_elevenlabs_tts.py

@@ -135,3 +135,38 @@ def test_elevenlabs_overlong_text_rejected_before_engine(monkeypatch):
     routes._handle_tts(h, None)
     assert h.status == 400
     assert "too long" in (h.payload() or {}).get("error", "")
+
+
+def test_elevenlabs_rejects_oversized_upstream_audio(monkeypatch):
+    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"}}
+    })
+
+    class _Resp:
+        def __init__(self):
+            self.headers = {"Content-Type": "audio/mpeg"}
+            self._chunks = [b"1234", b"5", 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_urlopen(req, timeout=30):
+        return _Resp()
+
+    import urllib.request as _ur
+    monkeypatch.setattr(routes, "_TTS_PROXY_MAX_BYTES", 4)
+    monkeypatch.setattr(_ur, "urlopen", _fake_urlopen)
+
+    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", "")

tests/test_issue4982_openai_tts.py

@@ -0,0 +1,255 @@
+"""OpenAI-compatible TTS endpoint and UI wiring coverage for #4982."""
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+import api.routes as routes
+
+
+STATIC_DIR = Path(__file__).resolve().parent.parent / "static"
+
+
+class _FakeHandler:
+    def __init__(self, body: bytes, command: str = "POST", headers=None, client="1.2.3.4"):
+        self.command = command
+        self.rfile = io.BytesIO(body)
+        self.wfile = io.BytesIO()
+        self.headers = headers or {}
+        self.headers.setdefault("Content-Length", str(len(body)))
+        self.client_address = (client, 12345)
+        self.status = None
+        self.sent_headers = {}
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, key, value):
+        self.sent_headers[key] = value
+
+    def end_headers(self):
+        pass
+
+    def payload(self):
+        try:
+            return json.loads(self.wfile.getvalue().decode("utf-8"))
+        except Exception:
+            return None
+
+
+class _StreamOnceResponse:
+    def __init__(self, chunks, headers=None):
+        self._chunks = list(chunks)
+        self.headers = headers or {}
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self, _size=-1):
+        if self._chunks:
+            return self._chunks.pop(0)
+        return b""
+
+
+def _post(body_dict, **kw):
+    body = json.dumps(body_dict).encode()
+    return _FakeHandler(body, **kw)
+
+
+def _reset_limiter():
+    if hasattr(routes._handle_tts, "_tts_limiter"):
+        del routes._handle_tts._tts_limiter
+
+
+@pytest.fixture(autouse=True)
+def _fresh_tts_limiter(monkeypatch):
+    import api.auth as _auth
+
+    monkeypatch.setattr(_auth, "is_auth_enabled", lambda: False)
+    monkeypatch.setattr(routes, "is_auth_enabled", lambda: False, raising=False)
+    monkeypatch.delenv("HERMES_WEBUI_TRUST_FORWARDED_FOR", raising=False)
+    monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
+    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+    _reset_limiter()
+    yield
+    _reset_limiter()
+
+
+def test_openai_tts_no_key_returns_503(monkeypatch):
+    import api.onboarding as onboarding
+
+    monkeypatch.setattr(onboarding, "_load_env_file", lambda *_args, **_kwargs: {})
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.1")
+    routes._handle_tts(h, None)
+    assert h.status == 503
+    assert "OpenAI API key not configured" in (h.payload() or {}).get("error", "")
+
+
+def test_openai_tts_success_returns_audio(monkeypatch):
+    captured = {}
+
+    def _fake_urlopen(req, timeout=0):
+        captured["auth"] = req.headers.get("Authorization")
+        captured["url"] = req.full_url
+        captured["timeout"] = timeout
+        captured["body"] = json.loads(req.data.decode("utf-8"))
+        return _StreamOnceResponse([b"audio-openai"])
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(routes, "_tts_open", lambda req, **kw: _fake_urlopen(req))
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.2")
+    routes._handle_tts(h, None)
+
+    assert h.status == 200
+    assert h.sent_headers["Content-Type"] == "audio/mpeg"
+    assert h.wfile.getvalue() == b"audio-openai"
+    assert captured["auth"] == "Bearer sk-openai"
+    assert captured["url"] == "https://api.openai.com/v1/audio/speech"
+    assert captured["body"] == {"model": "gpt-4o-mini-tts", "input": "Hello", "voice": "alloy"}
+
+
+def test_openai_tts_prefers_voice_tools_key_over_openai_key(monkeypatch):
+    captured = {}
+
+    def _fake_urlopen(req, timeout=0):
+        captured["auth"] = req.headers.get("Authorization")
+        return _StreamOnceResponse([b"preferred"])
+
+    monkeypatch.setenv("VOICE_TOOLS_OPENAI_KEY", "sk-voice-tools")
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(routes, "_tts_open", lambda req, **kw: _fake_urlopen(req))
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.3")
+    routes._handle_tts(h, None)
+
+    assert h.status == 200
+    assert captured["auth"] == "Bearer sk-voice-tools"
+
+
+def test_openai_tts_config_overrides(monkeypatch):
+    captured = {}
+    import api.config as config
+
+    def _fake_urlopen(req, timeout=0):
+        captured["url"] = req.full_url
+        captured["body"] = json.loads(req.data.decode("utf-8"))
+        return _StreamOnceResponse([b"custom"])
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(config, "get_config", lambda: {
+        "tts": {"openai": {"base_url": "https://custom.example.com/v1/", "model": "tts-custom", "voice": "nova"}}
+    })
+    monkeypatch.setattr(routes, "_tts_open", lambda req, **kw: _fake_urlopen(req))
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.4")
+    routes._handle_tts(h, None)
+
+    assert h.status == 200
+    assert captured["url"] == "https://custom.example.com/v1/audio/speech"
+    assert captured["body"] == {"model": "tts-custom", "input": "Hello", "voice": "nova"}
+
+
+@pytest.mark.parametrize("base_url", [
+    "http://169.254.169.254/v1",
+    "https://user:pass@api.example.com/v1",
+    "http://user:pass@localhost:8080/v1",
+    # SSRF: https to private / link-local / loopback / reserved literal IPs must
+    # be rejected even though the scheme is https (regression for #5079 gate).
+    "https://169.254.169.254/v1",
+    "https://10.0.0.5/v1",
+    "https://192.168.1.10/v1",
+    "https://127.0.0.1/v1",
+    "https://[::1]/v1",
+])
+def test_openai_tts_rejects_invalid_base_url_config(monkeypatch, base_url):
+    import api.config as config
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(config, "get_config", lambda: {
+        "tts": {"openai": {"base_url": base_url}}
+    })
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.5")
+    routes._handle_tts(h, None)
+
+    assert h.status == 400
+    assert "base_url" in (h.payload() or {}).get("error", "")
+
+
+def test_openai_tts_rejects_non_audio_upstream_response(monkeypatch):
+    def _fake_urlopen(_req, timeout=0):
+        return _StreamOnceResponse(
+            [b'{"error":"nope"}'],
+            headers={"Content-Type": "application/json"},
+        )
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(routes, "_tts_open", lambda req, **kw: _fake_urlopen(req))
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.6")
+    routes._handle_tts(h, None)
+
+    assert h.status == 502
+    assert "OpenAI TTS generation failed" in (h.payload() or {}).get("error", "")
+
+
+def test_openai_tts_does_not_follow_upstream_redirect(monkeypatch):
+    # SSRF / credential-leak regression (#5079 gate): an upstream redirect must
+    # NOT be followed — following it would carry the Authorization bearer to the
+    # redirect target and could bounce the request to a private/link-local host
+    # after base-url validation already passed. The no-redirect opener raises,
+    # which surfaces as a 502 (generation failed), never a second request.
+    import urllib.error
+
+    def _redirecting_open(req, **kw):
+        # Simulate urllib raising on a redirect the way HTTPRedirectHandler would
+        # when redirect_request() raises (the no-redirect handler's behavior).
+        raise urllib.error.HTTPError(
+            req.full_url, 302, "Found",
+            {"Location": "http://169.254.169.254/v1/audio/speech"}, None,
+        )
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(routes, "_tts_open", _redirecting_open)
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.9")
+    routes._handle_tts(h, None)
+
+    assert h.status in (500, 502)
+    assert "OpenAI TTS generation failed" in (h.payload() or {}).get("error", "")
+
+
+def test_openai_tts_rejects_oversized_upstream_audio(monkeypatch):
+    def _fake_urlopen(_req, timeout=0):
+        return _StreamOnceResponse([b"1234", b"5"], headers={"Content-Type": "audio/mpeg"})
+
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-openai")
+    monkeypatch.setattr(routes, "_TTS_PROXY_MAX_BYTES", 4)
+    monkeypatch.setattr(routes, "_tts_open", lambda req, **kw: _fake_urlopen(req))
+    h = _post({"text": "Hello", "engine": "openai"}, client="10.82.0.7")
+    routes._handle_tts(h, None)
+
+    assert h.status == 502
+    assert "OpenAI TTS generation failed" in (h.payload() or {}).get("error", "")
+
+
+def test_openai_option_in_html():
+    src = (STATIC_DIR / "index.html").read_text(encoding="utf-8")
+    assert '<option value="openai">OpenAI TTS (server)</option>' in src
+
+
+def test_openai_voice_placeholder_in_panels():
+    src = (STATIC_DIR / "panels.js").read_text(encoding="utf-8")
+    assert "engine==='openai'" in src
+    assert 'OpenAI voice (server-configured)' in src
+
+
+def test_play_openai_tts_exists_in_ui_js():
+    src = (STATIC_DIR / "ui.js").read_text(encoding="utf-8")
+    assert 'function _playOpenaiTts(text, btn)' in src
+    assert "body:JSON.stringify({text:text, engine:'openai'})" in src
+
+
+def test_boot_js_handles_openai_engine():
+    src = (STATIC_DIR / "boot.js").read_text(encoding="utf-8")
+    assert 'if(engine==="openai")' in src
+    assert "body: JSON.stringify({text: clean, engine: 'openai'})" in src