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

Author: AntonioABLima

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 @@
 
     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 @@ async def get_jobs_by_flow_id(self, flow_id: UUID | str, page: int = 1, page_siz
         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 @@ async def create_job(
         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 @@ async def create_job(
             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 @@ async def create_job(
                 type=job_type,
                 asset_id=asset_id,
                 asset_type=asset_type,
+                user_id=user_id,
             )
             session.add(job)
             await session.flush()