fix: validate step scopes in batch flows, fail closed on unvalidated step scopes

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014FHGE3ynoAu6yrgeg4kLwL

backend/windmill-api/src/jobs.rs

@@ -9746,6 +9746,11 @@ async fn add_batch_jobs(
                     "Path is required if no value is not provided"
                 ))?
             };
+            if windmill_common::scopes::validate_flow_step_job_token_scopes(&value)? {
+                windmill_common::min_version::MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES
+                    .assert()
+                    .await?;
+            }
             add_virtual_items_if_necessary(&mut value.modules);
             let flow_status = FlowStatus::new(&value);
             (

backend/windmill-common/src/scopes.rs

@@ -664,6 +664,13 @@ pub fn log_job_token_scopes_deploy(kind: &'static str, scopes: Option<&[String]>
     );
 }
 
+/// The cap a flow step or agent tool's own `job_token_scopes` puts on its jobs. A definition
+/// can reach the worker without the deploy-time validation (a raw flow), so a setting that
+/// does not validate restricts the step to no API access rather than being trusted as is.
+pub fn step_job_token_scopes(scopes: Option<&[String]>) -> Option<Vec<String>> {
+    scopes.map(|s| validate_job_token_scopes(s).unwrap_or_default())
+}
+
 /// The `scopes` claim of a job token minted from the job's effective scopes.
 pub fn job_token_jwt_scopes(effective: Option<Vec<String>>) -> Option<Vec<String>> {
     match effective {
@@ -732,5 +739,10 @@ mod job_token_scopes_tests {
             validate_job_token_scopes(&v(&["oidc:write", " oidc:write"])).unwrap(),
             v(&["oidc:write"])
         );
+        // A step setting that skipped deploy validation restricts rather than widens.
+        assert_eq!(
+            step_job_token_scopes(Some(&v(&["if_jobs:filter_tags:bun"]))),
+            Some(vec![])
+        );
     }
 }

backend/windmill-queue/src/jobs.rs

@@ -3741,7 +3741,7 @@ impl From<QueuedJobV2> for MiniCompletedJob {
             cache_ttl: job.cache_ttl,
             cache_ignore_s3_path: job.cache_ignore_s3_path,
             runnable_settings_handle: job.runnable_settings_handle,
-            // `QueuedJobV2` carries no args; a dependency job never completes through here.
+            // `QueuedJobV2` carries no args, which is what marks a binary-build job.
             build_binary_only: false,
             // Nor scopes: a caller whose completion can re-run the job (the monitor's zombie
             // recovery) fills them in from `job_perms` while the job is still queued.

backend/windmill-worker/src/ai/tools.rs

@@ -603,7 +603,10 @@ async fn enqueue_windmill_tool(
         // narrows that further.
         windmill_common::scopes::intersect_job_token_scopes(
             ctx.job.job_token_scopes.as_deref(),
-            tool_module.job_token_scopes.as_deref(),
+            windmill_common::scopes::step_job_token_scopes(
+                tool_module.job_token_scopes.as_deref(),
+            )
+            .as_deref(),
         )
         .as_deref(),
     )

backend/windmill-worker/src/worker_flow.rs

@@ -4756,7 +4756,10 @@ async fn push_next_flow_job(
             // narrows that further.
             windmill_common::scopes::intersect_job_token_scopes(
                 flow_job.job_token_scopes.as_deref(),
-                module.job_token_scopes.as_deref(),
+                windmill_common::scopes::step_job_token_scopes(
+                    module.job_token_scopes.as_deref(),
+                )
+                .as_deref(),
             )
             .as_deref(),
         )