diff --git a/backend/windmill-api/src/auth.rs b/backend/windmill-api/src/auth.rs new file mode 100644 index 0000000000..d03e138dba --- /dev/null +++ b/backend/windmill-api/src/auth.rs @@ -0,0 +1,582 @@ +use axum::{ + async_trait, + extract::{FromRequestParts, OriginalUri, Query}, + Extension, +}; +use chrono::TimeZone; +use http::{request::Parts, StatusCode}; +use quick_cache::sync::Cache; +use serde::Deserialize; +use tower_cookies::Cookies; +use tracing::Span; + +use crate::db::{ApiAuthed, DB}; +use std::sync::{ + atomic::{AtomicI64, AtomicU64, Ordering}, + Arc, +}; +use windmill_common::{ + auth::{get_folders_for_user, get_groups_for_user, JWTAuthClaims, JWT_SECRET}, + users::{COOKIE_NAME, SUPERADMIN_SECRET_EMAIL}, +}; + +#[derive(Clone)] +pub struct ExpiringAuthCache { + pub authed: ApiAuthed, + pub expiry: chrono::DateTime, +} + +pub struct AuthCache { + cache: Cache<(String, String), ExpiringAuthCache>, + db: DB, + superadmin_secret: Option, + #[cfg(feature = "enterprise")] + ext_jwks: Option>>, +} + +impl AuthCache { + pub fn new( + db: DB, + superadmin_secret: Option, + #[cfg(feature = "enterprise")] ext_jwks: Option>>, + ) -> Self { + AuthCache { + cache: Cache::new(300), + db, + superadmin_secret, + #[cfg(feature = "enterprise")] + ext_jwks, + } + } + + pub async fn invalidate(&self, w_id: &str, token: String) { + self.cache.remove(&(w_id.to_string(), token)); + } + + pub async fn get_authed(&self, w_id: Option, token: &str) -> Option { + let key = ( + w_id.as_ref().unwrap_or(&"".to_string()).to_string(), + token.to_string(), + ); + let s = self.cache.get(&key).map(|c| c.to_owned()); + match s { + Some(ExpiringAuthCache { authed, expiry }) if expiry > chrono::Utc::now() => { + Some(authed) + } + #[cfg(feature = "enterprise")] + _ if token.starts_with("jwt_ext_") => { + let authed_and_exp = match crate::ee::jwt_ext_auth( + w_id.as_ref(), + token.trim_start_matches("jwt_ext_"), + self.ext_jwks.clone(), + ) + .await + { + Ok(r) => Some(r), + Err(e) => { + tracing::error!("JWT_EXT auth error: {:?}", e); + None + } + }; + + if let Some((authed, exp)) = authed_and_exp.clone() { + self.cache.insert( + key, + ExpiringAuthCache { + authed: authed.clone(), + expiry: chrono::Utc.timestamp_nanos(exp as i64 * 1_000_000_000), + }, + ); + + Some(authed) + } else { + None + } + } + _ if token.starts_with("jwt_") => { + let jwt_secret = JWT_SECRET.read().await; + if !jwt_secret.is_empty() { + let jwt_token = token.trim_start_matches("jwt_"); + + let jwt_result = jsonwebtoken::decode::( + jwt_token, + &jsonwebtoken::DecodingKey::from_secret(jwt_secret.as_bytes()), + &jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::HS256), + ); + + match jwt_result { + Ok(payload) => { + if w_id.is_some_and(|w_id| w_id != payload.claims.workspace_id) { + tracing::error!("JWT auth error: workspace_id mismatch"); + return None; + } + + let username_override = + username_override_from_label(payload.claims.label); + let authed = crate::db::ApiAuthed { + email: payload.claims.email, + username: payload.claims.username, + is_admin: payload.claims.is_admin, + is_operator: payload.claims.is_operator, + groups: payload.claims.groups, + folders: payload.claims.folders, + scopes: None, + username_override, + }; + + self.cache.insert( + key, + ExpiringAuthCache { + authed: authed.clone(), + expiry: chrono::Utc + .timestamp_nanos(payload.claims.exp as i64 * 1_000_000_000), + }, + ); + + Some(authed) + } + Err(err) => { + tracing::error!("JWT auth error: {:?}", err); + None + } + } + } else { + tracing::error!("JWT auth error: no jwt secret set"); + None + } + } + _ => { + let user_o = sqlx::query_as::<_, (Option, Option, bool, Option>, Option)>( + "UPDATE token SET last_used_at = now() WHERE token = $1 AND (expiration > NOW() \ + OR expiration IS NULL) AND (workspace_id IS NULL OR workspace_id = $2) RETURNING owner, email, super_admin, scopes, label", + ) + .bind(token) + .bind(w_id.as_ref()) + .fetch_optional(&self.db) + .await + .ok() + .flatten(); + + if let Some(user) = user_o { + let authed_o = { + match user { + (Some(owner), Some(email), super_admin, _, label) if w_id.is_some() => { + let username_override = username_override_from_label(label); + if let Some((prefix, name)) = owner.split_once('/') { + if prefix == "u" { + let (is_admin, is_operator) = if super_admin { + (true, false) + } else { + let r = sqlx::query!( + "SELECT is_admin, operator FROM usr where username = $1 AND \ + workspace_id = $2 AND disabled = false", + name, + &w_id.as_ref().unwrap() + ) + .fetch_one(&self.db) + .await + .ok(); + if let Some(r) = r { + (r.is_admin, r.operator) + } else { + (false, true) + } + }; + + let w_id = &w_id.unwrap(); + let groups = + get_groups_for_user(w_id, &name, &email, &self.db) + .await + .ok() + .unwrap_or_default(); + + let folders = + get_folders_for_user(w_id, &name, &groups, &self.db) + .await + .ok() + .unwrap_or_default(); + + Some(ApiAuthed { + email: email, + username: name.to_string(), + is_admin, + is_operator, + groups, + folders, + scopes: None, + username_override, + }) + } else { + let groups = vec![name.to_string()]; + let folders = get_folders_for_user( + &w_id.unwrap(), + "", + &groups, + &self.db, + ) + .await + .ok() + .unwrap_or_default(); + Some(ApiAuthed { + email: email, + username: format!("group-{name}"), + is_admin: false, + groups, + is_operator: false, + folders, + scopes: None, + username_override, + }) + } + } else { + let groups = vec![]; + let folders = vec![]; + Some(ApiAuthed { + email: email, + username: owner, + is_admin: super_admin, + is_operator: true, + groups, + folders, + scopes: None, + username_override, + }) + } + } + (_, Some(email), super_admin, scopes, label) => { + let username_override = username_override_from_label(label); + if w_id.is_some() { + let row_o = sqlx::query_as::<_, (String, bool, bool)>( + "SELECT username, is_admin, operator FROM usr where email = $1 AND \ + workspace_id = $2 AND disabled = false", + ) + .bind(&email) + .bind(&w_id.as_ref().unwrap()) + .fetch_optional(&self.db) + .await + .unwrap_or(Some(("error".to_string(), false, false))); + + match row_o { + Some((username, is_admin, is_operator)) => { + let groups = get_groups_for_user( + &w_id.as_ref().unwrap(), + &username, + &email, + &self.db, + ) + .await + .ok() + .unwrap_or_default(); + + let folders = get_folders_for_user( + &w_id.unwrap(), + &username, + &groups, + &self.db, + ) + .await + .ok() + .unwrap_or_default(); + Some(ApiAuthed { + email, + username, + is_admin: is_admin || super_admin, + is_operator, + groups, + folders, + scopes, + username_override, + }) + } + None if super_admin => Some(ApiAuthed { + email: email.clone(), + username: email, + is_admin: super_admin, + is_operator: false, + groups: vec![], + folders: vec![], + scopes, + username_override, + }), + None => None, + } + } else { + Some(ApiAuthed { + email: email.to_string(), + username: email, + is_admin: super_admin, + is_operator: true, + groups: Vec::new(), + folders: Vec::new(), + scopes, + username_override, + }) + } + } + _ => None, + } + }; + if let Some(authed) = authed_o.as_ref() { + self.cache.insert( + key, + ExpiringAuthCache { + authed: authed.clone(), + expiry: chrono::Utc::now() + + chrono::Duration::try_seconds(120).unwrap(), + }, + ); + } + authed_o + } else if self + .superadmin_secret + .as_ref() + .map(|x| x == token) + .unwrap_or(false) + { + Some(ApiAuthed { + email: SUPERADMIN_SECRET_EMAIL.to_string(), + username: "superadmin_secret".to_string(), + is_admin: true, + is_operator: false, + groups: Vec::new(), + folders: Vec::new(), + scopes: None, + username_override: None, + }) + } else { + None + } + } + } + } +} + +async fn extract_token(parts: &mut Parts, state: &S) -> Option { + let auth_header = parts + .headers + .get(http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|s| s.strip_prefix("Bearer ")); + + let from_cookie = match auth_header { + Some(x) => Some(x.to_owned()), + None => Extension::::from_request_parts(parts, state) + .await + .ok() + .and_then(|cookies| cookies.get(COOKIE_NAME).map(|c| c.value().to_owned())), + }; + + #[derive(Deserialize)] + struct Token { + token: Option, + } + match from_cookie { + Some(token) => Some(token), + None => Query::::from_request_parts(parts, state) + .await + .ok() + .and_then(|token| token.token.clone()), + } +} + +#[derive(Clone, Debug)] +pub struct Tokened { + pub token: String, +} + +pub struct OptTokened { + #[allow(dead_code)] + pub token: Option, +} + +struct BruteForceCounter { + counter: AtomicU64, + last_reset: AtomicI64, +} + +lazy_static::lazy_static! { + static ref BRUTE_FORCE_COUNTER: BruteForceCounter = + BruteForceCounter { last_reset: AtomicI64::new(0), counter: AtomicU64::new(0) }; +} + +impl BruteForceCounter { + async fn increment(&self) { + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + if self.counter.fetch_add(1, Ordering::Relaxed) > 10000 { + tracing::error!( + "Brute force attack to find valid token detected, sleeping unauthorized response for 2 seconds" + ); + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + } + if now - self.last_reset.load(Ordering::Relaxed) > 60 { + self.counter.store(0, Ordering::Relaxed); + self.last_reset.store(now, Ordering::Relaxed); + } + } +} + +#[async_trait] +impl FromRequestParts for Tokened +where + S: Send + Sync, +{ + type Rejection = (StatusCode, String); + + async fn from_request_parts( + parts: &mut Parts, + state: &S, + ) -> std::result::Result { + if parts.method == http::Method::OPTIONS { + return Ok(Tokened { token: "".to_string() }); + }; + let already_tokened = parts.extensions.get::(); + if let Some(tokened) = already_tokened { + Ok(tokened.clone()) + } else { + let token_o = extract_token(parts, state).await; + if let Some(token) = token_o { + let tokened = Self { token }; + parts.extensions.insert(tokened.clone()); + Ok(tokened) + } else { + BRUTE_FORCE_COUNTER.increment().await; + Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned())) + } + } + } +} + +#[async_trait] +impl FromRequestParts for OptTokened +where + S: Send + Sync, +{ + type Rejection = (StatusCode, String); + + async fn from_request_parts( + parts: &mut Parts, + state: &S, + ) -> std::result::Result { + if parts.method == http::Method::OPTIONS { + return Ok(OptTokened { token: None }); + }; + let already_tokened = parts.extensions.get::(); + if let Some(tokened) = already_tokened { + Ok(OptTokened { token: Some(tokened.token.clone()) }) + } else { + let token_o = extract_token(parts, state).await; + Ok(OptTokened { token: token_o }) + } + } +} + +#[async_trait] +impl FromRequestParts for ApiAuthed +where + S: Send + Sync, +{ + type Rejection = (StatusCode, String); + + async fn from_request_parts( + parts: &mut Parts, + state: &S, + ) -> std::result::Result { + if parts.method == http::Method::OPTIONS { + return Ok(ApiAuthed { + email: "".to_owned(), + username: "".to_owned(), + is_admin: false, + is_operator: false, + groups: Vec::new(), + folders: Vec::new(), + scopes: None, + username_override: None, + }); + }; + let already_authed = parts.extensions.get::(); + if let Some(authed) = already_authed { + Ok(authed.clone()) + } else { + let already_tokened = parts.extensions.get::(); + let token_o = if let Some(token) = already_tokened { + Some(token.token.clone()) + } else { + extract_token(parts, state).await + }; + let original_uri = OriginalUri::from_request_parts(parts, state) + .await + .ok() + .map(|x| x.0) + .unwrap_or_default(); + let path_vec: Vec<&str> = original_uri.path().split("/").collect(); + + let workspace_id = if path_vec.len() >= 4 && path_vec[0] == "" && path_vec[2] == "w" { + Some(path_vec[3].to_owned()) + } else { + if path_vec.len() >= 5 + && path_vec[0] == "" + && path_vec[2] == "srch" + && path_vec[3] == "w" + { + Some(path_vec[4].to_string()) + } else { + None + } + }; + if let Some(token) = token_o { + if let Ok(Extension(cache)) = + Extension::>::from_request_parts(parts, state).await + { + if let Some(authed) = cache.get_authed(workspace_id.clone(), &token).await { + parts.extensions.insert(authed.clone()); + if authed.scopes.as_ref().is_some_and(|scopes| { + scopes + .iter() + .any(|s| s.starts_with("jobs:") || s.starts_with("run:")) + }) && (path_vec.len() < 3 + || (path_vec[4] != "jobs" && path_vec[4] != "jobs_u")) + { + BRUTE_FORCE_COUNTER.increment().await; + return Err(( + StatusCode::UNAUTHORIZED, + format!("Unauthorized scoped token: {:?}", authed.scopes), + )); + } + Span::current().record("username", &authed.username.as_str()); + Span::current().record("email", &authed.email); + + if let Some(workspace_id) = workspace_id { + Span::current().record("workspace_id", &workspace_id); + } + return Ok(authed); + } + } + } + BRUTE_FORCE_COUNTER.increment().await; + Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned())) + } + } +} + +fn username_override_from_label(label: Option) -> Option { + match label { + Some(label) + if label.starts_with("webhook-") + || label.starts_with("http-") + || label.starts_with("email-") + || label.starts_with("ws-") => + { + Some(label) + } + Some(label) if label.starts_with("ephemeral-script-end-user-") => Some( + label + .trim_start_matches("ephemeral-script-end-user-") + .to_string(), + ), + Some(label) if label == "Ephemeral lsp token" => Some("lsp".to_string()), + Some(label) if label != "ephemeral-script" && label != "session" && !label.is_empty() => { + Some(format!("label-{label}")) + } + _ => None, + } +} diff --git a/backend/windmill-api/src/folders.rs b/backend/windmill-api/src/folders.rs index a2ece7b55b..208ec63985 100644 --- a/backend/windmill-api/src/folders.rs +++ b/backend/windmill-api/src/folders.rs @@ -11,8 +11,9 @@ use std::sync::Arc; use crate::db::ApiAuthed; use crate::{ + auth::AuthCache, db::DB, - users::{AuthCache, Tokened}, + users::Tokened, webhook_util::{WebhookMessage, WebhookShared}, }; use axum::{ diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 723b5a5c95..96a0f7e3f6 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -55,6 +55,7 @@ mod ai; mod apps; mod args; mod audit; +mod auth; mod capture; mod concurrency_groups; mod configs; @@ -108,6 +109,8 @@ mod websocket_triggers; mod workers; mod workspaces; mod workspaces_ee; +mod workspaces_export; +mod workspaces_extra; pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB @@ -195,7 +198,7 @@ pub async fn run_server( #[cfg(feature = "enterprise")] let ext_jwks = ExternalJwks::load().await; - let auth_cache = Arc::new(users::AuthCache::new( + let auth_cache = Arc::new(crate::auth::AuthCache::new( db.clone(), std::env::var("SUPERADMIN_SECRET").ok(), #[cfg(feature = "enterprise")] diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 090bff24aa..69342120a5 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -12,7 +12,8 @@ use crate::{ triggers::{ get_triggers_count_internal, list_tokens_internal, TriggersCount, TruncatedTokenWithEmail, }, - users::{maybe_refresh_folders, require_owner_of_path, AuthCache}, + users::{maybe_refresh_folders, require_owner_of_path}, + auth::AuthCache, utils::WithStarredInfoQuery, webhook_util::{WebhookMessage, WebhookShared}, HTTP_CLIENT, diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index aaa407b6df..a725d75a37 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -8,11 +8,12 @@ #![allow(non_snake_case)] -use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; +use std::sync::atomic::AtomicBool; use std::sync::Arc; use crate::db::ApiAuthed; +pub use crate::auth::Tokened; #[cfg(feature = "enterprise")] use crate::ee::ExternalJwks; use crate::utils::{ @@ -24,16 +25,14 @@ use crate::{ use argon2::{Argon2, PasswordHash, PasswordVerifier}; use axum::{ async_trait, - extract::{Extension, FromRequestParts, OriginalUri, Path, Query}, + extract::{Extension, FromRequestParts, Path, Query}, http::request::Parts, response::{IntoResponse, Response}, routing::{delete, get, post}, Json, Router, }; -use chrono::TimeZone; use hyper::{header::LOCATION, StatusCode}; use lazy_static::lazy_static; -use quick_cache::sync::Cache; use regex::Regex; use serde::{Deserialize, Serialize}; use sqlx::FromRow; @@ -41,27 +40,25 @@ use time::OffsetDateTime; #[cfg(feature = "enterprise")] use tokio::sync::RwLock; use tower_cookies::{Cookie, Cookies}; -use tracing::{Instrument, Span}; +use tracing::Instrument; use windmill_audit::audit_ee::{audit_log, AuditAuthor}; use windmill_audit::ActionKind; use windmill_common::auth::fetch_authed_from_permissioned_as; use windmill_common::global_settings::AUTOMATE_USERNAME_CREATION_SETTING; use windmill_common::oauth2::InstanceEvent; +use windmill_common::users::COOKIE_NAME; use windmill_common::users::{truncate_token, username_to_permissioned_as}; use windmill_common::utils::paginate; use windmill_common::worker::CLOUD_HOSTED; use windmill_common::{ - auth::{get_folders_for_user, get_groups_for_user, JWTAuthClaims, JWT_SECRET}, + auth::{get_folders_for_user, get_groups_for_user}, db::UserDB, error::{self, Error, JsonResult, Result}, - users::SUPERADMIN_SECRET_EMAIL, utils::{not_found_if_none, rd_string, require_admin, Pagination, StripPath}, }; use windmill_git_sync::handle_deployment_metadata; - pub const TTL_TOKEN_DB_H: u32 = 72; -const COOKIE_NAME: &str = "token"; const COOKIE_PATH: &str = "/"; pub fn workspaced_service() -> Router { @@ -126,479 +123,6 @@ pub fn make_unauthed_service() -> Router { .route("/is_first_time_setup", get(is_first_time_setup)) } -fn username_override_from_label(label: Option) -> Option { - match label { - Some(label) - if label.starts_with("webhook-") - || label.starts_with("http-") - || label.starts_with("email-") - || label.starts_with("ws-") => - { - Some(label) - } - Some(label) if label.starts_with("ephemeral-script-end-user-") => Some( - label - .trim_start_matches("ephemeral-script-end-user-") - .to_string(), - ), - Some(label) if label == "Ephemeral lsp token" => Some("lsp".to_string()), - Some(label) if label != "ephemeral-script" && label != "session" && !label.is_empty() => { - Some(format!("label-{label}")) - } - _ => None, - } -} - -#[derive(Clone)] -pub struct ExpiringAuthCache { - pub authed: ApiAuthed, - pub expiry: chrono::DateTime, -} - -pub struct AuthCache { - cache: Cache<(String, String), ExpiringAuthCache>, - db: DB, - superadmin_secret: Option, - #[cfg(feature = "enterprise")] - ext_jwks: Option>>, -} - -impl AuthCache { - pub fn new( - db: DB, - superadmin_secret: Option, - #[cfg(feature = "enterprise")] ext_jwks: Option>>, - ) -> Self { - AuthCache { - cache: Cache::new(300), - db, - superadmin_secret, - #[cfg(feature = "enterprise")] - ext_jwks, - } - } - - pub async fn invalidate(&self, w_id: &str, token: String) { - self.cache.remove(&(w_id.to_string(), token)); - } - - pub async fn get_authed(&self, w_id: Option, token: &str) -> Option { - let key = ( - w_id.as_ref().unwrap_or(&"".to_string()).to_string(), - token.to_string(), - ); - let s = self.cache.get(&key).map(|c| c.to_owned()); - match s { - Some(ExpiringAuthCache { authed, expiry }) if expiry > chrono::Utc::now() => { - Some(authed) - } - #[cfg(feature = "enterprise")] - _ if token.starts_with("jwt_ext_") => { - let authed_and_exp = match crate::ee::jwt_ext_auth( - w_id.as_ref(), - token.trim_start_matches("jwt_ext_"), - self.ext_jwks.clone(), - ) - .await - { - Ok(r) => Some(r), - Err(e) => { - tracing::error!("JWT_EXT auth error: {:?}", e); - None - } - }; - - if let Some((authed, exp)) = authed_and_exp.clone() { - self.cache.insert( - key, - ExpiringAuthCache { - authed: authed.clone(), - expiry: chrono::Utc.timestamp_nanos(exp as i64 * 1_000_000_000), - }, - ); - - Some(authed) - } else { - None - } - } - _ if token.starts_with("jwt_") => { - let jwt_secret = JWT_SECRET.read().await; - if !jwt_secret.is_empty() { - let jwt_token = token.trim_start_matches("jwt_"); - - let jwt_result = jsonwebtoken::decode::( - jwt_token, - &jsonwebtoken::DecodingKey::from_secret(jwt_secret.as_bytes()), - &jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::HS256), - ); - - match jwt_result { - Ok(payload) => { - if w_id.is_some_and(|w_id| w_id != payload.claims.workspace_id) { - tracing::error!("JWT auth error: workspace_id mismatch"); - return None; - } - - let username_override = - username_override_from_label(payload.claims.label); - let authed = crate::db::ApiAuthed { - email: payload.claims.email, - username: payload.claims.username, - is_admin: payload.claims.is_admin, - is_operator: payload.claims.is_operator, - groups: payload.claims.groups, - folders: payload.claims.folders, - scopes: None, - username_override, - }; - - self.cache.insert( - key, - ExpiringAuthCache { - authed: authed.clone(), - expiry: chrono::Utc - .timestamp_nanos(payload.claims.exp as i64 * 1_000_000_000), - }, - ); - - Some(authed) - } - Err(err) => { - tracing::error!("JWT auth error: {:?}", err); - None - } - } - } else { - tracing::error!("JWT auth error: no jwt secret set"); - None - } - } - _ => { - let user_o = sqlx::query_as::<_, (Option, Option, bool, Option>, Option)>( - "UPDATE token SET last_used_at = now() WHERE token = $1 AND (expiration > NOW() \ - OR expiration IS NULL) AND (workspace_id IS NULL OR workspace_id = $2) RETURNING owner, email, super_admin, scopes, label", - ) - .bind(token) - .bind(w_id.as_ref()) - .fetch_optional(&self.db) - .await - .ok() - .flatten(); - - if let Some(user) = user_o { - let authed_o = { - match user { - (Some(owner), Some(email), super_admin, _, label) if w_id.is_some() => { - let username_override = username_override_from_label(label); - if let Some((prefix, name)) = owner.split_once('/') { - if prefix == "u" { - let (is_admin, is_operator) = if super_admin { - (true, false) - } else { - let r = sqlx::query!( - "SELECT is_admin, operator FROM usr where username = $1 AND \ - workspace_id = $2 AND disabled = false", - name, - &w_id.as_ref().unwrap() - ) - .fetch_one(&self.db) - .await - .ok(); - if let Some(r) = r { - (r.is_admin, r.operator) - } else { - (false, true) - } - }; - - let w_id = &w_id.unwrap(); - let groups = - get_groups_for_user(w_id, &name, &email, &self.db) - .await - .ok() - .unwrap_or_default(); - - let folders = - get_folders_for_user(w_id, &name, &groups, &self.db) - .await - .ok() - .unwrap_or_default(); - - Some(ApiAuthed { - email: email, - username: name.to_string(), - is_admin, - is_operator, - groups, - folders, - scopes: None, - username_override, - }) - } else { - let groups = vec![name.to_string()]; - let folders = get_folders_for_user( - &w_id.unwrap(), - "", - &groups, - &self.db, - ) - .await - .ok() - .unwrap_or_default(); - Some(ApiAuthed { - email: email, - username: format!("group-{name}"), - is_admin: false, - groups, - is_operator: false, - folders, - scopes: None, - username_override, - }) - } - } else { - let groups = vec![]; - let folders = vec![]; - Some(ApiAuthed { - email: email, - username: owner, - is_admin: super_admin, - is_operator: true, - groups, - folders, - scopes: None, - username_override, - }) - } - } - (_, Some(email), super_admin, scopes, label) => { - let username_override = username_override_from_label(label); - if w_id.is_some() { - let row_o = sqlx::query_as::<_, (String, bool, bool)>( - "SELECT username, is_admin, operator FROM usr where email = $1 AND \ - workspace_id = $2 AND disabled = false", - ) - .bind(&email) - .bind(&w_id.as_ref().unwrap()) - .fetch_optional(&self.db) - .await - .unwrap_or(Some(("error".to_string(), false, false))); - - match row_o { - Some((username, is_admin, is_operator)) => { - let groups = get_groups_for_user( - &w_id.as_ref().unwrap(), - &username, - &email, - &self.db, - ) - .await - .ok() - .unwrap_or_default(); - - let folders = get_folders_for_user( - &w_id.unwrap(), - &username, - &groups, - &self.db, - ) - .await - .ok() - .unwrap_or_default(); - Some(ApiAuthed { - email, - username, - is_admin: is_admin || super_admin, - is_operator, - groups, - folders, - scopes, - username_override, - }) - } - None if super_admin => Some(ApiAuthed { - email: email.clone(), - username: email, - is_admin: super_admin, - is_operator: false, - groups: vec![], - folders: vec![], - scopes, - username_override, - }), - None => None, - } - } else { - Some(ApiAuthed { - email: email.to_string(), - username: email, - is_admin: super_admin, - is_operator: true, - groups: Vec::new(), - folders: Vec::new(), - scopes, - username_override, - }) - } - } - _ => None, - } - }; - if let Some(authed) = authed_o.as_ref() { - self.cache.insert( - key, - ExpiringAuthCache { - authed: authed.clone(), - expiry: chrono::Utc::now() - + chrono::Duration::try_seconds(120).unwrap(), - }, - ); - } - authed_o - } else if self - .superadmin_secret - .as_ref() - .map(|x| x == token) - .unwrap_or(false) - { - Some(ApiAuthed { - email: SUPERADMIN_SECRET_EMAIL.to_string(), - username: "superadmin_secret".to_string(), - is_admin: true, - is_operator: false, - groups: Vec::new(), - folders: Vec::new(), - scopes: None, - username_override: None, - }) - } else { - None - } - } - } - } -} - -async fn extract_token(parts: &mut Parts, state: &S) -> Option { - let auth_header = parts - .headers - .get(http::header::AUTHORIZATION) - .and_then(|value| value.to_str().ok()) - .and_then(|s| s.strip_prefix("Bearer ")); - - let from_cookie = match auth_header { - Some(x) => Some(x.to_owned()), - None => Extension::::from_request_parts(parts, state) - .await - .ok() - .and_then(|cookies| cookies.get(COOKIE_NAME).map(|c| c.value().to_owned())), - }; - - #[derive(Deserialize)] - struct Token { - token: Option, - } - match from_cookie { - Some(token) => Some(token), - None => Query::::from_request_parts(parts, state) - .await - .ok() - .and_then(|token| token.token.clone()), - } -} - -#[derive(Clone, Debug)] -pub struct Tokened { - pub token: String, -} - -pub struct OptTokened { - #[allow(dead_code)] - pub token: Option, -} - -struct BruteForceCounter { - counter: AtomicU64, - last_reset: AtomicI64, -} - -lazy_static! { - static ref BRUTE_FORCE_COUNTER: BruteForceCounter = - BruteForceCounter { last_reset: AtomicI64::new(0), counter: AtomicU64::new(0) }; -} - -impl BruteForceCounter { - async fn increment(&self) { - let now = time::OffsetDateTime::now_utc().unix_timestamp(); - if self.counter.fetch_add(1, Ordering::Relaxed) > 10000 { - tracing::error!( - "Brute force attack to find valid token detected, sleeping unauthorized response for 2 seconds" - ); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - } - if now - self.last_reset.load(Ordering::Relaxed) > 60 { - self.counter.store(0, Ordering::Relaxed); - self.last_reset.store(now, Ordering::Relaxed); - } - } -} - -#[async_trait] -impl FromRequestParts for Tokened -where - S: Send + Sync, -{ - type Rejection = (StatusCode, String); - - async fn from_request_parts( - parts: &mut Parts, - state: &S, - ) -> std::result::Result { - if parts.method == http::Method::OPTIONS { - return Ok(Tokened { token: "".to_string() }); - }; - let already_tokened = parts.extensions.get::(); - if let Some(tokened) = already_tokened { - Ok(tokened.clone()) - } else { - let token_o = extract_token(parts, state).await; - if let Some(token) = token_o { - let tokened = Self { token }; - parts.extensions.insert(tokened.clone()); - Ok(tokened) - } else { - BRUTE_FORCE_COUNTER.increment().await; - Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned())) - } - } - } -} - -#[async_trait] -impl FromRequestParts for OptTokened -where - S: Send + Sync, -{ - type Rejection = (StatusCode, String); - - async fn from_request_parts( - parts: &mut Parts, - state: &S, - ) -> std::result::Result { - if parts.method == http::Method::OPTIONS { - return Ok(OptTokened { token: None }); - }; - let already_tokened = parts.extensions.get::(); - if let Some(tokened) = already_tokened { - Ok(OptTokened { token: Some(tokened.token.clone()) }) - } else { - let token_o = extract_token(parts, state).await; - Ok(OptTokened { token: token_o }) - } - } -} - pub async fn maybe_refresh_folders( path: &str, w_id: &str, @@ -629,94 +153,6 @@ pub async fn maybe_refresh_folders( } } -#[async_trait] -impl FromRequestParts for ApiAuthed -where - S: Send + Sync, -{ - type Rejection = (StatusCode, String); - - async fn from_request_parts( - parts: &mut Parts, - state: &S, - ) -> std::result::Result { - if parts.method == http::Method::OPTIONS { - return Ok(ApiAuthed { - email: "".to_owned(), - username: "".to_owned(), - is_admin: false, - is_operator: false, - groups: Vec::new(), - folders: Vec::new(), - scopes: None, - username_override: None, - }); - }; - let already_authed = parts.extensions.get::(); - if let Some(authed) = already_authed { - Ok(authed.clone()) - } else { - let already_tokened = parts.extensions.get::(); - let token_o = if let Some(token) = already_tokened { - Some(token.token.clone()) - } else { - extract_token(parts, state).await - }; - let original_uri = OriginalUri::from_request_parts(parts, state) - .await - .ok() - .map(|x| x.0) - .unwrap_or_default(); - let path_vec: Vec<&str> = original_uri.path().split("/").collect(); - - let workspace_id = if path_vec.len() >= 4 && path_vec[0] == "" && path_vec[2] == "w" { - Some(path_vec[3].to_owned()) - } else { - if path_vec.len() >= 5 - && path_vec[0] == "" - && path_vec[2] == "srch" - && path_vec[3] == "w" - { - Some(path_vec[4].to_string()) - } else { - None - } - }; - if let Some(token) = token_o { - if let Ok(Extension(cache)) = - Extension::>::from_request_parts(parts, state).await - { - if let Some(authed) = cache.get_authed(workspace_id.clone(), &token).await { - parts.extensions.insert(authed.clone()); - if authed.scopes.as_ref().is_some_and(|scopes| { - scopes - .iter() - .any(|s| s.starts_with("jobs:") || s.starts_with("run:")) - }) && (path_vec.len() < 3 - || (path_vec[4] != "jobs" && path_vec[4] != "jobs_u")) - { - BRUTE_FORCE_COUNTER.increment().await; - return Err(( - StatusCode::UNAUTHORIZED, - format!("Unauthorized scoped token: {:?}", authed.scopes), - )); - } - Span::current().record("username", &authed.username.as_str()); - Span::current().record("email", &authed.email); - - if let Some(workspace_id) = workspace_id { - Span::current().record("workspace_id", &workspace_id); - } - return Ok(authed); - } - } - } - BRUTE_FORCE_COUNTER.increment().await; - Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned())) - } - } -} - pub fn check_scopes(authed: &ApiAuthed, required: F) -> error::Result<()> where F: FnOnce() -> String, @@ -2514,7 +1950,7 @@ async fn get_all_runnables( Extension(db): Extension, authed: ApiAuthed, Tokened { token }: Tokened, - Extension(cache): Extension>, + Extension(cache): Extension>, ) -> JsonResult> { let mut tx = db.clone().begin(&authed).await?; let mut runnables = Vec::new(); diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 1519b9d46f..433450b6e8 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -14,10 +14,7 @@ use crate::users_ee::send_email_if_possible; use crate::utils::get_instance_username_or_create_pending; use crate::BASE_URL; use crate::{ - apps::AppWithLastVersion, db::DB, - folders::Folder, - resources::{Resource, ResourceType}, users::{WorkspaceInvite, VALID_USERNAME}, utils::require_super_admin, webhook_util::WebhookShared, @@ -25,14 +22,11 @@ use crate::{ use axum::{ extract::{Extension, Path, Query}, - response::IntoResponse, routing::{delete, get, post}, Json, Router, }; use chrono::Utc; -use http::HeaderName; -use itertools::Itertools; use regex::Regex; use uuid::Uuid; @@ -40,22 +34,18 @@ use windmill_audit::audit_ee::audit_log; use windmill_audit::ActionKind; use windmill_common::db::UserDB; use windmill_common::s3_helpers::LargeFileStorage; -use windmill_common::schedule::Schedule; use windmill_common::users::username_to_permissioned_as; use windmill_common::variables::build_crypt; -use windmill_common::worker::{to_raw_value, CLOUD_HOSTED}; +use windmill_common::worker::to_raw_value; #[cfg(feature = "enterprise")] use windmill_common::workspaces::WorkspaceDeploymentUISettings; #[cfg(feature = "enterprise")] use windmill_common::workspaces::WorkspaceGitSyncSettings; use windmill_common::{ - error::{to_anyhow, Error, JsonResult, Result}, - flows::Flow, + error::{Error, JsonResult, Result}, global_settings::AUTOMATE_USERNAME_CREATION_SETTING, oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH, - scripts::{Schema, Script, ScriptLang}, utils::{paginate, rd_string, require_admin, Pagination}, - variables::ExportableListableVariable, }; use windmill_git_sync::handle_deployment_metadata; @@ -63,13 +53,9 @@ use windmill_git_sync::handle_deployment_metadata; use windmill_common::utils::require_admin_or_devops; use crate::variables::{decrypt, encrypt}; -use hyper::{header, StatusCode}; +use hyper::StatusCode; use serde::{Deserialize, Serialize}; -use serde_json::Value; use sqlx::{FromRow, Postgres, Transaction}; -use tempfile::TempDir; -use tokio::fs::File; -use tokio_util::io::ReaderStream; use windmill_common::oauth2::InstanceEvent; use windmill_common::utils::not_found_if_none; @@ -95,7 +81,7 @@ pub fn workspaced_service() -> Router { .route("/edit_webhook", post(edit_webhook)) .route("/edit_auto_invite", post(edit_auto_invite)) .route("/edit_deploy_to", post(edit_deploy_to)) - .route("/tarball", get(tarball_workspace)) + .route("/tarball", get(crate::workspaces_export::tarball_workspace)) .route("/is_premium", get(is_premium)) .route("/edit_copilot_config", post(edit_copilot_config)) .route("/get_copilot_info", get(get_copilot_info)) @@ -120,7 +106,10 @@ pub fn workspaced_service() -> Router { .route("/leave", post(leave_workspace)) .route("/get_workspace_name", get(get_workspace_name)) .route("/change_workspace_name", post(change_workspace_name)) - .route("/change_workspace_id", post(change_workspace_id)) + .route( + "/change_workspace_id", + post(crate::workspaces_extra::change_workspace_id), + ) .route("/usage", get(get_usage)) .route("/used_triggers", get(get_used_triggers)) .route("/critical_alerts", get(get_critical_alerts)) @@ -152,7 +141,10 @@ pub fn global_service() -> Router { .route("/exists_username", post(exists_username)) .route("/allowed_domain_auto_invite", get(is_allowed_auto_domain)) .route("/unarchive/:workspace", post(unarchive_workspace)) - .route("/delete/:workspace", delete(delete_workspace)) + .route( + "/delete/:workspace", + delete(crate::workspaces_extra::delete_workspace), + ) .route( "/create_workspace_require_superadmin", get(create_workspace_require_superadmin), @@ -1679,141 +1671,6 @@ async fn unarchive_workspace( Ok(format!("Unarchived workspace {}", &w_id)) } -async fn delete_workspace( - Extension(db): Extension, - Path(w_id): Path, - authed: ApiAuthed, -) -> Result { - let w_id = match w_id.as_str() { - "starter" => Err(Error::BadRequest( - "starter workspace cannot be deleted".to_string(), - )), - "admins" => Err(Error::BadRequest( - "admins workspace cannot be deleted".to_string(), - )), - _ => Ok(w_id), - }?; - let mut tx = db.begin().await?; - require_super_admin(&db, &authed.email).await?; - - sqlx::query!("DELETE FROM dependency_map WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM queue WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM capture WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM draft WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM script WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM app WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM raw_app WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM input WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM variable WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - sqlx::query!("DELETE FROM resource WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM schedule WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM completed_job WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM job_stats WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "DELETE FROM deployment_metadata WHERE workspace_id = $1", - &w_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM usr WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM resource_type WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "DELETE FROM workspace_invite WHERE workspace_id = $1", - &w_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM usr_to_group WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM group_ WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM folder WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM account WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM workspace_key WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "DELETE FROM workspace_settings WHERE workspace_id = $1", - &w_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM token WHERE workspace_id = $1", &w_id) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM workspace WHERE id = $1", &w_id) - .execute(&mut *tx) - .await?; - - audit_log( - &mut *tx, - &authed, - "workspaces.delete", - ActionKind::Delete, - &w_id, - Some(&authed.email), - None, - ) - .await?; - tx.commit().await?; - - Ok(format!("Deleted workspace {}", &w_id)) -} - async fn invite_user( ApiAuthed { username, is_admin, .. }: ApiAuthed, Extension(db): Extension, @@ -2066,637 +1923,6 @@ async fn exists_username( Ok("valid username".to_string()) } -#[derive(Serialize)] -struct ScriptMetadata { - summary: String, - description: String, - schema: Option, - lock: Option, - kind: String, - #[serde(skip_serializing_if = "Option::is_none")] - envs: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - concurrent_limit: Option, - #[serde(skip_serializing_if = "Option::is_none")] - concurrency_time_window_s: Option, - #[serde(skip_serializing_if = "Option::is_none")] - cache_ttl: Option, - #[serde(skip_serializing_if = "Option::is_none")] - dedicated_worker: Option, - #[serde(skip_serializing_if = "is_none_or_false")] - ws_error_handler_muted: Option, - #[serde(skip_serializing_if = "Option::is_none")] - priority: Option, - #[serde(skip_serializing_if = "Option::is_none")] - tag: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub timeout: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub delete_after_use: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub restart_unless_cancelled: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub visible_to_runner_only: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub no_main_func: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub codebase: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub concurrency_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub has_preprocessor: Option, -} - -pub fn is_none_or_false(val: &Option) -> bool { - match val { - Some(val) => !val, - None => true, - } -} - -enum ArchiveImpl { - #[cfg(feature = "zip")] - Zip(async_zip::tokio::write::ZipFileWriter), - Tar(tokio_tar::Builder), -} - -impl ArchiveImpl { - async fn write_to_archive(&mut self, content: &str, path: &str) -> Result<()> { - match self { - ArchiveImpl::Tar(t) => { - let bytes = content.as_bytes(); - let mut header = tokio_tar::Header::new_gnu(); - header.set_size(bytes.len() as u64); - header.set_mtime(0); - header.set_uid(0); - header.set_gid(0); - header.set_mode(0o777); - header.set_cksum(); - t.append_data(&mut header, path, bytes).await?; - } - #[cfg(feature = "zip")] - ArchiveImpl::Zip(z) => { - let header = - async_zip::ZipEntryBuilder::new(path.into(), async_zip::Compression::Deflate) - .last_modification_date(Default::default()) - .unix_permissions(0o777) - .build(); - z.write_entry_whole(header, content.as_bytes()) - .await - .map_err(to_anyhow)?; - } - } - Ok(()) - } - async fn finish(self) -> Result<()> { - match self { - ArchiveImpl::Tar(t) => t.into_inner().await?, - #[cfg(feature = "zip")] - ArchiveImpl::Zip(z) => z.close().await.map_err(to_anyhow)?.into_inner(), - } - .sync_all() - .await?; - - Ok(()) - } -} - -#[derive(Deserialize)] -struct ArchiveQueryParams { - archive_type: Option, - plain_secret: Option, - plain_secrets: Option, - skip_secrets: Option, - skip_variables: Option, - skip_resources: Option, - include_schedules: Option, - include_users: Option, - include_groups: Option, - include_settings: Option, - include_key: Option, - default_ts: Option, -} - -#[inline] -pub fn to_string_without_metadata( - value: &T, - preserve_extra_perms: bool, - ignore_keys: Option>, -) -> Result -where - T: ?Sized + Serialize, -{ - let mut value = serde_json::to_value(value).map_err(to_anyhow)?; - value - .as_object_mut() - .map(|obj| { - let keys = [ - vec![ - "workspace_id", - "path", - "name", - "versions", - "id", - "created_at", - "updated_at", - "created_by", - "updated_by", - "edited_at", - "edited_by", - "archived", - "has_draft", - "draft_only", - "error", - ], - ignore_keys.unwrap_or(vec![]), - ] - .concat(); - - for key in keys { - if obj.contains_key(key) { - obj.remove(key); - } - } - - if let Some(o2) = obj.get_mut("policy").and_then(|x| x.as_object_mut()) { - o2.remove("on_behalf_of"); - o2.remove("on_behalf_of_email"); - } - if !preserve_extra_perms && obj.contains_key("extra_perms") { - obj.remove("extra_perms"); - } - - serde_json::to_string_pretty(&obj).ok() - }) - .flatten() - .ok_or_else(|| Error::BadRequest("Impossible to serialize value".to_string())) -} - -#[derive(Serialize)] -struct SimplifiedUser { - username: String, - role: String, - disabled: bool, - email: String, -} - -#[derive(Serialize)] -struct SimplifiedGroup { - name: String, - summary: Option, - members: Vec, - admins: Vec, -} - -#[derive(Serialize)] -struct SimplifiedSettings { - // slack_team_id: Option, - // slack_name: Option, - // slack_command_script: Option, - // slack_email: Option, - auto_invite_enabled: bool, - auto_invite_as: String, - auto_invite_mode: String, - webhook: Option, - deploy_to: Option, - error_handler: Option, - error_handler_extra_args: Option, - error_handler_muted_on_cancel: bool, - ai_resource: Option, - code_completion_enabled: bool, - large_file_storage: Option, - git_sync: Option, - default_app: Option, - default_scripts: Option, - name: String, -} - -async fn tarball_workspace( - authed: ApiAuthed, - Extension(user_db): Extension, - Extension(db): Extension, - Path(w_id): Path, - Query(ArchiveQueryParams { - archive_type, - plain_secret, - plain_secrets, - skip_resources, - skip_secrets, - skip_variables, - include_schedules, - include_users, - include_groups, - include_settings, - include_key, - default_ts, - }): Query, -) -> Result<([(HeaderName, String); 2], impl IntoResponse)> { - // require_admin(authed.is_admin, &authed.username)?; - - let mut tx = user_db.begin(&authed).await?; - - let tmp_dir = TempDir::new_in("/tmp/windmill/")?; - - let name = match archive_type.as_deref() { - Some("tar") | None => Ok(format!("windmill-{w_id}.tar")), - Some("zip") => Ok(format!("windmill-{w_id}.zip")), - Some(t) => Err(Error::BadRequest(format!("Invalid Archive Type {t}"))), - }?; - let file_path = tmp_dir.path().join(&name); - let mut archive = match archive_type.as_deref() { - Some("tar") | None => { - let file = File::create(&file_path).await?; - Ok(ArchiveImpl::Tar(tokio_tar::Builder::new(file))) - } - #[cfg(feature = "zip")] - Some("zip") => { - let file = tokio::fs::File::create(&file_path).await?; - Ok(ArchiveImpl::Zip( - async_zip::tokio::write::ZipFileWriter::with_tokio(file), - )) - } - Some(t) => Err(Error::BadRequest(format!("Invalid Archive Type {t}"))), - }?; - { - let folders = sqlx::query_as::<_, Folder>("SELECT * FROM folder WHERE workspace_id = $1") - .bind(&w_id) - .fetch_all(&mut *tx) - .await?; - - for folder in folders { - archive - .write_to_archive( - &to_string_without_metadata(&folder, true, None).unwrap(), - &format!("f/{}/folder.meta.json", folder.name), - ) - .await?; - } - } - - { - let scripts = sqlx::query_as::<_, Script>( - "SELECT * FROM script as o WHERE workspace_id = $1 AND archived = false - AND created_at = (select max(created_at) from script where path = o.path AND \ - workspace_id = $1)", - ) - .bind(&w_id) - .fetch_all(&mut *tx) - .await?; - - for script in scripts { - let ext = match script.language { - ScriptLang::Python3 => "py", - ScriptLang::Deno => { - if default_ts.as_ref().is_some_and(|x| x == "bun") { - "deno.ts" - } else { - "ts" - } - } - ScriptLang::Go => "go", - ScriptLang::Bash => "sh", - ScriptLang::Powershell => "ps1", - ScriptLang::Postgresql => "pg.sql", - ScriptLang::Mysql => "my.sql", - ScriptLang::Bigquery => "bq.sql", - ScriptLang::Snowflake => "sf.sql", - ScriptLang::Mssql => "ms.sql", - ScriptLang::Graphql => "gql", - ScriptLang::Nativets => "fetch.ts", - ScriptLang::Bun | ScriptLang::Bunnative => { - if default_ts.as_ref().is_some_and(|x| x == "bun") { - "ts" - } else { - "bun.ts" - } - } - ScriptLang::Php => "php", - ScriptLang::Rust => "rs", - ScriptLang::Ansible => "playbook.yml", - ScriptLang::CSharp => "cs", - }; - archive - .write_to_archive(&script.content, &format!("{}.{}", script.path, ext)) - .await?; - - let metadata = ScriptMetadata { - summary: script.summary, - description: script.description, - schema: script.schema, - kind: script.kind.to_string(), - lock: script.lock, - envs: script.envs, - concurrent_limit: script.concurrent_limit, - concurrency_time_window_s: script.concurrency_time_window_s, - cache_ttl: script.cache_ttl, - dedicated_worker: script.dedicated_worker, - ws_error_handler_muted: script.ws_error_handler_muted, - priority: script.priority, - tag: script.tag, - timeout: script.timeout, - delete_after_use: script.delete_after_use, - restart_unless_cancelled: script.restart_unless_cancelled, - visible_to_runner_only: script.visible_to_runner_only, - no_main_func: script.no_main_func, - codebase: script.codebase, - concurrency_key: script.concurrency_key, - has_preprocessor: script.has_preprocessor, - }; - let metadata_str = serde_json::to_string_pretty(&metadata).unwrap(); - archive - .write_to_archive(&metadata_str, &format!("{}.script.json", script.path)) - .await?; - } - } - - if !skip_resources.unwrap_or(false) { - let resources = sqlx::query_as!( - Resource, - "SELECT * FROM resource WHERE workspace_id = $1 AND resource_type != 'state' AND resource_type != 'cache'", - &w_id - ) - .fetch_all(&mut *tx) - .await?; - - for resource in resources { - let resource_str = &to_string_without_metadata(&resource, false, None).unwrap(); - archive - .write_to_archive(&resource_str, &format!("{}.resource.json", resource.path)) - .await?; - } - } - - if !skip_resources.unwrap_or(false) { - let resource_types = sqlx::query_as!( - ResourceType, - "SELECT * FROM resource_type WHERE workspace_id = $1", - &w_id - ) - .fetch_all(&mut *tx) - .await?; - - for resource_type in resource_types { - let resource_str = &to_string_without_metadata(&resource_type, false, None).unwrap(); - archive - .write_to_archive( - &resource_str, - &format!("{}.resource-type.json", resource_type.name), - ) - .await?; - } - } - - { - let flows = sqlx::query_as::<_, Flow>( - "SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by - FROM flow - LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] - WHERE flow.workspace_id = $1 AND flow.archived = false", - ) - .bind(&w_id) - .fetch_all(&mut *tx) - .await?; - - for flow in flows { - let flow_str = &to_string_without_metadata(&flow, false, None).unwrap(); - archive - .write_to_archive(&flow_str, &format!("{}.flow.json", flow.path)) - .await?; - } - } - - if !skip_variables.unwrap_or(false) { - let variables = - sqlx::query_as::<_, ExportableListableVariable>(if !skip_secrets.unwrap_or(false) { - "SELECT * FROM variable WHERE workspace_id = $1 AND expires_at IS NULL" - } else { - "SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL" - }) - .bind(&w_id) - .fetch_all(&mut *tx) - .await?; - - let mc = build_crypt(&db, &w_id).await?; - - for mut var in variables { - if plain_secret.or(plain_secrets).unwrap_or(false) - && var.value.is_some() - && var.is_secret - { - var.value = Some(decrypt(&mc, var.value.unwrap())?); - } - let var_str = &to_string_without_metadata(&var, false, None).unwrap(); - archive - .write_to_archive(&var_str, &format!("{}.variable.json", var.path)) - .await?; - } - } - - { - let apps = sqlx::query_as::<_, AppWithLastVersion>( - "SELECT app.id, app.path, app.summary, app.versions, app.policy, app.custom_path, - app.extra_perms, app_version.value, - app_version.created_at, app_version.created_by from app, app_version - WHERE app.workspace_id = $1 AND app_version.id = app.versions[array_upper(app.versions, 1)]", - ) - .bind(&w_id) - .fetch_all(&mut *tx) - .await?; - - for app in apps { - let app_str = &to_string_without_metadata(&app, false, None).unwrap(); - archive - .write_to_archive(&app_str, &format!("{}.app.json", app.path)) - .await?; - } - } - - if include_schedules.unwrap_or(false) { - let schedules = sqlx::query_as::<_, Schedule>( - "SELECT * FROM schedule - WHERE workspace_id = $1", - ) - .bind(&w_id) - .fetch_all(&mut *tx) - .await?; - - for schedule in schedules { - let app_str = &to_string_without_metadata(&schedule, false, None).unwrap(); - archive - .write_to_archive(&app_str, &format!("{}.schedule.json", schedule.path)) - .await?; - } - } - - if include_users.unwrap_or(false) { - let users = sqlx::query!( - "SELECT * FROM usr - WHERE workspace_id = $1", - &w_id - ) - .fetch_all(&mut *tx) - .await?; - - for user in users { - let user = SimplifiedUser { - username: user.username, - role: if user.is_admin { - "admin".to_string() - } else if user.operator { - "operator".to_string() - } else { - "developer".to_string() - }, - disabled: user.disabled, - email: user.email, - }; - let user_str = &to_string_without_metadata( - &user, - false, - Some(vec!["is_admin", "operator", "email"]), - ) - .unwrap(); - archive - .write_to_archive(&user_str, &format!("users/{}.user.json", user.email)) - .await?; - } - } - - if include_groups.unwrap_or(false) { - let groups = sqlx::query!( - r#"SELECT g_.workspace_id, name, summary, extra_perms, array_agg(u2g.usr) filter (where u2g.usr is not null) as members - FROM usr u - JOIN usr_to_group u2g ON u2g.usr = u.username AND u2g.workspace_id = u.workspace_id - RIGHT JOIN group_ g_ ON g_.workspace_id = u.workspace_id AND g_.name = u2g.group_ - WHERE g_.workspace_id = $1 AND g_.name != 'all' - GROUP BY g_.workspace_id, name, summary, extra_perms"#, - &w_id - ) - .fetch_all(&mut *tx) - .await?; - - for group in groups { - let extra_perms: HashMap = serde_json::from_value(group.extra_perms) - .map_err(|e| { - Error::InternalErr(format!( - "Error parsing extra_perms for group {}: {}", - group.name, e - )) - })?; - tracing::info!("{:?}", extra_perms); - let members = group.members.unwrap_or(vec![]); - let admins: Vec = extra_perms - .iter() - .filter_map(|(k, v)| { - // only consider extra_perms that concern actual members of the group - if members.contains(&k[2..].to_string()) && *v { - Some(k.clone()) - } else { - None - } - }) - .sorted() - .collect(); - let group = SimplifiedGroup { - name: group.name, - summary: group.summary, - members: members - .iter() - .filter_map(|x| { - // remove members that are also admins as they are already in the admins list - let full_name = format!("u/{}", x); - if !admins.contains(&full_name) { - Some(full_name) - } else { - None - } - }) - .collect(), - admins, - }; - - let group_str = &to_string_without_metadata(&group, true, None).unwrap(); - archive - .write_to_archive(&group_str, &format!("groups/{}.group.json", group.name)) - .await?; - } - } - - if include_settings.unwrap_or(false) { - let settings = sqlx::query_as!( - SimplifiedSettings, - r#"SELECT - -- slack_team_id, - -- slack_name, - -- slack_command_script, - -- CASE WHEN slack_email = 'missing@email.xyz' THEN NULL ELSE slack_email END AS slack_email, - auto_invite_domain IS NOT NULL AS "auto_invite_enabled!", - CASE WHEN auto_invite_operator IS TRUE THEN 'operator' ELSE 'developer' END AS "auto_invite_as!", - CASE WHEN auto_add IS TRUE THEN 'add' ELSE 'invite' END AS "auto_invite_mode!", - webhook, - deploy_to, - error_handler, - ai_resource, - code_completion_enabled, - error_handler_extra_args, - error_handler_muted_on_cancel, - large_file_storage, - git_sync, - default_app, - default_scripts, - workspace.name - FROM workspace_settings - LEFT JOIN workspace ON workspace.id = workspace_settings.workspace_id - WHERE workspace_id = $1"#, - &w_id - ).fetch_one(&mut *tx).await?; - - let settings_str = serde_json::to_value(settings) - .map(|v| serde_json::to_string_pretty(&v).ok()) - .ok() - .flatten() - .ok_or_else(|| Error::InternalErr("Error serializing settings".to_string()))?; - - archive - .write_to_archive(&settings_str, "settings.json") - .await?; - } - - if include_key.unwrap_or(false) { - let key = sqlx::query_scalar!( - "SELECT key FROM workspace_key WHERE workspace_id = $1", - &w_id - ) - .fetch_one(&mut *tx) - .await?; - - let key_json = serde_json::to_value(key) - .map(|v| serde_json::to_string_pretty(&v).ok()) - .ok() - .flatten() - .ok_or_else(|| Error::InternalErr("Error serializing enryption key".to_string()))?; - archive - .write_to_archive(&key_json, "encryption_key.json") - .await?; - } - - archive.finish().await?; - - let file = tokio::fs::File::open(&file_path).await?; - - let stream = ReaderStream::new(file); - let body = axum::body::Body::from_stream(stream); - - let headers = [ - (header::CONTENT_TYPE, "application/x-tar".to_string()), - ( - header::CONTENT_DISPOSITION, - format!("attachment; filename=\"{name}\""), - ), - ]; - Ok((headers, body)) -} - async fn get_workspace_name( authed: ApiAuthed, Path(w_id): Path, @@ -2751,355 +1977,6 @@ async fn change_workspace_name( Ok(format!("updated workspace name to {}", &rw.new_name)) } -#[derive(Deserialize)] -struct ChangeWorkspaceId { - new_id: String, - new_name: String, -} -async fn change_workspace_id( - authed: ApiAuthed, - Path(old_id): Path, - Extension(db): Extension, - Json(rw): Json, -) -> Result { - if *CLOUD_HOSTED { - return Err(Error::BadRequest( - "This feature is not available on the cloud".to_string(), - )); - } - - if *CREATE_WORKSPACE_REQUIRE_SUPERADMIN { - require_super_admin(&db, &authed.email).await?; - } else { - require_admin(authed.is_admin, &authed.username)?; - } - - let mut tx = db.begin().await?; - - let workspace_conflict = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM workspace WHERE id = $1)", - &rw.new_id, - ) - .fetch_one(&mut *tx) - .await? - .unwrap_or(false); - - if workspace_conflict { - return Err(Error::BadRequest(format!( - "workspace id {} already used", - &rw.new_id - ))); - } - - // duplicate workspace with new id name - sqlx::query!( - "INSERT INTO workspace SELECT $1, $2, owner, deleted, premium FROM workspace WHERE id = $3", - &rw.new_id, - &rw.new_name, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE account SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE app SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE audit SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE capture SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE completed_job SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE dependency_map SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE deployment_metadata SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE draft SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE favorite SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "INSERT INTO flow - (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at) - SELECT $1, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at - FROM flow WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE flow_version SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &old_id) - .execute(&mut *tx) - .await?; - - // have to duplicate group_ with new workspace id because of foreign key constraint - sqlx::query!( - "INSERT INTO group_ SELECT $1, name, summary, extra_perms FROM group_ WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE usr_to_group SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - // then delete old group_ - sqlx::query!("DELETE FROM group_ WHERE workspace_id = $1", &old_id) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE folder SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE input SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE job_logs SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE job_stats SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE queue SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE job SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE raw_app SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE resource SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE resource_type SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE schedule SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE script SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE token SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE usage SET id = $1 WHERE id = $2 AND is_workspace = true", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE usr SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE variable SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE workspace_env SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE workspace_invite SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE workspace_key SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - sqlx::query!( - "UPDATE workspace_settings SET workspace_id = $1 WHERE workspace_id = $2", - &rw.new_id, - &old_id - ) - .execute(&mut *tx) - .await?; - - // delete old workspace - sqlx::query!("DELETE FROM workspace WHERE id = $1", &old_id) - .execute(&mut *tx) - .await?; - - audit_log( - &mut *tx, - &authed, - "workspace.change_workspace_id", - ActionKind::Update, - &rw.new_id, - Some(&authed.email), - None, - ) - .await?; - tx.commit().await?; - Ok(format!( - "updated workspace from {} to {}", - &old_id, &rw.new_id - )) -} - async fn get_usage(Extension(db): Extension, Path(w_id): Path) -> Result { let usage = sqlx::query_scalar!( " diff --git a/backend/windmill-api/src/workspaces_export.rs b/backend/windmill-api/src/workspaces_export.rs new file mode 100644 index 0000000000..774bc97053 --- /dev/null +++ b/backend/windmill-api/src/workspaces_export.rs @@ -0,0 +1,676 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +use std::collections::HashMap; + +use crate::db::ApiAuthed; + +use crate::{ + apps::AppWithLastVersion, + db::DB, + folders::Folder, + resources::{Resource, ResourceType}, +}; + +use axum::{ + extract::{Extension, Path, Query}, + response::IntoResponse, +}; + +use http::HeaderName; +use itertools::Itertools; + +use windmill_common::db::UserDB; +use windmill_common::schedule::Schedule; +use windmill_common::variables::build_crypt; + +use windmill_common::{ + error::{to_anyhow, Error, Result}, + flows::Flow, + scripts::{Schema, Script, ScriptLang}, + variables::ExportableListableVariable, +}; + +use crate::variables::decrypt; +use hyper::header; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tempfile::TempDir; +use tokio::fs::File; +use tokio_util::io::ReaderStream; + +#[derive(Serialize)] +struct ScriptMetadata { + summary: String, + description: String, + schema: Option, + lock: Option, + kind: String, + #[serde(skip_serializing_if = "Option::is_none")] + envs: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + concurrency_time_window_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] + cache_ttl: Option, + #[serde(skip_serializing_if = "Option::is_none")] + dedicated_worker: Option, + #[serde(skip_serializing_if = "is_none_or_false")] + ws_error_handler_muted: Option, + #[serde(skip_serializing_if = "Option::is_none")] + priority: Option, + #[serde(skip_serializing_if = "Option::is_none")] + tag: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub timeout: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub delete_after_use: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub restart_unless_cancelled: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub visible_to_runner_only: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub no_main_func: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub codebase: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub has_preprocessor: Option, +} + +pub fn is_none_or_false(val: &Option) -> bool { + match val { + Some(val) => !val, + None => true, + } +} + +enum ArchiveImpl { + #[cfg(feature = "zip")] + Zip(async_zip::tokio::write::ZipFileWriter), + Tar(tokio_tar::Builder), +} + +impl ArchiveImpl { + async fn write_to_archive(&mut self, content: &str, path: &str) -> Result<()> { + match self { + ArchiveImpl::Tar(t) => { + let bytes = content.as_bytes(); + let mut header = tokio_tar::Header::new_gnu(); + header.set_size(bytes.len() as u64); + header.set_mtime(0); + header.set_uid(0); + header.set_gid(0); + header.set_mode(0o777); + header.set_cksum(); + t.append_data(&mut header, path, bytes).await?; + } + #[cfg(feature = "zip")] + ArchiveImpl::Zip(z) => { + let header = + async_zip::ZipEntryBuilder::new(path.into(), async_zip::Compression::Deflate) + .last_modification_date(Default::default()) + .unix_permissions(0o777) + .build(); + z.write_entry_whole(header, content.as_bytes()) + .await + .map_err(to_anyhow)?; + } + } + Ok(()) + } + async fn finish(self) -> Result<()> { + match self { + ArchiveImpl::Tar(t) => t.into_inner().await?, + #[cfg(feature = "zip")] + ArchiveImpl::Zip(z) => z.close().await.map_err(to_anyhow)?.into_inner(), + } + .sync_all() + .await?; + + Ok(()) + } +} + +#[derive(Deserialize)] +pub(crate) struct ArchiveQueryParams { + archive_type: Option, + plain_secret: Option, + plain_secrets: Option, + skip_secrets: Option, + skip_variables: Option, + skip_resources: Option, + include_schedules: Option, + include_users: Option, + include_groups: Option, + include_settings: Option, + include_key: Option, + default_ts: Option, +} + +#[inline] +pub fn to_string_without_metadata( + value: &T, + preserve_extra_perms: bool, + ignore_keys: Option>, +) -> Result +where + T: ?Sized + Serialize, +{ + let mut value = serde_json::to_value(value).map_err(to_anyhow)?; + value + .as_object_mut() + .map(|obj| { + let keys = [ + vec![ + "workspace_id", + "path", + "name", + "versions", + "id", + "created_at", + "updated_at", + "created_by", + "updated_by", + "edited_at", + "edited_by", + "archived", + "has_draft", + "draft_only", + "error", + ], + ignore_keys.unwrap_or(vec![]), + ] + .concat(); + + for key in keys { + if obj.contains_key(key) { + obj.remove(key); + } + } + + if let Some(o2) = obj.get_mut("policy").and_then(|x| x.as_object_mut()) { + o2.remove("on_behalf_of"); + o2.remove("on_behalf_of_email"); + } + if !preserve_extra_perms && obj.contains_key("extra_perms") { + obj.remove("extra_perms"); + } + + serde_json::to_string_pretty(&obj).ok() + }) + .flatten() + .ok_or_else(|| Error::BadRequest("Impossible to serialize value".to_string())) +} + +#[derive(Serialize)] +struct SimplifiedUser { + username: String, + role: String, + disabled: bool, + email: String, +} + +#[derive(Serialize)] +struct SimplifiedGroup { + name: String, + summary: Option, + members: Vec, + admins: Vec, +} + +#[derive(Serialize)] +struct SimplifiedSettings { + // slack_team_id: Option, + // slack_name: Option, + // slack_command_script: Option, + // slack_email: Option, + auto_invite_enabled: bool, + auto_invite_as: String, + auto_invite_mode: String, + webhook: Option, + deploy_to: Option, + error_handler: Option, + error_handler_extra_args: Option, + error_handler_muted_on_cancel: bool, + ai_resource: Option, + code_completion_enabled: bool, + large_file_storage: Option, + git_sync: Option, + default_app: Option, + default_scripts: Option, + name: String, +} + +pub(crate) async fn tarball_workspace( + authed: ApiAuthed, + Extension(user_db): Extension, + Extension(db): Extension, + Path(w_id): Path, + Query(ArchiveQueryParams { + archive_type, + plain_secret, + plain_secrets, + skip_resources, + skip_secrets, + skip_variables, + include_schedules, + include_users, + include_groups, + include_settings, + include_key, + default_ts, + }): Query, +) -> Result<([(HeaderName, String); 2], impl IntoResponse)> { + // require_admin(authed.is_admin, &authed.username)?; + + let mut tx = user_db.begin(&authed).await?; + + let tmp_dir = TempDir::new_in("/tmp/windmill/")?; + + let name = match archive_type.as_deref() { + Some("tar") | None => Ok(format!("windmill-{w_id}.tar")), + Some("zip") => Ok(format!("windmill-{w_id}.zip")), + Some(t) => Err(Error::BadRequest(format!("Invalid Archive Type {t}"))), + }?; + let file_path = tmp_dir.path().join(&name); + let mut archive = match archive_type.as_deref() { + Some("tar") | None => { + let file = File::create(&file_path).await?; + Ok(ArchiveImpl::Tar(tokio_tar::Builder::new(file))) + } + #[cfg(feature = "zip")] + Some("zip") => { + let file = tokio::fs::File::create(&file_path).await?; + Ok(ArchiveImpl::Zip( + async_zip::tokio::write::ZipFileWriter::with_tokio(file), + )) + } + Some(t) => Err(Error::BadRequest(format!("Invalid Archive Type {t}"))), + }?; + { + let folders = sqlx::query_as::<_, Folder>("SELECT * FROM folder WHERE workspace_id = $1") + .bind(&w_id) + .fetch_all(&mut *tx) + .await?; + + for folder in folders { + archive + .write_to_archive( + &to_string_without_metadata(&folder, true, None).unwrap(), + &format!("f/{}/folder.meta.json", folder.name), + ) + .await?; + } + } + + { + let scripts = sqlx::query_as::<_, Script>( + "SELECT * FROM script as o WHERE workspace_id = $1 AND archived = false + AND created_at = (select max(created_at) from script where path = o.path AND \ + workspace_id = $1)", + ) + .bind(&w_id) + .fetch_all(&mut *tx) + .await?; + + for script in scripts { + let ext = match script.language { + ScriptLang::Python3 => "py", + ScriptLang::Deno => { + if default_ts.as_ref().is_some_and(|x| x == "bun") { + "deno.ts" + } else { + "ts" + } + } + ScriptLang::Go => "go", + ScriptLang::Bash => "sh", + ScriptLang::Powershell => "ps1", + ScriptLang::Postgresql => "pg.sql", + ScriptLang::Mysql => "my.sql", + ScriptLang::Bigquery => "bq.sql", + ScriptLang::Snowflake => "sf.sql", + ScriptLang::Mssql => "ms.sql", + ScriptLang::Graphql => "gql", + ScriptLang::Nativets => "fetch.ts", + ScriptLang::Bun | ScriptLang::Bunnative => { + if default_ts.as_ref().is_some_and(|x| x == "bun") { + "ts" + } else { + "bun.ts" + } + } + ScriptLang::Php => "php", + ScriptLang::Rust => "rs", + ScriptLang::Ansible => "playbook.yml", + ScriptLang::CSharp => "cs", + }; + archive + .write_to_archive(&script.content, &format!("{}.{}", script.path, ext)) + .await?; + + let metadata = ScriptMetadata { + summary: script.summary, + description: script.description, + schema: script.schema, + kind: script.kind.to_string(), + lock: script.lock, + envs: script.envs, + concurrent_limit: script.concurrent_limit, + concurrency_time_window_s: script.concurrency_time_window_s, + cache_ttl: script.cache_ttl, + dedicated_worker: script.dedicated_worker, + ws_error_handler_muted: script.ws_error_handler_muted, + priority: script.priority, + tag: script.tag, + timeout: script.timeout, + delete_after_use: script.delete_after_use, + restart_unless_cancelled: script.restart_unless_cancelled, + visible_to_runner_only: script.visible_to_runner_only, + no_main_func: script.no_main_func, + codebase: script.codebase, + concurrency_key: script.concurrency_key, + has_preprocessor: script.has_preprocessor, + }; + let metadata_str = serde_json::to_string_pretty(&metadata).unwrap(); + archive + .write_to_archive(&metadata_str, &format!("{}.script.json", script.path)) + .await?; + } + } + + if !skip_resources.unwrap_or(false) { + let resources = sqlx::query_as!( + Resource, + "SELECT * FROM resource WHERE workspace_id = $1 AND resource_type != 'state' AND resource_type != 'cache'", + &w_id + ) + .fetch_all(&mut *tx) + .await?; + + for resource in resources { + let resource_str = &to_string_without_metadata(&resource, false, None).unwrap(); + archive + .write_to_archive(&resource_str, &format!("{}.resource.json", resource.path)) + .await?; + } + } + + if !skip_resources.unwrap_or(false) { + let resource_types = sqlx::query_as!( + ResourceType, + "SELECT * FROM resource_type WHERE workspace_id = $1", + &w_id + ) + .fetch_all(&mut *tx) + .await?; + + for resource_type in resource_types { + let resource_str = &to_string_without_metadata(&resource_type, false, None).unwrap(); + archive + .write_to_archive( + &resource_str, + &format!("{}.resource-type.json", resource_type.name), + ) + .await?; + } + } + + { + let flows = sqlx::query_as::<_, Flow>( + "SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by + FROM flow + LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] + WHERE flow.workspace_id = $1 AND flow.archived = false", + ) + .bind(&w_id) + .fetch_all(&mut *tx) + .await?; + + for flow in flows { + let flow_str = &to_string_without_metadata(&flow, false, None).unwrap(); + archive + .write_to_archive(&flow_str, &format!("{}.flow.json", flow.path)) + .await?; + } + } + + if !skip_variables.unwrap_or(false) { + let variables = + sqlx::query_as::<_, ExportableListableVariable>(if !skip_secrets.unwrap_or(false) { + "SELECT * FROM variable WHERE workspace_id = $1 AND expires_at IS NULL" + } else { + "SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL" + }) + .bind(&w_id) + .fetch_all(&mut *tx) + .await?; + + let mc = build_crypt(&db, &w_id).await?; + + for mut var in variables { + if plain_secret.or(plain_secrets).unwrap_or(false) + && var.value.is_some() + && var.is_secret + { + var.value = Some(decrypt(&mc, var.value.unwrap())?); + } + let var_str = &to_string_without_metadata(&var, false, None).unwrap(); + archive + .write_to_archive(&var_str, &format!("{}.variable.json", var.path)) + .await?; + } + } + + { + let apps = sqlx::query_as::<_, AppWithLastVersion>( + "SELECT app.id, app.path, app.summary, app.versions, app.policy, app.custom_path, + app.extra_perms, app_version.value, + app_version.created_at, app_version.created_by from app, app_version + WHERE app.workspace_id = $1 AND app_version.id = app.versions[array_upper(app.versions, 1)]", + ) + .bind(&w_id) + .fetch_all(&mut *tx) + .await?; + + for app in apps { + let app_str = &to_string_without_metadata(&app, false, None).unwrap(); + archive + .write_to_archive(&app_str, &format!("{}.app.json", app.path)) + .await?; + } + } + + if include_schedules.unwrap_or(false) { + let schedules = sqlx::query_as::<_, Schedule>( + "SELECT * FROM schedule + WHERE workspace_id = $1", + ) + .bind(&w_id) + .fetch_all(&mut *tx) + .await?; + + for schedule in schedules { + let app_str = &to_string_without_metadata(&schedule, false, None).unwrap(); + archive + .write_to_archive(&app_str, &format!("{}.schedule.json", schedule.path)) + .await?; + } + } + + if include_users.unwrap_or(false) { + let users = sqlx::query!( + "SELECT * FROM usr + WHERE workspace_id = $1", + &w_id + ) + .fetch_all(&mut *tx) + .await?; + + for user in users { + let user = SimplifiedUser { + username: user.username, + role: if user.is_admin { + "admin".to_string() + } else if user.operator { + "operator".to_string() + } else { + "developer".to_string() + }, + disabled: user.disabled, + email: user.email, + }; + let user_str = &to_string_without_metadata( + &user, + false, + Some(vec!["is_admin", "operator", "email"]), + ) + .unwrap(); + archive + .write_to_archive(&user_str, &format!("users/{}.user.json", user.email)) + .await?; + } + } + + if include_groups.unwrap_or(false) { + let groups = sqlx::query!( + r#"SELECT g_.workspace_id, name, summary, extra_perms, array_agg(u2g.usr) filter (where u2g.usr is not null) as members + FROM usr u + JOIN usr_to_group u2g ON u2g.usr = u.username AND u2g.workspace_id = u.workspace_id + RIGHT JOIN group_ g_ ON g_.workspace_id = u.workspace_id AND g_.name = u2g.group_ + WHERE g_.workspace_id = $1 AND g_.name != 'all' + GROUP BY g_.workspace_id, name, summary, extra_perms"#, + &w_id + ) + .fetch_all(&mut *tx) + .await?; + + for group in groups { + let extra_perms: HashMap = serde_json::from_value(group.extra_perms) + .map_err(|e| { + Error::InternalErr(format!( + "Error parsing extra_perms for group {}: {}", + group.name, e + )) + })?; + tracing::info!("{:?}", extra_perms); + let members = group.members.unwrap_or(vec![]); + let admins: Vec = extra_perms + .iter() + .filter_map(|(k, v)| { + // only consider extra_perms that concern actual members of the group + if members.contains(&k[2..].to_string()) && *v { + Some(k.clone()) + } else { + None + } + }) + .sorted() + .collect(); + let group = SimplifiedGroup { + name: group.name, + summary: group.summary, + members: members + .iter() + .filter_map(|x| { + // remove members that are also admins as they are already in the admins list + let full_name = format!("u/{}", x); + if !admins.contains(&full_name) { + Some(full_name) + } else { + None + } + }) + .collect(), + admins, + }; + + let group_str = &to_string_without_metadata(&group, true, None).unwrap(); + archive + .write_to_archive(&group_str, &format!("groups/{}.group.json", group.name)) + .await?; + } + } + + if include_settings.unwrap_or(false) { + let settings = sqlx::query_as!( + SimplifiedSettings, + r#"SELECT + -- slack_team_id, + -- slack_name, + -- slack_command_script, + -- CASE WHEN slack_email = 'missing@email.xyz' THEN NULL ELSE slack_email END AS slack_email, + auto_invite_domain IS NOT NULL AS "auto_invite_enabled!", + CASE WHEN auto_invite_operator IS TRUE THEN 'operator' ELSE 'developer' END AS "auto_invite_as!", + CASE WHEN auto_add IS TRUE THEN 'add' ELSE 'invite' END AS "auto_invite_mode!", + webhook, + deploy_to, + error_handler, + ai_resource, + code_completion_enabled, + error_handler_extra_args, + error_handler_muted_on_cancel, + large_file_storage, + git_sync, + default_app, + default_scripts, + workspace.name + FROM workspace_settings + LEFT JOIN workspace ON workspace.id = workspace_settings.workspace_id + WHERE workspace_id = $1"#, + &w_id + ).fetch_one(&mut *tx).await?; + + let settings_str = serde_json::to_value(settings) + .map(|v| serde_json::to_string_pretty(&v).ok()) + .ok() + .flatten() + .ok_or_else(|| Error::InternalErr("Error serializing settings".to_string()))?; + + archive + .write_to_archive(&settings_str, "settings.json") + .await?; + } + + if include_key.unwrap_or(false) { + let key = sqlx::query_scalar!( + "SELECT key FROM workspace_key WHERE workspace_id = $1", + &w_id + ) + .fetch_one(&mut *tx) + .await?; + + let key_json = serde_json::to_value(key) + .map(|v| serde_json::to_string_pretty(&v).ok()) + .ok() + .flatten() + .ok_or_else(|| Error::InternalErr("Error serializing enryption key".to_string()))?; + archive + .write_to_archive(&key_json, "encryption_key.json") + .await?; + } + + archive.finish().await?; + + let file = tokio::fs::File::open(&file_path).await?; + + let stream = ReaderStream::new(file); + let body = axum::body::Body::from_stream(stream); + + let headers = [ + (header::CONTENT_TYPE, "application/x-tar".to_string()), + ( + header::CONTENT_DISPOSITION, + format!("attachment; filename=\"{name}\""), + ), + ]; + Ok((headers, body)) +} diff --git a/backend/windmill-api/src/workspaces_extra.rs b/backend/windmill-api/src/workspaces_extra.rs new file mode 100644 index 0000000000..8fba5d0be5 --- /dev/null +++ b/backend/windmill-api/src/workspaces_extra.rs @@ -0,0 +1,506 @@ +use crate::db::ApiAuthed; + +use crate::workspaces::CREATE_WORKSPACE_REQUIRE_SUPERADMIN; +use crate::{db::DB, utils::require_super_admin}; + +use axum::{ + extract::{Extension, Path}, + Json, +}; + +use windmill_audit::audit_ee::audit_log; +use windmill_audit::ActionKind; + +use windmill_common::worker::CLOUD_HOSTED; + +use windmill_common::{ + error::{Error, Result}, + utils::require_admin, +}; + +use serde::Deserialize; + +#[derive(Deserialize)] +pub(crate) struct ChangeWorkspaceId { + new_id: String, + new_name: String, +} + +pub(crate) async fn change_workspace_id( + authed: ApiAuthed, + Path(old_id): Path, + Extension(db): Extension, + Json(rw): Json, +) -> Result { + if *CLOUD_HOSTED { + return Err(Error::BadRequest( + "This feature is not available on the cloud".to_string(), + )); + } + + if *CREATE_WORKSPACE_REQUIRE_SUPERADMIN { + require_super_admin(&db, &authed.email).await?; + } else { + require_admin(authed.is_admin, &authed.username)?; + } + + let mut tx = db.begin().await?; + + let workspace_conflict = sqlx::query_scalar!( + "SELECT EXISTS(SELECT 1 FROM workspace WHERE id = $1)", + &rw.new_id, + ) + .fetch_one(&mut *tx) + .await? + .unwrap_or(false); + + if workspace_conflict { + return Err(Error::BadRequest(format!( + "workspace id {} already used", + &rw.new_id + ))); + } + + // duplicate workspace with new id name + sqlx::query!( + "INSERT INTO workspace SELECT $1, $2, owner, deleted, premium FROM workspace WHERE id = $3", + &rw.new_id, + &rw.new_name, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE account SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE app SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE audit SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE capture SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE completed_job SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE dependency_map SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE deployment_metadata SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE draft SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE favorite SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "INSERT INTO flow + (workspace_id, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at) + SELECT $1, path, summary, description, archived, extra_perms, dependency_job, draft_only, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, concurrency_key, versions, value, schema, edited_by, edited_at + FROM flow WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE flow_version SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &old_id) + .execute(&mut *tx) + .await?; + + // have to duplicate group_ with new workspace id because of foreign key constraint + sqlx::query!( + "INSERT INTO group_ SELECT $1, name, summary, extra_perms FROM group_ WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE usr_to_group SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + // then delete old group_ + sqlx::query!("DELETE FROM group_ WHERE workspace_id = $1", &old_id) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE folder SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE input SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE job_logs SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE job_stats SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE queue SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE job SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE raw_app SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE resource SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE resource_type SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE schedule SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE script SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE token SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE usage SET id = $1 WHERE id = $2 AND is_workspace = true", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE usr SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE variable SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE workspace_env SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE workspace_invite SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE workspace_key SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "UPDATE workspace_settings SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + + // delete old workspace + sqlx::query!("DELETE FROM workspace WHERE id = $1", &old_id) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + &authed, + "workspace.change_workspace_id", + ActionKind::Update, + &rw.new_id, + Some(&authed.email), + None, + ) + .await?; + tx.commit().await?; + Ok(format!( + "updated workspace from {} to {}", + &old_id, &rw.new_id + )) +} + +pub(crate) async fn delete_workspace( + Extension(db): Extension, + Path(w_id): Path, + authed: ApiAuthed, +) -> Result { + let w_id = match w_id.as_str() { + "starter" => Err(Error::BadRequest( + "starter workspace cannot be deleted".to_string(), + )), + "admins" => Err(Error::BadRequest( + "admins workspace cannot be deleted".to_string(), + )), + _ => Ok(w_id), + }?; + let mut tx = db.begin().await?; + require_super_admin(&db, &authed.email).await?; + + sqlx::query!("DELETE FROM dependency_map WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM queue WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM capture WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM draft WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM script WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM app WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM raw_app WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM input WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM variable WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM resource WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM schedule WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM completed_job WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM job_stats WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "DELETE FROM deployment_metadata WHERE workspace_id = $1", + &w_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM usr WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM resource_type WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "DELETE FROM workspace_invite WHERE workspace_id = $1", + &w_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM usr_to_group WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM group_ WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM folder WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM account WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM workspace_key WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "DELETE FROM workspace_settings WHERE workspace_id = $1", + &w_id + ) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM token WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM workspace WHERE id = $1", &w_id) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + &authed, + "workspaces.delete", + ActionKind::Delete, + &w_id, + Some(&authed.email), + None, + ) + .await?; + tx.commit().await?; + + Ok(format!("Deleted workspace {}", &w_id)) +} diff --git a/backend/windmill-common/src/users.rs b/backend/windmill-common/src/users.rs index 7675c8799e..288337eb40 100644 --- a/backend/windmill-common/src/users.rs +++ b/backend/windmill-common/src/users.rs @@ -10,6 +10,8 @@ pub const SUPERADMIN_SECRET_EMAIL: &str = "superadmin_secret@windmill.dev"; pub const SUPERADMIN_NOTIFICATION_EMAIL: &str = "superadmin_notification@windmill.dev"; pub const SUPERADMIN_SYNC_EMAIL: &str = "superadmin_sync@windmill.dev"; +pub const COOKIE_NAME: &str = "token"; + pub fn username_to_permissioned_as(user: &str) -> String { if user.contains('@') { user.to_string()