Harden remote diarization adapter

app/services/diarization_service.py

@@ -5,20 +5,17 @@ import math
 import os
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlparse
 
 import httpx
 from pydantic import ValidationError
 
-from app.core.config import Settings
 from app.core.exceptions import InputError, ProcessingError
 from app.models.transcript import DiarizationSpan
 
 
 class DiarizationService:
-    """Optional provider-neutral HTTP adapter for speaker diarization."""
-
-    def __init__(self, settings: Settings) -> None:
-        self.settings = settings
+    """Optional authenticated HTTP adapter for speaker diarization."""
 
     @property
     def url(self) -> str:
@@ -29,10 +26,7 @@ class DiarizationService:
         return bool(self.url)
 
     async def diarize(self, path: Path, mime_type: str, filename: str) -> list[dict[str, Any]]:
-        if not self.is_configured:
-            raise ProcessingError(
-                "Diarization is enabled but no DIARIZATION_URL is configured"
-            )
+        url = self._validated_url()
         if not path.is_file():
             raise ProcessingError("Diarization input file is unavailable")
 
@@ -52,10 +46,13 @@ class DiarizationService:
         token = os.getenv("DIARIZATION_API_KEY", "").strip()
         if token:
             headers["Authorization"] = f"Bearer {token}"
+        elif self._bool_env("DIARIZATION_REQUIRE_AUTH", True):
+            raise ProcessingError("Diarization backend authentication is not configured")
 
         connect_timeout = self._float_env("DIARIZATION_CONNECT_TIMEOUT_SECONDS", 10.0, 0.1)
         timeout_seconds = self._float_env("DIARIZATION_TIMEOUT_SECONDS", 300.0, 1.0)
         max_response = self._int_env("DIARIZATION_MAX_RESPONSE_BYTES", 1_048_576, 1024)
+        max_spans = self._int_env("DIARIZATION_MAX_SPANS", 10_000, 1)
 
         timeout = httpx.Timeout(
             connect=connect_timeout,
@@ -67,18 +64,12 @@ class DiarizationService:
         try:
             async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client:
                 with path.open("rb") as handle:
-                    response = await client.post(
-                        self.url,
-                        content=handle,
-                        headers=headers,
-                    )
+                    response = await client.post(url, content=handle, headers=headers)
         except (OSError, httpx.HTTPError) as exc:
             raise ProcessingError("Diarization backend request failed") from exc
 
         if response.status_code >= 400:
-            raise ProcessingError(
-                f"Diarization backend returned HTTP {response.status_code}"
-            )
+            raise ProcessingError(f"Diarization backend returned HTTP {response.status_code}")
         if len(response.content) > max_response:
             raise ProcessingError("Diarization backend response exceeds the configured limit")
 
@@ -91,6 +82,8 @@ class DiarizationService:
             payload = payload.get("spans", payload.get("segments"))
         if not isinstance(payload, list):
             raise ProcessingError("Diarization backend must return a span array")
+        if len(payload) > max_spans:
+            raise ProcessingError("Diarization backend returned too many spans")
 
         normalized: list[dict[str, Any]] = []
         previous_start = -1.0
@@ -110,6 +103,24 @@ class DiarizationService:
 
         return normalized
 
+    def _validated_url(self) -> str:
+        url = self.url
+        if not url:
+            raise ProcessingError(
+                "Diarization is enabled but no DIARIZATION_URL is configured"
+            )
+        parsed = urlparse(url)
+        if parsed.scheme == "https" and parsed.hostname:
+            return url
+        environment = os.getenv("APP_ENVIRONMENT", "development").strip().lower()
+        if (
+            environment in {"development", "test"}
+            and parsed.scheme == "http"
+            and parsed.hostname in {"localhost", "127.0.0.1", "::1"}
+        ):
+            return url
+        raise ProcessingError("DIARIZATION_URL must use HTTPS")
+
     @staticmethod
     def _int_env(name: str, default: int, minimum: int) -> int:
         raw = os.getenv(name, "")
@@ -135,3 +146,10 @@ class DiarizationService:
         if not math.isfinite(value) or value < minimum:
             raise ProcessingError(f"{name} is below its minimum allowed value")
         return value
+
+    @staticmethod
+    def _bool_env(name: str, default: bool) -> bool:
+        raw = os.getenv(name, "")
+        if not raw:
+            return default
+        return raw.strip().lower() in {"1", "true", "yes", "on"}