From acc7bda015293e8dfceb1715b34d662f85fd84c7 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 10 Dec 2025 16:35:26 +0100 Subject: [PATCH] fix(backend): add presigned url support for object storage (#7328) * presigned * all * all * all * all * all * all * all * nit * nit * ee-ref * presigned * presigned --- backend/ee-repo-ref.txt | 2 +- backend/windmill-api/src/apps.rs | 405 ++++++++---------- backend/windmill-api/src/args.rs | 2 +- backend/windmill-api/src/db.rs | 13 +- backend/windmill-api/src/job_helpers_oss.rs | 15 +- backend/windmill-api/src/jobs.rs | 4 +- backend/windmill-api/src/resources.rs | 131 +++--- .../windmill-api/src/triggers/http/handler.rs | 1 - backend/windmill-api/src/users.rs | 10 +- backend/windmill-api/src/variables.rs | 42 +- backend/windmill-audit/src/audit_oss.rs | 37 +- backend/windmill-common/src/audit.rs | 20 + backend/windmill-common/src/db.rs | 51 ++- backend/windmill-common/src/lib.rs | 1 + backend/windmill-common/src/s3_helpers.rs | 2 + backend/windmill-queue/src/jobs.rs | 4 +- .../windmill-worker/src/duckdb_executor.rs | 7 +- backend/windmill-worker/src/worker_flow.rs | 4 +- frontend/package-lock.json | 15 - 19 files changed, 370 insertions(+), 396 deletions(-) create mode 100644 backend/windmill-common/src/audit.rs diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 1b5f682b77..62a5c1e615 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -a61644247e621f863a2a72626f47ee1c13267c49 +8f6e0bd9f273ac9da5f4fb3a44979f17b8e1b9b5 \ No newline at end of file diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 70f0ea5788..77e7763b01 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -55,7 +55,7 @@ use windmill_common::{ apps::{AppScriptId, ListAppQuery, APP_WORKSPACED_ROUTE}, auth::TOKEN_PREFIX_LEN, cache::{self, future::FutureCachedExt}, - db::UserDB, + db::{DbWithOptAuthed, UserDB}, error::{to_anyhow, Error, JsonResult, Result}, jobs::{get_payload_tag_from_prefixed_path, JobPayload, RawCode}, users::username_to_permissioned_as, @@ -1777,7 +1777,11 @@ async fn execute_component( .. } => ( &Policy { execution_mode: ExecutionMode::Viewer, ..Default::default() }, - &PolicyTriggerableInputs { static_inputs, one_of_inputs: force_viewer_one_of_fields.unwrap_or_default(), allow_user_resources: force_viewer_allow_user_resources.unwrap_or_default() }, + &PolicyTriggerableInputs { + static_inputs, + one_of_inputs: force_viewer_one_of_fields.unwrap_or_default(), + allow_user_resources: force_viewer_allow_user_resources.unwrap_or_default(), + }, ), // 2. "run" mode. _ => { @@ -2059,44 +2063,6 @@ async fn sign_s3_objects( Ok(Json(signed_s3_objects)) } -#[cfg(feature = "parquet")] -async fn validate_s3_signature(file_query: &AppS3FileQuery, w_id: &str, db: &DB) -> Result<()> { - let workspace_key = get_workspace_key(w_id, &db).await?; - - let Some(exp) = file_query - .exp - .as_ref() - .map(|e| e.parse::().unwrap_or_default()) - else { - return Err(Error::BadRequest("Missing exp".to_string())); - }; - - let Some(ref sig) = file_query.sig else { - return Err(Error::BadRequest("Missing signature".to_string())); - }; - - let mut message = format!("file_key={}&exp={}", file_query.s3, exp); - - if let Some(ref storage) = file_query.storage { - message = format!("{}&storage={}", message, storage); - } - - let mut mac = HmacSha256::new_from_slice(workspace_key.as_bytes()) - .map_err(|err| Error::internal_err(format!("Failed to create hmac: {}", err)))?; - - mac.update(message.as_bytes()); - - let sig_bytes = hex::decode(sig)?; - mac.verify_slice(&sig_bytes) - .map_err(|err| Error::BadRequest(format!("Invalid signature: {}", err)))?; - - if exp < chrono::Utc::now().timestamp() { - return Err(Error::BadRequest("Signature expired".to_string())); - } - - Ok(()) -} - #[cfg(not(feature = "parquet"))] async fn sign_s3_objects() -> Result<()> { return Err(Error::BadRequest( @@ -2155,89 +2121,66 @@ async fn upload_s3_file_from_app( let user_db = UserDB::new(db.clone()); - let (s3_resource_opt, file_key, on_behalf_of_email, permissioned_as, username) = - if policy.as_ref().is_some_and(|p| p.s3_inputs.is_some()) { - let policy = policy.unwrap(); - let s3_inputs = policy.s3_inputs.as_ref().unwrap(); + let (s3_resource_opt, file_key, on_behalf_of_email, permissioned_as, username) = if policy + .as_ref() + .is_some_and(|p| p.s3_inputs.is_some()) + { + let policy = policy.unwrap(); + let s3_inputs = policy.s3_inputs.as_ref().unwrap(); - let (username, permissioned_as, email) = - get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?; + let (username, permissioned_as, email) = + get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?; - let on_behalf_authed = fetch_api_authed_from_permissioned_as( - permissioned_as.clone(), - email.clone(), - &w_id, - &db, - Some(username.clone()), - ) - .await?; + let on_behalf_authed = fetch_api_authed_from_permissioned_as( + permissioned_as.clone(), + email.clone(), + &w_id, + &db, + Some(username.clone()), + ) + .await?; - if let Some(file_key) = query.file_key { - // file key is provided => requires workspace, user or list policy and must match the regex - let matching_s3_inputs = if let Some(ref s3_resource_path) = query.s3_resource_path - { - s3_inputs - .iter() - .filter(|s3_input| { - s3_input.allowed_resources.contains(s3_resource_path) - || s3_input.allow_user_resources - }) - .sorted_by_key(|i| i.allow_user_resources) // consider user resources last - .collect::>() - } else { - s3_inputs - .iter() - .filter(|s3_input| s3_input.allow_workspace_resource) - .collect::>() - }; + if let Some(file_key) = query.file_key { + // file key is provided => requires workspace, user or list policy and must match the regex + let matching_s3_inputs = if let Some(ref s3_resource_path) = query.s3_resource_path { + s3_inputs + .iter() + .filter(|s3_input| { + s3_input.allowed_resources.contains(s3_resource_path) + || s3_input.allow_user_resources + }) + .sorted_by_key(|i| i.allow_user_resources) // consider user resources last + .collect::>() + } else { + s3_inputs + .iter() + .filter(|s3_input| s3_input.allow_workspace_resource) + .collect::>() + }; - let matched_input = matching_s3_inputs.iter().find(|s3_input| { - match Regex::new(&s3_input.file_key_regex) { - Ok(re) => re.is_match(&file_key), - Err(e) => { - tracing::error!("Error compiling regex: {}", e); - false - } + let matched_input = matching_s3_inputs.iter().find(|s3_input| { + match Regex::new(&s3_input.file_key_regex) { + Ok(re) => re.is_match(&file_key), + Err(e) => { + tracing::error!("Error compiling regex: {}", e); + false } - }); + } + }); - if let Some(matched_input) = matched_input { - if let Some(ref s3_resource_path) = query.s3_resource_path { - if matched_input.allow_user_resources { - if let Some(authed) = opt_authed { - ( - Some( - get_s3_resource( - &authed, - &db, - Some(user_db), - "", - &w_id, - s3_resource_path, - None, - None, - ) - .await?, - ), - file_key, - email, - permissioned_as, - username, - ) - } else { - return Err(Error::BadRequest( - "User resources are not allowed without being logged in" - .to_string(), - )); - } - } else { + if let Some(matched_input) = matched_input { + if let Some(ref s3_resource_path) = query.s3_resource_path { + if matched_input.allow_user_resources { + if let Some(authed) = opt_authed { + let db_with_opt_authed = DbWithOptAuthed::from_authed( + &authed, + db.clone(), + Some(user_db.clone()), + ); ( Some( get_s3_resource( - &on_behalf_authed, - &db, - Some(user_db), - "", + &db_with_opt_authed, &w_id, s3_resource_path, None, @@ -2250,112 +2193,131 @@ async fn upload_s3_file_from_app( permissioned_as, username, ) + } else { + return Err(Error::BadRequest( + "User resources are not allowed without being logged in" + .to_string(), + )); } } else { - let (_, s3_resource_opt) = get_workspace_s3_resource_and_check_paths( + let db_with_opt_authed = DbWithOptAuthed::from_authed( &on_behalf_authed, - &db, - None, - "", - &w_id, - None, - &[(&file_key, S3Permission::WRITE)], + db.clone(), + Some(user_db.clone()), + ); + ( + Some( + get_s3_resource( + &db_with_opt_authed, + &w_id, + s3_resource_path, + None, + None, + ) + .await?, + ), + file_key, + email, + permissioned_as, + username, ) - .await?; - (s3_resource_opt, file_key, email, permissioned_as, username) } } else { - return Err(Error::BadRequest( - "No matching s3 resource found for the given file key".to_string(), - )); + let db_with_opt_authed = DbWithOptAuthed::from_authed( + &on_behalf_authed, + db.clone(), + Some(user_db.clone()), + ); + let (_, s3_resource_opt) = get_workspace_s3_resource_and_check_paths( + &db_with_opt_authed, + &w_id, + None, + &[(&file_key, S3Permission::WRITE)], + ) + .await?; + (s3_resource_opt, file_key, email, permissioned_as, username) } } else { - // no file key => requires unnamed upload policy => allow workspace resource and file_key_regex is empty - let has_unnamed_policy = s3_inputs.iter().any(|s3_input| { - s3_input.allow_workspace_resource && s3_input.file_key_regex.is_empty() - }); + return Err(Error::BadRequest( + "No matching s3 resource found for the given file key".to_string(), + )); + } + } else { + // no file key => requires unnamed upload policy => allow workspace resource and file_key_regex is empty + let has_unnamed_policy = s3_inputs.iter().any(|s3_input| { + s3_input.allow_workspace_resource && s3_input.file_key_regex.is_empty() + }); - if !has_unnamed_policy { - return Err(Error::BadRequest( - "no policy found for unnamed s3 file upload".to_string(), - )); - } + if !has_unnamed_policy { + return Err(Error::BadRequest( + "no policy found for unnamed s3 file upload".to_string(), + )); + } - // for now, we place all files into `windmill_uploads` folder with a random name - // TODO: make the folder configurable via the workspace settings - let file_key = get_random_file_name(query.file_extension); + // for now, we place all files into `windmill_uploads` folder with a random name + // TODO: make the folder configurable via the workspace settings + let file_key = get_random_file_name(query.file_extension); + let db_with_opt_authed = + DbWithOptAuthed::from_authed(&on_behalf_authed, db.clone(), None); + let (_, s3_resource_opt) = get_workspace_s3_resource_and_check_paths( + &db_with_opt_authed, + &w_id, + None, + &[(&file_key, S3Permission::WRITE)], + ) + .await?; - let (_, s3_resource_opt) = get_workspace_s3_resource_and_check_paths( - &on_behalf_authed, - &db, - None, - "", + (s3_resource_opt, file_key, email, permissioned_as, username) + } + } else { + // backward compatibility (no policy) + // if no policy but logged in, use the user's auth to get the s3 resource + if let Some(authed) = opt_authed { + let file_key = query + .file_key + .unwrap_or_else(|| get_random_file_name(query.file_extension)); + + let (on_behalf_of_email, permissioned_as, username) = ( + authed.email.clone(), + username_to_permissioned_as(&authed.username), + authed.display_username().to_string(), + ); + + if let Some(ref s3_resource_path) = query.s3_resource_path { + let db_with_opt_authed = + DbWithOptAuthed::from_authed(&authed, db.clone(), Some(user_db.clone())); + ( + Some( + get_s3_resource(&db_with_opt_authed, &w_id, s3_resource_path, None, None) + .await?, + ), + file_key, + on_behalf_of_email, + permissioned_as, + username, + ) + } else { + let db_with_opt_authed = DbWithOptAuthed::from_authed(&authed, db.clone(), None); + let (_, s3_resource) = get_workspace_s3_resource_and_check_paths( + &db_with_opt_authed, &w_id, None, &[(&file_key, S3Permission::WRITE)], ) .await?; - (s3_resource_opt, file_key, email, permissioned_as, username) + ( + s3_resource, + file_key, + on_behalf_of_email, + permissioned_as, + username, + ) } } else { - // backward compatibility (no policy) - // if no policy but logged in, use the user's auth to get the s3 resource - if let Some(authed) = opt_authed { - let file_key = query - .file_key - .unwrap_or_else(|| get_random_file_name(query.file_extension)); - - let (on_behalf_of_email, permissioned_as, username) = ( - authed.email.clone(), - username_to_permissioned_as(&authed.username), - authed.display_username().to_string(), - ); - - if let Some(ref s3_resource_path) = query.s3_resource_path { - ( - Some( - get_s3_resource( - &authed, - &db, - Some(user_db), - "", - &w_id, - s3_resource_path, - None, - None, - ) - .await?, - ), - file_key, - on_behalf_of_email, - permissioned_as, - username, - ) - } else { - let (_, s3_resource) = get_workspace_s3_resource_and_check_paths( - &authed, - &db, - None, - "", - &w_id, - None, - &[(&file_key, S3Permission::WRITE)], - ) - .await?; - - ( - s3_resource, - file_key, - on_behalf_of_email, - permissioned_as, - username, - ) - } - } else { - return Err(Error::BadRequest("Missing s3 policy".to_string())); - } - }; + return Err(Error::BadRequest("Missing s3 policy".to_string())); + } + }; let s3_resource = s3_resource_opt.ok_or(Error::internal_err( "No files storage resource defined at the workspace level".to_string(), @@ -2434,11 +2396,10 @@ async fn delete_s3_file_from_app( .await?; let s3_resource = if let Some(s3_resource_path) = s3_resource_path { + let db_with_opt_authed = + DbWithOptAuthed::from_authed(&on_behalf_authed, db.clone(), Some(user_db.clone())); get_s3_resource( - &on_behalf_authed, - &db, - Some(user_db), - "", + &db_with_opt_authed, &w_id, s3_resource_path.as_str(), None, @@ -2446,11 +2407,9 @@ async fn delete_s3_file_from_app( ) .await? } else { + let db_with_opt_authed = DbWithOptAuthed::from_authed(&on_behalf_authed, db.clone(), None); let (_, s3_resource) = get_workspace_s3_resource_and_check_paths( - &on_behalf_authed, - &db, - None, - "", + &db_with_opt_authed, &w_id, None, &[(&path.to_string(), S3Permission::DELETE)], @@ -2544,7 +2503,23 @@ async fn check_if_allowed_to_access_s3_file_from_app( // otherwise, if logged in, allow any file (TODO: change that when we implement better s3 policy) if file_query.sig.is_some() { - validate_s3_signature(file_query, w_id, &db).await + #[cfg(feature = "private")] + { + crate::s3_proxy_ee::validate_s3_signature( + &file_query.s3, + &file_query.sig, + &file_query.exp, + &file_query.storage, + w_id, + &db, + ) + .await?; + Ok(()) + } + #[cfg(not(feature = "private"))] + return Err(Error::InternalErr( + "Internal error: signature validation is not supported in open source mode".to_string(), + )); } else if opt_authed.is_some() { Ok(()) } else { @@ -2588,6 +2563,7 @@ struct AppS3FileQuery { s3: String, storage: Option, sig: Option, + #[cfg(feature = "private")] exp: Option, } @@ -2634,7 +2610,6 @@ async fn download_s3_file_from_app( on_behalf_authed, &db, None, - "", &w_id, DownloadFileQuery { file_key: query.file_query.s3, @@ -2721,14 +2696,14 @@ async fn build_args( key.and_then(|x| x.clone().strip_prefix("$res:").map(|x| x.to_string())) { if let Some(authed) = authed { + let db_with_opt_authed = + DbWithOptAuthed::from_authed(authed, db.clone(), Some(user_db.clone())); let res = get_resource_value_interpolated_internal( - authed, - Some(user_db.clone()), - db, + &db_with_opt_authed, w_id, &path, None, - "", + None, false, ) .await?; diff --git a/backend/windmill-api/src/args.rs b/backend/windmill-api/src/args.rs index 763074948f..bcbfba03d0 100644 --- a/backend/windmill-api/src/args.rs +++ b/backend/windmill-api/src/args.rs @@ -94,7 +94,7 @@ impl RawWebhookArgs { use object_store::{Attribute, Attributes}; use windmill_common::s3_helpers::build_object_store_client; - let (_, s3_resource) = get_workspace_s3_resource(authed, db, None, "", w_id, None).await?; + let (_, s3_resource) = get_workspace_s3_resource(authed, db, None, w_id, None).await?; if let Some(s3_resource) = s3_resource { let s3_client = build_object_store_client(&s3_resource).await?; diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 09da5d1b11..87dab1b188 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -14,7 +14,7 @@ use sqlx::{ }; use tokio::task::JoinHandle; -use windmill_audit::audit_oss::{AuditAuthor, AuditAuthorable}; +use windmill_audit::audit_oss::AuditAuthorable; pub use windmill_common::db::DB; use windmill_common::{ db::{Authable, Authed, AuthedRef}, @@ -309,17 +309,6 @@ impl From for ApiAuthed { } } -impl From<&ApiAuthed> for AuditAuthor { - fn from(value: &ApiAuthed) -> Self { - Self { - email: value.email.clone(), - username: value.username.clone(), - username_override: value.username_override.clone(), - token_prefix: value.token_prefix.clone(), - } - } -} - impl ApiAuthed { pub fn display_username(&self) -> &str { self.username_override.as_ref().unwrap_or(&self.username) diff --git a/backend/windmill-api/src/job_helpers_oss.rs b/backend/windmill-api/src/job_helpers_oss.rs index defb53db76..8a67ada9ad 100644 --- a/backend/windmill-api/src/job_helpers_oss.rs +++ b/backend/windmill-api/src/job_helpers_oss.rs @@ -51,7 +51,6 @@ pub async fn get_workspace_s3_resource<'c>( _authed: &ApiAuthed, _db: &DB, _user_db: Option, - _token: &str, _w_id: &str, _storage: Option, ) -> windmill_common::error::Result<(Option, Option)> { @@ -64,12 +63,12 @@ pub fn get_random_file_name(_file_extension: Option) -> String { unimplemented!("Not implemented in Windmill's Open Source repository") } +#[cfg(not(feature = "private"))] +use windmill_common::db::DbWithOptAuthed; + #[cfg(not(feature = "private"))] pub async fn get_s3_resource<'c>( - _authed: &ApiAuthed, - _db: &DB, - _user_db: Option, - _token: &str, + _db_with_opt_authed: &DbWithOptAuthed<'c, ApiAuthed>, _w_id: &str, _resource_path: &str, _resource_type: Option, @@ -109,7 +108,6 @@ pub async fn download_s3_file_internal( _authed: ApiAuthed, _db: &DB, _user_db: Option, - _token: &str, _w_id: &str, _query: DownloadFileQuery, ) -> error::Result { @@ -153,10 +151,7 @@ pub struct DeleteS3FileQuery { #[cfg(not(feature = "private"))] pub async fn get_workspace_s3_resource_and_check_paths<'c>( - _authed: &crate::db::ApiAuthed, - _db: &crate::db::DB, - _user_db: Option, - _token: &str, + _db_with_opt_authed: &DbWithOptAuthed<'c, ApiAuthed>, _w_id: &str, _storage: Option, _paths: &[(&str, windmill_common::s3_helpers::S3Permission)], diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 005c498a40..82a288abe3 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -81,9 +81,9 @@ use sqlx::types::JsonRawValue; use sqlx::{types::Uuid, FromRow, Postgres, Transaction}; use tower_http::cors::{Any, CorsLayer}; use urlencoding::encode; -use windmill_audit::audit_oss::{audit_log, AuditAuthor}; +use windmill_audit::audit_oss::audit_log; use windmill_audit::ActionKind; - +use windmill_common::audit::AuditAuthor; use windmill_common::worker::to_raw_value; use windmill_common::{ diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index 14c7054c3a..05f7127db2 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -28,18 +28,20 @@ use hyper::{header, StatusCode}; use serde::{Deserialize, Serialize}; use serde_json::{value::RawValue, Value}; use sql_builder::{bind::Bind, quote, SqlBuilder}; -use sqlx::{FromRow, Postgres, Transaction}; +use sqlx::{Acquire, FromRow, Postgres, Transaction}; use std::process::Stdio; use tokio::process::Command; use uuid::Uuid; -use windmill_audit::audit_oss::{audit_log, AuditAuthor}; +use windmill_audit::audit_oss::{audit_log, AuditAuthorable}; use windmill_audit::ActionKind; use windmill_common::{ - db::{UserDB, UserDbWithAuthed, UserDbWithOptAuthed}, + db::{DbWithOptAuthed, UserDB}, error::{self, Error, JsonResult, Result}, get_database_url, parse_postgres_url, - utils::get_custom_pg_instance_password, - utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath}, + utils::{ + get_custom_pg_instance_password, not_found_if_none, paginate, require_admin, Pagination, + StripPath, + }, variables, worker::{CLOUD_HOSTED, TMP_DIR}, }; @@ -442,14 +444,14 @@ async fn get_resource_value_interpolated( let path = path.to_path(); check_scopes(&authed, || format!("resources:read:{}", path))?; + let db_with_opt_authed = + DbWithOptAuthed::from_authed(&authed, db.clone(), Some(user_db.clone())); return get_resource_value_interpolated_internal( - &authed, - Some(user_db), - &db, + &db_with_opt_authed, w_id.as_str(), path, job_info.job_id, - token.as_str(), + Some(token.as_str()), job_info.allow_cache.unwrap_or(false), ) .await @@ -459,19 +461,18 @@ async fn get_resource_value_interpolated( use async_recursion::async_recursion; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; -pub async fn get_resource_value_interpolated_internal( - authed: &ApiAuthed, - user_db: Option, // if none, no permission will be checked to access the resource - db: &DB, +pub async fn get_resource_value_interpolated_internal<'a>( + db_with_opt_authed: &'a DbWithOptAuthed<'a, ApiAuthed>, workspace: &str, path: &str, job_id: Option, - token: &str, + token_for_context: Option<&str>, allow_cache: bool, ) -> Result> { // This is a special syntax to help debugging custom instance databases if let Some(dbname) = path.strip_prefix("CUSTOM_INSTANCE_DB/") { - require_super_admin(db, &authed.email).await?; + let db = db_with_opt_authed.db(); + require_super_admin(db_with_opt_authed.db(), &db_with_opt_authed.email()).await?; let pg_creds = parse_postgres_url(&get_database_url().await?.as_str().await)?; return Ok(Some(serde_json::json!({ "dbname": dbname, @@ -488,7 +489,8 @@ pub async fn get_resource_value_interpolated_internal( return Ok(Some(cached_value)); } } - let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?; + use sqlx::Acquire; + let mut tx = db_with_opt_authed.begin().await?; let value_o = sqlx::query_scalar!( "SELECT value from resource WHERE path = $1 AND workspace_id = $2", @@ -499,19 +501,20 @@ pub async fn get_resource_value_interpolated_internal( .await?; tx.commit().await?; if value_o.is_none() { - explain_resource_perm_error(path, workspace, db, &authed).await?; + if let Some(authed) = db_with_opt_authed.authed() { + let db = db_with_opt_authed.db(); + explain_resource_perm_error(path, workspace, db, authed).await?; + } } let value = not_found_if_none(value_o, "Resource", path)?; if let Some(value) = value { let r = transform_json_value( - authed, - user_db.clone(), - db, + &db_with_opt_authed, workspace, value, &job_id, - token, + token_for_context, ) .await?; if allow_cache { @@ -524,38 +527,20 @@ pub async fn get_resource_value_interpolated_internal( } #[async_recursion] -pub async fn transform_json_value<'c>( - authed: &ApiAuthed, - user_db: Option, // if none, no permission will be checked to access the resources/variables - db: &DB, +pub async fn transform_json_value( + db_with_opt_authed: &DbWithOptAuthed, workspace: &str, v: Value, job_id: &Option, - token: &str, + token: Option<&str>, ) -> Result { 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 v = crate::variables::get_value_internal( - &userdb_authed, - db, - workspace, - path, - &user_db - .clone() - .map(|_| authed.into()) - .unwrap_or(AuditAuthor { - email: "backend".to_string(), - username: "backend".to_string(), - username_override: None, - token_prefix: None, - }), - false, - ) - .await?; + let v = + crate::variables::get_value_internal(&db_with_opt_authed, workspace, path, false) + .await?; Ok(Value::String(v)) } Value::String(y) if y.starts_with("$res:") => { @@ -565,8 +550,7 @@ pub async fn transform_json_value<'c>( "Invalid resource path: {path}" ))); } - let mut tx: Transaction<'_, Postgres> = - authed_transaction_or_default(authed, user_db.clone(), db).await?; + let mut tx: Transaction<'_, Postgres> = db_with_opt_authed.begin().await?; let v = sqlx::query_scalar!( "SELECT value from resource WHERE path = $1 AND workspace_id = $2", path, @@ -577,13 +561,13 @@ 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.clone(), db, workspace, v, job_id, token).await + transform_json_value(db_with_opt_authed, workspace, v, job_id, token).await } else { Ok(Value::Null) } } Value::String(y) if y.starts_with("$") && job_id.is_some() => { - let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?; + let mut tx = db_with_opt_authed.begin().await?; let job_id = job_id.unwrap(); let job = sqlx::query!( "SELECT @@ -609,8 +593,7 @@ pub async fn transform_json_value<'c>( let job = not_found_if_none(job, "Job", job_id.to_string())?; let flow_path = if let Some(uuid) = job.parent_job { - let mut tx: Transaction<'_, Postgres> = - authed_transaction_or_default(authed, user_db.clone(), db).await?; + let mut tx: Transaction<'_, Postgres> = db_with_opt_authed.begin().await?; let p = sqlx::query_scalar!("SELECT runnable_path FROM v2_job WHERE id = $1", uuid) .fetch_optional(&mut *tx) .await? @@ -622,9 +605,9 @@ pub async fn transform_json_value<'c>( }; let variables = variables::get_reserved_variables( - &db.into(), + &db_with_opt_authed.db().into(), workspace, - token, + token.unwrap_or_else(|| "no_token_available"), &job.permissioned_as_email, &job.created_by, &job_id.to_string(), @@ -654,8 +637,7 @@ pub async fn transform_json_value<'c>( Value::Object(mut m) => { for (a, b) in m.clone().into_iter() { let v = - transform_json_value(authed, user_db.clone(), db, workspace, b, job_id, token) - .await?; + transform_json_value(db_with_opt_authed, workspace, b, job_id, token).await?; m.insert(a.clone(), v); } Ok(Value::Object(m)) @@ -664,17 +646,15 @@ pub async fn transform_json_value<'c>( } } -async fn authed_transaction_or_default<'c>( - authed: &ApiAuthed, - user_db: Option, - db: &DB, -) -> sqlx::error::Result> { - if let Some(user_db) = user_db { - user_db.begin(authed).await - } else { - db.clone().begin().await - } -} +// async fn authed_transaction_or_default<'c>( +// db_with_opt_authed: &'c DbWithOptAuthed, +// ) -> sqlx::error::Result> { +// if let Some(user_db) = user_db { +// user_db.begin(authed).await +// } else { +// db.clone().begin().await +// } +// } async fn check_path_conflict<'c>( tx: &mut Transaction<'c, Postgres>, @@ -1395,13 +1375,11 @@ where T: serde::de::DeserializeOwned, { let resource = get_resource_value_interpolated_internal( - &authed, - user_db, - &db, + &DbWithOptAuthed::from_authed(authed, db.clone(), user_db), &w_id, &resource_path, None, - "", + None, false, ) .await?; @@ -1501,7 +1479,6 @@ async fn get_git_commit_hash( authed: ApiAuthed, Extension(user_db): Extension, Extension(db): Extension, - Tokened { token }: Tokened, Path((w_id, path)): Path<(String, StripPath)>, Query(query): Query, ) -> JsonResult { @@ -1509,14 +1486,14 @@ async fn get_git_commit_hash( check_scopes(&authed, || format!("resources:read:{}", path))?; + let db_with_opt_authed = + DbWithOptAuthed::from_authed(&authed, db.clone(), Some(user_db.clone())); let git_repo_resource_value = get_resource_value_interpolated_internal( - &authed, - Some(user_db.clone()), - &db, + &db_with_opt_authed, &w_id, path, None, - &token, + None, false, ) .await @@ -1566,8 +1543,8 @@ async fn write_ssh_file( .join("ssh_ids") .join(id_file_name); - let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; - let mut content = get_value_internal(&userdb_authed, db, w_id, var_path, authed, false) + let userdb_authed = DbWithOptAuthed::from_authed(authed, db.clone(), Some(user_db.clone())); + let mut content = get_value_internal(&userdb_authed, &w_id, &var_path, false) .await .map_err(|e| { ( diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index a9a8b694fe..4ecc94d15a 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -924,7 +924,6 @@ async fn route_job( &authed, &db, None, - &"NO_TOKEN".to_string(), // no token is provided in this case &trigger.workspace_id, config.storage, ) diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 7ed3fe13f9..3e03fbd88f 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -43,8 +43,9 @@ use sqlx::FromRow; use time::OffsetDateTime; use tower_cookies::{Cookie, Cookies}; use tracing::Instrument; -use windmill_audit::audit_oss::{audit_log, AuditAuthor}; +use windmill_audit::audit_oss::audit_log; use windmill_audit::ActionKind; +use windmill_common::audit::AuditAuthor; use windmill_common::auth::{fetch_authed_from_permissioned_as, TOKEN_PREFIX_LEN}; use windmill_common::global_settings::AUTOMATE_USERNAME_CREATION_SETTING; use windmill_common::oauth2::InstanceEvent; @@ -574,17 +575,14 @@ async fn get_tutorial_progress( ) .fetch_optional(&db) .await?; - + if let Some(row) = row { Ok(Json(Progress { progress: row.progress.unwrap_or_default() as u64, skipped_all: row.skipped_all, })) } else { - Ok(Json(Progress { - progress: 0, - skipped_all: false, - })) + Ok(Json(Progress { progress: 0, skipped_all: false })) } } diff --git a/backend/windmill-api/src/variables.rs b/backend/windmill-api/src/variables.rs index 07d0572e6f..a5493cd2e5 100644 --- a/backend/windmill-api/src/variables.rs +++ b/backend/windmill-api/src/variables.rs @@ -25,7 +25,7 @@ use serde_json::Value; use windmill_audit::audit_oss::{audit_log, AuditAuthorable}; use windmill_audit::ActionKind; use windmill_common::{ - db::{UserDB, UserDbWithAuthed}, + db::{DbWithOptAuthed, UserDB}, error::{Error, JsonResult, Result}, scripts::ScriptHash, utils::{not_found_if_none, paginate, Pagination, StripPath, WarnAfterExt}, @@ -38,7 +38,7 @@ use windmill_common::{ use crate::var_resource_cache::{cache_variable, get_cached_variable}; use lazy_static::lazy_static; use serde::Deserialize; -use sqlx::{Postgres, Transaction}; +use sqlx::{Acquire, Postgres, Transaction}; use windmill_common::variables::{decrypt, encrypt}; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; @@ -244,19 +244,12 @@ async fn get_value( ) -> JsonResult { let path = path.to_path(); check_scopes(&authed, || format!("variables:read:{}", path))?; - let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() }; + let userdb_authed = DbWithOptAuthed::from_authed(&authed, db.clone(), Some(user_db.clone())); - return get_value_internal( - &userdb_authed, - &db, - &w_id, - &path, - &authed, - q.allow_cache.unwrap_or(false), - ) - .warn_after_seconds(10) - .await - .map(Json); + return get_value_internal(&userdb_authed, &w_id, &path, q.allow_cache.unwrap_or(false)) + .warn_after_seconds(10) + .await + .map(Json); } async fn explain_variable_perm_error( @@ -780,12 +773,10 @@ fn replace_path(v: serde_json::Value, path: &str, npath: &str) -> Value { } } -pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a>( - acquire_db: A, - db: &DB, +pub async fn get_value_internal<'a>( + db_with_opt_authed: &'a DbWithOptAuthed<'a, ApiAuthed>, w_id: &str, path: &str, - audit_author: &impl AuditAuthorable, allow_cache: bool, ) -> Result { if allow_cache { @@ -794,7 +785,7 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres } } - let mut tx = acquire_db.begin().await?; + let mut tx = db_with_opt_authed.begin().await?; let variable_o = sqlx::query!( "SELECT value, account, (now() > account.expires_at) as is_expired, is_secret, path from variable LEFT JOIN account ON variable.account = account.id WHERE variable.path = $1 AND variable.workspace_id = $2", path, w_id @@ -807,15 +798,16 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres let variable = if let Some(variable) = variable_o { variable } else { - explain_variable_perm_error(path, w_id, db).await?; + explain_variable_perm_error(path, w_id, &db_with_opt_authed.db()).await?; unreachable!() }; let r = if variable.is_secret { - let mut tx = db.begin().await?; + // let audit_author = + let mut tx = db_with_opt_authed.db().begin().await?; audit_log( &mut *tx, - audit_author, + db_with_opt_authed, "variables.decrypt_secret", ActionKind::Execute, &w_id, @@ -829,6 +821,7 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres if variable.is_expired.unwrap_or(false) && variable.account.is_some() { #[cfg(feature = "oauth2")] { + let db = db_with_opt_authed.db(); let tx = db.begin().await?; crate::oauth2_oss::_refresh_token( tx, @@ -842,7 +835,7 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres #[cfg(not(feature = "oauth2"))] return Err(Error::internal_err("Require oauth2 feature".to_string())); } else if !value.is_empty() { - let mc = build_crypt(&db, &w_id).await?; + let mc = build_crypt(db_with_opt_authed.db(), &w_id).await?; decrypt(&mc, value).map_err(|e| { Error::internal_err(format!( "Error decrypting variable {}: {}", @@ -858,9 +851,8 @@ pub async fn get_value_internal<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres // Cache the result when explicitly allowed and caching appropriate if allow_cache { - cache_variable(&w_id, &path, audit_author.email(), r.clone()); + cache_variable(&w_id, &path, db_with_opt_authed.email(), r.clone()); } Ok(r) } - diff --git a/backend/windmill-audit/src/audit_oss.rs b/backend/windmill-audit/src/audit_oss.rs index f5979f45d7..da36ada468 100644 --- a/backend/windmill-audit/src/audit_oss.rs +++ b/backend/windmill-audit/src/audit_oss.rs @@ -1,3 +1,8 @@ +use windmill_common::{ + audit::AuditAuthor, + db::{Authable, DbWithOptAuthed}, +}; + #[cfg(feature = "private")] #[allow(unused)] pub use crate::audit_ee::*; @@ -20,7 +25,27 @@ use { }, }; -#[cfg(not(feature = "private"))] +impl<'a, T: Authable + AuditAuthorable + Sync> AuditAuthorable for DbWithOptAuthed<'a, T> { + fn email(&self) -> &str { + match self { + DbWithOptAuthed::UserDB { authed, .. } => AuditAuthorable::email(*authed), + DbWithOptAuthed::DB { audit_author, .. } => audit_author.email(), + } + } + fn username(&self) -> &str { + match self { + DbWithOptAuthed::UserDB { authed, .. } => AuditAuthorable::username(*authed), + DbWithOptAuthed::DB { audit_author, .. } => audit_author.username(), + } + } + fn username_override(&self) -> Option<&str> { + match self { + DbWithOptAuthed::UserDB { authed, .. } => AuditAuthorable::username_override(*authed), + DbWithOptAuthed::DB { .. } => None, + } + } +} + impl AuditAuthorable for AuditAuthor { fn email(&self) -> &str { &self.email @@ -39,7 +64,6 @@ impl AuditAuthorable for AuditAuthor { } } -#[cfg(not(feature = "private"))] pub trait AuditAuthorable { fn username(&self) -> &str; fn email(&self) -> &str; @@ -49,15 +73,6 @@ pub trait AuditAuthorable { } } -#[derive(Clone)] -#[cfg(not(feature = "private"))] -pub struct AuditAuthor { - pub username: String, - pub email: String, - pub username_override: Option, - pub token_prefix: Option, -} - #[cfg(not(feature = "private"))] #[tracing::instrument(level = "trace", skip_all)] pub async fn audit_log<'c, E: sqlx::Executor<'c, Database = Postgres>>( diff --git a/backend/windmill-common/src/audit.rs b/backend/windmill-common/src/audit.rs new file mode 100644 index 0000000000..41fba29fce --- /dev/null +++ b/backend/windmill-common/src/audit.rs @@ -0,0 +1,20 @@ +use crate::db::Authable; + +#[derive(Clone)] +pub struct AuditAuthor { + pub username: String, + pub email: String, + pub username_override: Option, + pub token_prefix: Option, +} + +impl From<&T> for AuditAuthor { + fn from(value: &T) -> Self { + Self { + username: value.username().to_string(), + email: value.email().to_string(), + username_override: None, + token_prefix: None, + } + } +} diff --git a/backend/windmill-common/src/db.rs b/backend/windmill-common/src/db.rs index 1aa5e57ea0..49f3a87b12 100644 --- a/backend/windmill-common/src/db.rs +++ b/backend/windmill-common/src/db.rs @@ -1,5 +1,7 @@ use sqlx::{Acquire, Pool, Postgres, Transaction}; +use crate::audit::AuditAuthor; + pub type DB = Pool; #[derive(Clone, Debug, Hash, Eq, PartialEq)] @@ -121,22 +123,44 @@ impl<'c, 'd, T: Authable + Sync> Acquire<'c> for &'c UserDbWithAuthed<'d, T> { } } -pub struct UserDbWithOptAuthed<'c, T: Authable + Sync> { - pub authed: &'c T, - pub user_db: Option, - pub db: DB, +pub enum DbWithOptAuthed<'a, T: Authable + Sync> { + UserDB { authed: &'a T, user_db: UserDB, db: DB }, + DB { db: DB, audit_author: AuditAuthor }, +} +impl<'a, T: Authable + Sync> DbWithOptAuthed<'a, T> { + pub fn from_authed(authed: &'a T, db: DB, user_db: Option) -> Self { + if let Some(user_db) = user_db { + Self::UserDB { authed, user_db, db } + } else { + Self::DB { db, audit_author: AuditAuthor::from(authed) } + } + } + pub fn db(&self) -> &DB { + match self { + DbWithOptAuthed::UserDB { db, .. } => db, + DbWithOptAuthed::DB { db, .. } => db, + } + } + + pub fn authed(&self) -> Option<&T> { + match self { + DbWithOptAuthed::UserDB { authed, .. } => Some(authed), + DbWithOptAuthed::DB { .. } => None, + } + } } -impl<'c, 'd, T: Authable + Sync> Acquire<'c> for &'c UserDbWithOptAuthed<'d, T> { +impl<'c, 'd, T: Authable + Sync> Acquire<'c> for &'c DbWithOptAuthed<'d, T> { type Database = Postgres; type Connection = Transaction<'c, Postgres>; fn acquire(self) -> futures_core::future::BoxFuture<'c, Result> { Box::pin(async move { - if let Some(db) = &self.user_db { - db.clone().begin(self.authed).await - } else { - self.db.clone().begin().await + match self { + DbWithOptAuthed::UserDB { authed, user_db, .. } => { + user_db.clone().begin(&**authed).await + } + DbWithOptAuthed::DB { db, .. } => db.clone().begin().await, } }) } @@ -145,10 +169,11 @@ impl<'c, 'd, T: Authable + Sync> Acquire<'c> for &'c UserDbWithOptAuthed<'d, T> self, ) -> futures_core::future::BoxFuture<'c, Result, sqlx::Error>> { Box::pin(async move { - if let Some(db) = &self.user_db { - db.clone().begin(self.authed).await - } else { - self.db.clone().begin().await + match self { + DbWithOptAuthed::UserDB { authed, user_db, .. } => { + user_db.clone().begin(&**authed).await + } + DbWithOptAuthed::DB { db, .. } => db.clone().begin().await, } }) } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 73390ef6f2..c79799390a 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -28,6 +28,7 @@ use sqlx::{Acquire, Postgres}; pub mod agent_workers; pub mod ai_providers; pub mod apps; +pub mod audit; pub mod assets; pub mod auth; #[cfg(feature = "benchmark")] diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index 69fbaa4b29..da283573b0 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -546,6 +546,8 @@ impl BundleFormat { } } +pub const DEFAULT_STORAGE: &str = "_default_"; + pub async fn upload_artifact_to_store( path: &str, data: bytes::Bytes, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 326ee794a5..212045607d 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -30,11 +30,11 @@ use tokio::task::JoinHandle; use tokio::{sync::RwLock, time::sleep}; use ulid::Ulid; use uuid::Uuid; -use windmill_audit::audit_oss::{audit_log, AuditAuthor}; +use windmill_audit::audit_oss::audit_log; use windmill_audit::ActionKind; - #[cfg(feature = "benchmark")] use windmill_common::add_time; +use windmill_common::audit::AuditAuthor; use windmill_common::auth::JobPerms; #[cfg(feature = "benchmark")] use windmill_common::bench::BenchmarkIter; diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 8cd0380df6..44d6fdbc65 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -28,6 +28,7 @@ use crate::mysql_executor::MysqlDatabase; use crate::pg_executor::PgDatabase; use crate::sanitized_sql_params::sanitize_and_interpolate_unsafe_sql_args; use windmill_common::client::AuthedClient; +use windmill_common::s3_helpers::DEFAULT_STORAGE; pub async fn do_duckdb( job: &MiniPulledJob, @@ -88,7 +89,7 @@ pub async fn do_duckdb( })?; let uri = format!( "s3://{}/{}", - s3_obj.storage.as_deref().unwrap_or("_default_"), + s3_obj.storage.as_deref().unwrap_or(DEFAULT_STORAGE), s3_obj.s3 ); m.push(Arg { @@ -570,7 +571,7 @@ async fn transform_attach_ducklake( } let db_conn_str = format_attach_db_conn_str(ducklake.catalog_resource, db_type)?; - let storage = ducklake.storage.storage.as_deref().unwrap_or("_default_"); + let storage = ducklake.storage.storage.as_deref().unwrap_or(DEFAULT_STORAGE); let data_path = ducklake.storage.path; // Ducklake 0.3 only requires DATA_PATH at creation and then stores it internally in the catalog @@ -639,7 +640,7 @@ async fn transform_s3_uris(query: &str) -> Result { continue; } let original_str_lit: String = format!("'s3://{}/{}'", storage, s3_path); - storage = "_default_"; + storage = DEFAULT_STORAGE; let new_s3_lit = format!("'s3://{}/{}'", storage, s3_path); transformed_query = Some( diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 5b8ce19d92..bf0101458b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -69,10 +69,10 @@ use windmill_queue::{ MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError, }; -use windmill_audit::audit_oss::{audit_log, AuditAuthor}; +use windmill_audit::audit_oss::audit_log; use windmill_audit::ActionKind; +use windmill_common::audit::AuditAuthor; use windmill_queue::{canceled_job_to_result, push}; - // #[instrument(level = "trace", skip_all)] pub async fn update_flow_status_after_job_completion( db: &DB, diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 9d0fda39b6..906b167d20 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -13245,21 +13245,6 @@ } } }, - "node_modules/svelte-check/node_modules/picomatch": { - "version": "4.0.3", - "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", - "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", - "dev": true, - "license": "MIT", - "optional": true, - "peer": true, - "engines": { - "node": ">=12" - }, - "funding": { - "url": "https://github.com/sponsors/jonschlinkert" - } - }, "node_modules/svelte-eslint-parser": { "version": "0.43.0", "resolved": "https://registry.npmjs.org/svelte-eslint-parser/-/svelte-eslint-parser-0.43.0.tgz",