fix: harden flow-orchestration token refresh (mint from job_perms) (#10419)

* fix: harden flow-orchestration token refresh (mint from job_perms)

* refactor: address review nits on flow token refresh

backend/src/monitor.rs

@@ -46,7 +46,7 @@ use windmill_common::otel_oss::{
 use windmill_common::{
     agent_workers::DECODED_AGENT_TOKEN,
     apps::APP_WORKSPACED_ROUTE,
-    auth::create_token_for_owner,
+    auth::{create_token_for_owner, ephemeral_script_token_label},
     ee_oss::CriticalErrorChannel,
     email_oss::send_email_if_possible,
     error,
@@ -4662,13 +4662,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, node_n
             continue;
         }
         if let Some(job) = job.unwrap() {
-            let label = if job.permissioned_as != format!("u/{}", job.created_by)
-                && job.permissioned_as != job.created_by
-            {
-                format!("ephemeral-script-end-user-{}", job.created_by)
-            } else {
-                "ephemeral-script".to_string()
-            };
+            let label = ephemeral_script_token_label(&job.permissioned_as, &job.created_by);
             let token = create_token_for_owner(
                 &db,
                 &job.workspace_id,

backend/windmill-common/src/auth.rs

@@ -496,6 +496,31 @@ pub async fn get_job_perms<'a, E: sqlx::PgExecutor<'a>>(
     .await
 }
 
+/// A job token is refreshed once its remaining lifetime drops below this. It must exceed the
+/// 60s `jsonwebtoken` exp leeway, otherwise a token that still validates now could expire
+/// mid-orchestration after being judged fresh.
+pub const JOB_TOKEN_REFRESH_MARGIN_SECS: i64 = 120;
+
+/// Seconds until an internal job JWT expires, or `None` when `token` is not a decodable job
+/// JWT (e.g. the empty test token). The signature is intentionally not verified: the value only
+/// gates whether to refresh the token, never whose identity to assume.
+pub fn job_token_remaining_lifetime_secs(token: &str) -> Option<i64> {
+    let raw = token.strip_prefix("jwt_")?;
+    let claims: JWTAuthClaims = jwt::decode_without_verify(raw).ok()?;
+    Some(claims.exp as i64 - Utc::now().timestamp())
+}
+
+/// Label for an ephemeral job token. For a job run on behalf of an end user (its
+/// `permissioned_as` differs from its `created_by`) it encodes that end user so
+/// `username_override_from_label` can recover them; otherwise it is the plain script label.
+pub fn ephemeral_script_token_label(permissioned_as: &str, created_by: &str) -> String {
+    if permissioned_as != format!("u/{created_by}") && permissioned_as != created_by {
+        format!("ephemeral-script-end-user-{created_by}")
+    } else {
+        "ephemeral-script".to_string()
+    }
+}
+
 #[tracing::instrument(level = "trace", skip_all)]
 pub async fn create_token_for_owner(
     db: &DB,
@@ -679,6 +704,46 @@ pub mod aws {
 #[cfg(test)]
 mod tests {
     use super::is_user_token;
+    use super::{job_token_remaining_lifetime_secs, JWTAuthClaims, JOB_TOKEN_REFRESH_MARGIN_SECS};
+
+    fn job_jwt(exp_offset_secs: i64) -> String {
+        let claims = JWTAuthClaims {
+            email: String::new(),
+            username: String::new(),
+            is_admin: false,
+            is_operator: false,
+            groups: vec![],
+            folders: vec![],
+            label: None,
+            workspace_id: None,
+            workspace_ids: None,
+            exp: (chrono::Utc::now().timestamp() + exp_offset_secs) as usize,
+            job_id: None,
+            scopes: None,
+            audit_span: None,
+        };
+        // Signature is irrelevant — the gate decodes without verifying — so any key works.
+        let token = jsonwebtoken::encode(
+            &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::HS256),
+            &claims,
+            &jsonwebtoken::EncodingKey::from_secret(b"test"),
+        )
+        .unwrap();
+        format!("jwt_{token}")
+    }
+
+    #[test]
+    fn remaining_lifetime_reflects_exp_and_flags_near_expiry() {
+        // A token minted for less than the margin reads as needing a refresh...
+        let short = job_token_remaining_lifetime_secs(&job_jwt(30)).unwrap();
+        assert!(short < JOB_TOKEN_REFRESH_MARGIN_SECS);
+        // ...a long-lived one does not...
+        let long = job_token_remaining_lifetime_secs(&job_jwt(10_000)).unwrap();
+        assert!(long >= JOB_TOKEN_REFRESH_MARGIN_SECS);
+        // ...and a non-JWT token (e.g. the empty test-run token) yields no lifetime.
+        assert!(job_token_remaining_lifetime_secs("not-a-jwt").is_none());
+        assert!(job_token_remaining_lifetime_secs("").is_none());
+    }
 
     #[test]
     fn user_tokens_are_editable() {

backend/windmill-queue/src/jobs.rs

@@ -3397,13 +3397,10 @@ impl PulledJob {
 pub async fn create_token(db: &DB, job: &MiniPulledJob, perms: Option<JobPerms>) -> String {
     // skipping test runs
     if job.workspace_id != "" {
-        let label = if job.permissioned_as != format!("u/{}", job.created_by)
-            && job.permissioned_as != job.created_by
-        {
-            format!("ephemeral-script-end-user-{}", job.created_by)
-        } else {
-            "ephemeral-script".to_string()
-        };
+        let label = windmill_common::auth::ephemeral_script_token_label(
+            &job.permissioned_as,
+            &job.created_by,
+        );
         windmill_common::auth::create_token_for_owner(
             db,
             &job.workspace_id,

backend/windmill-worker/src/result_processor.rs

@@ -690,46 +690,60 @@ pub async fn handle_receive_completed_job(
     #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
 ) -> Option<Arc<MiniPulledJob>> {
     let workspace = jc.job.workspace_id.clone();
-    // The client built here drives post-completion orchestration (the next step's input
-    // transforms fetch prior step results), which outlives the finished step. The step's own
-    // token has a `SCRIPT_TOKEN_EXPIRY` lifetime, so reusing it would fail that orchestration
-    // once the step itself ran longer than the token lives; refresh it when the step is old enough.
-    let token_maybe_expired = jc
-        .duration
-        .is_some_and(|d| d as u64 >= *windmill_common::worker::SCRIPT_TOKEN_EXPIRY * 1000 / 2);
-    let token = if jc.job.is_flow_step() && token_maybe_expired {
-        // Mirror `create_token`'s label so run-on-behalf-of flows keep their end-user override.
-        let label = if jc.job.permissioned_as != format!("u/{}", jc.job.created_by)
-            && jc.job.permissioned_as != jc.job.created_by
-        {
-            format!("ephemeral-script-end-user-{}", jc.job.created_by)
-        } else {
-            "ephemeral-script".to_string()
-        };
-        match windmill_common::auth::create_token_for_owner(
-            db,
-            &jc.job.workspace_id,
+    // This client drives post-completion orchestration (the next step's input transforms fetch
+    // prior results) and outlives the finished step, so the step's own token can already be near
+    // expiry. Refresh it — but only from the server-written job_perms row, never from the
+    // completion payload's owner fields, which are untrusted on the agent-worker path.
+    let token = if jc.job.is_flow_step()
+        && windmill_common::auth::job_token_remaining_lifetime_secs(&jc.token)
+            .is_some_and(|r| r < windmill_common::auth::JOB_TOKEN_REFRESH_MARGIN_SECS)
+    {
+        let label = windmill_common::auth::ephemeral_script_token_label(
             &jc.job.permissioned_as,
-            &label,
-            *windmill_common::worker::SCRIPT_TOKEN_EXPIRY,
-            &jc.job.permissioned_as_email,
-            &jc.job.id,
-            None,
-            Some(format!(
-                "job-span-{}",
-                jc.job.flow_innermost_root_job.unwrap_or(jc.job.id)
-            )),
-        )
-        .warn_after_seconds(5)
-        .await
-        {
-            Ok(t) => t,
-            Err(e) => {
+            &jc.job.created_by,
+        );
+        match windmill_common::auth::get_job_perms(db, &jc.job.id, &jc.job.workspace_id).await {
+            Ok(Some(perms)) => windmill_common::auth::create_token_for_owner(
+                db,
+                &jc.job.workspace_id,
+                &jc.job.permissioned_as,
+                &label,
+                *windmill_common::worker::SCRIPT_TOKEN_EXPIRY,
+                &jc.job.permissioned_as_email,
+                &jc.job.id,
+                Some(perms),
+                Some(format!(
+                    "job-span-{}",
+                    jc.job.flow_innermost_root_job.unwrap_or(jc.job.id)
+                )),
+            )
+            .warn_after_seconds(5)
+            .await
+            .unwrap_or_else(|e| {
                 tracing::warn!(
                     "could not mint fresh flow-orchestration token for job {}, reusing step token: {e:#}",
                     jc.job.id
                 );
                 jc.token.clone()
+            }),
+            // No perms row (e.g. a zombie replay after the queue row was reaped): keep the step
+            // token rather than minting an identity from untrusted payload fields. The token is
+            // near expiry, so trace it — the downstream fetch may hit the original failure.
+            Ok(None) => {
+                tracing::warn!(
+                    "no job_perms row to refresh flow-orchestration token for job {}, reusing step token",
+                    jc.job.id
+                );
+                jc.token.clone()
+            }
+            // A transient DB error must not silently reuse the near-expired token without a trace,
+            // or the very failure this guards against recurs invisibly.
+            Err(e) => {
+                tracing::warn!(
+                    "could not load job_perms to refresh flow-orchestration token for job {}, reusing step token: {e:#}",
+                    jc.job.id
+                );
+                jc.token.clone()
             }
         }
     } else {