From 9a2db63c07d356594a98e9d13f8c4e4fbb6b9598 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Wed, 18 Mar 2026 15:45:53 -0700 Subject: [PATCH] fix: collector JWT expiry causes silent workstream data wipe (#126) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: collector JWT expiry causes silent workstream data wipe The console collector baked a one-time JWT snapshot into its httpx client headers at startup. After 1 hour (JWT expiry), every poll to server nodes returned 401. The error JSON was silently parsed as valid empty data, wiping all workstream state while nodes still appeared reachable — the cluster showed "10 nodes, 0 workstreams." Root causes fixed: - Collector: no auth baked into httpx.Client; per-request headers from ServiceTokenManager.token (auto-rotating) or static fallback - Proxy: same pattern — proxy_client/proxy_sse_client created without auth headers; _proxy_auth_headers() injects fresh token per-request - main(): static token snapshot only passed when no token_manager exists, preventing stale JWT from being stored anywhere - _fetch_node: raise_for_status() before .json() so 401s throw instead of returning error JSON as "0 workstreams" - Auth errors (401/403) logged at warning level for operator visibility * fix: address PR #126 review — type annotation, regression tests, log messages Tighten token_manager type from Any to ServiceTokenManager | None. Add two regression tests verifying 401/403 poll responses preserve existing workstream data and mark nodes unreachable. Fix misleading log messages: "jwt_minted" → "token_manager_created" since ServiceTokenManager mints lazily on first .token access. --- tests/test_console.py | 53 +++++++++++++++++++++++++++++++++- turnstone/console/collector.py | 49 ++++++++++++++++++++++++++----- turnstone/console/server.py | 23 ++++++--------- 3 files changed, 102 insertions(+), 23 deletions(-) diff --git a/tests/test_console.py b/tests/test_console.py index 9773e78c..fd52dd08 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -3,7 +3,7 @@ import asyncio import json import queue -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest @@ -264,6 +264,57 @@ class TestCollectorPolling: assert q.empty() assert len(c._nodes["node-a"].workstreams) == 0 + def test_poll_401_preserves_workstreams_and_marks_unreachable(self): + """A 401 from the server must NOT wipe workstream data.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + reachable=True, + workstreams={"ws1": {"id": "ws1", "name": "existing", "state": "idle"}}, + ) + + # Mock httpx to return 401 + import httpx as _httpx + + mock_response = _httpx.Response( + 401, + json={"error": "Unauthorized"}, + request=_httpx.Request("GET", "http://a:8080/v1/api/dashboard"), + ) + + with patch.object(c._http_client, "get", return_value=mock_response): + c._poll_all_nodes() + + # Workstream data must be preserved, node marked unreachable + assert c._nodes["node-a"].reachable is False + assert "ws1" in c._nodes["node-a"].workstreams + assert c._nodes["node-a"].workstreams["ws1"]["name"] == "existing" + + def test_poll_403_preserves_workstreams(self): + """A 403 should also preserve state and mark unreachable.""" + c = _make_collector() + c._nodes["node-a"] = NodeSnapshot( + node_id="node-a", + server_url="http://a:8080", + reachable=True, + workstreams={"ws1": {"id": "ws1", "name": "keep-me", "state": "running"}}, + ) + + import httpx as _httpx + + mock_response = _httpx.Response( + 403, + json={"error": "Forbidden"}, + request=_httpx.Request("GET", "http://a:8080/v1/api/dashboard"), + ) + + with patch.object(c._http_client, "get", return_value=mock_response): + c._poll_all_nodes() + + assert c._nodes["node-a"].reachable is False + assert "ws1" in c._nodes["node-a"].workstreams + class TestCollectorEvents: """Real-time event handling from cluster channel.""" diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index 9147547b..75e2173a 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -20,6 +20,7 @@ from typing import TYPE_CHECKING, Any import httpx if TYPE_CHECKING: + from turnstone.core.auth import ServiceTokenManager from turnstone.mq.broker import RedisBroker log = logging.getLogger("turnstone.console.collector") @@ -58,6 +59,7 @@ class ClusterCollector: max_poll_workers: int = 50, http_timeout: float = 5.0, auth_token: str = "", + token_manager: ServiceTokenManager | None = None, ): self._broker = broker self._prefix = prefix @@ -65,16 +67,20 @@ class ClusterCollector: self._discovery_interval = discovery_interval self._max_poll_workers = max_poll_workers self._http_timeout = http_timeout + self._token_manager = token_manager + # Static auth header — only used when no token_manager is present. + # When a token_manager exists, auth is injected per-request via + # extra_headers in _poll_all_nodes to avoid stale JWT expiry. + self._static_auth: dict[str, str] | None = None + if auth_token and token_manager is None: + self._static_auth = {"Authorization": f"Bearer {auth_token}"} self._lock = threading.Lock() self._nodes: dict[str, NodeSnapshot] = {} self._running = False self._threads: list[threading.Thread] = [] self._poll_pool = ThreadPoolExecutor(max_workers=max_poll_workers) - headers = {} - if auth_token: - headers["Authorization"] = f"Bearer {auth_token}" - self._http_client = httpx.Client(timeout=http_timeout, headers=headers) + self._http_client = httpx.Client(timeout=http_timeout) # SSE fan-out to browser clients self._listeners: list[queue.Queue[dict[str, Any]]] = [] @@ -236,6 +242,14 @@ class ClusterCollector: def _poll_all_nodes(self) -> None: """Fetch dashboard data from all known nodes in parallel.""" + # Snapshot current auth header for this poll cycle. Per-request + # headers avoid mutating shared client state (thread-safe). + if self._token_manager is not None: + poll_headers: dict[str, str] | None = { + "Authorization": f"Bearer {self._token_manager.token}" + } + else: + poll_headers = self._static_auth with self._lock: targets = [ (n.node_id, n.server_url) @@ -246,25 +260,44 @@ class ClusterCollector: if not targets: return - futures = {self._poll_pool.submit(self._fetch_node, nid, url): nid for nid, url in targets} + futures = { + self._poll_pool.submit(self._fetch_node, nid, url, poll_headers): nid + for nid, url in targets + } for future in as_completed(futures): nid = futures[future] try: dashboard, health = future.result() self._apply_poll(nid, dashboard, health) + except httpx.HTTPStatusError as exc: + if exc.response.status_code in (401, 403): + log.warning( + "Auth failure polling node %s: HTTP %d", nid, exc.response.status_code + ) + else: + log.debug("Failed to poll node %s: HTTP %d", nid, exc.response.status_code) + with self._lock: + if nid in self._nodes: + self._nodes[nid].reachable = False except Exception: log.debug("Failed to poll node %s", nid) with self._lock: if nid in self._nodes: self._nodes[nid].reachable = False - def _fetch_node(self, node_id: str, server_url: str) -> tuple[dict[str, Any], dict[str, Any]]: + def _fetch_node( + self, + node_id: str, + server_url: str, + extra_headers: dict[str, str] | None = None, + ) -> tuple[dict[str, Any], dict[str, Any]]: """Fetch /v1/api/dashboard and /health from a single node.""" base = server_url.rstrip("/") - dash_resp = self._http_client.get(f"{base}/v1/api/dashboard") + dash_resp = self._http_client.get(f"{base}/v1/api/dashboard", headers=extra_headers) + dash_resp.raise_for_status() dash_data: dict[str, Any] = dash_resp.json() try: - health_resp = self._http_client.get(f"{base}/health") + health_resp = self._http_client.get(f"{base}/health", headers=extra_headers) health_data: dict[str, Any] = health_resp.json() except Exception: health_data = {} diff --git a/turnstone/console/server.py b/turnstone/console/server.py index bd7543d3..c315c533 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -679,17 +679,13 @@ async def _proxy_sse( @asynccontextmanager async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: - # Create async HTTP client for proxy routes - headers: dict[str, str] = {} - token = app.state.proxy_auth_token - if token: - headers["Authorization"] = f"Bearer {token}" - app.state.proxy_client = httpx.AsyncClient(timeout=30, headers=headers) - # Separate client for SSE streams — longer read timeout, shared connection pool + # Create async HTTP clients for proxy routes. Auth headers are NOT baked + # in — _proxy_auth_headers() injects a fresh token per-request so JWTs + # auto-rotate via ServiceTokenManager instead of expiring after 1 hour. + app.state.proxy_client = httpx.AsyncClient(timeout=30) app.state.proxy_sse_client = httpx.AsyncClient( timeout=httpx.Timeout(connect=5, read=30, write=5, pool=5), limits=httpx.Limits(keepalive_expiry=30), - headers=headers, ) # Start scheduler if configured scheduler = getattr(app.state, "scheduler", None) @@ -4863,13 +4859,13 @@ def main() -> None: audience=JWT_AUD_SERVER, expiry_hours=1, ) - collector_token = collector_token_mgr.token - log.info("console.collector_jwt_minted") + log.info("console.collector_token_manager_created") collector = ClusterCollector( broker=broker, poll_interval=args.poll_interval, - auth_token=collector_token, + auth_token=collector_token if collector_token_mgr is None else "", + token_manager=collector_token_mgr, ) collector.start() @@ -4907,8 +4903,7 @@ def main() -> None: audience=JWT_AUD_SERVER, expiry_hours=1, ) - proxy_token = proxy_token_mgr.token - log.info("console.proxy_jwt_minted") + log.info("console.proxy_token_manager_created") from turnstone.core.web_helpers import parse_cors_origins @@ -4920,7 +4915,7 @@ def main() -> None: auth_config=auth_config, jwt_secret=jwt_secret, auth_storage=auth_storage, - proxy_auth_token=proxy_token, + proxy_auth_token=proxy_token if proxy_token_mgr is None else "", proxy_token_mgr=proxy_token_mgr, cors_origins=cors_origins, )