feat: discover and authenticate the Engine HTTP API with mTLS

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.

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:

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(

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

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(