From fbfd170ca6e66892980428ce556c93b25aee5d52 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Wed, 1 Jul 2026 15:36:42 -0700 Subject: [PATCH] feat(projects): enforce private-project workstream visibility server-side MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Workstreams attached to a private project were listed and reachable for every authenticated user — only the scope tier was checked. Add a tenancy predicate (WorkstreamProjectVisibility: private → project owner/members, the workstream's own creator, service scope, or admin.cluster.inspect; public/dangling/no project → unchanged trusted-team visibility; membership itself is the grant — deliberately NOT gated on the project.read capability, which guards the management API) and apply it at every surface: - listings: saved sessions (project_id + owner tail-appended to list_workstreams_with_history on both backends), active list, node dashboard, console cluster list (pre-pagination via a collector row_filter so totals stay honest), node detail - console tier-1 SSE: per-connection snapshot filtering + a hidden-set for sparse follow-up events; ws_created project lookups run on the executor, membership changes take effect on reconnect - row access: resolve_workstream_owner 403s private-project rows for non-members, covering every interactive ws-scoped verb via tenant_check (console coordinator lane stays on its privileged admin.coordinator gate) - create: ensure_project_attachable gates explicit and parent-inherited project_id on both create validators (unknown project 400s instead of minting a dangling link) --- tests/test_project_workstream_visibility.py | 294 ++++++++++++++++++++ tests/test_saved_handler_unified.py | 11 +- turnstone/console/collector.py | 9 + turnstone/console/server.py | 90 +++++- turnstone/core/auth.py | 150 ++++++++++ turnstone/core/memory.py | 14 + turnstone/core/session_routes.py | 33 ++- turnstone/core/storage/_postgresql.py | 2 +- turnstone/core/storage/_sqlite.py | 2 +- turnstone/core/web_helpers.py | 45 ++- turnstone/server.py | 27 +- 11 files changed, 650 insertions(+), 27 deletions(-) create mode 100644 tests/test_project_workstream_visibility.py diff --git a/tests/test_project_workstream_visibility.py b/tests/test_project_workstream_visibility.py new file mode 100644 index 00000000..b93d62c3 --- /dev/null +++ b/tests/test_project_workstream_visibility.py @@ -0,0 +1,294 @@ +"""Private-project workstream visibility enforcement. + +Covers the tenancy predicate (:class:`WorkstreamProjectVisibility`), the +create-time attach gate (:func:`ensure_project_attachable`), the row-access +gate in :func:`resolve_workstream_owner`, and the saved-list filter in +``_collect_saved_rows`` — the choke points that keep workstreams attached +to a private project out of non-members' listings and 403 their direct +access. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from turnstone.core.auth import ( + WorkstreamProjectVisibility, + ensure_project_attachable, +) + +pytestmark = pytest.mark.anyio + + +def _fake_storage( + *, + visibility: str = "private", + owner: str = "alice", + members: tuple[str, ...] = (), + missing: bool = False, +) -> MagicMock: + storage = MagicMock() + if missing: + storage.get_project.return_value = None + else: + storage.get_project.return_value = { + "project_id": "p1", + "name": "P1", + "owner_id": owner, + "visibility": visibility, + "state": "active", + } + storage.is_project_member.side_effect = lambda pid, uid: uid in members + return storage + + +class _FakeAuth: + def __init__( + self, + user_id: str, + scopes: tuple[str, ...] = (), + permissions: tuple[str, ...] = (), + ) -> None: + self.user_id = user_id + self._scopes = set(scopes) + self._permissions = set(permissions) + + def has_scope(self, scope: str) -> bool: + return scope in self._scopes + + def has_permission(self, permission: str) -> bool: + return permission in self._permissions + + +def _request_for( + uid: str, + scopes: tuple[str, ...] = (), + permissions: tuple[str, ...] = (), +) -> Any: + return SimpleNamespace(state=SimpleNamespace(auth_result=_FakeAuth(uid, scopes, permissions))) + + +class TestWsVisiblePredicate: + def test_no_project_always_visible(self) -> None: + vis = WorkstreamProjectVisibility("bob", storage=_fake_storage()) + assert vis.ws_visible(None) + assert vis.ws_visible("") + + def test_dangling_project_visible(self) -> None: + # Project deletion leaves ws links behind — no row, no privacy. + vis = WorkstreamProjectVisibility("bob", storage=_fake_storage(missing=True)) + assert vis.ws_visible("p1") + + def test_public_project_visible_to_anyone(self) -> None: + vis = WorkstreamProjectVisibility("bob", storage=_fake_storage(visibility="public")) + assert vis.ws_visible("p1") + + def test_private_hidden_from_non_member(self) -> None: + vis = WorkstreamProjectVisibility("bob", storage=_fake_storage()) + assert not vis.ws_visible("p1") + + def test_private_visible_to_project_owner(self) -> None: + vis = WorkstreamProjectVisibility("alice", storage=_fake_storage()) + assert vis.ws_visible("p1") + + def test_private_visible_to_member(self) -> None: + vis = WorkstreamProjectVisibility("bob", storage=_fake_storage(members=("bob",))) + assert vis.ws_visible("p1") + + def test_private_visible_to_ws_creator(self) -> None: + # A workstream's own creator never loses sight of it, even after + # a membership revoke leaves a legacy private-project link. + vis = WorkstreamProjectVisibility("bob", storage=_fake_storage()) + assert vis.ws_visible("p1", ws_owner="bob") + + def test_private_hidden_from_anonymous(self) -> None: + vis = WorkstreamProjectVisibility("", storage=_fake_storage()) + assert not vis.ws_visible("p1") + + def test_bypass_sees_everything(self) -> None: + vis = WorkstreamProjectVisibility("bob", bypass=True, storage=_fake_storage()) + assert vis.ws_visible("p1") + + def test_storage_error_fails_closed(self) -> None: + storage = MagicMock() + storage.get_project.side_effect = RuntimeError("db down") + vis = WorkstreamProjectVisibility("bob", storage=storage) + assert not vis.ws_visible("p1") + + def test_project_rows_memoized(self) -> None: + storage = _fake_storage(visibility="public") + vis = WorkstreamProjectVisibility("bob", storage=storage) + assert vis.ws_visible("p1") + assert vis.ws_visible("p1") + assert storage.get_project.call_count == 1 + + def test_for_request_bypass_rules(self) -> None: + assert WorkstreamProjectVisibility.for_request( + _request_for("bob", scopes=("service",)) + )._bypass + assert WorkstreamProjectVisibility.for_request( + _request_for("bob", permissions=("admin.cluster.inspect",)) + )._bypass + assert not WorkstreamProjectVisibility.for_request(_request_for("bob"))._bypass + + +class TestEnsureProjectAttachable: + def test_no_project_allowed(self) -> None: + assert ensure_project_attachable("bob", "", storage=_fake_storage()) is None + + def test_unknown_project_is_400(self) -> None: + denied = ensure_project_attachable("bob", "p1", storage=_fake_storage(missing=True)) + assert denied is not None and denied[0] == 400 + + def test_public_project_allowed(self) -> None: + assert ( + ensure_project_attachable("bob", "p1", storage=_fake_storage(visibility="public")) + is None + ) + + def test_private_member_and_owner_allowed(self) -> None: + assert ( + ensure_project_attachable("bob", "p1", storage=_fake_storage(members=("bob",))) is None + ) + assert ensure_project_attachable("alice", "p1", storage=_fake_storage()) is None + + def test_private_non_member_is_403(self) -> None: + denied = ensure_project_attachable("bob", "p1", storage=_fake_storage()) + assert denied is not None and denied[0] == 403 + + def test_anonymous_private_is_403(self) -> None: + denied = ensure_project_attachable("", "p1", storage=_fake_storage()) + assert denied is not None and denied[0] == 403 + + def test_storage_error_fails_closed(self) -> None: + storage = MagicMock() + storage.get_project.side_effect = RuntimeError("db down") + denied = ensure_project_attachable("bob", "p1", storage=storage) + assert denied is not None and denied[0] == 403 + + +class TestResolveWorkstreamOwnerProjectGate: + """Integration against the real (ephemeral) storage: the row-access + gate every interactive ws-scoped verb inherits via tenant_check.""" + + def _seed(self, *, member: bool) -> None: + from turnstone.core.memory import register_workstream + from turnstone.core.storage import get_storage + + storage = get_storage() + storage.create_project("p1", "Secret", "alice") + if member: + storage.add_project_member("p1", "bob") + register_workstream("ws-priv", user_id="alice", project_id="p1") + + def test_non_member_gets_403(self, tmp_db: str) -> None: + from turnstone.core.web_helpers import resolve_workstream_owner + + self._seed(member=False) + owner, err = resolve_workstream_owner(_request_for("bob"), "ws-priv") + assert err is not None and err.status_code == 403 + + def test_member_resolves_owner(self, tmp_db: str) -> None: + from turnstone.core.web_helpers import resolve_workstream_owner + + self._seed(member=True) + owner, err = resolve_workstream_owner(_request_for("bob"), "ws-priv") + assert err is None + assert owner == "alice" + + def test_ws_creator_bypasses(self, tmp_db: str) -> None: + from turnstone.core.memory import register_workstream + from turnstone.core.storage import get_storage + from turnstone.core.web_helpers import resolve_workstream_owner + + storage = get_storage() + storage.create_project("p1", "Secret", "alice") + # bob created a ws in alice's private project, then lost access — + # bob still reaches his own workstream. + register_workstream("ws-bob", user_id="bob", project_id="p1") + owner, err = resolve_workstream_owner(_request_for("bob"), "ws-bob") + assert err is None + assert owner == "bob" + + def test_admin_inspect_bypasses(self, tmp_db: str) -> None: + from turnstone.core.web_helpers import resolve_workstream_owner + + self._seed(member=False) + owner, err = resolve_workstream_owner( + _request_for("bob", permissions=("admin.cluster.inspect",)), "ws-priv" + ) + assert err is None + assert owner == "alice" + + def test_missing_ws_still_404s(self, tmp_db: str) -> None: + from turnstone.core.web_helpers import resolve_workstream_owner + + owner, err = resolve_workstream_owner(_request_for("bob"), "nope") + assert err is not None and err.status_code == 404 + + def test_public_project_ws_resolves(self, tmp_db: str) -> None: + from turnstone.core.memory import register_workstream + from turnstone.core.storage import get_storage + from turnstone.core.web_helpers import resolve_workstream_owner + + storage = get_storage() + storage.create_project("p1", "Open", "alice") + storage.update_project("p1", visibility="public") + register_workstream("ws-pub", user_id="alice", project_id="p1") + owner, err = resolve_workstream_owner(_request_for("bob"), "ws-pub") + assert err is None + assert owner == "alice" + + +class TestSavedListFilter: + """The saved-sessions collector drops private-project rows server-side + and carries project_id on surviving rows (real ephemeral DB).""" + + async def test_saved_rows_filtered_and_carry_project_id(self, tmp_db: str) -> None: + from turnstone.core.memory import register_workstream, save_message + from turnstone.core.session_routes import ( + SessionEndpointConfig, + _collect_saved_rows, + ) + from turnstone.core.storage import get_storage + from turnstone.core.workstream import WorkstreamKind + + storage = get_storage() + storage.create_project("p1", "Secret", "alice") + storage.create_project("p2", "Open", "alice") + storage.update_project("p2", visibility="public") + + register_workstream("ws-plain", user_id="alice") + register_workstream("ws-priv", user_id="alice", project_id="p1") + register_workstream("ws-pub", user_id="alice", project_id="p2") + register_workstream("ws-own", user_id="bob", project_id="p1") + for wid in ("ws-plain", "ws-priv", "ws-pub", "ws-own"): + save_message(wid, "user", "hello") + + cfg = SessionEndpointConfig( + permission_gate=None, + manager_lookup=lambda request: (None, None), + tenant_check=None, + not_found_label="Workstream not found", + audit_action_prefix="workstream", + list_kind=WorkstreamKind.INTERACTIVE, + saved_state_filter=None, + saved_loaded_lookup=None, + ) + + rows = await _collect_saved_rows(cfg, _request_for("bob")) + ids = {r["ws_id"] for r in rows} + # bob: no membership in p1 — alice's private ws is dropped; the + # public-project ws, the project-less ws, and bob's own + # private-project ws all survive. + assert ids == {"ws-plain", "ws-pub", "ws-own"} + by_id = {r["ws_id"]: r for r in rows} + assert by_id["ws-pub"]["project_id"] == "p2" + assert by_id["ws-plain"]["project_id"] is None + + rows_alice = await _collect_saved_rows(cfg, _request_for("alice")) + assert {r["ws_id"] for r in rows_alice} == {"ws-plain", "ws-priv", "ws-pub", "ws-own"} diff --git a/tests/test_saved_handler_unified.py b/tests/test_saved_handler_unified.py index 2423e043..02d059d5 100644 --- a/tests/test_saved_handler_unified.py +++ b/tests/test_saved_handler_unified.py @@ -6,7 +6,7 @@ for its L-shell dashboard, plus a regression guard for the single-kind :func:`turnstone.core.session_routes._collect_saved_rows`. Storage is mocked (``list_workstreams_with_history`` is patched to -return synthetic 15-tuples) — no real or dev database is touched. The +return synthetic 17-tuples) — no real or dev database is touched. The request is a :class:`unittest.mock.MagicMock`, matching how the body-level coordinator endpoint tests build request stubs; the saved path only reads ``request`` to pass it to ``saved_loaded_lookup`` / @@ -41,7 +41,7 @@ pytestmark = pytest.mark.anyio # Column order from list_workstreams_with_history (keep in sync with the # storage SELECT): ws_id, alias, title, name, created, updated, # message_count, node_id, state, kind, model_alias, launch_skill, -# child_count, context_tokens, context_window. +# child_count, context_tokens, context_window, project_id, owner. def _row( ws_id: str, *, @@ -49,8 +49,10 @@ def _row( kind: str, state: str = "closed", name: str | None = None, + project_id: str | None = None, + owner: str | None = None, ) -> tuple[Any, ...]: - """Build a synthetic storage row (15-tuple) for one workstream.""" + """Build a synthetic storage row (17-tuple) for one workstream.""" return ( ws_id, None, # alias @@ -67,6 +69,8 @@ def _row( 0, # child_count 1000, # context_tokens 4000, # context_window + project_id, # project_id + owner, # owner user_id ) @@ -343,6 +347,7 @@ async def test_single_kind_saved_unchanged(monkeypatch: pytest.MonkeyPatch) -> N "child_count", "context_tokens", "context_ratio", + "project_id", } diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index e1d77f43..80122882 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -932,6 +932,7 @@ class ClusterCollector: page: int = 1, per_page: int = 50, extra_rows: list[dict[str, Any]] | None = None, + row_filter: Callable[[dict[str, Any]], bool] | None = None, ) -> tuple[list[dict[str, Any]], int]: """Return filtered, sorted, paginated workstreams + total count. @@ -939,6 +940,10 @@ class ClusterCollector: filter / sort / paginate — used by callers that contribute console-local rows (e.g. coordinator workstreams) that aren't tracked on any node's SSE stream. + + ``row_filter`` (when provided) runs against the merged, + UNPAGINATED pool so dropped rows never skew ``total`` or page + boundaries — the private-project tenancy filter rides here. """ with self._lock: all_ws = [] @@ -955,6 +960,10 @@ class ClusterCollector: if extra_rows: all_ws.extend(dict(r) for r in extra_rows) + # Row-level tenancy filter first — before pagination math. + if row_filter is not None: + all_ws = [ws for ws in all_ws if row_filter(ws)] + # Filter if state: all_ws = [ws for ws in all_ws if ws.get("state") == state] diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 01edd7ae..1c155f58 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -995,6 +995,8 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]: async def cluster_workstreams(request: Request) -> JSONResponse: + from turnstone.core.auth import WorkstreamProjectVisibility + collector: ClusterCollector = request.app.state.collector params = dict(request.query_params) state = params.get("state") @@ -1004,6 +1006,7 @@ async def cluster_workstreams(request: Request) -> JSONResponse: page = _parse_int(params, "page", 1, minimum=1) per_page = _parse_int(params, "per_page", 50, minimum=1, maximum=200) extra_rows = _coordinator_rows(request) + visibility = WorkstreamProjectVisibility.for_request(request) ws_list, total = collector.get_workstreams( state=state, node=node, @@ -1012,6 +1015,9 @@ async def cluster_workstreams(request: Request) -> JSONResponse: page=page, per_page=per_page, extra_rows=extra_rows, + row_filter=lambda ws: visibility.ws_visible( + ws.get("project_id") or "", ws_owner=ws.get("user_id") or "" + ), ) pages = math.ceil(total / per_page) if per_page > 0 else 0 return JSONResponse( @@ -1506,6 +1512,8 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse: async def cluster_node_detail(request: Request) -> JSONResponse: + from turnstone.core.auth import WorkstreamProjectVisibility + collector: ClusterCollector = request.app.state.collector node_id = request.path_params["node_id"] nv = _validate_node_id(node_id) @@ -1514,6 +1522,13 @@ async def cluster_node_detail(request: Request) -> JSONResponse: detail = collector.get_node_detail(node_id) if not detail: return JSONResponse({"error": "Node not found"}, status_code=404) + # Private-project tenancy — same predicate as the cluster list. + visibility = WorkstreamProjectVisibility.for_request(request) + detail["workstreams"] = [ + ws + for ws in detail.get("workstreams", []) + if visibility.ws_visible(ws.get("project_id") or "", ws_owner=ws.get("user_id") or "") + ] # Attach metadata if available import json as _nd_json @@ -1564,12 +1579,61 @@ async def cluster_snapshot(request: Request) -> JSONResponse: async def cluster_events_sse(request: Request) -> Response: + from turnstone.core.auth import WorkstreamProjectVisibility + err = _collector_scope_error(request) if err is not None: return err collector: ClusterCollector = request.app.state.collector client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=2000) + # Per-connection private-project tenancy. The snapshot carries full + # rows (project_id + user_id); follow-up events are sparse (usually + # just ws_id), so invisible workstreams are recorded in ``hidden`` + # at snapshot/ws_created time and every later event naming them is + # swallowed. Access changes (membership grant/revoke) take effect + # on reconnect — same resolve-once precedent as session + # construction. The visibility instance memoizes project rows, so + # the steady-state per-event cost is a set lookup. + visibility = WorkstreamProjectVisibility.for_request(request) + hidden: set[str] = set() + + def _row_ws_id(ws: dict[str, Any]) -> str: + return str(ws.get("ws_id") or ws.get("id") or "") + + def _filter_snapshot(snap: dict[str, Any]) -> dict[str, Any]: + for node in snap.get("nodes", []): + kept: list[dict[str, Any]] = [] + for ws in node.get("workstreams", []): + if visibility.ws_visible( + ws.get("project_id") or "", ws_owner=ws.get("user_id") or "" + ): + kept.append(ws) + else: + wid = _row_ws_id(ws) + if wid: + hidden.add(wid) + node["workstreams"] = kept + return snap + + def _event_visible(event: dict[str, Any]) -> bool: + etype = event.get("type") + wid = str(event.get("ws_id") or "") + if etype == "ws_created": + if not visibility.ws_visible( + event.get("project_id") or "", ws_owner=event.get("user_id") or "" + ): + if wid: + hidden.add(wid) + return False + hidden.discard(wid) + return True + if wid and wid in hidden: + if etype == "ws_closed": + hidden.discard(wid) + return False + return True + async def event_generator() -> AsyncGenerator[dict[str, str], None]: loop = asyncio.get_running_loop() try: @@ -1578,6 +1642,9 @@ async def cluster_events_sse(request: Request) -> Response: None, collector.get_snapshot_and_register, client_queue ) snap["type"] = "snapshot" + # Executor: the snapshot filter resolves project rows from + # storage — never block the event loop on DB I/O. + snap = await loop.run_in_executor(None, _filter_snapshot, snap) yield {"data": json.dumps(snap)} while True: @@ -1585,7 +1652,15 @@ async def cluster_events_sse(request: Request) -> Response: event = await loop.run_in_executor( None, functools.partial(client_queue.get, timeout=5) ) - yield {"data": json.dumps(event)} + # Only ws_created can touch storage (project lookup); + # every other event type is a pure set-membership + # check and stays on the loop. + if event.get("type") == "ws_created": + visible = await loop.run_in_executor(None, _event_visible, event) + else: + visible = _event_visible(event) + if visible: + yield {"data": json.dumps(event)} except queue.Empty: pass # poll timeout, retry if await request.is_disconnected(): @@ -3393,6 +3468,19 @@ async def _coord_create_validate_request( """ if not uid: return JSONResponse({"error": "authentication required"}, status_code=401) + # Project attach gate — same rule as the interactive validator: a + # private project accepts new workstreams only from its owner or + # members, and a nonexistent project_id 400s rather than minting a + # dangling link. + project_raw = body.get("project_id") + attach_pid = (project_raw.strip() if isinstance(project_raw, str) else "") or "" + if attach_pid: + from turnstone.core.auth import ensure_project_attachable + + denied = ensure_project_attachable(uid, attach_pid) + if denied is not None: + status, message = denied + return JSONResponse({"error": message}, status_code=status) return None diff --git a/turnstone/core/auth.py b/turnstone/core/auth.py index 5f342d76..053d2378 100644 --- a/turnstone/core/auth.py +++ b/turnstone/core/auth.py @@ -253,6 +253,156 @@ def user_can_access_project( return acc.can_write if write else acc.can_read +class WorkstreamProjectVisibility: + """Per-request memoized visibility predicate for project-scoped workstreams. + + Answers "may *user_id* see a workstream attached to *project_id*?" for + listing filters and the row-access gate. Distinct from + :func:`user_can_access_project` on purpose: that composes the RBAC + capability (``project.read``, admin-default) with the ACL and gates the + project *management* surfaces, whereas workstream visibility is a + tenancy question — an explicit ``project_members`` row (or ownership) + IS the grant, no capability required, or members without ``project.read`` + would lose sight of their own shared workstreams. + + Rules (first match wins): + + * ``bypass`` instances see everything — service-scope callers (the + collector and other cluster machinery must never be blinded at the + node edge; user-facing filtering happens at the console edge) and + holders of ``admin.cluster.inspect`` (the existing cluster-wide + workstream-inspect surface). + * No / dangling ``project_id`` → visible (a deleted project leaves the + link behind by design — no row, no privacy to enforce). + * Non-private visibility → visible (trusted-team default). + * Private → the workstream's own creator, the project owner, or a + project member; anonymous callers fail closed. + * A storage failure while resolving a project fails closed (treated + as private-and-not-a-member) rather than leaking on a blip. + + Project rows and membership verdicts are memoized per instance — + construct one per request/connection, not per row. + """ + + def __init__(self, user_id: str, *, bypass: bool = False, storage: Any = None) -> None: + self._user_id = user_id or "" + self._bypass = bypass + self._storage = storage + self._projects: dict[str, dict[str, Any] | None] = {} + self._member: dict[str, bool] = {} + + @classmethod + def for_request(cls, request: Any, *, storage: Any = None) -> WorkstreamProjectVisibility: + """Build a filter for an HTTP request's authenticated principal. + + Service-scoped tokens and ``admin.cluster.inspect`` holders get a + bypass instance; everyone else filters as themselves. + """ + auth: AuthResult | None = getattr(getattr(request, "state", None), "auth_result", None) + uid = str(getattr(auth, "user_id", "") or "") + bypass = bool( + auth is not None + and (auth.has_scope("service") or auth.has_permission("admin.cluster.inspect")) + ) + return cls(uid, bypass=bypass, storage=storage) + + def _resolve_storage(self) -> Any: + if self._storage is None: + from turnstone.core.storage._registry import get_storage + + self._storage = get_storage() + return self._storage + + def _project(self, project_id: str) -> dict[str, Any] | None: + if project_id not in self._projects: + storage = self._resolve_storage() + if storage is None: + raise RuntimeError("storage unavailable for project visibility check") + self._projects[project_id] = storage.get_project(project_id) + return self._projects[project_id] + + def _is_member(self, project_id: str) -> bool: + if project_id not in self._member: + storage = self._resolve_storage() + self._member[project_id] = bool( + storage is not None and storage.is_project_member(project_id, self._user_id) + ) + return self._member[project_id] + + def ws_visible(self, project_id: str | None, ws_owner: str = "") -> bool: + """Apply the class rules to one workstream row.""" + if self._bypass: + return True + pid = (project_id or "").strip() + if not pid: + return True + if ws_owner and ws_owner == self._user_id: + return True + try: + project = self._project(pid) + if project is None: + return True + if (project.get("visibility") or "private") != "private": + return True + if not self._user_id: + return False + if project.get("owner_id") == self._user_id: + return True + return self._is_member(pid) + except Exception: + log.warning( + "workstream project-visibility check failed user=%s project=%s — failing closed", + self._user_id, + pid, + ) + return False + + +def ensure_project_attachable( + user_id: str, + project_id: str, + *, + storage: Any = None, +) -> tuple[int, str] | None: + """Gate attaching a NEW workstream to *project_id* at create time. + + Returns ``None`` when the attach is allowed, else ``(status_code, + message)`` for the handler to surface. Stricter than + :meth:`WorkstreamProjectVisibility.ws_visible` in one way — a + nonexistent project is a 400 (a dangling link on an EXISTING row is + tolerated because project deletion leaves links behind, but minting + a fresh dangling link is a caller error) — and shares its tenancy + rule: private projects accept workstreams only from their owner or + members; public/active-or-archived projects accept from anyone + (memory writes stay member-gated at the session layer). + + Fail-closed: empty ``user_id`` or a storage failure denies with 403. + """ + pid = (project_id or "").strip() + if not pid: + return None + if storage is None: + from turnstone.core.storage._registry import get_storage + + storage = get_storage() + if storage is None: + return (403, "project access could not be verified") + try: + project = storage.get_project(pid) + if project is None: + return (400, "unknown project_id") + if (project.get("visibility") or "private") != "private": + return None + if user_id and ( + project.get("owner_id") == user_id or bool(storage.is_project_member(pid, user_id)) + ): + return None + return (403, "cannot attach a workstream to a private project you don't belong to") + except Exception: + log.warning("project attach check failed user=%s project=%s — failing closed", user_id, pid) + return (403, "project access could not be verified") + + def _permissions_to_scopes(permissions: set[str]) -> frozenset[str]: """Derive legacy scopes from a granular permission set.""" scopes: set[str] = set() diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index 7e9f6720..ffaeb48f 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -614,6 +614,20 @@ def get_workstream_owner(ws_id: str) -> str | None: return None +def get_workstream_row(ws_id: str) -> dict[str, Any] | None: + """Return the full workstreams row dict, or None when missing/unreadable. + + Same fail-soft shape as :func:`get_workstream_owner` — access gates + treat ``None`` as not-found, so a storage blip degrades to a 404 + rather than a 500. + """ + try: + return get_storage().get_workstream(ws_id) + except Exception: + log.warning("Failed to get workstream row ws=%s", ws_id, exc_info=True) + return None + + def update_workstream_title(ws_id: str, title: str) -> None: """Set or update the auto-generated title for a workstream.""" try: diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index a4e56ace..544d5701 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -2646,12 +2646,20 @@ def make_list_handler(cfg: SessionEndpointConfig) -> Handler: resolve_titles = cfg.list_resolve_titles def _build_rows() -> list[dict[str, Any]]: + from turnstone.core.auth import WorkstreamProjectVisibility + + visibility = WorkstreamProjectVisibility.for_request(request) wss = mgr.list_all() titles: dict[str, str | None] = {} if resolve_titles is not None and wss: titles = resolve_titles([ws.id for ws in wss]) rows: list[dict[str, Any]] = [] for ws in wss: + project_id = getattr(ws, "project_id", "") or "" + # Private-project tenancy — drop rows the requester may + # not see (same predicate as the saved list). + if not visibility.ws_visible(project_id, ws_owner=ws.user_id or ""): + continue title = titles.get(ws.id) or ws.name rows.append( { @@ -2661,6 +2669,7 @@ def make_list_handler(cfg: SessionEndpointConfig) -> Handler: "kind": ws.kind, "parent_ws_id": ws.parent_ws_id, "user_id": ws.user_id, + "project_id": project_id or None, } ) return rows @@ -2720,12 +2729,17 @@ async def _collect_saved_rows( # Column order from list_workstreams_with_history (keep in sync with # the storage SELECT): ws_id, alias, title, name, created, updated, # message_count, node_id, state, kind, model_alias, launch_skill, - # child_count, context_tokens, context_window. The occupancy ratio - # is derived here (Python float division) rather than in SQL so the - # NULL / zero-window cases stay obvious and identical across backends. - # context_window is NULL for model aliases absent from - # model_definitions (e.g. config.toml-only models), so context_ratio - # degrades to 0.0 there rather than reporting a bogus occupancy. + # child_count, context_tokens, context_window, project_id, owner. + # The occupancy ratio is derived here (Python float division) rather + # than in SQL so the NULL / zero-window cases stay obvious and + # identical across backends. context_window is NULL for model + # aliases absent from model_definitions (e.g. config.toml-only + # models), so context_ratio degrades to 0.0 there rather than + # reporting a bogus occupancy. + from turnstone.core.auth import WorkstreamProjectVisibility + + visibility = WorkstreamProjectVisibility.for_request(request) + result: list[dict[str, Any]] = [] for row in rows: ( @@ -2744,9 +2758,15 @@ async def _collect_saved_rows( child_count, context_tokens, context_window, + project_id, + owner, ) = row if wid in loaded: continue + # Private-project tenancy: rows the requester may not see are + # dropped server-side, not hidden client-side. + if not visibility.ws_visible(project_id, ws_owner=owner or ""): + continue ctx_tokens = context_tokens or 0 context_ratio = ( round(ctx_tokens / context_window, 3) if ctx_tokens and context_window else 0.0 @@ -2768,6 +2788,7 @@ async def _collect_saved_rows( "child_count": child_count or 0, "context_tokens": ctx_tokens, "context_ratio": context_ratio, + "project_id": project_id or None, } ) return result diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index 5ff78b08..d5505832 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -622,7 +622,7 @@ class PostgreSQLBackend: "(SELECT ue.prompt_tokens FROM usage_events ue " " WHERE ue.ws_id = w.ws_id " " ORDER BY ue.timestamp DESC LIMIT 1), " - "md.context_window " + "md.context_window, w.project_id, w.user_id " "FROM workstreams w " "LEFT JOIN workstream_config wcm " " ON wcm.ws_id = w.ws_id AND wcm.key = 'model_alias' " diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 0890f50a..7135a989 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -726,7 +726,7 @@ class SQLiteBackend: "(SELECT ue.prompt_tokens FROM usage_events ue " " WHERE ue.ws_id = w.ws_id " " ORDER BY ue.timestamp DESC LIMIT 1), " - "md.context_window " + "md.context_window, w.project_id, w.user_id " "FROM workstreams w " "LEFT JOIN workstream_config wcm " " ON wcm.ws_id = w.ws_id AND wcm.key = 'model_alias' " diff --git a/turnstone/core/web_helpers.py b/turnstone/core/web_helpers.py index d3b591e2..f8b67fb2 100644 --- a/turnstone/core/web_helpers.py +++ b/turnstone/core/web_helpers.py @@ -316,17 +316,22 @@ def resolve_workstream_owner( """Resolve ``ws_id`` to its owner; 404 when the row doesn't exist. Turnstone is a trusted-team tool: scope-level auth (e.g. - ``admin.workstreams`` / ``admin.coordinator``) is the only gate; - row-level ownership is not enforced here. Returns + ``admin.coordinator``) is the primary gate and row-level OWNERSHIP + is still not enforced here. The one row-level check this performs + is PROJECT tenancy: a workstream attached to a *private* project is + only reachable by the project's owner/members, the workstream's own + creator, service-scope callers, and ``admin.cluster.inspect`` + holders — everyone else gets a 403 (see + :class:`turnstone.core.auth.WorkstreamProjectVisibility`). 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 memory, - trust its cached ``user_id`` instead of round-tripping storage — - keeps in-memory-only handlers functional during transient DB - outages and trims the hot-path by one query. + trust its cached ``user_id`` / ``project_id`` instead of + round-tripping storage — keeps in-memory-only handlers functional + during transient DB outages and trims the hot-path by one query. ``not_found_label`` is the message body the 404 carries — the interactive surface uses "Workstream not found"; coord uses @@ -334,20 +339,36 @@ def resolve_workstream_owner( """ from starlette.responses import JSONResponse as _JSONResponse + from turnstone.core.auth import WorkstreamProjectVisibility + caller = auth_user_id(request) + owner: str | None = None + project_id = "" if mgr is not None: ws_mem = mgr.get(ws_id) if ws_mem is not None: - return ws_mem.user_id or caller, None - # Not in memory — fall through to storage so persisted-but-not- - # loaded rows still resolve. + owner = ws_mem.user_id or "" + project_id = getattr(ws_mem, "project_id", "") or "" - from turnstone.core.memory import get_workstream_owner - - owner = get_workstream_owner(ws_id) if owner is None: - return "", _JSONResponse({"error": not_found_label}, status_code=404) + # Not in memory — storage resolves persisted-but-not-loaded rows. + from turnstone.core.memory import get_workstream_row + + row = get_workstream_row(ws_id) + if row is None: + return "", _JSONResponse({"error": not_found_label}, status_code=404) + owner = row.get("user_id") or "" + project_id = row.get("project_id") or "" + + if project_id: + visibility = WorkstreamProjectVisibility.for_request(request) + if not visibility.ws_visible(project_id, ws_owner=owner): + return "", _JSONResponse( + {"error": "Forbidden: workstream belongs to a private project"}, + status_code=403, + ) + return owner or caller, None diff --git a/turnstone/server.py b/turnstone/server.py index f0d06e41..5d0ad298 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -1065,12 +1065,19 @@ async def global_events_sse(request: Request) -> Response: async def dashboard(request: Request) -> JSONResponse: """GET /v1/api/dashboard — enriched workstream data + aggregate stats.""" + from turnstone.core.auth import WorkstreamProjectVisibility from turnstone.core.memory import get_workstream_display_name mgr: SessionManager = request.app.state.workstreams - # No per-user filter — see list_workstreams above for the rationale - # (trusted-team deployment shape; mutations stay owner-gated). - wss = mgr.list_all() + # No per-user OWNER filter (trusted-team deployment shape) — but + # private-project rows are dropped for non-members, same predicate + # as every other listing surface. + visibility = WorkstreamProjectVisibility.for_request(request) + wss = [ + ws + for ws in mgr.list_all() + if visibility.ws_visible(getattr(ws, "project_id", "") or "", ws_owner=ws.user_id or "") + ] total_tokens = 0 total_tool_calls = 0 active_count = 0 @@ -1936,6 +1943,20 @@ async def _interactive_create_validate_request( # (which forwards the body verbatim). if not (body.get("project_id") or "") and parent_row.get("project_id"): body["project_id"] = parent_row.get("project_id") + # Project attach gate (explicit or parent-inherited): a private + # project accepts new workstreams only from its owner/members, and a + # nonexistent project_id is a caller error rather than a silent + # dangling link. Re-checking the inherited value is deliberate — a + # coordinator owner whose membership was revoked fails the child + # spawn loudly here instead of minting rows they can no longer see. + attach_pid = str(body.get("project_id") or "") + if attach_pid: + from turnstone.core.auth import ensure_project_attachable + + denied = ensure_project_attachable(uid, attach_pid) + if denied is not None: + status, message = denied + return JSONResponse({"error": message}, status_code=status) notify_targets_raw = body.get("notify_targets", "[]") if isinstance(notify_targets_raw, list): notify_targets_raw = json.dumps(notify_targets_raw)