Enforce the checks the service edges only appeared to have

services/voice-gateway/.env.example

@@ -17,6 +17,11 @@ VOICE_GATEWAY_DEEPGRAM_API_KEY=
 VOICE_GATEWAY_DEEPGRAM_HOST=api.deepgram.com
 VOICE_GATEWAY_DEEPGRAM_TOKEN_TTL_SECONDS=30
 
+# Required to point DEEPGRAM_HOST at anything the URL builder does not know.
+# Off by default so a typo fails loudly instead of quietly sending audio
+# somewhere unintended; on, for the ADR-013 self-hosted deployment.
+VOICE_GATEWAY_DEEPGRAM_ALLOW_SELF_HOSTED_HOST=false
+
 VOICE_GATEWAY_STT_MODEL=nova-3-medical
 VOICE_GATEWAY_STT_ENDPOINTING_MS=400
 VOICE_GATEWAY_STT_UTTERANCE_END_MS=1500

services/voice-gateway/src/voice_gateway/app.py

@@ -103,11 +103,37 @@ async def send_audio(self, chunk: bytes) -> None:
         await self._ws.send_bytes(chunk)
 
 
+def _origin_is_allowed(origin: str | None, allowed: list[str]) -> bool:
+    """CORS does not cover this. The middleware above guards `fetch`; a browser
+    sends no preflight for a WebSocket handshake and applies no same-origin rule
+    to it, so an unchecked `accept()` lets any page on the internet open an
+    intake session. The Origin header is set by the browser and cannot be forged
+    from script, which is exactly the attacker this check is for — it is not a
+    substitute for the session auth that arrives with the real deployment.
+    """
+    if "*" in allowed:
+        return True
+    if origin is None:
+        # Not a browser. Native clients are the deployment's problem to
+        # authenticate, and here that means "not yet".
+        return False
+    return origin in allowed
+
+
 @app.websocket("/v1/intake/stream")
 async def intake_stream(websocket: WebSocket) -> None:
-    await websocket.accept()
     settings = app.state.settings
 
+    origin = websocket.headers.get("origin")
+    if not _origin_is_allowed(origin, settings.allowed_origins):
+        logger.warning("rejected an intake handshake from origin %r", origin)
+        # Before accept(), so the handshake fails outright rather than opening
+        # and immediately closing.
+        await websocket.close(code=1008, reason="origin not allowed")
+        return
+
+    await websocket.accept()
+
     session = IntakeSession(
         settings=settings,
         channel=_WebSocketChannel(websocket),

services/voice-gateway/src/voice_gateway/config.py

@@ -37,6 +37,13 @@ class Settings(BaseSettings):
     # holds is not a guarantee.
     deepgram_host: str = "api.deepgram.com"
 
+    # Setting a host the builder does not recognise is an error unless the
+    # operator says the unfamiliar host is deliberate. Off by default so a typo
+    # in `deepgram_host` fails loudly instead of quietly sending audio somewhere
+    # unintended; on, it is what makes the ADR-013 deployment configurable at
+    # all rather than a parameter nothing passes.
+    deepgram_allow_self_hosted_host: bool = False
+
     # Ephemeral tokens only need to be valid at handshake time (05 §2).
     deepgram_token_ttl_seconds: int = 30
 
@@ -88,7 +95,16 @@ class Settings(BaseSettings):
     # --- Server -----------------------------------------------------------
     host: str = "127.0.0.1"
     port: int = 8080
-    allowed_origins: list[str] = Field(default_factory=lambda: ["http://127.0.0.1:8080"])
+    # Gates both CORS and the WebSocket handshake. `localhost` is listed
+    # alongside `127.0.0.1` because they are different origins to a browser and
+    # the demo page is reachable at either — a check that rejects half the URLs
+    # the developer actually types gets widened to `*` and stays there.
+    allowed_origins: list[str] = Field(
+        default_factory=lambda: [
+            "http://127.0.0.1:8080",
+            "http://localhost:8080",
+        ]
+    )
 
     @field_validator("deepgram_host")
     @classmethod

services/voice-gateway/src/voice_gateway/deepgram/urls.py

@@ -88,7 +88,7 @@ def build_deepgram_url(
     params: dict[str, str | int | float | bool] | None = None,
     keyterms: tuple[str, ...] | list[str] = (),
     host: str | None = None,
-    allow_self_hosted_host: bool = False,
+    allow_self_hosted_host: bool | None = None,
     over_http: bool = False,
 ) -> str:
     """Build a Deepgram URL. Always opted out of model training.
@@ -107,18 +107,27 @@ def build_deepgram_url(
             unless `allow_self_hosted_host` is set.
         allow_self_hosted_host: permits a private host for the ADR-013
             self-hosted deployment. Note that the opt-out flag is still applied
-            — it is inert there, and harmless.
+            — it is inert there, and harmless. Defaults to the configured
+            setting, so the deployment is reachable by configuration rather than
+            only by a keyword argument every call site would have to remember to
+            thread through; pass it explicitly to override.
 
     Raises:
         DeepgramUrlError: on a keyterm or parameter that would fail silently.
     """
     from voice_gateway.config import get_settings
 
-    resolved_host = host or get_settings().deepgram_host
-    if resolved_host not in DEEPGRAM_HOSTS and not allow_self_hosted_host:
+    settings = get_settings()
+    resolved_host = host or settings.deepgram_host
+    allow_self_hosted = (
+        settings.deepgram_allow_self_hosted_host
+        if allow_self_hosted_host is None
+        else allow_self_hosted_host
+    )
+    if resolved_host not in DEEPGRAM_HOSTS and not allow_self_hosted:
         raise DeepgramUrlError(
             f"{resolved_host!r} is not a known Deepgram host. If this is the "
-            "self-hosted deployment, pass allow_self_hosted_host=True."
+            "self-hosted deployment, set deepgram_allow_self_hosted_host."
         )
 
     supplied = dict(params or {})
@@ -166,7 +175,7 @@ def build_listen_url(
     utterance_end_ms: int,
     keyterms: tuple[str, ...] = (),
     host: str | None = None,
-    allow_self_hosted_host: bool = False,
+    allow_self_hosted_host: bool | None = None,
 ) -> str:
     """Streaming STT connection parameters, exactly as pinned in 05 §3."""
     if not 1000 <= utterance_end_ms <= 5000:
@@ -207,7 +216,7 @@ def build_speak_url(
     container: str | None = None,
     over_http: bool = False,
     host: str | None = None,
-    allow_self_hosted_host: bool = False,
+    allow_self_hosted_host: bool | None = None,
 ) -> str:
     """TTS URL, for both the streaming socket and the REST pre-render.
 
@@ -238,7 +247,7 @@ def build_speak_url(
 
 
 def build_auth_grant_url(
-    *, host: str | None = None, allow_self_hosted_host: bool = False
+    *, host: str | None = None, allow_self_hosted_host: bool | None = None
 ) -> str:
     """Ephemeral token grant. Also goes through the builder — "all API
     requests" in 05 §6 has no carve-out for the ones that do not carry audio,

services/voice-gateway/src/voice_gateway/persistence/stub_server.py

@@ -17,10 +17,12 @@
 
 import argparse
 import logging
+import secrets
 
 import uvicorn
 from fastapi import FastAPI, Header, HTTPException
 
+from voice_gateway.config import get_settings
 from voice_gateway.contracts import TurnEvent
 
 logger = logging.getLogger(__name__)
@@ -40,6 +42,14 @@ async def persist_turn(
     if not authorization or not authorization.startswith("Bearer "):
         raise HTTPException(status_code=401, detail="missing bearer token")
 
+    # Compare it, rather than merely observing that it is shaped like a token.
+    # A stub that accepts any bearer teaches the shape of the check without the
+    # substance of it, and the TypeScript route gets written from this file.
+    # `compare_digest` because the comparison is against a shared secret.
+    expected = get_settings().persist_turn_token
+    if not secrets.compare_digest(authorization.removeprefix("Bearer "), expected):
+        raise HTTPException(status_code=401, detail="invalid bearer token")
+
     _received.append(event)
     # Identifiers only, matching what the real route may log. The transcript is
     # PHI and is not a log line here or there.

services/voice-gateway/tests/test_service_boundaries.py

@@ -0,0 +1,120 @@
+"""The three edges of this service, and what is checked at each of them.
+
+The gateway sits inside the PHI boundary and outside the persistence boundary
+(ADR-016). That makes its edges the interesting part: who may open an intake
+session, what it presents to the TypeScript tier, and where it is allowed to
+send audio. Each of these had a check that looked present and was not.
+"""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+from voice_gateway.app import _origin_is_allowed, app
+from voice_gateway.config import Settings
+from voice_gateway.deepgram.urls import DeepgramUrlError, build_auth_grant_url
+
+# --- who may open an intake session ---------------------------------------
+
+
+def test_a_page_on_another_origin_cannot_open_a_session() -> None:
+    """CORS does not cover this. A browser sends no preflight for a WebSocket
+    handshake and applies no same-origin rule to it, so the CORS middleware —
+    which is where `allowed_origins` was wired — never sees this request."""
+    with TestClient(app) as client:
+        with pytest.raises(WebSocketDisconnect) as caught:
+            with client.websocket_connect(
+                "/v1/intake/stream", headers={"Origin": "https://not-us.example"}
+            ):
+                pass
+
+    assert caught.value.code == 1008
+
+
+@pytest.mark.parametrize(
+    ("origin", "allowed", "expected"),
+    [
+        ("http://127.0.0.1:8080", ["http://127.0.0.1:8080"], True),
+        ("https://evil.example", ["http://127.0.0.1:8080"], False),
+        # No Origin means not a browser. Native clients are the deployment's
+        # problem to authenticate, and here that means "not yet".
+        (None, ["http://127.0.0.1:8080"], False),
+        (None, ["*"], True),
+        ("https://anything.example", ["*"], True),
+        # Not a prefix match: a domain that merely starts with an allowed one
+        # is a different origin.
+        ("http://127.0.0.1:8080.evil.example", ["http://127.0.0.1:8080"], False),
+    ],
+)
+def test_origin_matching(origin: str | None, allowed: list[str], expected: bool) -> None:
+    assert _origin_is_allowed(origin, allowed) is expected
+
+
+# --- what it presents to the TypeScript tier ------------------------------
+
+
+def _post(client: TestClient, token: str | None) -> int:
+    event = {
+        "session_id": "11111111-1111-1111-1111-111111111111",
+        "turn_index": 0,
+        "question_id": "knee.onset",
+        "prompt_version": "knee-v0-mock",
+        "outcome": "reprompt",
+        "stt_model": "nova-3-medical",
+        "tts_cache_hit": True,
+        "answer": None,
+    }
+    headers = {} if token is None else {"Authorization": f"Bearer {token}"}
+    return client.post("/api/internal/intake/turn", json=event, headers=headers).status_code
+
+
+def test_the_stub_checks_the_token_rather_than_its_shape() -> None:
+    """A stub that accepts any bearer teaches the shape of the check without
+    the substance of it — and the real TypeScript route gets written from this
+    file. `persist_turn_token` existing in config made it read as enforced."""
+    from voice_gateway.persistence.stub_server import app as stub
+
+    with TestClient(stub) as client:
+        assert _post(client, None) == 401
+        assert _post(client, "not-the-secret") == 401
+        assert _post(client, Settings().persist_turn_token) == 201
+
+
+# --- where it is allowed to send audio ------------------------------------
+
+
+def test_a_self_hosted_host_is_reachable_by_configuration(monkeypatch) -> None:
+    """ADR-013 points at self-hosted Deepgram in ca-central-1, because there is
+    no Canadian Deepgram region. The builder had an `allow_self_hosted_host`
+    parameter that no call site passed, so that deployment was unreachable —
+    every URL raised. A flag only the tests can set is not a flag."""
+    from voice_gateway import config
+
+    def _self_hosted() -> Settings:
+        return Settings(
+            deepgram_host="deepgram.internal.rehabify.ca",
+            deepgram_allow_self_hosted_host=True,
+        )
+
+    monkeypatch.setattr(config, "get_settings", _self_hosted)
+    url = build_auth_grant_url()
+
+    assert url.startswith("https://deepgram.internal.rehabify.ca/v1/auth/grant")
+    # Inert on a self-hosted deployment, and applied anyway. A guarantee that
+    # holds only while one deployment mode holds is not a guarantee.
+    assert "mip_opt_out=true" in url
+
+
+def test_an_unrecognised_host_still_fails_without_the_flag(monkeypatch) -> None:
+    """Off by default, so a typo in `deepgram_host` fails loudly instead of
+    quietly sending audio somewhere unintended."""
+    from voice_gateway import config
+
+    monkeypatch.setattr(
+        config, "get_settings", lambda: Settings(deepgram_host="api.deepgran.com")
+    )
+
+    with pytest.raises(DeepgramUrlError, match="not a known Deepgram host"):
+        build_auth_grant_url()