Files
turnstone/tests/test_service_auth_boundary.py
Patrick Buckley f4fd7e1f67 fix(security): classify outbound addresses by what they reach (GHSA-wm4f-79pw-pfr9) (#1003)
* fix(security): classify outbound addresses by what they reach (GHSA-wm4f-79pw-pfr9)

Five guards screened outbound URLs and each hand-rolled its own address
normalization and policy tests, so each had a different hole. An IPv6
transition address carries an IPv4 destination in its low bits and
`ipaddress` classifies the wrapper, not the destination: 64:ff9b::a9fe:a9fe
reports is_global because 64:ff9b::/96 is global unicast, while a NAT64
gateway routes it to the cloud metadata endpoint. CGNAT (100.64.0.0/10) is
neither is_private nor is_global, so a denylist built on is_private missed
it with no gateway involved at all.

Add turnstone/core/ip_classify.py as the single classifier. One function
returns exactly one policy lane — PUBLIC, PRIVATE (operator-approvable) or
NEVER — and every guard branches on the lane rather than re-deriving it.
Two overlapping booleans would make a verdict depend on which one a caller
tested first; several addresses are simultaneously globally routable and
metadata-reaching.

- Decode transition addresses per RFC 6052 §2.2 (NAT64 well-known and
  local-use prefixes, 6to4, Teredo, IPv4-mapped, IPv4-compatible) and judge
  them by the IPv4 they reach. The local-use prefix does not say which
  layout its gateway uses, so every length it can carry is decoded and the
  worst result classified.
- Share hostname resolution too. The five copies had already drifted on
  which failures they caught, and getaddrinfo raises UnicodeError — not an
  OSError — from the IDNA encoder.
- Resolution failure is a refusal, not a pass: the fetch resolves again, so
  an authority answering the guard with SERVFAIL and the fetch with an
  internal address would otherwise switch the guard off for that hop.
- Screen every redirect hop in every mode. allow_private_origin widens which
  lanes are acceptable rather than turning screening off, and the permission
  is revoked after any hop that is not wholly private.
- Cleartext http is allowed only for a hostname that RESOLVES to loopback.
  *.localhost is ordinary DNS, and trusting the name put an OIDC token
  exchange on the wire in the clear.
- Screen doctor and console-probe URLs through the classifier. Both used a
  host.startswith("169.254.") string test that never resolved, so any DNS
  name pointing at the metadata service passed and its body was returned to
  the model.
- Add known vendor metadata prefixes the stdlib does not flag, and place
  deprecated IPv6 site-local outside the public lane.

The operator's private-network opt-in still admits the whole home lab,
including IPv6 loopback, CGNAT and split-horizon hosts. Metadata,
link-local, multicast, unspecified and reserved addresses stay refused
regardless of the opt-in, including as a redirect target from an approved
private origin — the settings help and docs now say so.

Reported by @tonghuaroot.

* fix(security): close Azure/Oracle metadata gap and restore dual-stack origins

Review follow-ups on the address-classification rework.

Azure's host-agent endpoint (168.63.129.16) and Oracle Cloud's metadata
endpoint (192.0.0.192) sit in ordinary unicast space, so the stdlib reported
them as globally routable and both classified PUBLIC — reachable with no
opt-in at all, a worse position than the RFC 1918 host beside them, and
directly contradicting the "metadata stays refused even with the opt-in"
guarantee the settings help and docs now advertise. Both join the shared
vendor list.

Revoking the private-hop permission on the ORIGIN hop broke the case
`_screen_tool_url` deliberately admits: a dual-stack or split-horizon
home-lab host answering with both a LAN and a public record was approved,
then refused on its own `302 /login` — one hop was all it ever got. Track
the approved HOST instead, so redirects that stay on it remain covered while
a redirect to any other private host is still refused once the chain is no
longer wholly private.

Also:

- Try several registry candidates for the collector-scope probe instead of
  abandoning it when the first is unresolvable, which also stopped a healthy
  registry from logging as malformed.
- Bound the probe's name resolution with an explicit timeout matching the
  2s the httpx connect deadline used to provide; it runs before the console
  lifespan yields and getaddrinfo has no timeout of its own.
- Route doctor and the console probe through `web.screen_url` rather than
  keeping a third and fourth copy of parse/resolve/classify/fold, which had
  already diverged on default port and empty-hostname wording. An empty
  hostname no longer reports as a cloud-metadata refusal.
- Give `screen_url` a scheme-aware default port.
- Stop doubling the word "hostname" in the OAuth resolution refusal.
- Correct the `_screen_tool_url` docstring: it described `private_origin` as
  requiring every record to be private, which the mixed-record decision
  reversed, and `private_block` as a property of a refusal when it reports
  the lane on the success path too.
- Make the preview tests' screening stub opt-in rather than autouse — as a
  module-wide fixture it also stubbed the tests whose subject IS the screen,
  so one of them would have passed even if screening refused everything.
  Verified the module now passes with all name resolution blocked.

* fix(security): refuse mixed-record private origins instead of exempting them

The previous commit let an approved private origin redirect to itself by
exempting its hostname from the chain-wide revocation. That exemption was
wrong three ways: it was captured once and never cleared, so a public hop
could steer the fetcher back into the approved host at a path of its
choosing — reopening the private -> public -> private bypass; it was
re-entrant across same-host redirects with fresh DNS each time, so a
self-redirecting host could walk arbitrary internal addresses; and it
matched on bare hostname, so it spanned every port on the approved box.
All three were reproduced against the parent commit, which refuses them.

Delete the exemption rather than repair it. The case it existed for — a
dual-stack host answering with both a LAN and a public record — is now
refused where it is actually decidable, in `_screen_tool_url`, with the
remedy in the message: point the tool at the LAN address directly. A
granted chain therefore always starts wholly private, so the fetch guard
needs no notion of an approved host and stays one unconditional rule.

That the accommodation could not be expressed safely in the guard is the
signal: the connection may land on either record, so approving such a host
never described where the fetch would go.

Also from the same review:

- Walk the whole service registry for a collector-scope probe candidate
  instead of the first three, and split the outcome into three log lines,
  so entries that are merely unreachable stop raising the malformed-registry
  alarm and skipping the boot check cluster-wide.
- Stop the candidate walk on a resolver timeout. `asyncio.timeout` bounds
  the await, not the work, so continuing left one parked thread per timed-out
  candidate on the shared executor.
- Move the metadata-hostname denylist into `ip_classify` and enforce it in
  `screen_url`, so doctor and the console probe inherit it instead of each
  keeping a copy.
- Drop the scheme-aware default port: a numeric service does not change
  which addresses resolution returns, and classification reads only those.
  `parsed.port` is still touched so an out-of-range value refuses.
- Correct the vendor-metadata comment, which generalized a claim true of
  Azure's and Oracle's addresses to Alibaba's CGNAT one.
- Rename a test class that was still named for the rule it no longer tests.
2026-08-11 02:18:03 -07:00

701 lines
29 KiB
Python

"""Tests for the invariants that protect the console ↔ node
service-auth boundary from silent drift.
Covers:
- ``_effective_user_filter`` on ``turnstone.console.server`` — the
three-way return (None / caller_uid / DENY_EMPTY_SUB) and the four
decision branches (admin, service-scope, blank-sub non-service,
normal uid).
- ``_effective_user_filter`` on ``turnstone.server`` — mirror of the
above minus the admin bypass (node-side has no admin concept).
- ``_verify_collector_service_scope`` — 409 probe OK path,
403 drift → ``collector_scope_error`` set + ERROR log,
transient failures → no refuse-to-serve.
- ``cluster_snapshot`` + ``cluster_events_sse`` endpoints gate on
``collector_scope_error`` and return 503 with a remediation hint.
- ``_NodeDashboardCache.get`` logs 4xx at WARNING with status + body
preview.
- Cross-module identity of the ``DENY_EMPTY_SUB`` sentinel.
"""
from __future__ import annotations
import logging
from types import SimpleNamespace
from unittest.mock import MagicMock
import httpx
import pytest
from turnstone.core.auth import AuthResult
# ---------------------------------------------------------------------------
# Shared builders
# ---------------------------------------------------------------------------
def _request_with_auth(
*,
user_id: str = "",
scopes: frozenset[str] = frozenset(),
permissions: frozenset[str] = frozenset(),
) -> MagicMock:
"""Build a MagicMock Request with an AuthResult on ``request.state``.
Matches the shape the auth middleware attaches so the helpers
under test exercise the real auth-reading path.
"""
request = MagicMock()
request.state.auth_result = AuthResult(
user_id=user_id,
scopes=scopes,
token_source="test",
permissions=permissions,
)
return request
# ---------------------------------------------------------------------------
# _effective_user_filter — the console edition was deleted alongside the
# row-level ownership gates (trusted-team unification). Only the server
# edition survives — it still differentiates service callers (cluster-
# wide) from scoped users (tenant-pinned aggregates on node endpoints).
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# _effective_user_filter — server edition (service, uid, DENY — no admin)
# ---------------------------------------------------------------------------
class TestServerEffectiveUserFilter:
def test_service_scope_returns_none(self):
from turnstone.server import _effective_user_filter
req = _request_with_auth(user_id="console-proxy", scopes=frozenset({"service"}))
assert _effective_user_filter(req) is None
def test_scoped_caller_returns_uid(self):
from turnstone.server import _effective_user_filter
req = _request_with_auth(user_id="alice", scopes=frozenset({"read"}))
assert _effective_user_filter(req) == "alice"
def test_blank_sub_non_service_returns_deny(self):
from turnstone.server import DENY_EMPTY_SUB, _effective_user_filter
req = _request_with_auth(user_id="", scopes=frozenset({"read"}))
assert _effective_user_filter(req) is DENY_EMPTY_SUB
def test_server_has_no_admin_bypass(self):
"""Server-side has no admin-permissions concept — ``admin.users``
must NOT bypass the tenant filter on node endpoints. Only the
service scope crosses tenants."""
from turnstone.server import _effective_user_filter
req = _request_with_auth(
user_id="alice",
scopes=frozenset({"read"}),
permissions=frozenset({"admin.users"}),
)
# Admin perm is ignored; caller is a scoped user.
assert _effective_user_filter(req) == "alice"
# ---------------------------------------------------------------------------
# Boot self-check — _verify_collector_service_scope
# ---------------------------------------------------------------------------
def _scope_probe_app(
*,
services: list[dict] | None = None,
token: str = "probe-token",
) -> MagicMock:
"""Build a MagicMock ``app`` with the state the self-check reads."""
storage = MagicMock()
storage.list_services.return_value = services or []
token_mgr = SimpleNamespace(token=token)
app = MagicMock()
app.state.auth_storage = storage
app.state.collector_token_mgr = token_mgr
app.state.collector_scope_error = ""
return app
class TestVerifyCollectorServiceScope:
@pytest.mark.anyio
async def test_409_probe_leaves_scope_error_empty(self):
"""A 409 response means the scope gate passed; the probe's
deliberately-wrong node_id tripped the identity check only
after auth was accepted. This is the happy path."""
from turnstone.console.server import _verify_collector_service_scope
app = _scope_probe_app(
services=[
{"service_id": "node-1", "url": "http://127.0.0.1:8001"},
]
)
def handler(request: httpx.Request) -> httpx.Response:
# Caller MUST probe with expected_node_id set to the
# sentinel so the server returns 409 before opening a
# stream.
assert "expected_node_id=_scope-probe_" in str(request.url)
assert request.headers["authorization"] == "Bearer probe-token"
return httpx.Response(409, text='{"error":"node_id mismatch"}')
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
await _verify_collector_service_scope(app, client)
assert app.state.collector_scope_error == ""
@pytest.mark.anyio
async def test_403_probe_sets_scope_error_and_logs_error(self, caplog):
"""403 from the probe means the collector token is missing the
``service`` scope (or the JWT audience is misconfigured). The
probe must (1) set ``app.state.collector_scope_error`` non-empty
with a remediation hint and (2) log at ERROR so operators see
the drift at boot rather than chasing empty-dashboard reports."""
from turnstone.console.server import _verify_collector_service_scope
app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://127.0.0.1:8001"}])
def handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(403, text='{"error":"service scope required"}')
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
with caplog.at_level(logging.ERROR, logger="turnstone.console.server"):
await _verify_collector_service_scope(app, client)
err = app.state.collector_scope_error
assert err, "403 probe must populate collector_scope_error"
assert "collector token rejected" in err
assert "HTTP 403" in err
assert any(
rec.levelno == logging.ERROR and "collector_scope_probe.drift" in rec.getMessage()
for rec in caplog.records
), "403 drift must log at ERROR so operators see it at boot"
@pytest.mark.anyio
async def test_401_also_sets_scope_error(self):
"""401 (JWT audience / secret mismatch) is the same configuration
class as 403 — refuse to serve."""
from turnstone.console.server import _verify_collector_service_scope
app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://127.0.0.1:8001"}])
def handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(401, text="unauthorized")
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
await _verify_collector_service_scope(app, client)
assert "HTTP 401" in app.state.collector_scope_error
@pytest.mark.anyio
async def test_no_registered_nodes_skips_silently(self, caplog):
"""Single-node or pre-discovery states have no upstream to
probe. The self-check must NOT refuse to serve — the dashboard
simply has no cluster data to render yet."""
from turnstone.console.server import _verify_collector_service_scope
app = _scope_probe_app(services=[])
client = httpx.AsyncClient(transport=httpx.MockTransport(lambda _r: httpx.Response(500)))
with caplog.at_level(logging.INFO, logger="turnstone.console.server"):
await _verify_collector_service_scope(app, client)
assert app.state.collector_scope_error == ""
@pytest.mark.anyio
async def test_network_error_does_not_refuse(self):
"""Transient httpx.ConnectError during probe is a "cluster is
coming up" state; it must NOT be confused with scope drift.
Leave ``collector_scope_error`` empty and log a warning."""
from turnstone.console.server import _verify_collector_service_scope
app = _scope_probe_app(services=[{"service_id": "node-1", "url": "http://127.0.0.1:8001"}])
def handler(_req: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("connection refused")
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
await _verify_collector_service_scope(app, client)
assert app.state.collector_scope_error == ""
# ---------------------------------------------------------------------------
# Gated dashboard endpoints — cluster_snapshot + cluster_events_sse
# ---------------------------------------------------------------------------
class TestClusterDashboardGate:
@pytest.mark.anyio
async def test_cluster_snapshot_503_when_scope_error(self):
from turnstone.console.server import cluster_snapshot
request = MagicMock()
request.app.state.collector_scope_error = "collector token rejected by node-1"
resp = await cluster_snapshot(request)
assert resp.status_code == 503
import json
body = json.loads(resp.body)
assert body["reason"] == "collector_scope_drift"
assert "collector token rejected" in body["error"]
@pytest.mark.anyio
async def test_cluster_snapshot_200_when_scope_ok(self):
from turnstone.console.server import cluster_snapshot
request = MagicMock()
request.app.state.collector_scope_error = ""
request.app.state.collector.get_snapshot.return_value = {"nodes": []}
resp = await cluster_snapshot(request)
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Dashboard cache — 4xx log-warning floor (0d)
# ---------------------------------------------------------------------------
class TestDashboardCache4xxLogLevel:
@pytest.mark.anyio
async def test_403_logged_at_warning_with_preview(self, caplog):
"""A 4xx from the upstream dashboard fetch must log at WARNING
with the upstream body preview — silence here hides auth/scope
drift behind an empty dashboard."""
from turnstone.console.server import _NodeDashboardCache
cache = _NodeDashboardCache()
def handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(403, text='{"error":"service scope required"}')
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
with caplog.at_level(logging.WARNING, logger="turnstone.console.server"):
payload = await cache.get("node-1", "http://node-1:8001", client, {})
assert payload is None # 4xx → no payload cached
matches = [
r
for r in caplog.records
if r.levelno == logging.WARNING
and "dashboard_cache" in r.getMessage()
and "403" in r.getMessage()
]
assert matches, "4xx from dashboard fetch must log at WARNING"
assert "service scope required" in matches[0].getMessage()
@pytest.mark.anyio
async def test_200_does_not_log(self, caplog):
"""The happy path stays quiet — only 4xx raises the log floor."""
from turnstone.console.server import _NodeDashboardCache
cache = _NodeDashboardCache()
def handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"workstreams": []})
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
with caplog.at_level(logging.WARNING, logger="turnstone.console.server"):
payload = await cache.get("node-1", "http://node-1:8001", client, {})
assert payload == {"workstreams": []}
assert not [r for r in caplog.records if "dashboard_cache" in r.getMessage()]
@pytest.mark.anyio
async def test_4xx_does_not_cache_payload_none(self):
"""On 4xx the dashboard cache must skip the TTL write so an
operator scope fix shows up on the next request instead of
after the cache expires. Regression lock for the per-node
``asyncio.Lock`` already handling hot-loop protection."""
from turnstone.console.server import _NodeDashboardCache
cache = _NodeDashboardCache()
calls = {"n": 0}
def handler(_req: httpx.Request) -> httpx.Response:
calls["n"] += 1
if calls["n"] == 1:
return httpx.Response(403, text="forbidden")
return httpx.Response(200, json={"workstreams": [{"ws_id": "ws-1"}]})
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
first = await cache.get("node-1", "http://node-1:8001", client, {})
second = await cache.get("node-1", "http://node-1:8001", client, {})
assert first is None
assert second == {"workstreams": [{"ws_id": "ws-1"}]}
assert calls["n"] == 2, "4xx must bypass the cache so the retry reaches upstream"
# ---------------------------------------------------------------------------
# _fetch_live_block — 4xx log floor on the direct-fetch fallback (0d)
# ---------------------------------------------------------------------------
class TestFetchLiveBlock4xxLogLevel:
@pytest.mark.anyio
async def test_direct_fallback_logs_warning_on_4xx(self, caplog, monkeypatch):
"""Test harnesses / legacy embeddings skip the dashboard cache
and fall through to the direct-fetch path inside
``_fetch_live_block``. 4xx there must surface at WARNING —
the silence the cache path previously had also applied here."""
from turnstone.console.server import _fetch_live_block
def handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(401, text='{"error":"JWT audience mismatch"}')
# Build the request with explicit state so _proxy_auth_headers
# takes the empty-headers fallback (no auth_result, no
# jwt_secret, no service-token manager); we're exercising the
# 4xx branch, not the token-mint path.
request = MagicMock()
request.state = SimpleNamespace(auth_result=None)
request.app.state = SimpleNamespace(
proxy_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
proxy_token_mgr=None,
jwt_secret="",
dashboard_cache=None, # force the direct-fetch branch
coord_mgr=None,
)
# Shim _get_server_url so we don't need the full cluster-
# router wiring to resolve node_id → URL. monkeypatch handles
# the restore automatically.
monkeypatch.setattr(
"turnstone.console.server._get_server_url",
lambda _req, _nid: "http://node-1:8001",
)
with caplog.at_level(logging.WARNING, logger="turnstone.console.server"):
live = await _fetch_live_block(
request, {"node_id": "node-1", "kind": "interactive"}, "ws-abc"
)
assert live is None
matches = [
r
for r in caplog.records
if r.levelno == logging.WARNING and "proxy.live_block.4xx" in r.getMessage()
]
assert matches, "4xx from the direct fetch must log at WARNING"
assert "401" in matches[0].getMessage()
assert "JWT audience mismatch" in matches[0].getMessage()
# ---------------------------------------------------------------------------
# _proxy_sse — 4xx log floor on the streaming path (0d)
# ---------------------------------------------------------------------------
class TestProxySseNon200LogLevel:
@pytest.mark.anyio
async def test_non_200_upstream_logs_warning_with_preview(self, caplog):
"""Non-200 on a service-auth SSE proxy hop is operator-
actionable; the browser already sees the error event, but
operators need the drift in ops logs too."""
from starlette.requests import Request
from turnstone.console.server import _proxy_sse
def handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(403, text='{"error":"service scope required\\n"}')
sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
scope = {
"type": "http",
"method": "GET",
"path": "/node/n/api/events",
"headers": [],
"query_string": b"",
"app": MagicMock(
state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client)
),
}
async def _receive():
return {"type": "http.request", "body": b""}
request = Request(scope, receive=_receive)
# MagicMock on app.state.proxy_sse_client above is covered by
# the SimpleNamespace; auth headers fall through to the
# fallback empty-dict path since _proxy_auth_headers sees no
# auth_result.
with caplog.at_level(logging.WARNING, logger="turnstone.console.server"):
response = await _proxy_sse(request, "http://node-1:8001", "events", api_prefix="api")
# Drain the streaming body so the async gen executes.
async for _ in response.body_iterator: # type: ignore[attr-defined]
pass
matches = [
r
for r in caplog.records
if r.levelno == logging.WARNING and "proxy.sse.non_200" in r.getMessage()
]
assert matches, "non-200 from SSE proxy must log at WARNING"
# Control-char scrub replaces the literal \n byte with a
# space so the preview can't forge a log-line break.
assert "\n" not in matches[0].getMessage().split("body=", 1)[-1]
# ---------------------------------------------------------------------------
# _proxy_sse — Last-Event-ID forwarding (PR-D reconnect-with-replay)
# ---------------------------------------------------------------------------
class TestProxySseLastEventIdForwarding:
"""The console SSE proxy is the inbound SSE path for multi-node
deployments — every browser EventSource traverses it. Without
forwarding ``Last-Event-ID``, the per-ws / global SSE handlers on
the node would treat every reconnect as a fresh connect and silently
drop events emitted during the disconnect window. PR-D's whole
reconnect-with-replay foundation depends on these tests passing."""
@pytest.mark.anyio
async def test_forwards_last_event_id_header_to_upstream(self):
"""Browser sends ``Last-Event-ID``; upstream node must receive it."""
from starlette.requests import Request
from turnstone.console.server import _proxy_sse
captured_headers: dict[str, str] = {}
def handler(req: httpx.Request) -> httpx.Response:
# httpx headers are case-insensitive; capture lowercased.
captured_headers.update({k.lower(): v for k, v in req.headers.items()})
return httpx.Response(
200, text="data: {}\n\n", headers={"content-type": "text/event-stream"}
)
sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
scope = {
"type": "http",
"method": "GET",
"path": "/node/n/api/workstreams/ws-1/events",
"headers": [(b"last-event-id", b"42")],
"query_string": b"",
"app": MagicMock(
state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client)
),
}
async def _receive():
return {"type": "http.request", "body": b""}
request = Request(scope, receive=_receive)
response = await _proxy_sse(
request, "http://node-1:8001", "workstreams/ws-1/events", api_prefix="api"
)
# Drain so the upstream call actually fires.
async for _ in response.body_iterator: # type: ignore[attr-defined]
pass
assert captured_headers.get("last-event-id") == "42", (
f"Last-Event-ID not forwarded to upstream; got headers={captured_headers!r}"
)
@pytest.mark.anyio
async def test_omits_last_event_id_when_client_did_not_send_one(self):
"""Fresh connect (no header on the browser side) → no header
added on the upstream side either. Guards against
accidentally injecting a stale or fabricated value."""
from starlette.requests import Request
from turnstone.console.server import _proxy_sse
captured_headers: dict[str, str] = {}
def handler(req: httpx.Request) -> httpx.Response:
captured_headers.update({k.lower(): v for k, v in req.headers.items()})
return httpx.Response(
200, text="data: {}\n\n", headers={"content-type": "text/event-stream"}
)
sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
scope = {
"type": "http",
"method": "GET",
"path": "/node/n/api/workstreams/ws-1/events",
"headers": [],
"query_string": b"",
"app": MagicMock(
state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client)
),
}
async def _receive():
return {"type": "http.request", "body": b""}
request = Request(scope, receive=_receive)
response = await _proxy_sse(
request, "http://node-1:8001", "workstreams/ws-1/events", api_prefix="api"
)
async for _ in response.body_iterator: # type: ignore[attr-defined]
pass
assert "last-event-id" not in captured_headers, (
f"upstream got an unexpected Last-Event-ID; headers={captured_headers!r}"
)
# ---------------------------------------------------------------------------
# Gated cluster_events_sse — 503 on scope error (0a)
# ---------------------------------------------------------------------------
class TestClusterEventsSseGate:
@pytest.mark.anyio
async def test_cluster_events_sse_503_when_scope_error(self):
from turnstone.console.server import cluster_events_sse
request = MagicMock()
request.app.state.collector_scope_error = (
"collector token rejected by node-1 — upstream_body=<<<forbidden>>>"
)
resp = await cluster_events_sse(request)
assert resp.status_code == 503
import json
body = json.loads(resp.body)
assert body["reason"] == "collector_scope_drift"
# ---------------------------------------------------------------------------
# Cross-module identity of DENY_EMPTY_SUB (q-4)
# ---------------------------------------------------------------------------
class TestDenySentinelSharedIdentity:
def test_core_and_server_share_one_sentinel(self):
"""The sentinel is compared with ``is``; a future refactor
that re-introduced per-module duplicates would silently break
the identity check. Lock the cross-module invariant.
Only the server + core surfaces consume the sentinel after the
trusted-team unification — the console no longer gates on
row ownership, so its ``_effective_user_filter`` was removed."""
from turnstone.core.auth import DENY_EMPTY_SUB as CORE_DENY
from turnstone.server import DENY_EMPTY_SUB as SERVER_DENY
assert CORE_DENY is SERVER_DENY
# ---------------------------------------------------------------------------
# _bounded_body_preview control-char scrub (sec-1)
# ---------------------------------------------------------------------------
class TestBoundedBodyPreviewScrub:
def test_control_chars_replaced_with_space(self):
"""CR/LF/NUL/TAB in upstream bodies must not appear raw in
logs or in the operator-facing 503 ``collector_scope_error``
— otherwise an attacker-controllable upstream can forge
additional log lines or embed fake remediation text."""
from turnstone.console.server import _bounded_body_preview
preview = _bounded_body_preview("line-a\nline-b\r\nNUL\x00TAB\t")
assert "\n" not in preview
assert "\r" not in preview
assert "\x00" not in preview
assert "\t" not in preview
# Structure is preserved with spaces, so operators can still
# read the body preview meaningfully.
assert "line-a" in preview
assert "line-b" in preview
def test_accepts_bytes_and_decodes(self):
from turnstone.console.server import _bounded_body_preview
preview = _bounded_body_preview(b"hello\nworld")
assert preview == "hello world"
def test_caps_at_requested_length(self):
from turnstone.console.server import _bounded_body_preview
preview = _bounded_body_preview("x" * 1000, cap=50)
assert len(preview) == 50
# ---------------------------------------------------------------------------
# coordinator_metrics DENY short-circuit — shape matches happy path (bug-1)
# ---------------------------------------------------------------------------
class TestCoordinatorMetricsDenyShape:
def test_zero_payload_matches_success_keys(self):
"""The DENY short-circuit in coordinator_metrics must emit the
same key set as the success path so strict-schema consumers
don't break on the blank-sub branch."""
from turnstone.console.server import _coordinator_metrics_payload
zero = _coordinator_metrics_payload(ws_id="a" * 32)
happy = _coordinator_metrics_payload(
ws_id="a" * 32,
spawns_total=5,
spawns_last_hour=2,
child_state_counts={"idle": 3},
judge_fallback_rate=0.1,
intent_verdicts_sample=10,
)
assert set(zero.keys()) == set(happy.keys()), (
"DENY payload key set must match success payload — "
"otherwise a future field addition silently drifts"
)
# The zero payload carries ws_id through so consumers that
# key on it don't drop the response.
assert zero["ws_id"] == "a" * 32
# ---------------------------------------------------------------------------
# Probe URL allowlist (sec-3)
# ---------------------------------------------------------------------------
class TestProbeUrlAllowlist:
def test_rejects_non_http_scheme(self):
"""Probe URL picker must reject non-http(s) schemes so a
poisoned service-registry entry can't redirect the probe
through a ``file://`` or ``gs://`` transport."""
from turnstone.console.server import _probe_candidate_url
url, nid = _probe_candidate_url([{"service_id": "node-x", "url": "file:///etc/passwd"}])
assert (url, nid) == ("", "")
def test_rejects_link_local_host(self):
"""169.254.0.0/16 is the cloud metadata range; a poisoned
entry pointing there would turn the probe into an SSRF to
IMDS."""
from turnstone.console.server import _probe_candidate_url
url, nid = _probe_candidate_url(
[{"service_id": "node-x", "url": "http://169.254.169.254:80"}]
)
assert (url, nid) == ("", "")
def test_accepts_loopback_for_dev(self):
"""Single-box dev setups register the node at 127.0.0.1 — the
allowlist must let that through."""
from turnstone.console.server import _probe_candidate_url
url, nid = _probe_candidate_url([{"service_id": "node-x", "url": "http://127.0.0.1:8001"}])
assert nid == "node-x"
assert url == "http://127.0.0.1:8001"
def test_skips_malformed_entries(self):
"""Entries missing url or service_id are skipped so the loop
falls through to the next candidate."""
from turnstone.console.server import _probe_candidate_url
url, nid = _probe_candidate_url(
[
{"service_id": "", "url": "http://node-a:8001"},
{"service_id": "node-b", "url": ""},
{"service_id": "node-c", "url": "http://node-c:8001"},
]
)
assert nid == "node-c"