diff --git a/README b/README index af5018d..9145f27 100644 --- a/README +++ b/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. diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 699d042..f94d227 100644 --- a/src/livepeer_gateway/http.py +++ b/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: diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index dd4554f..05f6b64 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/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() diff --git a/src/livepeer_gateway/trickle_publisher.py b/src/livepeer_gateway/trickle_publisher.py index 1419720..17617ed 100644 --- a/src/livepeer_gateway/trickle_publisher.py +++ b/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: diff --git a/src/livepeer_gateway/trickle_subscriber.py b/src/livepeer_gateway/trickle_subscriber.py index 9869c2f..ff48f4d 100644 --- a/src/livepeer_gateway/trickle_subscriber.py +++ b/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: diff --git a/tests/test_tls.py b/tests/test_tls.py new file mode 100644 index 0000000..97d9ada --- /dev/null +++ b/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 == []