From dfb1d8fa44222f52b285a37d867a42cb1f27450d Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 20 Jul 2023 20:23:55 +0200 Subject: [PATCH] fix: remove lockfile on any rawinput change in flows --- backend/windmill-api/src/groups.rs | 186 ++++++++++++++++-- backend/windmill-api/src/jobs.rs | 91 +++++---- backend/windmill-api/src/lib.rs | 5 +- backend/windmill-api/src/scim.rs | 113 +++++++++-- .../flows/content/FlowModuleComponent.svelte | 5 +- 5 files changed, 327 insertions(+), 73 deletions(-) diff --git a/backend/windmill-api/src/groups.rs b/backend/windmill-api/src/groups.rs index f3e3569aeb..d0c9640d5d 100644 --- a/backend/windmill-api/src/groups.rs +++ b/backend/windmill-api/src/groups.rs @@ -9,6 +9,7 @@ use crate::{ db::{UserDB, DB}, users::{get_groups_for_user, Authed}, + utils::require_super_admin, }; use axum::{ extract::{Extension, Path, Query}, @@ -39,6 +40,15 @@ pub fn workspaced_service() -> Router { .route("/is_owner", get(is_owner)) } +pub fn global_service() -> Router { + Router::new() + .route("/list", get(list_igroups)) + .route("/create", post(create_igroup)) + .route("/delete/:name", delete(delete_igroup)) + .route("/adduser/:name", post(add_user_igroup)) + .route("/removeuser/:name", post(remove_user_igroup)) +} + #[derive(FromRow, Serialize, Deserialize)] pub struct Group { pub workspace_id: String, @@ -72,6 +82,11 @@ pub struct Username { pub username: String, } +#[derive(Deserialize)] +pub struct Email { + pub email: String, +} + async fn list_groups( Extension(db): Extension, Path(w_id): Path, @@ -193,8 +208,7 @@ async fn create_group( check_name_conflict(&mut tx, &w_id, &ng.name).await?; - sqlx::query_as!( - Group, + sqlx::query!( "INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES ($1, $2, $3, $4)", w_id, ng.name, @@ -204,8 +218,7 @@ async fn create_group( .execute(&mut *tx) .await?; - sqlx::query_as!( - Group, + sqlx::query!( "INSERT INTO usr_to_group (workspace_id, usr, group_) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", &w_id, &authed.username, @@ -229,6 +242,62 @@ async fn create_group( Ok(format!("Created group {}", ng.name)) } +async fn create_igroup( + authed: Authed, + Extension(db): Extension, + Json(ng): Json, +) -> Result { + let mut tx = db.begin().await?; + require_super_admin(&mut tx, &authed.email).await?; + sqlx::query!( + "INSERT INTO instance_group (name, summary) VALUES ($1, $2) ON CONFLICT DO NOTHING", + ng.name, + ng.summary, + ) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + &authed.username, + "igroup.create", + ActionKind::Create, + "global", + Some(&ng.name.to_string()), + None, + ) + .await?; + + tx.commit().await?; + Ok(format!("Created group {}", ng.name)) +} + +async fn delete_igroup( + authed: Authed, + Extension(db): Extension, + Json(ng): Json, +) -> Result { + let mut tx = db.begin().await?; + require_super_admin(&mut tx, &authed.email).await?; + sqlx::query!("DELETE FROM instance_group WHERE name = $1", ng.name,) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + &authed.username, + "igroup.delete", + ActionKind::Delete, + "global", + Some(&ng.name.to_string()), + None, + ) + .await?; + + tx.commit().await?; + Ok(format!("Created group {}", ng.name)) +} + pub async fn get_group_opt<'c>( db: &mut Transaction<'c, Postgres>, w_id: &str, @@ -294,7 +363,8 @@ async fn delete_group( if name == "all" { return Err(Error::BadRequest( - "The group 'all' is a special group that contains all users and cannot be deleted".to_string(), + "The group 'all' is a special group that contains all users and cannot be deleted" + .to_string(), )); } @@ -344,8 +414,7 @@ async fn update_group( } not_found_if_none(get_group_opt(&mut tx, &w_id, &name).await?, "Group", &name)?; - sqlx::query_as!( - Group, + sqlx::query!( "UPDATE group_ SET summary = $1 WHERE name = $2 AND workspace_id = $3", eg.summary, &name, @@ -382,8 +451,7 @@ async fn add_user( not_found_if_none(get_group_opt(&mut tx, &w_id, &name).await?, "Group", &name)?; - sqlx::query_as!( - Group, + sqlx::query!( "INSERT INTO usr_to_group (workspace_id, usr, group_) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", &w_id, user_username, @@ -406,6 +474,103 @@ async fn add_user( Ok(format!("Added {} to group {}", user_username, name)) } +async fn add_user_igroup( + authed: Authed, + Extension(db): Extension, + Path(name): Path, + Json(Email { email }): Json, +) -> Result { + let mut tx = db.begin().await?; + + require_super_admin(&mut tx, &authed.email).await?; + + let group_opt = sqlx::query_scalar!("SELECT name FROM instance_group WHERE name = $1", name) + .fetch_optional(&mut *tx) + .await?; + + not_found_if_none(group_opt, "IGroup", &name)?; + + sqlx::query!( + "INSERT INTO email_to_igroup (email, igroup) VALUES ($1, $2) ON CONFLICT DO NOTHING", + email, + name, + ) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + &authed.username, + "igroup.adduser", + ActionKind::Update, + "global", + Some(&name.to_string()), + Some([("email", email.as_str())].into()), + ) + .await?; + tx.commit().await?; + Ok(format!("Added {} to igroup {}", email, name)) +} + +#[derive(Serialize)] +struct IGroup { + name: String, + emails: Option>, +} +async fn list_igroups(authed: Authed, Extension(db): Extension) -> JsonResult> { + let mut tx = db.begin().await?; + + require_super_admin(&mut tx, &authed.email).await?; + let groups = sqlx::query_as!( + IGroup, + "SELECT igroup as name, array_agg(email_to_igroup.email) as emails FROM email_to_igroup GROUP BY igroup" + ) + .fetch_all(&mut *tx) + .await?; + + tx.commit().await?; + + return Ok(Json(groups)); +} + +async fn remove_user_igroup( + authed: Authed, + Extension(db): Extension, + Path(name): Path, + Json(Email { email }): Json, +) -> Result { + let mut tx = db.begin().await?; + + require_super_admin(&mut tx, &authed.email).await?; + + let group_opt = sqlx::query_scalar!("SELECT name FROM instance_group WHERE name = $1", name,) + .fetch_optional(&mut *tx) + .await?; + + not_found_if_none(group_opt, "IGroup", &name)?; + + sqlx::query!( + "DELETE FROM email_to_igroup WHERE email = $1 AND igroup = $2", + email, + name, + ) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + &authed.username, + "igroup.removeuser", + ActionKind::Update, + "global", + Some(&name.to_string()), + Some([("email", email.as_str())].into()), + ) + .await?; + tx.commit().await?; + Ok(format!("Added {} to igroup {}", email, name)) +} + async fn remove_user( authed: Authed, Extension(db): Extension, @@ -422,8 +587,7 @@ async fn remove_user( if &name == "all" { return Err(Error::BadRequest(format!("Cannot delete users from all"))); } - sqlx::query_as!( - Group, + sqlx::query!( "DELETE FROM usr_to_group WHERE usr = $1 AND group_ = $2 AND workspace_id = $3", user_username, name, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 6c30388327..e8fd45ba26 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1292,55 +1292,54 @@ where ) -> Result { let content_type_header = req.headers().get(CONTENT_TYPE); let content_type = content_type_header.and_then(|value| value.to_str().ok()); - if let Some(content_type) = content_type { - if content_type.starts_with("application/json") { - if req - .uri() - .query() - .map(|x| x.contains("raw=true")) - .unwrap_or(false) - { - let bytes = Bytes::from_request(req, _state) - .await - .map_err(IntoResponse::into_response)?; - let str = String::from_utf8(bytes.to_vec()).map_err(|e| { - Error::BadRequest(format!("invalid utf8: {}", e)).into_response() + if content_type.is_none() || content_type.unwrap().starts_with("application/json") { + if req + .uri() + .query() + .map(|x| x.contains("raw=true")) + .unwrap_or(false) + { + let bytes = Bytes::from_request(req, _state) + .await + .map_err(IntoResponse::into_response)?; + let str = String::from_utf8(bytes.to_vec()).map_err(|e| { + Error::BadRequest(format!("invalid utf8: {}", e)).into_response() + })?; + let payload = + serde_json::from_str::>(&str).map_err(|e| { + Error::BadRequest(format!("invalid json: {}", e)).into_response() })?; - let payload = - serde_json::from_str::>(&str).map_err(|e| { - Error::BadRequest(format!("invalid json: {}", e)).into_response() - })?; - return match payload { - Some(serde_json::Value::Object(map)) => Ok(Self(Some(map), Some(str))), - None => Ok(Self(None, Some(str))), - Some(x) => { - let mut map = serde_json::Map::new(); - map.insert("body".to_string(), x); - Ok(Self(Some(map), Some(str))) - } - }; - } else { - let Json(payload): Json> = - req.extract().await.map_err(IntoResponse::into_response)?; - return match payload { - Some(serde_json::Value::Object(map)) => Ok(Self(Some(map), None)), - None => Ok(Self(None, None)), - Some(x) => { - let mut map = serde_json::Map::new(); - map.insert("body".to_string(), x); - Ok(Self(Some(map), None)) - } - }; - } - } - - if content_type.starts_with("application/x-www-form-urlencoded") { - let Form(payload) = req.extract().await.map_err(IntoResponse::into_response)?; - return Ok(Self(Some(payload), None)); + return match payload { + Some(serde_json::Value::Object(map)) => Ok(Self(Some(map), Some(str))), + None => Ok(Self(None, Some(str))), + Some(x) => { + let mut map = serde_json::Map::new(); + map.insert("body".to_string(), x); + Ok(Self(Some(map), Some(str))) + } + }; + } else { + let Json(payload): Json> = + req.extract().await.map_err(IntoResponse::into_response)?; + return match payload { + Some(serde_json::Value::Object(map)) => Ok(Self(Some(map), None)), + None => Ok(Self(None, None)), + Some(x) => { + let mut map = serde_json::Map::new(); + map.insert("body".to_string(), x); + Ok(Self(Some(map), None)) + } + }; } + } else if content_type + .unwrap() + .starts_with("application/x-www-form-urlencoded") + { + let Form(payload) = req.extract().await.map_err(IntoResponse::into_response)?; + return Ok(Self(Some(payload), None)); + } else { + Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response()) } - - Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response()) } } pub struct QueryOrBody(pub Option); diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 51e91a31d8..a34c270e75 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -8,7 +8,7 @@ use crate::oauth2::AllClients; use crate::saml::{SamlSsoLogin, ServiceProviderExt}; -use crate::scim::has_basic_auth; +use crate::scim::has_scim_token; use crate::{ db::UserDB, oauth2::{build_oauth_clients, SlackVerifier}, @@ -220,6 +220,7 @@ pub async fn run_server( ) .nest("/workers", workers::global_service()) .nest("/scripts", scripts::global_service()) + .nest("/groups", groups::global_service()) .nest("/flows", flows::global_service()) .nest("/apps", apps::global_service().layer(cors.clone())) .nest("/schedules", schedule::global_service()) @@ -231,7 +232,7 @@ pub async fn run_server( ) .nest( "/scim", - saml::global_service().route_layer(axum::middleware::from_fn(has_basic_auth)), + scim::global_service().route_layer(axum::middleware::from_fn(has_scim_token)), ) .nest("/scripts_u", scripts::global_unauthed_service()) .nest( diff --git a/backend/windmill-api/src/scim.rs b/backend/windmill-api/src/scim.rs index 2e6c4ecd8b..c7065fbe6f 100644 --- a/backend/windmill-api/src/scim.rs +++ b/backend/windmill-api/src/scim.rs @@ -7,26 +7,37 @@ */ use axum::{ + extract::{Path, Query}, middleware::Next, response::{IntoResponse, Response}, routing::{get, post}, - Router, + Extension, Router, }; use bytes::{BufMut, BytesMut}; use hyper::{header, http::HeaderValue, Request, StatusCode}; use mime_guess::mime; -use serde::Serialize; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use sql_builder::SqlBuilder; use windmill_common::error::{Error, Result}; +use crate::db::DB; + lazy_static::lazy_static! { - static ref SCIM_PASSWORD: Option = std::env::var("SCIM_PASSWORD") + static ref SCIM_TOKEN: Option = std::env::var("SCIM_TOKEN") .ok(); } +pub fn global_service() -> Router { + Router::new().route("/Users", get(get_users)) + // .route("/Groups", get(get_groups).post(create_group)) + // .route("/Groups/:id", get(get_group)) +} + #[derive(Debug, Clone, Copy, Default)] pub struct JsonScim(pub T); -pub async fn has_basic_auth(request: Request, next: Next) -> Response { +pub async fn has_scim_token(request: Request, next: Next) -> Response { let header = request.headers().get("Authorization"); next.run(request).await } @@ -63,16 +74,94 @@ where } } -pub fn global_service() -> Router { - Router::new() - .route("/authorize", post(authorize)) - .route("/Users", get(get_users)) +#[derive(Serialize)] +struct User { + id: String, + userName: String, +} +pub fn resource_response(schema: &str, resources: Vec) -> JsonScim +where + S: Serialize, +{ + return JsonScim(json!({ + "schemas": [schema], + "totalResults": resources.len(), + "Resources": resources, + "startIndex": 1, + "itemsPerPage": 100, + })); } -pub async fn authorize() -> Result { - Ok("Hello, World!".to_string()) +#[derive(Deserialize)] +pub struct ScimQuery { + startIndex: Option, + count: Option, } -pub async fn get_users() -> JsonScimResult { - Ok(JsonScim("Hello, World!".to_string())) +pub async fn get_users( + Extension(db): Extension, + Query(query): Query, +) -> Result> { + let sqlb = SqlBuilder::select_from("usr") + .fields(&["email"]) + .limit(query.count.unwrap_or(100000)) + .offset(query.startIndex.map(|x| x - 1).unwrap_or(0)) + .clone(); + + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let users = sqlx::query_scalar(&sql) + .fetch_all(&db) + .await? + .into_iter() + .map(|x: String| User { id: x.clone(), userName: x }) + .collect(); + Ok(resource_response( + "urn:ietf:params:scim:api:messages:2.0:ListResponse", + users, + )) } + +pub async fn get_groups( + Extension(db): Extension, + Query(query): Query, +) -> Result> { + let sqlb = SqlBuilder::select_from("instance_group") + .fields(&["email"]) + .limit(query.count.unwrap_or(100000)) + .offset(query.startIndex.map(|x| x - 1).unwrap_or(0)) + .clone(); + + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let users = sqlx::query_scalar(&sql) + .fetch_all(&db) + .await? + .into_iter() + .map(|x: String| User { id: x.clone(), userName: x }) + .collect(); + Ok(resource_response( + "urn:ietf:params:scim:api:messages:2.0:ListResponse", + users, + )) +} + +// pub async fn get_group( +// Extension(db): Extension, +// Query(query): Query, +// Path(id): Path, +// ) -> Result> { +// let groups = sqlx::query_as!( +// IGroup, +// "SELECT igroup as name, array_agg(email_to_igroup.email) as emails FROM email_to_igroup GROUP BY igroup" +// ) +// .fetch_all(&mut *tx) +// .await?; +// Ok(resource_response( +// "urn:ietf:params:scim:api:messages:2.0:ListResponse", +// groups, +// )) +// } +// pub async fn create_group( +// Extension(db): Extension, +// Query(query): Query, +// ) -> Result> { +// } diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index c010e742fe..77b3ba6177 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -97,7 +97,9 @@ validCode = true inputTransformSchemaForm?.setArgs(input_transforms) - + if (flowModule.value.type == 'rawscript' && flowModule.value.lock != undefined) { + flowModule.value.lock = undefined + } await tick() if (!deepEqual(schema, $flowStateStore[flowModule.id]?.schema)) { if (!$flowStateStore[flowModule.id]) { @@ -150,7 +152,6 @@ }} on:reload={async () => { if (flowModule.value.type == 'script') { - console.log('reload') if (flowModule.value.hash != undefined) { flowModule.value.hash = await getLatestHashForScript(flowModule.value.path) }