From e7165f3357a2ba7a690accd78a03c2518aa61860 Mon Sep 17 00:00:00 2001 From: HugoCasa Date: Tue, 3 Oct 2023 18:14:37 +0200 Subject: [PATCH] fix: add reserved variables in args (#2371) * fix: add reserved variables in args * fix: only get job when necessary --- backend/Cargo.lock | 1 + backend/windmill-api/Cargo.toml | 3 +- backend/windmill-api/openapi.yaml | 8 ++- backend/windmill-api/src/resources.rs | 68 +++++++++++++++++-- backend/windmill-worker/src/bash_executor.rs | 4 ++ .../windmill-worker/src/bigquery_executor.rs | 3 +- backend/windmill-worker/src/bun_executor.rs | 2 +- backend/windmill-worker/src/common.rs | 43 +++++++++++- backend/windmill-worker/src/deno_executor.rs | 2 +- backend/windmill-worker/src/go_executor.rs | 4 +- .../windmill-worker/src/graphql_executor.rs | 3 +- backend/windmill-worker/src/js_eval.rs | 2 +- backend/windmill-worker/src/mysql_executor.rs | 3 +- backend/windmill-worker/src/pg_executor.rs | 3 +- .../windmill-worker/src/python_executor.rs | 2 +- .../windmill-worker/src/snowflake_executor.rs | 3 +- backend/windmill-worker/src/worker.rs | 23 ++++--- 17 files changed, 148 insertions(+), 29 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index f661a5b32c..aff0f8af1f 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -7182,6 +7182,7 @@ dependencies = [ "tracing", "tracing-subscriber", "urlencoding", + "uuid 1.4.1", "windmill-audit", "windmill-common", "windmill-parser", diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 8cdcd80f6c..13c4b7fdee 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -74,4 +74,5 @@ bytes.workspace = true mail-send.workspace = true samael = { workspace = true, optional = true } async-recursion.workspace = true -rsa.workspace = true \ No newline at end of file +rsa.workspace = true +uuid.workspace = true \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 15af01f70c..62de5448c0 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -678,7 +678,6 @@ paths: schema: type: string - /users/tutorial_progress: get: summary: get tutorial progress @@ -2065,6 +2064,12 @@ paths: parameters: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/Path" + - name: job_id + description: job id + in: query + schema: + type: string + format: uuid responses: "200": description: resource value @@ -2608,7 +2613,6 @@ paths: - path - content - /w/{workspace}/scripts/list: get: summary: list all available scripts diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index aa698db650..a5a5422d27 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -8,7 +8,7 @@ use crate::{ db::{ApiAuthed, DB}, - users::{maybe_refresh_folders, require_owner_of_path}, + users::{maybe_refresh_folders, require_owner_of_path, Tokened}, webhook_util::{WebhookMessage, WebhookShared}, }; use axum::{ @@ -21,11 +21,14 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use sql_builder::{bind::Bind, SqlBuilder}; use sqlx::{FromRow, Postgres, Transaction}; +use uuid::Uuid; use windmill_audit::{audit_log, ActionKind}; use windmill_common::{ db::UserDB, error::{Error, JsonResult, Result}, + jobs::QueuedJob, utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath}, + variables, }; pub fn workspaced_service() -> Router { @@ -303,10 +306,16 @@ async fn get_resource_value( Ok(Json(value)) } +#[derive(Deserialize)] +struct JobInfo { + job_id: Option, +} async fn get_resource_value_interpolated( authed: ApiAuthed, Extension(user_db): Extension, + Tokened { token }: Tokened, Path((w_id, path)): Path<(String, StripPath)>, + Query(job_info): Query, ) -> JsonResult> { let path = path.to_path(); let mut tx = user_db.clone().begin(&authed).await?; @@ -323,7 +332,7 @@ async fn get_resource_value_interpolated( let value = not_found_if_none(value_o, "Resource", path)?; if let Some(value) = value { Ok(Json(Some( - transform_json_value(&authed, &user_db, &w_id, value).await?, + transform_json_value(&authed, &user_db, &w_id, value, &job_info.job_id, &token).await?, ))) } else { Ok(Json(None)) @@ -338,6 +347,8 @@ pub async fn transform_json_value<'c>( user_db: &UserDB, workspace: &str, v: Value, + job_id: &Option, + token: &str, ) -> Result { match v { Value::String(y) if y.starts_with("$var:") => { @@ -363,16 +374,65 @@ pub async fn transform_json_value<'c>( tx.commit().await?; let v = not_found_if_none(v, "Resource", path)?; if let Some(v) = v { - transform_json_value(authed, user_db, workspace, v).await + transform_json_value(authed, user_db, workspace, v, job_id, token).await } else { Ok(Value::Null) } } + Value::String(y) if y.starts_with("$") && job_id.is_some() => { + let mut tx = user_db.clone().begin(authed).await?; + let job = sqlx::query_as::<_, QueuedJob>( + "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2", + ) + .bind(job_id.unwrap()) + .bind(workspace) + .fetch_optional(&mut *tx) + .await?; + tx.commit().await?; + + let job = not_found_if_none(job, "Job", job_id.unwrap().to_string())?; + + let flow_path = if let Some(uuid) = job.parent_job { + let mut tx: Transaction<'_, Postgres> = user_db.clone().begin(authed).await?; + let p = sqlx::query_scalar!("SELECT script_path FROM queue WHERE id = $1", uuid) + .fetch_optional(&mut *tx) + .await? + .flatten(); + tx.commit().await?; + p + } else { + None + }; + + let variables = variables::get_reserved_variables( + &job.workspace_id, + token, + &job.email, + &job.created_by, + &job.id.to_string(), + &job.permissioned_as, + job.script_path.clone(), + job.parent_job.map(|x| x.to_string()), + flow_path, + job.schedule_path.clone(), + job.flow_step_id.clone(), + ) + .await; + + let name = y.strip_prefix("$").unwrap(); + + let value = variables + .iter() + .find(|x| x.name == name) + .map(|x| x.value.clone()) + .unwrap_or_else(|| y); + Ok(serde_json::json!(value)) + } Value::Object(mut m) => { for (a, b) in m.clone().into_iter() { m.insert( a.clone(), - transform_json_value(authed, user_db, workspace, b).await?, + transform_json_value(authed, user_db, workspace, b, job_id, token).await?, ); } Ok(Value::Object(m)) diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 173fc5ad72..c54d3c05a7 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -53,6 +53,8 @@ pub async fn handle_bash_job( &client, &job.workspace_id, job.args.clone().unwrap_or_else(|| json!({})), + job, + db, ) .await? { @@ -173,6 +175,8 @@ pub async fn handle_powershell_job( &client, &job.workspace_id, job.args.clone().unwrap_or_else(|| json!({})), + job, + db, ) .await? { diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index fd2615c3ac..cc2570cfcc 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -55,9 +55,10 @@ pub async fn do_bigquery( job: QueuedJob, client: &AuthedClient, query: &str, + db: &sqlx::Pool, ) -> windmill_common::error::Result { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), &job, db).await?) } else { None }; diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 3e67382a85..a02f1ea8c5 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -385,7 +385,7 @@ run().catch(async (e) => {{ let reserved_variables_args_out_f = async { let client = client.get_authed().await; let args_and_out_f = async { - create_args_and_out_file(&client, job, job_dir).await?; + create_args_and_out_file(&client, job, job_dir, db).await?; Ok(()) as Result<()> }; let reserved_variables_f = async { diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 2e2827d6c7..1840f6079c 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -48,9 +48,10 @@ pub async fn create_args_and_out_file( client: &AuthedClient, job: &QueuedJob, job_dir: &str, + db: &Pool, ) -> Result<(), Error> { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), job, db).await?) } else { None }; @@ -84,6 +85,8 @@ pub async fn transform_json_value( client: &AuthedClient, workspace: &str, v: Value, + job: &QueuedJob, + db: &Pool, ) -> error::Result { match v { Value::String(y) if y.starts_with("$var:") => { @@ -104,16 +107,50 @@ pub async fn transform_json_value( } Ok(client .get_client() - .get_resource_value_interpolated(workspace, path) + .get_resource_value_interpolated(workspace, path, Some(&job.id)) .await .map_err(|_| Error::NotFound(format!("Resource {path} not found for `{name}`")))? .into_inner()) } + Value::String(y) if y.starts_with("$") => { + let flow_path = if let Some(uuid) = job.parent_job { + sqlx::query_scalar!("SELECT script_path FROM queue WHERE id = $1", uuid) + .fetch_optional(db) + .await? + .flatten() + } else { + None + }; + + let variables = variables::get_reserved_variables( + &job.workspace_id, + &client.token, + &job.email, + &job.created_by, + &job.id.to_string(), + &job.permissioned_as, + job.script_path.clone(), + job.parent_job.map(|x| x.to_string()), + flow_path, + job.schedule_path.clone(), + job.flow_step_id.clone(), + ) + .await; + + let name = y.strip_prefix("$").unwrap(); + + let value = variables + .iter() + .find(|x| x.name == name) + .map(|x| x.value.clone()) + .unwrap_or_else(|| y); + Ok(json!(value)) + } Value::Object(mut m) => { for (a, b) in m.clone().into_iter() { m.insert( a.clone(), - transform_json_value(&a, client, workspace, b).await?, + transform_json_value(&a, client, workspace, b, job, &db).await?, ); } Ok(Value::Object(m)) diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 377a120857..106e498a79 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -248,7 +248,7 @@ run().catch(async (e) => {{ let reserved_variables_args_out_f = async { let client = client.get_authed().await; let args_and_out_f = async { - create_args_and_out_file(&client, job, job_dir).await?; + create_args_and_out_file(&client, job, job_dir, db).await?; Ok(()) as Result<()> }; let reserved_variables_f = async { diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 2ec04e4c06..d1fbf34439 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -90,7 +90,7 @@ pub async fn handle_go_job( logs.push_str("\n\n--- GO CODE EXECUTION ---\n"); set_logs(logs, &job.id, db).await; - create_args_and_out_file(client, job, job_dir).await?; + create_args_and_out_file(client, job, job_dir, db).await?; { let sig = windmill_parser_go::parse_go_sig(&inner_content)?; @@ -212,7 +212,7 @@ func Run(req Req) (interface{{}}, error){{ tokio::fs::copy(path, format!("{job_dir}/main")).await?; logs.push_str("\n\n--- GO CODE EXECUTION ---\n"); set_logs(logs, &job.id, db).await; - create_args_and_out_file(client, job, job_dir).await?; + create_args_and_out_file(client, job, job_dir, db).await?; } let reserved_variables = get_reserved_variables(job, &client.token, db).await?; diff --git a/backend/windmill-worker/src/graphql_executor.rs b/backend/windmill-worker/src/graphql_executor.rs index 6857178eca..7a6b359f82 100644 --- a/backend/windmill-worker/src/graphql_executor.rs +++ b/backend/windmill-worker/src/graphql_executor.rs @@ -31,9 +31,10 @@ pub async fn do_graphql( job: QueuedJob, client: &AuthedClient, query: &str, + db: &sqlx::Pool, ) -> windmill_common::error::Result { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), &job, db).await?) } else { None }; diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index afc50a3719..6e77daab6d 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -402,7 +402,7 @@ async fn op_resource( if let Some(client) = client { let result = client .get_client() - .get_resource_value_interpolated(&client.workspace, path) + .get_resource_value_interpolated(&client.workspace, path, None) .await?; Ok(result.into_inner()) } else { diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index d9b626673e..36d8154a0c 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -26,9 +26,10 @@ pub async fn do_mysql( job: QueuedJob, client: &AuthedClient, query: &str, + db: &sqlx::Pool, ) -> windmill_common::error::Result { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), &job, db).await?) } else { None }; diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 08d0f02380..5fd8989d75 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -36,9 +36,10 @@ pub async fn do_postgresql( job: QueuedJob, client: &AuthedClient, query: &str, + db: &sqlx::Pool, ) -> error::Result { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), &job, db).await?) } else { None }; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 975b73d173..dac0d68fd1 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -289,7 +289,7 @@ pub async fn handle_python_job( .collect::>() .join(""); let client = client.get_authed().await; - create_args_and_out_file(&client, job, job_dir).await?; + create_args_and_out_file(&client, job, job_dir, db).await?; let import_loader = if relative_imports { "import loader" diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index 17266b2de0..8fec698f73 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -64,9 +64,10 @@ pub async fn do_snowflake( job: QueuedJob, client: &AuthedClient, query: &str, + db: &sqlx::Pool, ) -> windmill_common::error::Result { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), &job, db).await?) } else { None }; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 47c6176e75..8c3334e5ba 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1532,9 +1532,10 @@ async fn do_nativets( logs: String, client: &AuthedClient, code: String, + db: &Pool, ) -> windmill_common::error::Result<(serde_json::Value, String)> { let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + Some(transform_json_value("args", client, &job.workspace_id, args.clone(), &job, db).await?) } else { None }; @@ -1920,9 +1921,9 @@ async fn handle_code_execution_job( }; if language == Some(ScriptLang::Postgresql) { - return do_postgresql(job.clone(), &client.get_authed().await, &inner_content).await; + return do_postgresql(job.clone(), &client.get_authed().await, &inner_content, db).await; } else if language == Some(ScriptLang::Mysql) { - return do_mysql(job.clone(), &client.get_authed().await, &inner_content).await; + return do_mysql(job.clone(), &client.get_authed().await, &inner_content, db).await; } else if language == Some(ScriptLang::Bigquery) { #[cfg(not(feature = "enterprise"))] { @@ -1933,7 +1934,7 @@ async fn handle_code_execution_job( #[cfg(feature = "enterprise")] { - return do_bigquery(job.clone(), &client.get_authed().await, &inner_content).await; + return do_bigquery(job.clone(), &client.get_authed().await, &inner_content, db).await; } } else if language == Some(ScriptLang::Snowflake) { #[cfg(not(feature = "enterprise"))] @@ -1945,10 +1946,10 @@ async fn handle_code_execution_job( #[cfg(feature = "enterprise")] { - return do_snowflake(job.clone(), &client.get_authed().await, &inner_content).await; + return do_snowflake(job.clone(), &client.get_authed().await, &inner_content, db).await; } } else if language == Some(ScriptLang::Graphql) { - return do_graphql(job.clone(), &client.get_authed().await, &inner_content).await; + return do_graphql(job.clone(), &client.get_authed().await, &inner_content, db).await; } else if language == Some(ScriptLang::Nativets) { logs.push_str("\n--- FETCH TS EXECUTION ---\n"); let code = format!( @@ -1956,8 +1957,14 @@ async fn handle_code_execution_job( &client.get_token().await, inner_content ); - let (result, ts_logs) = - do_nativets(job.clone(), logs.clone(), &client.get_authed().await, code).await?; + let (result, ts_logs) = do_nativets( + job.clone(), + logs.clone(), + &client.get_authed().await, + code, + db, + ) + .await?; *logs = ts_logs; return Ok(result); }