diff --git a/CHANGELOG.md b/CHANGELOG.md index de749df7..91640e7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ The bundled TTS model package (`pyproject.toml`) is versioned independently. ### Fixed +- GPT-SoVITS connections now stay on loopback or explicitly trusted networks and cannot escape through redirects or DNS rebinding. (#1463) - Engine discovery no longer exposes probe exceptions, local paths or credentials in API responses and logs. (#1460) - Failed gallery, batch-video, and desktop-log cleanup is now reported instead of silently claiming success, and diagnostic redaction fails closed if a scrubber breaks. (#1458) - Remote backends can no longer probe or overwrite arbitrary host files through native-only tools, and imported or persisted paths cannot escape their VoiceStudio data folders. (#1455) diff --git a/README.md b/README.md index cd21b315..00acb7fa 100644 --- a/README.md +++ b/README.md @@ -255,6 +255,11 @@ Professional-grade voice AI, minus the subscription and the cloud. | **dots.tts** ⚡ (2B) | 24 | ✅ | — | ✅ CUDA/CPU | ✅ CPU | ❌ | Apache-2.0 | | **Confucius4-TTS** ⚡ | 14 | ✅ | — | ✅ CUDA/CPU | ✅ CPU | ✅ CUDA/CPU | Apache-2.0 | +GPT-SoVITS connects to `http://127.0.0.1:9880` by default. To use a server on +another machine, set `OMNIVOICE_GPTSOVITS_URL` to its credential-free +`http://` or `https://` origin and add that machine's CIDR to +`OMNIVOICE_TRUSTED_NETWORKS`; redirects and untrusted destinations are rejected. + > **CUDA** = GPU-accelerated · **MPS** = Apple Silicon Metal · **CPU** = runs everywhere, slower for large models · KittenTTS and MOSS-TTS-Nano run realtime on CPU · MLX-Audio is Apple Silicon only · ⚡ = lazy-registered (installed on first use) > > **Clone** matters beyond single-clip generation: Video Dubbing (and any Batch job with a pinned voice) needs reference-audio cloning to preserve speaker identity, so picking a Clone-less engine (KittenTTS, Sherpa-ONNX, Supertonic 3) as the active engine fails those jobs up front with an actionable message instead of silently falling back to VoiceStudio. diff --git a/backend/services/outbound_http.py b/backend/services/outbound_http.py new file mode 100644 index 00000000..b3304589 --- /dev/null +++ b/backend/services/outbound_http.py @@ -0,0 +1,113 @@ +"""Pinned HTTP transport for explicitly configured local/trusted services.""" +from __future__ import annotations + +import http.client +import re +import socket +from dataclasses import dataclass +from urllib.parse import urlsplit + +from api.dependencies import is_local_host + + +class UnsafeEndpoint(ValueError): + """The configured endpoint is outside VoiceStudio's trusted networks.""" + + +@dataclass(frozen=True) +class ResolvedEndpoint: + scheme: str + host: str + port: int + ip: str + + +_IP_PREFIX_HOST_RE = re.compile(r"^(?:\d{1,3}\.){3}\d{1,3}\.") + + +def resolve_trusted_endpoint(url: str) -> ResolvedEndpoint: + """Validate and resolve a root HTTP(S) endpoint to one trusted address. + + Loopback is trusted by default. Non-loopback targets require an explicit + match in ``OMNIVOICE_TRUSTED_NETWORKS``, the same policy used for remote + inference consumers. Every DNS answer must be trusted; mixed answers are + rejected rather than choosing a convenient one. + """ + try: + parsed = urlsplit(url) + port = parsed.port or (443 if parsed.scheme == "https" else 80) + except (TypeError, ValueError) as exc: + raise UnsafeEndpoint("invalid endpoint URL") from exc + if ( + parsed.scheme not in {"http", "https"} + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.path not in {"", "/"} + or parsed.query + or parsed.fragment + or parsed.hostname.lower().startswith("localhost.") + or _IP_PREFIX_HOST_RE.match(parsed.hostname) + ): + raise UnsafeEndpoint("endpoint must be a credential-free HTTP(S) origin") + try: + answers = socket.getaddrinfo(parsed.hostname, port, type=socket.SOCK_STREAM) + except OSError as exc: + raise UnsafeEndpoint("endpoint host could not be resolved") from exc + ips = list(dict.fromkeys(answer[4][0] for answer in answers)) + if not ips or any(not is_local_host(ip) for ip in ips): + raise UnsafeEndpoint("endpoint is outside loopback or OMNIVOICE_TRUSTED_NETWORKS") + return ResolvedEndpoint(parsed.scheme, parsed.hostname, port, ips[0]) + + +class _PinnedHTTPConnection(http.client.HTTPConnection): + def __init__(self, endpoint: ResolvedEndpoint, timeout: float): + super().__init__(endpoint.host, endpoint.port, timeout=timeout) + self._pinned_ip = endpoint.ip + + def connect(self) -> None: + self.sock = self._create_connection( + (self._pinned_ip, self.port), self.timeout, self.source_address + ) + + +class _PinnedHTTPSConnection(http.client.HTTPSConnection): + def __init__(self, endpoint: ResolvedEndpoint, timeout: float): + super().__init__(endpoint.host, endpoint.port, timeout=timeout) + self._pinned_ip = endpoint.ip + + def connect(self) -> None: + sock = self._create_connection( + (self._pinned_ip, self.port), self.timeout, self.source_address + ) + self.sock = self._context.wrap_socket(sock, server_hostname=self.host) + + +def open_trusted_endpoint( + base_url: str, + *, + method: str, + query: str = "", + timeout: float, +) -> http.client.HTTPResponse: + """Open one request without redirects, pinned to the validated DNS answer.""" + endpoint = resolve_trusted_endpoint(base_url) + conn_cls = _PinnedHTTPSConnection if endpoint.scheme == "https" else _PinnedHTTPConnection + conn = conn_cls(endpoint, timeout) + target = "/" + (f"?{query}" if query else "") + # Let http.client format the authority from the validated host and port. + # Supplying the hostname ourselves drops non-default ports and IPv6 + # brackets, which can make Host-aware inference servers misroute requests. + conn.request(method, target) + response = conn.getresponse() + # Redirects are never followed: a configured inference origin must answer + # directly, so a Location header cannot escape the validated connection. + if 300 <= response.status < 400: + response.close() + conn.close() + raise UnsafeEndpoint("endpoint redirects are not allowed") + if response.status >= 400: + response.close() + conn.close() + raise OSError(f"endpoint returned HTTP {response.status}") + return response diff --git a/backend/services/tts_backend.py b/backend/services/tts_backend.py index ebeea21c..55473cad 100644 --- a/backend/services/tts_backend.py +++ b/backend/services/tts_backend.py @@ -1695,11 +1695,11 @@ class GPTSoVITSBackend(TTSBackend): @classmethod def is_available(cls) -> tuple[bool, str]: # GPT-SoVITS runs as an external API server — check if it's reachable. - import urllib.request + from services.outbound_http import open_trusted_endpoint url = os.environ.get("OMNIVOICE_GPTSOVITS_URL", "http://127.0.0.1:9880") try: - req = urllib.request.Request(f"{url}/", method="GET") - urllib.request.urlopen(req, timeout=2) + with open_trusted_endpoint(url, method="GET", timeout=2): + pass return True, "ready (server reachable)" except Exception: return False, ( @@ -1717,8 +1717,8 @@ class GPTSoVITSBackend(TTSBackend): return ["zh", "en", "ja", "yue", "ko"] def generate(self, text: str, **kw) -> torch.Tensor: - import urllib.request import urllib.parse + from services.outbound_http import open_trusted_endpoint ref_audio = kw.get("ref_audio") ref_text = kw.get("ref_text", "") @@ -1746,11 +1746,10 @@ class GPTSoVITSBackend(TTSBackend): params["speed_factor"] = str(speed) query = urllib.parse.urlencode(params) - url = f"{self._url}/?{query}" - try: - req = urllib.request.Request(url, method="POST") - with urllib.request.urlopen(req, timeout=120) as resp: + with open_trusted_endpoint( + self._url, method="POST", query=query, timeout=120 + ) as resp: audio_bytes = resp.read() except Exception as e: raise RuntimeError( diff --git a/tests/test_gptsovits_endpoint_security.py b/tests/test_gptsovits_endpoint_security.py new file mode 100644 index 00000000..0d150eb2 --- /dev/null +++ b/tests/test_gptsovits_endpoint_security.py @@ -0,0 +1,205 @@ +"""GPT-SoVITS outbound requests stay on loopback or explicit trusted CIDRs.""" +import importlib +import socket + +import pytest + + +@pytest.fixture +def outbound_http(): + return importlib.import_module("services.outbound_http") + + +def _answer(ip: str, port: int = 9880): + family = socket.AF_INET6 if ":" in ip else socket.AF_INET + return [(family, socket.SOCK_STREAM, 6, "", (ip, port))] + + +@pytest.mark.parametrize( + "url", + [ + "file:///etc/passwd", + "ftp://127.0.0.1/resource", + "http://127.0.0.1.evil.example:9880", + "http://127.0.0.1@evil.example:9880", + "http://user:secret@127.0.0.1:9880", + "http://127.0.0.1:9880/admin", + "http://127.0.0.1:9880/?next=http://169.254.169.254", + ], +) +def test_rejects_non_origin_and_host_spoof_urls(outbound_http, monkeypatch, url): + monkeypatch.setattr(socket, "getaddrinfo", lambda *_args, **_kwargs: _answer("127.0.0.1")) + with pytest.raises(outbound_http.UnsafeEndpoint): + outbound_http.resolve_trusted_endpoint(url) + + +def test_private_network_requires_explicit_existing_trust_policy(outbound_http, monkeypatch): + monkeypatch.setattr(socket, "getaddrinfo", lambda *_args, **_kwargs: _answer("192.168.4.20")) + monkeypatch.delenv("OMNIVOICE_TRUSTED_NETWORKS", raising=False) + with pytest.raises(outbound_http.UnsafeEndpoint): + outbound_http.resolve_trusted_endpoint("http://gptsovits.lan:9880") + + monkeypatch.setenv("OMNIVOICE_TRUSTED_NETWORKS", "192.168.4.0/24") + endpoint = outbound_http.resolve_trusted_endpoint("http://gptsovits.lan:9880") + assert endpoint.ip == "192.168.4.20" + + +def test_mixed_dns_answers_are_rejected(outbound_http, monkeypatch): + monkeypatch.setenv("OMNIVOICE_TRUSTED_NETWORKS", "10.0.0.0/8") + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda *_args, **_kwargs: _answer("10.2.3.4") + _answer("169.254.169.254"), + ) + with pytest.raises(outbound_http.UnsafeEndpoint): + outbound_http.resolve_trusted_endpoint("http://gptsovits.internal:9880") + + +class _Response: + def __init__(self, status=200): + self.status = status + self.closed = False + + def close(self): + self.closed = True + + +class _Connection: + instances = [] + + def __init__(self, endpoint, timeout): + self.endpoint = endpoint + self.timeout = timeout + self.request_args = None + self.response = _Response() + self.closed = False + self.instances.append(self) + + def request(self, *args, **kwargs): + self.request_args = (args, kwargs) + + def getresponse(self): + return self.response + + def close(self): + self.closed = True + + +class _CaptureSocket: + def __init__(self): + self.chunks = [] + + def sendall(self, data): + self.chunks.append(data) + + +@pytest.mark.parametrize( + ("connection_kind", "endpoint_args", "expected_host"), + [ + ( + "http", + ("http", "127.0.0.1", 80, "127.0.0.1"), + b"Host: 127.0.0.1\r\n", + ), + ( + "http", + ("http", "localhost", 9880, "127.0.0.1"), + b"Host: localhost:9880\r\n", + ), + ( + "http", + ("http", "::1", 9880, "::1"), + b"Host: [::1]:9880\r\n", + ), + ( + "https", + ("https", "localhost", 443, "127.0.0.1"), + b"Host: localhost\r\n", + ), + ], +) +def test_http_client_builds_complete_host_authority( + outbound_http, connection_kind, endpoint_args, expected_host +): + connection_cls = ( + outbound_http._PinnedHTTPSConnection + if connection_kind == "https" + else outbound_http._PinnedHTTPConnection + ) + endpoint = outbound_http.ResolvedEndpoint(*endpoint_args) + connection = connection_cls(endpoint, timeout=2) + capture = _CaptureSocket() + connection.sock = capture + + connection.request("GET", "/") + + wire = b"".join(capture.chunks) + assert expected_host in wire + + +def test_valid_endpoint_is_pinned_to_the_single_validated_dns_answer( + outbound_http, monkeypatch +): + calls = 0 + + def changing_dns(*_args, **_kwargs): + nonlocal calls + calls += 1 + return _answer("127.0.0.1" if calls == 1 else "169.254.169.254") + + _Connection.instances.clear() + monkeypatch.setattr(socket, "getaddrinfo", changing_dns) + monkeypatch.setattr(outbound_http, "_PinnedHTTPConnection", _Connection) + response = outbound_http.open_trusted_endpoint( + "http://localhost:9880", method="POST", query="text=hello", timeout=5 + ) + + connection = _Connection.instances[0] + assert calls == 1 + assert connection.endpoint.ip == "127.0.0.1" + assert connection.request_args[0] == ("POST", "/?text=hello") + assert connection.request_args[1] == {} + assert response.status == 200 + + +def test_redirect_is_rejected_without_following_location(outbound_http, monkeypatch): + _Connection.instances.clear() + monkeypatch.setattr(socket, "getaddrinfo", lambda *_args, **_kwargs: _answer("127.0.0.1")) + monkeypatch.setattr(outbound_http, "_PinnedHTTPConnection", _Connection) + original_init = _Connection.__init__ + + def redirecting_init(self, endpoint, timeout): + original_init(self, endpoint, timeout) + self.response = _Response(302) + + monkeypatch.setattr(_Connection, "__init__", redirecting_init) + with pytest.raises(outbound_http.UnsafeEndpoint, match="redirects"): + outbound_http.open_trusted_endpoint( + "http://127.0.0.1:9880", method="GET", timeout=2 + ) + assert _Connection.instances[0].closed is True + + +def test_gptsovits_availability_uses_valid_configured_endpoint( + outbound_http, monkeypatch +): + from services.tts_backend import GPTSoVITSBackend + + calls = [] + + class _ContextResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + monkeypatch.setenv("OMNIVOICE_GPTSOVITS_URL", "http://127.0.0.1:9880") + monkeypatch.setattr( + outbound_http, + "open_trusted_endpoint", + lambda url, **kwargs: calls.append((url, kwargs)) or _ContextResponse(), + ) + + assert GPTSoVITSBackend.is_available() == (True, "ready (server reachable)") + assert calls == [("http://127.0.0.1:9880", {"method": "GET", "timeout": 2})]