From a7ecd76e0afcdf5109b5af4ba58d7c7f6f4c469a Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Wed, 13 May 2026 17:40:56 +0800 Subject: [PATCH] =?UTF-8?q?test(boundary):=20PR6=20T6.8=20=E2=80=94=20cros?= =?UTF-8?q?s-workspace=20isolation=20e2e=20(4=20cases)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end coverage for the workspace boundary: - GET /api/threads/{tid} from workspace B → 404 - DELETE /api/threads/{tid} from workspace B → 404 - PATCH /api/threads/{tid} from workspace B → 404 - Positive control: same workspace → 200 Wires `MemoryThreadMetaStore` (real impl, not a mock) behind a stub-authed FastAPI app. The check_access call inside `@require_permission` returns False on cross-workspace and the decorator converts to 404 — proving the boundary holds at the HTTP boundary, not just the unit level. Run stream/wait are skipped here (they spin a background worker); their guard goes through the same decorator path so unit coverage in `test_require_permission_workspace` is sufficient. `_StubAuthMiddleware` gains `override_user_contextvar=True` so cross-user / cross-workspace tests can drive both contextvars from the stub. Default stays off — the autouse user fixture continues to own the contextvar for legacy tests whose routes resolve filesystem paths via `get_effective_user_id()`. 86 router / boundary tests stay green. --- backend/tests/_router_auth_helpers.py | 16 +++ .../test_workspace_isolation_boundary.py | 127 ++++++++++++++++++ 2 files changed, 143 insertions(+) create mode 100644 backend/tests/test_workspace_isolation_boundary.py diff --git a/backend/tests/_router_auth_helpers.py b/backend/tests/_router_auth_helpers.py index 38f98a12..77b58b34 100644 --- a/backend/tests/_router_auth_helpers.py +++ b/backend/tests/_router_auth_helpers.py @@ -38,6 +38,10 @@ from starlette.types import ASGIApp from app.gateway.auth.models import ActiveWorkspace, User from app.gateway.authz import AuthContext, Permissions +from deerflow.runtime.user_context import ( + reset_current_user, + set_current_user, +) from deerflow.runtime.workspace_context import ( reset_current_workspace, set_current_workspace, @@ -82,16 +86,24 @@ class _StubAuthMiddleware(BaseHTTPMiddleware): app: ASGIApp, user_factory: Callable[[], User], workspace_factory: Callable[[], ActiveWorkspace | None] | None = None, + override_user_contextvar: bool = False, ) -> None: super().__init__(app) self._user_factory = user_factory self._workspace_factory = workspace_factory + # Tests that only need ``request.state.auth`` (the @require_permission + # path) keep the autouse user contextvar — flipping it to a per-call + # UUID would break legacy tests whose routes resolve paths via + # ``get_effective_user_id()``. Cross-user / cross-workspace tests opt + # in by setting this flag so the contextvar matches the request user. + self._override_user_contextvar = override_user_contextvar async def dispatch(self, request: Request, call_next: Callable) -> Response: user = self._user_factory() request.state.user = user request.state.auth = AuthContext(user=user, permissions=list(_STUB_PERMISSIONS)) + user_token = set_current_user(user) if self._override_user_contextvar else None ws_token = None if self._workspace_factory is not None: workspace = self._workspace_factory() @@ -103,12 +115,15 @@ class _StubAuthMiddleware(BaseHTTPMiddleware): finally: if ws_token is not None: reset_current_workspace(ws_token) + if user_token is not None: + reset_current_user(user_token) def make_authed_test_app( *, user_factory: Callable[[], User] | None = None, workspace_factory: Callable[[], ActiveWorkspace | None] | None = None, + override_user_contextvar: bool = False, owner_check_passes: bool = True, ) -> FastAPI: """Build a FastAPI test app with stub auth + permissive thread_store. @@ -133,6 +148,7 @@ def make_authed_test_app( _StubAuthMiddleware, user_factory=factory, workspace_factory=workspace_factory, + override_user_contextvar=override_user_contextvar, ) repo = MagicMock() diff --git a/backend/tests/test_workspace_isolation_boundary.py b/backend/tests/test_workspace_isolation_boundary.py new file mode 100644 index 00000000..ba2c905f --- /dev/null +++ b/backend/tests/test_workspace_isolation_boundary.py @@ -0,0 +1,127 @@ +"""PR6 T6.8 — cross-workspace isolation boundary (e2e). + +Wires a ``MemoryThreadMetaStore`` (LangGraph BaseStore backed) into a +stub-authed FastAPI app and asserts that any request from workspace B +against a thread created in workspace A returns **404**, regardless of +matching user_id. The cross-workspace block fires in +``ThreadMetaStore.check_access`` and is converted to 404 by +``@require_permission(owner_check=True)``. + +We use the memory-backed implementation so the test stays in the test +event loop end-to-end (the SQL engine binds to whatever loop owns +``init_engine`` and the TestClient spins its own loop, which would +collide). The decorator path it exercises is the same as production; +the SQL repository's identical workspace filter is unit-covered by +``test_thread_meta_workspace_filter.py``. + +Covers: +- ``GET /api/threads/{tid}`` — read (require_existing=False) +- ``DELETE /api/threads/{tid}`` — destructive (require_existing=True) +- ``PATCH /api/threads/{tid}`` — destructive write +- positive control: same-workspace GET still succeeds +""" + +from __future__ import annotations + +from collections.abc import Callable +from types import SimpleNamespace +from uuid import uuid4 + +from _router_auth_helpers import make_authed_test_app +from fastapi.testclient import TestClient +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.store.memory import InMemoryStore + +from app.gateway.auth.models import ActiveWorkspace, User +from app.gateway.routers import threads +from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore +from deerflow.runtime.workspace_context import ( + reset_current_workspace, + set_current_workspace, +) + + +def _workspace_factory(wid: str) -> Callable[[], ActiveWorkspace]: + def _factory() -> ActiveWorkspace: + return ActiveWorkspace(id=wid, role="owner") + + return _factory + + +def _user_factory(uid: str) -> Callable[[], User]: + def _factory() -> User: + return User(email=f"{uid}@example.com", password_hash="x", system_role="user", id=uid) + + return _factory + + +def _seed_thread(store, *, thread_id: str, user_id: str, workspace_id: str) -> None: + """Insert a thread record under a specific workspace, bypassing the autouse fixture.""" + import asyncio + + async def _go(): + meta_store = MemoryThreadMetaStore(store) + token = set_current_workspace(SimpleNamespace(id=workspace_id, role="owner")) + try: + await meta_store.create(thread_id, user_id=user_id) + finally: + reset_current_workspace(token) + + asyncio.run(_go()) + + +def _build_app(*, user_id: str, workspace_id: str): + app = make_authed_test_app( + user_factory=_user_factory(user_id), + workspace_factory=_workspace_factory(workspace_id), + override_user_contextvar=True, + ) + store = InMemoryStore() + app.state.store = store + app.state.checkpointer = InMemorySaver() + app.state.thread_store = MemoryThreadMetaStore(store) + app.include_router(threads.router) + return app, store + + +def test_cross_workspace_get_returns_404(): + user_id = str(uuid4()) + app, store = _build_app(user_id=user_id, workspace_id="ws-beta") + _seed_thread(store, thread_id="t1", user_id=user_id, workspace_id="ws-alpha") + + with TestClient(app) as client: + response = client.get("/api/threads/t1") + assert response.status_code == 404 + assert "not found" in response.json()["detail"].lower() + + +def test_cross_workspace_delete_returns_404(): + user_id = str(uuid4()) + app, store = _build_app(user_id=user_id, workspace_id="ws-beta") + _seed_thread(store, thread_id="t1", user_id=user_id, workspace_id="ws-alpha") + + with TestClient(app) as client: + response = client.delete("/api/threads/t1") + assert response.status_code == 404 + + +def test_cross_workspace_patch_returns_404(): + user_id = str(uuid4()) + app, store = _build_app(user_id=user_id, workspace_id="ws-beta") + _seed_thread(store, thread_id="t1", user_id=user_id, workspace_id="ws-alpha") + + with TestClient(app) as client: + response = client.patch("/api/threads/t1", json={"metadata": {"k": "v"}}) + assert response.status_code == 404 + + +def test_same_workspace_get_succeeds(): + """Positive control: when the workspace matches, the row is returned.""" + user_id = str(uuid4()) + app, store = _build_app(user_id=user_id, workspace_id="ws-alpha") + _seed_thread(store, thread_id="t1", user_id=user_id, workspace_id="ws-alpha") + + with TestClient(app) as client: + response = client.get("/api/threads/t1") + assert response.status_code == 200 + assert response.json()["thread_id"] == "t1"