address review (rubenfiszel): gate on token lifetime, mint only from job_perms row

Security: on the agent-worker path the completion payload's owner fields
(permissioned_as/email) are attacker-supplied, and create_token_for_owner(perms:
None) falls back to fetch_authed_from_permissioned_as — letting a completion mint
a token for an arbitrary identity. Instead fetch the server-written job_perms row
and pass Some(perms); when the row is absent (e.g. zombie replay after the queue
row was reaped) keep the step token rather than trusting the payload.

Gating: refresh based on the token's actual remaining lifetime (DB-free
decode_without_verify) instead of step duration. This confines the extra
get_job_perms lookup to near-expiry completions and also covers completions with
no recorded duration (Noop/cache), which the duration guard missed.

Add job_token_remaining_lifetime_secs + JOB_TOKEN_REFRESH_MARGIN_SECS (must
exceed the 60s JWT leeway) in windmill-common::auth, with unit tests pinning the
margin-vs-leeway invariant and the near-expiry detection.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/windmill-common/src/auth.rs

@@ -496,6 +496,20 @@ 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())
+}
+
 #[tracing::instrument(level = "trace", skip_all)]
 pub async fn create_token_for_owner(
     db: &DB,
@@ -679,6 +693,52 @@ 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};
+
+    // The refresh margin only prevents a mid-orchestration expiry if it stays above the JWT leeway.
+    #[test]
+    fn refresh_margin_exceeds_jwt_leeway() {
+        assert!(JOB_TOKEN_REFRESH_MARGIN_SECS > 60);
+    }
+
+    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-worker/src/result_processor.rs

@@ -690,14 +690,14 @@ 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 {
+    // 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)
+    {
         // 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
@@ -706,31 +706,33 @@ pub async fn handle_receive_completed_job(
         } else {
             "ephemeral-script".to_string()
         };
-        match 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,
-            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) => {
+        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) or a DB error:
+            // keep the step token rather than minting an identity from untrusted payload fields.
+            _ => jc.token.clone(),
         }
     } else {
         jc.token.clone()