fix(security): close worker transport disclosure flows

backend/api/routers/audiobook.py

@@ -1114,6 +1114,25 @@ async def _render_longform_sse(
         yield _emit({"type": "error", "error": "render failed (see backend log)"})
 
 
+async def _public_longform_stream(stream):
+    """Keep generator diagnostics local if setup fails before its own guard."""
+    try:
+        async for event in stream:
+            yield event
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:
+        from core.public_errors import public_failure
+
+        error = public_failure(
+            logger,
+            "Longform response stream failed",
+            exc,
+            response="Render failed; check the backend log for details.",
+        )
+        yield f"data: {json.dumps({'type': 'error', 'error': error})}\n\n"
+
+
 @router.post("/audiobook")
 async def audiobook_synthesize(req: AudiobookRequest, request: Request = None):
     """Synthesize a chapterized audiobook from a script, streaming SSE progress."""
@@ -1122,14 +1141,14 @@ async def audiobook_synthesize(req: AudiobookRequest, request: Request = None):
     # to a direct in-process call, e.g. a unit test); its disconnect poll is what
     # lets Stop cancel the render mid-book (#1216).
     return StreamingResponse(
-        _render_longform_sse(
+        _public_longform_stream(_render_longform_sse(
             plan, default_voice=req.default_voice, language=req.language,
             fmt=req.format, bitrate=req.bitrate,
             loudness=req.loudness, cover_path=req.cover_path, metadata=req.metadata,
             lexicon=req.lexicon, opts=_expressive_opts(req), voice_map=req.voice_map,
             job_type="audiobook",
             is_disconnected=request.is_disconnected if request is not None else None,
-        ),
+        )),
         media_type="text/event-stream",
     )
 
@@ -1183,14 +1202,14 @@ async def longform_render(req: LongformRenderRequest, request: Request = None):
             chapters.append(Chapter(title=c.title or f"Chapter {i + 1}", spans=spans))
     plan = AudiobookPlan(chapters=chapters)
     return StreamingResponse(
-        _render_longform_sse(
+        _public_longform_stream(_render_longform_sse(
             plan, default_voice=req.default_voice, language=req.language,
             fmt=req.format, bitrate=req.bitrate,
             loudness=req.loudness, cover_path=req.cover_path, metadata=req.metadata,
             lexicon=req.lexicon, opts=_expressive_opts(req), voice_map=req.voice_map,
             job_type="story",
             is_disconnected=request.is_disconnected if request is not None else None,
-        ),
+        )),
         media_type="text/event-stream",
     )
 
@@ -1276,7 +1295,7 @@ async def resume_longform(job_id: str, request: Request = None):
     # unrendered ones synthesize. Using a fresh id means the request's job_id
     # never names a work dir / output file (defence-in-depth path-injection).
     return StreamingResponse(
-        _render_longform_sse(
+        _public_longform_stream(_render_longform_sse(
             plan, default_voice=p.get("default_voice"), language=p.get("language"),
             fmt=p.get("fmt", "m4b"), bitrate=p.get("bitrate", "128k"),
             loudness=p.get("loudness"), cover_path=p.get("cover_path"),
@@ -1285,6 +1304,6 @@ async def resume_longform(job_id: str, request: Request = None):
             voice_map=p.get("voice_map"),
             job_type=entry["job_type"],
             is_disconnected=request.is_disconnected if request is not None else None,
-        ),
+        )),
         media_type="text/event-stream",
     )

backend/api/routers/workers.py

@@ -401,7 +401,18 @@ async def set_inbound_enabled(request: InboundEnableRequest) -> dict:
     if inbound_service.enabled():
         await inbound_service.node.start()
         if inbound_service.node.startup_error:
-            raise HTTPException(status_code=409, detail=inbound_service.node.startup_error)
+            from core.public_errors import public_failure
+
+            detail = public_failure(
+                logger,
+                "Inbound worker listener failed to start",
+                inbound_service.node.startup_error,
+                response=(
+                    "The inbound worker listener could not start; "
+                    "check the backend log for details."
+                ),
+            )
+            raise HTTPException(status_code=409, detail=detail)
     else:
         await inbound_service.node.stop()
     return inbound_service.node.snapshot()

backend/worker/agent.py

@@ -98,9 +98,9 @@ def fetch_server_certificate(endpoint: str, *, timeout: float = 10.0) -> bytes:
     host, _, port = endpoint.rpartition(":")
     if not host:
         raise ValueError(f"Endpoint must be host:port — got {endpoint!r}")
-    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-    context.check_hostname = False
-    context.verify_mode = ssl.CERT_NONE
+    from worker import tls  # noqa: PLC0415
+
+    context = tls.unverified_client_context()
     with ssl.create_connection((host, int(port)), timeout=timeout) as raw:
         with context.wrap_socket(raw, server_hostname=host) as tls:
             der = tls.getpeercert(binary_form=True)

backend/worker/capabilities.py

@@ -139,7 +139,7 @@ def repo_ids_for(entry: dict) -> list[str]:
                 return []
             return [spec.weights_repo_id]
     except Exception:
-        logger.debug("Sidecar repository probe failed for %s", engine_id, exc_info=True)
+        logger.debug("Sidecar repository probe failed", exc_info=True)
     return []
 
 

backend/worker/inbound/connector.py

@@ -47,9 +47,7 @@ def _fetch_pinned_certificate(
     a self-signed certificate. The copied fingerprint is the trust anchor; the
     verified leaf is then the sole root trusted by the real gRPC channel.
     """
-    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-    context.check_hostname = False
-    context.verify_mode = ssl.CERT_NONE
+    context = tls.unverified_client_context()
     with socket.create_connection(
         (connection.host, connection.port), timeout=timeout
     ) as raw, context.wrap_socket(raw, server_hostname=connection.host) as secured:
@@ -109,12 +107,12 @@ class NodeConnection:
                 raise
             except Exception as exc:
                 attempt += 1
-                self._last_error = str(exc)
+                self._last_error = "Connection failed; check the backend log for details."
                 delay = backoff_delay(attempt)
                 logger.warning(
-                    "Connection to %s failed (%s). Retrying in %.1fs.",
+                    "Connection to %s failed (class=%s; details withheld). Retrying in %.1fs.",
                     self._connection.redacted(),
-                    exc,
+                    type(exc).__name__,
                     delay,
                 )
                 with contextlib.suppress(asyncio.TimeoutError):

backend/worker/registry.py

@@ -340,7 +340,7 @@ def revoke(worker_id: str, *, now: Optional[float] = None) -> bool:
             (stamp, worker_id),
         )
     if cur.rowcount:
-        logger.info("Revoked remote worker %s", worker_id)
+        logger.info("Remote worker revoked")
     return bool(cur.rowcount)
 
 

backend/worker/tls.py

@@ -24,6 +24,7 @@ import ipaddress
 import logging
 import os
 import socket
+import ssl
 from dataclasses import dataclass
 from typing import Optional
 
@@ -40,6 +41,15 @@ _CERT_VALID_DAYS = 825  # the CA/Browser Forum maximum; long enough to be quiet
 _RENEW_WITHIN_DAYS = 30
 
 
+def unverified_client_context() -> ssl.SSLContext:
+    """Build the pin-bootstrap context without permitting legacy TLS."""
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+    return context
+
+
 @dataclass(frozen=True)
 class ServerCredentials:
     """A control plane's certificate and its pinnable fingerprint."""

tests/test_worker_security_boundaries.py

@@ -0,0 +1,92 @@
+"""Security regressions shared by inbound and outbound worker transports."""
+
+from __future__ import annotations
+
+import logging
+import ssl
+
+import pytest
+
+
+def test_pin_bootstrap_rejects_legacy_tls():
+    from worker import tls
+
+    context = tls.unverified_client_context()
+
+    assert context.minimum_version >= ssl.TLSVersion.TLSv1_2
+    assert context.verify_mode == ssl.CERT_NONE
+
+
+@pytest.mark.asyncio
+async def test_connector_never_logs_or_exposes_private_failures(monkeypatch, caplog):
+    from worker.inbound.connection_string import Connection
+    from worker.inbound.connector import NodeConnection
+
+    private = "ovnode_private-secret"
+    connection = NodeConnection(
+        object(),
+        Connection(
+            host="127.0.0.1",
+            port=7444,
+            secret="ovnode_" + "s" * 40,
+            fingerprint="a" * 64,
+        ),
+    )
+
+    async def fail_once():
+        connection._stop.set()
+        raise RuntimeError(private)
+
+    monkeypatch.setattr(connection, "_connect_once", fail_once)
+    with caplog.at_level(logging.WARNING):
+        await connection.run_forever()
+
+    assert private not in caplog.text
+    assert private not in connection.last_error
+
+
+@pytest.mark.asyncio
+async def test_longform_stream_returns_only_a_fixed_public_failure():
+    from api.routers.audiobook import _public_longform_stream
+
+    private = "trace /home/alice ovnode_private-secret"
+
+    async def broken_stream():
+        raise RuntimeError(private)
+        yield "unreachable"
+
+    frames = [frame async for frame in _public_longform_stream(broken_stream())]
+
+    assert len(frames) == 1
+    assert private not in frames[0]
+    assert "Render failed" in frames[0]
+
+
+@pytest.mark.asyncio
+async def test_inbound_startup_error_is_not_returned_to_the_api(monkeypatch):
+    from api.routers import workers
+    from worker.inbound import service as inbound_service
+
+    private = "bind failed at /home/alice with ovnode_private-secret"
+
+    class Node:
+        startup_error = private
+
+        async def start(self):
+            return None
+
+        def snapshot(self):
+            return {}
+
+    monkeypatch.setattr(inbound_service, "enabled_override", lambda: None)
+    monkeypatch.setattr(inbound_service, "set_enabled", lambda _value: None)
+    monkeypatch.setattr(inbound_service, "enabled", lambda: True)
+    monkeypatch.setattr(inbound_service, "node", Node())
+
+    with pytest.raises(workers.HTTPException) as exc:
+        await workers.set_inbound_enabled(
+            workers.InboundEnableRequest(enabled=True, bind="", port=0)
+        )
+
+    assert exc.value.status_code == 409
+    assert private not in exc.value.detail