atlas-iac/testing/tests/test_hermes_tts_language_routing.py

"""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_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


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)