From 3907c9f9512ebd73daf0a2f3ee2e8db6fb9f4df6 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 3 Oct 2025 19:27:58 +0200 Subject: [PATCH] feat: end user email env var (#6750) * feat: end user email env var * nit * nits --- backend/ee-repo-ref.txt | 2 +- ...03145612_add_end_user_to_job_perms.down.sql | 2 ++ ...1003145612_add_end_user_to_job_perms.up.sql | 2 ++ backend/tests/common/mod.rs | 1 + backend/windmill-api/src/apps.rs | 5 +++++ backend/windmill-api/src/flows.rs | 2 ++ backend/windmill-api/src/jobs.rs | 15 +++++++++++++++ backend/windmill-api/src/resources.rs | 4 +++- backend/windmill-api/src/scripts.rs | 16 +++++++++++++--- .../src/triggers/trigger_helpers.rs | 4 +++- backend/windmill-api/src/variables.rs | 18 ++++++++---------- backend/windmill-common/src/variables.rs | 17 ++++++++++++----- backend/windmill-common/src/worker.rs | 2 +- backend/windmill-queue/src/jobs.rs | 17 +++++++++++++---- backend/windmill-queue/src/schedule.rs | 1 + backend/windmill-worker/src/ai_executor.rs | 1 + backend/windmill-worker/src/bun_executor.rs | 9 ++++++--- backend/windmill-worker/src/common.rs | 1 + backend/windmill-worker/src/deno_executor.rs | 1 + backend/windmill-worker/src/python_executor.rs | 2 ++ backend/windmill-worker/src/worker_flow.rs | 1 + .../windmill-worker/src/worker_lockfiles.rs | 1 + 22 files changed, 95 insertions(+), 29 deletions(-) create mode 100644 backend/migrations/20251003145612_add_end_user_to_job_perms.down.sql create mode 100644 backend/migrations/20251003145612_add_end_user_to_job_perms.up.sql diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 7df83e8211..71d25622e3 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -e4e3056eeaf12d7fe71579c0d1f6aee828fe5ea7 +33bdef405c678616b12084cf779a68a62d1f477e \ No newline at end of file diff --git a/backend/migrations/20251003145612_add_end_user_to_job_perms.down.sql b/backend/migrations/20251003145612_add_end_user_to_job_perms.down.sql new file mode 100644 index 0000000000..a6834882c7 --- /dev/null +++ b/backend/migrations/20251003145612_add_end_user_to_job_perms.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE job_perms DROP COLUMN end_user_email; \ No newline at end of file diff --git a/backend/migrations/20251003145612_add_end_user_to_job_perms.up.sql b/backend/migrations/20251003145612_add_end_user_to_job_perms.up.sql new file mode 100644 index 0000000000..775fea57f5 --- /dev/null +++ b/backend/migrations/20251003145612_add_end_user_to_job_perms.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE job_perms ADD COLUMN end_user_email VARCHAR(255); \ No newline at end of file diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 748fae04c1..026eee12f7 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -156,6 +156,7 @@ impl RunJob { None, None, false, + None, ) .await .expect("push has to succeed"); diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 21acc76b88..d3e6d549b3 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -1243,6 +1243,7 @@ async fn create_app_internal<'a>( None, Some(&authed.clone().into()), false, + None, ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1622,6 +1623,7 @@ async fn update_app_internal<'a>( None, Some(&authed.clone().into()), false, + None, ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1908,6 +1910,8 @@ async fn execute_component( (email.as_str(), permissioned_as) }; + let end_user_email = opt_authed.as_ref().map(|a| a.email.clone()); + let (uuid, tx) = push( &db, tx, @@ -1937,6 +1941,7 @@ async fn execute_component( None, None, false, + end_user_email, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index d2c93adc92..caf65a473e 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -545,6 +545,7 @@ async fn create_flow( None, Some(&authed.clone().into()), false, + None, ) .await?; @@ -1012,6 +1013,7 @@ async fn update_flow( None, Some(&authed.clone().into()), false, + None, ) .await?; sqlx::query!( diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index bc9d63ce9e..e721facc6e 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -4103,6 +4103,7 @@ pub async fn run_flow_by_path_inner( None, push_authed.as_ref(), false, + None, ) .await?; @@ -4219,6 +4220,7 @@ pub async fn restart_flow( completed_job.priority, Some(&authed.clone().into()), false, + None, ) .await?; tx.commit().await?; @@ -4321,6 +4323,7 @@ pub async fn run_script_by_path_inner( None, push_authed.as_ref(), false, + None, ) .await?; tx.commit().await?; @@ -4473,6 +4476,7 @@ pub async fn run_workflow_as_code( None, push_authed.as_ref(), false, + None, ) .await?; @@ -5016,6 +5020,7 @@ pub async fn run_wait_result_job_by_path_get( None, push_authed.as_ref(), false, + None, ) .await?; tx.commit().await?; @@ -5168,6 +5173,7 @@ pub async fn run_wait_result_script_by_path_internal( None, push_authed.as_ref(), false, + None, ) .await?; tx.commit().await?; @@ -5284,6 +5290,7 @@ pub async fn run_wait_result_script_by_hash( None, push_authed.as_ref(), false, + None, ) .await?; tx.commit().await?; @@ -5595,6 +5602,7 @@ pub async fn run_wait_result_flow_by_path_internal( None, push_authed.as_ref(), false, + None, ) .await?; @@ -5686,6 +5694,7 @@ async fn run_preview_script( None, Some(&authed.clone().into()), false, + None, ) .await?; tx.commit().await?; @@ -5802,6 +5811,7 @@ async fn run_bundle_preview_script( None, Some(&authed.clone().into()), false, + None, ) .await?; job_id = Some(uuid); @@ -5939,6 +5949,7 @@ async fn run_dependencies_job( None, Some(&authed.clone().into()), false, + None, ) .await?; tx.commit().await?; @@ -6006,6 +6017,7 @@ async fn run_flow_dependencies_job( None, Some(&authed.clone().into()), false, + None, ) .await?; tx.commit().await?; @@ -6349,6 +6361,7 @@ async fn run_preview_flow_job( None, Some(&authed.clone().into()), false, + None, ) .await?; tx.commit().await?; @@ -6522,6 +6535,7 @@ async fn run_dynamic_select( None, Some(&authed.clone().into()), false, + None, ) .await?; tx.commit().await?; @@ -6649,6 +6663,7 @@ pub async fn run_job_by_hash_inner( None, push_authed.as_ref(), false, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index 087d31cdcf..fc762ec451 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -515,7 +515,8 @@ pub async fn transform_json_value<'c>( match v { Value::String(y) if y.starts_with("$var:") => { let path = y.strip_prefix("$var:").unwrap(); - let userdb_authed = UserDbWithOptAuthed { authed: authed, user_db: user_db.clone(), db: db.clone() }; + let userdb_authed = + UserDbWithOptAuthed { authed: authed, user_db: user_db.clone(), db: db.clone() }; let v = crate::variables::get_value_internal( &userdb_authed, @@ -616,6 +617,7 @@ pub async fn transform_json_value<'c>( job.root_job.map(|x| x.to_string()), Some(job.scheduled_for.clone()), None, + None, ) .await; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 40657b4510..7791104e11 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -41,7 +41,12 @@ use windmill_audit::ActionKind; use windmill_worker::process_relative_imports; use windmill_common::{ - assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType}, error::to_anyhow, s3_helpers::upload_artifact_to_store, scripts::hash_script, utils::WarnAfterExt, worker::CLOUD_HOSTED + assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType}, + error::to_anyhow, + s3_helpers::upload_artifact_to_store, + scripts::hash_script, + utils::WarnAfterExt, + worker::CLOUD_HOSTED, }; use windmill_common::{ @@ -418,7 +423,12 @@ async fn create_snapshot_script( uploaded = true; let path = windmill_common::s3_helpers::bundle(&w_id, &hash); - upload_artifact_to_store(&path, data, &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR).await?; + upload_artifact_to_store( + &path, + data, + &windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR, + ) + .await?; } // println!("Length of `{}` is {} bytes", name, data.len()); } @@ -438,7 +448,6 @@ async fn create_snapshot_script( return Ok((StatusCode::CREATED, format!("{}", script_hash.unwrap()))); } - async fn list_paths_from_workspace_runnable( authed: ApiAuthed, Extension(user_db): Extension, @@ -980,6 +989,7 @@ async fn create_script_internal<'c>( None, Some(&authed.clone().into()), false, + None, ) .await?; Ok((hash, new_tx, None)) diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index a1a5f62e50..16e4140534 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -684,7 +684,8 @@ pub async fn trigger_runnable_and_wait_for_raw_result_with_error_ctx( error_handler_path, error_handler_args, trigger_path, - ).await?; + ) + .await?; if !success { Err(windmill_common::error::Error::internal_err(format!( @@ -857,6 +858,7 @@ async fn trigger_script_with_retry_and_error_handler( None, push_authed.as_ref(), false, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/variables.rs b/backend/windmill-api/src/variables.rs index 30cc9c0df3..80eac1c9be 100644 --- a/backend/windmill-api/src/variables.rs +++ b/backend/windmill-api/src/variables.rs @@ -83,6 +83,7 @@ async fn list_contextual_variables( Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()), Some(chrono::offset::Utc::now()), Some(ScriptHash(1234567890)), + None, ) .await .to_vec(), @@ -843,11 +844,11 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres } else if !value.is_empty() { let mc = build_crypt(&db, &w_id).await?; decrypt(&mc, value).map_err(|e| { - Error::internal_err(format!( - "Error decrypting variable {}: {}", - variable.path, e - )) - })? + Error::internal_err(format!( + "Error decrypting variable {}: {}", + variable.path, e + )) + })? } else { "".to_string() } @@ -884,11 +885,8 @@ pub async fn get_variable_or_self(path: String, db: &DB, w_id: &str) -> Result, scheduled_for: Option>, runnable_id: Option, + end_user_email: Option, ) -> Vec { let state_path = { let trigger = if schedule_path.is_some() { @@ -389,6 +390,12 @@ pub async fn get_reserved_variables( description: "Hash of the script. Useful as cache key for cache that should be runnable specific.".to_string(), is_custom: false, }, + ContextualVariable { + name: "WM_END_USER_EMAIL".to_string(), + value: end_user_email.unwrap_or_else(|| "".to_string()), + description: "Email of the end user that executed the current script. Only available when triggered from an app.".to_string(), + is_custom: false, + }, ].into_iter().chain(custom_envs.into_iter().map(|(name, value)| ContextualVariable { name, value, diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 5c1381ecd6..912ba5c1f7 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -395,7 +395,7 @@ fn format_pull_query(peek: String) -> String { j.timeout, j.flow_step_id, j.cache_ttl, j.priority, j.raw_code, j.raw_lock, j.raw_flow, j.script_entrypoint_override, j.preprocessed, pj.runnable_path as parent_runnable_path, COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin, - p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders + p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email FROM q, j LEFT JOIN v2_job_status f USING (id) LEFT JOIN job_perms p ON p.job_id = j.id diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index b82b3d6b00..3010874ac5 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -446,6 +446,7 @@ pub async fn push_init_job<'c>( None, None, false, + None, ) .await?; inner_tx.commit().await?; @@ -500,6 +501,7 @@ pub async fn push_periodic_bash_job<'c>( None, None, false, + None, ) .await?; inner_tx.commit().await?; @@ -1371,6 +1373,7 @@ async fn restart_job_if_perpetual_inner( queued_job.priority, None, false, + None, ) .await?; tx.commit().await?; @@ -1858,6 +1861,7 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( priority, None, false, + None, ) .await?; tx.commit().await?; @@ -1913,6 +1917,7 @@ pub struct MiniPulledJob { pub trigger: Option, pub trigger_kind: Option, pub visible_to_owner: bool, + pub permissioned_as_end_user_email: Option, } #[derive(Serialize, Deserialize, Debug, Clone)] @@ -1989,6 +1994,7 @@ impl MiniPulledJob { None }, visible_to_owner: job.visible_to_owner.clone(), + permissioned_as_end_user_email: None, } } pub fn is_flow(&self) -> bool { @@ -2189,7 +2195,8 @@ pub async fn get_mini_pulled_job<'c>( script_entrypoint_override, trigger, trigger_kind as \"trigger_kind: JobTriggerKind\", - visible_to_owner + visible_to_owner, + NULL as permissioned_as_end_user_email FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1", job_id, ) @@ -3147,6 +3154,7 @@ pub async fn push<'c, 'd>( _priority_override: Option, authed: Option<&Authed>, running: bool, // whether the job is already running: only set this to true if you don't want the job to be picked up by a worker from the queue. It will also set started_at to now. + end_user_email: Option, ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { #[cfg(feature = "cloud")] if *CLOUD_HOSTED { @@ -4233,8 +4241,8 @@ pub async fn push<'c, 'd>( INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null) ), inserted_job_perms AS ( - INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) - values ($1, $32, $33, $34, $35, $36, $37, $2) + INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email) + values ($1, $32, $33, $34, $35, $36, $37, $2, $41) ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2 ) INSERT INTO v2_job_queue @@ -4284,6 +4292,7 @@ pub async fn push<'c, 'd>( root_job, trigger_kind as Option, running, + end_user_email, ) .execute(&mut *tx) .warn_after_seconds(1) @@ -4697,7 +4706,7 @@ pub async fn get_same_worker_job( v2_job.raw_flow, pj.runnable_path as parent_runnable_path, p.email as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin, - p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders + p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 6594837ffa..0870d2f24a 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -331,6 +331,7 @@ pub async fn push_scheduled_job<'c>( None, push_authed, false, + None, ) .await?; diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index c7555e6334..08c8ede63a 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -855,6 +855,7 @@ pub async fn run_agent( job_priority, job_perms.as_ref(), true, + None, ) .await?; diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 215d360879..4e03043c85 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -25,7 +25,11 @@ use crate::{ DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_BIN_PATH, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, TZ_ENV, }; -use windmill_common::{client::AuthedClient, s3_helpers::BundleFormat, scripts::{id_to_codebase_info, CodebaseInfo}}; +use windmill_common::{ + client::AuthedClient, + s3_helpers::BundleFormat, + scripts::{id_to_codebase_info, CodebaseInfo}, +}; #[cfg(windows)] use crate::SYSTEM_ROOT; @@ -910,7 +914,6 @@ pub async fn handle_bun_job( let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(Some(&base_internal_url)).await; - let main_override = job.script_entrypoint_override.as_deref(); let apply_preprocessor = job.flow_step_id.as_deref() != Some("preprocessor") && job.preprocessed == Some(false); @@ -1621,11 +1624,11 @@ pub async fn start_worker( None, None, None, + None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; - let mut format = BundleFormat::Cjs; if let Some(codebase) = codebase.as_ref() { let pulled_codebase = pull_codebase(w_id, codebase, job_dir).await?; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index a90d14ca46..ff58d3a41e 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -452,6 +452,7 @@ pub async fn get_reserved_variables( Some(get_root_job_id(job).to_string()), Some(job.scheduled_for.clone()), job.runnable_id, + job.permissioned_as_end_user_email.clone(), ) .await .to_vec(); diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index fffe236fe9..cda1f78fdc 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -556,6 +556,7 @@ pub async fn start_worker( None, None, None, + None, ) .await; let context_envs = build_envs_map(context.to_vec()).await; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 80b577f5f3..c4e88e0853 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -2173,6 +2173,7 @@ pub async fn start_worker( None, None, None, + None, ) .await .to_vec(); @@ -2294,6 +2295,7 @@ for line in sys.stdin: None, None, None, + None, ) .await; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 1587b95844..1e6a7befe2 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -3153,6 +3153,7 @@ async fn push_next_flow_job( new_job_priority_override, job_perms.as_ref(), false, + None, ) .warn_after_seconds(2) .await?; diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 43a5125d79..59f0f87552 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -833,6 +833,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( None, None, false, + None, ) .await?; tracing::info!(