From e0d6dc1a1997514bfcaa8615de61daff4a2f81ca Mon Sep 17 00:00:00 2001 From: Alexander Petric Date: Fri, 31 Jul 2026 00:05:15 +0200 Subject: [PATCH] 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 | 10 +-- backend/windmill-common/src/auth.rs | 65 +++++++++++++++ backend/windmill-queue/src/jobs.rs | 11 +-- .../windmill-worker/src/result_processor.rs | 82 +++++++++++-------- 4 files changed, 119 insertions(+), 49 deletions(-) diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index ccff5c2c7b..336956415e 100644 --- a/backend/src/monitor.rs +++ b/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, 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, diff --git a/backend/windmill-common/src/auth.rs b/backend/windmill-common/src/auth.rs index bd6bf1ef9d..cbaa392cd4 100644 --- a/backend/windmill-common/src/auth.rs +++ b/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 { + 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() { diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index c4f83f4c4b..ebae8a2647 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3397,13 +3397,10 @@ impl PulledJob { pub async fn create_token(db: &DB, job: &MiniPulledJob, perms: Option) -> 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, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index bf3993bfe6..ccf12939d7 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/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> { 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 {