diff --git a/README.md b/README.md index 362ebb3fe..cdd9843ae 100644 --- a/README.md +++ b/README.md @@ -799,6 +799,14 @@ LangSmith and Langfuse attach as LangChain callbacks, so you can enable both and For Docker deployments, tracing is disabled by default. Set `LANGSMITH_TRACING=true` and `LANGSMITH_API_KEY` in your `.env` to enable it. +#### Existing-Run Stream Actions + +Existing-run SSE joins are observation-only on `GET`: supplying +`action=interrupt|rollback` returns `405`. Cancellation on this stream route is +`POST`-only and requires the `runs:cancel` permission. Accordingly, the OpenAPI +contract exposes `action` and `wait` only on `POST`; the `GET` operation exposes +only its path parameters. + #### Personal Access Tokens Non-interactive clients (CI pipelines, scripts, server-to-server integrations) diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 6929040c8..bb23f8fd7 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -57,7 +57,7 @@ owner-scoped assistant version selection remains enabled. | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update explicitly. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | -| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, an empty run-event message feed is seeded from an existing checkpoint head so legacy checkpoint-only history receives earlier thread-global sequence numbers and remains visible after the new run; a thread with no checkpoint or an already-populated feed skips this compatibility path. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`. | +| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty event feed from an existing checkpoint so legacy checkpoint-only history keeps earlier thread-global ordering and stays visible; skip without a checkpoint or when the feed is populated. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`. | | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | | **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback | | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index f257820c5..66b5b3d3a 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -18,7 +18,7 @@ from copy import deepcopy from datetime import UTC, datetime from typing import Any, Literal -from fastapi import APIRouter, HTTPException, Query, Request +from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import Response, StreamingResponse from langchain_core.messages import BaseMessage from pydantic import BaseModel, Field @@ -1019,26 +1019,34 @@ async def join_run(thread_id: ThreadId, run_id: str, request: Request) -> Stream ) -# Register GET and POST as separate routes so each method gets a unique OpenAPI -# operationId. ``api_route(methods=["GET", "POST"])`` shares one route registration -# across both methods, which makes FastAPI emit the same ``operationId`` twice and -# warn about a duplicate operation id during OpenAPI generation. -@router.get("/{thread_id}/runs/{run_id}/stream", response_model=None) -@router.post("/{thread_id}/runs/{run_id}/stream", response_model=None) -@require_permission("runs", "read", owner_check=True) -async def stream_existing_run( +def _reject_get_stream_action( + action: Literal["interrupt", "rollback"] | None = Query(default=None, include_in_schema=False), +) -> None: + """Keep the GET join read-only before thread ownership or run lookup.""" + if action is not None: + # SameSite=Lax still sends the session cookie on a cross-site top-level + # safe navigation. Reject the state-changing action before the endpoint + # wrapper performs its thread ownership lookup. + raise HTTPException( + status_code=405, + detail="`action` is only supported on POST requests", + headers={"Allow": "POST"}, + ) + + +async def _stream_existing_run( thread_id: ThreadId, run_id: str, request: Request, - action: Literal["interrupt", "rollback"] | None = Query(default=None, description="Cancel action"), - wait: int = Query(default=0, description="Block until cancelled (1) or return immediately (0)"), -): - """Join an existing run's SSE stream (GET), or cancel-then-stream (POST). + *, + action: Literal["interrupt", "rollback"] | None, + wait: int, +) -> Response: + """Join an existing run's SSE stream, optionally cancelling it first. - The LangGraph SDK's ``joinStream`` and ``useStream`` stop button both use - ``POST`` to this endpoint. When ``action=interrupt`` or ``action=rollback`` - is present the run is cancelled first; the response then streams any - remaining buffered events so the client observes a clean shutdown. + When ``action=interrupt`` or ``action=rollback`` is present the run is + cancelled first; the response then streams any remaining buffered events + so the client observes a clean shutdown. """ require_cancel_permission_when_action(request, action) @@ -1099,6 +1107,34 @@ async def stream_existing_run( ) +# Register POST before GET to preserve the historical route precedence and +# Allow header, while separate signatures keep cancel-only parameters off the +# GET schema. The shared route name keeps generated operationIds stable. +@router.post("/{thread_id}/runs/{run_id}/stream", response_model=None, name="stream_existing_run") +@require_permission("runs", "read", owner_check=True) +async def stream_existing_run( + thread_id: ThreadId, + run_id: str, + request: Request, + action: Literal["interrupt", "rollback"] | None = Query(default=None, description="Cancel action"), + wait: int = Query(default=0, description="Block until cancelled (1) or return immediately (0)"), +) -> Response: + """Join an existing run's SSE stream, optionally cancelling it first.""" + return await _stream_existing_run(thread_id, run_id, request, action=action, wait=wait) + + +@router.get( + "/{thread_id}/runs/{run_id}/stream", + response_model=None, + dependencies=[Depends(_reject_get_stream_action)], + name="stream_existing_run", +) +@require_permission("runs", "read", owner_check=True) +async def join_existing_run_stream(thread_id: ThreadId, run_id: str, request: Request) -> Response: + """Join an existing run's observation-only SSE stream.""" + return await _stream_existing_run(thread_id, run_id, request, action=None, wait=0) + + # --------------------------------------------------------------------------- # Messages / Events / Token usage endpoints # --------------------------------------------------------------------------- diff --git a/backend/tests/test_openapi_operation_ids.py b/backend/tests/test_openapi_operation_ids.py index d6bb46162..318f06606 100644 --- a/backend/tests/test_openapi_operation_ids.py +++ b/backend/tests/test_openapi_operation_ids.py @@ -77,3 +77,17 @@ def test_stream_existing_run_exposes_distinct_get_and_post(openapi_spec: dict) - post_op_id = path_item["post"].get("operationId") assert get_op_id and post_op_id, "Both GET and POST must have operationIds" assert get_op_id != post_op_id, f"GET and POST share operationId {get_op_id!r}, which breaks OpenAPI codegen" + + +def test_stream_existing_run_exposes_method_specific_query_parameters(openapi_spec: dict) -> None: + """Only POST advertises the cancel-then-stream query contract.""" + path = "/api/threads/{thread_id}/runs/{run_id}/stream" + path_item = openapi_spec["paths"][path] + + get_parameters = {(parameter["in"], parameter["name"]) for parameter in path_item["get"].get("parameters", [])} + post_parameters = {(parameter["in"], parameter["name"]) for parameter in path_item["post"].get("parameters", [])} + + assert ("query", "action") not in get_parameters + assert ("query", "wait") not in get_parameters + assert ("query", "action") in post_parameters + assert ("query", "wait") in post_parameters diff --git a/backend/tests/test_pat_auth.py b/backend/tests/test_pat_auth.py index 8d460c15c..eeaa05c4b 100644 --- a/backend/tests/test_pat_auth.py +++ b/backend/tests/test_pat_auth.py @@ -1,8 +1,9 @@ """Integration tests for PAT authentication (#4849). Covers credential precedence in AuthMiddleware, the CSRF boundary for -Bearer-authenticated requests, scope intersection, PAT management routes, -and the self-protection rules (a PAT may not manage PATs or auth state). +Bearer-authenticated requests and for the safe-method stream join (#5092), +scope intersection, PAT management routes, and the self-protection rules +(a PAT may not manage PATs or auth state). """ from __future__ import annotations @@ -12,7 +13,7 @@ from datetime import UTC, datetime, timedelta from types import SimpleNamespace import pytest -from fastapi import FastAPI, Request +from fastapi import Depends, FastAPI, Request from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from starlette.testclient import TestClient @@ -115,6 +116,20 @@ def _make_pat_app(with_pat_repo: bool = True): require_cancel_permission_when_action(request, action) return {"ok": True} + # Mirrors the real GET-only join surface (thread_runs.py + # join_existing_run_stream): registers the production route dependency + # that rejects cancel actions before thread ownership or run lookup, so + # the guard is exercised through the production middleware order above. + from app.gateway.routers.thread_runs import _reject_get_stream_action + + @app.get( + "/api/threads/{thread_id}/runs/{run_id}/stream", + dependencies=[Depends(_reject_get_stream_action)], + ) + @require_permission("runs", "read") + async def join_stream(thread_id: str, run_id: str, request: Request): + return {"ok": True} + # Mirrors the real run-creation entrypoints (thread_runs.py / runs.py): # runs:create at the decorator, plus the cancel-capability gate that # start_run applies to mutating multitask strategies. RunCreateRequest is @@ -311,6 +326,26 @@ def test_auth_endpoint_origin_check_not_bypassed_by_bearer(client): assert response.json()["detail"] == "Cross-site auth request denied." +def test_session_get_stream_action_dies_at_route_gate_not_csrf(client): + """#5092 defence-in-depth, end-to-end through the production middleware + order: SameSite=Lax still attaches the session cookie to a cross-site + top-level GET, and CSRF exempts safe methods — so the route gate is the + only thing standing between that navigation and a run cancel. An + authenticated GET with ?action=interrupt is answered 405 + Allow: POST by + the real _reject_get_stream_action dependency, while the same + unauthenticated GET dies at AuthMiddleware's 401 before any route logic + runs.""" + _session_cookie(client) + denied = client.get("/api/threads/t1/runs/run-1/stream?action=interrupt") + assert denied.status_code == 405 + assert denied.headers["allow"] == "POST" + assert denied.json()["detail"] == "`action` is only supported on POST requests" + + client.cookies.clear() + unauthed = client.get("/api/threads/t1/runs/run-1/stream?action=interrupt") + assert unauthed.status_code == 401 + + # ── Management routes + self-protection (#4849 point 6) ─────────────────── diff --git a/backend/tests/test_sse_observer_disconnect.py b/backend/tests/test_sse_observer_disconnect.py index b45ccbbb1..99e90db97 100644 --- a/backend/tests/test_sse_observer_disconnect.py +++ b/backend/tests/test_sse_observer_disconnect.py @@ -104,7 +104,7 @@ def test_join_routes_wire_sse_consumer_as_observers(): from app.gateway.routers import thread_runs thread_runs_source = inspect.getsource(thread_runs) - # join_run + stream_existing_run (GET and POST share one handler) + # join_run + the shared existing-run stream implementation assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)") == 2 # stream_run — the creator's create-and-stream endpoint assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr),") == 1 diff --git a/backend/tests/test_stream_get_action_rejected.py b/backend/tests/test_stream_get_action_rejected.py new file mode 100644 index 000000000..ac53d0512 --- /dev/null +++ b/backend/tests/test_stream_get_action_rejected.py @@ -0,0 +1,116 @@ +"""GET on the join-stream route must not carry cancel actions. + +The existing-run stream path supports both GET and POST; POST's ``action`` +branch cancels the run. The CSRF middleware exempts GET, while a SameSite=Lax +session cookie still accompanies a cross-site top-level safe navigation. An +attacker-induced navigation to +``GET .../runs/{run_id}/stream?action=interrupt|rollback`` was therefore a +state-changing GET that bypassed the CSRF protection guarding the POST +variant. These tests pin that GET stays a read-only join and POST keeps +cancelling. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from _router_auth_helpers import make_authed_test_app +from fastapi.testclient import TestClient + +from app.gateway.routers import thread_runs +from deerflow.runtime import RunManager, RunStatus +from deerflow.runtime.stream_bridge import MemoryStreamBridge + +THREAD_ID = "thread-get-action" + + +def _make_seeded_run_client(run_status: RunStatus = RunStatus.running) -> tuple[TestClient, RunManager, str]: + mgr = RunManager() + + async def _seed(): + record = await mgr.create(THREAD_ID) + await mgr.set_status(record.run_id, run_status) + return record.run_id + + run_id = asyncio.run(_seed()) + app = make_authed_test_app() + app.include_router(thread_runs.router) + app.state.run_manager = mgr + app.state.stream_bridge = MemoryStreamBridge() + return TestClient(app, raise_server_exceptions=False), mgr, run_id + + +def _get_run_status(mgr: RunManager, run_id: str) -> RunStatus: + async def _read_status() -> RunStatus: + record = await mgr.get(run_id) + assert record is not None + return record.status + + return asyncio.run(_read_status()) + + +@pytest.mark.parametrize("action", ("interrupt", "rollback")) +def test_get_with_cancel_action_is_rejected(action: str): + """GET + action=interrupt|rollback must answer 405, not cancel.""" + client, mgr, run_id = _make_seeded_run_client() + response = client.get(f"/api/threads/{THREAD_ID}/runs/{run_id}/stream?action={action}") + assert response.status_code == 405 + assert response.headers["allow"] == "POST" + assert "POST" in response.json()["detail"] + assert _get_run_status(mgr, run_id) == RunStatus.running + + +def test_get_with_invalid_action_has_one_validation_error(): + """The dependency must not duplicate the endpoint's query validation.""" + client, mgr, run_id = _make_seeded_run_client() + + response = client.get(f"/api/threads/{THREAD_ID}/runs/{run_id}/stream?action=invalid") + + assert response.status_code == 422 + assert len(response.json()["detail"]) == 1 + assert _get_run_status(mgr, run_id) == RunStatus.running + + +def test_unsupported_method_preserves_post_allow_header(): + """Splitting the handlers must not change Starlette's route precedence.""" + client, _, run_id = _make_seeded_run_client() + + response = client.put(f"/api/threads/{THREAD_ID}/runs/{run_id}/stream") + + assert response.status_code == 405 + assert response.headers["allow"] == "POST" + + +def test_get_with_action_is_rejected_before_owner_lookup(): + """The method gate must not reveal whether a thread metadata row exists.""" + app = make_authed_test_app(owner_check_passes=False) + app.include_router(thread_runs.router) + client = TestClient(app, raise_server_exceptions=False) + + response = client.get(f"/api/threads/{THREAD_ID}/runs/missing-run/stream?action=interrupt") + + assert response.status_code == 405 + assert response.headers["allow"] == "POST" + app.state.thread_store.check_access.assert_not_awaited() + + +def test_get_without_action_still_joins(): + """The method guard must not break the plain read-only GET join. The + seeded run is terminal so the SSE stream emits `end` and completes.""" + client, _, run_id = _make_seeded_run_client(run_status=RunStatus.success) + with client.stream("GET", f"/api/threads/{THREAD_ID}/runs/{run_id}/stream") as response: + assert response.status_code == 200 + events = [line for line in response.iter_lines() if line.startswith("event:")] + + assert events[-1].strip() == "event: end" + + +@pytest.mark.parametrize("action", ("interrupt", "rollback")) +def test_post_with_cancel_action_still_cancels(action: str): + """The documented POST cancel-then-stream flow is unchanged.""" + client, mgr, run_id = _make_seeded_run_client() + with client.stream("POST", f"/api/threads/{THREAD_ID}/runs/{run_id}/stream?action={action}") as response: + assert response.status_code == 200 + + assert _get_run_status(mgr, run_id) == RunStatus.interrupted