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

refactor(auth): drop workstream row-level ownership gates

Browse Source 
Turnstone is a trusted-team tool (per #400). user_id stays as
metadata for audit + display; it no longer rejects requests. Scope-
level auth via admin.workstreams / admin.coordinator tokens is the
only gate now.

Solves sec-1 (cross-tenant delete via collision on caller-supplied
ws_id, because the gate was half-implemented) and sec-2 (blank-sub
JWT bypass on empty-owner rows). Net: 359 lines of defensive
empty-string comparisons and admin=True bypass plumbing deleted.
This commit is contained in:
Patrick Buckley
2026-04-23 22:04:54 -07:00
parent 965e0b6427
commit a46dab1ac8
11 changed files with 264 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_coordinator_end_to_end.py
+7 -4
View File
@@ -178,12 +178,13 @@ def test_create_list_detail_lifecycle(tmp_path):
ids = {c["ws_id"] for c in coordinators}
assert ws_id in ids
# Coordinator created by a different user is invisible to our caller.
# Trusted-team visibility: every ``admin.coordinator`` caller sees
# every active coordinator regardless of owner.
mgr.create(user_id="other-user", name="not-mine")
resp = client.get("/v1/api/coordinator", headers=_COORD_HEADERS)
assert resp.status_code == 200
names = {c["name"] for c in resp.json()["coordinators"]}
assert "not-mine" not in names
assert "not-mine" in names
# --- Detail ---
resp = client.get(f"/v1/api/coordinator/{ws_id}", headers=_COORD_HEADERS)
@@ -431,12 +432,14 @@ def test_lazy_rehydration_on_detail_get(tmp_path):
# The endpoint triggers lazy rehydration — manager now tracks it.
assert mgr.get("persisted-coord") is not None
# Non-owner cannot reach the same endpoint (returns 404 — no existence leak).
# Trusted-team visibility: any admin.coordinator caller can read
# the coordinator's detail, regardless of ``user_id``.
resp_stranger = client.get(
"/v1/api/coordinator/persisted-coord",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp_stranger.status_code == 404
assert resp_stranger.status_code == 200
assert resp_stranger.json()["user_id"] == "user-1"
# A workstream with kind='interactive' is not reachable via the coordinator
# endpoint even when it exists in storage.
tests/test_coordinator_endpoints.py
+62 -184
View File
@@ -260,7 +260,10 @@ def test_create_returns_ws_id_and_records_audit(storage):
assert "coordinator.create" in actions
def test_list_filters_by_caller(storage):
def test_list_returns_cluster_wide(storage):
# Trusted-team visibility: any caller with admin.coordinator sees
# every active coordinator regardless of owner. ``user_id`` stays
# on the response as metadata.
mgr = _build_mgr(storage)
mgr.create(user_id="user-1", name="mine")
mgr.create(user_id="user-2", name="theirs")
@@ -269,23 +272,7 @@ def test_list_filters_by_caller(storage):
assert resp.status_code == 200
body = resp.json()
names = {c["name"] for c in body["coordinators"]}
assert names == {"mine"}
def test_list_admin_sees_all(storage):
mgr = _build_mgr(storage)
mgr.create(user_id="user-1", name="mine")
mgr.create(user_id="user-2", name="theirs")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get(
"/v1/api/coordinator",
headers={
"X-Test-User": "admin-1",
"X-Test-Perms": "admin.coordinator,admin.users",
},
)
assert resp.status_code == 200
assert len(resp.json()["coordinators"]) == 2
assert names == {"mine", "theirs"}
def _seed_closed_coord_with_history(
@@ -329,53 +316,19 @@ def saved_storage(tmp_path):
reset_storage()
def test_saved_filters_by_caller(saved_storage):
storage = saved_storage
mgr = _build_mgr(storage)
mine_id = _seed_closed_coord_with_history(mgr, storage, user_id="user-1", name="mine")
_seed_closed_coord_with_history(mgr, storage, user_id="user-2", name="theirs")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get("/v1/api/coordinator/saved", headers=_COORD_HEADERS)
assert resp.status_code == 200
body = resp.json()
assert {c["ws_id"] for c in body["coordinators"]} == {mine_id}
def test_saved_admin_sees_all(saved_storage):
def test_saved_returns_cluster_wide(saved_storage):
# Trusted-team visibility: every ``admin.coordinator`` caller sees
# every closed coordinator.
storage = saved_storage
mgr = _build_mgr(storage)
a = _seed_closed_coord_with_history(mgr, storage, user_id="user-1", name="a")
b = _seed_closed_coord_with_history(mgr, storage, user_id="user-2", name="b")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get(
"/v1/api/coordinator/saved",
headers={
"X-Test-User": "admin-1",
"X-Test-Perms": "admin.coordinator,admin.users",
},
)
resp = client.get("/v1/api/coordinator/saved", headers=_COORD_HEADERS)
assert resp.status_code == 200
assert {c["ws_id"] for c in resp.json()["coordinators"]} == {a, b}
def test_saved_blank_uid_returns_empty(saved_storage):
"""Non-admin caller with no sub gets fail-closed empty list.
Mirrors list_saved_workstreams — empty user_id must not fall through
to a cluster-wide query (would leak orphan / migration rows).
"""
storage = saved_storage
mgr = _build_mgr(storage)
_seed_closed_coord_with_history(mgr, storage, user_id="someone", name="x")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get(
"/v1/api/coordinator/saved",
headers={"X-Test-User": "", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 200
assert resp.json() == {"coordinators": []}
def test_saved_excludes_currently_loaded(saved_storage):
"""A coordinator currently in coord_mgr must NOT appear in saved cards.
@@ -426,7 +379,10 @@ def test_saved_excludes_active_state_rows(saved_storage):
assert saved_ids == {closed_id}
def test_send_to_someone_elses_coord_returns_404(storage):
def test_send_any_admin_coordinator_caller_can_send(storage):
# Trusted-team model: send is gated on admin.coordinator scope,
# not on per-row ownership. Any caller with the scope can post
# to any coordinator.
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner", name="theirs")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -435,7 +391,7 @@ def test_send_to_someone_elses_coord_returns_404(storage):
json={"message": "hi"},
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 404 # not 403 — don't leak existence
assert resp.status_code == 200
def test_send_requires_message(storage):
@@ -513,7 +469,10 @@ def test_detail_triggers_lazy_rehydration(storage):
assert mgr.get("persisted-coord") is not None
def test_detail_404_when_not_owned(storage):
def test_detail_any_admin_coordinator_caller_can_open(storage):
# Trusted-team model: any ``admin.coordinator`` caller can rehydrate
# any persisted coordinator regardless of owner. ``user_id`` stays
# on the response as metadata.
mgr = _build_mgr(storage)
storage.register_workstream("coord-x", kind="coordinator", user_id="owner")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -521,7 +480,8 @@ def test_detail_404_when_not_owned(storage):
"/v1/api/coordinator/coord-x",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 404
assert resp.status_code == 200
assert resp.json()["user_id"] == "owner"
def test_detail_404_when_kind_interactive(storage):
@@ -551,15 +511,19 @@ def test_history_returns_messages(storage):
assert any(m.get("role") == "user" and m.get("content") == "hello" for m in body["messages"])
def test_history_404_for_stranger(storage):
def test_history_any_admin_coordinator_caller_can_read(storage):
# Trusted-team visibility: history is readable by any
# ``admin.coordinator`` caller.
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner")
storage.save_message(ws.id, "user", "hello")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get(
f"/v1/api/coordinator/{ws.id}/history",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 404
assert resp.status_code == 200
assert resp.json()["ws_id"] == ws.id
# ---------------------------------------------------------------------------
@@ -595,7 +559,9 @@ def test_open_returns_already_loaded_when_in_memory(storage):
assert body.get("already_loaded") is True
def test_open_returns_404_on_ownership_mismatch_in_memory(storage):
def test_open_any_admin_coordinator_caller_succeeds_in_memory(storage):
# Trusted-team model: open is gated by admin.coordinator scope
# only; any authenticated caller can open any in-memory coordinator.
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner", name="theirs")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -603,7 +569,8 @@ def test_open_returns_404_on_ownership_mismatch_in_memory(storage):
f"/v1/api/coordinator/{ws.id}/open",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 404 # not 403 — don't leak existence
assert resp.status_code == 200
assert resp.json().get("already_loaded") is True
def test_open_rehydrates_when_not_in_memory(storage, monkeypatch):
@@ -620,28 +587,9 @@ def test_open_rehydrates_when_not_in_memory(storage, monkeypatch):
assert body["ws_id"] == "coord-rehy"
assert body["name"] == "rehydrated"
assert "already_loaded" not in body
mgr.open.assert_called_once_with("coord-rehy", user_id="user-1")
def test_open_admin_uses_open_admin(storage, monkeypatch):
mgr = _build_mgr(storage)
rehydrated = MagicMock()
rehydrated.id = "coord-rehy"
rehydrated.name = "r"
rehydrated.user_id = "someone-else"
# SessionManager.open accepts (ws_id, *, user_id, admin) — admin
# callers pass admin=True instead of the old ``open_admin`` helper.
monkeypatch.setattr(mgr, "open", MagicMock(return_value=rehydrated))
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.post(
"/v1/api/coordinator/coord-rehy/open",
headers={
"X-Test-User": "admin-1",
"X-Test-Perms": "admin.coordinator,admin.users",
},
)
assert resp.status_code == 200
mgr.open.assert_called_once_with("coord-rehy", user_id="", admin=True)
# SessionManager.open takes a single positional arg now — no
# per-caller ownership / admin plumbing.
mgr.open.assert_called_once_with("coord-rehy")
def test_open_returns_404_when_unknown_ws_id(storage, monkeypatch):
@@ -721,28 +669,16 @@ def test_children_returns_interactive_children(storage):
assert kinds == {"interactive"}
def test_children_ownership_404(storage):
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get(
f"/v1/api/coordinator/{ws.id}/children",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 404
def test_children_admin_bypass(storage):
def test_children_any_admin_coordinator_caller_sees_subtree(storage):
# Trusted-team visibility: the children subtree is readable by any
# ``admin.coordinator`` caller.
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner")
_seed_child(storage, ws.id, "a" * 32)
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
resp = client.get(
f"/v1/api/coordinator/{ws.id}/children",
headers={
"X-Test-User": "admin-1",
"X-Test-Perms": "admin.coordinator,admin.users",
},
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 200
assert len(resp.json()["items"]) == 1
@@ -804,7 +740,9 @@ def test_tasks_corrupt_envelope_returns_empty(storage):
assert resp.json() == {"version": 1, "tasks": []}
def test_tasks_ownership_404(storage):
def test_tasks_any_admin_coordinator_caller_can_read(storage):
# Trusted-team visibility: any ``admin.coordinator`` caller can
# read the tasks envelope.
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -812,7 +750,7 @@ def test_tasks_ownership_404(storage):
f"/v1/api/coordinator/{ws.id}/tasks",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 404
assert resp.status_code == 200
# ---------------------------------------------------------------------------
@@ -844,7 +782,8 @@ def test_cluster_inspect_invalid_ws_id_400(storage):
assert resp.status_code == 400
def test_cluster_inspect_ownership_404(storage):
def test_cluster_inspect_any_inspect_caller_sees_detail(storage):
# Trusted-team visibility: admin.cluster.inspect sees every row.
mgr = _build_mgr(storage)
ws = mgr.create(user_id="owner")
client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -852,7 +791,8 @@ def test_cluster_inspect_ownership_404(storage):
f"/v1/api/cluster/ws/{ws.id}/detail",
headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.cluster.inspect"},
)
assert resp.status_code == 404
assert resp.status_code == 200
assert resp.json()["persisted"]["ws_id"] == ws.id
def test_cluster_inspect_coordinator_self_path(storage):
@@ -1116,41 +1056,10 @@ def test_coordinator_rows_filters_by_caller_identity(storage):
)
return request
alice_rows = _coordinator_rows(
_request_for("alice", frozenset({"read"})),
)
names = {r["name"] for r in alice_rows}
assert names == {"alice-coord"}, "non-admin alice must NOT see bob's coordinator"
bob_rows = _coordinator_rows(_request_for("bob", frozenset({"read"})))
assert {r["name"] for r in bob_rows} == {"bob-coord"}
admin_rows = _coordinator_rows(
_request_for("admin-1", frozenset({"read", "admin.users"})),
)
assert {r["name"] for r in admin_rows} == {"alice-coord", "bob-coord"}
def test_coordinator_rows_empty_user_id_returns_empty(storage):
"""Defense-in-depth: a request with no user_id (shouldn't reach this
path through the auth middleware, but defensive) gets zero rows,
not a list_all leak."""
from unittest.mock import MagicMock
from turnstone.console.server import _coordinator_rows
mgr = _build_mgr(storage)
mgr.create(user_id="alice", name="alice-coord")
request = MagicMock()
request.app.state.coord_mgr = mgr
request.state.auth_result = AuthResult(
user_id="",
scopes=frozenset({"read"}),
token_source="test",
permissions=frozenset({"read"}),
)
assert _coordinator_rows(request) == []
# Trusted-team visibility: every caller sees every coordinator.
for caller in ("alice", "bob", "admin-1"):
rows = _coordinator_rows(_request_for(caller, frozenset({"read"})))
assert {r["name"] for r in rows} == {"alice-coord", "bob-coord"}
def _persisted_rows_request(storage, mgr, user_id: str, perms: frozenset[str]):
@@ -1232,11 +1141,10 @@ def test_coordinator_rows_dedupes_by_ws_id_in_memory_wins(storage):
assert rows[0]["state"] == "idle"
def test_coordinator_rows_persisted_respects_tenant_filter(storage):
"""Non-admin callers must not see OTHER tenants' persisted
(closed) coordinators either. Regression lock — the user_id
kwarg on list_workstreams is pushed through; the defense-in-depth
client-side empty-string check catches the tail."""
def test_coordinator_rows_persisted_cluster_wide(storage):
# Trusted-team visibility: every caller sees every persisted row,
# including rows from other identities and orphan (empty-user_id)
# rows. ``user_id`` stays on the response as metadata.
from turnstone.console.server import _coordinator_rows
from turnstone.core.workstream import WorkstreamKind
@@ -1259,33 +1167,6 @@ def test_coordinator_rows_persisted_respects_tenant_filter(storage):
kind=WorkstreamKind.COORDINATOR,
parent_ws_id=None,
)
alice_req = _persisted_rows_request(storage, mgr, "alice", frozenset({"read"}))
alice_rows = _coordinator_rows(alice_req)
assert {r["name"] for r in alice_rows} == {"alice-closed"}
bob_req = _persisted_rows_request(storage, mgr, "bob", frozenset({"read"}))
bob_rows = _coordinator_rows(bob_req)
assert {r["name"] for r in bob_rows} == {"bob-closed"}
admin_req = _persisted_rows_request(storage, mgr, "admin-1", frozenset({"read", "admin.users"}))
admin_rows = _coordinator_rows(admin_req)
assert {r["name"] for r in admin_rows} == {"alice-closed", "bob-closed"}
def test_coordinator_rows_persisted_skips_orphan_rows_for_non_admin(storage):
"""Defense-in-depth empty-string tenancy check — a persisted row
with empty user_id (migration-artifact / system-owned) must NOT
be visible to a non-admin caller with empty caller_uid either.
The SQL user_id filter above already enforces this, but duplicate
the check client-side so orphan rows never leak to any
hypothetical empty-sub JWT."""
from unittest.mock import MagicMock
from turnstone.console.server import _coordinator_rows
from turnstone.core.workstream import WorkstreamKind
mgr = _build_mgr(storage)
storage.register_workstream(
"c" * 32,
node_id="console",
@@ -1296,14 +1177,11 @@ def test_coordinator_rows_persisted_skips_orphan_rows_for_non_admin(storage):
parent_ws_id=None,
)
# Non-admin with empty caller_uid must not see the orphan.
request = MagicMock()
request.app.state.coord_mgr = mgr
request.app.state.auth_storage = storage
request.state.auth_result = AuthResult(
user_id="",
scopes=frozenset({"read"}),
token_source="test",
permissions=frozenset({"read"}),
)
assert _coordinator_rows(request) == []
for caller, perms in (
("alice", frozenset({"read"})),
("bob", frozenset({"read"})),
("admin-1", frozenset({"read", "admin.users"})),
):
request = _persisted_rows_request(storage, mgr, caller, perms)
rows = _coordinator_rows(request)
assert {r["name"] for r in rows} == {"alice-closed", "bob-closed", "orphan-closed"}
tests/test_coordinator_governance.py
+5 -2
View File
@@ -340,7 +340,10 @@ def test_restrict_rejects_non_object_body(storage):
assert resp.status_code == 400
def test_trust_toggle_tenant_404_on_foreign_coord(storage):
def test_trust_toggle_cluster_wide_access(storage):
# Trusted-team model: the trust toggle is gated on the scope
# permission, not on row-level ownership. A caller holding
# ``coordinator.trust.send`` may toggle any coord's trust state.
mgr = _build_mgr(storage)
coord = mgr.create(user_id="user-owner", name="coord-a")
coord.session, _ = _make_session_mock()
@@ -353,7 +356,7 @@ def test_trust_toggle_tenant_404_on_foreign_coord(storage):
"X-Test-Perms": "admin.coordinator,coordinator.trust.send",
},
)
assert resp.status_code == 404
assert resp.status_code == 200
def test_trust_toggle_404_when_session_not_loaded(storage):
tests/test_phase6_endpoints.py
+31 -56
View File
@@ -230,11 +230,11 @@ def test_bulk_live_admin_bypass_returns_live(storage):
assert body["denied"] == []
def test_bulk_live_tenant_filter_marks_foreign_rows_denied(storage):
"""A non-admin caller whose user_id doesn't match the row's owner
gets the ws_id in ``denied`` rather than ``results`` — no
existence-oracle leak."""
# Seed a foreign-owned interactive workstream.
def test_bulk_live_cluster_wide_visibility(storage):
"""Trusted-team visibility: any ``admin.cluster.inspect`` caller
sees every row in ``results``. ``denied`` is reserved for ids
that don't correspond to a persisted workstream (no existence
oracle for unknown ids)."""
ws_id = "b" * 32
_seed_workstream(storage, ws_id=ws_id, node_id="node-a", user_id="stranger")
client = _make_client(storage, coord_mgr=_build_mgr(storage))
@@ -244,22 +244,18 @@ def test_bulk_live_tenant_filter_marks_foreign_rows_denied(storage):
)
assert resp.status_code == 200
body = resp.json()
assert body["denied"] == [ws_id]
assert body["results"] == {}
assert ws_id in body["results"]
assert body["denied"] == []
def test_bulk_live_empty_caller_uid_denies_empty_owner_rows(storage):
"""Regression for #bug-3 / #sec-2: a caller with empty user_id
must NOT see rows with empty user_id (orphan / system-owned).
Either side empty → denied. Admin bypass honoured (tested
elsewhere)."""
ws_id = "c" * 32
_seed_workstream(storage, ws_id=ws_id, node_id="node-a", user_id="")
def test_bulk_live_unknown_ids_route_to_denied(storage):
"""Unknown ids (not in storage) land in ``denied`` so the endpoint
can't be used as an existence oracle."""
ws_id = "c" * 32 # not seeded
client = _make_client(storage, coord_mgr=_build_mgr(storage))
# caller_uid="" (empty X-Test-User) + non-admin perm.
resp = client.get(
f"/v1/api/cluster/ws/live?ids={ws_id}",
headers={"X-Test-User": "", "X-Test-Perms": "admin.cluster.inspect"},
headers={"X-Test-User": "user-1", "X-Test-Perms": "admin.cluster.inspect"},
)
assert resp.status_code == 200
body = resp.json()
@@ -315,9 +311,9 @@ def test_metrics_invalid_ws_id_400(storage):
assert resp.status_code == 400
def test_metrics_ownership_404_mask(storage):
"""A ws_id owned by another tenant returns 404, not 403 — no
existence-oracle leak (mirrors coordinator_detail)."""
def test_metrics_any_admin_coordinator_caller_can_read(storage):
"""Trusted-team visibility: metrics are readable by any caller
with ``admin.coordinator`` regardless of the coordinator owner."""
mgr = _build_mgr(storage)
ws = mgr.create(user_id="stranger")
client = _make_client(storage, coord_mgr=mgr)
@@ -325,7 +321,8 @@ def test_metrics_ownership_404_mask(storage):
f"/v1/api/coordinator/{ws.id}/metrics",
headers=_METRICS_HEADERS,
)
assert resp.status_code == 404
assert resp.status_code == 200
assert resp.json()["ws_id"] == ws.id
def test_metrics_empty_coordinator_defaults(storage):
@@ -393,20 +390,13 @@ def test_metrics_spawns_and_state_counts(storage):
assert body["child_state_counts"] == {"idle": 1, "running": 1, "closed": 1}
def test_metrics_tenant_filter_excludes_forged_cross_tenant_child(storage):
"""Defense-in-depth: a non-admin caller's aggregate counts must
exclude children whose parent_ws_id matches the coord but whose
user_id drifted to another tenant (forged / migration-era rows).
The primary defense is the 404-mask on coord ownership; this is
the secondary defense inside the aggregate queries (Copilot
review finding on PR #381).
Admin bypass sees the raw aggregate (no tenant filter) — same
pattern coordinator_children follows.
def test_metrics_cluster_wide_aggregates(storage):
"""Trusted-team model: aggregates are cluster-wide across every
caller with ``admin.coordinator``. Every child under the
coordinator counts, regardless of the ``user_id`` on the row.
"""
mgr = _build_mgr(storage)
ws = mgr.create(user_id="alice")
# Legitimate child owned by alice.
_seed_workstream(
storage,
ws_id="aa" * 16,
@@ -415,7 +405,6 @@ def test_metrics_tenant_filter_excludes_forged_cross_tenant_child(storage):
parent_ws_id=ws.id,
state="idle",
)
# Forged / drifted child — same parent_ws_id but foreign owner.
_seed_workstream(
storage,
ws_id="bb" * 16,
@@ -426,30 +415,16 @@ def test_metrics_tenant_filter_excludes_forged_cross_tenant_child(storage):
)
client = _make_client(storage, coord_mgr=mgr)
# Alice (non-admin) — counts must exclude bob's forged row.
resp = client.get(
f"/v1/api/coordinator/{ws.id}/metrics",
headers={"X-Test-User": "alice", "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 200
body = resp.json()
assert body["spawns_total"] == 1
assert body["child_state_counts"] == {"idle": 1}
# "running" (bob's forged child) filtered out.
assert "running" not in body["child_state_counts"]
# Admin sees both.
resp_admin = client.get(
f"/v1/api/coordinator/{ws.id}/metrics",
headers={
"X-Test-User": "admin-1",
"X-Test-Perms": "admin.coordinator,admin.users",
},
)
assert resp_admin.status_code == 200
body_admin = resp_admin.json()
assert body_admin["spawns_total"] == 2
assert body_admin["child_state_counts"] == {"idle": 1, "running": 1}
# Every admin.coordinator caller sees both children.
for caller in ("alice", "bob", "admin-1"):
resp = client.get(
f"/v1/api/coordinator/{ws.id}/metrics",
headers={"X-Test-User": caller, "X-Test-Perms": "admin.coordinator"},
)
assert resp.status_code == 200, caller
body = resp.json()
assert body["spawns_total"] == 2, caller
assert body["child_state_counts"] == {"idle": 1, "running": 1}, caller
def test_metrics_judge_fallback_rate_substring_match(storage):
tests/test_server_attachments_endpoints.py
+25 -14
View File
@@ -210,16 +210,17 @@ class TestUploadRejections:
)
assert resp.status_code == 404
def test_foreign_workstream_is_not_found(self, app_client):
def test_any_caller_can_attach_to_workstream(self, app_client):
# Trusted-team model: attaching to any workstream is gated on
# scope auth, not ownership. The attachment is filed under
# the ws's persisted owner so existing storage shape holds.
client, _ = app_client
# userA tries to attach to ws-B (owned by userB) — we mask this as
# 404 to avoid leaking workstream existence to non-owners.
resp = client.post(
"/v1/api/workstreams/ws-B/attachments",
files={"file": ("x.md", b"x", "text/markdown")},
headers=_auth("userA"),
)
assert resp.status_code == 404
assert resp.status_code == 200
class TestPendingCap:
@@ -298,13 +299,17 @@ class TestListAttachments:
assert all("content" not in a for a in atts)
assert {a["filename"] for a in atts} == {"a.md", "b.md"}
def test_list_isolated_per_user(self, app_client):
def test_list_visible_cluster_wide(self, app_client):
# Trusted-team visibility: any authenticated caller can list
# the attachments on any workstream. Attachments are filed
# under the ws's owner uid so a cross-caller lister still sees
# the owner's pending uploads.
client, _ = app_client
_upload(client, "ws-A", "userA", "mine.md", b"mine", "text/markdown")
# userB can't even GET listing on ws-A (not their workstream);
# masked as 404 to avoid existence-leak.
resp = client.get("/v1/api/workstreams/ws-A/attachments", headers=_auth("userB"))
assert resp.status_code == 404
assert resp.status_code == 200
atts = resp.json()["attachments"]
assert {a["filename"] for a in atts} == {"mine.md"}
class TestGetContent:
@@ -343,15 +348,19 @@ class TestGetContent:
assert resp.headers["content-type"].startswith("text/plain")
assert resp.headers.get("x-content-type-options") == "nosniff"
def test_get_content_wrong_user_is_not_found(self, app_client):
def test_get_content_visible_cluster_wide(self, app_client):
# Trusted-team visibility: any authenticated caller can fetch
# the content of an attachment on any workstream. Attachments
# are keyed by the ws's persisted owner uid so userB still
# resolves userA's blob via _require_ws_access's owner return.
client, _ = app_client
aid = _upload(client, "ws-A", "userA", "t.md", b"x", "text/markdown")
resp = client.get(
f"/v1/api/workstreams/ws-A/attachments/{aid}/content",
headers=_auth("userB"),
)
# 404 rather than 403 — caller can't distinguish from "ws doesn't exist".
assert resp.status_code == 404
assert resp.status_code == 200
assert resp.content == b"x"
def test_get_content_cross_workstream_id_404(self, app_client):
client, _ = app_client
@@ -400,12 +409,14 @@ class TestDelete:
resp = client.delete(f"/v1/api/workstreams/ws-A/attachments/{aid}", headers=_auth("userA"))
assert resp.status_code == 404
def test_delete_wrong_user_is_not_found(self, app_client):
def test_delete_cluster_wide(self, app_client):
# Trusted-team model: any authenticated caller can delete an
# attachment on any workstream. The filed ``user_id`` stays
# for audit even after a cross-caller delete.
client, _ = app_client
aid = _upload(client, "ws-A", "userA", "t.md", b"x", "text/markdown")
resp = client.delete(f"/v1/api/workstreams/ws-A/attachments/{aid}", headers=_auth("userB"))
# userB doesn't own ws-A — masked as 404 to avoid existence-leak.
assert resp.status_code == 404
assert resp.status_code == 200
# ---------------------------------------------------------------------------
tests/test_server_authz.py
+22 -10
View File
@@ -272,7 +272,10 @@ def _register_ws(storage: Any, ws_id: str, owner: str) -> None:
class TestCrossTenantDelete:
def test_non_owner_cannot_delete(self, app_client):
def test_any_caller_can_delete(self, app_client):
# Trusted-team model: scope auth gates the endpoint, not
# row-level ownership. ``user_id`` stays on audit + storage
# metadata.
from turnstone.core.storage import get_storage
client, _mgr = app_client
@@ -283,9 +286,7 @@ class TestCrossTenantDelete:
"/v1/api/workstreams/ws-victim/delete",
headers=_auth("attacker-user"),
)
assert resp.status_code == 404
# Victim's workstream still present in storage.
assert storage.get_workstream("ws-victim") is not None
assert resp.status_code == 200
def test_owner_delete_records_audit(self, app_client):
from turnstone.core.storage import get_storage
@@ -336,7 +337,11 @@ class TestCrossTenantClose:
class TestCrossTenantTitle:
def test_non_owner_cannot_refresh_title(self, app_client):
def test_refresh_title_requires_live_session(self, app_client):
# Trusted-team model: scope-level auth is the gate; any caller
# can hit the endpoint. A not-currently-active workstream
# still 404s because the refresh needs the live session, not
# because of tenant mismatch.
from turnstone.core.storage import get_storage
client, _mgr = app_client
@@ -348,8 +353,13 @@ class TestCrossTenantTitle:
headers=_auth("attacker-user"),
)
assert resp.status_code == 404
assert "not active" in resp.json().get("error", "") or "not found" in resp.json().get(
"error", ""
)
def test_non_owner_cannot_set_title(self, app_client):
def test_any_caller_can_set_title(self, app_client):
# Trusted-team model: title is editable by any authenticated
# caller; ``user_id`` remains metadata.
from turnstone.core.storage import get_storage
client, _mgr = app_client
@@ -358,14 +368,16 @@ class TestCrossTenantTitle:
_register_ws(storage, "ws-victim", "victim-user")
resp = client.post(
"/v1/api/workstreams/ws-victim/title",
json={"title": "phishing title"},
json={"title": "updated title"},
headers=_auth("attacker-user"),
)
assert resp.status_code == 404
assert resp.status_code == 200
class TestCrossTenantOpen:
def test_non_owner_cannot_open_persisted(self, app_client):
def test_any_caller_can_open_persisted(self, app_client):
# Trusted-team model: open is gated on scope auth, not on row
# ownership. The persisted ``user_id`` stays as metadata.
from turnstone.core.storage import get_storage
client, _mgr = app_client
@@ -376,7 +388,7 @@ class TestCrossTenantOpen:
"/v1/api/workstreams/ws-victim/open",
headers=_auth("attacker-user"),
)
assert resp.status_code == 404
assert resp.status_code == 200
class TestListWorkstreamsTrustedTeamVisibility:
tests/test_service_auth_boundary.py
+10 -53
View File
@@ -57,58 +57,13 @@ def _request_with_auth(
# ---------------------------------------------------------------------------
# _effective_user_filter — console edition (admin, service, uid, DENY)
# _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).
# ---------------------------------------------------------------------------
class TestConsoleEffectiveUserFilter:
def test_admin_returns_none(self):
from turnstone.console.server import _effective_user_filter
req = _request_with_auth(user_id="alice", permissions=frozenset({"admin.users"}))
assert _effective_user_filter(req) is None
def test_admin_roles_perm_also_bypasses(self):
from turnstone.console.server import _effective_user_filter
req = _request_with_auth(user_id="carol", permissions=frozenset({"admin.roles"}))
assert _effective_user_filter(req) is None
def test_service_scope_returns_none(self):
from turnstone.console.server import _effective_user_filter
req = _request_with_auth(user_id="svc-proxy", scopes=frozenset({"service"}))
assert _effective_user_filter(req) is None
def test_scoped_caller_returns_uid(self):
from turnstone.console.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_sentinel(self):
from turnstone.console.server import DENY_EMPTY_SUB, _effective_user_filter
req = _request_with_auth(user_id="", scopes=frozenset({"read"}))
result = _effective_user_filter(req)
assert result is DENY_EMPTY_SUB, (
"blank-sub non-service callers must fail closed — "
"passing through to storage with user_id=None is a "
"service escape and user_id='' matches legacy orphans"
)
def test_deny_sentinel_is_singleton(self):
"""Callers compare with ``is``; equality against a bare object()
must never match the sentinel, and two separate reads of the
attribute return the same instance (ruling out a property /
factory that would break ``is`` identity)."""
from turnstone.console.server import DENY_EMPTY_SUB as FIRST_READ
from turnstone.console.server import DENY_EMPTY_SUB as SECOND_READ
assert FIRST_READ is not object()
assert FIRST_READ is SECOND_READ
# ---------------------------------------------------------------------------
# _effective_user_filter — server edition (service, uid, DENY — no admin)
# ---------------------------------------------------------------------------
@@ -513,15 +468,17 @@ class TestClusterEventsSseGate:
class TestDenySentinelSharedIdentity:
def test_console_and_server_share_one_sentinel(self):
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."""
from turnstone.console.server import DENY_EMPTY_SUB as CONSOLE_DENY
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 CONSOLE_DENY
assert CORE_DENY is SERVER_DENY
tests/test_session_manager.py
+19 -11
View File
@@ -277,9 +277,12 @@ def test_create_rolls_back_slot_on_session_failure() -> None:
mgr, _, storage = _make_manager(adapter=adapter)
with pytest.raises(RuntimeError, match="build_session forced failure"):
mgr.create(user_id="u1")
# Slot freed, row deleted, no dangling capacity consumption.
# Slot freed — no dangling capacity consumption. The storage row
# survives construction failure on purpose: the next ``open(ws_id)``
# retries build_session rather than forcing the user to create a
# brand-new workstream.
assert mgr.count == 0
assert len(storage.rows) == 0
assert len(storage.rows) == 1
def test_create_rolls_back_slot_on_persist_failure() -> None:
@@ -329,7 +332,7 @@ def test_concurrent_create_does_not_exceed_max_active() -> None:
def test_open_returns_none_for_missing_row() -> None:
mgr, _, _ = _make_manager()
assert mgr.open("missing", user_id="u1") is None
assert mgr.open("missing") is None
def test_open_blocks_deleted_state() -> None:
@@ -338,7 +341,7 @@ def test_open_blocks_deleted_state() -> None:
mgr.close(ws.id)
# Flip the row to the tombstone state — open must refuse it.
storage.rows[ws.id].state = "deleted"
assert mgr.open(ws.id, user_id="u1") is None
assert mgr.open(ws.id) is None
def test_open_resurrects_closed_state() -> None:
@@ -348,7 +351,7 @@ def test_open_resurrects_closed_state() -> None:
mgr.close(ws_id)
assert mgr.get(ws_id) is None
reopened = mgr.open(ws_id, user_id="u1")
reopened = mgr.open(ws_id)
assert reopened is not None
assert reopened.id == ws_id
assert reopened.session is not None
@@ -357,13 +360,18 @@ def test_open_resurrects_closed_state() -> None:
assert ws_id in [e.ws_id for e in adapter.events_of("created")]
def test_open_rejects_non_matching_user_when_not_admin() -> None:
def test_open_ignores_owner_mismatch() -> None:
# Turnstone is a trusted-team tool; row-level ownership is
# metadata, not an access boundary. ``open`` no longer cares
# who the caller is — any authenticated caller can rehydrate
# any persisted workstream. Scope-level auth at the HTTP
# layer is the only gate.
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1")
mgr.close(ws.id)
assert mgr.open(ws.id, user_id="u2", admin=False) is None
# Admin bypass works.
assert mgr.open(ws.id, user_id="u2", admin=True) is not None
reopened = mgr.open(ws.id)
assert reopened is not None
assert reopened.user_id == "u1" # metadata preserved
def test_open_rejects_wrong_kind() -> None:
@@ -372,7 +380,7 @@ def test_open_rejects_wrong_kind() -> None:
# Storage row claims a different kind than our adapter's.
storage.rows[ws.id].kind = WorkstreamKind.COORDINATOR.value
mgr.close(ws.id)
assert mgr.open(ws.id, user_id="u1") is None
assert mgr.open(ws.id) is None
def test_concurrent_open_for_same_ws_id_returns_same_session() -> None:
@@ -387,7 +395,7 @@ def test_concurrent_open_for_same_ws_id_returns_same_session() -> None:
lock = threading.Lock()
def _open() -> None:
r = mgr.open(ws_id, user_id="u1")
r = mgr.open(ws_id)
with lock:
results.append(r)
turnstone/console/server.py
+36 -226
View File
@@ -46,11 +46,9 @@ from turnstone.console.metrics import ConsoleMetrics
from turnstone.console.router import ConsoleRouter
from turnstone.core.audit import record_audit
from turnstone.core.auth import (
DENY_EMPTY_SUB,
JWT_AUD_CONSOLE,
JWT_AUD_SERVER,
AuthMiddleware,
_DenyFilter,
create_jwt,
jwt_version_slot,
require_permission,
@@ -450,26 +448,15 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
In-memory wins on ws_id conflict so live state stays authoritative
for active sessions.
Ownership filter: non-admin callers must not see other tenants'
coordinator rows. Admins (``admin.users`` / ``admin.roles``) get
the full set. Unauthenticated callers shouldn't reach this path
— the endpoint sits behind global auth — but the filter defaults
to empty on a missing ``user_id`` rather than leaking.
Trusted-team visibility (post-#400): the cluster dashboard shows
every coordinator regardless of caller identity; ``user_id`` is
surfaced on each row as display metadata.
"""
coord_mgr = getattr(request.app.state, "coord_mgr", None)
if coord_mgr is None:
return []
filt = _effective_user_filter(request)
if filt is DENY_EMPTY_SUB:
return []
# filt is now ``None`` (admin / service) or a non-empty caller uid.
# ``list_for_user`` lived on the old CoordinatorManager; SessionManager
# only exposes ``list_all`` — inline the owner filter at the call site.
try:
all_wss = coord_mgr.list_all()
wss = (
all_wss if filt is None else [ws for ws in all_wss if ws.user_id and ws.user_id == filt]
)
wss = coord_mgr.list_all()
except Exception:
log.debug("cluster_workstreams.coord_list_failed", exc_info=True)
return []
@@ -506,16 +493,15 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
# Second lane — persisted coordinator rows, used to surface
# closed / error / deleted coordinators the manager has already
# evicted from ``self._workstreams``. ``filt`` carries the same
# tenant decision as the in-memory lane above (None = admin /
# service; str = scoped caller).
# evicted from ``self._workstreams``. Cluster-wide (trusted-team
# visibility).
storage = getattr(request.app.state, "auth_storage", None)
if storage is None:
return rows
try:
persisted = storage.list_workstreams(
kind=WorkstreamKind.COORDINATOR,
user_id=filt,
user_id=None,
limit=200,
)
except Exception:
@@ -534,12 +520,6 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
if not row_id or row_id in seen:
continue
row_owner = m.get("user_id") or ""
# Defense-in-depth empty-string tenancy check. The SQL user_id
# filter above already enforces the match when ``filt`` is set;
# drop any rows whose owner still lands empty (migration
# artifacts) for non-admin callers.
if filt is not None and (not row_owner or row_owner != filt):
continue
rows.append(
{
"id": row_id,
@@ -894,15 +874,6 @@ async def cluster_ws_detail(request: Request) -> JSONResponse:
if row is None:
return JSONResponse({"error": "workstream not found"}, status_code=404)
user_id = _auth_user_id(request)
err404 = _check_row_owner_or_404(
request,
row.get("user_id") or "",
user_id,
)
if err404 is not None:
return err404
try:
live = await _fetch_live_block(request, row, ws_id)
except Exception:
@@ -991,14 +962,6 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
if not cleaned:
return JSONResponse({"results": {}, "denied": [], "truncated": False})
filt = _effective_user_filter(request)
if filt is DENY_EMPTY_SUB:
# Non-admin, non-service caller with a blank sub — every row
# is denied by the empty-string rule. Skip the batch fetch
# entirely and route all ids to ``denied`` (no existence
# oracle).
return JSONResponse({"results": {}, "denied": cleaned, "truncated": truncated})
try:
rows = await asyncio.to_thread(storage.get_workstreams_batch, cleaned)
except Exception:
@@ -1020,14 +983,9 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
for wid in cleaned:
row = rows.get(wid)
if row is None:
denied.append(wid)
continue
# Tenant check — admin / service (``filt is None``) sees all
# rows; scoped callers (``filt`` is a uid) see only matching
# rows. An orphan / migration-artifact row with ``user_id=""``
# is denied for scoped callers by the ``not row_owner`` guard.
row_owner = row.get("user_id") or ""
if filt is not None and (not row_owner or row_owner != filt):
# Missing rows route to ``denied`` rather than ``results``
# so the endpoint can't be used as an existence oracle for
# ids outside the caller's knowledge.
denied.append(wid)
continue
owned_rows.append((wid, row))
@@ -2368,28 +2326,6 @@ def _require_admin_coordinator(
)
def _check_row_owner_or_404(
request: Request,
row_owner: str,
user_id: str,
*,
error_msg: str = "workstream not found",
) -> JSONResponse | None:
"""Ownership gate with the empty-string defense.
Non-admin callers need BOTH ``user_id`` and ``row_owner`` to be
non-empty AND equal. Either side being empty is a 404 — otherwise
an orphan / migration-artifact / system-owned row with
``user_id=""`` would leak to a (hypothetical) caller whose JWT
carries an empty ``sub`` claim. Admin bypass honoured.
"""
if _is_admin(request):
return None
if not user_id or not row_owner or row_owner != user_id:
return JSONResponse({"error": error_msg}, status_code=404)
return None
def _resolve_coordinator_or_404(
request: Request,
coord_mgr: Any,
@@ -2397,21 +2333,22 @@ def _resolve_coordinator_or_404(
ws_id: str,
user_id: str,
) -> tuple[Any, JSONResponse | None]:
"""Resolve a coordinator workstream and check ownership.
"""Resolve a coordinator workstream by id.
Returns ``(ws, None)`` on success — ``ws`` is the in-memory
``Workstream`` when present, ``None`` when the coordinator is
persisted but not loaded (callers may then fall through to
``storage`` directly or trigger lazy rehydration). Returns
``(None, 404)`` on missing row / wrong kind / storage unavailable
/ ownership mismatch.
``(None, 404)`` on missing row / wrong kind / storage unavailable.
Centralises the manager-first, storage-fallback, 404-mask ladder
previously duplicated across ``coordinator_children`` /
``coordinator_tasks`` / ``coordinator_history``. Ownership uses
:func:`_check_row_owner_or_404` so the empty-string defense
applies uniformly.
``coordinator_tasks`` / ``coordinator_history``. Turnstone is a
trusted-team tool — ``user_id`` is metadata, not an access
boundary, so this helper no longer gates on row ownership; scope
auth (``admin.coordinator``) upstream is the gate.
"""
del user_id # retained in signature for caller-site clarity; not consulted here
miss = JSONResponse({"error": "coordinator not found"}, status_code=404)
ws = coord_mgr.get(ws_id) if coord_mgr is not None else None
if ws is None:
@@ -2424,23 +2361,7 @@ def _resolve_coordinator_or_404(
return None, miss
if row is None or row.get("kind") != WorkstreamKind.COORDINATOR:
return None, miss
err = _check_row_owner_or_404(
request,
row.get("user_id") or "",
user_id,
error_msg="coordinator not found",
)
if err is not None:
return None, err
return None, None
err = _check_row_owner_or_404(
request,
ws.user_id or "",
user_id,
error_msg="coordinator not found",
)
if err is not None:
return None, err
return ws, None
@@ -2454,53 +2375,6 @@ def _auth_scopes(request: Request) -> set[str]:
return set(getattr(auth, "scopes", []) or [])
def _is_admin(request: Request) -> bool:
auth = getattr(getattr(request, "state", None), "auth_result", None)
perms: frozenset[str] = getattr(auth, "permissions", frozenset())
return "admin.users" in perms or "admin.roles" in perms
def _effective_user_filter(request: Request) -> str | None | _DenyFilter:
"""Resolve the effective ``user_id`` filter for a tenant-scoped aggregate.
Returns one of:
- ``None`` — admin (``admin.users`` / ``admin.roles``) or
service-scoped caller; no tenant filter — storage helpers see
cluster-wide rows.
- ``str`` — non-admin, non-service caller with a resolved uid;
storage helpers MUST receive ``user_id=<uid>`` and push the
filter into SQL.
- :data:`DENY_EMPTY_SUB` — non-admin, non-service caller whose
``sub`` claim is blank. Callers MUST short-circuit with their
endpoint's empty-shape response; passing ``None`` through to
storage would be a service-escape and passing ``""`` would
accidentally match legacy orphan rows with empty ``user_id``.
.. note::
The service-scope bypass is end-to-end only on endpoints that
do NOT first gate on :func:`_check_row_owner_or_404` (which
currently bypasses for ``admin.*`` permissions but not for
service scope). A service-scoped non-admin caller to
``coordinator_children`` / ``coordinator_metrics`` is 404'd
before the filter runs; the bypass there is reachable only by
admin callers. ``_coordinator_rows`` and
``cluster_ws_live_bulk`` honour the full three-way return.
See the class-level tenancy contract on
:class:`~turnstone.core.storage._protocol.StorageBackend` for the
storage-side requirement.
"""
if _is_admin(request):
return None
if "service" in _auth_scopes(request):
return None
uid = _auth_user_id(request)
if not uid:
return DENY_EMPTY_SUB
return uid
async def coordinator_create(request: Request) -> JSONResponse:
"""POST /v1/api/coordinator/new — create a new coordinator session."""
from turnstone.core.audit import record_audit
@@ -2603,16 +2477,9 @@ async def coordinator_send(request: Request) -> JSONResponse:
message = (body.get("message") or "").strip()
if not message:
return JSONResponse({"error": "message is required"}, status_code=400)
user_id = _auth_user_id(request)
ws = coord_mgr.get(ws_id)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
# Strict equality (not short-circuit on empty ws.user_id) —
# empty-owner rows would otherwise leak ws_id/state/history
# across tenants to anyone with admin.coordinator. 404 (not
# 403) to avoid leaking existence.
return JSONResponse({"error": "coordinator not found"}, status_code=404)
coord_adapter = getattr(request.app.state, "coord_adapter", None)
if coord_adapter is None or not coord_adapter.send(ws_id, message):
# Distinguish "worker busy + queue full" from "ws not loaded".
@@ -2642,13 +2509,9 @@ async def coordinator_approve(request: Request) -> JSONResponse:
approved = bool(body.get("approved", False))
feedback = body.get("feedback")
always = bool(body.get("always", False))
user_id = _auth_user_id(request)
ws = coord_mgr.get(ws_id)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
# Strict equality — empty-owner rows must not leak across tenants.
return JSONResponse({"error": "coordinator not found"}, status_code=404)
ui = ws.ui
if ui is None or not hasattr(ui, "resolve_approval"):
return JSONResponse(
@@ -2682,9 +2545,6 @@ async def coordinator_cancel(request: Request) -> JSONResponse:
ws = coord_mgr.get(ws_id)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
# Strict equality — empty-owner rows must not leak across tenants.
return JSONResponse({"error": "coordinator not found"}, status_code=404)
coord_mgr.cancel(ws_id)
storage = getattr(request.app.state, "auth_storage", None)
if storage is not None:
@@ -2718,9 +2578,6 @@ async def coordinator_close(request: Request) -> JSONResponse:
ws = coord_mgr.get(ws_id)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
# Strict equality — empty-owner rows must not leak across tenants.
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if not coord_mgr.close(ws_id):
return JSONResponse({"error": "close failed"}, status_code=500)
storage = getattr(request.app.state, "auth_storage", None)
@@ -2749,13 +2606,9 @@ async def coordinator_events(request: Request) -> Response:
if err503 is not None:
return err503
ws_id = request.path_params.get("ws_id", "")
user_id = _auth_user_id(request)
ws = coord_mgr.get(ws_id)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
# Strict equality — empty-owner rows must not leak across tenants.
return JSONResponse({"error": "coordinator not found"}, status_code=404)
ui = ws.ui
if ui is None or not hasattr(ui, "_register_listener"):
return JSONResponse({"error": "coordinator has no UI"}, status_code=409)
@@ -2831,13 +2684,10 @@ async def coordinator_list(request: Request) -> JSONResponse:
coord_mgr, err503 = _require_coord_mgr(request)
if err503 is not None:
return err503
user_id = _auth_user_id(request)
all_rows = coord_mgr.list_all()
rows = (
all_rows
if _is_admin(request)
else [ws for ws in all_rows if ws.user_id and ws.user_id == user_id]
)
# Trusted-team visibility: any admin.coordinator-scoped caller
# sees cluster-wide active coordinators. ``user_id`` stays on the
# response row as metadata for the UI.
rows = coord_mgr.list_all()
return JSONResponse(
{
"coordinators": [
@@ -2879,17 +2729,10 @@ async def coordinator_saved(request: Request) -> JSONResponse:
if err is not None:
return err
if _is_admin(request):
user_filter: str | None = None
else:
caller_uid = _auth_user_id(request)
if not caller_uid:
# Blank sub on a non-service / non-admin token — fail closed
# rather than match every orphan / migration row with empty
# user_id. Mirrors list_saved_workstreams.
return JSONResponse({"coordinators": []})
user_filter = caller_uid
# Trusted-team visibility (post-#400): any caller with the
# ``admin.coordinator`` scope sees cluster-wide saved coordinators;
# ``user_id`` stays as metadata on the persisted row but isn't a
# filter here.
# Offload the blocking storage call + the lock-acquiring list_all
# off the event loop, matching coordinator_create's pattern (#perf-2
# from the saved-coordinators review). list_workstreams_with_history
@@ -2900,7 +2743,7 @@ async def coordinator_saved(request: Request) -> JSONResponse:
list_workstreams_with_history,
limit=50,
kind=WorkstreamKind.COORDINATOR,
user_id=user_filter,
user_id=None,
state="closed",
)
@@ -2974,7 +2817,6 @@ async def coordinator_detail(request: Request) -> JSONResponse:
if err503 is not None:
return err503
ws_id = request.path_params.get("ws_id", "")
user_id = _auth_user_id(request)
ws = coord_mgr.get(ws_id)
if ws is None:
# Lazy rehydration goes through the session factory, which can
@@ -2982,11 +2824,7 @@ async def coordinator_detail(request: Request) -> JSONResponse:
# the correlation-id mask so stack traces don't leak through
# the detail endpoint either.
try:
ws = (
coord_mgr.open(ws_id, user_id="", admin=True)
if _is_admin(request)
else coord_mgr.open(ws_id, user_id=user_id)
)
ws = coord_mgr.open(ws_id)
except ValueError as exc:
return JSONResponse({"error": str(exc)}, status_code=503)
except Exception:
@@ -3008,9 +2846,6 @@ async def coordinator_detail(request: Request) -> JSONResponse:
)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
# Strict equality — empty-owner rows must not leak across tenants.
return JSONResponse({"error": "coordinator not found"}, status_code=404)
return JSONResponse(
{
"ws_id": ws.id,
@@ -3040,18 +2875,11 @@ async def coordinator_open(request: Request) -> JSONResponse:
ws_id = request.path_params.get("ws_id", "")
if not ws_id:
return JSONResponse({"error": "ws_id is required"}, status_code=400)
user_id = _auth_user_id(request)
ws = coord_mgr.get(ws_id)
if ws is not None:
if ws.user_id != user_id and not _is_admin(request):
return JSONResponse({"error": "coordinator not found"}, status_code=404)
return JSONResponse({"ws_id": ws.id, "name": ws.name, "already_loaded": True})
try:
ws = (
coord_mgr.open(ws_id, user_id="", admin=True)
if _is_admin(request)
else coord_mgr.open(ws_id, user_id=user_id)
)
ws = coord_mgr.open(ws_id)
except ValueError as exc:
return JSONResponse({"error": str(exc)}, status_code=503)
except Exception:
@@ -3072,8 +2900,6 @@ async def coordinator_open(request: Request) -> JSONResponse:
)
if ws is None:
return JSONResponse({"error": "coordinator not found"}, status_code=404)
if ws.user_id != user_id and not _is_admin(request):
return JSONResponse({"error": "coordinator not found"}, status_code=404)
return JSONResponse({"ws_id": ws.id, "name": ws.name})
@@ -3147,21 +2973,15 @@ async def coordinator_children(request: Request) -> JSONResponse:
if err404 is not None:
return err404
# Tenant filter: push the caller's user_id into SQL so forged /
# migration-era rows with the same parent_ws_id but a different
# owner can't leak through. Admins / service scope bypass the
# filter (they're expected to see the full subtree). The prior
# _resolve_coordinator_or_404 already rejected non-admin callers
# with a blank sub, so DENY here is defense-in-depth.
filter_user_id = _effective_user_filter(request)
if filter_user_id is DENY_EMPTY_SUB:
return JSONResponse({"items": [], "truncated": False})
# Trusted-team visibility: any caller with admin.coordinator sees
# the full child subtree. ``user_id`` stays on each row as
# metadata, not a filter.
try:
raw = storage.list_workstreams(
limit=_CHILDREN_PAGE_LIMIT + 1,
parent_ws_id=ws_id,
kind=None,
user_id=filter_user_id,
user_id=None,
)
except Exception:
correlation_id = secrets.token_hex(4)
@@ -3281,27 +3101,17 @@ async def coordinator_metrics(request: Request) -> JSONResponse:
# (#perf-1). Two cheap queries instead of a ``list_workstreams``
# scan up to 10k rows.
#
# Tenant filter on the aggregates — push user_id into SQL so a
# non-admin caller can't observe cross-tenant counts via forged /
# migration-era rows sharing parent_ws_id. The 404-mask above
# already rejected foreign coord_ws_id, so the filter here is
# defense-in-depth against child rows with drifted user_id.
# Trusted-team visibility: aggregates run cluster-wide per the
# unified ownership model; ``user_id`` is not a filter here.
from datetime import UTC, datetime
filter_user_id = _effective_user_filter(request)
if filter_user_id is DENY_EMPTY_SUB:
# Prior _resolve_coordinator_or_404 already rejected this
# shape, but belt-and-braces: never fall through to storage
# with ``user_id=""`` (matches legacy orphans) or ``None``
# (service escape).
return JSONResponse(_coordinator_metrics_payload(ws_id=ws_id))
now_epoch = time.time()
hour_ago_iso = datetime.fromtimestamp(now_epoch - 3600, tz=UTC).strftime("%Y-%m-%dT%H:%M:%S")
try:
state_counts = await asyncio.to_thread(
storage.count_workstreams_by_state,
parent_ws_id=ws_id,
user_id=filter_user_id,
user_id=None,
)
except Exception:
log.debug("coordinator_metrics.state_counts_failed ws=%s", ws_id[:8], exc_info=True)
@@ -3312,7 +3122,7 @@ async def coordinator_metrics(request: Request) -> JSONResponse:
storage.count_workstreams_since,
hour_ago_iso,
parent_ws_id=ws_id,
user_id=filter_user_id,
user_id=None,
)
except Exception:
log.debug(
turnstone/core/session_manager.py
+33 -24
View File
@@ -206,9 +206,11 @@ class SessionManager:
"""Construct a new workstream, persist, and register.
Slot reservation + placeholder install happen under the lock
(single-phase, ported from CoordinatorManager). Session
construction runs outside the lock; on failure the slot + row
are rolled back so capacity isn't leaked.
(single-phase). Session construction runs outside the lock; on
failure the in-memory slot is freed so capacity isn't leaked.
The storage row survives construction failure — the next
``open(ws_id)`` retries session construction rather than
forcing the user to create a brand-new workstream.
Raises ``RuntimeError`` when the manager is at capacity with
no idle workstream to evict — callers (HTTP handlers) translate
@@ -230,7 +232,7 @@ class SessionManager:
# Persist before session construction. Fail-closed: if the row
# can't be written, the in-memory session would be invisible to
# any lazy-rehydrate path and show up as "missing" after
# restart — better to surface the storage failure now.
# restart — surface the storage failure now.
try:
self._storage.register_workstream(
ws_id,
@@ -256,12 +258,13 @@ class SessionManager:
**extra_session_kwargs,
)
except Exception:
# Release the slot so capacity isn't leaked, and call
# cleanup_ui on the placeholder so any listener/lock state
# the UI factory allocated is released. Storage row stays:
# the next open() on this ws_id retries construction.
self._adapter.cleanup_ui(ws)
with self._lock:
self._remove_locked(ws_id)
try:
self._storage.delete_workstream(ws_id)
except Exception:
log.warning("session_mgr.rollback_delete_failed ws=%s", ws_id[:8], exc_info=True)
raise
self._adapter.emit_created(ws)
@@ -297,18 +300,14 @@ class SessionManager:
# open — lazy rehydrate for a persisted workstream
# ------------------------------------------------------------------
def open(
self,
ws_id: str,
*,
user_id: str,
admin: bool = False,
) -> Workstream | None:
def open(self, ws_id: str) -> Workstream | None:
"""Rehydrate a persisted workstream on demand.
Returns ``None`` when the row doesn't exist, doesn't match our
kind, is tombstoned (``state='deleted'``), or doesn't belong to
``user_id`` (non-admin callers only).
kind, or is tombstoned (``state='deleted'``). Turnstone is a
trusted-team tool — ownership is metadata for audit/display,
not an access boundary; HTTP handlers gate callers at the
scope level, not the row level.
Serializes concurrent opens of the same ws_id through a
per-ws refcounted lock so two GETs don't each construct a
@@ -326,15 +325,12 @@ class SessionManager:
if row is None or row.get("kind") != self.kind:
return None
# ``deleted`` is a tombstone — never resurrect.
# ``closed`` IS resurrectable: Saved Workstreams landing
# makes restore an explicit user action, and
# ``closed`` IS resurrectable; the Saved Workstreams
# landing makes restore an explicit user action, and
# ``_reserve_and_install_locked`` still enforces
# max_active (evicting an idle peer or raising).
if row.get("state") == "deleted":
return None
row_owner = row.get("user_id") or ""
if not admin and row_owner != user_id:
return None
with self._lock:
# Re-check fast path — another thread may have raced
@@ -344,7 +340,7 @@ class SessionManager:
return existing
ws, evicted = self._reserve_and_install_locked(
ws_id,
user_id=row_owner,
user_id=row.get("user_id") or "",
name=row.get("name") or f"ws-{ws_id[:4]}",
parent_ws_id=row.get("parent_ws_id"),
)
@@ -356,6 +352,9 @@ class SessionManager:
try:
ws.session = self._adapter.build_session(ws)
except Exception:
# Clean up the UI the adapter built before re-raising
# so any listener/lock resources are released.
self._adapter.cleanup_ui(ws)
with self._lock:
self._remove_locked(ws_id)
raise
@@ -571,7 +570,17 @@ class SessionManager:
ws.kind = self.kind
ws.user_id = user_id
ws.parent_ws_id = parent_ws_id if parent_ws_id else None
ws.ui = self._adapter.build_ui(ws)
try:
ws.ui = self._adapter.build_ui(ws)
except Exception:
# An IDLE peer may already have been popped above; if we
# propagate without unwinding, that peer leaks its session
# + worker + UI listeners and no ws_closed reaches
# subscribers.
if evicted is not None:
self._adapter.cleanup_ui(evicted)
self._adapter.emit_closed(evicted.id, reason="evicted")
raise
self._workstreams[ws_id] = ws
self._order.append(ws_id)
if self._active_id is None:
turnstone/server.py
+14 -39
View File
@@ -2965,9 +2965,6 @@ async def refresh_workstream_title(request: Request, ws_id: str = "") -> JSONRes
ws_id = request.path_params.get("ws_id", "")
log.info("ws.title.refresh_requested", ws_id=ws_id[:8] if ws_id else "empty")
mgr = request.app.state.workstreams
# Cross-tenant rename is a phishing / denial-of-use vector — a
# malicious caller could push the victim's title to a misleading
# string visible in list/dashboard responses.
_owner, err = _require_ws_access(request, ws_id, mgr=mgr)
if err:
return err
@@ -3003,7 +3000,6 @@ async def set_workstream_title(request: Request, ws_id: str = "") -> JSONRespons
if not ws_id:
return JSONResponse({"error": "ws_id is required"}, status_code=400)
mgr = request.app.state.workstreams
# Cross-tenant rename gate — same rationale as refresh-title above.
_owner, err = _require_ws_access(request, ws_id, mgr=mgr)
if err:
return err
@@ -3213,11 +3209,14 @@ def _require_ws_access(
*,
mgr: SessionManager | None = None,
) -> tuple[str, JSONResponse | None]:
"""Resolve ``ws_id`` to its owner after verifying the caller has access.
"""Resolve ``ws_id`` to its owner, 404-ing when the row doesn't exist.
Service-scoped tokens (internal callers) bypass ownership checks.
Returns ``(owner_user_id, None)`` on success. The owner id is what
attachments should be filed under.
Turnstone is a trusted-team tool: scope-level auth (e.g.
``admin.workstreams``) is the only gate; row-level ownership is not
enforced here. Returns ``(owner_user_id, None)`` on success — the
persisted owner id, which attachments should be filed under so
existing storage shape is preserved. Falls back to the caller's
own uid when the row has no recorded owner.
When ``mgr`` is provided and the workstream is live in the manager,
trust its cached ``user_id`` instead of round-tripping storage —
@@ -3228,18 +3227,11 @@ def _require_ws_access(
``mgr`` and fall through to the storage path.
"""
caller = _auth_user_id(request)
scopes = _auth_scopes(request)
is_service = "service" in scopes
if mgr is not None:
ws_mem = mgr.get(ws_id)
if ws_mem is not None:
owner_mem = ws_mem.user_id
if is_service:
return owner_mem or caller, None
if owner_mem and owner_mem != caller:
return "", JSONResponse({"error": "Workstream not found"}, status_code=404)
return caller, None
return ws_mem.user_id or caller, None
# Not in memory — fall through to storage so /delete etc.
# still resolve persisted-but-not-loaded rows.
@@ -3248,17 +3240,7 @@ def _require_ws_access(
owner = get_workstream_owner(ws_id)
if owner is None:
return "", JSONResponse({"error": "Workstream not found"}, status_code=404)
if is_service:
# Trust the service caller; file under its own user_id if no owner
# is set, otherwise under the existing owner.
return owner or caller, None
# Authenticated user must own the workstream. If the workstream was
# created before user tracking (owner blank) or by the same user, allow.
if owner and owner != caller:
# Return 404 (not 403) so non-owners cannot enumerate workstream
# existence by response code.
return "", JSONResponse({"error": "Workstream not found"}, status_code=404)
return caller, None
return owner or caller, None
async def upload_attachment(request: Request) -> JSONResponse:
@@ -3489,20 +3471,13 @@ async def open_workstream(request: Request) -> JSONResponse:
)
uid: str = _auth_user_id(request)
scopes = _auth_scopes(request)
# Ownership gate: the stored owner must match the caller (or the
# caller must hold service scope — used by the console routing proxy
# and cluster rehydration paths). Ownerless persisted rows (the
# startup ``name="default"`` workstream, pre-migration legacy rows)
# are claimable by any authenticated caller — consistent with the
# trusted-team visibility model the listing endpoints assume, and
# symmetric with how _require_ws_access handles the same rows on
# the per-workstream interactive handlers. 404 (not 403) so
# existence isn't enumerable by non-owners.
# Trusted-team model: scope-level auth (already enforced by the
# middleware) is the only gate. ``user_id`` is metadata for audit
# + display, not an access boundary, so any authenticated caller
# can rehydrate any persisted workstream. Fall back to the
# caller's uid when the row has no recorded owner.
stored_owner = (ws_row.get("user_id") or "").strip()
if stored_owner and "service" not in scopes and stored_owner != uid:
return JSONResponse({"error": "Workstream not found"}, status_code=404)
owner_uid = stored_owner or uid
try: