fix: enforce IDOR protection on v2 workflow job endpoints (#12398)

* 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

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

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

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

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

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)