ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(console): service scope on collector token + surface upstream 4xx (#379)

Browse Source 
* fix(console): service scope on collector token + surface upstream 4xx

CRITICAL: the console's ClusterCollector ServiceTokenManager was
configured with only frozenset({"read"}) scope, but every upstream
node's /v1/api/events/global hard-gates on "service" scope (added in
PR #375 for cross-tenant authz hardening).  Every console→upstream
SSE connect 403'd, the collector never populated node state, and the
failure was silent — node health, idle workstreams, and interactive-
kind workstream rows all disappeared from the console dashboard with
no user-visible error.  The only surface was a log.debug line in the
collector's _node_sse_task that operators had to opt into via DEBUG
logging or browser DevTools.

Fix:

- Add "service" to the collector_token_mgr scopes
  (turnstone/console/server.py).  Matches the proxy_token_mgr (which
  already has it) and the existing cli / admin / channel-gateway
  service tokens.  Restores /v1/api/events/global SSE subscription
  and /v1/api/dashboard visibility (which silently tenant-filters
  non-service callers to zero rows).

- Upgrade the 4xx path in _node_sse_task to log.warning with the
  status code + 200-char body preview, so configuration-level
  failures (scope misconfig, JWT secret mismatch, expired token)
  show up in operator logs instead of being masked by the generic
  except-block debug line.  Keep transient network errors
  (CancelledError, ConnectError) at debug so the log doesn't flood
  during brief node restarts.

- Add reachable_reason field to NodeSnapshot + surface via
  get_nodes / get_node_detail / get_snapshot (and the browser's
  buildNodeInfoFromSnapshot).  Operators now see the failure cause
  on the cluster node list without tailing the log.  Cleared on
  successful reconnect in _apply_snapshot.

- Test coverage: test_server_authz.py TestGlobalEventsServiceGate
  gains a positive-path test asserting that a token with exactly
  the collector's scope set ({"read", "service"}) is accepted by
  /v1/api/events/global.  Locks in the scope contract so any future
  rename breaks the test before it breaks the dashboard.

Gate: ruff + mypy + pytest -m "not live" (4309 passed) all clean.

* fix(console): address Copilot review on PR #379

Two review comments folded in:

- collector.py — bounded body read for 4xx SSE error previews.  The
  prior ``await source.response.aread()`` buffered the entire
  upstream error body into memory just to log a 200-char preview; a
  malicious / oversized upstream response (HTML error page, proxy-
  generated body) could have forced the collector to download an
  arbitrary amount of bytes.  Iterate ``aiter_bytes()`` and stop once
  the preview cap (256 bytes, ~200 chars after UTF-8 decode) is
  satisfied.

- test_server_authz.py — tighten the service-scope positive test.
  The prior ``assert resp.status_code != 403`` could pass on
  unrelated 500s AND left an SSE stream open indefinitely.  Send
  ``?expected_node_id=definitely-wrong-node-id`` so the handler
  passes the scope gate, hits the post-auth node-identity check, and
  returns 409.  Now ``assert resp.status_code == 409`` proves the
  scope contract precisely and terminates the request immediately.

Gate: ruff + mypy + pytest -m "not live" (4309 passed) all clean.
This commit is contained in:
Patrick Buckley
2026-04-18 02:48:57 -07:00
committed by GitHub
parent 553d73109b
commit c17eddbbd8
4 changed files with 119 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_server_authz.py
+30
View File
@@ -420,6 +420,36 @@ class TestGlobalEventsServiceGate:
assert resp.status_code == 403
assert "service" in resp.json()["error"].lower()
def test_service_scope_accepted(self, app_client):
"""Regression for the console-collector 403 footgun: the
collector's ServiceTokenManager is configured in console/server.py
with scopes ``{"read", "service"}``. This gate must accept
exactly that scope set so the collector's SSE subscription
doesn't silently 403 out (#sev-0). Any future scope renaming
that would drop ``"service"`` from the node-side check breaks
this test before it breaks the dashboard.
Probe a deliberately-wrong ``expected_node_id`` — the handler
runs the scope gate first, then the node-identity check. A
409 response proves we made it past the scope gate (which is
what this test is asserting), while also avoiding an
indefinitely-open SSE stream the TestClient would never close.
"""
client, _mgr = app_client
# Exact scope set the collector uses today.
collector_scopes = frozenset({"read", "service"})
resp = client.get(
"/v1/api/events/global?expected_node_id=definitely-wrong-node-id",
headers=_auth("console-collector", scopes=collector_scopes),
)
# 409 = the scope gate passed and we hit the node-identity
# mismatch branch. Anything else (403 / 500 / 200 stream)
# is a failure for this contract.
assert resp.status_code == 409, (
f"service-scoped token did not reach node-id check: "
f"{resp.status_code} {resp.text[:120]}"
)
class TestPerWsSseGate:
def test_non_owner_rejected(self, app_client):
turnstone/console/collector.py
+78 -8
View File
@@ -46,6 +46,13 @@ class NodeSnapshot:
health: dict[str, Any] = field(default_factory=dict)
aggregate: dict[str, Any] = field(default_factory=dict)
reachable: bool = True
# Last unreachable-reason string (e.g. ``"HTTP 403"``,
# ``"ConnectError"``, ``"node_id mismatch"``). Surfaced through
# ``get_snapshot`` / ``get_nodes`` / ``get_node_detail`` so ops
# dashboards + the console node-list can show WHY a node is down
# without operators having to tail the collector log. Cleared
# when the node reconnects successfully.
reachable_reason: str = ""
class ClusterCollector:
@@ -236,10 +243,55 @@ class ClusterCollector:
params={"expected_node_id": node_id},
headers=self._auth_headers(),
) as source:
if source.response.status_code == 409:
status = source.response.status_code
if status == 409:
log.warning("Node identity mismatch for %s at %s", node_id, url)
self._mark_unreachable(node_id)
self._mark_unreachable(node_id, reason="node_id mismatch")
break # stop reconnecting — wrong node at this URL
# 4xx from upstream is ALWAYS an operator-actionable
# configuration problem (missing service scope,
# expired JWT secret mismatch, tenant misconfig) —
# surface at warning so it shows up in ops logs
# instead of silently burning SSE reconnect budget
# at debug. 403 in particular was the long-standing
# "console dashboard is empty" footgun when the
# collector token lacked ``service`` scope.
if 400 <= status < 500:
# Bounded body read — iterate aiter_bytes() up
# to the preview cap so a malicious / oversized
# upstream can't force the collector to buffer
# an arbitrary HTML error page just to log a
# 200-char preview. Stops pulling bytes as
# soon as we have enough.
body_preview = ""
try:
preview_cap = 256 # >200 chars after UTF-8 decode
chunks: list[bytes] = []
bytes_read = 0
async for chunk in source.response.aiter_bytes():
if not chunk:
continue
remaining = preview_cap - bytes_read
if remaining <= 0:
break
chunks.append(chunk[:remaining])
bytes_read += len(chunks[-1])
if bytes_read >= preview_cap:
break
body_preview = b"".join(chunks).decode("utf-8", "replace")[:200]
except Exception:
body_preview = "<unreadable>"
log.warning(
"SSE %d from node %s at %s — %s",
status,
node_id,
url,
body_preview,
)
self._mark_unreachable(node_id, reason=f"HTTP {status}")
await asyncio.sleep(min(backoff, 30) + random.random())
backoff = min(backoff * 2, 30)
continue
source.response.raise_for_status()
async for sse in source.aiter_sse():
if stop_event.is_set():
@@ -260,7 +312,7 @@ class ClusterCollector:
node_id,
data.get("node_id"),
)
self._mark_unreachable(node_id)
self._mark_unreachable(node_id, reason="node_id mismatch")
break
self._apply_snapshot(node_id, data)
backoff = 1.0
@@ -268,9 +320,14 @@ class ClusterCollector:
self._apply_delta(node_id, data)
except asyncio.CancelledError:
raise
except Exception:
log.debug("SSE error for node %s", node_id, exc_info=True)
self._mark_unreachable(node_id)
except Exception as exc:
# Network / timeout / TLS errors — expected during brief
# node restarts. Keep at debug so the log doesn't flood
# on every backoff cycle; the warning above already
# covers configuration-level failures operators need to
# see.
log.debug("SSE error for node %s: %r", node_id, exc, exc_info=True)
self._mark_unreachable(node_id, reason=type(exc).__name__)
await asyncio.sleep(min(backoff, 30) + random.random())
backoff = min(backoff * 2, 30)
@@ -280,12 +337,19 @@ class ClusterCollector:
node = self._nodes.get(node_id)
return node.server_url if node else ""
def _mark_unreachable(self, node_id: str) -> None:
"""Mark a node as unreachable (thread-safe)."""
def _mark_unreachable(self, node_id: str, reason: str = "") -> None:
"""Mark a node as unreachable (thread-safe).
``reason`` is a short human-readable diagnostic (e.g.
``"HTTP 403"``, ``"ConnectError"``) surfaced via the snapshot
+ node endpoints so operators can see WHY a node is down.
"""
with self._lock:
node = self._nodes.get(node_id)
if node:
node.reachable = False
if reason:
node.reachable_reason = reason
# -- node discovery ------------------------------------------------------
@@ -441,6 +505,9 @@ class ClusterCollector:
return
node.last_seen = time.monotonic()
node.reachable = True
# Clear the diagnostic on successful reconnect so the
# snapshot doesn't keep reporting a stale cause.
node.reachable_reason = ""
node.health = data.get("health", {})
node.aggregate = data.get("aggregate", {})
pending_events = self._reconcile_node(node_id, node, data.get("workstreams", []))
@@ -685,6 +752,7 @@ class ClusterCollector:
"started": node.started,
"last_seen": node.last_seen,
"reachable": node.reachable,
"reachable_reason": node.reachable_reason,
"health": node.health,
"version": node.health.get("version", ""),
}
@@ -781,6 +849,7 @@ class ClusterCollector:
"workstreams": [dict(ws) for ws in node.workstreams.values()],
"aggregate": dict(node.aggregate),
"reachable": node.reachable,
"reachable_reason": node.reachable_reason,
}
def get_snapshot(self) -> dict[str, Any]:
@@ -848,6 +917,7 @@ class ClusterCollector:
"server_url": node.server_url,
"max_ws": node.max_ws,
"reachable": node.reachable,
"reachable_reason": node.reachable_reason,
"version": ver,
"health": dict(node.health),
"aggregate": dict(node.aggregate),
turnstone/console/server.py
+10 -1
View File
@@ -9712,9 +9712,18 @@ def main() -> None:
from turnstone.core.auth import JWT_AUD_SERVER, ServiceTokenManager
# ``service`` scope is REQUIRED — ``/v1/api/events/global`` on every
# upstream node hard-gates on it (server.py global_events_sse), and
# ``/v1/api/dashboard`` / ``/v1/api/workstreams`` silently tenant-
# filter away all rows for non-service callers (server.py
# _visible_workstreams). Without ``service`` here, every
# console→upstream SSE connect 403s, no dashboard rows flow, the
# browser's cluster-view is empty, and the failure surfaces only
# as DEBUG-level log lines in the collector (#sev-0). ``read`` is
# kept for legacy read-path compatibility.
collector_token_mgr = ServiceTokenManager(
user_id="console-collector",
scopes=frozenset({"read"}),
scopes=frozenset({"read", "service"}),
source="console",
secret=jwt_secret,
audience=JWT_AUD_SERVER,
turnstone/console/static/app.js
+1
View File
@@ -265,6 +265,7 @@ function buildNodeInfoFromSnapshot(node) {
max_ws: node.max_ws || 10,
started: node.started || 0,
reachable: node.reachable !== false,
reachable_reason: node.reachable_reason || "",
health: node.health || {},
version: node.version || "",
};