From 0456606dc132ea9d865a0c3f39b5fcbcc59c88e5 Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Wed, 13 May 2026 17:33:02 +0800 Subject: [PATCH] =?UTF-8?q?test(authz):=20PR6=20T6.6=20=E2=80=94=20@requir?= =?UTF-8?q?e=5Fpermission=20workspace=5Fid=20propagation=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The decorator path was already updated in T6.4 to read `get_effective_workspace_id()` and forward it as the third positional to `check_access`. T6.6 closes the loop with dedicated coverage: - `_StubAuthMiddleware` and `make_authed_test_app` now accept an optional `workspace_factory` so router tests can drive the active workspace contextvar end-to-end through the FastAPI middleware stack. - 5 new probe tests assert: cross-workspace → 404 (not 403), same workspace → 200, the third positional reaching `check_access` carries the contextvar id, no-context tests (marked `no_auto_workspace`) fall back to "default", and read-style routes (`require_existing=False`) thread workspace_id too. 75 existing router tests (artifacts / runs / threads / uploads / suggestions) stay green. --- backend/tests/_router_auth_helpers.py | 38 ++++- .../test_require_permission_workspace.py | 136 ++++++++++++++++++ 2 files changed, 170 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_require_permission_workspace.py diff --git a/backend/tests/_router_auth_helpers.py b/backend/tests/_router_auth_helpers.py index 2bd2ebde..38f98a12 100644 --- a/backend/tests/_router_auth_helpers.py +++ b/backend/tests/_router_auth_helpers.py @@ -36,8 +36,12 @@ from fastapi import FastAPI, Request, Response from starlette.middleware.base import BaseHTTPMiddleware from starlette.types import ASGIApp -from app.gateway.auth.models import User +from app.gateway.auth.models import ActiveWorkspace, User from app.gateway.authz import AuthContext, Permissions +from deerflow.runtime.workspace_context import ( + reset_current_workspace, + set_current_workspace, +) # Default permission set granted to the stub user. Mirrors `_ALL_PERMISSIONS` # in authz.py — kept inline so the tests don't import a private symbol. @@ -67,22 +71,44 @@ class _StubAuthMiddleware(BaseHTTPMiddleware): Mirrors what production ``AuthMiddleware`` does after the JWT decode + DB lookup short-circuit, so ``@require_permission`` finds an authenticated context and skips its own re-authentication path. + + Optionally stamps the workspace contextvar too — needed by the PR6 + decorator path that calls ``get_effective_workspace_id()`` before + delegating to ``check_access``. """ - def __init__(self, app: ASGIApp, user_factory: Callable[[], User]) -> None: + def __init__( + self, + app: ASGIApp, + user_factory: Callable[[], User], + workspace_factory: Callable[[], ActiveWorkspace | None] | None = None, + ) -> None: super().__init__(app) self._user_factory = user_factory + self._workspace_factory = workspace_factory 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)) - return await call_next(request) + + ws_token = None + if self._workspace_factory is not None: + workspace = self._workspace_factory() + if workspace is not None: + request.state.workspace = workspace + ws_token = set_current_workspace(workspace) + try: + return await call_next(request) + finally: + if ws_token is not None: + reset_current_workspace(ws_token) def make_authed_test_app( *, user_factory: Callable[[], User] | None = None, + workspace_factory: Callable[[], ActiveWorkspace | None] | None = None, owner_check_passes: bool = True, ) -> FastAPI: """Build a FastAPI test app with stub auth + permissive thread_store. @@ -103,7 +129,11 @@ def make_authed_test_app( """ factory = user_factory or _make_stub_user app = FastAPI() - app.add_middleware(_StubAuthMiddleware, user_factory=factory) + app.add_middleware( + _StubAuthMiddleware, + user_factory=factory, + workspace_factory=workspace_factory, + ) repo = MagicMock() repo.check_access = AsyncMock(return_value=owner_check_passes) diff --git a/backend/tests/test_require_permission_workspace.py b/backend/tests/test_require_permission_workspace.py new file mode 100644 index 00000000..d4884f22 --- /dev/null +++ b/backend/tests/test_require_permission_workspace.py @@ -0,0 +1,136 @@ +"""PR6 T6.6 — `@require_permission(owner_check=True)` workspace upgrade. + +The decorator must: + +1. Pull workspace_id from ``get_effective_workspace_id()`` (set by + AuthMiddleware per request) and pass it as the third positional to + ``ThreadMetaStore.check_access``. +2. Raise **HTTPException 404** when check_access returns False — never + 403 — so a cross-workspace request cannot distinguish "thread exists + in another tenant" from "thread does not exist". + +These tests build a fake router with the same decorator usage as the +production code and verify the decorator's behaviour via a Mock +``thread_store`` whose ``check_access`` call we inspect, plus a TestClient +exercising the full HTTP boundary. +""" + +from __future__ import annotations + +from collections.abc import Callable + +import pytest +from _router_auth_helpers import make_authed_test_app +from fastapi import APIRouter, Request +from fastapi.testclient import TestClient + +from app.gateway.auth.models import ActiveWorkspace +from app.gateway.authz import require_permission +from deerflow.runtime.workspace_context import ( + reset_current_workspace, + set_current_workspace, +) + + +def _make_workspace(wid: str) -> Callable[[], ActiveWorkspace]: + """Factory closure stable per call so the middleware reinjects the same id.""" + + def _factory() -> ActiveWorkspace: + return ActiveWorkspace(id=wid, role="owner") + + return _factory + + +def _mount_routes(app): + router = APIRouter() + + @router.delete("/probe/{thread_id}") + @require_permission("threads", "delete", owner_check=True, require_existing=True) + async def _delete_probe(thread_id: str, request: Request): # noqa: ARG001 + return {"ok": True, "thread_id": thread_id} + + @router.get("/probe/{thread_id}") + @require_permission("threads", "read", owner_check=True) + async def _get_probe(thread_id: str, request: Request): # noqa: ARG001 + return {"ok": True, "thread_id": thread_id} + + app.include_router(router) + return app + + +def test_cross_workspace_returns_404(): + """check_access returning False surfaces as 404, never 403.""" + app = make_authed_test_app( + workspace_factory=_make_workspace("ws-alpha"), + owner_check_passes=False, + ) + _mount_routes(app) + with TestClient(app) as client: + response = client.delete("/probe/t1") + assert response.status_code == 404 + assert "not found" in response.json()["detail"].lower() + + +def test_same_workspace_delete_allowed(): + """check_access returning True lets the route execute.""" + app = make_authed_test_app( + workspace_factory=_make_workspace("ws-alpha"), + owner_check_passes=True, + ) + _mount_routes(app) + with TestClient(app) as client: + response = client.delete("/probe/t1") + assert response.status_code == 200 + assert response.json()["ok"] is True + + +def test_workspace_id_passed_to_check_access(): + """check_access receives the contextvar workspace_id as the 3rd positional.""" + app = make_authed_test_app( + workspace_factory=_make_workspace("ws-alpha"), + owner_check_passes=True, + ) + _mount_routes(app) + with TestClient(app) as client: + client.delete("/probe/t1") + + call = app.state.thread_store.check_access.call_args + assert call is not None + args = call.args + # (thread_id, user_id, workspace_id) + assert args[0] == "t1" + assert args[2] == "ws-alpha" + + +@pytest.mark.no_auto_workspace +def test_no_workspace_in_context_falls_back_to_default(): + """No-auth dev mode (no workspace contextvar) uses DEFAULT_WORKSPACE_ID.""" + app = make_authed_test_app(workspace_factory=None, owner_check_passes=True) + _mount_routes(app) + with TestClient(app) as client: + client.delete("/probe/t1") + + args = app.state.thread_store.check_access.call_args.args + assert args[2] == "default" + + +def test_get_route_also_uses_workspace_id(): + """Read-style routes (require_existing=False) also pass workspace_id through.""" + app = make_authed_test_app( + workspace_factory=_make_workspace("ws-beta"), + owner_check_passes=True, + ) + _mount_routes(app) + with TestClient(app) as client: + client.get("/probe/t-read") + args = app.state.thread_store.check_access.call_args.args + assert args[2] == "ws-beta" + + +@pytest.fixture +def _reset_ws(): + """Helper for direct-call paths that mutate the contextvar.""" + tokens: list = [] + yield lambda wid: tokens.append(set_current_workspace(ActiveWorkspace(id=wid, role="owner"))) + for token in reversed(tokens): + reset_current_workspace(token)