diff --git a/backend/rustfmt.toml b/backend/rustfmt.toml index c3bdd32db2..d862e08106 100644 --- a/backend/rustfmt.toml +++ b/backend/rustfmt.toml @@ -1,7 +1,2 @@ -imports_granularity = "Crate" max_width = 100 use_small_heuristics = "Default" -indent_style = "Block" -fn_single_line = false -force_multiline_blocks = true -format_strings = true diff --git a/backend/src/flows.rs b/backend/src/flows.rs index 2ab80ea1d2..095c8999e4 100644 --- a/backend/src/flows.rs +++ b/backend/src/flows.rs @@ -228,7 +228,8 @@ async fn create_flow( check_schedule_conflict(&mut tx, &w_id, &nf.path).await?; sqlx::query!( - "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, schema) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::text::json)", + "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, \ + schema) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::text::json)", w_id, nf.path, nf.summary, @@ -267,7 +268,8 @@ async fn check_schedule_conflict<'c>( path: &str, ) -> error::Result<()> { let exists_flow = sqlx::query_scalar!( - "SELECT EXISTS (SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2 AND path != script_path)", + "SELECT EXISTS (SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2 AND path != \ + script_path)", path, w_id ) @@ -276,7 +278,8 @@ async fn check_schedule_conflict<'c>( .unwrap_or(false); if exists_flow { return Err(error::Error::BadConfig(format!( - "A flow cannot have the same path as a schedule if the schedule does not trigger that same flow: {path}", + "A flow cannot have the same path as a schedule if the schedule does not trigger that \ + same flow: {path}", ))); }; Ok(()) @@ -295,7 +298,8 @@ async fn update_flow( let schema = nf.schema.map(|x| x.0); let flow = sqlx::query_scalar!( - "UPDATE flow SET path = $1, summary = $2, description = $3, value = $4, edited_by = $5, edited_at = $6, schema = $7 WHERE path = $8 AND workspace_id = $9 RETURNING path", + "UPDATE flow SET path = $1, summary = $2, description = $3, value = $4, edited_by = $5, \ + edited_at = $6, schema = $7 WHERE path = $8 AND workspace_id = $9 RETURNING path", nf.path, nf.summary, nf.description, @@ -358,7 +362,8 @@ async fn exists_flow_by_path( let path = path.to_path(); let exists = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM flow WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter'))", + "SELECT EXISTS(SELECT 1 FROM flow WHERE path = $1 AND (workspace_id = $2 OR workspace_id \ + = 'starter'))", path, w_id ) diff --git a/backend/src/granular_acls.rs b/backend/src/granular_acls.rs index 0b2b03b585..d34e89e636 100644 --- a/backend/src/granular_acls.rs +++ b/backend/src/granular_acls.rs @@ -46,7 +46,8 @@ async fn add_granular_acl( let identifier = if kind == "group_" { "name" } else { "path" }; let obj_o = sqlx::query_scalar::<_, serde_json::Value>(&format!( - "UPDATE {kind} SET extra_perms = jsonb_set(extra_perms, '{{\"{owner}\"}}', to_jsonb($1), true) WHERE {identifier} = $2 AND workspace_id = $3 RETURNING extra_perms" + "UPDATE {kind} SET extra_perms = jsonb_set(extra_perms, '{{\"{owner}\"}}', to_jsonb($1), \ + true) WHERE {identifier} = $2 AND workspace_id = $3 RETURNING extra_perms" )) .bind(write.unwrap_or(false)) .bind(path) @@ -74,7 +75,8 @@ async fn remove_granular_acl( let identifier = if kind == "group_" { "name" } else { "path" }; let obj_o = sqlx::query_scalar::<_, serde_json::Value>(&format!( - "UPDATE {kind} SET extra_perms = extra_perms - $1 WHERE {identifier} = $2 AND workspace_id = $3 RETURNING extra_perms" + "UPDATE {kind} SET extra_perms = extra_perms - $1 WHERE {identifier} = $2 AND \ + workspace_id = $3 RETURNING extra_perms" )) .bind(owner) .bind(path) diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index 4e9eeb33ac..5c6996ddf5 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -13,19 +13,17 @@ use sqlx::{query_scalar, Postgres, Transaction}; use std::collections::HashMap; use tracing::instrument; -use crate::error::to_anyhow; -use crate::scripts::{get_hub_script_by_path, ScriptLang}; -use crate::worker_flow::init_flow_status; use crate::{ audit::{audit_log, ActionKind}, db::{UserDB, DB}, error, - error::Error, + error::{to_anyhow, Error}, flows::FlowValue, schedule::get_schedule_opt, - scripts::ScriptHash, + scripts::{get_hub_script_by_path, ScriptHash, ScriptLang}, users::{owner_to_token_owner, Authed}, utils::{require_admin, Pagination, StripPath}, + worker_flow::init_flow_status, }; use axum::{ extract::{Extension, Path, Query}, @@ -269,8 +267,10 @@ pub async fn get_latest_hash_for_path<'c>( script_path: &str, ) -> error::Result { let script_hash_o = sqlx::query_scalar!( - "select hash from script where path = $1 AND (workspace_id = $2 OR workspace_id = 'starter') AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter')) AND + "select hash from script where path = $1 AND (workspace_id = $2 OR workspace_id = \ + 'starter') AND + created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND (workspace_id = $2 OR \ + workspace_id = 'starter')) AND deleted = false", script_path, w_id @@ -319,7 +319,8 @@ pub async fn get_path_for_hash<'c>( hash: i64, ) -> error::Result { let path = sqlx::query_scalar!( - "select path from script where hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", + "select path from script where hash = $1 AND (workspace_id = $2 OR workspace_id = \ + 'starter')", hash, w_id ) @@ -703,9 +704,8 @@ async fn cancel_job( let mut tx = user_db.begin(&authed).await?; let job_option = sqlx::query_scalar!( - "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2 \ - WHERE id = $3 AND schedule_path IS NULL AND workspace_id = $4\ - RETURNING id", + "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2 WHERE id = $3 \ + AND schedule_path IS NULL AND workspace_id = $4RETURNING id", &authed.username, reason, id, @@ -756,7 +756,8 @@ async fn delete_completed_job( require_admin(authed.is_admin, &authed.username)?; let job_o = sqlx::query_as::<_, CompletedJob>( - "UPDATE completed_job SET logs = '', deleted = true WHERE id = $1 AND workspace_id = $2 RETURNING *", + "UPDATE completed_job SET logs = '', deleted = true WHERE id = $1 AND workspace_id = $2 \ + RETURNING *", ) .bind(id) .bind(&w_id) @@ -821,7 +822,8 @@ async fn get_job_update( })) } else { let logs = query_scalar!( - "SELECT substr(logs, $1) as logs FROM completed_job WHERE workspace_id = $2 AND id = $3", + "SELECT substr(logs, $1) as logs FROM completed_job WHERE workspace_id = $2 AND id = \ + $3", log_offset, &w_id, &id @@ -1075,29 +1077,41 @@ pub async fn push<'c>( if let Some(nb_jobs) = rate_limiting_queue { if nb_jobs > MAX_NB_OF_JOBS_IN_Q_PER_USER { return Err(error::Error::ExecutionErr(format!( - "You have exceeded the number of authorized elements of queue at any given time: {}", MAX_NB_OF_JOBS_IN_Q_PER_USER))); + "You have exceeded the number of authorized elements of queue at any given \ + time: {}", + MAX_NB_OF_JOBS_IN_Q_PER_USER + ))); } } let rate_limiting_duration = sqlx::query_scalar!( - "SELECT SUM(duration) FROM completed_job WHERE created_by = $1 AND created_at > NOW() - INTERVAL '1200 seconds' AND workspace_id = $2", - user, - workspace_id - ) - .fetch_one(&mut tx) - .await?; + "SELECT SUM(duration) FROM completed_job WHERE created_by = $1 AND created_at > NOW() \ + - INTERVAL '1200 seconds' AND workspace_id = $2", + user, + workspace_id + ) + .fetch_one(&mut tx) + .await?; if let Some(sum_duration) = rate_limiting_duration { if sum_duration > MAX_DURATION_LAST_1200 { return Err(error::Error::ExecutionErr(format!( - "You have exceeded the scripts cumulative duration limit over the last 20m which is: {}", MAX_DURATION_LAST_1200))); + "You have exceeded the scripts cumulative duration limit over the last 20m \ + which is: {}", + MAX_DURATION_LAST_1200 + ))); } } } let (script_hash, script_path, raw_code, job_kind, raw_flow, language) = match job_payload { JobPayload::ScriptHash { hash, path } => { - let language = sqlx::query_scalar!("SELECT language as \"language: ScriptLang\" FROM script WHERE hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", hash.0, workspace_id) + let language = sqlx::query_scalar!( + "SELECT language as \"language: ScriptLang\" FROM script WHERE hash = $1 AND \ + (workspace_id = $2 OR workspace_id = 'starter')", + hash.0, + workspace_id + ) .fetch_one(&mut tx) .await?; ( @@ -1168,11 +1182,15 @@ pub async fn push<'c>( (None, path, None, JobKind::FlowPreview, Some(value), None) } JobPayload::Flow(flow) => { - let value_json = sqlx::query_scalar!("SELECT value FROM flow WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", - flow, workspace_id) - .fetch_optional(&mut tx) - .await? - .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", flow)))?; + let value_json = sqlx::query_scalar!( + "SELECT value FROM flow WHERE path = $1 AND (workspace_id = $2 OR workspace_id = \ + 'starter')", + flow, + workspace_id + ) + .fetch_optional(&mut tx) + .await? + .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", flow)))?; let value = serde_json::from_value::(value_json).map_err(|err| { Error::InternalErr(format!( "could not convert json to flow for {flow}: {err:?}" @@ -1186,8 +1204,10 @@ pub async fn push<'c>( let uuid = sqlx::query_scalar!( "INSERT INTO queue (workspace_id, id, parent_job, created_by, permissioned_as, scheduled_for, - script_hash, script_path, raw_code, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) RETURNING id", + script_hash, script_path, raw_code, args, job_kind, schedule_path, raw_flow, \ + flow_status, is_flow_step, language) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) \ + RETURNING id", workspace_id, job_id, parent_job, @@ -1284,13 +1304,13 @@ pub async fn add_completed_job( .num_seconds() as i32; let _ = sqlx::query!( "INSERT INTO completed_job as cj - (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, script_path, \ - args, result, logs, - raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, permissioned_as, flow_status, raw_flow, \ - is_flow_step, is_skipped) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23) \ - ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) \ - RETURNING id", + (workspace_id, id, parent_job, created_by, created_at, duration, success, script_hash, \ + script_path, args, result, logs, + raw_code, canceled, canceled_by, canceled_reason, job_kind, schedule_path, \ + permissioned_as, flow_status, raw_flow, is_flow_step, is_skipped) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, \ + $18, $19, $20, $21, $22, $23) ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, \ + logs = concat(cj.logs, $12) RETURNING id", queued_job.workspace_id, queued_job.id, queued_job.parent_job, diff --git a/backend/src/js_eval.rs b/backend/src/js_eval.rs index f6b42ccd1a..c1970ddfa6 100644 --- a/backend/src/js_eval.rs +++ b/backend/src/js_eval.rs @@ -5,21 +5,13 @@ * LICENSE-AGPL for a copy of the license. */ -use deno_core::op; -use deno_core::serde_v8; -use deno_core::v8; -use deno_core::v8::IsolateHandle; -use deno_core::Extension; -use deno_core::JsRuntime; -use deno_core::RuntimeOptions; +use deno_core::{op, serde_v8, v8, v8::IsolateHandle, Extension, JsRuntime, RuntimeOptions}; use itertools::Itertools; use regex::Regex; use serde_json::Value; -use tokio::sync::oneshot; -use tokio::time::timeout; +use tokio::{sync::oneshot, time::timeout}; -use crate::client; -use crate::error::Error; +use crate::{client, error::Error}; pub struct EvalCreds { pub workspace: String, @@ -193,11 +185,13 @@ async function resource(path) {{ }})() "#, env.into_iter() - .map(|(a, b)| format!( - "let {a} = {};\n", - serde_json::to_string(&b) - .unwrap_or_else(|_| "\"error serializing value\"".to_string()) - )) + .map(|(a, b)| { + format!( + "let {a} = {};\n", + serde_json::to_string(&b) + .unwrap_or_else(|_| "\"error serializing value\"".to_string()) + ) + }) .join(""), ); tracing::debug!("{}", code); diff --git a/backend/src/main.rs b/backend/src/main.rs index 569bea9f77..283afd4a77 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -76,7 +76,9 @@ async fn main() -> anyhow::Result<()> { .unwrap_or(false); tracing::info!( - "DISABLE_NSJAIL: {disable_nsjail}, DISABLE_NUSER: {disable_nuser}, BASE_URL: {base_url}, SLEEP_QUEUE: {sleep_queue}, NUM_WORKERS: {num_workers}, TIMEOUT: {timeout}" + "DISABLE_NSJAIL: {disable_nsjail}, DISABLE_NUSER: {disable_nuser}, BASE_URL: \ + {base_url}, SLEEP_QUEUE: {sleep_queue}, NUM_WORKERS: {num_workers}, TIMEOUT: \ + {timeout}" ); windmill::run_workers( db.clone(), diff --git a/backend/src/oauth2.rs b/backend/src/oauth2.rs index cbe7a29142..b927c4c355 100644 --- a/backend/src/oauth2.rs +++ b/backend/src/oauth2.rs @@ -1,37 +1,39 @@ -use std::collections::HashMap; -use std::fmt::Debug; +use std::{collections::HashMap, fmt::Debug}; use std::sync::Arc; -use axum::body::Bytes; -use axum::extract::{Extension, FromRequest, Path, Query, RequestParts}; -use axum::response::Redirect; -use axum::routing::{get, post}; -use axum::{async_trait, Json, Router}; +use axum::{ + async_trait, + body::Bytes, + extract::{Extension, FromRequest, Path, Query, RequestParts}, + response::Redirect, + routing::{get, post}, + Json, Router, +}; use chrono::{Duration, Utc}; use hyper::StatusCode; use itertools::Itertools; use oauth2::{Client as OClient, *}; use reqwest::Client; -use serde::de::DeserializeOwned; -use serde::{Deserialize, Serialize}; +use serde::{de::DeserializeOwned, Deserialize, Serialize}; use slack_http_verifier::SlackVerifier; use sqlx::{Postgres, Transaction}; -use tokio::fs::File; -use tokio::io::AsyncReadExt; +use tokio::{fs::File, io::AsyncReadExt}; use tower_cookies::{Cookie, Cookies}; -use crate::audit::{audit_log, ActionKind}; -use crate::db::{UserDB, DB}; -use crate::error::{self, to_anyhow, Error, Result}; -use crate::jobs; -use crate::jobs::{get_latest_hash_for_path, JobPayload}; -use crate::users::Authed; -use crate::utils::not_found_if_none; -use crate::variables::{build_crypt, encrypt}; -use crate::workspaces::WorkspaceSettings; -use crate::BaseUrl; +use crate::{ + audit::{audit_log, ActionKind}, + db::{UserDB, DB}, + error::{self, to_anyhow, Error, Result}, + jobs, + jobs::{get_latest_hash_for_path, JobPayload}, + users::Authed, + utils::not_found_if_none, + variables::{build_crypt, encrypt}, + workspaces::WorkspaceSettings, + BaseUrl, +}; pub fn global_service() -> Router { Router::new() @@ -278,7 +280,8 @@ async fn create_account( let expires_at = chrono::Utc::now() + Duration::seconds(payload.expires_in); let id = sqlx::query_scalar!( - "INSERT INTO account (workspace_id, client, owner, expires_at, refresh_token) VALUES ($1, $2, $3, $4, $5) RETURNING id", + "INSERT INTO account (workspace_id, client, owner, expires_at, refresh_token) VALUES ($1, \ + $2, $3, $4, $5) RETURNING id", w_id, payload.client, payload.owner, @@ -469,8 +472,12 @@ pub async fn _refresh_token<'c>( .unwrap(), ); sqlx::query!( - "UPDATE account SET refresh_token = $1, expires_at = $2 WHERE workspace_id = $3 AND id = $4", - token.refresh_token.map(|x| x.to_string()).unwrap_or(account.refresh_token), + "UPDATE account SET refresh_token = $1, expires_at = $2 WHERE workspace_id = $3 AND id = \ + $4", + token + .refresh_token + .map(|x| x.to_string()) + .unwrap_or(account.refresh_token), expires_at, w_id, id @@ -548,7 +555,8 @@ async fn set_workspace_slack( sqlx::query!( "INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name) - VALUES ($1, $2, $3) ON CONFLICT (workspace_id) DO UPDATE SET slack_team_id = $2, slack_name = $3", + VALUES ($1, $2, $3) ON CONFLICT (workspace_id) DO UPDATE SET slack_team_id = $2, \ + slack_name = $3", &w_id, token.team_id, token.team_name @@ -720,15 +728,18 @@ async fn login_callback( crate::users::create_session_token(&email, super_admin, &mut tx, cookies).await?; } else { return Err(error::Error::BadRequest(format!( - "an user with the email associated to this login exists but with a different login type {login_type}") - )); + "an user with the email associated to this login exists but with a different \ + login type {login_type}" + ))); } } else { let user = get_user_info(&http_client, &client_name, &token).await?; - sqlx::query( - &format!("INSERT INTO password (email, name, company, login_type, verified) VALUES ($1, $2, $3, '{}', true)", &client_name) - ) + sqlx::query(&format!( + "INSERT INTO password (email, name, company, login_type, verified) VALUES ($1, \ + $2, $3, '{}', true)", + &client_name + )) .bind(&email) .bind(&user.name) .bind(user.company) diff --git a/backend/src/parser.rs b/backend/src/parser.rs index 8ceefa116a..bb229331dd 100644 --- a/backend/src/parser.rs +++ b/backend/src/parser.rs @@ -127,8 +127,7 @@ pub fn parse_python_signature(code: &str) -> error::Result { } } -use swc_common::sync::Lrc; -use swc_common::{FileName, SourceMap}; +use swc_common::{sync::Lrc, FileName, SourceMap}; use swc_ecma_ast::{ AssignPat, BindingIdent, Decl, ExportDecl, FnDecl, Ident, ModuleDecl, ModuleItem, Pat, TsArrayType, TsEntityName, TsKeywordTypeKind, TsType, TsTypeRef, @@ -240,70 +239,74 @@ fn binding_ident_to_arg( id.sym.to_string(), type_ann .as_ref() - .map(|x| match &*x.type_ann { - TsType::TsKeywordType(t) => match t.kind { - TsKeywordTypeKind::TsObjectKeyword => Typ::Dict, - TsKeywordTypeKind::TsBooleanKeyword => Typ::Bool, - TsKeywordTypeKind::TsBigIntKeyword => Typ::Int, - TsKeywordTypeKind::TsNumberKeyword => Typ::Float, - TsKeywordTypeKind::TsStringKeyword => Typ::Str, - _ => Typ::Unknown, - }, - // TODO: we can do better here and extract the inner type of array - TsType::TsArrayType(TsArrayType { span: _, elem_type }) => { - match &**elem_type { - TsType::TsTypeRef(TsTypeRef { - span: _, - type_name: - TsEntityName::Ident(Ident { - span: _, - sym, - optional: _, - }), - type_params: _, - }) => match sym.to_string().as_str() { - "Base64" => Typ::List(InnerTyp::Bytes), - "Email" => Typ::List(InnerTyp::Email), - "bigint" => Typ::List(InnerTyp::Int), - "number" => Typ::List(InnerTyp::Float), + .map(|x| { + match &*x.type_ann { + TsType::TsKeywordType(t) => match t.kind { + TsKeywordTypeKind::TsObjectKeyword => Typ::Dict, + TsKeywordTypeKind::TsBooleanKeyword => Typ::Bool, + TsKeywordTypeKind::TsBigIntKeyword => Typ::Int, + TsKeywordTypeKind::TsNumberKeyword => Typ::Float, + TsKeywordTypeKind::TsStringKeyword => Typ::Str, + _ => Typ::Unknown, + }, + // TODO: we can do better here and extract the inner type of array + TsType::TsArrayType(TsArrayType { span: _, elem_type }) => { + match &**elem_type { + TsType::TsTypeRef(TsTypeRef { + span: _, + type_name: + TsEntityName::Ident(Ident { + span: _, + sym, + optional: _, + }), + type_params: _, + }) => match sym.to_string().as_str() { + "Base64" => Typ::List(InnerTyp::Bytes), + "Email" => Typ::List(InnerTyp::Email), + "bigint" => Typ::List(InnerTyp::Int), + "number" => Typ::List(InnerTyp::Float), + _ => Typ::List(InnerTyp::Str), + }, + //TsType::TsKeywordType(()) _ => Typ::List(InnerTyp::Str), - }, - //TsType::TsKeywordType(()) - _ => Typ::List(InnerTyp::Str), + } } - } - TsType::TsTypeRef(TsTypeRef { - span: _, - type_name, - type_params, - }) => { - let sym = match type_name { - TsEntityName::Ident(Ident { - span: _, - sym, - optional: _, - }) => sym, - TsEntityName::TsQualifiedName(p) => &*p.right.sym, - }; - match sym.to_string().as_str() { - "Resource" => Typ::Resource( - type_params - .as_ref() - .and_then(|x| { - x.params.get(0).and_then(|y| { - y.as_ts_lit_type().and_then(|z| { - z.lit.as_str().map(|a| a.to_owned().value.to_string()) + TsType::TsTypeRef(TsTypeRef { + span: _, + type_name, + type_params, + }) => { + let sym = match type_name { + TsEntityName::Ident(Ident { + span: _, + sym, + optional: _, + }) => sym, + TsEntityName::TsQualifiedName(p) => &*p.right.sym, + }; + match sym.to_string().as_str() { + "Resource" => Typ::Resource( + type_params + .as_ref() + .and_then(|x| { + x.params.get(0).and_then(|y| { + y.as_ts_lit_type().and_then(|z| { + z.lit + .as_str() + .map(|a| a.to_owned().value.to_string()) + }) }) }) - }) - .unwrap_or_else(|| "unknown".to_string()), - ), - "Base64" => Typ::Bytes, - "Email" => Typ::Email, - _ => Typ::Unknown, + .unwrap_or_else(|| "unknown".to_string()), + ), + "Base64" => Typ::Bytes, + "Email" => Typ::Email, + _ => Typ::Unknown, + } } + _ => Typ::Unknown, } - _ => Typ::Unknown, }) .unwrap_or(Typ::Unknown), )) diff --git a/backend/src/resources.rs b/backend/src/resources.rs index c366eda69a..a539192e0a 100644 --- a/backend/src/resources.rs +++ b/backend/src/resources.rs @@ -140,7 +140,8 @@ async fn get_resource( let resource_o = sqlx::query_as!( Resource, - "SELECT * from resource WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", + "SELECT * from resource WHERE path = $1 AND (workspace_id = $2 OR workspace_id = \ + 'starter')", path.to_owned(), &w_id ) @@ -179,7 +180,8 @@ async fn get_resource_value( let mut tx = user_db.begin(&authed).await?; let value_o = sqlx::query_scalar!( - "SELECT value from resource WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", + "SELECT value from resource WHERE path = $1 AND (workspace_id = $2 OR workspace_id = \ + 'starter')", path.to_owned(), &w_id ) @@ -312,9 +314,14 @@ async fn list_resource_types( Extension(db): Extension, Path(w_id): Path, ) -> JsonResult> { - let rows = sqlx::query_as!(ResourceType, "SELECT * from resource_type WHERE (workspace_id = $1 OR workspace_id = 'starter') ORDER BY name", &w_id) - .fetch_all(&db) - .await?; + let rows = sqlx::query_as!( + ResourceType, + "SELECT * from resource_type WHERE (workspace_id = $1 OR workspace_id = 'starter') ORDER \ + BY name", + &w_id + ) + .fetch_all(&db) + .await?; Ok(Json(rows)) } @@ -323,9 +330,13 @@ async fn list_resource_types_names( Extension(db): Extension, Path(w_id): Path, ) -> JsonResult> { - let rows = sqlx::query_scalar!("SELECT name from resource_type WHERE (workspace_id = $1 OR workspace_id = 'starter') ORDER BY name", &w_id) - .fetch_all(&db) - .await?; + let rows = sqlx::query_scalar!( + "SELECT name from resource_type WHERE (workspace_id = $1 OR workspace_id = 'starter') \ + ORDER BY name", + &w_id + ) + .fetch_all(&db) + .await?; Ok(Json(rows)) } @@ -339,7 +350,8 @@ async fn get_resource_type( let resource_type_o = sqlx::query_as!( ResourceType, - "SELECT * from resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", + "SELECT * from resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = \ + 'starter')", &name, &w_id ) diff --git a/backend/src/schedule.rs b/backend/src/schedule.rs index a68131b7e2..31d40ef5a8 100644 --- a/backend/src/schedule.rs +++ b/backend/src/schedule.rs @@ -133,8 +133,10 @@ async fn create_schedule( check_flow_conflict(&mut tx, &w_id, &ns.path, ns.is_flow, &ns.script_path).await?; - let schedule = sqlx::query_as!(Schedule, - "INSERT INTO schedule (workspace_id, path, schedule, offset_, edited_by, script_path, is_flow, args, enabled) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING *", + let schedule = sqlx::query_as!( + Schedule, + "INSERT INTO schedule (workspace_id, path, schedule, offset_, edited_by, script_path, \ + is_flow, args, enabled) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING *", w_id, ns.path, ns.schedule, @@ -195,9 +197,10 @@ async fn check_flow_conflict<'c>( .unwrap_or(false); if exists_flow { return Err(error::Error::BadConfig(format!( - "If a schedule has the same path as or a flow, it must be its primary schedule and hence can only trigger it. + "If a schedule has the same path as or a flow, it must be its primary schedule \ + and hence can only trigger it. However the provided path is: {script_path} and is_flow is: {is_flow}", - ))); + ))); }; } Ok(()) @@ -233,8 +236,10 @@ async fn edit_schedule( check_flow_conflict(&mut tx, &w_id, &path, es.is_flow, &es.script_path).await?; clear_schedule(&mut tx, path).await?; - let schedule = sqlx::query_as!(Schedule, - "UPDATE schedule SET schedule = $1, script_path = $2, is_flow = $3, args = $4 WHERE path = $5 AND workspace_id = $6 RETURNING *", + let schedule = sqlx::query_as!( + Schedule, + "UPDATE schedule SET schedule = $1, script_path = $2, is_flow = $3, args = $4 WHERE path \ + = $5 AND workspace_id = $6 RETURNING *", es.schedule, es.script_path, es.is_flow, diff --git a/backend/src/scripts.rs b/backend/src/scripts.rs index f6bbf27b2e..6410127bdc 100644 --- a/backend/src/scripts.rs +++ b/backend/src/scripts.rs @@ -342,7 +342,7 @@ async fn create_script( if let Some(clashing_hash) = clashing_hash_o { return Err(Error::BadRequest(format!( "A script with hash {} with same parent_hash has been found. However, the \ - lineage must be linear: no 2 scripts can have the same parent", + lineage must be linear: no 2 scripts can have the same parent", ScriptHash(clashing_hash) ))); }; @@ -395,9 +395,9 @@ async fn create_script( }; //::text::json is to ensure we use serde_json with preserve order sqlx::query!( - "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, \ - created_by, schema, is_template, extra_perms, lock, language, is_trigger) VALUES \ - ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14)", + "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \ + content, created_by, schema, is_template, extra_perms, lock, language, is_trigger) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14)", &w_id, &hash.0, ns.path, @@ -515,8 +515,10 @@ async fn get_script_by_path( let mut tx = user_db.begin(&authed).await?; let script_o = sqlx::query_as::<_, Script>( - "SELECT * FROM script WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter') AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND archived = false AND (workspace_id = $2 OR workspace_id = 'starter'))", + "SELECT * FROM script WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter') \ + AND + created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND archived = false AND \ + (workspace_id = $2 OR workspace_id = 'starter'))", ) .bind(path) .bind(w_id) @@ -535,9 +537,12 @@ async fn exists_script_by_path( let path = path.to_path(); let exists = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM script WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter') AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND (workspace_id = $2 OR workspace_id = 'starter')))", - path, w_id + "SELECT EXISTS(SELECT 1 FROM script WHERE path = $1 AND (workspace_id = $2 OR \ + workspace_id = 'starter') AND + created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND (workspace_id = $2 \ + OR workspace_id = 'starter')))", + path, + w_id ) .fetch_one(&db) .await? @@ -586,8 +591,10 @@ async fn get_deployment_status( Path((w_id, hash)): Path<(String, ScriptHash)>, ) -> JsonResult { let mut tx = user_db.begin(&authed).await?; - let status_o: Option = sqlx::query_as!(DeploymentStatus, - "SELECT lock, lock_error_logs FROM script WHERE hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')", + let status_o: Option = sqlx::query_as!( + DeploymentStatus, + "SELECT lock, lock_error_logs FROM script WHERE hash = $1 AND (workspace_id = $2 OR \ + workspace_id = 'starter')", hash.0, w_id, ) @@ -669,8 +676,8 @@ async fn delete_script_by_hash( require_admin(authed.is_admin, &authed.username)?; let script = sqlx::query_as::<_, Script>( - "UPDATE script SET content = '', archived = true, deleted = true WHERE hash = $1 AND workspace_id = $2\ - RETURNING *", + "UPDATE script SET content = '', archived = true, deleted = true WHERE hash = $1 AND \ + workspace_id = $2RETURNING *", ) .bind(&hash.0) .bind(&w_id) diff --git a/backend/src/users.rs b/backend/src/users.rs index c8bd906378..817507d223 100644 --- a/backend/src/users.rs +++ b/backend/src/users.rs @@ -10,13 +10,13 @@ use std::{sync::Arc, time::Duration}; use crate::{ audit::{audit_log, ActionKind}, db::{UserDB, DB}, - error::{Error, JsonResult, Result, self}, - utils::{require_admin, require_super_admin, Pagination} + error::{self, Error, JsonResult, Result}, + utils::{require_admin, require_super_admin, Pagination}, }; use argon2::{password_hash::SaltString, Argon2, PasswordHash, PasswordHasher, PasswordVerifier}; use axum::{ async_trait, - extract::{Extension, FromRequest, Path, RequestParts, Query}, + extract::{Extension, FromRequest, Path, Query, RequestParts}, http, routing::{delete, get, post}, Json, Router, @@ -25,7 +25,7 @@ use hyper::StatusCode; use rand::rngs::OsRng; use retainer::Cache; use serde::{Deserialize, Serialize}; -use sqlx::{FromRow}; +use sqlx::FromRow; use time::OffsetDateTime; use tower_cookies::{Cookie, Cookies}; use tracing::Span; @@ -48,8 +48,6 @@ pub fn workspaced_service() -> Router { .route("/leave", post(leave_workspace)) } - - pub fn global_service() -> Router { Router::new() .route("/email", get(get_email)) @@ -65,13 +63,11 @@ pub fn global_service() -> Router { .route("/tokens/create", post(create_token)) .route("/tokens/delete/:token_prefix", delete(delete_token)) .route("/tokens/list", get(list_tokens)) - // .route("/list_invite_codes", get(list_invite_codes)) - // .route("/create_invite_code", post(create_invite_code)) - // .route("/signup", post(signup)) - // .route("/lost_password", post(lost_password)) - // .route("/use_magic_link", get(use_magic_link)) - - + // .route("/list_invite_codes", get(list_invite_codes)) + // .route("/create_invite_code", post(create_invite_code)) + // .route("/signup", post(signup)) + // .route("/lost_password", post(lost_password)) + // .route("/use_magic_link", get(use_magic_link)) } pub fn make_unauthed_service() -> Router { @@ -92,15 +88,17 @@ impl AuthCache { } 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 key = ( + w_id.as_ref().unwrap_or(&"".to_string()).to_string(), + token.to_string(), + ); let s = self.cache.get(&key).await.map(|c| c.to_owned()); match s { - a @ Some(_) => { - a - }, + a @ Some(_) => a, None => { let user_o = sqlx::query_as::<_, (Option, Option, bool)>( - "UPDATE token SET last_used_at = $1 WHERE token = $2 AND (expiration > NOW() OR expiration IS NULL) RETURNING owner, email, super_admin", + "UPDATE token SET last_used_at = $1 WHERE token = $2 AND (expiration > NOW() \ + OR expiration IS NULL) RETURNING owner, email, super_admin", ) .bind(chrono::Utc::now()) .bind(token) @@ -114,40 +112,44 @@ impl AuthCache { match user { (_, Some(email), super_admin) => { if w_id.is_some() { - let row_o = - sqlx::query_as::<_, (String, bool)>( - "SELECT username, is_admin FROM usr where email = $1 AND workspace_id = $2", - ) - .bind(&email) - .bind(&w_id.as_ref().unwrap()) - .fetch_optional(&self.db) - .await - .unwrap_or(Some(("error".to_string(), false))); + let row_o = sqlx::query_as::<_, (String, bool)>( + "SELECT username, is_admin FROM usr where email = $1 AND \ + workspace_id = $2", + ) + .bind(&email) + .bind(&w_id.as_ref().unwrap()) + .fetch_optional(&self.db) + .await + .unwrap_or(Some(("error".to_string(), false))); - match row_o { - Some((username, is_admin)) => { - let groups = get_groups_for_user(&w_id.as_ref().unwrap(), - &username, &self.db) - .await - .ok() - .unwrap_or_default(); + match row_o { + Some((username, is_admin)) => { + let groups = get_groups_for_user( + &w_id.as_ref().unwrap(), + &username, + &self.db, + ) + .await + .ok() + .unwrap_or_default(); - Some(Authed { - email: Some(email), - username, - is_admin: is_admin || super_admin, - groups, - }) - }, - None if super_admin || w_id.unwrap() == "starter" => Some(Authed { - email: Some(email.to_string()), - username: email, - is_admin: super_admin, - groups: vec![], - }), - None => None - - } + Some(Authed { + email: Some(email), + username, + is_admin: is_admin || super_admin, + groups, + }) + } + None if super_admin || w_id.unwrap() == "starter" => { + Some(Authed { + email: Some(email.to_string()), + username: email, + is_admin: super_admin, + groups: vec![], + }) + } + None => None, + } } else { Some(Authed { email: Some(email.to_string()), @@ -160,21 +162,23 @@ impl AuthCache { (Some(owner), _, super_admin) if w_id.is_some() => { if let Some((prefix, name)) = owner.split_once('/') { if prefix == "u" { - - let is_admin = super_admin || sqlx::query_scalar!( - "SELECT is_admin FROM usr where username = $1 AND workspace_id = $2", - name, - &w_id.as_ref().unwrap() - ) - .fetch_one(&self.db) - .await - .ok() - .unwrap_or(false); - - let groups = get_groups_for_user(&w_id.unwrap(), &name, &self.db) + let is_admin = super_admin + || sqlx::query_scalar!( + "SELECT is_admin FROM usr where username = $1 AND \ + workspace_id = $2", + name, + &w_id.as_ref().unwrap() + ) + .fetch_one(&self.db) .await .ok() - .unwrap_or_default(); + .unwrap_or(false); + + let groups = + get_groups_for_user(&w_id.unwrap(), &name, &self.db) + .await + .ok() + .unwrap_or_default(); Some(Authed { email: None, @@ -217,7 +221,8 @@ impl AuthCache { async fn extract_token(req: &mut RequestParts) -> Option { let auth_header = req - .headers().get(http::header::AUTHORIZATION) + .headers() + .get(http::header::AUTHORIZATION) .and_then(|value| value.to_str().ok()) .and_then(|s| s.strip_prefix("Bearer ")); @@ -248,14 +253,14 @@ where Ok(tokened.clone()) } else { let token_o = extract_token(req).await; - if let Some(token) = token_o { - let tokened = Self { token }; - req.extensions_mut().insert(tokened.clone()); - Ok(tokened) - } else { - Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned())) - } + if let Some(token) = token_o { + let tokened = Self { token }; + req.extensions_mut().insert(tokened.clone()); + Ok(tokened) + } else { + Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned())) } + } } } @@ -320,10 +325,9 @@ pub struct User { pub created_at: chrono::DateTime, pub operator: bool, pub disabled: bool, - pub role: Option + pub role: Option, } - #[derive(FromRow, Serialize)] pub struct GlobalUserInfo { email: String, @@ -334,7 +338,6 @@ pub struct GlobalUserInfo { company: Option, } - #[derive(Serialize)] pub struct UserInfo { pub workspace_id: String, @@ -346,7 +349,7 @@ pub struct UserInfo { pub groups: Vec, pub operator: bool, pub disabled: bool, - pub role: Option + pub role: Option, } #[derive(FromRow, Serialize)] @@ -388,7 +391,7 @@ pub struct NewUser { pub password: String, pub super_admin: bool, pub name: Option, - pub company: Option + pub company: Option, } #[derive(Deserialize)] @@ -439,7 +442,6 @@ pub struct Login { pub password: String, } - #[derive(Deserialize)] pub struct Signup { pub email: String, @@ -458,20 +460,21 @@ struct WorkspaceUsername { pub username: String, } - async fn exists_username( authed: Authed, Extension(user_db): Extension, Path(w_id): Path, - Json(WorkspaceUsername { username }): Json + Json(WorkspaceUsername { username }): Json, ) -> JsonResult { let mut tx = user_db.begin(&authed).await?; let exists = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM usr WHERE workspace_id = $1 AND username = $2)", - &w_id, &username) - .fetch_one(&mut tx) - .await? - .unwrap_or(false); + "SELECT EXISTS(SELECT 1 FROM usr WHERE workspace_id = $1 AND username = $2)", + &w_id, + &username + ) + .fetch_one(&mut tx) + .await? + .unwrap_or(false); tx.commit().await?; Ok(Json(exists)) } @@ -479,7 +482,7 @@ async fn exists_username( async fn list_users( authed: Authed, Extension(user_db): Extension, - Path(w_id): Path + Path(w_id): Path, ) -> JsonResult> { let mut tx = user_db.begin(&authed).await?; let rows = sqlx::query_as!(User, "SELECT * from usr WHERE workspace_id = $1", &w_id) @@ -492,20 +495,25 @@ async fn list_users( async fn list_users_as_super_admin( authed: Authed, Extension(db): Extension, - Query(pagination): Query + Query(pagination): Query, ) -> JsonResult> { let mut tx = db.begin().await?; require_super_admin(&mut tx, authed.email).await?; let (per_page, offset) = crate::utils::paginate(pagination); - let rows = sqlx::query_as!(GlobalUserInfo, "SELECT email, login_type::text, verified, super_admin, name, company from password LIMIT $1 OFFSET $2", per_page as i32, offset as i32) - .fetch_all(&mut tx) - .await?; + let rows = sqlx::query_as!( + GlobalUserInfo, + "SELECT email, login_type::text, verified, super_admin, name, company from password LIMIT \ + $1 OFFSET $2", + per_page as i32, + offset as i32 + ) + .fetch_all(&mut tx) + .await?; tx.commit().await?; Ok(Json(rows)) } - // async fn list_invite_codes( // authed: Authed, // Extension(db): Extension, @@ -522,11 +530,10 @@ async fn list_users_as_super_admin( // Ok(Json(rows)) // } - async fn list_usernames( authed: Authed, Extension(user_db): Extension, - Path(w_id): Path + Path(w_id): Path, ) -> JsonResult> { let mut tx = user_db.begin(&authed).await?; let rows = sqlx::query_scalar!("SELECT username from usr WHERE workspace_id = $1", &w_id) @@ -541,14 +548,17 @@ async fn list_invites( Extension(db): Extension, ) -> JsonResult> { let mut tx = db.begin().await?; - let rows = sqlx::query_as!(WorkspaceInvite, "SELECT * from workspace_invite WHERE email = $1", authed.email) - .fetch_all(&mut tx) - .await?; + let rows = sqlx::query_as!( + WorkspaceInvite, + "SELECT * from workspace_invite WHERE email = $1", + authed.email + ) + .fetch_all(&mut tx) + .await?; tx.commit().await?; Ok(Json(rows)) } - async fn logout( Tokened { token }: Tokened, cookies: Cookies, @@ -577,7 +587,12 @@ async fn logout( async fn whoami( Extension(db): Extension, Path(w_id): Path, - Authed { username, email, is_admin, groups }: Authed, + Authed { + username, + email, + is_admin, + groups, + }: Authed, ) -> JsonResult { let user = get_user(&w_id, &username, &db).await?; if let Some(user) = user { @@ -603,27 +618,40 @@ async fn global_whoami( Extension(db): Extension, Authed { email, .. }: Authed, ) -> JsonResult { - let user: GlobalUserInfo = sqlx::query_as!(GlobalUserInfo, "SELECT email, login_type::TEXT, super_admin, verified, name, company FROM password WHERE email = $1", email) - .fetch_one(&db) - .await?; + let user: GlobalUserInfo = sqlx::query_as!( + GlobalUserInfo, + "SELECT email, login_type::TEXT, super_admin, verified, name, company FROM password WHERE \ + email = $1", + email + ) + .fetch_one(&db) + .await?; Ok(Json(user)) } -async fn get_email( - Authed { email, .. }: Authed, -) -> Result { - let email = email.ok_or(Error::BadRequest("current session does not correspond to an user with email".to_string()))?; +async fn get_email(Authed { email, .. }: Authed) -> Result { + let email = email.ok_or(Error::BadRequest( + "current session does not correspond to an user with email".to_string(), + ))?; Ok(email) } async fn get_user(w_id: &str, username: &str, db: &DB) -> Result> { - let user = sqlx::query_as!(User, "SELECT * FROM usr where username = $1 AND workspace_id = $2", username, w_id) - .fetch_optional(db) - .await?; - let is_super_admin = sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", user.as_ref().map(|x| &x.email)) - .fetch_optional(db) - .await? - .unwrap_or(false); + let user = sqlx::query_as!( + User, + "SELECT * FROM usr where username = $1 AND workspace_id = $2", + username, + w_id + ) + .fetch_optional(db) + .await?; + let is_super_admin = sqlx::query_scalar!( + "SELECT super_admin FROM password WHERE email = $1", + user.as_ref().map(|x| &x.email) + ) + .fetch_optional(db) + .await? + .unwrap_or(false); let groups = get_groups_for_user(&w_id, username, db).await?; Ok(user.map(|usr| UserInfo { groups, @@ -635,33 +663,36 @@ async fn get_user(w_id: &str, username: &str, db: &DB) -> Result Result> { - let groups = sqlx::query_scalar!("SELECT group_ FROM usr_to_group where usr = $1 AND workspace_id = $2", username, w_id) - .fetch_all(db) - .await?; + let groups = sqlx::query_scalar!( + "SELECT group_ FROM usr_to_group where usr = $1 AND workspace_id = $2", + username, + w_id + ) + .fetch_all(db) + .await?; Ok(groups) } -async fn whois(Extension(db): Extension, Path((w_id, username)): Path<(String, String)>) -> JsonResult { +async fn whois( + Extension(db): Extension, + Path((w_id, username)): Path<(String, String)>, +) -> JsonResult { let user_o = get_user(&w_id, &username, &db).await?; let user = crate::utils::not_found_if_none(user_o, "User", username)?; Ok(Json(user)) } - - - // async fn create_invite_code( // Authed { email, .. }: Authed, // Extension(db): Extension, // Json(nu): Json, // ) -> Result<(StatusCode, String)> { - // let mut tx = db.begin().await?; // require_super_admin(&mut tx, email).await?; @@ -688,7 +719,6 @@ async fn decline_invite( Extension(db): Extension, Json(nu): Json, ) -> Result<(StatusCode, String)> { - let mut tx = db.begin().await?; let email = email.unwrap_or("".to_string()); @@ -701,8 +731,8 @@ async fn decline_invite( .await?; audit_log( - &mut tx, - &email, + &mut tx, + &email, "users.decline_invite", ActionKind::Delete, &nu.workspace_id, @@ -715,7 +745,10 @@ async fn decline_invite( if is_admin.is_some() { Ok(( StatusCode::OK, - format!("user {} declined invite to workspace {}", &email, nu.workspace_id), + format!( + "user {} declined invite to workspace {}", + &email, nu.workspace_id + ), )) } else { Err(Error::NotFound(format!("invite for {email} not found"))) @@ -728,7 +761,7 @@ async fn accept_invite( Json(nu): Json, ) -> Result<(StatusCode, String)> { if &nu.username == "bot" { - return Err(Error::BadRequest("bot is a reserved username".to_string())) + return Err(Error::BadRequest("bot is a reserved username".to_string())); } let mut tx = db.begin().await?; @@ -746,7 +779,7 @@ async fn accept_invite( } audit_log( - &mut tx, + &mut tx, &nu.username, "users.accept_invite", ActionKind::Create, @@ -760,14 +793,23 @@ async fn accept_invite( if is_admin.is_some() { Ok(( StatusCode::CREATED, - format!("user {} accepted invite to workspace {}", &email, nu.workspace_id), + format!( + "user {} accepted invite to workspace {}", + &email, nu.workspace_id + ), )) } else { Err(Error::NotFound(format!("invite for {email} not found"))) } } -async fn add_user_to_workspace<'c>(w_id: &str, email: &str, username: &str, is_admin: bool, mut tx: sqlx::Transaction<'c, sqlx::Postgres>) -> error::Result> { +async fn add_user_to_workspace<'c>( + w_id: &str, + email: &str, + username: &str, + is_admin: bool, + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, +) -> error::Result> { sqlx::query!( "INSERT INTO usr (workspace_id, email, username, is_admin) @@ -789,7 +831,7 @@ async fn add_user_to_workspace<'c>(w_id: &str, email: &str, username: &str, is_a .execute(&mut tx) .await?; audit_log( - &mut tx, + &mut tx, username, "users.add_to_workspace", ActionKind::Create, @@ -802,7 +844,9 @@ async fn add_user_to_workspace<'c>(w_id: &str, email: &str, username: &str, is_a } async fn update_workspace_user( - Authed { username, is_admin, .. }: Authed, + Authed { + username, is_admin, .. + }: Authed, Extension(db): Extension, Path((w_id, username_to_update)): Path<(String, String)>, Json(eu): Json, @@ -812,7 +856,7 @@ async fn update_workspace_user( require_admin(is_admin, &username)?; if let Some(a) = eu.is_admin { - sqlx::query_scalar!( + sqlx::query_scalar!( "UPDATE usr SET is_admin = $1 WHERE username = $2 AND workspace_id = $3", a, &username_to_update, @@ -847,7 +891,7 @@ async fn update_user( require_super_admin(&mut tx, email.clone()).await?; if let Some(sa) = eu.is_super_admin { - sqlx::query_scalar!( + sqlx::query_scalar!( "UPDATE password SET super_admin = $1 WHERE email = $2", sa, &email_to_update @@ -881,7 +925,8 @@ async fn create_user( require_super_admin(&mut tx, email.clone()).await?; sqlx::query!( - "INSERT INTO password(email, verified, password_hash, login_type, super_admin, name, company) + "INSERT INTO password(email, verified, password_hash, login_type, super_admin, name, \ + company) VALUES ($1, $2, $3, 'password', $4, $5, $6)", &nu.email, true, @@ -893,7 +938,6 @@ async fn create_user( .execute(&mut tx) .await?; - audit_log( &mut tx, &email.unwrap(), @@ -908,14 +952,15 @@ async fn create_user( Ok((StatusCode::CREATED, format!("email {} created", nu.email))) } - pub fn owner_to_token_owner(user: &str, is_group: bool) -> String { let prefix = if is_group { 'g' } else { 'u' }; format!("{}/{}", prefix, user) } async fn delete_user( - Authed { username, is_admin, .. }: Authed, + Authed { + username, is_admin, .. + }: Authed, Extension(db): Extension, Path((w_id, username_to_delete)): Path<(String, String)>, ) -> Result { @@ -923,7 +968,6 @@ async fn delete_user( require_admin(is_admin, &username)?; - let email_to_delete_o = sqlx::query_scalar!( "SELECT email FROM usr where username = $1 AND workspace_id = $2", username_to_delete, @@ -956,22 +1000,29 @@ async fn delete_user( async fn set_password( Extension(db): Extension, Extension(argon2): Extension>>, - Authed { username, email, .. }: Authed, + Authed { + username, email, .. + }: Authed, Json(EditPassword { password }): Json, ) -> Result { let mut tx = db.begin().await?; - let email = email.ok_or("no_email").map_err(|e| Error::NotAuthorized(e.to_string()))?; + let email = email + .ok_or("no_email") + .map_err(|e| Error::NotAuthorized(e.to_string()))?; let custom_type = sqlx::query_scalar!( - "SELECT login_type::TEXT FROM password WHERE email = $1", - &email) - .fetch_one(&mut tx) - .await? - .unwrap_or("".to_string()); - + "SELECT login_type::TEXT FROM password WHERE email = $1", + &email + ) + .fetch_one(&mut tx) + .await? + .unwrap_or("".to_string()); + if custom_type != "password".to_string() { - return Err(Error::BadRequest(format!("login type for {email} is of type {custom_type}. Cannot set password."))) - } + return Err(Error::BadRequest(format!( + "login type for {email} is of type {custom_type}. Cannot set password." + ))); + } sqlx::query!( "UPDATE password SET password_hash = $1 WHERE email = $2", @@ -1012,7 +1063,6 @@ pub fn hash_password(argon2: Arc, password: String) -> Result { Ok(password_hash) } - // async fn lost_password( // Extension(db): Extension, // Extension(es): Extension>, @@ -1032,7 +1082,7 @@ pub fn hash_password(argon2: Arc, password: String) -> Result { // if !exists { // return Err(Error::NotFound(format!("no user found at email {email}"))) -// } +// } // let already = sqlx::query_scalar!( // "SELECT EXISTS(SELECT 1 FROM magic_link WHERE email = $1)", @@ -1082,7 +1132,6 @@ pub fn hash_password(argon2: Arc, password: String) -> Result { // } // } - // async fn signup( // TypedHeader(host): TypedHeader, // Extension(db): Extension, @@ -1097,14 +1146,12 @@ pub fn hash_password(argon2: Arc, password: String) -> Result { // ) -> Result<(StatusCode, String)> { // let mut tx = db.begin().await?; - // let email = sqlx::query_scalar!( // "INSERT INTO password (email, password_hash, name, company) VALUES ($1, $2, $3, $4) RETURNING email", // &email, &hash_password(argon2, password)?, name, company) // .fetch_optional(&mut tx) // .await?; - // if let Some(email) = email { // let tx = create_magic_link(&host.hostname(), &email, &es, tx).await?; // tx.commit().await?; @@ -1118,7 +1165,6 @@ pub fn hash_password(argon2: Arc, password: String) -> Result { // } // } - // async fn create_magic_link<'c>(host: &str, email: &str, es: &EmailSender, mut tx: sqlx::Transaction<'c, sqlx::Postgres>) -> error::Result> { // let token = gen_token(); @@ -1131,7 +1177,7 @@ pub fn hash_password(argon2: Arc, password: String) -> Result { // ) // .execute(&mut tx) // .await?; - + // let encoded_token = urlencoding::encode(&token); // let encoded_email = urlencoding::encode(email); // es.send_email(Message::builder() @@ -1157,20 +1203,17 @@ async fn login( cookies: Cookies, Extension(db): Extension, Extension(argon2): Extension>>, - Json(Login { - email, - password, - }): Json, + Json(Login { email, password }): Json, ) -> Result { let mut tx = db.begin().await?; - - let email_w_h: Option<(String, String, bool)> = sqlx::query_as( - "SELECT email, password_hash, super_admin FROM password WHERE email = $1 AND login_type = 'password'", - ) - .bind(&email) - .fetch_optional(&mut tx) - .await?; + let email_w_h: Option<(String, String, bool)> = sqlx::query_as( + "SELECT email, password_hash, super_admin FROM password WHERE email = $1 AND login_type = \ + 'password'", + ) + .bind(&email) + .fetch_optional(&mut tx) + .await?; if let Some((email, hash, super_admin)) = email_w_h { let parsed_hash = @@ -1190,7 +1233,12 @@ async fn login( } } -pub async fn create_session_token<'c>(email: &str, super_admin: bool, tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, cookies: Cookies) -> Result { +pub async fn create_session_token<'c>( + email: &str, + super_admin: bool, + tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, + cookies: Cookies, +) -> Result { let token = gen_token(); sqlx::query!( "INSERT INTO token @@ -1238,8 +1286,11 @@ pub async fn create_token_for_owner( .map(char::from) .collect(); let mut tx = db.begin().await?; - let is_super_admin = username.contains('@') && sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", - owner.split_once('/').map(|x| x.1).unwrap_or("")) + let is_super_admin = username.contains('@') + && sqlx::query_scalar!( + "SELECT super_admin FROM password WHERE email = $1", + owner.split_once('/').map(|x| x.1).unwrap_or("") + ) .fetch_optional(&mut tx) .await? .unwrap_or(false); @@ -1284,16 +1335,19 @@ pub async fn create_token_for_owner( async fn create_token( Extension(db): Extension, - Authed { email,.. }: Authed, + Authed { email, .. }: Authed, Json(new_token): Json, ) -> Result<(StatusCode, String)> { let token = gen_token(); let mut tx = db.begin().await?; - let email = email.ok_or_else(|| error::Error::BadRequest(format!("Only users with email can create tokens")))?; - let is_super_admin = sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email) - .fetch_optional(&mut tx) - .await? - .unwrap_or(false); + let email = email.ok_or_else(|| { + error::Error::BadRequest(format!("Only users with email can create tokens")) + })?; + let is_super_admin = + sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email) + .fetch_optional(&mut tx) + .await? + .unwrap_or(false); sqlx::query!( "INSERT INTO token (token, email, label, expiration, super_admin) @@ -1327,8 +1381,8 @@ async fn list_tokens( ) -> JsonResult> { let rows = sqlx::query_as!( TruncatedToken, - "SELECT label, concat(substring(token for 10)) as token_prefix, expiration, created_at, last_used_at FROM token \ - WHERE email = $1", + "SELECT label, concat(substring(token for 10)) as token_prefix, expiration, created_at, \ + last_used_at FROM token WHERE email = $1", email, ) .fetch_all(&db) @@ -1342,7 +1396,9 @@ async fn delete_token( Path(token_prefix): Path, ) -> Result { let mut tx = db.begin().await?; - let email = email.ok_or_else(|| error::Error::BadRequest(format!("Only users with email can create tokens")))?; + let email = email.ok_or_else(|| { + error::Error::BadRequest(format!("Only users with email can create tokens")) + })?; let tokens_deleted: Vec = sqlx::query_scalar( "DELETE FROM token WHERE email = $1 AND token LIKE concat($3, '%') RETURNING concat(substring(token for 10), '*****')", @@ -1387,13 +1443,13 @@ async fn leave_workspace( .await?; audit_log( - &mut tx, + &mut tx, &username, "users.leave_workspace", ActionKind::Delete, &w_id, -None, -None, + None, + None, ) .await?; tx.commit().await?; @@ -1401,7 +1457,6 @@ None, Ok(format!("left workspace {w_id}")) } - pub async fn delete_expired_items_perdiodically( db: &DB, mut rx: tokio::sync::broadcast::Receiver<()>, @@ -1415,12 +1470,11 @@ pub async fn delete_expired_items_perdiodically( .fetch_all(db) .await; - match tokens_deleted_r { Ok(tokens) => tracing::debug!("deleted {} tokens: {:?}", tokens.len(), tokens), Err(e) => tracing::error!("Error deleting token: {}", e.to_string()), } - + let magic_links_deleted_r: std::result::Result, _> = sqlx::query_scalar( "DELETE FROM magic_link WHERE expiration <= $1 RETURNING concat(substring(token for 10), '*****')", diff --git a/backend/src/variables.rs b/backend/src/variables.rs index 22a5f7d1a7..db91de05b5 100644 --- a/backend/src/variables.rs +++ b/backend/src/variables.rs @@ -93,47 +93,51 @@ pub fn get_reserved_variables( ContextualVariable { name: "WM_WORKSPACE".to_string(), value: w_id.to_string(), - description: "Workspace id of the current script".to_string() + description: "Workspace id of the current script".to_string(), }, ContextualVariable { name: "WM_TOKEN".to_string(), value: token.to_string(), - description: "Token ephemeral to the current script with equal permission to the permission of the run (Usable as a bearer token)".to_string() + description: "Token ephemeral to the current script with equal permission to the \ + permission of the run (Usable as a bearer token)" + .to_string(), }, ContextualVariable { name: "WM_EMAIL".to_string(), value: email.to_string(), - description: "Email of the user that executed the current script".to_string() + description: "Email of the user that executed the current script".to_string(), }, ContextualVariable { name: "WM_USERNAME".to_string(), value: username.to_string(), - description: "Username of the user that executed the current script".to_string() + description: "Username of the user that executed the current script".to_string(), }, ContextualVariable { name: "WM_JOB_ID".to_string(), value: job_id.to_string(), - description: "Job id of the current script".to_string() + description: "Job id of the current script".to_string(), }, ContextualVariable { name: "WM_JOB_PATH".to_string(), value: path.unwrap_or_else(|| "".to_string()), - description: "Path of the script or flow being run if any".to_string() + description: "Path of the script or flow being run if any".to_string(), }, ContextualVariable { name: "WM_FLOW_PATH".to_string(), value: flow_path.unwrap_or_else(|| "".to_string()), - description: "Path of the encapsulating flow if the job is a flow step".to_string() + description: "Path of the encapsulating flow if the job is a flow step".to_string(), }, ContextualVariable { name: "WM_SCHEDULE_PATH".to_string(), value: schedule_path.unwrap_or_else(|| "".to_string()), - description: "Path of the schedule if the job of the step or encapsulating step has been triggered by a schedule".to_string() + description: "Path of the schedule if the job of the step or encapsulating step has \ + been triggered by a schedule" + .to_string(), }, ContextualVariable { name: "WM_PERMISSIONED_AS".to_string(), value: permissioned_as.to_string(), - description: "Fully Qualified (u/g) owner name of executor of the job".to_string() + description: "Fully Qualified (u/g) owner name of executor of the job".to_string(), }, ] } @@ -168,8 +172,11 @@ async fn list_variables( let mut tx = user_db.begin(&authed).await?; let rows = sqlx::query_as::<_, ListableVariable>( - "SELECT workspace_id, path, CASE WHEN is_secret IS TRUE THEN null ELSE value::text END as value, is_secret, description, extra_perms, account, is_oauth, false as is_expired from variable - WHERE (workspace_id = $1 OR (is_secret IS NOT TRUE AND workspace_id = 'starter')) ORDER BY path", + "SELECT workspace_id, path, CASE WHEN is_secret IS TRUE THEN null ELSE value::text END as \ + value, is_secret, description, extra_perms, account, is_oauth, false as is_expired from \ + variable + WHERE (workspace_id = $1 OR (is_secret IS NOT TRUE AND workspace_id = 'starter')) ORDER \ + BY path", ) .bind(&w_id) .fetch_all(&mut tx) @@ -198,7 +205,8 @@ async fn get_variable( let variable_o = sqlx::query_as::<_, ListableVariable>( "SELECT variable.*, (now() > account.expires_at) as is_expired from variable LEFT JOIN account ON variable.account = account.id - WHERE variable.path = $1 AND (variable.workspace_id = $2 OR (is_secret IS NOT TRUE AND variable.workspace_id = 'starter')) + WHERE variable.path = $1 AND (variable.workspace_id = $2 OR (is_secret IS NOT TRUE AND \ + variable.workspace_id = 'starter')) LIMIT 1", ) .bind(&path) diff --git a/backend/src/worker.rs b/backend/src/worker.rs index e32f76c9bf..1f6ff31fc2 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -404,7 +404,10 @@ async fn handle_nondep_job( }; (code, reqs, job.language.to_owned()) } else { - sqlx::query_as::<_, (String, Option, Option)>("SELECT content, lock, language FROM script WHERE hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')") + sqlx::query_as::<_, (String, Option, Option)>( + "SELECT content, lock, language FROM script WHERE hash = $1 AND (workspace_id = $2 OR \ + workspace_id = 'starter')", + ) .bind(&job.script_hash.unwrap_or(ScriptHash(0)).0) .bind(&job.workspace_id) .fetch_optional(db) @@ -474,11 +477,29 @@ async fn handle_nondep_job( let _ = write_file(job_dir, "inner.py", &inner_content).await?; let sig = crate::parser::parse_python_signature(&inner_content)?; - let transforms = sig.args.into_iter().map(|x| match x.typ { - Typ::Bytes => format!("if \"{}\" in kwargs and kwargs[\"{}\"] is not None:\n kwargs[\"{}\"] = base64.b64decode(kwargs[\"{}\"])\n", x.name, x.name, x.name, x.name), - Typ::Datetime => format!("if \"{}\" in kwargs and kwargs[\"{}\"] is not None:\n kwargs[\"{}\"] = datetime.strptime(kwargs[\"{}\"], '%Y-%m-%dT%H:%M')\n", x.name, x.name, x.name, x.name), - _ => "".to_string() - }).collect::>().join(""); + let transforms = sig + .args + .into_iter() + .map(|x| match x.typ { + Typ::Bytes => { + format!( + "if \"{}\" in kwargs and kwargs[\"{}\"] is not None:\n \ + kwargs[\"{}\"] = base64.b64decode(kwargs[\"{}\"])\n", + x.name, x.name, x.name, x.name + ) + } + Typ::Datetime => { + format!( + "if \"{}\" in kwargs and kwargs[\"{}\"] is not None:\n \ + kwargs[\"{}\"] = datetime.strptime(kwargs[\"{}\"], \ + '%Y-%m-%dT%H:%M')\n", + x.name, x.name, x.name, x.name + ) + } + _ => "".to_string(), + }) + .collect::>() + .join(""); let tx = db.begin().await?; diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index 80c7eb7085..bb3d4c7d60 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -1,21 +1,19 @@ use std::collections::HashMap; -use crate::flows::{FlowModuleValue, FlowValue, InputTransform}; -use crate::jobs::{ - add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push, - script_path_to_payload, JobPayload, -}; -use crate::js_eval::{eval_timeout, EvalCreds}; -use crate::users::create_token_for_owner; use crate::{ db::DB, error::{self, Error}, - jobs::QueuedJob, + flows::{FlowModuleValue, FlowValue, InputTransform}, + jobs::{ + add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push, + script_path_to_payload, JobPayload, QueuedJob, + }, + js_eval::{eval_timeout, EvalCreds}, + users::create_token_for_owner, }; use async_recursion::async_recursion; use serde::{Deserialize, Serialize}; -use serde_json::json; -use serde_json::{Map, Value}; +use serde_json::{json, Map, Value}; use tracing::instrument; use uuid::Uuid; @@ -134,20 +132,22 @@ pub async fn update_flow_status_after_job_completion( ); let prev_step = old_status.step; - let (stop_early_expr, skip_if_stop_early) = sqlx::query_as::<_, (Option, Option)>(&format!( - "UPDATE queue + let (stop_early_expr, skip_if_stop_early) = + sqlx::query_as::<_, (Option, Option)>(&format!( + "UPDATE queue SET - flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {prev_step}}}', $1), '{{\"step\"}}', $2) + flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {prev_step}}}', $1), \ + '{{\"step\"}}', $2) WHERE id = $3 RETURNING (raw_flow->'modules'->{prev_step}->>'stop_after_if_expr'), (raw_flow->'modules'->{prev_step}->>'skip_if_stopped')::bool", - )) - .bind(serde_json::json!(new_status)) - .bind(serde_json::json!(step_counter)) - .bind(flow) - .fetch_one(&mut tx) - .await?; + )) + .bind(serde_json::json!(new_status)) + .bind(serde_json::json!(step_counter)) + .bind(flow) + .fetch_one(&mut tx) + .await?; tracing::debug!("UPDATE: {:?}", new_status); diff --git a/backend/src/workspaces.rs b/backend/src/workspaces.rs index e6bae4e17f..0d9d803adc 100644 --- a/backend/src/workspaces.rs +++ b/backend/src/workspaces.rs @@ -6,15 +6,27 @@ */ use crate::{ + audit::{audit_log, ActionKind}, db::{UserDB, DB}, error::{Error, JsonResult, Result}, - users::{Authed, WorkspaceInvite}, utils::{require_admin, require_super_admin, Pagination}, audit::{audit_log, ActionKind}, scripts::{Script, Schema}, resources::{Resource, ResourceType}, flows::Flow, variables::ListableVariable, + flows::Flow, + resources::{Resource, ResourceType}, + scripts::{Schema, Script}, + users::{Authed, WorkspaceInvite}, + utils::{require_admin, require_super_admin, Pagination}, + variables::ListableVariable, +}; +use axum::{ + body::StreamBody, + extract::{Extension, Path, Query}, + response::IntoResponse, + routing::{delete, get, post}, + Json, Router, }; -use axum::{extract::{Extension, Path, Query}, routing::{get, post, delete}, Json, Router, response::{IntoResponse}, body::StreamBody}; -use hyper::{StatusCode, header}; +use hyper::{header, StatusCode}; use serde::{Deserialize, Serialize}; -use sqlx::{FromRow}; +use sqlx::FromRow; use tempfile::TempDir; use tokio::fs::File; use tokio_util::io::ReaderStream; @@ -29,19 +41,16 @@ pub fn workspaced_service() -> Router { .route("/get_settings", get(get_settings)) .route("/edit_slack_command", post(edit_slack_command)) .route("/tarball", get(tarball_workspace)) - - - } pub fn global_service() -> Router { Router::new() - .route("/list_as_superadmin", get(list_workspaces_as_super_admin)) - .route("/list", get(list_workspaces)) - .route("/users", get(user_workspaces)) - .route("/create", post(create_workspace)) - .route("/exists", post(exists_workspace)) - .route("/exists_username", post(exists_username)) + .route("/list_as_superadmin", get(list_workspaces_as_super_admin)) + .route("/list", get(list_workspaces)) + .route("/users", get(user_workspaces)) + .route("/create", post(create_workspace)) + .route("/exists", post(exists_workspace)) + .route("/exists_username", post(exists_username)) } #[derive(FromRow, Serialize)] @@ -51,7 +60,7 @@ struct Workspace { owner: String, domain: Option, deleted: bool, - premium: bool + premium: bool, } #[derive(FromRow, Serialize, Debug)] @@ -59,20 +68,18 @@ pub struct WorkspaceSettings { pub workspace_id: String, pub slack_team_id: Option, pub slack_name: Option, - pub slack_command_script: Option + pub slack_command_script: Option, } - #[derive(sqlx::Type, Serialize, Deserialize, Debug)] #[sqlx(type_name = "WORKSPACE_KEY_KIND", rename_all = "lowercase")] pub enum WorkspaceKeyKind { - Cloud + Cloud, } - #[derive(Deserialize)] struct EditCommandScript { - slack_command_script: Option + slack_command_script: Option, } #[derive(Deserialize)] struct CreateWorkspace { @@ -82,15 +89,13 @@ struct CreateWorkspace { domain: Option, } - #[derive(Deserialize)] struct EditWorkspace { name: String, owner: String, - domain: Option + domain: Option, } - #[derive(Serialize)] struct WorkspaceList { pub email: String, @@ -104,7 +109,6 @@ struct UserWorkspace { pub username: String, } - #[derive(Deserialize)] struct WorkspaceId { pub id: String, @@ -116,14 +120,12 @@ struct ValidateUsername { pub username: String, } - #[derive(Deserialize)] pub struct NewWorkspaceInvite { pub email: String, pub is_admin: bool, } - async fn list_pending_invites( authed: Authed, Extension(user_db): Extension, @@ -131,26 +133,30 @@ async fn list_pending_invites( ) -> JsonResult> { require_admin(authed.is_admin, &authed.username)?; let mut tx = user_db.begin(&authed).await?; - let rows = sqlx::query_as!(WorkspaceInvite, "SELECT * from workspace_invite WHERE workspace_id = $1", w_id) - .fetch_all(&mut tx) - .await?; + let rows = sqlx::query_as!( + WorkspaceInvite, + "SELECT * from workspace_invite WHERE workspace_id = $1", + w_id + ) + .fetch_all(&mut tx) + .await?; tx.commit().await?; Ok(Json(rows)) } - async fn exists_workspace( authed: Authed, Extension(user_db): Extension, - Json(WorkspaceId { id }): Json + Json(WorkspaceId { id }): Json, ) -> JsonResult { let mut tx = user_db.begin(&authed).await?; let exists = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM workspace WHERE workspace.id = $1)", - id) - .fetch_one(&mut tx) - .await? - .unwrap_or(false); + "SELECT EXISTS(SELECT 1 FROM workspace WHERE workspace.id = $1)", + id + ) + .fetch_one(&mut tx) + .await? + .unwrap_or(false); tx.commit().await?; Ok(Json(exists)) } @@ -161,16 +167,17 @@ async fn list_workspaces( ) -> JsonResult> { let mut tx = user_db.begin(&authed).await?; let workspaces = sqlx::query_as!( - Workspace, - "SELECT workspace.* FROM workspace, usr WHERE usr.workspace_id = workspace.id AND usr.email = $1 AND deleted = false", - authed.email.as_ref()) - .fetch_all(&mut tx) - .await?; + Workspace, + "SELECT workspace.* FROM workspace, usr WHERE usr.workspace_id = workspace.id AND \ + usr.email = $1 AND deleted = false", + authed.email.as_ref() + ) + .fetch_all(&mut tx) + .await?; tx.commit().await?; Ok(Json(workspaces)) } - async fn get_settings( authed: Authed, Path(w_id): Path, @@ -178,11 +185,12 @@ async fn get_settings( ) -> JsonResult { let mut tx = user_db.begin(&authed).await?; let settings = sqlx::query_as!( - WorkspaceSettings, - "SELECT * FROM workspace_settings WHERE workspace_id = $1", - &w_id) - .fetch_one(&mut tx) - .await?; + WorkspaceSettings, + "SELECT * FROM workspace_settings WHERE workspace_id = $1", + &w_id + ) + .fetch_one(&mut tx) + .await?; tx.commit().await?; Ok(Json(settings)) } @@ -191,8 +199,10 @@ async fn edit_slack_command( authed: Authed, Extension(db): Extension, Path(w_id): Path, - Authed { is_admin, username, .. }: Authed, - Json(es): Json + Authed { + is_admin, username, .. + }: Authed, + Json(es): Json, ) -> Result { require_admin(is_admin, &username)?; let mut tx = db.begin().await?; @@ -205,15 +215,21 @@ async fn edit_slack_command( .await?; audit_log( - &mut tx, + &mut tx, &authed.username, "workspaces.edit_command_script", ActionKind::Update, &w_id, -Some(&authed.email.unwrap()), -Some([ - ("script", es.slack_command_script.unwrap_or("NO_SCRIPT".to_string()).as_str()) - ].into()), + Some(&authed.email.unwrap()), + Some( + [( + "script", + es.slack_command_script + .unwrap_or("NO_SCRIPT".to_string()) + .as_str(), + )] + .into(), + ), ) .await?; tx.commit().await?; @@ -221,7 +237,6 @@ Some([ Ok(format!("Edit command script {}", &w_id)) } - async fn list_workspaces_as_super_admin( authed: Authed, Extension(user_db): Extension, @@ -233,10 +248,13 @@ async fn list_workspaces_as_super_admin( let (per_page, offset) = crate::utils::paginate(pagination); let workspaces = sqlx::query_as!( - Workspace, - "SELECT * FROM workspace LIMIT $1 OFFSET $2", per_page as i32, offset as i32) - .fetch_all(&mut tx) - .await?; + Workspace, + "SELECT * FROM workspace LIMIT $1 OFFSET $2", + per_page as i32, + offset as i32 + ) + .fetch_all(&mut tx) + .await?; tx.commit().await?; Ok(Json(workspaces)) } @@ -250,12 +268,14 @@ async fn user_workspaces( .map_err(|x| Error::NotAuthorized(x.to_string()))?; let mut tx = db.begin().await?; let workspaces = sqlx::query_as!( - UserWorkspace, - "SELECT workspace.id, workspace.name, usr.username - FROM workspace, usr WHERE usr.workspace_id = workspace.id AND usr.email = $1 AND deleted = false", - email) - .fetch_all(&mut tx) - .await?; + UserWorkspace, + "SELECT workspace.id, workspace.name, usr.username + FROM workspace, usr WHERE usr.workspace_id = workspace.id AND usr.email = $1 AND deleted = \ + false", + email + ) + .fetch_all(&mut tx) + .await?; tx.commit().await?; Ok(Json(WorkspaceList { email, workspaces })) } @@ -263,10 +283,10 @@ async fn user_workspaces( async fn create_workspace( authed: Authed, Extension(db): Extension, - Json(nw): Json + Json(nw): Json, ) -> Result { if &nw.username == "bot" { - return Err(Error::BadRequest("bot is a reserved username".to_string())) + return Err(Error::BadRequest("bot is a reserved username".to_string())); } let mut tx = db.begin().await?; sqlx::query!( @@ -293,7 +313,8 @@ async fn create_workspace( "INSERT INTO workspace_key (workspace_id, kind, key) VALUES ($1, 'cloud', $2)", - nw.id, &key + nw.id, + &key ) .execute(&mut tx) .await?; @@ -341,7 +362,7 @@ async fn create_workspace( .await?; audit_log( - &mut tx, + &mut tx, &authed.username, "workspaces.create", ActionKind::Create, @@ -351,7 +372,6 @@ async fn create_workspace( ) .await?; tx.commit().await?; - Ok(format!("Created workspace {}", &nw.id)) } @@ -360,8 +380,10 @@ async fn edit_workspace( authed: Authed, Extension(db): Extension, Path(w_id): Path, - Authed { is_admin, username, .. }: Authed, - Json(ew): Json + Authed { + is_admin, username, .. + }: Authed, + Json(ew): Json, ) -> Result { require_admin(is_admin, &username)?; let mut tx = db.begin().await?; @@ -376,15 +398,19 @@ async fn edit_workspace( .await?; audit_log( - &mut tx, + &mut tx, &authed.username, "workspaces.update", ActionKind::Update, &w_id, -Some(&authed.email.unwrap()), -Some([ - ("domain", ew.domain.unwrap_or("NO_DOMAIN".to_string()).as_str()) - ].into()), + Some(&authed.email.unwrap()), + Some( + [( + "domain", + ew.domain.unwrap_or("NO_DOMAIN".to_string()).as_str(), + )] + .into(), + ), ) .await?; tx.commit().await?; @@ -395,25 +421,27 @@ Some([ async fn delete_workspace( Extension(db): Extension, Path(w_id): Path, - Authed { is_admin, username, email, .. }: Authed, + Authed { + is_admin, + username, + email, + .. + }: Authed, ) -> Result { require_admin(is_admin, &username)?; let mut tx = db.begin().await?; - sqlx::query!( - "UPDATE workspace SET deleted = true WHERE id = $1", - &w_id - ) - .execute(&mut tx) - .await?; + sqlx::query!("UPDATE workspace SET deleted = true WHERE id = $1", &w_id) + .execute(&mut tx) + .await?; audit_log( - &mut tx, + &mut tx, &username, "workspaces.delete", ActionKind::Update, &w_id, -Some(&email.unwrap_or("noemail".to_string())), -None, + Some(&email.unwrap_or("noemail".to_string())), + None, ) .await?; tx.commit().await?; @@ -421,14 +449,14 @@ None, Ok(format!("Deleted workspace {}", &w_id)) } - async fn invite_user( - Authed { username, is_admin, .. }: Authed, + Authed { + username, is_admin, .. + }: Authed, Extension(db): Extension, Path(w_id): Path, Json(nu): Json, ) -> Result<(StatusCode, String)> { - require_admin(is_admin, &username)?; let mut tx = db.begin().await?; @@ -452,14 +480,14 @@ async fn invite_user( )) } - async fn delete_invite( - Authed { username, is_admin, .. }: Authed, + Authed { + username, is_admin, .. + }: Authed, Extension(db): Extension, Path(w_id): Path, Json(nu): Json, ) -> Result<(StatusCode, String)> { - require_admin(is_admin, &username)?; let mut tx = db.begin().await?; @@ -486,7 +514,6 @@ async fn exists_username( Extension(db): Extension, Json(vu): Json, ) -> Result { - let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM usr WHERE username = $1 AND workspace_id = $2)", vu.username, @@ -497,7 +524,7 @@ async fn exists_username( .unwrap_or(true); if exists { - return Err(Error::BadRequest("username already taken".to_string())) + return Err(Error::BadRequest("username already taken".to_string())); } Ok("valid username".to_string()) @@ -509,55 +536,67 @@ struct ScriptMetadata { description: String, schema: Option, is_template: bool, - lock: Vec + lock: Vec, } async fn tarball_workspace( authed: Authed, Extension(db): Extension, Path(w_id): Path, -) -> Result<([(headers::HeaderName, String); 2], impl IntoResponse)> { +) -> Result<([(headers::HeaderName, String); 2], impl IntoResponse)> { require_admin(authed.is_admin, &authed.username)?; let tmp_dir = TempDir::new_in(".")?; - + let name = format!("windmill-{w_id}.tar"); let file_path = tmp_dir.path().join(&name); let file = File::create(&file_path).await?; let mut a = tokio_tar::Builder::new(file); - { 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(&db) - .await?; - - for script in scripts { - write_to_archive(script.content, format!("scripts/{}.py", script.path), &mut a).await?; - - let lock = script.lock.unwrap_or_else(|| "".to_string()) - .lines() - .map(|x| x.to_string()) - .collect(); - let metadata = ScriptMetadata { - summary: script.summary, description: script.description, schema: script.schema, is_template: script.is_template, lock }; - let metadata_str = serde_json::to_string_pretty(&metadata).unwrap(); - write_to_archive(metadata_str, format!("scripts/{}.json", script.path), &mut a).await?; - - }; -} - - + "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(&db) + .await?; + for script in scripts { + write_to_archive( + script.content, + format!("scripts/{}.py", script.path), + &mut a, + ) + .await?; + let lock = script + .lock + .unwrap_or_else(|| "".to_string()) + .lines() + .map(|x| x.to_string()) + .collect(); + let metadata = ScriptMetadata { + summary: script.summary, + description: script.description, + schema: script.schema, + is_template: script.is_template, + lock, + }; + let metadata_str = serde_json::to_string_pretty(&metadata).unwrap(); + write_to_archive( + metadata_str, + format!("scripts/{}.json", script.path), + &mut a, + ) + .await?; + } + } { - - let resources = sqlx::query_as!(Resource, + let resources = sqlx::query_as!( + Resource, "SELECT * FROM resource WHERE workspace_id = $1", &w_id ) @@ -566,27 +605,38 @@ async fn tarball_workspace( for resource in resources { let resource_str = serde_json::to_string_pretty(&resource).unwrap(); - write_to_archive(resource_str, format!("resources/{}.json", resource.path), &mut a).await?; + write_to_archive( + resource_str, + format!("resources/{}.json", resource.path), + &mut a, + ) + .await?; } } { - let resource_types = sqlx::query_as!(ResourceType, - "SELECT * FROM resource_type WHERE workspace_id = $1", - &w_id - ) - .fetch_all(&db) - .await?; + let resource_types = sqlx::query_as!( + ResourceType, + "SELECT * FROM resource_type WHERE workspace_id = $1", + &w_id + ) + .fetch_all(&db) + .await?; for resource_type in resource_types { let resource_str = serde_json::to_string_pretty(&resource_type).unwrap(); - write_to_archive(resource_str, format!("resource_types/{}.json", resource_type.name), &mut a).await?; + write_to_archive( + resource_str, + format!("resource_types/{}.json", resource_type.name), + &mut a, + ) + .await?; } } { let flows = sqlx::query_as::<_, Flow>( - "SELECT * FROM flow WHERE workspace_id = $1 AND archived = false" + "SELECT * FROM flow WHERE workspace_id = $1 AND archived = false", ) .bind(&w_id) .fetch_all(&db) @@ -599,8 +649,8 @@ async fn tarball_workspace( } { - let variables = sqlx::query_as::<_, ListableVariable>( - "SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false" + let variables = sqlx::query_as::<_, ListableVariable>( + "SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false", ) .bind(&w_id) .fetch_all(&db) @@ -629,7 +679,11 @@ async fn tarball_workspace( Ok((headers, body)) } -async fn write_to_archive(content: String, path: String, a: &mut tokio_tar::Builder) -> Result<()> { +async fn write_to_archive( + content: String, + path: String, + a: &mut tokio_tar::Builder, +) -> Result<()> { let bytes = content.as_bytes(); let mut header = tokio_tar::Header::new_gnu(); header.set_size(bytes.len() as u64);