starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-24 00:00:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(backend): add presigned url support for object storage (#7328)

Browse Source 
* presigned

* all

* all

* all

* all

* all

* all

* all

* nit

* nit

* ee-ref

* presigned

* presigned
This commit is contained in:
Ruben Fiszel
2025-12-10 16:35:26 +01:00
committed by GitHub
parent 1857ff820a
commit acc7bda015
19 changed files with 370 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
a61644247e621f863a2a72626f47ee1c13267c49
8f6e0bd9f273ac9da5f4fb3a44979f17b8e1b9b5
backend/windmill-api/src/apps.rs
+190 -215
View File
@@ -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::<i64>().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::<Vec<_>>()
} else {
s3_inputs
.iter()
.filter(|s3_input| s3_input.allow_workspace_resource)
.collect::<Vec<_>>()
};
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::<Vec<_>>()
} else {
s3_inputs
.iter()
.filter(|s3_input| s3_input.allow_workspace_resource)
.collect::<Vec<_>>()
};
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<String>,
sig: Option<String>,
#[cfg(feature = "private")]
exp: Option<String>,
}
@@ -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?;
backend/windmill-api/src/args.rs
+1 -1
View File
@@ -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?;
backend/windmill-api/src/db.rs
+1 -12
View File
@@ -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<Authed> 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)
backend/windmill-api/src/job_helpers_oss.rs
+5 -10
View File
@@ -51,7 +51,6 @@ pub async fn get_workspace_s3_resource<'c>(
_authed: &ApiAuthed,
_db: &DB,
_user_db: Option<UserDB>,
_token: &str,
_w_id: &str,
_storage: Option<String>,
) -> windmill_common::error::Result<(Option<bool>, Option<ObjectStoreResource>)> {
@@ -64,12 +63,12 @@ pub fn get_random_file_name(_file_extension: Option<String>) -> 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<UserDB>,
_token: &str,
_db_with_opt_authed: &DbWithOptAuthed<'c, ApiAuthed>,
_w_id: &str,
_resource_path: &str,
_resource_type: Option<StorageResourceType>,
@@ -109,7 +108,6 @@ pub async fn download_s3_file_internal(
_authed: ApiAuthed,
_db: &DB,
_user_db: Option<UserDB>,
_token: &str,
_w_id: &str,
_query: DownloadFileQuery,
) -> error::Result<Response> {
@@ -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<windmill_common::db::UserDB>,
_token: &str,
_db_with_opt_authed: &DbWithOptAuthed<'c, ApiAuthed>,
_w_id: &str,
_storage: Option<String>,
_paths: &[(&str, windmill_common::s3_helpers::S3Permission)],
backend/windmill-api/src/jobs.rs
+2 -2
View File
@@ -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::{
backend/windmill-api/src/resources.rs
+54 -77
View File
@@ -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<UserDB>, // 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<Uuid>,
token: &str,
token_for_context: Option<&str>,
allow_cache: bool,
) -> Result<Option<serde_json::Value>> {
// 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<UserDB>, // 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<ApiAuthed>,
workspace: &str,
v: Value,
job_id: &Option<Uuid>,
token: &str,
token: Option<&str>,
) -> Result<Value> {
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<UserDB>,
db: &DB,
) -> sqlx::error::Result<Transaction<'c, Postgres>> {
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<ApiAuthed>,
// ) -> sqlx::error::Result<Transaction<'c, Postgres>> {
// 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<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path((w_id, path)): Path<(String, StripPath)>,
Query(query): Query<GitCommitHashQuery>,
) -> JsonResult<GitCommitHashResponse> {
@@ -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| {
(
backend/windmill-api/src/triggers/http/handler.rs
-1
View File
@@ -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,
)
backend/windmill-api/src/users.rs
+4 -6
View File
@@ -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 }))
}
}
backend/windmill-api/src/variables.rs
+17 -25
View File
@@ -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<String> {
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<String> {
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)
}
backend/windmill-audit/src/audit_oss.rs
+26 -11
View File
@@ -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<String>,
pub token_prefix: Option<String>,
}
#[cfg(not(feature = "private"))]
#[tracing::instrument(level = "trace", skip_all)]
pub async fn audit_log<'c, E: sqlx::Executor<'c, Database = Postgres>>(
backend/windmill-common/src/audit.rs
+20
View File
@@ -0,0 +1,20 @@
use crate::db::Authable;
#[derive(Clone)]
pub struct AuditAuthor {
pub username: String,
pub email: String,
pub username_override: Option<String>,
pub token_prefix: Option<String>,
}
impl<T: Authable> 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,
}
}
}
backend/windmill-common/src/db.rs
+38 -13
View File
@@ -1,5 +1,7 @@
use sqlx::{Acquire, Pool, Postgres, Transaction};
use crate::audit::AuditAuthor;
pub type DB = Pool<Postgres>;
#[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<UserDB>,
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<UserDB>) -> 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<Self::Connection, 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,
}
})
}
@@ -145,10 +169,11 @@ impl<'c, 'd, T: Authable + Sync> Acquire<'c> for &'c UserDbWithOptAuthed<'d, T>
self,
) -> futures_core::future::BoxFuture<'c, Result<Transaction<'c, Postgres>, 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,
}
})
}
backend/windmill-common/src/lib.rs
+1
View File
@@ -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")]
backend/windmill-common/src/s3_helpers.rs
+2
View File
@@ -546,6 +546,8 @@ impl BundleFormat {
}
}
pub const DEFAULT_STORAGE: &str = "_default_";
pub async fn upload_artifact_to_store(
path: &str,
data: bytes::Bytes,
backend/windmill-queue/src/jobs.rs
+2 -2
View File
@@ -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;
backend/windmill-worker/src/duckdb_executor.rs
+4 -3
View File
@@ -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<String> {
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(
backend/windmill-worker/src/worker_flow.rs
+2 -2
View File
@@ -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,
frontend/package-lock.json
Generated
-15
View File
@@ -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",