"""Language allow-list contracts for the private Hermes chat TTS voice policy.""" from __future__ import annotations import importlib.util import io import json import sys from types import SimpleNamespace import pytest from testing.tests.test_hermes_chat_support import ROOT AMY = "en_US-amy-medium" IRINA = "ru_RU-irina-medium" CLAUDE = "es_MX-claude-high" def _load_tts_server(monkeypatch): server_path = ROOT / "dockerfiles" / "hermes-jetson-tts-server.py" spec = importlib.util.spec_from_file_location("hermes_jetson_tts_server", server_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) class _FakeSessionOptions: def __init__(self) -> None: self.intra_op_num_threads = None self.inter_op_num_threads = None fake_onnxruntime = SimpleNamespace( SessionOptions=_FakeSessionOptions, InferenceSession=lambda *a, **k: SimpleNamespace(), ) fake_piper = SimpleNamespace( PiperConfig=SimpleNamespace(from_dict=lambda d: d), PiperVoice=lambda **kwargs: SimpleNamespace(**kwargs), SynthesisConfig=lambda **kwargs: SimpleNamespace(**kwargs), ) monkeypatch.setitem(sys.modules, "onnxruntime", fake_onnxruntime) monkeypatch.setitem(sys.modules, "piper", fake_piper) spec.loader.exec_module(module) return module @pytest.fixture def tts(monkeypatch): return _load_tts_server(monkeypatch) @pytest.mark.parametrize( "language,expected", [ ("en", AMY), ("en-US", AMY), ("en_US", AMY), ("EN", AMY), ("En-Us", AMY), ("ru", IRINA), ("ru-RU", IRINA), ("ru_RU", IRINA), ("RU", IRINA), ("es", CLAUDE), ("es-MX", CLAUDE), ("es_MX", CLAUDE), ("es-ES", CLAUDE), ("es_ES", CLAUDE), ("ES", CLAUDE), ], ) def test_allow_listed_languages_resolve_to_the_approved_voice(tts, language, expected): assert tts.resolve_voice_name(language) == expected @pytest.mark.parametrize( "language", [ None, "", " ", "fr", "fr-FR", "de-DE", "xx", "en-GB", "es-AR", "english", 123, 1.5, True, [], {}, {"lang": "ru"}, "../../etc/passwd", "en_US-amy-medium/../../ru_RU-irina-medium", "\x00ru", "ru\x00", ], ) def test_unknown_missing_or_malformed_language_falls_back_to_amy(tts, language): assert tts.resolve_voice_name(language) == AMY def test_default_voice_name_matches_the_dockerfile_env_default(tts): assert tts.DEFAULT_VOICE_NAME == AMY def test_deployment_does_not_override_the_image_default() -> None: """An old single-voice image and the new image can cross the Flux transition.""" manifest = (ROOT / "services/hermes/voice-deployment.yaml").read_text() assert "HERMES_TTS_VOICE" not in manifest assert "piper-multilingual-en-ru-es" in manifest def test_resolved_voice_is_always_one_of_the_three_baked_names(tts): assert frozenset({AMY, IRINA, CLAUDE}) == tts.BAKED_VOICE_NAMES fuzz_inputs = [ "en", "ru", "es", "unknown", "", None, 42, "../../../etc/shadow", "en_US-amy-medium\x00; rm -rf /", "RU-ru", "Es-Es", "en-us-extra", ] for value in fuzz_inputs: assert tts.resolve_voice_name(value) in tts.BAKED_VOICE_NAMES def test_client_voice_field_cannot_override_the_language_policy(tts): """The POST handler must select the voice from "language" only. A malicious or stale "voice" field in a hostile/legacy request must never change which baked model answers the request. """ calls: list[str] = [] class _RecordingVoice: def __init__(self, name: str) -> None: self.name = name def synthesize_wav(self, text, wav_file, syn_config) -> None: calls.append(self.name) wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(16_000) wav_file.writeframes(b"\x00\x00") class _RecordingHandler(tts.SpeechHandler): def __init__(self, payload): request = json.dumps(payload).encode("utf-8") self.path = "/v1/audio/speech" self.headers = {"Content-Length": str(len(request))} self.rfile = io.BytesIO(request) self.wfile = io.BytesIO() self.status = None self.response_headers = {} self.server = SimpleNamespace( voices={ AMY: _RecordingVoice(AMY), IRINA: _RecordingVoice(IRINA), CLAUDE: _RecordingVoice(CLAUDE), }, default_voice_name=AMY, ) def send_response(self, status, message=None): self.status = status def send_header(self, name, value): self.response_headers[name] = value def end_headers(self): return None # A payload that supplies an attacker/legacy "voice" value but no # language must resolve to the safe default, never the "voice" value. handler = _RecordingHandler({"input": "hi", "voice": IRINA}) handler.do_POST() assert handler.status == 200 assert handler.response_headers["X-TTS-Voice"] == AMY # A payload supplying both must still be governed by "language" alone. handler = _RecordingHandler({"input": "hi", "voice": CLAUDE, "language": "ru"}) handler.do_POST() assert handler.status == 200 assert handler.response_headers["X-TTS-Voice"] == IRINA assert calls == [AMY, IRINA] def test_no_client_string_reaches_a_filesystem_path(tts): """resolve_voice_name must only ever return a fixed, baked literal. This is the property that keeps a client from ever causing the server to build a Path out of attacker-controlled text: the return value is always a member of the fixed allow-list, regardless of input shape. """ hostile_inputs = [ "../../../../etc/passwd", "/etc/passwd", "en_US-amy-medium/../../../etc/passwd", "ru_RU-irina-medium\x00.onnx", "es_MX-claude-high; cat /etc/shadow", "\n\ren", "en" + "/" * 200, " ", ] for value in hostile_inputs: result = tts.resolve_voice_name(value) assert result in tts.BAKED_VOICE_NAMES assert "/" not in result assert ".." not in result assert "\x00" not in result def test_normalize_language_rejects_non_string_input(tts): assert tts.normalize_language(None) is None assert tts.normalize_language(123) is None assert tts.normalize_language([]) is None assert tts.normalize_language("") is None assert tts.normalize_language(" ") is None assert tts.normalize_language("En_US") == "en-us" def test_default_voice_name_is_one_of_the_baked_voices(tts): assert tts.DEFAULT_VOICE_NAME in tts.BAKED_VOICE_NAMES @pytest.mark.parametrize( "value,expected", [ ("turn-42", "turn-42"), (" session_1:chunk.3 ", "session_1:chunk.3"), (None, None), (42, None), ("", None), ("turn\r\nX-Injected: true", None), ("/not/header-safe", None), ("x" * 129, None), ], ) def test_turn_ids_are_safe_to_echo_in_response_headers(tts, value, expected): assert tts.normalize_turn_id(value) == expected def _decode_chunked_response(data: bytes) -> bytes: decoded = bytearray() cursor = 0 while True: line_end = data.index(b"\r\n", cursor) size = int(data[cursor:line_end], 16) cursor = line_end + 2 if size == 0: assert data[cursor:] == b"\r\n" break decoded.extend(data[cursor : cursor + size]) cursor += size assert data[cursor : cursor + 2] == b"\r\n" cursor += 2 return bytes(decoded) def test_stream_endpoint_returns_progressive_pcm_and_preserves_voice_policy(tts): events: list[str] = [] first_pcm = b"\x01\x00" * (tts.STREAM_WRITE_BYTES // 2 + 3) second_pcm = b"\x02\x00\x03\x00" class _StreamingVoice: config = SimpleNamespace(sample_rate=22_050) def synthesize(self, text, syn_config): assert text == "Hola mundo. Otra frase." assert syn_config.length_scale == pytest.approx(0.8) assert "headers" in events events.append("synthesis") yield SimpleNamespace( sample_rate=22_050, sample_width=2, sample_channels=1, audio_int16_bytes=first_pcm, ) yield SimpleNamespace( sample_rate=22_050, sample_width=2, sample_channels=1, audio_int16_bytes=second_pcm, ) class _StreamingHandler(tts.SpeechHandler): def __init__(self): payload = json.dumps( { "text": "Hola mundo. Otra frase.", "language": "es", "speed": 1.25, "turn_id": "turn-9:chunk.1", } ).encode("utf-8") self.path = "/v1/audio/speech/stream" self.headers = {"Content-Length": str(len(payload))} self.rfile = io.BytesIO(payload) self.wfile = io.BytesIO() self.status = None self.response_headers = {} self.close_connection = False self.server = SimpleNamespace( voices={CLAUDE: _StreamingVoice()}, default_voice_name=AMY, ) def send_response(self, status, message=None): self.status = status def send_header(self, name, value): self.response_headers[name] = value def end_headers(self): events.append("headers") handler = _StreamingHandler() handler.do_POST() assert handler.status == 200 assert handler.response_headers["Transfer-Encoding"] == "chunked" assert handler.response_headers["X-Audio-Format"] == "pcm_s16le" assert handler.response_headers["X-Audio-Sample-Rate"] == "22050" assert handler.response_headers["X-Audio-Channels"] == "1" assert handler.response_headers["X-Audio-Sample-Width"] == "2" assert handler.response_headers["X-TTS-Voice"] == CLAUDE assert handler.response_headers["X-TTS-Turn-ID"] == "turn-9:chunk.1" assert events == ["headers", "synthesis"] assert _decode_chunked_response(handler.wfile.getvalue()) == first_pcm + second_pcm def test_stream_disconnect_stops_before_synthesizing_another_sentence(tts): generated: list[int] = [] class _StreamingVoice: config = SimpleNamespace(sample_rate=16_000) def synthesize(self, text, syn_config): for sentence in (1, 2): generated.append(sentence) yield SimpleNamespace( sample_rate=16_000, sample_width=2, sample_channels=1, audio_int16_bytes=b"\x00\x00", ) class _DisconnectedHandler(tts.SpeechHandler): def __init__(self): self.wfile = io.BytesIO() self.response_headers = {} self.close_connection = False def send_response(self, status, message=None): return None def send_header(self, name, value): self.response_headers[name] = value def end_headers(self): return None def _write_stream_chunk(self, audio): self.close_connection = True return False handler = _DisconnectedHandler() handler._stream_pcm("hello", 1.0, AMY, _StreamingVoice(), "turn-1") assert generated == [1] assert handler.close_connection is True def test_load_voices_fails_closed_when_a_baked_model_is_missing(tts, tmp_path): with pytest.raises(RuntimeError, match="baked Piper voice is missing"): tts.load_voices(tmp_path, threads=1)