Verify TLS by default with an opt-out

README

@@ -173,3 +173,20 @@ Selection/discovery precedence (highest -> lowest):
 5) signer-derived discovery endpoint
 
 `signer_headers` are sent with requests to the signer service. `discovery_headers` are only used when an explicit `discovery_url` is provided (and not when using the signer service as a discovery fallback).
+
+## TLS certificate verification
+
+By default the SDK verifies HTTPS certificates against the system trust store.
+Orchestrators serve self-signed certificates, so to reach a stock orchestrator
+(and the signer, discovery, runner and trickle endpoints behind it) turn
+verification off for the process:
+
+```bash
+export LIVEPEER_GATEWAY_VERIFY_TLS=0   # "0", "false", "no" or "off"
+```
+
+or at runtime with `livepeer_gateway.http.VERIFY_TLS = False`.
+
+The setting is process-wide and all-or-nothing: there is no per-endpoint or
+per-call override, so leave it on only when every HTTPS endpoint the SDK reaches
+presents a publicly trusted certificate.

src/livepeer_gateway/http.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import functools
 import json
+import os
 import ssl
 from typing import Any
 from urllib.error import HTTPError, URLError
@@ -18,6 +20,39 @@
 
 _REFRESH_SESSION_ORCHESTRATOR_URL_HEADER = "Livepeer-Orchestrator-URL"
 
+VERIFY_TLS_ENV = "LIVEPEER_GATEWAY_VERIFY_TLS"
+
+
+def _verify_tls_from_env() -> bool:
+    """``LIVEPEER_GATEWAY_VERIFY_TLS``: ``0``, ``false``, ``no`` or ``off`` turn verification off."""
+    return os.environ.get(VERIFY_TLS_ENV, "").strip().lower() not in {"0", "false", "no", "off"}
+
+
+# HTTPS certificates are verified against the system trust store by default. Set
+# LIVEPEER_GATEWAY_VERIFY_TLS=0, or assign VERIFY_TLS, to turn verification off
+# for the process when any HTTPS endpoint the SDK calls (signer, discovery,
+# orchestrators, runners, trickle) serves a self-signed certificate, as stock
+# orchestrators do.
+VERIFY_TLS: bool = _verify_tls_from_env()
+
+
+@functools.cache
+def _ssl_context_for(verify: bool) -> ssl.SSLContext:
+    ctx = ssl.create_default_context()
+    if not verify:
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+    return ctx
+
+
+def _ssl_context() -> ssl.SSLContext:
+    """TLS context for an outgoing HTTPS request, following ``VERIFY_TLS`` at call time.
+
+    ``VERIFY_TLS`` is process-wide and all-or-nothing: it governs every HTTPS
+    request the SDK makes, with no per-endpoint or per-call override.
+    """
+    return _ssl_context_for(VERIFY_TLS)
+
 
 def _truncate(s: str, max_len: int = 2000) -> str:
     if len(s) <= max_len:
@@ -160,9 +195,7 @@ def request_json_sync(
         headers=headers,
     )
     req = Request(url, data=body, headers=req_headers, method=resolved_method)
-
-    # Always ignore HTTPS certificate validation (matches our gRPC behavior).
-    ssl_ctx = ssl._create_unverified_context()
+    ssl_ctx = _ssl_context()
 
     try:
         with urlopen(req, timeout=timeout, context=ssl_ctx) as resp:
@@ -266,7 +299,7 @@ async def _request_body(
 
     try:
         client_timeout = aiohttp.ClientTimeout(total=timeout)
-        connector = aiohttp.TCPConnector(ssl=False)
+        connector = aiohttp.TCPConnector(ssl=_ssl_context())
         async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session:
             async with session.request(resolved_method, url, data=body, headers=req_headers) as resp:
                 raw = await resp.read()
@@ -356,7 +389,7 @@ async def open_stream(
     )
 
     timeout = aiohttp.ClientTimeout(total=None, sock_connect=connect_timeout, sock_read=None)
-    session = aiohttp.ClientSession(timeout=timeout, connector=aiohttp.TCPConnector(ssl=False))
+    session = aiohttp.ClientSession(timeout=timeout, connector=aiohttp.TCPConnector(ssl=_ssl_context()))
     try:
         resp = await session.request(resolved_method, url, data=body, headers=req_headers)
     except (TimeoutError, aiohttp.ClientError) as e:

src/livepeer_gateway/remote_signer.py

@@ -5,7 +5,6 @@
 import json
 import logging
 import re
-import ssl
 from dataclasses import dataclass, replace
 from functools import lru_cache
 from typing import Any, Optional
@@ -488,7 +487,7 @@ def send_payment(self) -> None:
         Generate a payment (via get_payment) and forward it
         to the orchestrator via POST {orch}/payment.
         """
-        from .http import _extract_error_message, _http_origin
+        from .http import _extract_error_message, _http_origin, _ssl_context
 
         p = self.get_payment()
         if not self._info.transcoder:
@@ -500,7 +499,7 @@ def send_payment(self) -> None:
             "Livepeer-Segment": p.seg_creds or "",
         }
         req = Request(url, data=b"", headers=headers, method="POST")
-        ssl_ctx = ssl._create_unverified_context()
+        ssl_ctx = _ssl_context()
         try:
             with urlopen(req, timeout=5.0, context=ssl_ctx) as resp:
                 resp.read()

src/livepeer_gateway/trickle_publisher.py

@@ -9,6 +9,7 @@
 import aiohttp
 
 from .errors import LivepeerGatewayError
+from .http import _ssl_context
 
 
 _LOG = logging.getLogger(__name__)
@@ -164,8 +165,7 @@ async def _ensure_runtime(self) -> None:
         if self._lock is None:
             self._lock = asyncio.Lock()
         if self._session is None:
-            # Ignore TLS validation (matches the rest of this repo).
-            connector = aiohttp.TCPConnector(ssl=False)
+            connector = aiohttp.TCPConnector(ssl=_ssl_context())
             self._session = aiohttp.ClientSession(connector=connector)
 
     def _stream_url(self, seq: int) -> str:

src/livepeer_gateway/trickle_subscriber.py

@@ -8,6 +8,7 @@
 
 import aiohttp
 
+from .http import _ssl_context
 from .segment_reader import SegmentReader
 
 
@@ -104,7 +105,7 @@ async def _ensure_runtime(self) -> None:
         if self._lock is None:
             self._lock = asyncio.Lock()
         if self._session is None:
-            connector = aiohttp.TCPConnector(ssl=False)
+            connector = aiohttp.TCPConnector(ssl=_ssl_context())
             self._session = aiohttp.ClientSession(connector=connector)
 
     def _segment_url(self, seq: int) -> str:

tests/test_tls.py

@@ -0,0 +1,205 @@
+"""Opt-out TLS certificate verification.
+
+The tests serve a real aiohttp application over a self-signed certificate
+generated for the test. By default the SDK verifies certificates, so every HTTP
+path refuses it before sending a request; with ``VERIFY_TLS`` off, the SDK
+accepts it, as it accepts the self-signed certificates orchestrators serve.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import pathlib
+import ssl
+import subprocess
+import tempfile
+from collections.abc import AsyncIterator, Iterator
+
+import pytest
+from aiohttp import web
+
+from livepeer_gateway import http
+from livepeer_gateway.discovery import discover_runners
+from livepeer_gateway.errors import LivepeerGatewayError
+from livepeer_gateway.live_runner import call_runner
+from livepeer_gateway.remote_signer import RemoteSignerError, get_signer_info
+
+
+def _write_self_signed(directory: pathlib.Path) -> tuple[pathlib.Path, pathlib.Path]:
+    """Write a throwaway self-signed cert and key for ``127.0.0.1``."""
+    cert = directory / "selfsigned.crt"
+    key = directory / "selfsigned.key"
+    subprocess.run(
+        [
+            "openssl",
+            "req",
+            "-x509",
+            "-newkey",
+            "ec",
+            "-pkeyopt",
+            "ec_paramgen_curve:prime256v1",
+            "-keyout",
+            str(key),
+            "-out",
+            str(cert),
+            "-days",
+            "1",
+            "-nodes",
+            "-subj",
+            "/CN=127.0.0.1",
+            "-addext",
+            "subjectAltName=IP:127.0.0.1,DNS:localhost",
+        ],
+        check=True,
+        capture_output=True,
+    )
+    return cert, key
+
+
+class TestVerifyTlsSetting:
+    @pytest.mark.parametrize("value", ["0", "false", "no", "off", " FALSE ", "Off"])
+    def test_env_turns_verification_off(self, monkeypatch: pytest.MonkeyPatch, value: str) -> None:
+        monkeypatch.setenv(http.VERIFY_TLS_ENV, value)
+        assert http._verify_tls_from_env() is False
+
+    @pytest.mark.parametrize("value", [None, "", "1", "true", "yes", "anything"])
+    def test_anything_else_leaves_it_on(
+        self, monkeypatch: pytest.MonkeyPatch, value: str | None
+    ) -> None:
+        if value is None:
+            monkeypatch.delenv(http.VERIFY_TLS_ENV, raising=False)
+        else:
+            monkeypatch.setenv(http.VERIFY_TLS_ENV, value)
+        assert http._verify_tls_from_env() is True
+
+    def test_context_follows_the_setting_at_call_time(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(http, "VERIFY_TLS", False)
+        assert http._ssl_context().verify_mode == ssl.CERT_NONE
+        assert http._ssl_context().check_hostname is False
+        monkeypatch.setattr(http, "VERIFY_TLS", True)
+        assert http._ssl_context().verify_mode == ssl.CERT_REQUIRED
+        assert http._ssl_context().check_hostname is True
+
+
+@pytest.fixture
+def verify_off(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    monkeypatch.setattr(http, "VERIFY_TLS", False)
+    get_signer_info.cache_clear()  # type: ignore[attr-defined]
+    yield
+    get_signer_info.cache_clear()  # type: ignore[attr-defined]
+
+
+@pytest.fixture
+def verify_on(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    monkeypatch.setattr(http, "VERIFY_TLS", True)
+    get_signer_info.cache_clear()  # type: ignore[attr-defined]
+    yield
+    get_signer_info.cache_clear()  # type: ignore[attr-defined]
+
+
+@contextlib.asynccontextmanager
+async def _serve_tls(app: web.Application) -> AsyncIterator[str]:
+    """Serve ``app`` over a generated self-signed certificate on an ephemeral port."""
+    with tempfile.TemporaryDirectory() as directory:
+        cert, key = _write_self_signed(pathlib.Path(directory))
+        ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ssl_ctx.load_cert_chain(cert, key)
+        runner = web.AppRunner(app)
+        await runner.setup()
+        site = web.TCPSite(runner, "127.0.0.1", 0, ssl_context=ssl_ctx)
+        await site.start()
+        port = site._server.sockets[0].getsockname()[1]  # type: ignore[union-attr]
+        try:
+            yield f"https://127.0.0.1:{port}"
+        finally:
+            await runner.cleanup()
+
+
+def _app(calls: list[str]) -> web.Application:
+    """A runner, signer and discovery endpoint in one."""
+
+    async def call(request: web.Request) -> web.Response:
+        calls.append(request.path)
+        return web.json_response({"text": "hello"})
+
+    async def sse(request: web.Request) -> web.StreamResponse:
+        calls.append(request.path)
+        resp = web.StreamResponse(headers={"Content-Type": "text/event-stream"})
+        await resp.prepare(request)
+        await resp.write(b"data: one\n\n")
+        await resp.write_eof()
+        return resp
+
+    async def sign(request: web.Request) -> web.Response:
+        calls.append(request.path)
+        return web.json_response({"address": "0xpayer", "signature": "0xsig"})
+
+    async def discover(request: web.Request) -> web.Response:
+        calls.append(request.path)
+        return web.json_response(
+            [{"address": "https://orch.example.com", "runners": [{"url": "https://orch.example.com/r", "app": "a"}]}]
+        )
+
+    app = web.Application()
+    app.router.add_post("/call", call)
+    app.router.add_post("/sse", sse)
+    app.router.add_post("/sign-orchestrator-info", sign)
+    app.router.add_get("/discover-orchestrators", discover)
+    return app
+
+
+class TestVerifyOffAcceptsSelfSigned:
+    async def test_runner_signer_and_discovery(self, verify_off: None) -> None:
+        calls: list[str] = []
+        async with _serve_tls(_app(calls)) as base:
+            result = await call_runner(f"{base}/call", payload={"x": 1})
+            async with await call_runner(f"{base}/sse", payload={"x": 1}, stream=True) as stream:
+                lines = [line async for line in stream.aiter_lines() if line]
+            signer = await get_signer_info(base)
+            entries = await discover_runners(signer_url=base)
+        assert result.data == {"text": "hello"}
+        assert lines == ["data: one"]
+        assert signer.address == "0xpayer"
+        assert entries[0]["runners"][0]["app"] == "a"
+        assert calls == ["/call", "/sse", "/sign-orchestrator-info", "/discover-orchestrators"]
+
+
+class TestDefaultRejectsSelfSigned:
+    async def test_call_runner(self, verify_on: None) -> None:
+        calls: list[str] = []
+        async with _serve_tls(_app(calls)) as base:
+            with pytest.raises(LivepeerGatewayError, match="certificate"):
+                await call_runner(f"{base}/call", payload={"x": 1})
+            with pytest.raises(LivepeerGatewayError, match="certificate"):
+                await call_runner(f"{base}/sse", payload={"x": 1}, stream=True)
+        assert calls == []
+
+    async def test_signer(self, verify_on: None) -> None:
+        calls: list[str] = []
+        async with _serve_tls(_app(calls)) as base:
+            with pytest.raises(LivepeerGatewayError, match="certificate"):
+                await get_signer_info(base)
+        assert calls == []
+
+    async def test_discovery(self, verify_on: None) -> None:
+        calls: list[str] = []
+        async with _serve_tls(_app(calls)) as base:
+            with pytest.raises(RemoteSignerError, match="certificate"):
+                await discover_runners(signer_url=base)
+        assert calls == []
+
+    async def test_sync_request(self, verify_on: None) -> None:
+        calls: list[str] = []
+
+        def probe(base: str) -> str:
+            try:
+                http.post_json_sync(f"{base}/sign-orchestrator-info", {})
+            except LivepeerGatewayError as e:
+                return str(e)
+            return ""
+
+        async with _serve_tls(_app(calls)) as base:
+            message = await asyncio.to_thread(probe, base)
+        assert "certificate" in message
+        assert calls == []