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

fix: s3 resource is accessed by backend with admin permissions (#3011)

Browse Source
This commit is contained in:
Guillaume Bouvignies
2024-01-16 18:05:40 +01:00
committed by GitHub
parent bd8193defe
commit 602afbe457
2 changed files with 69 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/job_helpers.rs
+22 -30
View File
@@ -132,8 +132,8 @@ struct DuckdbConnectionSettingsQueryV2 {
async fn duckdb_connection_settings_v2(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Json(query): Json<DuckdbConnectionSettingsQueryV2>,
@@ -142,15 +142,15 @@ async fn duckdb_connection_settings_v2(
Some(s3_resource_path) => {
get_s3_resource(
&authed,
&user_db,
&db,
Some(user_db),
&token,
&w_id,
s3_resource_path.as_str(),
)
.await?
}
None => get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?,
None => get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?,
};
let s3_resource = s3_resource_opt.ok_or(error::Error::NotFound(
"No datasets storage resource defined at the workspace level".to_string(),
@@ -225,8 +225,8 @@ struct PolarsStorageOptions {
async fn polars_connection_settings_v2(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Json(query): Json<PolarsConnectionSettingsQueryV2>,
@@ -235,15 +235,15 @@ async fn polars_connection_settings_v2(
Some(s3_resource_path) => {
get_s3_resource(
&authed,
&user_db,
&db,
Some(user_db),
&token,
&w_id,
s3_resource_path.as_str(),
)
.await?
}
None => get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?,
None => get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?,
};
let s3_resource = s3_resource_opt.ok_or(error::Error::NotFound(
"No datasets storage resource defined at the workspace level".to_string(),
@@ -274,8 +274,8 @@ struct S3ResourceInfoQuery {
async fn s3_resource_info(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Json(query): Json<S3ResourceInfoQuery>,
@@ -284,15 +284,15 @@ async fn s3_resource_info(
Some(s3_resource_path) => {
get_s3_resource(
&authed,
&user_db,
&db,
Some(user_db),
&token,
&w_id,
s3_resource_path.as_str(),
)
.await?
}
None => get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?,
None => get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?,
};
let s3_resource = s3_resource_opt.ok_or(error::Error::NotFound(
"No datasets storage resource defined at the workspace level".to_string(),
@@ -307,12 +307,11 @@ struct WindmillLargeFile {
async fn test_connection(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
) -> error::JsonResult<()> {
let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
if s3_resource_opt.is_none() {
return Err(error::Error::NotFound(
"No datasets storage resource defined at the workspace level".to_string(),
@@ -350,13 +349,12 @@ struct ListStoredDatasetsResponse {
async fn list_stored_files(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Query(query): Query<ListStoredFilesQuery>,
) -> error::JsonResult<ListStoredDatasetsResponse> {
let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
@@ -464,14 +462,13 @@ enum WindmillContentType {
async fn load_file_metadata(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Query(query): Query<LoadFileMetadataQuery>,
) -> error::JsonResult<LoadFileMetadataResponse> {
let file_key = query.file_key.clone();
let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
@@ -505,7 +502,6 @@ async fn load_file_metadata(
async fn load_file_preview(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
@@ -519,7 +515,7 @@ async fn load_file_preview(
}
let file_key = query.file_key.clone();
let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
@@ -669,14 +665,13 @@ struct DeleteS3FileQuery {
async fn delete_s3_file(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Query(query): Query<DeleteS3FileQuery>,
) -> error::JsonResult<()> {
let file_key = query.file_key.clone();
let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
@@ -705,13 +700,12 @@ struct MoveS3FileQuery {
async fn move_s3_file(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Query(query): Query<MoveS3FileQuery>,
) -> error::JsonResult<()> {
let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
@@ -774,8 +768,8 @@ struct UploadFileResponse {
async fn multipart_upload_s3_file(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Tokened { token }: Tokened,
Path(w_id): Path<String>,
Json(query): Json<UploadFileQuery>,
@@ -791,15 +785,15 @@ async fn multipart_upload_s3_file(
Some(s3_resource_path) => {
get_s3_resource(
&authed,
&user_db,
&db,
Some(user_db),
&token,
&w_id,
s3_resource_path.as_str(),
)
.await?
}
None => get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?,
None => get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?,
};
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
@@ -920,20 +914,18 @@ pub struct S3Object {
async fn get_workspace_s3_resource<'c>(
authed: &ApiAuthed,
user_db: &UserDB,
db: &DB,
user_db: Option<UserDB>,
token: &str,
w_id: &str,
) -> error::Result<Option<S3Resource>> {
let mut tx = user_db.clone().begin(authed).await?;
let raw_lfs_opt = sqlx::query_scalar!(
"SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1",
w_id
)
.fetch_optional(&mut *tx)
.fetch_optional(db)
.await?
.flatten();
tx.commit().await?;
if raw_lfs_opt.is_none() {
return Ok(None);
@@ -959,13 +951,13 @@ async fn get_workspace_s3_resource<'c>(
Some(stripped) => stripped,
None => s3_lfs.s3_resource_path.as_str(),
};
return get_s3_resource(authed, user_db, db, token, w_id, stripped_resource_path).await;
return get_s3_resource(authed, db, user_db, token, w_id, stripped_resource_path).await;
}
async fn get_s3_resource<'c>(
authed: &ApiAuthed,
user_db: &UserDB,
db: &DB,
user_db: Option<UserDB>,
token: &str,
w_id: &str,
s3_resource_path: &str,
backend/windmill-api/src/resources.rs
+47 -13
View File
@@ -401,7 +401,7 @@ async fn get_resource_value_interpolated(
) -> JsonResult<Option<serde_json::Value>> {
return get_resource_value_interpolated_internal(
&authed,
&user_db,
Some(user_db),
&db,
w_id.as_str(),
path.to_path(),
@@ -416,14 +416,14 @@ use async_recursion::async_recursion;
pub async fn get_resource_value_interpolated_internal(
authed: &ApiAuthed,
user_db: &UserDB,
user_db: Option<UserDB>, // if none, no permission will be checked to access the resource
db: &DB,
workspace: &str,
path: &str,
job_id: Option<Uuid>,
token: &str,
) -> Result<Option<serde_json::Value>> {
let mut tx = user_db.clone().begin(authed).await?;
let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?;
let value_o = sqlx::query_scalar!(
"SELECT value from resource WHERE path = $1 AND workspace_id = $2",
@@ -440,7 +440,16 @@ pub async fn get_resource_value_interpolated_internal(
let value = not_found_if_none(value_o, "Resource", path)?;
if let Some(value) = value {
Ok(Some(
transform_json_value(authed, user_db, db, workspace, value, &job_id, token).await?,
transform_json_value(
authed,
user_db.clone(),
db,
workspace,
value,
&job_id,
token,
)
.await?,
))
} else {
Ok(None)
@@ -450,7 +459,7 @@ pub async fn get_resource_value_interpolated_internal(
#[async_recursion]
pub async fn transform_json_value<'c>(
authed: &ApiAuthed,
user_db: &UserDB,
user_db: Option<UserDB>, // if none, no permission will be checked to access the resources/variables
db: &DB,
workspace: &str,
v: Value,
@@ -460,9 +469,19 @@ pub async fn transform_json_value<'c>(
match v {
Value::String(y) if y.starts_with("$var:") => {
let path = y.strip_prefix("$var:").unwrap();
let tx: Transaction<'_, Postgres> = user_db.clone().begin(authed).await?;
let v = crate::variables::get_value_internal(tx, db, workspace, path, &authed.username)
.await?;
let tx: Transaction<'_, Postgres> =
authed_transaction_or_default(authed, user_db.clone(), db).await?;
let v = crate::variables::get_value_internal(
tx,
db,
workspace,
path,
user_db
.clone()
.map(|_| authed.username.as_str())
.unwrap_or("backend"),
)
.await?;
Ok(Value::String(v))
}
Value::String(y) if y.starts_with("$res:") => {
@@ -470,7 +489,8 @@ pub async fn transform_json_value<'c>(
if path.split("/").count() < 2 {
return Err(Error::InternalErr(format!("Invalid resource path: {path}")));
}
let mut tx: Transaction<'_, Postgres> = user_db.clone().begin(authed).await?;
let mut tx: Transaction<'_, Postgres> =
authed_transaction_or_default(authed, user_db.clone(), db).await?;
let v = sqlx::query_scalar!(
"SELECT value from resource WHERE path = $1 AND workspace_id = $2",
path,
@@ -481,13 +501,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, db, workspace, v, job_id, token).await
transform_json_value(authed, user_db.clone(), 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 mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?;
let job = sqlx::query_as::<_, QueuedJob>(
"SELECT * FROM queue WHERE id = $1 AND workspace_id = $2",
)
@@ -500,7 +520,8 @@ pub async fn transform_json_value<'c>(
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 mut tx: Transaction<'_, Postgres> =
authed_transaction_or_default(authed, user_db.clone(), db).await?;
let p = sqlx::query_scalar!("SELECT script_path FROM queue WHERE id = $1", uuid)
.fetch_optional(&mut *tx)
.await?
@@ -539,7 +560,8 @@ pub async fn transform_json_value<'c>(
for (a, b) in m.clone().into_iter() {
m.insert(
a.clone(),
transform_json_value(authed, user_db, db, workspace, b, job_id, token).await?,
transform_json_value(authed, user_db.clone(), db, workspace, b, job_id, token)
.await?,
);
}
Ok(Value::Object(m))
@@ -548,6 +570,18 @@ 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 check_path_conflict<'c>(
tx: &mut Transaction<'c, Postgres>,
w_id: &str,