diff --git a/CHANGELOG.md b/CHANGELOG.md index 1909d4a..a067f9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Changed + +- `discover_orchestrators` is now asynchronous and must be awaited, including + when given explicit orchestrator addresses. + ### Removed +- Synchronous HTTP helpers `request_json_sync`, `get_json_sync`, and + `post_json_sync`. All HTTP requests now use the asynchronous transport. +- TLS certificate-verification bypasses in HTTP requests, streaming responses, + and trickle transports. HTTPS now verifies certificates and hostnames using + the standard trust configuration; explicit HTTP URLs remain supported. - gRPC orchestrator information and selection including protobuf bindings - LV2V and Scope startup APIs, including 720p pricing - BYOC (bring your own container) inference and training jobs: the diff --git a/examples/echo/README.md b/examples/echo/README.md index e500a1d..68cb6b9 100644 --- a/examples/echo/README.md +++ b/examples/echo/README.md @@ -9,7 +9,7 @@ your `PATH`, then run from the repository root: ```sh uv sync --extra examples LIVEPEER_ORCHESTRATOR_BOOTSTRAP_SECRET=abcdef livepeer orchestrator \ - --config="" --listen 127.0.0.1:8935 --runner-grants 127.0.0.1:8989 + --runner-grants 127.0.0.1:8989 ``` Start the runner in another terminal: diff --git a/examples/ping-pong/README.md b/examples/ping-pong/README.md index 34cbc7a..a649932 100644 --- a/examples/ping-pong/README.md +++ b/examples/ping-pong/README.md @@ -14,7 +14,7 @@ your `PATH`, and start its local orchestrator: ```sh LIVEPEER_ORCHESTRATOR_BOOTSTRAP_SECRET=abcdef livepeer orchestrator \ - --config="" --listen 127.0.0.1:8935 --runner-grants 127.0.0.1:8991 + --runner-grants 127.0.0.1:8991 ``` Start the runner: diff --git a/src/livepeer_gateway/discovery.py b/src/livepeer_gateway/discovery.py index e11352e..06da4b5 100644 --- a/src/livepeer_gateway/discovery.py +++ b/src/livepeer_gateway/discovery.py @@ -8,7 +8,7 @@ from .errors import LivepeerGatewayError from .remote_signer import RemoteSignerError -from .http import _http_origin, _parse_http_url, get_json, get_json_sync +from .http import _http_origin, _parse_http_url, get_json _LOG = logging.getLogger(__name__) @@ -49,7 +49,7 @@ def _append_runner_filters( return _append_query_values(url, values) -def discover_orchestrators( +async def discover_orchestrators( orchestrators: Sequence[str] | str | None = None, *, signer_url: str | None = None, @@ -92,7 +92,7 @@ def discover_orchestrators( try: _LOG.debug("discover_orchestrators running discovery: %s", discovery_endpoint) - data = get_json_sync(discovery_endpoint, headers=request_headers) + data = await get_json(discovery_endpoint, headers=request_headers) except LivepeerGatewayError as e: _LOG.debug("discover_orchestrators discovery failed: %s", e) raise RemoteSignerError( diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 4558c73..964a32e 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -1,11 +1,8 @@ from __future__ import annotations import json -import ssl from typing import Any -from urllib.error import HTTPError, URLError from urllib.parse import ParseResult, urlparse -from urllib.request import Request, urlopen import aiohttp @@ -23,21 +20,6 @@ def _truncate(s: str, max_len: int = 2000) -> str: return s[:max_len] + f"...(+{len(s) - max_len} chars)" -def _http_error_body(e: HTTPError) -> str: - """ - Best-effort read of an HTTPError response body for debugging. - """ - try: - b = e.read() - if not b: - return "" - if isinstance(b, bytes): - return b.decode("utf-8", errors="replace") - return str(b) - except Exception: - return "" - - def _extract_error_message_from_body(body: str) -> str: """ Best-effort extraction of a useful error message from an HTTP error body. @@ -66,13 +48,6 @@ def _extract_error_message_from_body(body: str) -> str: return _truncate(body) -def _extract_error_message(e: HTTPError) -> str: - """ - Best-effort extraction of a useful error message from an HTTPError body. - """ - return _extract_error_message_from_body(_http_error_body(e)) - - def _json_request_parts( url: str, *, @@ -126,103 +101,6 @@ def _ensure_json_object(data: Any, *, url: str) -> dict[str, Any]: return data -def request_json_sync( - url: str, - *, - method: str | None = None, - payload: dict[str, Any] | None = None, - headers: dict[str, str] | None = None, - timeout: float = 5.0, -) -> Any: - """ - Make a JSON HTTP request and parse the JSON response. - - If method is None, defaults to POST when payload is provided, otherwise GET. - - Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. - """ - resolved_method, req_headers, body = _json_request_parts( - url, - method=method, - payload=payload, - headers=headers, - ) - req = Request(url, data=body, headers=req_headers, method=resolved_method) - - # Always ignore HTTPS certificate validation. - ssl_ctx = ssl._create_unverified_context() - - try: - with urlopen(req, timeout=timeout, context=ssl_ctx) as resp: - raw = resp.read().decode("utf-8") - data: Any = json.loads(raw) - except HTTPError as e: - raw_body = _http_error_body(e) - body_text = _extract_error_message_from_body(raw_body) - body_part = f"; body={body_text!r}" if body_text else "" - if e.code == 480: - raise SignerRefreshRequired( - f"Signer returned HTTP 480 (refresh session required) (url={url}){body_part}" - ) from e - if e.code == 482: - raise SkipPaymentCycle( - f"Signer returned HTTP 482 (skip payment cycle) (url={url}){body_part}" - ) from e - raise LivepeerHTTPError( - e.code, - url, - raw_body, - f"HTTP {e.code} from endpoint (url={url}){body_part}", - ) from e - except ConnectionRefusedError as e: - raise LivepeerGatewayError( - f"HTTP JSON error: connection refused (is the server running? is the host/port correct?) (url={url})" - ) from e - except URLError as e: - raise LivepeerGatewayError( - f"HTTP JSON error: failed to reach endpoint: {getattr(e, 'reason', e)} (url={url})" - ) from e - except json.JSONDecodeError as e: - raise LivepeerGatewayError(f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})") from e - except Exception as e: - raise LivepeerGatewayError( - f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})" - ) from e - - return data - - -def post_json_sync( - url: str, - payload: dict[str, Any], - *, - headers: dict[str, str] | None = None, - timeout: float = 5.0, -) -> dict[str, Any]: - """ - POST JSON to `url` and parse a JSON object response. - """ - data = request_json_sync( - url, - payload=payload, - headers=headers, - timeout=timeout, - ) - return _ensure_json_object(data, url=url) - - -def get_json_sync( - url: str, - *, - headers: dict[str, str] | None = None, - timeout: float = 5.0, -) -> Any: - """ - GET JSON from `url` and parse the response. - """ - return request_json_sync(url, headers=headers, timeout=timeout) - - async def _request_body( url: str, *, @@ -250,8 +128,7 @@ async def _request_body( try: client_timeout = aiohttp.ClientTimeout(total=timeout) - connector = aiohttp.TCPConnector(ssl=False) - async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: + async with aiohttp.ClientSession(timeout=client_timeout) as session: async with session.request(resolved_method, url, data=body, headers=req_headers) as resp: raw = await resp.read() content_type = resp.content_type or "" @@ -340,7 +217,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) 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/trickle_publisher.py b/src/livepeer_gateway/trickle_publisher.py index 1419720..7b64ee7 100644 --- a/src/livepeer_gateway/trickle_publisher.py +++ b/src/livepeer_gateway/trickle_publisher.py @@ -164,9 +164,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) - self._session = aiohttp.ClientSession(connector=connector) + self._session = aiohttp.ClientSession() def _stream_url(self, seq: int) -> str: return f"{self.url}/{seq}" diff --git a/src/livepeer_gateway/trickle_subscriber.py b/src/livepeer_gateway/trickle_subscriber.py index 9869c2f..a8add94 100644 --- a/src/livepeer_gateway/trickle_subscriber.py +++ b/src/livepeer_gateway/trickle_subscriber.py @@ -104,8 +104,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) - self._session = aiohttp.ClientSession(connector=connector) + self._session = aiohttp.ClientSession() def _segment_url(self, seq: int) -> str: return f"{self.base_url}/{seq}" diff --git a/tests/fixtures/untrusted-localhost-cert.pem b/tests/fixtures/untrusted-localhost-cert.pem new file mode 100644 index 0000000..29a1e82 --- /dev/null +++ b/tests/fixtures/untrusted-localhost-cert.pem @@ -0,0 +1,19 @@ +-----BEGIN CERTIFICATE----- +MIIDJDCCAgygAwIBAgITViirpxrPGUVZW30UZ0Yeg4A+wzANBgkqhkiG9w0BAQsF +ADAUMRIwEAYDVQQDDAlsb2NhbGhvc3QwHhcNMjYwODE4MjIzMDMyWhcNMzYwODE1 +MjIzMDMyWjAUMRIwEAYDVQQDDAlsb2NhbGhvc3QwggEiMA0GCSqGSIb3DQEBAQUA +A4IBDwAwggEKAoIBAQC80HLfE3k4Lo76h/o1gYuwAwd9yftEyLJTb+pGZibCL9z9 +dRQUgiYecfIRw1S3lLx5Pj4Vsk+gE9UdYG68q6sY2GUP2NwcAn4MBuryp0OG6Q9n +KDPISBr+9q6lzNc060tHwk2j/FRuQf1a3RHMkjj1PWm/uR+6FX2K0MV7LdWJ8hcb +2Bz4S5Gx+aPAYDNuP0IFL6E34F3cbbgHyozMjVez7o64NVezz/aPdz8VtHylmoPJ +1lUjiSGo5d35m23OA1a0LaWF6CXQIGjOfY0uulxB+kIlHEKmeiHW12Z4s2aDtByz +vcuBkLZH9ReASN2XAHFnIUmW+L0Sof/xl3tEf3qNAgMBAAGjbzBtMB0GA1UdDgQW +BBQdtsh+/4E10qiMgarMw0okWgyZ0DAfBgNVHSMEGDAWgBQdtsh+/4E10qiMgarM +w0okWgyZ0DAPBgNVHRMBAf8EBTADAQH/MBoGA1UdEQQTMBGCCWxvY2FsaG9zdIcE +fwAAATANBgkqhkiG9w0BAQsFAAOCAQEAbSsJTV2EDrgpjVJHnXtvGqQv0B8um+NW +3lfqM6f7rp6f0vgo2kKWqcW0iJsvACJmX+5WpD8uINmNpT6+q+1jAZRvytD8ojwa +CrQ6eB8LvVJq0sH4mtr18+6Xt0MigenYFN3UQOBNAWvlEVg17niFZlC5BGtfkDpN +WnpGb7xgLNcjHw+fvi3nwNtD63TWJKKYnfL9nB1RBkHKGwbPts+OW1LUj6m76T8H +oxBav9dE2aD/A4BVjECDM9B1FYIavcIKT5goaOj+iBu8+ukThBnRH6znsY2ox2/p +FE8/5lz7oeRwr/nu49zeRXQbjrpSbw6sjw73dhwlja20YhXhkFHwDQ== +-----END CERTIFICATE----- diff --git a/tests/fixtures/untrusted-localhost-key.pem b/tests/fixtures/untrusted-localhost-key.pem new file mode 100644 index 0000000..5b2b1b6 --- /dev/null +++ b/tests/fixtures/untrusted-localhost-key.pem @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC80HLfE3k4Lo76 +h/o1gYuwAwd9yftEyLJTb+pGZibCL9z9dRQUgiYecfIRw1S3lLx5Pj4Vsk+gE9Ud +YG68q6sY2GUP2NwcAn4MBuryp0OG6Q9nKDPISBr+9q6lzNc060tHwk2j/FRuQf1a +3RHMkjj1PWm/uR+6FX2K0MV7LdWJ8hcb2Bz4S5Gx+aPAYDNuP0IFL6E34F3cbbgH +yozMjVez7o64NVezz/aPdz8VtHylmoPJ1lUjiSGo5d35m23OA1a0LaWF6CXQIGjO +fY0uulxB+kIlHEKmeiHW12Z4s2aDtByzvcuBkLZH9ReASN2XAHFnIUmW+L0Sof/x +l3tEf3qNAgMBAAECggEAQYTdz53syvuQ/suvneCJrfanI1Wy/P/drYlg1LP+0BSW +zVbStHkjLB5jHxnWbGvO6Q2ZsqKw8b3gDnUXiB3T3ee9Tf31Bz0wKxE9dx1ml8qI +30V6jo3C3GWSXhNSPFlHwqtgkL5vN+FrPpwG4KN+406/wtlE8qgha8mdNWzGtDp1 +jyAY+9dUi1kYKRkws39h8Png24Zv7LrxKbIg/14tLXl6KDjU55ygg31fAOSsVYrm +jd3Z0n2hD8dAuI2XOVCGmMi/wrtPRjo3483oPHyqHH3PHehLFV6kMLkbO0atCFUw +38vxvhf8YmItwXTB0KYNuVJeigKuYvSdkwJMotT+6wKBgQD2Hp4DpR+nZjeq/Ezz +A+ZjBAmg2fma+qotj7t287bgpjZYxcmot1V7ykL+HzIy0SaPzEYKDxaJZnxZDiDf +3wKmJMOOLk9lqEGNtD532EJZmw7Mn6JA804Tm+qpNXP5vkhC1u5UjldBlXkiJyaO +lhbZVn+LWrfglMjLiFAZHGBcFwKBgQDEZObXXY7krTxYkllrILhY/2fYYKfysaRQ +uCFWGJs/UZd1B1eBB8ZBHGP42RB70dDwp8lnPx4fOziYtMQ0nzutV8XlkdykJ1X+ +FVAB+hy7dKcZma2esAGgoN1ItaLmK/PmdY7cTTig8yNHuZyRPCMEZFwneOuUbEyA +9QYqJXJQ+wKBgG+r5LZyrbIexxNl9uQs2NQEePnq7T5MjfquRLjsTAYSZIZS9/T9 +emP4mf0D2psEGyMT5JOb1PXy8mjV6ESrw1/vPmu8x3mwpk3NgqqODUzq9gQSSmR+ +AOcUz5yHDGA83jy1mN3019KCJ7w/V1dK8oMapDRurQ/RcmonJ3C9V1txAoGANnSz +/WuXv5i5buXFnYrJEwReywSiDQ1svmUP+UuBYKFDGdQbU0/QjSVOIyYtB+g5w0EC ++3no5EgLfWxCK0UryvjXirxTLcEqevV15dZwORoza0nNdn7fNKbmNM8IiEIPswtN +vZA774T7d1vkKfE+1GzvkJJa4CDXZbqLJwyTEhkCgYEA1DlrzDB1ZPoIbY4EGiXq +XOtvBikS5741LUjymR1XZAiqtp/7gMY7iC6Oolu4UFFRbNQ/JoLR7EUW9LYphwN9 +Gegy5O9aP5XPtxCzW6cZC9RRcNry6F2IHh7Xlj4MQgLPA7INrFMyUor/WKKNtjyJ +DF6rC3VMlrx7wpq/qA2YFDo= +-----END PRIVATE KEY----- diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 213b8d4..cee120b 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -70,6 +70,22 @@ async def scenario(base): assert result.data == {} +def test_streaming_response_returns_raw(): + body = b'{"ok":true}' + + async def handler(request): + return web.Response(body=body, content_type="application/json") + + app = web.Application() + app.router.add_post("/call", handler) + + async def scenario(base): + async with await call_runner(f"{base}/call", stream=True) as result: + return b"".join([chunk async for chunk in result.aiter_bytes()]) + + assert _run(app, scenario) == body + + def test_json_suffix_content_type_is_parsed(): """RFC 6839 ``+json`` types are single JSON documents, so they parse.""" diff --git a/tests/test_discovery.py b/tests/test_discovery.py index e31e1c0..61b584e 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -5,10 +5,59 @@ import pytest -from livepeer_gateway import discovery +from livepeer_gateway import discovery, discover_orchestrators from livepeer_gateway.remote_signer import RemoteSignerError +class TestOrchestratorDiscovery: + @pytest.mark.parametrize( + "orchestrators", + [ + [" https://orch-a.example.com ", "", "http://orch-b.example.com"], + " https://orch-a.example.com, , http://orch-b.example.com ", + ], + ) + async def test_explicit_orchestrators_are_awaited_and_skip_discovery( + self, orchestrators: list[str] | str + ) -> None: + with mock.patch.object(discovery, "get_json") as request: + result = await discover_orchestrators( + orchestrators, discovery_url="https://discovery.example.com" + ) + + assert result == ["https://orch-a.example.com", "http://orch-b.example.com"] + request.assert_not_called() + + @pytest.mark.parametrize("discovery_url", [None, "https://discovery.example.com/base"]) + async def test_discovers_addresses_with_selected_endpoint_and_headers( + self, discovery_url: str | None + ) -> None: + signer_headers = {"Authorization": "signer-token"} + discovery_headers = {"Authorization": "discovery-token"} + with mock.patch.object( + discovery, + "get_json", + return_value=[ + {"address": " https://orch.example.com "}, + {"address": " "}, + {}, + "invalid", + ], + ) as request: + result = await discover_orchestrators( + signer_url="https://signer.example.com/base", + signer_headers=signer_headers, + discovery_url=discovery_url, + discovery_headers=discovery_headers, + ) + + assert result == ["https://orch.example.com"] + request.assert_awaited_once_with( + discovery_url or "https://signer.example.com/discover-orchestrators", + headers=discovery_headers if discovery_url else signer_headers, + ) + + class TestRunnerDiscoveryQuery: def test_append_runner_filters_preserves_queries_and_ignores_empty_values( self, diff --git a/tests/test_tls.py b/tests/test_tls.py new file mode 100644 index 0000000..a0bbae4 --- /dev/null +++ b/tests/test_tls.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import ssl +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path + +import aiohttp +import pytest +from aiohttp import web + +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.live_runner import call_runner +from livepeer_gateway.trickle_publisher import TricklePublisher +from livepeer_gateway.trickle_subscriber import TrickleSubscriber + + +_FIXTURES = Path(__file__).with_name("fixtures") + + +@asynccontextmanager +async def _untrusted_tls_server() -> AsyncIterator[str]: + runner = web.AppRunner(web.Application()) + await runner.setup() + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ssl_context.load_cert_chain( + _FIXTURES / "untrusted-localhost-cert.pem", + _FIXTURES / "untrusted-localhost-key.pem", + ) + site = web.TCPSite(runner, "127.0.0.1", 0, ssl_context=ssl_context) + await site.start() + server = site._server + assert server is not None + port = server.sockets[0].getsockname()[1] + try: + yield f"https://127.0.0.1:{port}" + finally: + await runner.cleanup() + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_call_runner_rejects_untrusted_certificate(stream: bool) -> None: + async with _untrusted_tls_server() as base_url: + with pytest.raises(LivepeerGatewayError) as raised: + result = await call_runner(f"{base_url}/runner", stream=stream) + if stream: + await result.aclose() + + assert isinstance(raised.value.__cause__, aiohttp.ClientConnectorCertificateError) + + +async def test_trickle_publisher_rejects_untrusted_certificate() -> None: + async with _untrusted_tls_server() as base_url: + async with TricklePublisher( + f"{base_url}/trickle", "application/octet-stream" + ) as publisher: + with pytest.raises(aiohttp.ClientConnectorCertificateError): + await publisher.create() + + +async def test_trickle_subscriber_rejects_untrusted_certificate( + caplog: pytest.LogCaptureFixture, +) -> None: + async with _untrusted_tls_server() as base_url: + async with TrickleSubscriber( + f"{base_url}/trickle", start_seq=0, max_retries=1 + ) as subscriber: + assert await subscriber.next() is None + + assert any( + record.exc_info + and isinstance(record.exc_info[1], aiohttp.ClientConnectorCertificateError) + for record in caplog.records + )