fix(security): constrain GPT-SoVITS endpoints (#1463)
* fix(security): constrain GPT-SoVITS endpoints * docs: note trusted GPT-SoVITS transport * fix(security): preserve trusted endpoint authority * test(security): patch live outbound transport * test: load outbound security seam at runtime
This commit is contained in:
Palash Debnath
@@ -41,6 +41,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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
@@ -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(
|
||||
|
||||
@@ -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})]
|
||||
Reference in New Issue
Block a user