fix: fail closed on perms read errors, refuse restricted queue imports, gate step scopes in previews

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

backend/tests/job_token_scopes.rs

@@ -148,6 +148,46 @@ async fn test_restricted_job_token_is_confined(db: Pool<Postgres>) -> anyhow::Re
         assert_eq!(resp.status() == StatusCode::FORBIDDEN, refused, "{flow}");
     }
 
+    // Imported queue rows bypass push, so a restricted token cannot import, even an admin's.
+    let admin_job = Uuid::parse_str("b0000000-0000-0000-0000-000000000004")?;
+    sqlx::query(
+        "INSERT INTO v2_job (id, workspace_id, created_by, permissioned_as, permissioned_as_email,
+            kind, script_lang, runnable_path, tag)
+        VALUES ($1, 'test-workspace', 'test-user', 'u/test-user', 'test@windmill.dev',
+            'script', 'deno', 'u/test-user/agent', 'deno')",
+    )
+    .bind(admin_job)
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups,
+            workspace_id, job_token_scopes)
+        VALUES ($1, 'test@windmill.dev', 'test-user', true, false, '{}', '{}', 'test-workspace',
+            '{jobs:write}')",
+    )
+    .bind(admin_job)
+    .execute(&db)
+    .await?;
+    let admin_token = windmill_common::auth::create_token_for_owner(
+        &db,
+        "test-workspace",
+        "u/test-user",
+        "ephemeral-script",
+        300,
+        "test@windmill.dev",
+        &admin_job,
+        None,
+        None,
+    )
+    .await?;
+    let resp = client
+        .post(format!("{base}/jobs/queue/import"))
+        .bearer_auth(&admin_token)
+        .json(&json!([]))
+        .send()
+        .await?;
+    assert_eq!(resp.status(), StatusCode::FORBIDDEN, "{}", resp.text().await?);
+
     // A job it starts is capped at its own scopes, intersected with the target's setting.
     let run_token = job_token(&db, RUN_JOB).await?;
     for (path, expected) in [

backend/windmill-api-jobs/src/jobs_export.rs

@@ -449,6 +449,13 @@ pub async fn import_queued_jobs(
         ));
     }
 
+    // Imported rows bypass `push`, so nothing would cap their tokens: a restricted job token
+    // could otherwise queue arbitrary jobs that run with its owner's full permissions.
+    if authed.job_id.is_some() && authed.scopes.is_some() {
+        return Err(error::Error::PermissionDenied(
+            "A restricted job token cannot import queued jobs".to_string(),
+        ));
+    }
     // An imported job gets no `job_perms` row, so its token would be minted without the
     // restriction it was queued with.
     if let Some(job) = jobs.iter().find(|job| job.job_token_scopes.is_some()) {

backend/windmill-api/src/jobs.rs

@@ -9759,6 +9759,13 @@ async fn run_preview_flow_job(
     // jobs:run scope so a narrowly-scoped token cannot escape its scope. See run_preview_script.
     check_scopes(&authed, || format!("jobs:run"))?;
     require_path_read_access_for_preview(&authed, &raw_flow.path)?;
+    // Step restrictions apply to a preview as they do to a deployed run, so they are checked
+    // the same way: an invalid entry or an older worker would leave a step unrestricted.
+    if windmill_common::scopes::validate_flow_step_job_token_scopes(&raw_flow.value)? {
+        windmill_common::min_version::MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES
+            .assert()
+            .await?;
+    }
     // A builder must be able to test what it composes, but the submitted value is not the stored
     // one: without this the preview is a way to run inline code the write path refuses.
     if authed.is_operator {

backend/windmill-common/src/auth.rs

@@ -756,9 +756,15 @@ pub async fn create_token_for_owner(
             let scopes = jp.job_token_scopes.take();
             (jp.into(), scopes)
         }
+        // A failed read must not mint as if the job had no row: that would drop its restriction.
+        Err(e) => {
+            return Err(Error::internal_err(format!(
+                "Could not read permissions for job {job_id}: {e:#}"
+            )))
+        }
         // Push writes a job's `job_perms` row and its scopes in one statement, so a job with no
         // row was never restricted.
-        _ => {
+        Ok(None) => {
             tracing::warn!("Could not get permissions for job {job_id} from job_perms table, getting permissions directly...");
             let authed = fetch_authed_from_permissioned_as(owner, email, w_id, db)
                 .await

backend/windmill-queue/src/jobs.rs

@@ -2146,9 +2146,16 @@ async fn restart_job_if_perpetual_inner(
                 // TODO(debouncing): handle properly
                 debouncing_settings: DebouncingSettings::default(),
                 labels: None, // labels already set on original job
-                // The run it replaces caps it (`scope_ceiling`), and that run already
-                // carried this script's setting.
-                job_token_scopes: None,
+                // Read from the script, not only from the run it replaces: that run's
+                // `job_perms` row (the ceiling) may already be swept once it left the queue.
+                job_token_scopes: sqlx::query_scalar!(
+                    "SELECT job_token_scopes FROM script WHERE hash = $1 AND workspace_id = $2",
+                    hash.0,
+                    &queued_job.workspace_id
+                )
+                .fetch_optional(db)
+                .await?
+                .flatten(),
             },
             PushArgs::from(&args.0),
             &queued_job.created_by,

backend/windmill-worker/src/worker_flow.rs

@@ -4329,14 +4329,15 @@ async fn push_next_flow_job(
     };
 
     // only start runners if we're not already in a squash for loop
-    // Runners would run its steps with their own unscoped token: a restricted flow runs them
-    // as regular jobs.
+    // Runners would run its steps with their own unscoped token: a restricted flow, or a loop
+    // that restricts itself or any step in it, runs them as regular jobs.
     let start_runners = flow_runners.is_none()
         && flow_job.job_token_scopes.is_none()
         && matches!(
             next_status,
             NextStatus::NextLoopIteration { start_runners: true, .. }
-        );
+        )
+        && !restricts_any_step(module);
 
     let do_not_pass_runners = matches!(next_status, NextStatus::NextStep { .. })
         && flow_runners
@@ -6342,6 +6343,16 @@ pub fn raw_script_to_payload(
     }
 }
 
+/// Whether `module`, or any step or agent tool under it, sets `job_token_scopes`.
+fn restricts_any_step(module: &FlowModule) -> bool {
+    let mut any = false;
+    let _ = FlowModule::traverse_modules(&vec![module.clone()], &mut |m: &FlowModule| {
+        any |= m.job_token_scopes.is_some();
+        Ok(())
+    });
+    any
+}
+
 async fn flow_to_payload(
     path: String,
     delete_after_use: bool,