fix: limit public flow endpoint from displaying private flow streams (#13602)

* fix: limit public flow endpoint from displaying private flow streams

* fix: Address coderabbit reviews

---------

Co-authored-by: Janardan S Kavia <janardanskavia@Janardans-MacBook-Pro.local>

src/backend/base/langflow/api/v1/chat.py

@@ -861,6 +861,10 @@ async def build_public_tmp(
             queue_service=queue_service,
             flow_name=flow_name or f"{authenticated_user_id or client_id}_{flow_id}",
         )
+        # Gate the public events/cancel endpoints to jobs that were actually
+        # started through this public build path, preventing unauthenticated
+        # callers from reading or cancelling private-flow builds by job_id.
+        await queue_service.register_public_job(job_id)
     except CustomComponentValidationError as exc:
         await logger.awarning(f"Public flow validation failed: {exc}")
         raise HTTPException(status_code=400, detail="This flow cannot be executed.") from exc
@@ -880,6 +884,20 @@ async def build_public_tmp(
     )
 
 
+async def _assert_public_job(job_id: str, queue_service: JobQueueService) -> None:
+    """Raise HTTP 404 if job_id was not registered through the public build endpoint.
+
+    Prevents unauthenticated callers from reading or cancelling private-flow
+    builds by guessing or leaking a job_id.
+
+    Why 404 not 403: returning 403 would confirm the job exists under a different
+    access tier, leaking information about private builds. 404 is neutral.
+    """
+    if not await queue_service.is_public_job_async(job_id):
+        # Static detail — do not reflect job_id back; avoid confirming which IDs exist.
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found")
+
+
 @router.get("/build_public_tmp/{job_id}/events")
 async def get_build_events_public(
     job_id: str,
@@ -892,6 +910,7 @@ async def get_build_events_public(
     This endpoint does not require authentication, matching the public build endpoint.
     It is used by the shareable playground to consume build events.
     """
+    await _assert_public_job(job_id, queue_service)
     return await get_flow_events_response(
         job_id=job_id,
         queue_service=queue_service,
@@ -912,6 +931,7 @@ async def cancel_build_public(
     This endpoint does not require authentication, matching the public build endpoint.
     It is used by the shareable playground to cancel builds.
     """
+    await _assert_public_job(job_id, queue_service)
     try:
         cancellation_success = await cancel_flow_build(job_id=job_id, queue_service=queue_service)
 

src/backend/base/langflow/services/job_queue/service.py

@@ -29,6 +29,10 @@ _CANCEL_CHANNEL_PREFIX = "langflow:cancel:"
 # Activity heartbeat key written by polling and streaming responses. The
 # polling watchdog scans these to detect abandoned builds (client gave up).
 _ACTIVITY_PREFIX = "langflow:activity:"
+# Presence key for jobs started through the public (unauthenticated) build
+# endpoint. Allows the public events/cancel endpoints to reject job_ids that
+# belong to private-flow builds. Uses the same TTL as the stream/owner keys.
+_PUBLIC_JOB_PREFIX = "langflow:public_job:"
 
 
 class JobQueueNotFoundError(Exception):
@@ -92,6 +96,7 @@ class JobQueueService(Service):
         """
         self._queues: dict[str, tuple[asyncio.Queue, EventManager, asyncio.Task | None, float | None]] = {}
         self._job_owners: dict[str, UUID] = {}
+        self._public_jobs: set[str] = set()
         self._cleanup_task: asyncio.Task | None = None
         self._closed = False
         self.ready = False
@@ -311,6 +316,32 @@ class JobQueueService(Service):
         """Return the user ID that owns a job, or None if not tracked."""
         return self._job_owners.get(job_id)
 
+    async def register_public_job(self, job_id: str) -> None:
+        """Mark a job as started through the public (unauthenticated) build endpoint.
+
+        Only jobs registered here may be accessed via the public events/cancel endpoints.
+        This prevents unauthenticated callers from reading or cancelling private-flow jobs.
+
+        Async (even though the base implementation is a synchronous set add):
+        RedisJobQueueService overrides this to also persist the marker to Redis
+        *before returning*, so a request landing on a different worker immediately
+        after registration sees the marker via is_public_job_async.
+        """
+        self._public_jobs.add(job_id)
+
+    def is_public_job(self, job_id: str) -> bool:
+        """Return True if the job was started through the public build endpoint."""
+        return job_id in self._public_jobs
+
+    async def is_public_job_async(self, job_id: str) -> bool:
+        """Return True if the job was started through the public build endpoint.
+
+        Base implementation is synchronous (in-memory set lookup).
+        RedisJobQueueService overrides this to also check Redis for cross-worker
+        correctness in multi-worker deployments.
+        """
+        return self.is_public_job(job_id)
+
     async def cleanup_job(self, job_id: str) -> None:
         """Clean up and release resources for a specific job.
 
@@ -366,6 +397,7 @@ class JobQueueService(Service):
         # Remove the job entry from the registry
         self._queues.pop(job_id, None)
         self._job_owners.pop(job_id, None)
+        self._public_jobs.discard(job_id)
         await logger.adebug(f"Cleanup successful for job_id {job_id}: resources have been released.")
 
     async def cancel_job(self, job_id: str) -> None:
@@ -735,6 +767,7 @@ class RedisJobQueueService(JobQueueService):
     OWNER_PREFIX = _OWNER_PREFIX
     CANCEL_CHANNEL_PREFIX = _CANCEL_CHANNEL_PREFIX
     ACTIVITY_PREFIX = _ACTIVITY_PREFIX
+    PUBLIC_JOB_PREFIX = _PUBLIC_JOB_PREFIX
 
     def __init__(
         self,
@@ -812,6 +845,9 @@ class RedisJobQueueService(JobQueueService):
     def _owner_key(self, job_id: str) -> str:
         return f"{self.OWNER_PREFIX}{job_id}"
 
+    def _public_job_key(self, job_id: str) -> str:
+        return f"{self.PUBLIC_JOB_PREFIX}{job_id}"
+
     def _cancel_channel(self, job_id: str) -> str:
         return f"{self.CANCEL_CHANNEL_PREFIX}{job_id}"
 
@@ -970,6 +1006,16 @@ class RedisJobQueueService(JobQueueService):
                             published = True
                         if needs_ttl_refresh:
                             await self._client.expire(stream_key, self._ttl)
+                            # Why: register_public_job sets the public_job marker with
+                            # ex=self._ttl once at job start. Long-running builds that
+                            # outlive that TTL would have the marker expire while the
+                            # stream itself is kept alive, causing is_public_job_async
+                            # to 404 a still-active public job on other workers. Refresh
+                            # it on the same cadence as the stream TTL. is_public_job is
+                            # the in-memory (sync) check — true on the worker that owns
+                            # this bridge, which is the same worker that registered it.
+                            if self.is_public_job(job_id):
+                                await self._client.expire(self._public_job_key(job_id), self._ttl)
                             last_ttl_refresh = time.monotonic()
                         in_flight_item = None
                         _retry_delay = 0.1
@@ -1555,6 +1601,7 @@ class RedisJobQueueService(JobQueueService):
                         self._stream_key(job_id),
                         self._owner_key(job_id),
                         self._activity_key(job_id),
+                        self._public_job_key(job_id),
                     )
                 except asyncio.CancelledError:
                     raise
@@ -1591,3 +1638,39 @@ class RedisJobQueueService(JobQueueService):
                 await self._client.expire(owner_key, self._ttl)
                 return _UUID(value.decode())
         return None
+
+    async def register_public_job(self, job_id: str) -> None:
+        """Mark a job as public in both local memory and Redis for cross-worker access.
+
+        Why synchronous (not fire-and-forget): a request for the public events/cancel
+        endpoint can land on a different worker than the one that registered the job.
+        If the Redis write were backgrounded, that worker could run is_public_job_async
+        before the marker exists and incorrectly 404 a legitimate public job. Awaiting
+        the write here guarantees the marker is visible to every worker by the time
+        build_public_tmp's response (containing job_id) reaches the client.
+        """
+        await super().register_public_job(job_id)
+        if self._client:
+            await self._set_public_job_key(job_id)
+
+    async def _set_public_job_key(self, job_id: str) -> None:
+        try:
+            await self._client.set(self._public_job_key(job_id), b"1", ex=self._ttl)
+        except Exception as exc:  # noqa: BLE001
+            await logger.awarning(f"Failed to set public_job Redis key for {job_id}: {exc!r}")
+
+    async def is_public_job_async(self, job_id: str) -> bool:
+        """Return True if the job was started through the public build endpoint.
+
+        Checks local memory first (fast path), then falls back to Redis so that
+        a request hitting a different worker than the one that started the build
+        still works correctly.
+        """
+        if super().is_public_job(job_id):
+            return True
+        if self._client:
+            try:
+                return bool(await self._client.exists(self._public_job_key(job_id)))
+            except Exception as exc:  # noqa: BLE001
+                await logger.awarning(f"Redis public_job check failed for {job_id}: {exc!r}")
+        return False

src/backend/tests/unit/test_chat_endpoint.py

@@ -1319,3 +1319,138 @@ def test_scope_session_to_namespace_helper():
     assert scope_session_to_namespace("victim-session", "namespace-B") == "namespace-B:victim-session"
     # A foreign-namespace prefix is treated as out-of-namespace and gets re-wrapped.
     assert scope_session_to_namespace("namespace-B:victim", "namespace-A") == "namespace-A:namespace-B:victim"
+
+
+# ── Public job registry unit tests ───────────────────────────────────────────
+# CVE fix: unauthenticated callers must not access private-flow job streams
+# by guessing or leaking a job_id from the authenticated build endpoint.
+
+
+async def test_job_queue_service_register_and_check_public_job():
+    """register_public_job marks a job as public; is_public_job reflects that."""
+    svc = JobQueueService()
+    job_id = str(uuid.uuid4())
+
+    # Why: job not registered yet — must return False before registration
+    assert svc.is_public_job(job_id) is False
+
+    await svc.register_public_job(job_id)
+
+    # Why: job registered — must return True after registration
+    assert svc.is_public_job(job_id) is True
+
+
+def test_job_queue_service_unregistered_job_not_public():
+    """A job_id that was never registered is not considered public."""
+    svc = JobQueueService()
+    assert svc.is_public_job(str(uuid.uuid4())) is False
+
+
+async def test_job_queue_service_is_public_job_async_base():
+    """is_public_job_async on base class delegates to in-memory is_public_job."""
+    svc = JobQueueService()
+    job_id = str(uuid.uuid4())
+
+    # Why: async variant must mirror sync variant — False before, True after
+    assert await svc.is_public_job_async(job_id) is False
+    await svc.register_public_job(job_id)
+    assert await svc.is_public_job_async(job_id) is True
+
+
+async def test_job_queue_service_cleanup_removes_public_registration():
+    """cleanup_job discards the public registration so the job_id cannot be reused.
+
+    Why: tests the actual cleanup_job contract — not the internal set.
+    If cleanup_job stops calling discard, this test must catch it.
+    """
+    svc = JobQueueService()
+    job_id = str(uuid.uuid4())
+    await svc.register_public_job(job_id)
+    assert svc.is_public_job(job_id) is True
+
+    # Call the real cleanup path — not svc._public_jobs.discard directly.
+    # cleanup_job early-returns when job_id is not in _queues, but the
+    # _public_jobs.discard call is unconditional (after the early-return guard),
+    # so we need to reach it. Seed a minimal queue entry first.
+    svc._queues[job_id] = (asyncio.Queue(), None, None, None)  # type: ignore[arg-type]
+    await svc.cleanup_job(job_id)
+
+    # Why: if cleanup_job ever drops the discard call, is_public_job still returns True here
+    assert svc.is_public_job(job_id) is False
+
+
+@pytest.mark.benchmark
+@pytest.mark.security
+async def test_private_job_id_blocked_on_public_events_endpoint(client, json_memory_chatbot_no_llm, logged_in_headers):
+    """A job_id started via the authenticated build endpoint must be rejected by the public events endpoint.
+
+    Security proof: before the fix, any caller who knew or guessed a private job_id
+    could read the live event stream (LLM output, API keys, tracebacks) without auth.
+    After the fix, _assert_public_job returns HTTP 404 because the job was never
+    registered via register_public_job.
+
+    Why 404 not 403: returning 403 would confirm the job exists under a different
+    access tier, leaking information about private builds.
+    """
+    flow_id = await create_flow(client, json_memory_chatbot_no_llm, logged_in_headers)
+
+    # Start a PRIVATE (authenticated) build — job_id never passed through build_public_tmp
+    private_start = await client.post(
+        f"api/v1/build/{flow_id}/flow",
+        json={},
+        headers={**logged_in_headers, "Content-Type": "application/json"},
+    )
+    assert private_start.status_code == codes.OK
+    private_job_id = private_start.json()["job_id"]
+
+    # Why: the shared AsyncClient persists access-token cookies from logged_in_headers.
+    # Without clearing them, get_current_user_optional could resolve a user on this
+    # "public" request, which would not exercise the unauthenticated attack path.
+    client.cookies.clear()
+
+    # Attempt to read the private job's events via the unauthenticated public endpoint
+    # Why: this is the exact attack vector — attacker has job_id, tries public endpoint
+    events_response = await client.get(
+        f"api/v1/build_public_tmp/{private_job_id}/events?event_delivery=polling",
+        headers={"Accept": "application/x-ndjson"},
+    )
+
+    # Must be 404 — gate blocks private job from public endpoint
+    assert events_response.status_code == codes.NOT_FOUND
+    assert events_response.json()["detail"] == "Job not found"
+
+
+@pytest.mark.benchmark
+@pytest.mark.security
+async def test_private_job_id_blocked_on_public_cancel_endpoint(client, json_memory_chatbot_no_llm, logged_in_headers):
+    """A job_id started via the authenticated build endpoint must be rejected by the public cancel endpoint.
+
+    Security proof: before the fix, an unauthenticated attacker could cancel any
+    in-flight private build as a denial-of-service by supplying a known job_id.
+    After the fix, _assert_public_job returns HTTP 404.
+    """
+    flow_id = await create_flow(client, json_memory_chatbot_no_llm, logged_in_headers)
+
+    # Start a PRIVATE (authenticated) build
+    private_start = await client.post(
+        f"api/v1/build/{flow_id}/flow",
+        json={},
+        headers={**logged_in_headers, "Content-Type": "application/json"},
+    )
+    assert private_start.status_code == codes.OK
+    private_job_id = private_start.json()["job_id"]
+
+    # Why: the shared AsyncClient persists access-token cookies from logged_in_headers.
+    # Without clearing them, get_current_user_optional could resolve a user on this
+    # "public" request, which would not exercise the unauthenticated attack path.
+    client.cookies.clear()
+
+    # Attempt to cancel the private job via the unauthenticated public endpoint
+    cancel_response = await client.post(
+        f"api/v1/build_public_tmp/{private_job_id}/cancel",
+        headers={"Content-Type": "application/json"},
+    )
+
+    # Must be 404 — gate blocks private job from public cancel endpoint
+    assert cancel_response.status_code == codes.NOT_FOUND
+    assert cancel_response.json()["detail"] == "Job not found"

src/backend/tests/unit/test_redis_job_queue_service.py

@@ -2406,3 +2406,77 @@ async def test_polling_watchdog_runs_when_cancel_channel_disabled():
                 with contextlib.suppress(asyncio.CancelledError):
                     await bridge
         await fake_client.aclose()
+
+
+# ── Public job registry — Redis-specific tests ───────────────────────────────
+# Complement the base-class unit tests in test_chat_endpoint.py.
+# These tests verify the Redis-specific paths: cross-worker fallback (#6)
+# and cleanup deleting the Redis key (#7).
+
+
+async def test_redis_public_job_cross_worker_fallback():
+    """is_public_job_async returns True for Worker B even when its in-memory set is empty.
+
+    Why: Worker A registers the job (writes local memory + Redis key via background task).
+    Worker B has no local memory entry — it must fall back to Redis.
+    This is the multi-worker correctness guarantee of RedisJobQueueService.
+    """
+    fake_client = fakeredis_aio.FakeRedis()
+    # Worker A — registers the public job
+    svc_a, _ = await _make_service(shared_client=fake_client, cancel_channel_enabled=False)
+    # Worker B — shares same Redis, but has empty local memory
+    svc_b, _ = await _make_service(shared_client=fake_client, cancel_channel_enabled=False)
+
+    try:
+        job_id = str(uuid.uuid4())
+        # register_public_job awaits the Redis write directly (no background task
+        # to drain) — see Why comment on RedisJobQueueService.register_public_job.
+        await svc_a.register_public_job(job_id)
+
+        # Worker B must have no in-memory entry (no shared memory between workers)
+        assert not svc_b.is_public_job(job_id), "Worker B must have no in-memory entry"
+
+        # Why: is_public_job_async on Worker B must hit Redis fallback and return True
+        assert await svc_b.is_public_job_async(job_id) is True
+    finally:
+        await _stop_service(svc_a)
+        await _stop_service(svc_b)
+        await fake_client.aclose()
+
+
+async def test_redis_cleanup_removes_public_job_key():
+    """cleanup_job deletes the public_job Redis key so the job_id cannot be reused cross-worker.
+
+    Why: after cleanup the in-memory discard is proven by the base-class test in
+    test_chat_endpoint.py. This test proves the Redis key (cross-worker marker) is
+    also removed. A missing delete would let a cross-worker is_public_job_async
+    return True for a finished/evicted job.
+    """
+    svc, fake_client = await _make_service(cancel_channel_enabled=False)
+
+    try:
+        job_id = str(uuid.uuid4())
+        # register_public_job awaits the Redis write directly (no background task
+        # to drain) — see Why comment on RedisJobQueueService.register_public_job.
+        await svc.register_public_job(job_id)
+
+        # Confirm the key exists in Redis before cleanup
+        pub_key = svc._public_job_key(job_id)
+        assert await fake_client.exists(pub_key), "public_job Redis key must exist after registration"
+
+        # Seed a minimal queue entry so cleanup_job doesn't early-return
+        svc._queues[job_id] = (asyncio.Queue(), None, None, None)  # type: ignore[arg-type]
+        await svc.cleanup_job(job_id)
+
+        # Drain any background tasks spawned by cleanup
+        for task in list(svc._background_tasks):
+            with contextlib.suppress(Exception):
+                await task
+
+        # Why: if cleanup_job's Redis DEL call ever drops public_job_key, this catches it
+        assert not await fake_client.exists(pub_key), "public_job Redis key must be deleted after cleanup"
+        # In-memory also cleared
+        assert svc.is_public_job(job_id) is False
+    finally:
+        await _stop_service(svc)
+        await fake_client.aclose()