From 0745fb268f93a2e93467e1534b5a9c7cb1fb8c0b Mon Sep 17 00:00:00 2001 From: xiaodu55 <1704788159@qq.com> Date: Wed, 16 Sep 2026 15:56:25 +0800 Subject: [PATCH] fix(gateway): scope runs read endpoints by data identity, not authorization identity (#5448) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(gateway): scope runs read endpoints by data identity, not authorization identity Trusted internal callers are authorized as a synthetic internal user (id "default", or the make_safe_user_id-normalized owner with an owner header), while start_run stamps run rows with the raw trusted-owner value. list_runs / get_run / list_runs_page filtered by the authorization identity, so the store-side user filter never matched and internal callers always saw an empty runs list (or 404) for threads they are authorized to read. The three read endpoints now resolve their filter id through _run_scope_user_id: internal-role callers skip the per-user filter (thread visibility is already authorized by owner_check=True), and browser/API sessions keep the existing per-user filter unchanged. Regression tests cover both identities across the three endpoints (list, keyset page, single get) with a MemoryRunStore seeded with mixed-owner rows; without the fix the four internal-caller cases fail while the browser-session isolation case passes. * fix(gateway): route the message read endpoints through the same data-identity scoping Review follow-up on #5448: list_thread_messages and list_thread_messages_page resolved get_current_user and passed it as the data filter to the event-store scan, hidden-run lookups, turn-duration injection and the feedback queries — the same authorization-vs-data identity conflation fixed for the runs endpoints, leaving the #5437 empty-read symptom in place for lossy owner values. Both endpoints now resolve their filter id through _run_scope_user_id as well. Regression tests extend to the two message endpoints, asserting the resolved filter identity at the runs-store and feedback-repo boundaries (None for internal callers, the session user id for browser sessions). * fix(feedback): deterministic per-run collapse for unfiltered feedback reads Review follow-up on #5448: with _run_scope_user_id returning None for internal callers, the feedback lookups now receive an explicit-None user id, which skips the user_id WHERE in FeedbackRepository. On shared/NULL-owner threads several browser users can hold feedback on the same run, and list_by_thread_grouped / list_by_run_ids collapsed rows per run_id via a dict comprehension over unordered results — the feedback attached to the last AI message would be an arbitrary user's row. Both methods now order by created_at ASC with feedback_id as the tie-break, so the collapse deterministically keeps the most recently created feedback. _run_scope_user_id's docstring now documents that the resolved id also scopes feedback and event-store reads, not just run rows. Regression test seeds multi-user feedback on one run and asserts the collapse outcome is stable across repeated unfiltered reads. * docs(feedback): the collapse keeps the most recently written feedback created_at is refreshed on upsert, so the surviving row per run is the most recently written (created or updated), not the most recently created — align both docstrings with the ordering key's actual semantics. --- backend/app/gateway/routers/thread_runs.py | 43 ++- .../deerflow/persistence/feedback/sql.py | 19 +- backend/tests/test_feedback.py | 25 ++ .../tests/test_thread_runs_internal_scope.py | 317 ++++++++++++++++++ 4 files changed, 394 insertions(+), 10 deletions(-) create mode 100644 backend/tests/test_thread_runs_internal_scope.py diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 3b4966897..3b509364b 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -48,7 +48,7 @@ from app.gateway.conversation_reader import ( scan_visible_thread_messages as _scan_visible_thread_messages, ) from app.gateway.deps import get_current_user, get_feedback_repo, get_run_event_store, get_run_manager, get_run_store, get_stream_bridge -from app.gateway.internal_auth import get_trusted_internal_owner_user_id +from app.gateway.internal_auth import INTERNAL_SYSTEM_ROLE, get_trusted_internal_owner_user_id from app.gateway.pagination import trim_run_message_page from app.gateway.run_models import RunCreateRequest from app.gateway.services import abuild_checkpoint_state_accessor, build_thread_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion @@ -1062,12 +1062,38 @@ def _parse_run_page_created_at(value: str) -> str: return normalized +async def _run_scope_user_id(request: Request) -> str | None: + """Resolve the data-filter id for run and message reads, not for authorization. + + Thread visibility on these endpoints is already authorized by + ``@require_permission(..., owner_check=True)``. Trusted internal callers + are authorized as a synthetic internal user instead — ``id="default"`` + without an owner header, or the ``make_safe_user_id``-normalized owner + otherwise — while ``start_run`` stamps run rows and run-event rows with + the raw trusted-owner value. Filtering by the authorization identity + therefore never matches the persisted rows (#5437), so internal callers + read the authorized thread's runs, event-store messages, hidden-run + lookups, turn durations and feedback unfiltered; browser/API sessions + keep the per-user filter. + + Feedback note: an explicit ``None`` also skips the ``user_id`` WHERE in + ``FeedbackRepository``, so on shared/NULL-owner threads several users' + feedback rows collapse per run — ``FeedbackRepository.list_by_thread_grouped`` + / ``list_by_run_ids`` order deterministically (latest wins, ``feedback_id`` + breaks ties) to keep that well-defined. + """ + user = getattr(request.state, "user", None) + if getattr(user, "system_role", None) == INTERNAL_SYSTEM_ROLE: + return None + return await get_current_user(request) + + @router.get("/{thread_id}/runs", response_model=list[RunResponse]) @require_permission("runs", "read", owner_check=True) async def list_runs(thread_id: ThreadId, request: Request) -> list[RunResponse]: """List the newest runs for a thread (default 100, as a bare array).""" run_mgr = get_run_manager(request) - user_id = await get_current_user(request) + user_id = await _run_scope_user_id(request) records = await run_mgr.list_by_thread(thread_id, user_id=user_id) return [_record_to_response(r) for r in records] @@ -1095,7 +1121,7 @@ async def list_runs_page( before_created_at = _parse_run_page_created_at(before_created_at) run_mgr = get_run_manager(request) - user_id = await get_current_user(request) + user_id = await _run_scope_user_id(request) records = await run_mgr.list_by_thread( thread_id, user_id=user_id, @@ -1119,7 +1145,7 @@ async def list_runs_page( async def get_run(thread_id: ThreadId, run_id: str, request: Request) -> RunResponse: """Get details of a specific run.""" run_mgr = get_run_manager(request) - user_id = await get_current_user(request) + user_id = await _run_scope_user_id(request) record = await run_mgr.get(run_id, user_id=user_id) if record is None or record.thread_id != thread_id: raise HTTPException(status_code=404, detail=f"Run {run_id} not found") @@ -1342,9 +1368,10 @@ async def list_thread_messages( after_seq: int | None = Query(default=None, ge=1), ) -> list[dict]: """Return displayable messages for a thread (across all runs), with feedback attached.""" - # Resolve the caller once; it is needed both to scope the feedback query - # below and to list the thread's runs for turn-duration injection. - user_id = await get_current_user(request) + # Resolve the data-filter id once (None for internal callers — same + # rationale as the runs endpoints above); it scopes the feedback query, + # the hidden-run lookup, the event-store scan and turn-duration injection. + user_id = await _run_scope_user_id(request) run_mgr = get_run_manager(request) hidden_run_ids = await _default_history_hidden_run_ids(run_mgr, thread_id, user_id=user_id) messages, _ = await _scan_visible_thread_messages( @@ -1482,7 +1509,7 @@ async def list_thread_messages_page( if "after_seq" in request.query_params: raise HTTPException(status_code=422, detail="after_seq is not supported by this backward-only endpoint") - user_id = await get_current_user(request) + user_id = await _run_scope_user_id(request) rows, has_more = await _scan_thread_message_page( thread_id, limit=limit, diff --git a/backend/packages/harness/deerflow/persistence/feedback/sql.py b/backend/packages/harness/deerflow/persistence/feedback/sql.py index 5cd03d0e9..1c8a0f087 100644 --- a/backend/packages/harness/deerflow/persistence/feedback/sql.py +++ b/backend/packages/harness/deerflow/persistence/feedback/sql.py @@ -193,11 +193,19 @@ class FeedbackRepository: *, user_id: str | None | _AutoSentinel = AUTO, ) -> dict[str, dict]: - """Return feedback grouped by run_id for a thread: {run_id: feedback_dict}.""" + """Return feedback grouped by run_id for a thread: {run_id: feedback_dict}. + + With an explicit ``None`` user id (unfiltered reads) several users may + hold feedback on the same run, so order deterministically — the + per-run collapse below keeps the last row per ``run_id``, i.e. the + most recently written feedback (``created_at`` is refreshed on + update), with ``feedback_id`` breaking ties. + """ resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_thread_grouped") stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id) if resolved_user_id is not None: stmt = stmt.where(FeedbackRow.user_id == resolved_user_id) + stmt = stmt.order_by(FeedbackRow.created_at.asc(), FeedbackRow.feedback_id.asc()) async with self._sf() as session: result = await session.execute(stmt) return {row.run_id: self._row_to_dict(row) for row in result.scalars()} @@ -209,7 +217,13 @@ class FeedbackRepository: *, user_id: str | None | _AutoSentinel = AUTO, ) -> dict[str, dict]: - """Return feedback for only the selected runs in one thread.""" + """Return feedback for only the selected runs in one thread. + + Same deterministic ordering as :meth:`list_by_thread_grouped`: with an + explicit ``None`` user id the per-run collapse keeps the most recently + written feedback (``created_at`` is refreshed on update), ties broken + by ``feedback_id``. + """ if not run_ids: return {} resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_run_ids") @@ -219,6 +233,7 @@ class FeedbackRepository: ) if resolved_user_id is not None: stmt = stmt.where(FeedbackRow.user_id == resolved_user_id) + stmt = stmt.order_by(FeedbackRow.created_at.asc(), FeedbackRow.feedback_id.asc()) async with self._sf() as session: result = await session.execute(stmt) return {row.run_id: self._row_to_dict(row) for row in result.scalars()} diff --git a/backend/tests/test_feedback.py b/backend/tests/test_feedback.py index d9a34fbf1..ef5d62950 100644 --- a/backend/tests/test_feedback.py +++ b/backend/tests/test_feedback.py @@ -244,6 +244,31 @@ class TestFeedbackRepository: assert grouped["r2"]["rating"] == -1 await _cleanup() + @pytest.mark.anyio + async def test_unfiltered_reads_collapse_multi_user_feedback_deterministically(self, tmp_path): + """Explicit-None reads (internal callers) collapse same-run feedback rows. + + Several browser users can hold feedback on one run of a shared + NULL-owner thread; the per-run collapse must be deterministic — + the most recently created feedback wins, ``feedback_id`` breaks ties. + """ + import asyncio + + repo = await _make_feedback_repo(tmp_path) + await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1") + # Guarantee a strictly later created_at than the first row. + await asyncio.sleep(0.01) + second = await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u2") + + for _ in range(3): + grouped = await repo.list_by_thread_grouped("t1", user_id=None) + by_run_ids = await repo.list_by_run_ids("t1", {"r1"}, user_id=None) + assert grouped["r1"]["feedback_id"] == second["feedback_id"] + assert grouped["r1"]["rating"] == -1 + assert grouped["r1"]["user_id"] == "u2" + assert by_run_ids["r1"]["feedback_id"] == second["feedback_id"] + await _cleanup() + @pytest.mark.anyio async def test_list_by_run_ids_empty_skips_query(self, tmp_path): repo = await _make_feedback_repo(tmp_path) diff --git a/backend/tests/test_thread_runs_internal_scope.py b/backend/tests/test_thread_runs_internal_scope.py new file mode 100644 index 000000000..9754b6858 --- /dev/null +++ b/backend/tests/test_thread_runs_internal_scope.py @@ -0,0 +1,317 @@ +"""Regression coverage for the runs/messages read endpoints' identity scoping (#5437). + +Trusted internal callers are *authorized* as a synthetic internal user +(``system_role="internal"``) whose id is ``"default"`` or the +``make_safe_user_id``-normalized owner, while ``start_run`` stamps run rows +with the raw trusted-owner value. Filtering the reads by the authorization +identity therefore never matches the persisted rows. The read endpoints must +skip the per-user filter for internal callers — thread visibility is already +authorized by ``@require_permission(..., owner_check=True)`` — and keep it for +browser/API sessions. +""" + +from __future__ import annotations + +import asyncio +from uuid import UUID + +import pytest +from fastapi import FastAPI, Request, Response +from fastapi.testclient import TestClient +from starlette.middleware.base import BaseHTTPMiddleware + +from app.gateway.auth.models import User +from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL, AUTH_SOURCE_SESSION +from app.gateway.authz import AuthContext, Permissions +from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, get_internal_user +from app.gateway.routers import thread_runs +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.runs.manager import RunManager +from deerflow.runtime.runs.store.memory import MemoryRunStore + +THREAD_ID = "thread-scope" +BROWSER_USER_ID = UUID("00000000-0000-0000-0000-00000000000a") +# A lossy trusted-owner value: make_safe_user_id normalizes it to +# "feishu-owner-777-", which can never equal the raw value stamped on +# the run row — the exact mismatch class reported in #5437. +OWNER_RAW = "feishu:owner-777" +RUN_BROWSER = "run-browser-row" +RUN_OWNER = "run-owner-row" + +_STUB_PERMISSIONS: list[str] = [ + Permissions.THREADS_READ, + Permissions.RUNS_READ, + Permissions.RUNS_CANCEL, +] + + +class _ScopeAuthMiddleware(BaseHTTPMiddleware): + """Stamp the same state trio production ``AuthMiddleware`` stamps.""" + + def __init__(self, app, *, user, auth_source: str) -> None: + super().__init__(app) + self._user = user + self._auth_source = auth_source + + async def dispatch(self, request: Request, call_next) -> Response: + request.state.user = self._user + request.state.auth_source = self._auth_source + request.state.auth = AuthContext(user=self._user, permissions=list(_STUB_PERMISSIONS)) + return await call_next(request) + + +class _PermissiveThreadStore: + """Stands in for the thread store behind ``owner_check=True``.""" + + async def check_access(self, _thread_id: str, _user_id: str, *, require_existing: bool = False) -> bool: + return True + + +class _RecordingRunStore(MemoryRunStore): + """Records the per-user filter identity each read resolves to. + + ``MemoryRunEventStore.list_messages`` ignores ``user_id`` (only the SQL + backends honor it), so the runs store is where the resolved filter id is + observable in-memory: hidden-run lookups and turn-duration injection both + flow through ``list_by_thread``/``get`` with the endpoint's filter id. + """ + + def __init__(self) -> None: + super().__init__() + self.list_by_thread_user_ids: list[str | None] = [] + self.get_user_ids: list[str | None] = [] + + async def list_by_thread(self, thread_id, *, user_id=None, **kwargs): + self.list_by_thread_user_ids.append(user_id) + return await super().list_by_thread(thread_id, user_id=user_id, **kwargs) + + async def get(self, run_id, *, user_id=None, **kwargs): + self.get_user_ids.append(user_id) + return await super().get(run_id, user_id=user_id, **kwargs) + + +class _RecordingFeedbackRepo: + """Records the per-user identity the feedback queries are scoped with.""" + + def __init__(self) -> None: + self.list_by_thread_user_ids: list[str | None] = [] + self.list_by_run_ids_user_ids: list[str | None] = [] + + async def list_by_thread_grouped(self, thread_id, *, user_id=None): + self.list_by_thread_user_ids.append(user_id) + return {} + + async def list_by_run_ids(self, thread_id, run_ids, *, user_id=None): + self.list_by_run_ids_user_ids.append(user_id) + return {} + + +def _browser_user() -> User: + return User(id=BROWSER_USER_ID, email="scope-test@example.com", password_hash="x", system_role="user") + + +def _internal_user(owner_raw: str | None): + # Mirrors AuthMiddleware + get_internal_user: the synthetic internal user + # carries the safe-spelled owner id, or "default" without an owner header. + return get_internal_user(owner_user_id=owner_raw) + + +def _seed_run(store: MemoryRunStore, run_id: str, *, user_id: str | None) -> None: + asyncio.run(store.put(run_id, thread_id=THREAD_ID, user_id=user_id, status="success")) + + +def _seed_message(event_store: MemoryRunEventStore, run_id: str, message_id: str) -> None: + asyncio.run( + event_store.put( + thread_id=THREAD_ID, + run_id=run_id, + event_type="llm.ai.response", + category="message", + content={"type": "ai", "id": message_id, "content": message_id, "additional_kwargs": {}}, + metadata={}, + ) + ) + + +def _make_app( + *, + user, + auth_source: str, + run_store: MemoryRunStore, + event_store: MemoryRunEventStore | None = None, + feedback_repo: _RecordingFeedbackRepo | None = None, +) -> TestClient: + app = FastAPI() + app.add_middleware(_ScopeAuthMiddleware, user=user, auth_source=auth_source) + app.state.thread_store = _PermissiveThreadStore() + app.state.run_manager = RunManager(store=run_store) + if event_store is not None: + app.state.run_event_store = event_store + if feedback_repo is not None: + app.state.feedback_repo = feedback_repo + app.include_router(thread_runs.router) + return TestClient(app) + + +@pytest.fixture() +def mixed_owner_store() -> MemoryRunStore: + store = MemoryRunStore() + _seed_run(store, RUN_BROWSER, user_id=str(BROWSER_USER_ID)) + _seed_run(store, RUN_OWNER, user_id=OWNER_RAW) + return store + + +def test_internal_caller_lists_owner_stamped_runs(mixed_owner_store: MemoryRunStore) -> None: + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=mixed_owner_store, + ) + + with client: + response = client.get( + f"/api/threads/{THREAD_ID}/runs", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert response.status_code == 200 + assert {row["run_id"] for row in response.json()} == {RUN_BROWSER, RUN_OWNER} + + +def test_internal_caller_get_run_owner_stamped(mixed_owner_store: MemoryRunStore) -> None: + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=mixed_owner_store, + ) + + with client: + response = client.get( + f"/api/threads/{THREAD_ID}/runs/{RUN_OWNER}", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert response.status_code == 200 + assert response.json()["run_id"] == RUN_OWNER + + +def test_internal_caller_runs_page_owner_stamped(mixed_owner_store: MemoryRunStore) -> None: + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=mixed_owner_store, + ) + + with client: + response = client.get( + f"/api/threads/{THREAD_ID}/runs/page", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert response.status_code == 200 + assert {row["run_id"] for row in response.json()["data"]} == {RUN_BROWSER, RUN_OWNER} + assert response.json()["has_more"] is False + + +def test_internal_caller_without_owner_header_sees_authorized_thread_runs(mixed_owner_store: MemoryRunStore) -> None: + """No owner header ⇒ synthetic id "default", which matches nothing either. + + The thread is authorized via owner_check, so its runs stay listable. + """ + client = _make_app( + user=_internal_user(None), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=mixed_owner_store, + ) + + with client: + response = client.get(f"/api/threads/{THREAD_ID}/runs") + + assert response.status_code == 200 + assert {row["run_id"] for row in response.json()} == {RUN_BROWSER, RUN_OWNER} + + +def test_browser_session_keeps_per_user_filter(mixed_owner_store: MemoryRunStore) -> None: + """Browser sessions keep filtering by their own data identity.""" + client = _make_app( + user=_browser_user(), + auth_source=AUTH_SOURCE_SESSION, + run_store=mixed_owner_store, + ) + + with client: + listed = client.get(f"/api/threads/{THREAD_ID}/runs") + cross_user = client.get(f"/api/threads/{THREAD_ID}/runs/{RUN_OWNER}") + + assert listed.status_code == 200 + assert [row["run_id"] for row in listed.json()] == [RUN_BROWSER] + assert cross_user.status_code == 404 + + +def test_internal_caller_messages_skip_per_user_filter(mixed_owner_store: MemoryRunStore) -> None: + """Internal callers read the authorized thread's messages unfiltered. + + The observable filter identity is what reaches the scoped queries — the + runs store (hidden-run lookups, turn durations) and the feedback repo — + ``None`` for internal callers. + """ + event_store = MemoryRunEventStore() + _seed_message(event_store, RUN_OWNER, "msg-owner") + _seed_message(event_store, RUN_BROWSER, "msg-browser") + run_store = _RecordingRunStore() + for run_id, user_id in ((RUN_BROWSER, str(BROWSER_USER_ID)), (RUN_OWNER, OWNER_RAW)): + _seed_run(run_store, run_id, user_id=user_id) + feedback_repo = _RecordingFeedbackRepo() + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=event_store, + feedback_repo=feedback_repo, + ) + + with client: + response = client.get( + f"/api/threads/{THREAD_ID}/messages", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + page = client.get( + f"/api/threads/{THREAD_ID}/messages/page", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert response.status_code == 200 + assert {row["content"]["id"] for row in response.json()} == {"msg-owner", "msg-browser"} + assert page.status_code == 200 + assert {row["content"]["id"] for row in page.json()["data"]} == {"msg-owner", "msg-browser"} + assert run_store.list_by_thread_user_ids and all(uid is None for uid in run_store.list_by_thread_user_ids) + assert feedback_repo.list_by_thread_user_ids == [None] + assert feedback_repo.list_by_run_ids_user_ids == [None] + + +def test_browser_session_messages_keep_per_user_filter(mixed_owner_store: MemoryRunStore) -> None: + """Browser sessions keep passing their own id to the messages pipeline.""" + event_store = MemoryRunEventStore() + _seed_message(event_store, RUN_OWNER, "msg-owner") + _seed_message(event_store, RUN_BROWSER, "msg-browser") + run_store = _RecordingRunStore() + for run_id, user_id in ((RUN_BROWSER, str(BROWSER_USER_ID)), (RUN_OWNER, OWNER_RAW)): + _seed_run(run_store, run_id, user_id=user_id) + feedback_repo = _RecordingFeedbackRepo() + + client = _make_app( + user=_browser_user(), + auth_source=AUTH_SOURCE_SESSION, + run_store=run_store, + event_store=event_store, + feedback_repo=feedback_repo, + ) + + with client: + response = client.get(f"/api/threads/{THREAD_ID}/messages") + + assert response.status_code == 200 + assert {row["content"]["id"] for row in response.json()} == {"msg-owner", "msg-browser"} + assert run_store.list_by_thread_user_ids and all(uid == str(BROWSER_USER_ID) for uid in run_store.list_by_thread_user_ids) + assert feedback_repo.list_by_thread_user_ids == [str(BROWSER_USER_ID)]