diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 2ac87dd037..6c30388327 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1986,6 +1986,11 @@ async fn run_preview_job( Json(preview): Json, ) -> error::Result<(StatusCode, String)> { check_scopes(&authed, || format!("runscript"))?; + if authed.is_operator { + return Err(error::Error::NotAuthorized( + "Operators cannot run preview jobs for security reasons".to_string(), + )); + } let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default()); @@ -2075,6 +2080,11 @@ async fn run_preview_flow_job( Json(raw_flow): Json, ) -> error::Result<(StatusCode, String)> { check_scopes(&authed, || format!("runflow"))?; + if authed.is_operator { + return Err(error::Error::NotAuthorized( + "Operators cannot run preview jobs for security reasons".to_string(), + )); + } let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default()); diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index 751463099a..38d18170cd 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -169,9 +169,11 @@ async fn edit_schedule( path, w_id ) - .fetch_one(&mut tx) + .fetch_optional(&mut tx) .await?; + let is_flow = not_found_if_none(is_flow, "Schedule", &path)?; + clear_schedule(tx.transaction_mut(), path, is_flow).await?; let schedule = sqlx::query_as!( Schedule, diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 994122d6b6..b8839f2a0e 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -136,17 +136,24 @@ impl AuthCache { (Some(owner), email, 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 \ + let (is_admin, is_operator) = if super_admin { + (true, false) + } else { + let r = sqlx::query!( + "SELECT is_admin, operator FROM usr where username = $1 AND \ workspace_id = $2 AND disabled = false", name, &w_id.as_ref().unwrap() ) .fetch_one(&self.db) .await - .ok() - .unwrap_or(false); + .ok(); + if let Some(r) = r { + (r.is_admin, r.operator) + } else { + (false, true) + } + }; let w_id = &w_id.unwrap(); let groups = get_groups_for_user(w_id, &name, &self.db) @@ -165,6 +172,7 @@ impl AuthCache { .unwrap_or_else(|| "missing@email.xyz".to_string()), username: name.to_string(), is_admin, + is_operator, groups, folders, scopes: None, @@ -186,6 +194,7 @@ impl AuthCache { username: format!("group-{name}"), is_admin: false, groups, + is_operator: false, folders, scopes: None, }) @@ -198,6 +207,7 @@ impl AuthCache { .unwrap_or_else(|| "missing@email.xyz".to_string()), username: owner, is_admin: super_admin, + is_operator: true, groups, folders, scopes: None, @@ -206,18 +216,18 @@ impl AuthCache { } (_, Some(email), super_admin, scopes) => { if w_id.is_some() { - let row_o = sqlx::query_as::<_, (String, bool)>( - "SELECT username, is_admin FROM usr where email = $1 AND \ + let row_o = sqlx::query_as::<_, (String, bool, bool)>( + "SELECT username, is_admin, operator FROM usr where email = $1 AND \ workspace_id = $2 AND disabled = false", ) .bind(&email) .bind(&w_id.as_ref().unwrap()) .fetch_optional(&self.db) .await - .unwrap_or(Some(("error".to_string(), false))); + .unwrap_or(Some(("error".to_string(), false, false))); match row_o { - Some((username, is_admin)) => { + Some((username, is_admin, is_operator)) => { let groups = get_groups_for_user( &w_id.as_ref().unwrap(), &username, @@ -240,6 +250,7 @@ impl AuthCache { email, username, is_admin: is_admin || super_admin, + is_operator, groups, folders, scopes, @@ -249,6 +260,7 @@ impl AuthCache { email: email.clone(), username: email, is_admin: super_admin, + is_operator: false, groups: vec![], folders: vec![], scopes, @@ -260,6 +272,7 @@ impl AuthCache { email: email.to_string(), username: email, is_admin: super_admin, + is_operator: true, groups: Vec::new(), folders: Vec::new(), scopes, @@ -285,6 +298,7 @@ impl AuthCache { email: SUPERADMIN_SECRET_EMAIL.to_string(), username: "superadmin_secret".to_string(), is_admin: true, + is_operator: false, groups: Vec::new(), folders: Vec::new(), scopes: None, @@ -366,6 +380,7 @@ pub struct Authed { pub email: String, pub username: String, pub is_admin: bool, + pub is_operator: bool, pub groups: Vec, // (folder name, can write, is owner) pub folders: Vec<(String, bool, bool)>, diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 961232f383..5e374e5a80 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -34,6 +34,7 @@ use magic_crypt::MagicCryptTrait; #[cfg(feature = "enterprise")] use stripe::CustomerId; use windmill_audit::{audit_log, ActionKind}; +use windmill_common::schedule::Schedule; use windmill_common::users::username_to_permissioned_as; use windmill_common::{ error::{to_anyhow, Error, JsonResult, Result}, @@ -1368,6 +1369,7 @@ struct ArchiveQueryParams { skip_secrets: Option, skip_variables: Option, skip_resources: Option, + include_schedules: Option, } #[inline] @@ -1395,6 +1397,7 @@ where "archived", "has_draft", "draft_only", + "error" ] { if obj.contains_key(key) { obj.remove(key); @@ -1422,6 +1425,7 @@ async fn tarball_workspace( skip_resources, skip_secrets, skip_variables, + include_schedules, }): Query, ) -> Result<([(headers::HeaderName, String); 2], impl IntoResponse)> { require_admin(authed.is_admin, &authed.username)?; @@ -1605,6 +1609,25 @@ async fn tarball_workspace( .await?; } } + + if include_schedules.unwrap_or(false) { + let schedules = sqlx::query_as!( + Schedule, + "SELECT * FROM schedule + WHERE workspace_id = $1", + &w_id + ) + .fetch_all(&db) + .await?; + + for schedule in schedules { + let app_str = &to_string_without_metadata(&schedule, false).unwrap(); + archive + .write_to_archive(&app_str, &format!("{}.schedule.json", schedule.path)) + .await?; + } + } + archive.finish().await?; let file = tokio::fs::File::open(file_path).await?; diff --git a/backend/windmill-common/src/schedule.rs b/backend/windmill-common/src/schedule.rs index bf2c5c267d..6b4b3b3de4 100644 --- a/backend/windmill-common/src/schedule.rs +++ b/backend/windmill-common/src/schedule.rs @@ -24,6 +24,7 @@ pub struct Schedule { pub args: Option, pub extra_perms: serde_json::Value, pub email: String, + #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, pub on_failure: Option, } diff --git a/cli/folder.ts b/cli/folder.ts index c2a898334d..39d7afa592 100644 --- a/cli/folder.ts +++ b/cli/folder.ts @@ -64,22 +64,32 @@ export async function pushFolder( return; } log.debug(`Folder ${name} is not up-to-date, updating...`); - await FolderService.updateFolder({ - workspace: workspace, - name: name, - requestBody: { - ...localFolder, - }, - }); + try { + await FolderService.updateFolder({ + workspace: workspace, + name: name, + requestBody: { + ...localFolder, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } } else { console.log(colors.bold.yellow("Creating new folder: " + name)); - await FolderService.createFolder({ - workspace: workspace, - requestBody: { - name: name, - ...localFolder, - }, - }); + try { + await FolderService.createFolder({ + workspace: workspace, + requestBody: { + name: name, + ...localFolder, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } } } diff --git a/cli/main.ts b/cli/main.ts index 6b8ab39711..b71a0581b4 100644 --- a/cli/main.ts +++ b/cli/main.ts @@ -16,6 +16,7 @@ import push from "./push.ts"; import pull from "./pull.ts"; import hub from "./hub.ts"; import folder from "./folder.ts"; +import schedule from "./schedule.ts"; import sync from "./sync.ts"; import dev from "./dev.ts"; import { tryResolveVersion } from "./context.ts"; @@ -61,6 +62,7 @@ let command: any = new Command() .command("variable", variable) .command("hub", hub) .command("folder", folder) + .command("schedule", schedule) .command("dev", dev) .command("sync", sync) .command("version", "Show version information") diff --git a/cli/pull.ts b/cli/pull.ts index 1230bdd069..ed40b487b4 100644 --- a/cli/pull.ts +++ b/cli/pull.ts @@ -8,7 +8,8 @@ export async function downloadZip( plainSecrets: boolean | undefined, skipVariables?: boolean, skipResources?: boolean, - skipSecrets?: boolean + skipSecrets?: boolean, + includeSchedules?: boolean ): Promise { const requestHeaders: HeadersInit = new Headers(); requestHeaders.set("Authorization", "Bearer " + workspace.token); @@ -22,7 +23,9 @@ export async function downloadZip( plainSecrets ?? false }&skip_variables=${skipVariables ?? false}&skip_resources=${ skipResources ?? false - }&skip_secrets=${skipSecrets ?? false}`, + }&skip_secrets=${skipSecrets ?? false}&include_schedules=${ + includeSchedules ?? false + }`, { headers: requestHeaders, method: "GET", diff --git a/cli/schedule.ts b/cli/schedule.ts new file mode 100644 index 0000000000..ba76ac70e9 --- /dev/null +++ b/cli/schedule.ts @@ -0,0 +1,135 @@ +// deno-lint-ignore-file no-explicit-any +import { + colors, + Command, + Schedule, + ScheduleService, + log, + Table, +} from "./deps.ts"; +import { requireLogin, resolveWorkspace, validatePath } from "./context.ts"; +import { + GlobalOptions, + isSuperset, + parseFromFile, + removeType, +} from "./types.ts"; + +export interface ScheduleFile { + schedule: string; + on_failure: string; + script_path: string; + args: any; + timezone: string; + is_flow: boolean; +} + +async function list(opts: GlobalOptions) { + const workspace = await resolveWorkspace(opts); + await requireLogin(opts); + + const schedules = await ScheduleService.listSchedules({ + workspace: workspace.workspaceId, + }); + + new Table() + .header(["Path", "Schedule"]) + .padding(2) + .border(true) + .body(schedules.map((x) => [x.path, x.schedule])) + .render(); +} + +export async function pushSchedule( + workspace: string, + path: string, + schedule: Schedule | ScheduleFile | undefined, + localSchedule: ScheduleFile, + raw: boolean +): Promise { + path = removeType(path, "schedule"); + + log.debug(`Processing local schedule ${path}`); + + if (raw) { + // deleting old app if it exists in raw mode + try { + schedule = await ScheduleService.getSchedule({ workspace, path }); + log.debug(`Schedule ${path} exists on remote`); + } catch { + log.debug(`Schedule ${path} does not exist on remote`); + //ignore + } + } + + if (schedule) { + if (isSuperset(localSchedule, schedule)) { + log.debug(`Schedule ${path} is up to date`); + return; + } + log.debug(`Schedule ${path} is not up-to-date, updating...`); + try { + await ScheduleService.updateSchedule({ + workspace: workspace, + path, + requestBody: { + ...localSchedule, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } + } else { + console.log(colors.bold.yellow("Creating new schedule: " + path)); + try { + await ScheduleService.createSchedule({ + workspace: workspace, + requestBody: { + path: path, + ...localSchedule, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } + } +} + +async function push(opts: GlobalOptions, filePath: string, remotePath: string) { + const workspace = await resolveWorkspace(opts); + await requireLogin(opts); + + if (!validatePath(remotePath)) { + return; + } + + const fstat = await Deno.stat(filePath); + if (!fstat.isFile) { + throw new Error("file path must refer to a file."); + } + + console.log(colors.bold.yellow("Pushing schedule...")); + + await pushSchedule( + workspace.workspaceId, + remotePath, + undefined, + parseFromFile(filePath), + false + ); + console.log(colors.bold.underline.green("Schedule pushed")); +} + +const command = new Command() + .description("schedule related commands") + .action(list as any) + .command( + "push", + "push a local schedule spec. This overrides any remote versions." + ) + .arguments(" ") + .action(push as any); + +export default command; diff --git a/cli/sync.ts b/cli/sync.ts index 850be983cc..cd7a186792 100644 --- a/cli/sync.ts +++ b/cli/sync.ts @@ -19,6 +19,7 @@ import { log, yamlStringify, yamlParse, + ScheduleService, } from "./deps.ts"; import { getTypeStrFromPath, @@ -398,6 +399,7 @@ async function pull( skipVariables?: boolean; skipResources?: boolean; skipSecrets?: boolean; + includeSchedules?: boolean; } ) { if (!opts.raw) { @@ -418,7 +420,8 @@ async function pull( opts.plainSecrets, opts.skipVariables, opts.skipResources, - opts.skipSecrets + opts.skipSecrets, + opts.includeSchedules ))!, !opts.json ); @@ -606,6 +609,7 @@ async function push( skipVariables?: boolean; skipResources?: boolean; skipSecrets?: boolean; + includeSchedules?: boolean; } ) { if (!opts.raw) { @@ -635,7 +639,8 @@ async function push( opts.plainSecrets, opts.skipVariables, opts.skipResources, - opts.skipSecrets + opts.skipSecrets, + opts.includeSchedules ))!, !opts.json ); @@ -788,6 +793,12 @@ async function push( path: removeSuffix(change.path, ".app.json"), }); break; + case "schedule": + await ScheduleService.deleteSchedule({ + workspace: workspaceId, + path: removeSuffix(change.path, ".schedule.json"), + }); + break; case "variable": await VariableService.deleteVariable({ workspace: workspaceId, @@ -828,6 +839,7 @@ const command = new Command() .option("--skip-variables", "Skip syncing variables (including secrets)") .option("--skip-secrets", "Skip syncing only secrets variables") .option("--skip-resources", "Skip syncing resources") + .option("--include-schedules", "Include syncing schedules") // deno-lint-ignore no-explicit-any .action(pull as any) .command("push") @@ -846,6 +858,7 @@ const command = new Command() .option("--skip-variables", "Skip syncing variables (including secrets)") .option("--skip-secrets", "Skip syncing only secrets variables") .option("--skip-resources", "Skip syncing resources") + .option("--include-schedules", "Include syncing schedules") // deno-lint-ignore no-explicit-any .action(push as any); diff --git a/cli/types.ts b/cli/types.ts index ae4acbf42e..344294507a 100644 --- a/cli/types.ts +++ b/cli/types.ts @@ -11,6 +11,7 @@ import * as Diff from "npm:diff"; import { yamlOptions } from "./sync.ts"; import { showDiffs } from "./main.ts"; import { deepEqual } from "./utils.ts"; +import { pushSchedule } from "./schedule.ts"; export interface DifferenceCreate { type: "CREATE"; @@ -115,6 +116,8 @@ export function pushObj( pushResource(workspace, p, befObj, newObj, checkForCreate); } else if (typeEnding === "resource-type") { pushResourceType(workspace, p, befObj, newObj, checkForCreate); + } else if (typeEnding === "schedule") { + pushSchedule(workspace, p, befObj, newObj, checkForCreate); } else { throw new Error("infer type unreachable"); } @@ -145,7 +148,8 @@ export function getTypeStrFromPath( | "resource" | "resource-type" | "folder" - | "app" { + | "app" + | "schedule" { if (p.includes(".flow/")) { return "flow"; } @@ -170,7 +174,8 @@ export function getTypeStrFromPath( typeEnding === "variable" || typeEnding === "resource" || typeEnding === "resource-type" || - typeEnding === "app" + typeEnding === "app" || + typeEnding === "schedule" ) { return typeEnding; } else {