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

cargo fmt

Browse Source
This commit is contained in:
Ruben Fiszel
2022-07-29 21:29:59 +02:00
parent fed1849256
commit d4b5a0ea3d
16 changed files with 731 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/rustfmt.toml
-5
View File
@@ -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
backend/src/flows.rs
+10 -5
View File
@@ -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
)
backend/src/granular_acls.rs
+4 -2
View File
@@ -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)
backend/src/jobs.rs
+56 -36
View File
@@ -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<ScriptHash> {
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<String> {
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::<FlowValue>(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,
backend/src/js_eval.rs
+10 -16
View File
@@ -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);
backend/src/main.rs
+3 -1
View File
@@ -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(),
backend/src/oauth2.rs
+41 -30
View File
@@ -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)
backend/src/parser.rs
+63 -60
View File
@@ -127,8 +127,7 @@ pub fn parse_python_signature(code: &str) -> error::Result<MainArgSignature> {
}
}
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),
))
backend/src/resources.rs
+21 -9
View File
@@ -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<DB>,
Path(w_id): Path<String>,
) -> JsonResult<Vec<ResourceType>> {
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<DB>,
Path(w_id): Path<String>,
) -> JsonResult<Vec<String>> {
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
)
backend/src/schedule.rs
+11 -6
View File
@@ -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,
backend/src/scripts.rs
+20 -13
View File
@@ -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<DeploymentStatus> {
let mut tx = user_db.begin(&authed).await?;
let status_o: Option<DeploymentStatus> = 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<DeploymentStatus> = 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)
backend/src/users.rs
+237 -183
View File
@@ -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<String>, token: &str) -> Option<Authed> {
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<String>, Option<String>, 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<B: Send>(req: &mut RequestParts<B>) -> Option<String> {
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<chrono::Utc>,
pub operator: bool,
pub disabled: bool,
pub role: Option<String>
pub role: Option<String>,
}
#[derive(FromRow, Serialize)]
pub struct GlobalUserInfo {
email: String,
@@ -334,7 +338,6 @@ pub struct GlobalUserInfo {
company: Option<String>,
}
#[derive(Serialize)]
pub struct UserInfo {
pub workspace_id: String,
@@ -346,7 +349,7 @@ pub struct UserInfo {
pub groups: Vec<String>,
pub operator: bool,
pub disabled: bool,
pub role: Option<String>
pub role: Option<String>,
}
#[derive(FromRow, Serialize)]
@@ -388,7 +391,7 @@ pub struct NewUser {
pub password: String,
pub super_admin: bool,
pub name: Option<String>,
pub company: Option<String>
pub company: Option<String>,
}
#[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<UserDB>,
Path(w_id): Path<String>,
Json(WorkspaceUsername { username }): Json<WorkspaceUsername>
Json(WorkspaceUsername { username }): Json<WorkspaceUsername>,
) -> JsonResult<bool> {
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<UserDB>,
Path(w_id): Path<String>
Path(w_id): Path<String>,
) -> JsonResult<Vec<User>> {
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<DB>,
Query(pagination): Query<Pagination>
Query(pagination): Query<Pagination>,
) -> JsonResult<Vec<GlobalUserInfo>> {
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<DB>,
@@ -522,11 +530,10 @@ async fn list_users_as_super_admin(
// Ok(Json(rows))
// }
async fn list_usernames(
authed: Authed,
Extension(user_db): Extension<UserDB>,
Path(w_id): Path<String>
Path(w_id): Path<String>,
) -> JsonResult<Vec<String>> {
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<DB>,
) -> JsonResult<Vec<WorkspaceInvite>> {
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<DB>,
Path(w_id): Path<String>,
Authed { username, email, is_admin, groups }: Authed,
Authed {
username,
email,
is_admin,
groups,
}: Authed,
) -> JsonResult<UserInfo> {
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<DB>,
Authed { email, .. }: Authed,
) -> JsonResult<GlobalUserInfo> {
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<String> {
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<String> {
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<Option<UserInfo>> {
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<Option<UserInfo
created_at: usr.created_at,
operator: usr.operator,
disabled: usr.disabled,
role: usr.role
role: usr.role,
}))
}
async fn get_groups_for_user(w_id: &str, username: &str, db: &DB) -> Result<Vec<String>> {
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<DB>, Path((w_id, username)): Path<(String, String)>) -> JsonResult<UserInfo> {
async fn whois(
Extension(db): Extension<DB>,
Path((w_id, username)): Path<(String, String)>,
) -> JsonResult<UserInfo> {
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<DB>,
// Json(nu): Json<NewInviteCode>,
// ) -> 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<DB>,
Json(nu): Json<DeclineInvite>,
) -> 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<AcceptInvite>,
) -> 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<sqlx::Transaction<'c, sqlx::Postgres>> {
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::Transaction<'c, sqlx::Postgres>> {
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<DB>,
Path((w_id, username_to_update)): Path<(String, String)>,
Json(eu): Json<EditWorkspaceUser>,
@@ -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<DB>,
Path((w_id, username_to_delete)): Path<(String, String)>,
) -> Result<String> {
@@ -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<DB>,
Extension(argon2): Extension<Arc<Argon2<'_>>>,
Authed { username, email, .. }: Authed,
Authed {
username, email, ..
}: Authed,
Json(EditPassword { password }): Json<EditPassword>,
) -> Result<String> {
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<Argon2>, password: String) -> Result<String> {
Ok(password_hash)
}
// async fn lost_password(
// Extension(db): Extension<DB>,
// Extension(es): Extension<Arc<EmailSender>>,
@@ -1032,7 +1082,7 @@ pub fn hash_password(argon2: Arc<Argon2>, password: String) -> Result<String> {
// 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<Argon2>, password: String) -> Result<String> {
// }
// }
// async fn signup(
// TypedHeader(host): TypedHeader<headers::Host>,
// Extension(db): Extension<DB>,
@@ -1097,14 +1146,12 @@ pub fn hash_password(argon2: Arc<Argon2>, password: String) -> Result<String> {
// ) -> 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<Argon2>, password: String) -> Result<String> {
// }
// }
// async fn create_magic_link<'c>(host: &str, email: &str, es: &EmailSender, mut tx: sqlx::Transaction<'c, sqlx::Postgres>) -> error::Result<sqlx::Transaction<'c, sqlx::Postgres>> {
// let token = gen_token();
@@ -1131,7 +1177,7 @@ pub fn hash_password(argon2: Arc<Argon2>, password: String) -> Result<String> {
// )
// .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<DB>,
Extension(argon2): Extension<Arc<Argon2<'_>>>,
Json(Login {
email,
password,
}): Json<Login>,
Json(Login { email, password }): Json<Login>,
) -> Result<String> {
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<String> {
pub async fn create_session_token<'c>(
email: &str,
super_admin: bool,
tx: &mut sqlx::Transaction<'c, sqlx::Postgres>,
cookies: Cookies,
) -> Result<String> {
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<DB>,
Authed { email,.. }: Authed,
Authed { email, .. }: Authed,
Json(new_token): Json<NewToken>,
) -> 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<Vec<TruncatedToken>> {
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<String>,
) -> Result<String> {
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<String> = 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<Vec<String>, _> = sqlx::query_scalar(
"DELETE FROM magic_link WHERE expiration <= $1
RETURNING concat(substring(token for 10), '*****')",
backend/src/variables.rs
+20 -12
View File
@@ -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)
backend/src/worker.rs
+27 -6
View File
@@ -404,7 +404,10 @@ async fn handle_nondep_job(
};
(code, reqs, job.language.to_owned())
} else {
sqlx::query_as::<_, (String, Option<String>, Option<ScriptLang>)>("SELECT content, lock, language FROM script WHERE hash = $1 AND (workspace_id = $2 OR workspace_id = 'starter')")
sqlx::query_as::<_, (String, Option<String>, Option<ScriptLang>)>(
"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::<Vec<String>>().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::<Vec<String>>()
.join("");
let tx = db.begin().await?;
backend/src/worker_flow.rs
+19 -19
View File
@@ -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<String>, Option<bool>)>(&format!(
"UPDATE queue
let (stop_early_expr, skip_if_stop_early) =
sqlx::query_as::<_, (Option<String>, Option<bool>)>(&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);
backend/src/workspaces.rs
+189 -135
View File
@@ -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<String>,
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<String>,
pub slack_name: Option<String>,
pub slack_command_script: Option<String>
pub slack_command_script: Option<String>,
}
#[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<String>
slack_command_script: Option<String>,
}
#[derive(Deserialize)]
struct CreateWorkspace {
@@ -82,15 +89,13 @@ struct CreateWorkspace {
domain: Option<String>,
}
#[derive(Deserialize)]
struct EditWorkspace {
name: String,
owner: String,
domain: Option<String>
domain: Option<String>,
}
#[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<UserDB>,
@@ -131,26 +133,30 @@ async fn list_pending_invites(
) -> JsonResult<Vec<WorkspaceInvite>> {
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<UserDB>,
Json(WorkspaceId { id }): Json<WorkspaceId>
Json(WorkspaceId { id }): Json<WorkspaceId>,
) -> JsonResult<bool> {
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<Vec<Workspace>> {
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<String>,
@@ -178,11 +185,12 @@ async fn get_settings(
) -> JsonResult<WorkspaceSettings> {
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<DB>,
Path(w_id): Path<String>,
Authed { is_admin, username, .. }: Authed,
Json(es): Json<EditCommandScript>
Authed {
is_admin, username, ..
}: Authed,
Json(es): Json<EditCommandScript>,
) -> Result<String> {
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<UserDB>,
@@ -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<DB>,
Json(nw): Json<CreateWorkspace>
Json(nw): Json<CreateWorkspace>,
) -> Result<String> {
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<DB>,
Path(w_id): Path<String>,
Authed { is_admin, username, .. }: Authed,
Json(ew): Json<EditWorkspace>
Authed {
is_admin, username, ..
}: Authed,
Json(ew): Json<EditWorkspace>,
) -> Result<String> {
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<DB>,
Path(w_id): Path<String>,
Authed { is_admin, username, email, .. }: Authed,
Authed {
is_admin,
username,
email,
..
}: Authed,
) -> Result<String> {
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<DB>,
Path(w_id): Path<String>,
Json(nu): Json<NewWorkspaceInvite>,
) -> 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<DB>,
Path(w_id): Path<String>,
Json(nu): Json<NewWorkspaceInvite>,
) -> 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<DB>,
Json(vu): Json<ValidateUsername>,
) -> Result<String> {
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<Schema>,
is_template: bool,
lock: Vec<String>
lock: Vec<String>,
}
async fn tarball_workspace(
authed: Authed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
) -> 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<File>) -> Result<()> {
async fn write_to_archive(
content: String,
path: String,
a: &mut tokio_tar::Builder<File>,
) -> Result<()> {
let bytes = content.as_bytes();
let mut header = tokio_tar::Header::new_gnu();
header.set_size(bytes.len() as u64);