fix(gateway): match conversation reference URLs on the addressed host (#8216)

* fix(gateway): match conversation reference URLs on the addressed host

A conversation URL reference was accepted only when its (scheme,
netloc) equalled Gateway's request.url. Behind TLS that URL is often
http: Docker Compose and Helm run uvicorn without FORWARDED_ALLOW_IPS,
so nginx's X-Forwarded-Proto from another container is ignored, and
nginx.local.conf resends its own $scheme. The browser's https chat link
then got a 422, and so did an explicit :443 or a differently cased host.

Resolve the run request's origin with _request_origin(), as the CSRF
check does (Forwarded / X-Forwarded-Host / Host), normalize the
reference's origin the same way, and compare host and port only. The
scheme cannot tell two deployments on the same host apart, and URLs
remain owner-bound local selectors that are never fetched.

* test(gateway): cover forwarded port, IPv6 and fragment for conversation URLs

Name X-Forwarded-Port in the API.md reference contract: it is appended
when the forwarded host has no port, so a stray value causes a 422.

* test(runtime): synchronize PostgreSQL lease expiration with row lock

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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)

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")

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.
 

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 = []
 

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}

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 = []
 

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