diff --git a/src/backend/base/langflow/api/v1/chat.py b/src/backend/base/langflow/api/v1/chat.py index a18c3dbb37..7ad4c9b18a 100644 --- a/src/backend/base/langflow/api/v1/chat.py +++ b/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) diff --git a/src/backend/base/langflow/services/job_queue/service.py b/src/backend/base/langflow/services/job_queue/service.py index f08774dad5..1611cf5c9d 100644 --- a/src/backend/base/langflow/services/job_queue/service.py +++ b/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 diff --git a/src/backend/tests/unit/test_chat_endpoint.py b/src/backend/tests/unit/test_chat_endpoint.py index 3df3aafacd..50476a5532 100644 --- a/src/backend/tests/unit/test_chat_endpoint.py +++ b/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" diff --git a/src/backend/tests/unit/test_redis_job_queue_service.py b/src/backend/tests/unit/test_redis_job_queue_service.py index e17f5d5a8a..5f3a120129 100644 --- a/src/backend/tests/unit/test_redis_job_queue_service.py +++ b/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()