From 2f6400db89bd04072fdec5435f21506758db6091 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ant=C3=B4nio=20Alexandre=20Borges=20Lima?= <104531655+AntonioABLima@users.noreply.github.com> Date: Mon, 6 Apr 2026 15:15:11 -0300 Subject: [PATCH] fix: enforce IDOR protection on v2 workflow job endpoints (#12398) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: enforce ownership check and pass user_id in workflow job creation - Add _assert_job_owner helper that raises 403 for non-owners (legacy jobs with user_id=None are allowed through) - Move ownership check before job type check in stop_workflow to avoid leaking job.type to non-owners - Pass user_id to create_job in both sync and background execution paths - Add user_id parameter to JobService.create_job signature * test: add ownership and legacy job coverage for workflow endpoints - Add TestWorkflowIDORProtection class with tests for 403 on cross-user access - Add test for stop_workflow with legacy user_id=None job (should not return 403) * fix: pass user_id to create_job in knowledge_bases ingestion endpoint Prevents IDOR vulnerability where ingestion jobs created without user_id would bypass ownership checks, matching the fix applied to workflow jobs. * refactor: move job ownership check to JobService layer Moves _assert_job_owner from workflow.py into JobService.assert_job_owner so any future job-consuming endpoint can reuse the check without duplicating logic. Both get_workflow_status and stop_workflow now delegate to the service. * test: fix mocks for assert_job_owner after service layer refactor MagicMock blocks attributes starting with 'assert' by default. Added explicit mock_service.assert_job_owner = MagicMock() to each test that mocks get_job_service so the ownership check is a no-op for tests not focused on IDOR behavior. * refactor: enforce job ownership at SQL level, remove assert_job_owner Move IDOR ownership check from application layer into the DB query. `get_job_by_job_id` now accepts an optional `user_id` and filters by `job_id AND (user_id = ? OR user_id IS NULL)`, so unauthorized access returns 404 instead of 403. Removes `JobService.assert_job_owner`. - Strengthen legacy-job test assertions (!=403 → ==200) - Add missing GET test for non-WORKFLOW job type → 404 --- .../base/langflow/api/v1/knowledge_bases.py | 7 +- src/backend/base/langflow/api/v2/workflow.py | 21 +- .../services/database/models/jobs/crud.py | 10 +- .../base/langflow/services/jobs/service.py | 11 +- .../tests/unit/api/v2/test_workflow.py | 422 +++++++++++++++++- 5 files changed, 459 insertions(+), 12 deletions(-) diff --git a/src/backend/base/langflow/api/v1/knowledge_bases.py b/src/backend/base/langflow/api/v1/knowledge_bases.py index 4767e89d89..44d6fdfb27 100644 --- a/src/backend/base/langflow/api/v1/knowledge_bases.py +++ b/src/backend/base/langflow/api/v1/knowledge_bases.py @@ -367,7 +367,12 @@ async def ingest_files_to_knowledge_base( # Create job record in database for both async and sync paths await job_service.create_job( - job_id=job_id, flow_id=job_id, job_type=JobType.INGESTION, asset_id=asset_id, asset_type="knowledge_base" + job_id=job_id, + flow_id=job_id, + job_type=JobType.INGESTION, + asset_id=asset_id, + asset_type="knowledge_base", + user_id=current_user.id, ) # Always use async path: fire and forget the ingestion logic wrapped in status updates diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index a0ab72d6c3..8bd7749ed1 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -378,7 +378,7 @@ async def execute_sync_workflow( # Execute graph - component errors are caught and returned in response body job_service = get_job_service() - await job_service.create_job(job_id=job_id, flow_id=flow_id_str) + await job_service.create_job(job_id=job_id, flow_id=flow_id_str, user_id=api_key_user.id) try: task_result, execution_session_id = await job_service.execute_with_status( job_id=job_id, @@ -467,6 +467,7 @@ async def execute_workflow_background( await job_service.create_job( job_id=job_id, flow_id=flow_id_str, + user_id=api_key_user.id, ) await task_service.fire_and_forget_task( @@ -535,7 +536,7 @@ async def get_workflow_status( job_service = get_job_service() try: - job = await job_service.get_job_by_job_id(job_id=job_id) + job = await job_service.get_job_by_job_id(job_id=job_id, user_id=api_key_user.id) except Exception as exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -648,7 +649,7 @@ async def get_workflow_status( ) async def stop_workflow( request: WorkflowStopRequest, - api_key_user: Annotated[UserRead, Depends(api_key_security)], # noqa: ARG001 + api_key_user: Annotated[UserRead, Depends(api_key_security)], ) -> WorkflowStopResponse: """Stop a running workflow execution by job_id. @@ -673,7 +674,7 @@ async def stop_workflow( try: # 1. Fetch Job - job = await job_service.get_job_by_job_id(job_id) + job = await job_service.get_job_by_job_id(job_id, user_id=api_key_user.id) except Exception as exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -695,6 +696,18 @@ async def stop_workflow( }, ) + # Verify this is a workflow job + if job.type != JobType.WORKFLOW: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={ + "error": "Job not found", + "code": "JOB_NOT_FOUND", + "message": f"Job {job_id} is not a workflow job (type: {job.type})", + "job_id": str(job_id), + }, + ) + if job.status == JobStatus.CANCELLED: return WorkflowStopResponse(job_id=str(job_id), message=f"Job {job_id} is already cancelled.") diff --git a/src/backend/base/langflow/services/database/models/jobs/crud.py b/src/backend/base/langflow/services/database/models/jobs/crud.py index 88eb0bc99a..4e8e156585 100644 --- a/src/backend/base/langflow/services/database/models/jobs/crud.py +++ b/src/backend/base/langflow/services/database/models/jobs/crud.py @@ -8,7 +8,7 @@ if TYPE_CHECKING: from sqlmodel.ext.asyncio.session import AsyncSession -from sqlmodel import col, select +from sqlmodel import col, or_, select from langflow.services.database.models.jobs.model import Job, JobStatus @@ -37,17 +37,21 @@ async def get_jobs_by_flow_id(db: AsyncSession, flow_id: UUID, page: int = 1, si return list(result.all()) -async def get_job_by_job_id(db: AsyncSession, job_id: UUID) -> Job | None: +async def get_job_by_job_id(db: AsyncSession, job_id: UUID, user_id: UUID | None = None) -> Job | None: """Get a single job by its UUID. Args: db: Async database session job_id: The job ID to fetch + user_id: When provided, restricts the result to jobs owned by this user + or legacy jobs with no owner (user_id IS NULL). Returns: - Job object or None if not found + Job object or None if not found (or not accessible by the given user) """ statement = select(Job).where(Job.job_id == job_id) + if user_id is not None: + statement = statement.where(or_(Job.user_id == user_id, col(Job.user_id).is_(None))) result = await db.exec(statement) return result.first() diff --git a/src/backend/base/langflow/services/jobs/service.py b/src/backend/base/langflow/services/jobs/service.py index a3ed31bdcd..914e683e83 100644 --- a/src/backend/base/langflow/services/jobs/service.py +++ b/src/backend/base/langflow/services/jobs/service.py @@ -47,20 +47,22 @@ class JobService(Service): async with session_scope() as session: return await get_jobs_by_flow_id(session, flow_id, page=page, size=page_size) - async def get_job_by_job_id(self, job_id: UUID | str) -> Job | None: + async def get_job_by_job_id(self, job_id: UUID | str, user_id: UUID | None = None) -> Job | None: """Get job for a specific job ID. Args: job_id: The job ID to filter jobs by + user_id: When provided, restricts the result to jobs owned by this user + or legacy jobs with no owner (user_id IS NULL). Returns: - Job object for the specified job ID + Job object for the specified job ID, or None if not found or not accessible """ if isinstance(job_id, str): job_id = UUID(job_id) async with session_scope() as session: - return await get_job_by_job_id(session, job_id) + return await get_job_by_job_id(session, job_id, user_id=user_id) async def create_job( self, @@ -69,6 +71,7 @@ class JobService(Service): job_type: JobType = JobType.WORKFLOW, asset_id: UUID | None = None, asset_type: str | None = None, + user_id: UUID | None = None, ) -> Job: """Create a new job record with QUEUED status. @@ -78,6 +81,7 @@ class JobService(Service): job_type: The job type asset_id: The asset ID asset_type: The asset type + user_id: The user ID who owns this job Returns: Created Job object @@ -96,6 +100,7 @@ class JobService(Service): type=job_type, asset_id=asset_id, asset_type=asset_type, + user_id=user_id, ) session.add(job) await session.flush() diff --git a/src/backend/tests/unit/api/v2/test_workflow.py b/src/backend/tests/unit/api/v2/test_workflow.py index 99a39201f0..940f38b101 100644 --- a/src/backend/tests/unit/api/v2/test_workflow.py +++ b/src/backend/tests/unit/api/v2/test_workflow.py @@ -34,7 +34,7 @@ import pytest from httpx import AsyncClient from langflow.exceptions.api import WorkflowValidationError from langflow.services.database.models.flow.model import Flow -from langflow.services.database.models.jobs.model import JobType +from langflow.services.database.models.jobs.model import Job, JobType from lfx.schema.workflow import JobStatus from lfx.services.deps import session_scope from sqlalchemy.exc import OperationalError @@ -1339,6 +1339,7 @@ class TestWorkflowStatus: mock_job.flow_id = flow_id mock_job.status = JobStatus.QUEUED mock_job.type = JobType.WORKFLOW + mock_job.user_id = None mock_job.created_timestamp = datetime.now(timezone.utc) with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: @@ -1389,6 +1390,7 @@ class TestWorkflowStatus: mock_job.job_id = job_id mock_job.status = JobStatus.FAILED mock_job.type = JobType.WORKFLOW + mock_job.user_id = None with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() @@ -1418,6 +1420,7 @@ class TestWorkflowStatus: mock_job.flow_id = flow_id mock_job.status = JobStatus.COMPLETED mock_job.type = JobType.WORKFLOW + mock_job.user_id = None with ( patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service, @@ -1457,6 +1460,7 @@ class TestWorkflowStatus: mock_job.flow_id = flow_id mock_job.status = JobStatus.TIMED_OUT mock_job.type = JobType.WORKFLOW + mock_job.user_id = None with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() @@ -1500,6 +1504,8 @@ class TestWorkflowStop: mock_job = MagicMock() mock_job.job_id = job_id mock_job.status = JobStatus.IN_PROGRESS + mock_job.type = JobType.WORKFLOW + mock_job.user_id = None with ( patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service, @@ -1557,6 +1563,8 @@ class TestWorkflowStop: mock_job = MagicMock() mock_job.job_id = job_id mock_job.status = JobStatus.CANCELLED + mock_job.type = JobType.WORKFLOW + mock_job.user_id = None with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() @@ -1570,3 +1578,415 @@ class TestWorkflowStop: result = response.json() assert result["job_id"] == job_id assert "already cancelled" in result["message"] + + +@pytest.mark.security +class TestWorkflowIDORProtection: + """Security regression tests for IDOR on workflow job endpoints. + + Covers GHSA-qfw4-cjhf-3g3q: background workflow jobs are queryable and + cancellable cross-user unless an ownership check is enforced. + """ + + @pytest.fixture + def mock_settings_dev_api_enabled(self): + with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings: + mock_service = MagicMock() + mock_settings = MagicMock() + mock_settings.developer_api_enabled = True + mock_service.settings = mock_settings + mock_get_settings.return_value = mock_service + yield mock_settings + + @pytest.mark.security + async def test_get_workflow_status_forbidden_for_other_user_job( + self, + client: AsyncClient, + created_api_key, + created_user_two_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """GET /api/v2/workflows returns 404 when the job belongs to a different user. + + GHSA-qfw4-cjhf-3g3q: job status must not be visible cross-user. + Ownership is enforced at the SQL level — unauthorized access returns 404. + """ + job_id = uuid4() + other_user_id = created_user_two_api_key.user_id + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.QUEUED, + type=JobType.WORKFLOW, + user_id=other_user_id, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) + + assert response.status_code == 404 + result = response.json() + assert result["detail"]["code"] == "JOB_NOT_FOUND" + assert str(job_id) in result["detail"]["job_id"] + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_get_workflow_status_allowed_for_own_job( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """GET /api/v2/workflows returns 200 when the job belongs to the requesting user.""" + job_id = uuid4() + owner_user_id = created_api_key.user_id + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.QUEUED, + type=JobType.WORKFLOW, + user_id=owner_user_id, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) + + assert response.status_code == 200 + result = response.json() + assert str(job_id) in result["job_id"] + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_stop_workflow_forbidden_for_other_user_job( + self, + client: AsyncClient, + created_api_key, + created_user_two_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """POST /api/v2/workflows/stop returns 404 when the job belongs to a different user. + + GHSA-qfw4-cjhf-3g3q: job cancellation must not be allowed cross-user. + Ownership is enforced at the SQL level — unauthorized access returns 404. + """ + job_id = uuid4() + other_user_id = created_user_two_api_key.user_id + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.IN_PROGRESS, + type=JobType.WORKFLOW, + user_id=other_user_id, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflows/stop", + json={"job_id": str(job_id)}, + headers=headers, + ) + + assert response.status_code == 404 + result = response.json() + assert result["detail"]["code"] == "JOB_NOT_FOUND" + assert str(job_id) in result["detail"]["job_id"] + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_stop_workflow_allowed_for_own_job( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """POST /api/v2/workflows/stop succeeds when the job belongs to the requesting user.""" + job_id = uuid4() + owner_user_id = created_api_key.user_id + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.CANCELLED, + type=JobType.WORKFLOW, + user_id=owner_user_id, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflows/stop", + json={"job_id": str(job_id)}, + headers=headers, + ) + + assert response.status_code == 200 + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_get_workflow_status_allowed_for_legacy_job_with_no_user_id( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """GET /api/v2/workflows does NOT block legacy jobs where user_id is NULL. + + Jobs created before the fix have user_id=None and must not be broken. + """ + job_id = uuid4() + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.QUEUED, + type=JobType.WORKFLOW, + user_id=None, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) + + assert response.status_code == 200 + result = response.json() + assert result["job_id"] == str(job_id) + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_stop_workflow_allowed_for_legacy_job_with_no_user_id( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """POST /api/v2/workflows/stop does NOT block legacy jobs where user_id is NULL. + + Jobs created before the ownership fix have user_id=None and must not be + broken by the ownership check (parity with the equivalent GET test). + """ + job_id = uuid4() + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.IN_PROGRESS, + type=JobType.WORKFLOW, + user_id=None, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflows/stop", + json={"job_id": str(job_id)}, + headers=headers, + ) + + assert response.status_code == 200, ( + "Legacy jobs with user_id=None must not be blocked by the ownership check" + ) + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_stop_workflow_returns_404_for_non_workflow_job_type( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """POST /api/v2/workflows/stop returns 404 for non-WORKFLOW job types. + + Prevents stop endpoint from cancelling ingestion or evaluation jobs. + """ + job_id = uuid4() + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.IN_PROGRESS, + type=JobType.INGESTION, + user_id=None, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflows/stop", + json={"job_id": str(job_id)}, + headers=headers, + ) + + assert response.status_code == 404 + result = response.json() + assert result["detail"]["code"] == "JOB_NOT_FOUND" + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_get_workflow_status_returns_404_for_non_workflow_job_type( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """GET /api/v2/workflows returns 404 for non-WORKFLOW job types. + + Prevents status endpoint from exposing ingestion or evaluation jobs. + """ + job_id = uuid4() + + async with session_scope() as session: + job = Job( + job_id=job_id, + flow_id=uuid4(), + status=JobStatus.IN_PROGRESS, + type=JobType.INGESTION, + user_id=None, + ) + session.add(job) + await session.flush() + + try: + headers = {"x-api-key": created_api_key.api_key} + response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) + + assert response.status_code == 404 + result = response.json() + assert result["detail"]["code"] == "JOB_NOT_FOUND" + finally: + async with session_scope() as session: + db_job = await session.get(Job, job_id) + if db_job: + await session.delete(db_job) + + @pytest.mark.security + async def test_background_job_stores_user_id_and_blocks_cross_user_access( + self, + client: AsyncClient, + created_api_key, + created_user_two_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """End-to-end: POST /workflows (background=true) stores user_id and blocks cross-user GET. + + This test exercises the real create_job code path to catch regressions + where user_id is not passed to create_job (causing user_id=NULL and + bypassing the ownership check — the exact bug found in Postman). + + Steps: + 1. Alice creates a background job via the real endpoint (no mock on create_job) + 2. Bob queries the job_id returned — must get 404 (ownership enforced at SQL level) + """ + flow_id = uuid4() + job_id_str = None + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Alice End-to-End Flow", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + + try: + # Mock only the task service to prevent real background execution + with patch("langflow.api.v2.workflow.get_task_service") as mock_task_svc: + mock_task_service = MagicMock() + mock_task_service.fire_and_forget_task = AsyncMock() + mock_task_svc.return_value = mock_task_service + + alice_headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflows", + json={ + "flow_id": str(flow_id), + "background": True, + "stream": False, + "inputs": {}, + }, + headers=alice_headers, + ) + + assert response.status_code in (200, 202), ( + f"Expected 200/202 from Alice's background job creation, got {response.status_code}: {response.text}" + ) + job_id_str = response.json()["job_id"] + + # Bob queries Alice's job — must be 404 (ownership enforced at SQL level) + bob_headers = {"x-api-key": created_user_two_api_key.api_key} + response = await client.get( + f"api/v2/workflows?job_id={job_id_str}", + headers=bob_headers, + ) + + assert response.status_code == 404, ( + f"Expected 404 Not Found but got {response.status_code}. " + "user_id is not being persisted in create_job — Bob can access Alice's job." + ) + assert response.json()["detail"]["code"] == "JOB_NOT_FOUND" + + finally: + async with session_scope() as session: + if job_id_str: + db_job = await session.get(Job, UUID(job_id_str)) + if db_job: + await session.delete(db_job) + db_flow = await session.get(Flow, flow_id) + if db_flow: + await session.delete(db_flow)