fix(security): constrain endpoint probes to trusted origins (#1447)

Validate HTTPS probe destinations against the shipped origin allowlist and clean up the related CodeQL test findings. Reviewed by Greptile; CodeRabbit was harvested but rate-limited. CI, Security, and cross-platform smoke checks are green.

CHANGELOG.md

@@ -41,6 +41,7 @@ The bundled TTS model package (`pyproject.toml`) is versioned independently.
 
 ### Fixed
 
+- Automatic model-mirror checks now reject untrusted URLs before opening a network connection. (#1447)
 - Sidecar engines no longer break when a library they load prints to the console. Those bytes landed in the middle of the engine's data stream, failing the generation and leaving the connection scrambled for every request after it. (#1428) — thanks @1335-Group!
 - A generation abandoned while stuck on an internal lock now says so, instead of blaming your hardware and suggesting shorter text. Nothing had been computed, so none of that advice applied. (#1416, #1419)
 - A machine with a GPU that ends up on CPU now says why — a missing device node, a permissions problem, a card newer than the installed ROCm, an `HSA_OVERRIDE_GFX_VERSION` that is doing more harm than good, or an NVIDIA driver the container can't reach each read differently. Before, all of them looked identical to having no GPU at all. (#1274, #1228)

backend/services/endpoint_race.py

@@ -83,6 +83,26 @@ _race_lock = threading.Lock()
 _FAILOVER_ATTEMPTED: set[str] = set()
 
 
+def _is_allowed_probe_endpoint(endpoint: str) -> bool:
+    """Only probe the two fixed HTTPS origins shipped by VoiceStudio."""
+    try:
+        parsed = urlsplit(endpoint)
+        port = parsed.port
+    except (TypeError, ValueError):
+        return False
+    return (
+        parsed.scheme == "https"
+        and parsed.hostname in {urlsplit(CANONICAL_ENDPOINT).hostname,
+                                urlsplit(COMMUNITY_MIRROR).hostname}
+        and port in (None, 443)
+        and parsed.username is None
+        and parsed.password is None
+        and parsed.path in ("", "/")
+        and not parsed.query
+        and not parsed.fragment
+    )
+
+
 @dataclass
 class ProbeResult:
     endpoint: str
@@ -120,11 +140,13 @@ def probe_endpoint(endpoint: str, timeout: float = PROBE_TIMEOUT_S) -> ProbeResu
     Any HTTP response (even an error status) counts as reachable — the probe
     measures whether the network path works, not whether a specific resource
     exists. Never raises."""
+    if not _is_allowed_probe_endpoint(endpoint):
+        return ProbeResult(endpoint=endpoint, reachable=False, error="invalid_endpoint")
     url = endpoint.rstrip("/") + "/"
     req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "VoiceStudio-endpoint-probe"})
     start = time.monotonic()
     try:
-        with urllib.request.urlopen(req, timeout=timeout):
+        with urllib.request.urlopen(req, timeout=timeout):  # nosec B310 -- fixed HTTPS allowlist above
             pass
     except urllib.error.HTTPError:
         pass  # the server answered → reachable
@@ -143,6 +165,8 @@ def throughput_probe(endpoint: str, timeout: float = PROBE_TIMEOUT_S) -> Optiona
     Used only as a tiebreak confirmation when latency says the mirror is
     decisively faster — throughput is what a multi-GB download actually
     feels. Best-effort; any failure returns None (tiebreak skipped)."""
+    if not _is_allowed_probe_endpoint(endpoint):
+        return None
     url = endpoint.rstrip("/") + _THROUGHPUT_SAMPLE_PATH
     req = urllib.request.Request(
         url,
@@ -155,7 +179,7 @@ def throughput_probe(endpoint: str, timeout: float = PROBE_TIMEOUT_S) -> Optiona
     total = 0
     start = time.monotonic()
     try:
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:  # nosec B310 -- fixed HTTPS allowlist above
             while total < _THROUGHPUT_SAMPLE_BYTES and time.monotonic() < deadline:
                 chunk = resp.read(min(65536, _THROUGHPUT_SAMPLE_BYTES - total))
                 if not chunk:

backend/tests/test_batch.py

@@ -180,7 +180,8 @@ class TestDeleteJob:
         assert client.get(f"/batch/jobs/{r['job_id']}").status_code == 404
 
     def test_delete_not_found(self, client):
-        assert client.delete("/batch/jobs/nope").status_code == 404
+        response = client.delete("/batch/jobs/nope")
+        assert response.status_code == 404
 
 
 class TestSetProgress:

backend/tests/test_run_sentinel.py

@@ -15,6 +15,7 @@ import os
 import subprocess
 import sys
 import time
+from pathlib import Path
 
 import pytest
 from fastapi import FastAPI
@@ -244,12 +245,13 @@ def test_version_gate_hides_other_release_records_and_reads_never_write(sentinel
     store["records"][0]["version"] = "0.0.1"
     with open(run_sentinel.CRASH_RECORD_PATH, "w", encoding="utf-8") as f:
         json.dump(store, f)
-    before = open(run_sentinel.CRASH_RECORD_PATH, "rb").read()
+    before = Path(run_sentinel.CRASH_RECORD_PATH).read_bytes()
 
     assert run_sentinel.newest_record("9.9.9") is None, "other release = stale"
     # Preview stamps match their base release (X.Y.Z-N == X.Y.Z).
     assert run_sentinel.newest_record("0.0.1-7") is not None
-    assert open(run_sentinel.CRASH_RECORD_PATH, "rb").read() == before, (
+    actual = Path(run_sentinel.CRASH_RECORD_PATH).read_bytes()
+    assert actual == before, (
         "the read path must never write (crash.rs read-only contract)"
     )
     # Versionless legacy records never surface either.

tests/test_endpoint_race.py

@@ -79,6 +79,30 @@ def test_hint_only_reorders_never_drops(er):
         assert set(prober.calls) == {er.CANONICAL_ENDPOINT, er.COMMUNITY_MIRROR}
 
 
+@pytest.mark.parametrize(
+    "endpoint",
+    [
+        "file:///etc/passwd",
+        "http://huggingface.co",
+        "https://huggingface.co.evil.example",
+        "https://huggingface.co@evil.example",
+        "https://huggingface.co:444",
+        "https://huggingface.co/model",
+        "https://huggingface.co?redirect=file:///etc/passwd",
+    ],
+)
+def test_probe_rejects_unapproved_origins(er, endpoint):
+    # The suite-wide network guard replaces the actual prober, so exercise the
+    # validation chokepoint directly. Both real network helpers call it before
+    # constructing a Request or reaching urlopen.
+    assert er._is_allowed_probe_endpoint(endpoint) is False
+
+
+@pytest.mark.parametrize("endpoint", ["https://huggingface.co", "https://hf-mirror.com/"])
+def test_probe_allows_only_shipped_https_origins(er, endpoint):
+    assert er._is_allowed_probe_endpoint(endpoint) is True
+
+
 # ── Decision policy matrix ──────────────────────────────────────────────────
 
 def test_both_reachable_similar_latency_prefers_canonical(er):