diff --git a/backend/app/gateway/context_usage.py b/backend/app/gateway/context_usage.py index 04c4634ab..ef0d6730c 100644 --- a/backend/app/gateway/context_usage.py +++ b/backend/app/gateway/context_usage.py @@ -32,10 +32,14 @@ async def _load_checkpoint_messages(accessor: Any, config: dict[str, Any]) -> li return list(values.get("messages") or []) -async def _resolve_thread_model_name(run_store: Any, thread_id: str, app_config: Any) -> str | None: - """Prefer the latest run's model, then fall back to the first configured model.""" +async def _resolve_thread_model_name(run_store: Any, thread_id: str, app_config: Any, user_id: str | None = None) -> str | None: + """Prefer the latest run's model, then fall back to the first configured model. + + ``user_id`` scopes the latest-run lookup (``None`` = unfiltered, matching + the thread-scoped aggregate semantics). + """ try: - runs = await run_store.list_by_thread(thread_id, limit=1) + runs = await run_store.list_by_thread(thread_id, limit=1, user_id=user_id) except Exception: runs = [] if runs: @@ -59,7 +63,7 @@ def build_context_usage_payload(*, token_count: int, max_context_tokens: int | N } -async def build_context_usage(request: Request, thread_id: str, run_store: Any) -> dict[str, Any] | None: +async def build_context_usage(request: Request, thread_id: str, run_store: Any, user_id: str | None = None) -> dict[str, Any] | None: """Return approximate usage for the latest materialized thread checkpoint.""" try: app_config = get_config() @@ -79,7 +83,7 @@ async def build_context_usage(request: Request, thread_id: str, run_store: Any) logger.warning("Failed to count context messages for thread %s", thread_id, exc_info=True) return None - model_name = await _resolve_thread_model_name(run_store, thread_id, app_config) + model_name = await _resolve_thread_model_name(run_store, thread_id, app_config, user_id=user_id) model_config = app_config.get_model_config(model_name) if model_name else None configured_window = getattr(model_config, "context_window", None) if model_config is not None else None max_context_tokens = int(configured_window) if configured_window else None diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index ae33a0ed9..a672af6fb 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -590,7 +590,7 @@ async def _find_target_run_id( return source_run_id run_mgr = get_run_manager(request) - user_id = await _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) records = await run_mgr.list_by_thread(thread_id, user_id=user_id, limit=10) fallback_record = next( (record for record in records if record.status == RunStatus.success and _run_last_ai_matches_message(record, target_message)), @@ -677,7 +677,7 @@ def _run_status_value(record: Any) -> str | None: async def _require_successful_source_run(thread_id: str, run_id: str, request: Request) -> RunRecord: run_mgr = get_run_manager(request) - user_id = await _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) record = await run_mgr.get(run_id, user_id=user_id) if record is None: # The run-event journal is the authoritative lookup above. This fallback @@ -704,7 +704,7 @@ async def _find_interrupted_target_run_id( return None run_mgr = get_run_manager(request) - user_id = await _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) record = await run_mgr.get(source_run_id, user_id=user_id) if record is None: records = await run_mgr.list_by_thread(thread_id, user_id=user_id, limit=20) @@ -1062,7 +1062,22 @@ def _parse_run_page_created_at(value: str) -> str: return normalized -async def _run_scope_user_id(request: Request) -> str | None: +async def _thread_ownership_established(request: Request, thread_id: str) -> bool: + """Whether an existing meta row with a concrete owner covers ``thread_id``. + + Missing rows (legacy compatibility) and NULL-owner rows (shared/pre-auth + data) do **not** establish ownership, even though ``owner_check=True`` + still authorizes access to them. + """ + thread_store = getattr(request.app.state, "thread_store", None) + if thread_store is None: + return False + meta = await thread_store.get(thread_id, user_id=None) + meta_owner = meta.get("user_id") if isinstance(meta, dict) else getattr(meta, "user_id", None) + return meta is not None and bool(meta_owner) + + +async def _run_scope_user_id(request: Request, thread_id: str) -> str | None: """Resolve the data-filter id for run and message reads, not for authorization. Thread visibility on these endpoints is already authorized by @@ -1071,10 +1086,17 @@ async def _run_scope_user_id(request: Request) -> str | None: 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. + therefore never matches the persisted rows (#5437). + + Owner isolation (#5448 review P1): ``owner_check=True`` also authorizes + threads whose meta row is missing (legacy compatibility) or NULL-owner + (shared/pre-auth data). On those, an unfiltered read would expose other + users' persisted runs to the acting owner's internal caller, so the + per-user filter is only dropped when the thread's meta row exists with an + established owner; otherwise the raw trusted owner — the exact value + ``start_run`` stamps — is retained as the filter. Browser/API sessions + always keep the per-user filter. The thread token-usage aggregate is + scoped the same way. Feedback note: an explicit ``None`` also skips the ``user_id`` WHERE in ``FeedbackRepository``, so on shared/NULL-owner threads several users' @@ -1082,18 +1104,56 @@ async def _run_scope_user_id(request: Request) -> str | None: / ``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: + # Tolerate state-less request stand-ins used by focused unit tests. + state = getattr(request, "state", None) + user = getattr(state, "user", None) + if getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE: + return await get_current_user(request) + if await _thread_ownership_established(request, thread_id): return None + # Missing or NULL-owner meta row: ownership was never established, so the + # isolation boundary is the acting owner's raw stamp (the exact value + # start_run writes) — or, without an owner header, the synthetic + # "default" identity, which only matches legacy default-stamped rows. + owner = get_trusted_internal_owner_user_id(request) + if owner is not None: + return owner return await get_current_user(request) +async def _require_run_visible_to_scope(run_id: str, thread_id: str, request: Request) -> None: + """Gate run-scoped sub-resource reads and writes (events, messages, join, + stream, cancel, artifact archive). + + These routes query or mutate by ``(thread_id, run_id)`` without a + per-user filter of their own. For trusted internal callers on threads + without established ownership, that let an internal caller acting for + owner A read or cancel owner B's run by id (#5448 review P1 follow-up). + The run's own stamp must therefore match the acting owner's raw value (or + the legacy ``"default"`` stamp); every other caller and every + established-ownership thread keeps its existing semantics. + """ + state = getattr(request, "state", None) + user = getattr(state, "user", None) + if getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE: + return + if await _thread_ownership_established(request, thread_id): + return + scope = get_trusted_internal_owner_user_id(request) or "default" + record = await get_run_manager(request).get(run_id) + if record is None: + return + record_owner = getattr(record, "user_id", None) or "default" + if record.thread_id != thread_id or record_owner != scope: + raise HTTPException(status_code=404, detail=f"Run {run_id} not found") + + @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 _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) records = await run_mgr.list_by_thread(thread_id, user_id=user_id) return [_record_to_response(r) for r in records] @@ -1121,7 +1181,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 _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) records = await run_mgr.list_by_thread( thread_id, user_id=user_id, @@ -1145,7 +1205,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 _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) 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") @@ -1172,6 +1232,7 @@ async def cancel_run( durably notifies the owner when its lease is live, or takes over and terminalizes the run when that lease has expired. """ + await _require_run_visible_to_scope(run_id, thread_id, request) run_mgr = get_run_manager(request) record = await run_mgr.get(run_id) if record is None or record.thread_id != thread_id: @@ -1216,6 +1277,7 @@ async def cancel_run( @require_permission("runs", "read", owner_check=True) async def join_run(thread_id: ThreadId, run_id: str, request: Request) -> StreamingResponse: """Join an existing run's SSE stream.""" + await _require_run_visible_to_scope(run_id, thread_id, request) run_mgr = get_run_manager(request) record = await run_mgr.get(run_id) if record is None or record.thread_id != thread_id: @@ -1268,6 +1330,7 @@ async def _stream_existing_run( """ require_cancel_permission_when_action(request, action) + await _require_run_visible_to_scope(run_id, thread_id, request) run_mgr = get_run_manager(request) record = await run_mgr.get(run_id) if record is None or record.thread_id != thread_id: @@ -1368,10 +1431,11 @@ 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 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) + # Resolve the data-filter id once (None for internal callers on threads + # with established ownership — see `_run_scope_user_id`); 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, thread_id) 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( @@ -1509,7 +1573,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 _run_scope_user_id(request) + user_id = await _run_scope_user_id(request, thread_id) rows, has_more = await _scan_thread_message_page( thread_id, limit=limit, @@ -1539,6 +1603,7 @@ async def list_run_messages( Response: { data: [...], has_more: bool } """ + await _require_run_visible_to_scope(run_id, thread_id, request) event_store = get_run_event_store(request) rows = await event_store.list_messages_by_run( thread_id, @@ -1645,6 +1710,7 @@ async def get_run_artifact_archive_manifest( request: Request, ) -> ArtifactArchiveManifestResponse: """Return the verified terminal delivery count used by the archive.""" + await _require_run_visible_to_scope(run_id, thread_id, request) presented_paths = await _archive_presented_paths(thread_id, run_id, request) return ArtifactArchiveManifestResponse(file_count=len(dict.fromkeys(presented_paths))) @@ -1657,6 +1723,7 @@ async def create_run_artifact_archive( request: Request, ) -> StreamingResponse: """Download the current contents of the files presented by one terminal run.""" + await _require_run_visible_to_scope(run_id, thread_id, request) presented_paths = await _archive_presented_paths(thread_id, run_id, request) raw_owner_user_id = get_trusted_internal_owner_user_id(request) @@ -1723,6 +1790,7 @@ async def list_run_events( ``task_id`` + ``after_seq`` let the subtask card page through one subagent task's persisted steps without the run-wide ``limit`` truncating the tail (#3779). """ + await _require_run_visible_to_scope(run_id, thread_id, request) event_store = get_run_event_store(request) types = event_types.split(",") if event_types else None events = await event_store.list_events( @@ -1754,6 +1822,7 @@ async def get_run_workspace_changes( include_diff: bool = Query(default=True), ) -> dict: """Return workspace/output file changes recorded for one run.""" + await _require_run_visible_to_scope(run_id, thread_id, request) event_store = get_run_event_store(request) return await get_workspace_changes_response( event_store, @@ -1773,9 +1842,10 @@ async def thread_token_usage( ) -> ThreadTokenUsageResponse: """Thread-level token usage aggregation.""" run_store = get_run_store(request) + scope_user_id = await _run_scope_user_id(request, thread_id) if include_active: - agg = await run_store.aggregate_tokens_by_thread(thread_id, include_active=True) + agg = await run_store.aggregate_tokens_by_thread(thread_id, include_active=True, user_id=scope_user_id) else: - agg = await run_store.aggregate_tokens_by_thread(thread_id) - context_usage = await build_context_usage(request, thread_id, run_store) + agg = await run_store.aggregate_tokens_by_thread(thread_id, user_id=scope_user_id) + context_usage = await build_context_usage(request, thread_id, run_store, user_id=scope_user_id) return ThreadTokenUsageResponse(thread_id=thread_id, context_usage=context_usage, **agg) diff --git a/backend/packages/harness/deerflow/persistence/run/sql.py b/backend/packages/harness/deerflow/persistence/run/sql.py index f08d6a0dc..52ea202bc 100644 --- a/backend/packages/harness/deerflow/persistence/run/sql.py +++ b/backend/packages/harness/deerflow/persistence/run/sql.py @@ -495,7 +495,13 @@ class RunRepository(RunStore): await session.execute(update(RunRow).where(RunRow.run_id == run_id, RunRow.status == "running").values(**values)) await session.commit() - async def aggregate_tokens_by_thread(self, thread_id: str, *, include_active: bool = False) -> dict[str, Any]: + async def aggregate_tokens_by_thread( + self, + thread_id: str, + *, + include_active: bool = False, + user_id: str | None | _AutoSentinel = AUTO, + ) -> dict[str, Any]: """Aggregate token usage for a thread. ``by_model`` is reduced in Python from each row's ``token_usage_by_model`` @@ -513,6 +519,7 @@ class RunRepository(RunStore): _completed = RunRow.status.in_(statuses) _thread = RunRow.thread_id == thread_id _run_operation = RunRow.operation_kind == "run" + resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.aggregate_tokens_by_thread") stmt = select( RunRow.model_name, @@ -524,6 +531,8 @@ class RunRepository(RunStore): RunRow.middleware_tokens, RunRow.token_usage_by_model, ).where(_thread, _run_operation, _completed) + if resolved_user_id is not None: + stmt = stmt.where(RunRow.user_id == resolved_user_id) async with self._sf() as session: rows = (await session.execute(stmt)).all() diff --git a/backend/packages/harness/deerflow/runtime/runs/store/base.py b/backend/packages/harness/deerflow/runtime/runs/store/base.py index 45316b932..f8f33489d 100644 --- a/backend/packages/harness/deerflow/runtime/runs/store/base.py +++ b/backend/packages/harness/deerflow/runtime/runs/store/base.py @@ -299,7 +299,7 @@ class RunStore(abc.ABC): pass @abc.abstractmethod - async def aggregate_tokens_by_thread(self, thread_id: str, *, include_active: bool = False) -> dict[str, Any]: + async def aggregate_tokens_by_thread(self, thread_id: str, *, include_active: bool = False, user_id: str | None = None) -> dict[str, Any]: """Aggregate token usage for completed runs in a thread. Returns a dict with keys: total_tokens, total_input_tokens, diff --git a/backend/packages/harness/deerflow/runtime/runs/store/memory.py b/backend/packages/harness/deerflow/runtime/runs/store/memory.py index 4c91d45f7..8c2fe05c4 100644 --- a/backend/packages/harness/deerflow/runtime/runs/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/runs/store/memory.py @@ -252,12 +252,12 @@ class MemoryRunStore(RunStore): results.sort(key=lambda r: r["created_at"]) return results - async def aggregate_tokens_by_thread(self, thread_id: str, *, include_active: bool = False) -> dict[str, Any]: + async def aggregate_tokens_by_thread(self, thread_id: str, *, include_active: bool = False, user_id: str | None = None) -> dict[str, Any]: statuses = ("success", "error", "running") if include_active else ("success", "error") # Use the thread index for an O(runs-in-thread) lookup instead of # scanning every run in the process (mirrors ``list_by_thread``). run_ids = self._runs_by_thread.get(thread_id) or () - completed = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and run.get("operation_kind", "run") == "run" and run.get("status") in statuses] + completed = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and run.get("operation_kind", "run") == "run" and run.get("status") in statuses and (user_id is None or run.get("user_id") == user_id)] by_model: dict[str, dict] = {} for r in completed: usage_by_model = r.get("token_usage_by_model") or {} diff --git a/backend/tests/test_thread_runs_internal_scope.py b/backend/tests/test_thread_runs_internal_scope.py index 4bb65fd57..f6aea056e 100644 --- a/backend/tests/test_thread_runs_internal_scope.py +++ b/backend/tests/test_thread_runs_internal_scope.py @@ -19,6 +19,7 @@ from uuid import UUID import pytest from fastapi import FastAPI, HTTPException, Request, Response from fastapi.testclient import TestClient +from langgraph.store.memory import InMemoryStore from starlette.middleware.base import BaseHTTPMiddleware from app.gateway.auth.models import User @@ -26,6 +27,7 @@ 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.persistence.thread_meta.memory import MemoryThreadMetaStore from deerflow.runtime.events.store.memory import MemoryRunEventStore from deerflow.runtime.runs.manager import RunManager from deerflow.runtime.runs.store.memory import MemoryRunStore @@ -62,11 +64,18 @@ class _ScopeAuthMiddleware(BaseHTTPMiddleware): class _PermissiveThreadStore: - """Stands in for the thread store behind ``owner_check=True``.""" + """Stands in for the thread store behind ``owner_check=True``. + + The existing scope tests exercise the established-ownership path, so + ``get`` reports an existing, owner-established meta row. + """ async def check_access(self, _thread_id: str, _user_id: str, *, require_existing: bool = False) -> bool: return True + async def get(self, _thread_id: str, *, user_id: str | None | object = None) -> dict | None: + return {"thread_id": THREAD_ID, "user_id": "established-owner"} + class _RecordingRunStore(MemoryRunStore): """Records the per-user filter identity each read resolves to. @@ -141,10 +150,12 @@ def _make_app( run_store: MemoryRunStore, event_store: MemoryRunEventStore | None = None, feedback_repo: _RecordingFeedbackRepo | None = None, + thread_store=None, ) -> TestClient: app = FastAPI() app.add_middleware(_ScopeAuthMiddleware, user=user, auth_source=auth_source) - app.state.thread_store = _PermissiveThreadStore() + app.state.thread_store = thread_store if thread_store is not None else _PermissiveThreadStore() + app.state.run_store = run_store app.state.run_manager = RunManager(store=run_store) if event_store is not None: app.state.run_event_store = event_store @@ -319,16 +330,294 @@ def test_browser_session_messages_keep_per_user_filter(mixed_owner_store: Memory # --------------------------------------------------------------------------- +# owner isolation on threads without established metadata (#5448 review P1) +# --------------------------------------------------------------------------- + + +def test_missing_thread_meta_keeps_owner_isolation_for_internal_callers() -> None: + """owner_check also authorizes missing-meta (legacy shared) threads. + + There, unfiltered reads would expose other users' persisted runs to the + acting owner's internal caller, so the raw trusted owner stays the filter + — the exact value ``start_run`` stamps on run rows (#5448 review P1). + """ + thread_store = MemoryThreadMetaStore(InMemoryStore()) # no metadata row at all + run_store = _RecordingRunStore() + _seed_run(run_store, "run-other-user", user_id=str(BROWSER_USER_ID), status="success") + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + thread_store=thread_store, + ) + + with client: + listed = client.get( + f"/api/threads/{THREAD_ID}/runs", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + page = client.get( + f"/api/threads/{THREAD_ID}/runs/page", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + single = client.get( + f"/api/threads/{THREAD_ID}/runs/run-other-user", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert listed.status_code == 200 + assert listed.json() == [] + assert page.status_code == 200 + assert page.json()["data"] == [] + assert single.status_code == 404 + # The acting owner's own raw-stamped runs remain visible: seed one and + # confirm it comes back through the same endpoints. + owned_store = _RecordingRunStore() + _seed_run(owned_store, "run-own-owner-stamp", user_id=OWNER_RAW, status="success") + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=owned_store, + thread_store=thread_store, + ) + with client: + own = client.get( + f"/api/threads/{THREAD_ID}/runs/run-own-owner-stamp", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + assert own.status_code == 200 + assert own.json()["run_id"] == "run-own-owner-stamp" + + +def test_null_owner_thread_meta_keeps_owner_isolation_for_internal_callers() -> None: + """NULL-owner meta rows (shared/pre-auth data) isolate by raw owner too.""" + thread_store = MemoryThreadMetaStore(InMemoryStore()) + asyncio.run( + thread_store.create( + THREAD_ID, + assistant_id=None, + user_id=None, # shared / pre-auth: meta row exists with NULL owner + ) + ) + run_store = _RecordingRunStore() + _seed_run(run_store, "run-other-user", user_id=str(BROWSER_USER_ID), status="success") + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + thread_store=thread_store, + ) + + with client: + listed = client.get( + f"/api/threads/{THREAD_ID}/runs", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + single = client.get( + f"/api/threads/{THREAD_ID}/runs/run-other-user", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert listed.status_code == 200 + assert listed.json() == [] + assert single.status_code == 404 + + +def test_established_ownership_still_reads_thread_runs_unfiltered(mixed_owner_store: MemoryRunStore) -> None: + """Established meta ownership keeps the #5437 unfiltered-read behavior.""" + thread_store = MemoryThreadMetaStore(InMemoryStore()) + asyncio.run(thread_store.create(THREAD_ID, assistant_id=None, user_id=OWNER_RAW)) + run_store = _RecordingRunStore() + _seed_run(run_store, RUN_OWNER, user_id=OWNER_RAW, status="success") + _seed_run(run_store, "run-legacy-default", user_id="default", status="success") + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + thread_store=thread_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_OWNER, "run-legacy-default"} + + +def test_ownerless_internal_caller_default_filter_on_missing_meta() -> None: + """Without an owner header the synthetic "default" identity is the filter. + + Pins the owner-less fallback branch of ``_run_scope_user_id``: a run + stamped with another owner's raw id stays hidden on missing-meta threads. + """ + thread_store = MemoryThreadMetaStore(InMemoryStore()) # no meta row + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + + client = _make_app( + user=_internal_user(None), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + thread_store=thread_store, + ) + + with client: + response = client.get(f"/api/threads/{THREAD_ID}/runs") + + assert response.status_code == 200 + assert response.json() == [] + + +def test_subresource_reads_stay_owner_isolated_without_meta() -> None: + """Run-scoped sub-resources must respect the acting owner's stamp. + + These reads query by ``(thread_id, run_id)`` with no per-user filter of + their own; on missing-meta threads an internal caller acting for owner A + could otherwise read owner B's run content by id (#5448 review P1 + follow-up). + """ + thread_store = MemoryThreadMetaStore(InMemoryStore()) # no meta row + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + event_store = MemoryRunEventStore() + _seed_message(event_store, "run-owner-777", "msg-owner-run") + + stranger_headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: "feishu:owner-999"} + owner_headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW} + base = f"/api/threads/{THREAD_ID}/runs/run-owner-777" + + stranger = _make_app( + user=_internal_user("feishu:owner-999"), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=event_store, + thread_store=thread_store, + ) + owner_client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=event_store, + thread_store=thread_store, + ) + + with stranger: + assert stranger.get(base + "/messages", headers=stranger_headers).status_code == 404 + assert stranger.get(base + "/events", headers=stranger_headers).status_code == 404 + assert stranger.get(base + "/workspace-changes", headers=stranger_headers).status_code == 404 + assert stranger.get(base + "/join", headers=stranger_headers).status_code == 404 + + with owner_client: + messages = owner_client.get(base + "/messages", headers=owner_headers) + events = owner_client.get(base + "/events", headers=owner_headers) + + assert messages.status_code == 200 + assert [row["content"]["id"] for row in messages.json()["data"]] == ["msg-owner-run"] + assert events.status_code == 200 + assert any(event.get("run_id") == "run-owner-777" for event in events.json()) + + +def test_null_owner_thread_gates_cancel_and_archive_for_internal_callers() -> None: + """NULL-owner meta rows gate POST /cancel and the archive pair too. + + The round-2 findings: cancel resolved runs unscoped (an interrupt-vs-join + inconsistency) and the archive manifest leaked the other owner's + delivered-file count plus a 200-vs-409 delivery oracle on shared threads. + """ + thread_store = MemoryThreadMetaStore(InMemoryStore()) + asyncio.run(thread_store.create(THREAD_ID, assistant_id=None, user_id=None)) + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + event_store = MemoryRunEventStore() + + stranger_headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: "feishu:owner-999"} + stranger = _make_app( + user=_internal_user("feishu:owner-999"), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=event_store, + thread_store=thread_store, + ) + + with stranger: + cancel = stranger.post( + f"/api/threads/{THREAD_ID}/runs/run-owner-777/cancel?action=interrupt", + headers=stranger_headers, + ) + manifest = stranger.get( + f"/api/threads/{THREAD_ID}/runs/run-owner-777/artifacts/archive", + headers=stranger_headers, + ) + archive = stranger.post( + f"/api/threads/{THREAD_ID}/runs/run-owner-777/artifacts/archive", + headers=stranger_headers, + ) + + assert cancel.status_code == 404 + assert manifest.status_code == 404 + assert archive.status_code == 404 + + +def test_null_owner_thread_matching_owner_cancels_and_reads_manifest() -> None: + """The acting owner keeps cancel and archive access on shared threads.""" + thread_store = MemoryThreadMetaStore(InMemoryStore()) + asyncio.run(thread_store.create(THREAD_ID, assistant_id=None, user_id=None)) + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + event_store = MemoryRunEventStore() + asyncio.run( + event_store.put( + thread_id=THREAD_ID, + run_id="run-owner-777", + event_type="run.delivery", + category="outputs", + content={"presented": 2, "by_tool": {"present_files": ["/mnt/user-data/outputs/a.txt", "/mnt/user-data/outputs/b.txt"]}}, + ) + ) + + owner_client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + event_store=event_store, + thread_store=thread_store, + ) + + with owner_client: + manifest = owner_client.get( + f"/api/threads/{THREAD_ID}/runs/run-owner-777/artifacts/archive", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + cancel = owner_client.post( + f"/api/threads/{THREAD_ID}/runs/run-owner-777/cancel?action=interrupt", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert manifest.status_code == 200 + assert manifest.json() == {"file_count": 2} + # A terminal run cannot be cancelled again: the acting owner reaches the + # real conflict path instead of a 404 anti-enumeration answer. + assert cancel.status_code == 409 + + # edit/regenerate helper fallback paths (#5482) # --------------------------------------------------------------------------- -def _helper_request(*, user, auth_source: str, run_store, event_store): - """Minimal Request stand-in: the helpers only touch state and app.state.""" +def _helper_request(*, user, auth_source: str, run_store, event_store, owner_header: str | None = None): + """Minimal Request stand-in: the helpers touch state, app.state and headers.""" app_state = SimpleNamespace(run_manager=RunManager(store=run_store), run_event_store=event_store) + headers = {INTERNAL_OWNER_USER_ID_HEADER_NAME: owner_header} if owner_header else {} return SimpleNamespace( state=SimpleNamespace(user=user, auth_source=auth_source), app=SimpleNamespace(state=app_state), + headers=headers, ) @@ -342,12 +631,13 @@ def test_helper_fallback_paths_resolve_internal_caller_runs() -> None: auth_source=AUTH_SOURCE_INTERNAL, run_store=store, event_store=MemoryRunEventStore(), + owner_header=OWNER_RAW, ) # The owner-stamped interrupted run resolves through the raw owner stamp. interrupted = asyncio.run(thread_runs._find_interrupted_target_run_id(THREAD_ID, {"additional_kwargs": {"run_id": RUN_OWNER}}, request)) assert interrupted == RUN_OWNER - assert store.get_user_ids[-1] is None + assert store.get_user_ids[-1] == OWNER_RAW # An interrupted run is not an editable source run, but the lookup itself # must have reached it (409 for status, not for a missing record). @@ -355,10 +645,11 @@ def test_helper_fallback_paths_resolve_internal_caller_runs() -> None: asyncio.run(thread_runs._require_successful_source_run(THREAD_ID, RUN_OWNER, request)) assert exc.value.status_code == 409 assert "successful" in exc.value.detail - assert store.get_user_ids[-1] is None + assert store.get_user_ids[-1] == OWNER_RAW # Fallback scan without any event-store or kwargs anchors still scans the - # authorized thread unfiltered (and 409s on the miss). + # authorized thread through the acting owner's raw-stamp scope (and 409s + # on the miss). with pytest.raises(HTTPException) as exc2: asyncio.run( thread_runs._find_target_run_id( @@ -370,7 +661,7 @@ def test_helper_fallback_paths_resolve_internal_caller_runs() -> None: ) ) assert exc2.value.status_code == 409 - assert store.list_by_thread_user_ids and store.list_by_thread_user_ids[-1] is None + assert store.list_by_thread_user_ids and store.list_by_thread_user_ids[-1] == OWNER_RAW def test_helper_fallback_paths_keep_per_user_filter_for_browser_sessions() -> None: @@ -396,3 +687,94 @@ def test_helper_fallback_paths_keep_per_user_filter_for_browser_sessions() -> No asyncio.run(thread_runs._require_successful_source_run(THREAD_ID, RUN_OWNER, request)) assert exc.value.status_code == 409 assert store.get_user_ids[-1] == str(BROWSER_USER_ID) + + +def test_token_usage_isolated_without_meta_for_internal_callers() -> None: + """Token-usage aggregate honors the acting owner's raw stamp (#5484 r4).""" + thread_store = MemoryThreadMetaStore(InMemoryStore()) # no meta row + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + _seed_run(run_store, "run-other-user", user_id=str(BROWSER_USER_ID), status="success") + run_store._runs["run-owner-777"]["token_usage_by_model"] = {"gpt-x": {"total_tokens": 111}} + run_store._runs["run-owner-777"]["total_tokens"] = 111 + run_store._runs["run-other-user"]["token_usage_by_model"] = {"gpt-x": {"total_tokens": 999}} + run_store._runs["run-other-user"]["total_tokens"] = 999 + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + thread_store=thread_store, + ) + + with client: + response = client.get( + f"/api/threads/{THREAD_ID}/token-usage", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert response.status_code == 200 + body = response.json() + assert body["total_tokens"] == 111 + assert body["total_runs"] == 1 + + +def test_token_usage_narrowed_for_browser_sessions_without_meta() -> None: + """Browser sessions on shared threads see only their own spend too.""" + thread_store = MemoryThreadMetaStore(InMemoryStore()) # no meta row + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + _seed_run(run_store, "run-browser", user_id=str(BROWSER_USER_ID), status="success") + run_store._runs["run-owner-777"]["token_usage_by_model"] = {"gpt-x": {"total_tokens": 111}} + run_store._runs["run-owner-777"]["total_tokens"] = 111 + run_store._runs["run-browser"]["token_usage_by_model"] = {"gpt-x": {"total_tokens": 222}} + run_store._runs["run-browser"]["total_tokens"] = 222 + + client = _make_app( + user=_browser_user(), + auth_source=AUTH_SOURCE_SESSION, + run_store=run_store, + thread_store=thread_store, + ) + + with client: + response = client.get(f"/api/threads/{THREAD_ID}/token-usage") + + assert response.status_code == 200 + assert response.json()["total_tokens"] == 222 + + +def test_token_usage_unfiltered_on_established_ownership_for_internal_callers() -> None: + """Established meta ownership keeps the unfiltered aggregate. + + Pins the other half of the scoping contract: on an established thread the + internal caller's aggregate folds runs stamped by different identities + (the store must receive ``user_id=None``), mirroring + ``test_established_ownership_still_reads_thread_runs_unfiltered``. + """ + thread_store = MemoryThreadMetaStore(InMemoryStore()) + asyncio.run(thread_store.create(THREAD_ID, assistant_id=None, user_id=OWNER_RAW)) + run_store = _RecordingRunStore() + _seed_run(run_store, "run-owner-777", user_id=OWNER_RAW, status="success") + _seed_run(run_store, "run-legacy-default", user_id="default", status="success") + run_store._runs["run-owner-777"]["token_usage_by_model"] = {"gpt-x": {"total_tokens": 111}} + run_store._runs["run-owner-777"]["total_tokens"] = 111 + run_store._runs["run-legacy-default"]["token_usage_by_model"] = {"gpt-x": {"total_tokens": 55}} + run_store._runs["run-legacy-default"]["total_tokens"] = 55 + + client = _make_app( + user=_internal_user(OWNER_RAW), + auth_source=AUTH_SOURCE_INTERNAL, + run_store=run_store, + thread_store=thread_store, + ) + + with client: + response = client.get( + f"/api/threads/{THREAD_ID}/token-usage", + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: OWNER_RAW}, + ) + + assert response.status_code == 200 + body = response.json() + assert body["total_tokens"] == 166 # both stamps fold when ownership is established diff --git a/backend/tests/test_thread_token_usage.py b/backend/tests/test_thread_token_usage.py index 636d453f0..c19b1d62a 100644 --- a/backend/tests/test_thread_token_usage.py +++ b/backend/tests/test_thread_token_usage.py @@ -58,7 +58,7 @@ def test_thread_token_usage_returns_stable_shape(monkeypatch: pytest.MonkeyPatch **_aggregate_result(), "context_usage": None, } - run_store.aggregate_tokens_by_thread.assert_awaited_once_with("thread-1") + run_store.aggregate_tokens_by_thread.assert_awaited_once_with("thread-1", user_id=None) build_context_usage.assert_awaited_once() @@ -72,7 +72,7 @@ def test_thread_token_usage_can_include_active_runs(monkeypatch: pytest.MonkeyPa response = client.get("/api/threads/thread-1/token-usage?include_active=true") assert response.status_code == 200 - run_store.aggregate_tokens_by_thread.assert_awaited_once_with("thread-1", include_active=True) + run_store.aggregate_tokens_by_thread.assert_awaited_once_with("thread-1", include_active=True, user_id=None) def test_thread_token_usage_serializes_context_percentage(monkeypatch: pytest.MonkeyPatch) -> None: