From d99cb7564474af974ce4f09e02a05467df30c34e Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 10 Oct 2026 22:31:03 +0200 Subject: [PATCH] feat: discover and authenticate the Engine HTTP API with mTLS --- README.md | 23 ++++ src/rstream/client.py | 6 + src/rstream/engine_api.py | 11 +- src/rstream/engine_discovery.py | 146 ++++++++++++++++++++ tests/e2e/test_real_engine.py | 16 ++- tests/integration/test_engine_mtls.py | 191 ++++++++++++++++++++++++++ 6 files changed, 385 insertions(+), 8 deletions(-) create mode 100644 src/rstream/engine_discovery.py create mode 100644 tests/integration/test_engine_mtls.py diff --git a/README.md b/README.md index 8aa999d..b4fe3fa 100644 --- a/README.md +++ b/README.md @@ -287,3 +287,26 @@ the repository. ## License Apache-2.0. See [LICENSE](LICENSE). + +## Engine HTTP access with certificates + +The existing `list_clients`, `list_tunnels` and `watch` methods support enrolled +file-based mTLS identities. Keep one `Client` in an `async with` block. It discovers +the dedicated project API through anonymous, verified HTTPS and caches the address +in memory, normally for one hour. Concurrent callers share discovery; cancellation +of one waiter does not cancel others, and closing the client cancels pending +lookup work. No YAML field, persistent cache or new login is required. + +Discovery never sends the certificate, a token, cookies or Control plane headers, +refuses redirects and verifies the server even when runtime TLS is marked insecure. +The ordinary Engine hostname never requests a certificate. Runtime requests use +the discovered authority with the existing identity and project permissions. +Both SSE and WebSocket certificate watches use TLS/TCP. Python does not implement +HTTP/3 or a TURN generation helper; use Go/JavaScript or the documented Engine API +for TURN issuance. Control plane administration and region selectors still require +a token; provision an explicit Engine address for a certificate-only device. + +Tests in `tests/integration/test_engine_mtls.py` use real TLS servers for discovery, +inventory, both watch transports, trust validation and cancellation. Opt-in +real-Engine inventory checks use `RSTREAM_PYTHON_E2E_ENGINE_API=1` with the +existing E2E credential configuration. diff --git a/src/rstream/client.py b/src/rstream/client.py index 9346e4a..e39d298 100644 --- a/src/rstream/client.py +++ b/src/rstream/client.py @@ -30,6 +30,7 @@ watch_events, watch_params_json, ) +from rstream.engine_discovery import EngineAPIDiscovery, has_client_certificate from rstream.errors import ConfigurationError, ProtocolError, RuntimeError from rstream.protocol import ( engine_error_from_pb, @@ -119,6 +120,7 @@ def __init__( self._controls: set[ControlChannel] = set() self._control_watchers: set[asyncio.Task[None]] = set() self._closed = False + self._engine_discovery = EngineAPIDiscovery() @classmethod def from_env( @@ -380,15 +382,19 @@ async def generate() -> AsyncIterator[Event]: return EventStream(generate()) async def _engine_api_target(self) -> tuple[str, str | None, TLSOptions | None]: + self._ensure_open() resolved = await self._get_resolved() engine = await self._resolve_engine(resolved) token = await self._resolve_token(resolved, engine) + if has_client_certificate(resolved.tls): + engine = await self._engine_discovery.resolve(engine, resolved.tls) return engine, token, resolved.tls async def close(self) -> None: if self._closed: return self._closed = True + await self._engine_discovery.close() failure: BaseException | None = None for control in tuple(self._controls): try: diff --git a/src/rstream/engine_api.py b/src/rstream/engine_api.py index 2d01570..7508761 100644 --- a/src/rstream/engine_api.py +++ b/src/rstream/engine_api.py @@ -14,7 +14,7 @@ from datetime import datetime from typing import Any, Literal -from rstream.config import TLSOptions +from rstream.config import TLSOptions, create_ssl_context from rstream.errors import RuntimeError from rstream.types import TunnelProperties @@ -239,12 +239,9 @@ def _require_httpx() -> Any: def _httpx_verify(tls: TLSOptions | None) -> object: if tls is None: return True - if tls.insecure_skip_verify: - return False - if tls.ca_file is not None: - context = ssl.create_default_context(cafile=tls.ca_file) - return context - return True + context, _ = create_ssl_context("localhost", tls) + context.set_alpn_protocols(["http/1.1"]) + return context async def list_inventory( diff --git a/src/rstream/engine_discovery.py b/src/rstream/engine_discovery.py new file mode 100644 index 0000000..6145166 --- /dev/null +++ b/src/rstream/engine_discovery.py @@ -0,0 +1,146 @@ +"""Anonymous, bounded Engine API discovery cached by each runtime client.""" + +from __future__ import annotations + +import asyncio +import json +import re +import ssl +import time +from contextlib import suppress +from urllib.parse import urlsplit + +from rstream.config import TLSOptions +from rstream.errors import RuntimeError + + +def has_client_certificate(tls: TLSOptions | None) -> bool: + return tls is not None and bool(tls.cert_file or tls.certificate) + + +def _invalid() -> RuntimeError: + return RuntimeError( + "Engine returned invalid or unavailable mTLS API discovery.", + code="ERR_RSTREAM_ENGINE_DISCOVERY", + ) + + +def _target(value: object, engine: str) -> str: + if not isinstance(value, dict): + raise _invalid() + project = urlsplit("https://" + engine).hostname + project = project.split(".")[0] if project else None + api_url = value.get("mtlsApiUrl") + capabilities = value.get("capabilities") + if ( + type(value.get("version")) is not int + or value.get("version") != 1 + or value.get("projectEndpoint") != project + or not isinstance(api_url, str) + or not isinstance(capabilities, list) + or not all(isinstance(item, str) for item in capabilities) + or "engine-api-mtls" not in capabilities + ): + raise _invalid() + try: + parsed = urlsplit(api_url) + if ( + parsed.scheme != "https" + or parsed.path != "/api" + or parsed.query + or parsed.fragment + or parsed.username is not None + or parsed.password is not None + or not parsed.hostname + or parsed.hostname.split(".")[0] != project + or (parsed.port is not None and parsed.port == 0) + or re.search(r"[\s\\%]", parsed.netloc) + ): + raise _invalid() + return parsed.netloc + except ValueError as error: + raise _invalid() from error + + +class EngineAPIDiscovery: + def __init__(self) -> None: + self._cache: tuple[str, str, float] | None = None + self._pending: asyncio.Task[str] | None = None + self._closed = False + + async def resolve(self, engine: str, tls: TLSOptions | None) -> str: + if self._closed: + raise RuntimeError( + "rstream client is closed.", code="ERR_RSTREAM_CLIENT_CLOSED" + ) + if ( + self._cache + and self._cache[0] == engine + and self._cache[2] > time.monotonic() + ): + return self._cache[1] + if self._pending is None: + self._pending = asyncio.create_task(self._discover(engine, tls)) + self._pending.add_done_callback(self._finished) + return await asyncio.shield(self._pending) + + def _finished(self, task: asyncio.Task[str]) -> None: + if self._pending is task: + self._pending = None + if not task.cancelled(): + task.exception() + + def invalidate(self) -> None: + self._cache = None + + async def close(self) -> None: + self._closed = True + self._cache = None + pending = self._pending + if pending is not None: + pending.cancel() + with suppress(asyncio.CancelledError, Exception): + await pending + + async def _discover(self, engine: str, tls: TLSOptions | None) -> str: + return await asyncio.wait_for(self._request(engine, tls), timeout=5) + + async def _request(self, engine: str, tls: TLSOptions | None) -> str: + try: + import httpx + except ImportError as error: + raise RuntimeError( + "Install rstream[api] to call the engine API.", + code="ERR_RSTREAM_API_EXTRA_REQUIRED", + ) from error + trust = ssl.create_default_context(cafile=tls.ca_file if tls else None) + async with ( + httpx.AsyncClient( + verify=trust, timeout=5, follow_redirects=False, trust_env=False + ) as client, + client.stream( + "GET", f"https://{engine}/.well-known/rstream/engine" + ) as response, + ): + if response.status_code != 200: + raise _invalid() + body = bytearray() + async for chunk in response.aiter_bytes(): + body.extend(chunk) + if len(body) > 16_384: + raise _invalid() + try: + value: object = json.loads(body) + except (ValueError, UnicodeError) as error: + raise _invalid() from error + target = _target(value, engine) + cache_control = response.headers.get("Cache-Control", "").lower() + age = response.headers.get("Age", "0") + max_age = re.search(r"(?:^|,)\s*max-age=(\d+)(?:\s*,|$)", cache_control) + ttl = min(int(max_age[1]), 86_400) if max_age else 3_600 + ttl = max(0, ttl - int(age)) if age.isdigit() else 0 + if "no-store" in cache_control or "no-cache" in cache_control: + ttl = 0 + if not self._closed: + self._cache = (engine, target, time.monotonic() + ttl) + return target diff --git a/tests/e2e/test_real_engine.py b/tests/e2e/test_real_engine.py index 71783a0..07388b9 100644 --- a/tests/e2e/test_real_engine.py +++ b/tests/e2e/test_real_engine.py @@ -15,6 +15,16 @@ import rstream.wsgi as rstream_wsgi +async def test_real_engine_certificate_http_inventory() -> None: + if os.environ.get("RSTREAM_PYTHON_E2E_ENGINE_API") != "1": + pytest.skip( + "Set RSTREAM_PYTHON_E2E_ENGINE_API=1 for Engine HTTP qualification." + ) + async with real_engine_client(zero_rtt=False) as client: + await client.list_clients() + await client.list_tunnels() + + @pytest.mark.asyncio async def test_real_engine_private_bytestream_matrix() -> None: if not real_engine_enabled(): @@ -377,13 +387,17 @@ def real_engine_client(*, zero_rtt: bool) -> rstream.Client: ca_file = os.environ.get("RSTREAM_PYTHON_E2E_CA_FILE") insecure_skip_verify = os.environ.get("RSTREAM_PYTHON_E2E_TLS_INSECURE") == "1" server_name = os.environ.get("RSTREAM_PYTHON_E2E_SERVER_NAME") + cert_file = os.environ.get("RSTREAM_PYTHON_E2E_CERT_FILE") + key_file = os.environ.get("RSTREAM_PYTHON_E2E_KEY_FILE") tls = ( rstream.TLSOptions( ca_file=ca_file, insecure_skip_verify=insecure_skip_verify, server_name=server_name, + cert_file=cert_file, + key_file=key_file, ) - if ca_file or insecure_skip_verify or server_name + if ca_file or insecure_skip_verify or server_name or cert_file or key_file else None ) return rstream.Client( diff --git a/tests/integration/test_engine_mtls.py b/tests/integration/test_engine_mtls.py new file mode 100644 index 0000000..c15b8bb --- /dev/null +++ b/tests/integration/test_engine_mtls.py @@ -0,0 +1,191 @@ +from __future__ import annotations + +import asyncio +import ssl +from collections.abc import AsyncIterator +from pathlib import Path + +import pytest +import trustme +from aiohttp import web + +from rstream.client import Client +from rstream.config import TLSOptions +from rstream.engine_discovery import EngineAPIDiscovery, _target +from rstream.errors import RstreamRuntimeError + + +@pytest.fixture +async def engine_pair( + tmp_path: Path, +) -> AsyncIterator[tuple[str, TLSOptions, dict[str, int]]]: + ca = trustme.CA() + ca.cert_pem.write_to_path(tmp_path / "ca.pem") + certificate = ca.issue_cert("device.test") + certificate.cert_chain_pems[0].write_to_path(tmp_path / "client.pem") + certificate.private_key_pem.write_to_path(tmp_path / "client.key") + server_tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ca.issue_cert("localhost").configure_cert(server_tls) + ca.configure_trust(server_tls) + server_tls.verify_mode = ssl.CERT_OPTIONAL + calls = {"discovery": 0, "authenticated": 0} + + async def authenticated(request: web.Request) -> web.StreamResponse: + assert request.transport is not None + assert request.transport.get_extra_info("peercert") + assert "Authorization" not in request.headers + calls["authenticated"] += 1 + if request.path == "/api/sse": + return web.Response( + text='data: {"type":"client.connected"}\n\n', + content_type="text/event-stream", + ) + if request.path == "/api/websocket": + ws = web.WebSocketResponse() + await ws.prepare(request) + await ws.send_json({"type": "client.connected"}) + await ws.close() + return ws + return web.json_response([{"id": "device"}]) + + api = web.Application() + api.router.add_get("/api/{path}", authenticated) + api_runner = web.AppRunner(api) + await api_runner.setup() + api_site = web.TCPSite(api_runner, "127.0.0.1", 0, ssl_context=server_tls) + await api_site.start() + api_port = api_runner.addresses[0][1] + + async def discover(request: web.Request) -> web.Response: + assert request.transport is not None + assert not request.transport.get_extra_info("peercert") + assert "Authorization" not in request.headers + assert "Cookie" not in request.headers + calls["discovery"] += 1 + await asyncio.sleep(0.02) + return web.json_response( + { + "version": 1, + "projectEndpoint": "localhost", + "mtlsApiUrl": f"https://localhost:{api_port}/api", + "capabilities": ["engine-api-mtls"], + }, + headers={"Cache-Control": "public, max-age=3600"}, + ) + + public = web.Application() + public.router.add_get("/.well-known/rstream/engine", discover) + public_runner = web.AppRunner(public) + await public_runner.setup() + public_site = web.TCPSite(public_runner, "127.0.0.1", 0, ssl_context=server_tls) + await public_site.start() + public_port = public_runner.addresses[0][1] + try: + yield ( + f"localhost:{public_port}", + TLSOptions( + ca_file=str(tmp_path / "ca.pem"), + cert_file=str(tmp_path / "client.pem"), + key_file=str(tmp_path / "client.key"), + ), + calls, + ) + finally: + await public_runner.cleanup() + await api_runner.cleanup() + + +async def test_mtls_inventory_and_watch_reuse_anonymous_discovery( + engine_pair: tuple[str, TLSOptions, dict[str, int]], +) -> None: + engine, tls, calls = engine_pair + async with Client(engine=engine, tls=tls, read_config_file=False) as client: + inventories = await asyncio.gather(*(client.list_clients() for _ in range(12))) + assert all(items[0]["id"] == "device" for items in inventories) + for events in [ + client.watch(transport="sse"), + client.watch(transport="websocket"), + ]: + async with events: + assert ( + await asyncio.wait_for(anext(events), 5) + ).type == "client.connected" + assert calls == {"discovery": 1, "authenticated": 14} + with pytest.raises(RstreamRuntimeError, match="closed"): + await client.list_clients() + + +async def test_discovery_verifies_tls_and_isolates_waiter_cancellation( + engine_pair: tuple[str, TLSOptions, dict[str, int]], +) -> None: + engine, tls, calls = engine_pair + discovery = EngineAPIDiscovery() + first = asyncio.create_task(discovery.resolve(engine, tls)) + second = asyncio.create_task(discovery.resolve(engine, tls)) + await asyncio.sleep(0.005) + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + assert (await second).startswith("localhost:") + assert calls["discovery"] == 1 + await discovery.close() + import httpx + + untrusted = EngineAPIDiscovery() + try: + with pytest.raises(httpx.ConnectError): + await untrusted.resolve(engine, TLSOptions(insecure_skip_verify=True)) + finally: + await untrusted.close() + + +async def test_close_cancels_discovery_and_joins_its_task( + monkeypatch: pytest.MonkeyPatch, +) -> None: + started = asyncio.Event() + stopped = asyncio.Event() + + async def pending_request( + self: EngineAPIDiscovery, engine: str, tls: TLSOptions | None + ) -> str: + started.set() + try: + await asyncio.Event().wait() + finally: + stopped.set() + raise AssertionError("pending discovery should have been cancelled") + + monkeypatch.setattr(EngineAPIDiscovery, "_request", pending_request) + discovery = EngineAPIDiscovery() + waiter = asyncio.create_task(discovery.resolve("project.engine.test", None)) + await asyncio.wait_for(started.wait(), 1) + await discovery.close() + assert stopped.is_set() + with pytest.raises(asyncio.CancelledError): + await waiter + with pytest.raises(RstreamRuntimeError, match="closed"): + await discovery.resolve("project.engine.test", None) + + +@pytest.mark.parametrize( + "url", + [ + "http://project.mtls.test/api", + "https://other.mtls.test/api", + "https://user@project.mtls.test/api", + "https://project.mtls.test/api?token=x", + "https://project.mtls.test/other", + "https://project.mtls.test:0/api", + ], +) +def test_discovery_rejects_untrusted_authorities_and_paths(url: str) -> None: + with pytest.raises(RstreamRuntimeError): + _target( + { + "version": 1, + "projectEndpoint": "project", + "mtlsApiUrl": url, + "capabilities": ["engine-api-mtls"], + }, + "project.engine.test:443", + )