diff --git a/backend/app/gateway/conversation_access.py b/backend/app/gateway/conversation_access.py index 213aa55a6d..7aadc333b0 100644 --- a/backend/app/gateway/conversation_access.py +++ b/backend/app/gateway/conversation_access.py @@ -12,6 +12,7 @@ from urllib.parse import urlsplit from fastapi import HTTPException, Request from app.gateway.conversation_reader import read_visible_message_page +from app.gateway.csrf_middleware import _normalize_origin, _request_origin from deerflow.constants import CONVERSATION_TOOL_NAME, CONVERSATION_TOOL_USE from deerflow.utils.llm_text import strip_think_blocks from deerflow.utils.thread_id import validate_thread_id @@ -43,13 +44,20 @@ _BUDGET_TOO_SMALL = { } -def _source_id(reference: str, request_url: str) -> str: - """URLs are same-origin local selectors, never network fetch targets.""" +def _source_id(reference: str, request_origin: str | None) -> str: + """URLs are local selectors on this DeerFlow's host, never network fetch targets. + + *request_origin* is the origin the client addressed, as the CSRF check + resolves it (forwarded host and port included). Only the host and port are + compared, not the scheme: behind a TLS proxy Gateway can see ``http`` for a + browser's ``https`` link, because uvicorn ignores ``X-Forwarded-Proto`` + from an untrusted peer and a local nginx resends its own scheme. + """ if "://" not in reference: return validate_thread_id(reference) parsed = urlsplit(reference) - origin = urlsplit(request_url) - if parsed.scheme not in {"http", "https"} or (parsed.scheme, parsed.netloc) != (origin.scheme, origin.netloc) or parsed.query or parsed.fragment: + origin = _normalize_origin(f"{parsed.scheme}://{parsed.netloc}") + if origin is None or request_origin is None or urlsplit(origin).netloc != urlsplit(request_origin).netloc or parsed.query or parsed.fragment: raise ValueError("Use a thread ID or a conversation URL from this DeerFlow origin") match = re.fullmatch(r"/workspace/(?:agents/[^/]+/)?chats/([^/]+)", parsed.path) if match is None: @@ -181,7 +189,8 @@ def prepare_conversation_reader( if auth is None or not auth.is_authenticated or not auth.has_permission("runs", "read") or not user_id: raise HTTPException(status_code=403, detail="Permission denied: runs:read") try: - ids = tuple(dict.fromkeys(_source_id(reference, str(request.url)) for reference in references)) + request_origin = _request_origin(request) + ids = tuple(dict.fromkeys(_source_id(reference, request_origin) for reference in references)) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc return _prepare_owner_bound_conversation_reader(ids, user_id=user_id, run_context=run_context, run_manager=run_manager, app_config=app_config) diff --git a/backend/app/gateway/run_models.py b/backend/app/gateway/run_models.py index 36353110cb..9a9bd8984e 100644 --- a/backend/app/gateway/run_models.py +++ b/backend/app/gateway/run_models.py @@ -40,7 +40,7 @@ class RunCreateRequest(BaseModel): conversation_references: list[ConversationReference] = Field( default_factory=list, max_length=MAX_CONVERSATION_REFERENCES, - description="Explicit thread IDs or same-origin chat URLs readable only during this run (opt-in read_conversation tool); SDK clients may send the same list as context.conversation_references", + description="Explicit thread IDs or chat URLs on this DeerFlow host readable only during this run (opt-in read_conversation tool); SDK clients may send the same list as context.conversation_references", ) webhook: None = Field(default=None, description="Compatibility placeholder; completion callbacks are not supported") checkpoint_id: str | None = Field(default=None, description="Resume from checkpoint") diff --git a/backend/docs/API.md b/backend/docs/API.md index 8b23eab12c..b8aa522186 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -510,10 +510,15 @@ Gateway API callers can attach up to three explicit references to create/stream/ A reference is a valid thread ID or an absolute `/workspace/chats/{thread_id}` URL (also `/workspace/agents/{agent_name}/chats/{thread_id}` for custom agents) -with the same scheme and authority as the run request, without query or fragment. -URLs are parsed as local selectors and are never fetched. For split-origin clients -or internal proxies, pass the thread ID. The field is separate from message text: -links in pasted documents, tool results, or previous messages grant no access. +on the host and port the client sent the run request to, without query or fragment. +Gateway takes that host from `Forwarded` / `X-Forwarded-Host` / `Host`, as the CSRF +check does, and appends `X-Forwarded-Port` when that host carries no port of its own. +A stray `X-Forwarded-Port: 8443` therefore turns `deerflow.example` into +`deerflow.example:8443`, and a link without that port gets a 422. The scheme is +not compared, since behind a TLS proxy Gateway can see `http` for an `https` link. URLs are parsed as local selectors and are never fetched. +For split-origin clients or internal proxies, pass the thread ID. The field is +separate from message text: links in pasted documents, tool results, or previous +messages grant no access. The server supplies source IDs to the model as background user-role data and binds the reader to this run's references and authenticated identity. diff --git a/backend/tests/test_conversation_access.py b/backend/tests/test_conversation_access.py index 27b3dfee87..260d46a6de 100644 --- a/backend/tests/test_conversation_access.py +++ b/backend/tests/test_conversation_access.py @@ -9,10 +9,11 @@ from types import SimpleNamespace from unittest.mock import AsyncMock import pytest -from fastapi import HTTPException +from fastapi import HTTPException, Request from langchain_core.messages import HumanMessage from langgraph.store.memory import InMemoryStore from pydantic import ValidationError +from starlette.datastructures import URL from app.gateway.authz import AuthContext from app.gateway.run_models import RunCreateRequest @@ -33,7 +34,8 @@ def _setup(*, user_id="alice", permissions=("runs:read",), enabled=True, tool_ou } ) user = SimpleNamespace(id=user_id, system_role="admin") - request = SimpleNamespace(state=SimpleNamespace(auth=AuthContext(user, list(permissions))), url="https://deerflow.example/api/threads/current/runs") + request = Request({"type": "http", "method": "POST", "scheme": "https", "server": ("deerflow.example", 443), "path": "/api/threads/current/runs", "headers": [(b"host", b"deerflow.example")]}) + request.state.auth = AuthContext(user, list(permissions)) events = MemoryRunEventStore() threads = MemoryThreadMetaStore(InMemoryStore()) manager = AsyncMock() @@ -227,7 +229,7 @@ def test_start_run_installs_fresh_capability_without_persisting_it(monkeypatch): request.state.user = user request.state.auth = AuthContext(user, ["runs:create", "runs:read"]) request.state.auth_source = "session" - request.url = "https://deerflow.example/api/threads/current/runs" + request.url = URL("https://deerflow.example/api/threads/current/runs") await threads.create("source", user_id="alice") captured = [] diff --git a/backend/tests/test_conversation_reference_proxy_origin.py b/backend/tests/test_conversation_reference_proxy_origin.py new file mode 100644 index 0000000000..7416e9d0db --- /dev/null +++ b/backend/tests/test_conversation_reference_proxy_origin.py @@ -0,0 +1,102 @@ +"""A conversation URL is matched against the host the client addressed. + +Gateway runs behind nginx and often behind a TLS proxy too. uvicorn only +applies ``X-Forwarded-Proto`` from ``FORWARDED_ALLOW_IPS`` (``127.0.0.1`` by +default), so with nginx in another container or pod the request URL stays +``http`` while the browser's link is ``https``. A local nginx resends its own +``$scheme``. These cases send the request through uvicorn's real +``ProxyHeadersMiddleware`` and the real ``prepare_conversation_reader``. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx +import pytest +from fastapi import FastAPI, HTTPException, Request +from langgraph.store.memory import InMemoryStore +from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware + +from app.gateway.authz import AuthContext +from app.gateway.conversation_access import prepare_conversation_reader +from deerflow.config.app_config import AppConfig +from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.runs.manager import EditReplayVisibility + +pytestmark = pytest.mark.asyncio + +_NGINX_CONTAINER = "172.18.0.5" +_CHAT = "/workspace/chats/source" + + +def _app() -> FastAPI: + config = AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "tools": [{"name": "read_conversation", "group": "conversation", "use": "deerflow.tools.conversation:read_conversation"}], + } + ) + manager = AsyncMock() + manager.list_edit_replay_visibility.return_value = EditReplayVisibility() + run_context = SimpleNamespace(event_store=MemoryRunEventStore(), thread_store=MemoryThreadMetaStore(InMemoryStore())) + app = FastAPI() + + @app.post("/api/threads/current/runs") + async def create_run(request: Request, body: dict) -> dict: + request.state.auth = AuthContext(SimpleNamespace(id="alice", system_role="user"), ["runs:read"]) + try: + _, ids = prepare_conversation_reader(body["references"], request=request, user_id="alice", run_context=run_context, run_manager=manager, app_config=config) + except HTTPException as exc: + return {"status": exc.status_code} + return {"status": 200, "ids": list(ids)} + + return app + + +async def _submit(reference: str, headers: dict[str, str], *, peer: str = _NGINX_CONTAINER, trusted: str = "127.0.0.1") -> dict: + # ``trusted`` is uvicorn's FORWARDED_ALLOW_IPS; ``peer`` is the address nginx connects from. + transport = httpx.ASGITransport(app=ProxyHeadersMiddleware(_app(), trusted_hosts=trusted), client=(peer, 50000)) + async with httpx.AsyncClient(transport=transport, base_url="http://gateway:8001") as client: + response = await client.post("/api/threads/current/runs", json={"references": [reference]}, headers=headers) + return response.json() + + +_TLS_EDGE = {"host": "deerflow.example", "x-forwarded-proto": "https", "x-forwarded-for": "203.0.113.9"} + + +@pytest.mark.parametrize( + ("reference", "headers", "peer", "trusted"), + [ + pytest.param(f"https://deerflow.example{_CHAT}", _TLS_EDGE, _NGINX_CONTAINER, "127.0.0.1", id="docker-nginx-untrusted-peer"), + pytest.param(f"https://deerflow.example{_CHAT}", {**_TLS_EDGE, "x-forwarded-proto": "http"}, "127.0.0.1", "127.0.0.1", id="local-nginx-resends-http"), + pytest.param(f"https://deerflow.example{_CHAT}", _TLS_EDGE, _NGINX_CONTAINER, _NGINX_CONTAINER, id="trusted-proxy"), + pytest.param(f"https://deerflow.example{_CHAT}", {**_TLS_EDGE, "host": "gateway:8001", "x-forwarded-host": "deerflow.example"}, _NGINX_CONTAINER, "127.0.0.1", id="forwarded-host"), + pytest.param(f"https://deerflow.example:443{_CHAT}", _TLS_EDGE, _NGINX_CONTAINER, "127.0.0.1", id="explicit-default-port"), + pytest.param(f"https://DeerFlow.Example{_CHAT}", _TLS_EDGE, _NGINX_CONTAINER, "127.0.0.1", id="host-case"), + pytest.param(f"http://localhost:2026{_CHAT}", {"host": "localhost:2026", "x-forwarded-proto": "http"}, "127.0.0.1", "127.0.0.1", id="plain-http-dev"), + pytest.param(f"https://deerflow.example:8443{_CHAT}", {**_TLS_EDGE, "x-forwarded-port": "8443"}, _NGINX_CONTAINER, "127.0.0.1", id="forwarded-port"), + pytest.param(f"http://[::1]:2026{_CHAT}", {"host": "[::1]:2026", "x-forwarded-proto": "http"}, "127.0.0.1", "127.0.0.1", id="ipv6-host"), + ], +) +async def test_a_link_from_the_addressed_host_is_accepted(reference, headers, peer, trusted) -> None: + assert await _submit(reference, headers, peer=peer, trusted=trusted) == {"status": 200, "ids": ["source"]} + + +@pytest.mark.parametrize( + "reference", + [ + pytest.param(f"https://other.example{_CHAT}", id="other-host"), + pytest.param(f"https://deerflow.example:8443{_CHAT}", id="other-port"), + pytest.param(f"https://alice@deerflow.example{_CHAT}", id="credentials"), + pytest.param(f"ftp://deerflow.example{_CHAT}", id="other-scheme"), + pytest.param(f"https://deerflow.example:notaport{_CHAT}", id="bad-port"), + pytest.param(f"https://{_CHAT}", id="no-host"), + pytest.param(f"https://deerflow.example{_CHAT}?tab=1", id="query"), + pytest.param(f"https://deerflow.example{_CHAT}#tab", id="fragment"), + ], +) +async def test_a_link_from_anywhere_else_is_still_rejected(reference) -> None: + assert await _submit(reference, _TLS_EDGE) == {"status": 422} diff --git a/backend/tests/test_conversation_references_context_channel.py b/backend/tests/test_conversation_references_context_channel.py index 749b3e11d6..5a1675820f 100644 --- a/backend/tests/test_conversation_references_context_channel.py +++ b/backend/tests/test_conversation_references_context_channel.py @@ -17,6 +17,7 @@ import pytest from fastapi import HTTPException from langchain_core.messages import HumanMessage from pydantic import ValidationError +from starlette.datastructures import URL from app.gateway.authz import AuthContext from app.gateway.run_models import MAX_CONVERSATION_REFERENCES, RunCreateRequest @@ -172,7 +173,7 @@ def test_start_run_grants_through_context_without_forwarding_the_key(monkeypatch request.state.user = user request.state.auth = AuthContext(user, ["runs:create", "runs:read"]) request.state.auth_source = "session" - request.url = "https://deerflow.example/api/threads/current/runs" + request.url = URL("https://deerflow.example/api/threads/current/runs") await threads.create("source", user_id="alice") captured = [] diff --git a/backend/tests/test_terminal_store_contract.py b/backend/tests/test_terminal_store_contract.py index 01c8c37dd9..451e8a3563 100644 --- a/backend/tests/test_terminal_store_contract.py +++ b/backend/tests/test_terminal_store_contract.py @@ -106,38 +106,38 @@ async def test_postgres_checks_database_time_after_row_lock(store, monkeypatch, pytest.skip("requires PostgreSQL row locks and clock_timestamp") import asyncio - from sqlalchemy import func, select + from sqlalchemy import func, select, update - expiry = datetime.now(UTC) + timedelta(milliseconds=300) + expiry = datetime.now(UTC) + timedelta(hours=1) await store.put("locked", thread_id="locked-thread", status="running", owner_worker_id="owner", lease_expires_at=expiry.isoformat()) - entered = asyncio.Event() + entered = asyncio.get_running_loop().create_future() lock = store._lock_owned_row async def observe_transaction_start(session, run_id): - assert await session.scalar(select(func.clock_timestamp())) < expiry - entered.set() + pid, transaction_started_at = (await session.execute(select(func.pg_backend_pid(), func.transaction_timestamp()))).one() + entered.set_result((pid, transaction_started_at)) await lock(session, run_id) monkeypatch.setattr(store, "_lock_owned_row", observe_transaction_start) - operation = None - async with store._sf() as blocker: - await blocker.execute(select(RunRow).where(RunRow.run_id == "locked").with_for_update()) - try: + async with asyncio.timeout(10), asyncio.TaskGroup() as tasks: + async with store._sf() as blocker: + await blocker.execute(select(RunRow.run_id).where(RunRow.run_id == "locked").with_for_update()) + blocker_pid = await blocker.scalar(select(func.pg_backend_pid())) if method in ("renew_lease", "update_lease"): - operation = asyncio.create_task(getattr(store, method)("locked", owner_worker_id="owner", lease_expires_at=(expiry + timedelta(seconds=30)).isoformat())) + operation = tasks.create_task(getattr(store, method)("locked", owner_worker_id="owner", lease_expires_at=(expiry + timedelta(seconds=30)).isoformat())) elif method == "update_status": - operation = asyncio.create_task(store.update_status("locked", "error", owner_worker_id="owner")) + operation = tasks.create_task(store.update_status("locked", "error", owner_worker_id="owner")) else: - operation = asyncio.create_task(store.finalize_if_not_cancelled("locked", status="success", owner_worker_id="owner")) - await asyncio.wait_for(entered.wait(), 1) - async with asyncio.timeout(2): - while await blocker.scalar(select(func.clock_timestamp())) <= expiry: - await asyncio.sleep(0.01) - assert not operation.done() - finally: - await blocker.rollback() - assert operation is not None - result = await asyncio.wait_for(operation, 1) + operation = tasks.create_task(store.finalize_if_not_cancelled("locked", status="success", owner_worker_id="owner")) + pid, transaction_started_at = await entered + while blocker_pid not in await blocker.scalar(select(func.pg_blocking_pids(pid))): + await asyncio.sleep(0.01) + # Expire the lease while the writer is blocked. Its transaction clock + # still sees a live lease, but the database clock has reached expiry. + expired_at = await blocker.scalar(update(RunRow).where(RunRow.run_id == "locked").values(lease_expires_at=func.clock_timestamp()).returning(RunRow.lease_expires_at)) + assert transaction_started_at < expired_at < expiry + await blocker.commit() + result = await operation if method == "renew_lease": assert not result.renewed elif method == "finalize_if_not_cancelled": @@ -146,7 +146,7 @@ async def test_postgres_checks_database_time_after_row_lock(store, monkeypatch, assert result is False row = await store.get("locked") assert row["status"] == "running" - assert datetime.fromisoformat(row["lease_expires_at"]) == expiry + assert datetime.fromisoformat(row["lease_expires_at"]) == expired_at @pytest.mark.anyio