diff --git a/backend/.sqlx/query-42cb4bc3abc8cba2193a7fbbae8275c1bc830081a43e3a4a897c41b6b3099b1d.json b/backend/.sqlx/query-42cb4bc3abc8cba2193a7fbbae8275c1bc830081a43e3a4a897c41b6b3099b1d.json deleted file mode 100644 index 37515811a3..0000000000 --- a/backend/.sqlx/query-42cb4bc3abc8cba2193a7fbbae8275c1bc830081a43e3a4a897c41b6b3099b1d.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO token\n (workspace_id, token, owner, label, expiration, super_admin, email, job)\n VALUES ($1, $2, $3, $4, now() + ($5 || ' seconds')::interval, $6, $7, $8)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Varchar", - "Text", - "Bool", - "Varchar", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "42cb4bc3abc8cba2193a7fbbae8275c1bc830081a43e3a4a897c41b6b3099b1d" -} diff --git a/backend/.sqlx/query-64a3b92ce4366518423bc536efe33415b43b325d7380e0c7062c392c7ba51e59.json b/backend/.sqlx/query-64a3b92ce4366518423bc536efe33415b43b325d7380e0c7062c392c7ba51e59.json new file mode 100644 index 0000000000..8912828eb2 --- /dev/null +++ b/backend/.sqlx/query-64a3b92ce4366518423bc536efe33415b43b325d7380e0c7062c392c7ba51e59.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_perms WHERE job_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "64a3b92ce4366518423bc536efe33415b43b325d7380e0c7062c392c7ba51e59" +} diff --git a/backend/.sqlx/query-8efd06387ded837d7849adafe5bc93acb882ef90fc58b023650c875e0fd17047.json b/backend/.sqlx/query-8efd06387ded837d7849adafe5bc93acb882ef90fc58b023650c875e0fd17047.json new file mode 100644 index 0000000000..005e8fedc9 --- /dev/null +++ b/backend/.sqlx/query-8efd06387ded837d7849adafe5bc93acb882ef90fc58b023650c875e0fd17047.json @@ -0,0 +1,21 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $2, $3, $4, $5, $6, $7, $8) \n ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Varchar", + "Varchar", + "Bool", + "Bool", + "JsonbArray", + "TextArray", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "8efd06387ded837d7849adafe5bc93acb882ef90fc58b023650c875e0fd17047" +} diff --git a/backend/.sqlx/query-a00e61e770e20157bbd9e4cdedf7fb5f9de7c8c9e50282e3ecf2e3ce917ec37a.json b/backend/.sqlx/query-a00e61e770e20157bbd9e4cdedf7fb5f9de7c8c9e50282e3ecf2e3ce917ec37a.json new file mode 100644 index 0000000000..9f95781136 --- /dev/null +++ b/backend/.sqlx/query-a00e61e770e20157bbd9e4cdedf7fb5f9de7c8c9e50282e3ecf2e3ce917ec37a.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO global_settings (name, value) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET value = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "a00e61e770e20157bbd9e4cdedf7fb5f9de7c8c9e50282e3ecf2e3ce917ec37a" +} diff --git a/backend/.sqlx/query-d2def87d7f7901eebc65082f7df5e0a33e5702b25c3db3affa06155e90480e42.json b/backend/.sqlx/query-d2def87d7f7901eebc65082f7df5e0a33e5702b25c3db3affa06155e90480e42.json new file mode 100644 index 0000000000..58f03dbec2 --- /dev/null +++ b/backend/.sqlx/query-d2def87d7f7901eebc65082f7df5e0a33e5702b25c3db3affa06155e90480e42.json @@ -0,0 +1,71 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT * FROM job_perms WHERE job_id = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "email", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "username", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "is_admin", + "type_info": "Bool" + }, + { + "ordinal": 4, + "name": "is_operator", + "type_info": "Bool" + }, + { + "ordinal": 5, + "name": "created_at", + "type_info": "Timestamp" + }, + { + "ordinal": 6, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 7, + "name": "groups", + "type_info": "TextArray" + }, + { + "ordinal": 8, + "name": "folders", + "type_info": "JsonbArray" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + false, + false, + false + ] + }, + "hash": "d2def87d7f7901eebc65082f7df5e0a33e5702b25c3db3affa06155e90480e42" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 616fbcbee9..599ac8915b 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -10185,6 +10185,7 @@ dependencies = [ "http 1.1.0", "hyper 1.3.1", "itertools 0.13.0", + "jsonwebtoken", "lazy_static", "magic-crypt", "mime_guess", diff --git a/backend/build.rs b/backend/build.rs index 7609593841..d5068697c7 100644 --- a/backend/build.rs +++ b/backend/build.rs @@ -2,4 +2,4 @@ fn main() { // trigger recompilation when a new migration is added println!("cargo:rerun-if-changed=migrations"); -} \ No newline at end of file +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 4105caad36..a10eea8256 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -50a2701b6bf16c55677066ed4aeeaf32cd5a66a8 \ No newline at end of file +9f179a95ce20062411cd73dfe5cbc4f22ef522e2 \ No newline at end of file diff --git a/backend/migrations/20240613150524_add_job_perms.down.sql b/backend/migrations/20240613150524_add_job_perms.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20240613150524_add_job_perms.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20240613150524_add_job_perms.up.sql b/backend/migrations/20240613150524_add_job_perms.up.sql new file mode 100644 index 0000000000..decb49a010 --- /dev/null +++ b/backend/migrations/20240613150524_add_job_perms.up.sql @@ -0,0 +1,13 @@ +-- Add up migration script here +CREATE TABLE public.job_perms ( + job_id uuid NOT NULL, + email varchar(255) NOT NULL, + username varchar(50) NOT NULL, + is_admin bool NOT NULL, + is_operator bool NOT NULL, + created_at timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL, + workspace_id varchar(50) NOT NULL, + groups _text NOT NULL, + folders _jsonb NOT NULL, + CONSTRAINT job_perms_pk PRIMARY KEY (job_id) +); \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 708c612ed8..1a2ed655a3 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -25,10 +25,11 @@ use windmill_common::{ BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, - HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, - LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, - REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, + HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, + KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, + PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, + SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, stats_ee::schedule_stats, utils::{rd_string, Mode}, @@ -63,9 +64,10 @@ use crate::monitor::{ load_tag_per_workspace_enabled, monitor_db, monitor_pool, reload_base_url_setting, reload_bunfig_install_scopes_setting, reload_critical_error_channels_setting, reload_extra_pip_index_url_setting, reload_hub_base_url_setting, - reload_job_default_timeout_setting, reload_license_key, reload_npm_config_registry_setting, - reload_pip_index_url_setting, reload_retention_period_setting, reload_scim_token_setting, - reload_server_config, reload_worker_config, + reload_job_default_timeout_setting, reload_jwt_secret_setting, reload_license_key, + reload_npm_config_registry_setting, reload_pip_index_url_setting, + reload_retention_period_setting, reload_scim_token_setting, reload_server_config, + reload_worker_config, }; #[cfg(feature = "parquet")] @@ -163,7 +165,7 @@ async fn windmill_main() -> anyhow::Result<()> { panic!("BASE_INTERNAL_URL is required in agent mode") } if std::env::var("JOB_TOKEN").is_err() { - tracing::warn!("JOB_TOKEN is not passed, hence workers will still create one ephemeral token per job and the DATABASE_URL need to be of a role that can INSERT into the token table") + tracing::warn!("JOB_TOKEN is not passed, hence workers will still need to create permissions for each job and the DATABASE_URL needs to be of a role that can INSERT into the job_perms table") } #[cfg(not(feature = "enterprise"))] @@ -536,6 +538,11 @@ Windmill Community Edition {GIT_VERSION} tracing::error!(error = %e, "Could not reload critical error emails setting"); } }, + JWT_SECRET_SETTING => { + if let Err(e) = reload_jwt_secret_setting(&db).await { + tracing::error!(error = %e, "Could not reload jwt secret setting"); + } + }, a @_ => { tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a); } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 0a7dd14388..542d40f1c5 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -22,18 +22,31 @@ use windmill_api::{ DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN, }; use windmill_common::{ - ee::CriticalErrorChannel, error, flow_status::FlowStatusModule, global_settings::{ + auth::JWT_SECRET, + ee::CriticalErrorChannel, + error, + flow_status::FlowStatusModule, + global_settings::{ BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, - KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, - PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, + OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, - }, jobs::QueuedJob, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, server::load_server_config, stats_ee::get_user_usage, users::truncate_token, utils::now_from_db, worker::{ + }, + jobs::QueuedJob, + oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, + server::load_server_config, + stats_ee::get_user_usage, + users::truncate_token, + utils::{now_from_db, rd_string}, + worker::{ load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE, SERVER_CONFIG, WORKER_CONFIG, - }, BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, METRICS_DEBUG_ENABLED, METRICS_ENABLED + }, + BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, + METRICS_DEBUG_ENABLED, METRICS_ENABLED, }; use windmill_queue::cancel_job; use windmill_worker::{ @@ -140,6 +153,10 @@ pub async fn initial_load( tracing::error!("Could not reload critical error emails setting: {:?}", e); } + if let Err(e) = reload_jwt_secret_setting(&db).await { + tracing::error!("Could not reload jwt secret setting: {:?}", e); + } + #[cfg(feature = "parquet")] if !_is_agent { reload_s3_cache_setting(&db).await; @@ -200,7 +217,7 @@ pub async fn load_metrics_debug_enabled(db: &DB) -> error::Result<()> { tracing::error!("Error setting jemalloc prof_active: {e:?}"); } } - }, + } _ => (), }; Ok(()) @@ -208,7 +225,9 @@ pub async fn load_metrics_debug_enabled(db: &DB) -> error::Result<()> { #[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))] #[derive(Debug, Clone)] -pub struct MallctlError { pub code: i32 } +pub struct MallctlError { + pub code: i32, +} #[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))] fn set_prof_active(new_value: bool) -> Result<(), MallctlError> { @@ -216,13 +235,12 @@ fn set_prof_active(new_value: bool) -> Result<(), MallctlError> { tracing::info!("Setting jemalloc prof_active to {}", new_value); let result = unsafe { - tikv_jemalloc_sys::mallctl( - option_name.as_ptr(), // const char *name - std::ptr::null_mut(), // void *oldp - std::ptr::null_mut(), // size_t *oldlenp - &new_value as *const _ as *mut _, // void *newp - std::mem::size_of_val(&new_value) // size_t newlen + option_name.as_ptr(), // const char *name + std::ptr::null_mut(), // void *oldp + std::ptr::null_mut(), // size_t *oldlenp + &new_value as *const _ as *mut _, // void *newp + std::mem::size_of_val(&new_value), // size_t newlen ) }; @@ -239,15 +257,12 @@ pub fn bytes_to_mb(bytes: u64) -> f64 { bytes as f64 / BYTES_PER_MB } - #[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))] pub async fn monitor_mem() { - use std::time::Duration; - use tikv_jemalloc_ctl::{stats, epoch}; + use tikv_jemalloc_ctl::{epoch, stats}; tokio::spawn(async move { - // Obtain a MIB for the `epoch`, `stats.allocated`, and // `atats.resident` keys: let e = match epoch::mib() { @@ -270,19 +285,22 @@ pub async fn monitor_mem() { tracing::error!("Error getting jemalloc resident mib: {:?}", e); return; } - }; - - + }; + loop { - // Many statistics are cached and only updated + // Many statistics are cached and only updated // when the epoch is advanced: match e.advance() { Ok(_) => { // Read statistics using MIB key: let allocated = allocated.read().unwrap_or_default(); let resident = resident.read().unwrap_or_default(); - tracing::info!("{} mb allocated/{} mb resident", bytes_to_mb(allocated as u64), bytes_to_mb(resident as u64)); - }, + tracing::info!( + "{} mb allocated/{} mb resident", + bytes_to_mb(allocated as u64), + bytes_to_mb(resident as u64) + ); + } Err(e) => { tracing::error!("Error advancing jemalloc epoch: {:?}", e); } @@ -799,7 +817,6 @@ pub async fn expose_queue_metrics(db: &Pool) { .unwrap_or(true); if metrics_enabled || save_metrics { - let queue_counts = sqlx::query!( "SELECT tag, count(*) as count FROM queue WHERE scheduled_for <= now() - ('3 seconds')::interval AND running = false @@ -845,20 +862,21 @@ pub async fn expose_queue_metrics(db: &Pool) { // clean queue metrics older than 14 days sqlx::query!( "DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'" - ).execute(&mut *tx).await.ok(); + ) + .execute(&mut *tx) + .await + .ok(); tx.commit().await.ok(); } } - #[derive(Serialize)] struct WorkerUsage { worker: String, worker_instance: String, vcpus: Option, memory: Option, - } pub async fn save_usage_metrics(db: &Pool) { @@ -876,7 +894,10 @@ pub async fn save_usage_metrics(db: &Pool) { // save author and operator count every ~24 hours if last_check - .map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::hours(24) - chrono::Duration::minutes(random_nb % 60)) + .map(|last_check| { + chrono::Utc::now() - last_check + > chrono::Duration::hours(24) - chrono::Duration::minutes(random_nb % 60) + }) .unwrap_or(true) { let user_usage = get_user_usage(&mut *tx).await.ok(); @@ -884,7 +905,7 @@ pub async fn save_usage_metrics(db: &Pool) { if let Some(user_usage) = user_usage { sqlx::query!( "INSERT INTO metrics (id, value) VALUES ('author_count', $1), ('operator_count', $2)", - serde_json::json!(user_usage.author_count.unwrap_or(0)), + serde_json::json!(user_usage.author_count.unwrap_or(0)), serde_json::json!(user_usage.operator_count.unwrap_or(0)) ) .execute(&mut *tx) @@ -892,7 +913,6 @@ pub async fn save_usage_metrics(db: &Pool) { .ok(); } - // clean metrics older than 6 months (including worker usage) sqlx::query!( "DELETE FROM metrics @@ -905,7 +925,10 @@ pub async fn save_usage_metrics(db: &Pool) { // save worker usage every ~60 minutes if last_check - .map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::minutes(60) - chrono::Duration::seconds(random_nb % 300)) + .map(|last_check| { + chrono::Utc::now() - last_check + > chrono::Duration::minutes(60) - chrono::Duration::seconds(random_nb % 300) + }) .unwrap_or(true) { let worker_usage = sqlx::query_as!( @@ -929,7 +952,6 @@ pub async fn save_usage_metrics(db: &Pool) { tx.commit().await.ok(); } - } pub async fn reload_server_config(db: &Pool) { @@ -1103,7 +1125,9 @@ async fn handle_zombie_jobs mpsc::channel::(1); let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::(1); - let label = if job.permissioned_as != format!("u/{}", job.created_by) && job.permissioned_as != job.created_by { + let label = if job.permissioned_as != format!("u/{}", job.created_by) + && job.permissioned_as != job.created_by + { format!("ephemeral-script-end-user-{}", job.created_by) } else { "ephemeral-script".to_string() @@ -1256,7 +1280,7 @@ async fn cancel_zombie_flow_job( db, rsmq.clone(), false, - false + false, ) .await?; sqlx::query!( @@ -1334,3 +1358,35 @@ pub async fn reload_critical_error_channels_setting(db: &DB) -> error::Result<() Ok(()) } + +async fn generate_and_save_jwt_secret(db: &DB) -> error::Result { + let secret = rd_string(32); + sqlx::query!( + "INSERT INTO global_settings (name, value) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET value = $2", + JWT_SECRET_SETTING, + serde_json::to_value(&secret).unwrap() + ).execute(db).await?; + + Ok(secret) +} + +pub async fn reload_jwt_secret_setting(db: &DB) -> error::Result<()> { + let jwt_secret = load_value_from_global_settings(db, JWT_SECRET_SETTING).await?; + + let jwt_secret = if let Some(q) = jwt_secret { + if let Ok(v) = serde_json::from_value::(q.clone()) { + v + } else { + tracing::error!("Could not parse jwt_secret setting, generating new one"); + generate_and_save_jwt_secret(db).await? + } + } else { + tracing::info!("Not jwt secret found, generating one"); + generate_and_save_jwt_secret(db).await? + }; + + let mut l = JWT_SECRET.write().await; + *l = jwt_secret; + + Ok(()) +} diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 43c302bfde..52cad583ad 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1,5 +1,5 @@ -use windmill_api_client::types::{NewScript, NewScriptLanguage}; use std::str::FromStr; +use windmill_api_client::types::{NewScript, NewScriptLanguage}; #[cfg(feature = "enterprise")] use chrono::Timelike; @@ -13,26 +13,23 @@ use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres}; #[cfg(feature = "enterprise")] use tokio::time::{timeout, Duration}; -use windmill_api_client::types::{ - CreateFlowBody, RawScript -}; +use windmill_api_client::types::{CreateFlowBody, RawScript}; use sqlx::query; #[cfg(feature = "enterprise")] use windmill_api_client::types::{EditSchedule, NewSchedule, ScriptArgs}; - -use windmill_common::worker::{WORKER_CONFIG, PriorityTags}; +use serde::Serialize; +use windmill_common::auth::JWT_SECRET; +use windmill_common::worker::{PriorityTags, WORKER_CONFIG}; use windmill_common::{ flow_status::{FlowStatus, FlowStatusModule, RestartedFrom}, flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform}, - jobs::{JobPayload, RawCode, JobKind}, - scripts::{ScriptLang, ScriptHash} + jobs::{JobKind, JobPayload, RawCode}, + scripts::{ScriptHash, ScriptLang}, }; use windmill_queue::PushIsolationLevel; -use serde::Serialize; - #[derive(Debug, sqlx::FromRow, Serialize)] pub struct CompletedJob { @@ -66,7 +63,6 @@ pub struct CompletedJob { pub script_hash: Option, pub language: Option, pub job_kind: JobKind, - } impl CompletedJob { @@ -168,6 +164,12 @@ fn find_module_in_vec(modules: Vec, id: &str) -> Option () { + let secret = "mytestsecret".to_string(); + let mut l = JWT_SECRET.write().await; + *l = secret; +} + mod suspend_resume { use serde_json::json; @@ -272,11 +274,12 @@ mod suspend_resume { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) - .arg("n", json!(1)) - .arg("port", json!(port)) - .push(&db) - .await; + let flow = + RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) + .arg("n", json!(1)) + .arg("port", json!(port)) + .push(&db) + .await; let mut completed = listen_for_completed_jobs(&db).await; let queue = listen_for_queue(&db).await; @@ -353,14 +356,15 @@ mod suspend_resume { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) - .arg("n", json!(1)) - .arg("op", json!("cancel")) - .arg("port", json!(port)) - .run_until_complete(&db, port) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) + .arg("n", json!(1)) + .arg("op", json!("cancel")) + .arg("port", json!(port)) + .run_until_complete(&db, port) + .await + .json_result() + .unwrap(); server.close().await.unwrap(); @@ -377,11 +381,12 @@ mod suspend_resume { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) - .arg("n", json!(1)) - .arg("port", json!(port)) - .push(&db) - .await; + let flow = + RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) + .arg("n", json!(1)) + .arg("port", json!(port)) + .push(&db) + .await; let mut completed = listen_for_completed_jobs(&db).await; let queue = listen_for_queue(&db).await; @@ -571,13 +576,17 @@ def main(last, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None }) - .arg("items", json!(["unused", "unused", "unused"])) - .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) - .await - .json_result() - .unwrap(); + let result = RunJob::from(JobPayload::RawFlow { + value: flow_forloop_retry(), + path: None, + restarted_from: None, + }) + .arg("items", json!(["unused", "unused", "unused"])) + .arg("port", json!(server.addr.port())) + .run_until_complete(&db, server.addr.port()) + .await + .json_result() + .unwrap(); assert_eq!(server.close().await, attempts); assert_eq!(json!([3, 5, 7, 9]), result); @@ -600,13 +609,17 @@ def main(last, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None }) - .arg("items", json!(["unused", "unused", "unused"])) - .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) - .await - .json_result() - .unwrap(); + let result = RunJob::from(JobPayload::RawFlow { + value: flow_forloop_retry(), + path: None, + restarted_from: None, + }) + .arg("items", json!(["unused", "unused", "unused"])) + .arg("port", json!(server.addr.port())) + .run_until_complete(&db, server.addr.port()) + .await + .json_result() + .unwrap(); assert_eq!(server.close().await, attempts); @@ -641,11 +654,15 @@ def main(last, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let job = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None }) - .arg("items", json!(["unused", "unused", "unused"])) - .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) - .await; + let job = RunJob::from(JobPayload::RawFlow { + value: flow_forloop_retry(), + path: None, + restarted_from: None, + }) + .arg("items", json!(["unused", "unused", "unused"])) + .arg("port", json!(server.addr.port())) + .run_until_complete(&db, server.addr.port()) + .await; let result = job.json_result().unwrap(); assert_eq!(server.close().await, attempts); @@ -765,21 +782,23 @@ async fn test_iteration(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("items", json!([])) - .run_until_complete(&db, server.addr.port()) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("items", json!([])) + .run_until_complete(&db, server.addr.port()) + .await + .json_result() + .unwrap(); assert_eq!(result, serde_json::json!([])); /* Don't actually test that this does 257 jobs or that will take forever. */ - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("items", json!((0..257).collect::>())) - .run_until_complete(&db, server.addr.port()) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("items", json!((0..257).collect::>())) + .run_until_complete(&db, server.addr.port()) + .await + .json_result() + .unwrap(); assert!(matches!(result, serde_json::Value::Array(_))); assert!(result[2]["error"] .as_object() @@ -822,19 +841,21 @@ async fn test_iteration_parallel(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("items", json!([])) - .run_until_complete(&db, server.addr.port()) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("items", json!([])) + .run_until_complete(&db, server.addr.port()) + .await + .json_result() + .unwrap(); assert_eq!(result, serde_json::json!([])); /* Don't actually test that this does 257 jobs or that will take forever. */ - let job = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("items", json!((0..50).collect::>())) - .run_until_complete(&db, server.addr.port()) - .await; + let job = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("items", json!((0..50).collect::>())) + .run_until_complete(&db, server.addr.port()) + .await; // println!("{:#?}", job); let result = job.json_result().unwrap(); assert!(matches!(result, serde_json::Value::Array(_))); @@ -870,7 +891,8 @@ impl RunJob { let mut hm_args = std::collections::HashMap::new(); for (k, v) in args { hm_args.insert(k, windmill_common::worker::to_raw_value(&v)); - } + } + let tx = PushIsolationLevel::IsolatedRoot(db.clone(), None); let (uuid, tx) = windmill_queue::push::( &db, @@ -894,6 +916,7 @@ impl RunJob { None, None, None, + None, ) .await .expect("push has to succeed"); @@ -928,6 +951,7 @@ async fn in_test_worker( inner: Fut, port: u16, ) -> ::Output { + set_jwt_secret().await; let (quit, worker) = spawn_test_worker(db, port); let worker = tokio::time::timeout(std::time::Duration::from_secs(60), worker); tokio::pin!(worker); @@ -959,7 +983,10 @@ fn spawn_test_worker( tokio::sync::broadcast::Sender<()>, tokio::task::JoinHandle<()>, ) { - for x in [windmill_worker::LOCK_CACHE_DIR, windmill_worker::GO_BIN_CACHE_DIR] { + for x in [ + windmill_worker::LOCK_CACHE_DIR, + windmill_worker::GO_BIN_CACHE_DIR, + ] { std::fs::DirBuilder::new() .recursive(true) .create(x) @@ -976,9 +1003,10 @@ fn spawn_test_worker( let future = async move { let base_internal_url = format!("http://localhost:{}", port); { - let mut wc = WORKER_CONFIG.write().await; - (*wc).worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); - (*wc).priority_tags_sorted = vec![PriorityTags { priority: 0, tags: (*wc).worker_tags.clone()} ] + let mut wc = WORKER_CONFIG.write().await; + (*wc).worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); + (*wc).priority_tags_sorted = + vec![PriorityTags { priority: 0, tags: (*wc).worker_tags.clone() }] } windmill_worker::run_worker::( &db, @@ -1028,11 +1056,13 @@ async fn listen_for_uuid_on( } async fn completed_job(uuid: Uuid, db: &Pool) -> CompletedJob { - - sqlx::query_as::<_, CompletedJob>("SELECT *, result->'wm_labels' as labels FROM completed_job WHERE id = $1").bind(uuid) - .fetch_one(db) - .await - .unwrap() + sqlx::query_as::<_, CompletedJob>( + "SELECT *, result->'wm_labels' as labels FROM completed_job WHERE id = $1", + ) + .bind(uuid) + .fetch_one(db) + .await + .unwrap() } #[axum::async_trait(?Send)] @@ -1073,7 +1103,8 @@ async fn test_deno_flow(db: Pool) { custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, - }.into(), + } + .into(), stop_after_if: Default::default(), summary: Default::default(), suspend: Default::default(), @@ -1111,7 +1142,8 @@ async fn test_deno_flow(db: Pool) { custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, - }.into(), + } + .into(), stop_after_if: Default::default(), summary: Default::default(), suspend: Default::default(), @@ -1124,7 +1156,8 @@ async fn test_deno_flow(db: Pool) { delete_after_use: None, continue_on_error: None, }], - }.into(), + } + .into(), stop_after_if: Default::default(), summary: Default::default(), suspend: Default::default(), @@ -1183,11 +1216,12 @@ async fn test_identity(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .run_until_complete(&db, server.addr.port()) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .run_until_complete(&db, server.addr.port()) + .await + .json_result() + .unwrap(); assert_eq!(result, serde_json::json!(42)); } @@ -1681,7 +1715,7 @@ func main(derp string) (string, error) { concurrent_limit: None, concurrency_time_window_s: None, cache_ttl: None, - dedicated_worker: None + dedicated_worker: None, })) .arg("derp", json!("world")) .run_until_complete(&db, port) @@ -1714,7 +1748,7 @@ echo "hello $msg" concurrent_limit: None, concurrency_time_window_s: None, cache_ttl: None, - dedicated_worker: None + dedicated_worker: None, })) .arg("msg", json!("world")) .run_until_complete(&db, port) @@ -1744,7 +1778,7 @@ def main(): concurrent_limit: None, concurrency_time_window_s: None, cache_ttl: None, - dedicated_worker: None + dedicated_worker: None, }); let result = run_job_in_new_worker_until_complete(&db, job, port) @@ -1780,7 +1814,7 @@ def main(): concurrent_limit: None, concurrency_time_window_s: None, cache_ttl: None, - dedicated_worker: None + dedicated_worker: None, }); let result = run_job_in_new_worker_until_complete(&db, job, port) @@ -1815,7 +1849,7 @@ def main(): concurrent_limit: None, concurrency_time_window_s: None, cache_ttl: None, - dedicated_worker: None + dedicated_worker: None, }); let result = run_job_in_new_worker_until_complete(&db, job, port) @@ -2457,12 +2491,13 @@ async fn test_failure_module(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("n", json!(0)) - .run_until_complete(&db, port) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("n", json!(0)) + .run_until_complete(&db, port) + .await + .json_result() + .unwrap(); assert!(result["from failure module"]["error"] .as_object() @@ -2473,12 +2508,13 @@ async fn test_failure_module(db: Pool) { .unwrap() .contains("[]")); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("n", json!(1)) - .run_until_complete(&db, port) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("n", json!(1)) + .run_until_complete(&db, port) + .await + .json_result() + .unwrap(); assert!(result["from failure module"]["error"] .as_object() @@ -2489,12 +2525,13 @@ async fn test_failure_module(db: Pool) { .unwrap() .contains("[0]")); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("n", json!(2)) - .run_until_complete(&db, port) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("n", json!(2)) + .run_until_complete(&db, port) + .await + .json_result() + .unwrap(); assert!(result["from failure module"]["error"] .as_object() @@ -2505,12 +2542,13 @@ async fn test_failure_module(db: Pool) { .unwrap() .contains("[0,1]")); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .arg("n", json!(3)) - .run_until_complete(&db, port) - .await - .json_result() - .unwrap(); + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .arg("n", json!(3)) + .run_until_complete(&db, port) + .await + .json_result() + .unwrap(); assert_eq!(json!({ "l": [0, 1, 2] }), result); } @@ -2819,11 +2857,10 @@ async fn test_complex_flow_restart(db: Pool) { })) .unwrap(); - let first_run_result = RunJob::from(JobPayload::RawFlow { - value: flow.clone(), - path: None, - restarted_from: None - }).run_until_complete(&db, port).await; + let first_run_result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .run_until_complete(&db, port) + .await; let restarted_flow_result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), @@ -2833,10 +2870,16 @@ async fn test_complex_flow_restart(db: Pool) { step_id: "h".to_owned(), branch_or_iteration_n: None, }), - }).run_until_complete(&db, port).await; + }) + .run_until_complete(&db, port) + .await; - let first_run_result_int = serde_json::from_value::(first_run_result.json_result().unwrap()).expect("first_run_result was not an int"); - let restarted_flow_result_int = serde_json::from_value::(restarted_flow_result.json_result().unwrap()).expect("restarted_flow_result was not an int"); + let first_run_result_int = + serde_json::from_value::(first_run_result.json_result().unwrap()) + .expect("first_run_result was not an int"); + let restarted_flow_result_int = + serde_json::from_value::(restarted_flow_result.json_result().unwrap()) + .expect("restarted_flow_result was not an int"); assert_eq!(first_run_result_int, restarted_flow_result_int); } @@ -2855,7 +2898,6 @@ async fn test_rust_client(db: Pool) { .unwrap(); } - #[cfg(feature = "enterprise")] #[sqlx::test(fixtures("base", "schedule"))] async fn test_script_schedule_handlers(db: Pool) { @@ -2980,7 +3022,7 @@ async fn test_script_schedule_handlers(db: Pool) { } let uuid = uuid.unwrap().unwrap(); - + let completed_job = query!("SELECT script_path FROM completed_job WHERE id = $1", uuid) .fetch_one(&db2) @@ -2999,7 +3041,6 @@ async fn test_script_schedule_handlers(db: Pool) { .await; } - #[cfg(feature = "enterprise")] #[sqlx::test(fixtures("base", "schedule"))] async fn test_flow_schedule_handlers(db: Pool) { @@ -3126,7 +3167,7 @@ async fn test_flow_schedule_handlers(db: Pool) { } let uuid = uuid.unwrap().unwrap(); - + let completed_job = query!("SELECT script_path FROM completed_job WHERE id = $1", uuid) .fetch_one(&db2) @@ -3145,8 +3186,11 @@ async fn test_flow_schedule_handlers(db: Pool) { .await; } - -async fn run_deployed_relative_imports(db: &Pool, script_content: String, language: ScriptLang) { +async fn run_deployed_relative_imports( + db: &Pool, + script_content: String, + language: ScriptLang, +) { initialize_tracing().await; let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); @@ -3155,109 +3199,138 @@ async fn run_deployed_relative_imports(db: &Pool, script_content: Stri "SECRET_TOKEN".to_string(), ); - client.create_script( - "test-workspace", - &NewScript { - language: NewScriptLanguage::from_str(language.as_str()).unwrap(), - content: script_content, - path: "f/system/test_import".to_string(), - concurrent_limit: None, - concurrency_time_window_s: None, - cache_ttl: None, - dedicated_worker: None, - description: "".to_string(), - draft_only: None, - envs: vec![], - is_template: None, - kind: None, - parent_hash: None, - lock: None, - summary: "".to_string(), - tag: None, - schema: std::collections::HashMap::new(), - ws_error_handler_muted: Some(false), - priority: None, - delete_after_use: None, - timeout: None, - restart_unless_cancelled: None, - deployment_message: None, - concurrency_key: None, - visible_to_runner_only: None, - no_main_func: None, - codebase: None - }, - ).await.unwrap(); + client + .create_script( + "test-workspace", + &NewScript { + language: NewScriptLanguage::from_str(language.as_str()).unwrap(), + content: script_content, + path: "f/system/test_import".to_string(), + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + description: "".to_string(), + draft_only: None, + envs: vec![], + is_template: None, + kind: None, + parent_hash: None, + lock: None, + summary: "".to_string(), + tag: None, + schema: std::collections::HashMap::new(), + ws_error_handler_muted: Some(false), + priority: None, + delete_after_use: None, + timeout: None, + restart_unless_cancelled: None, + deployment_message: None, + concurrency_key: None, + visible_to_runner_only: None, + no_main_func: None, + codebase: None, + }, + ) + .await + .unwrap(); let mut completed = listen_for_completed_jobs(&db).await; let db2 = db.clone(); - in_test_worker(&db, async move { - completed.next().await; // deployed script + in_test_worker( + &db, + async move { + completed.next().await; // deployed script - let script = - query!("SELECT hash FROM script WHERE path = $1", "f/system/test_import".to_string()) - .fetch_one(&db2) - .await - .unwrap(); + let script = query!( + "SELECT hash FROM script WHERE path = $1", + "f/system/test_import".to_string() + ) + .fetch_one(&db2) + .await + .unwrap(); - let job = RunJob::from(JobPayload::ScriptHash { - path: "f/system/test_import".to_string(), - hash: ScriptHash(script.hash), - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, - cache_ttl: None, - dedicated_worker: None, - language, - priority: None, - }).push(&db2).await; + let job = RunJob::from(JobPayload::ScriptHash { + path: "f/system/test_import".to_string(), + hash: ScriptHash(script.hash), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language, + priority: None, + }) + .push(&db2) + .await; + completed.next().await; // completed job - completed.next().await; // completed job + let result = completed_job(job, &db2).await.json_result().unwrap(); - - let result = completed_job(job, &db2).await.json_result().unwrap(); - - - assert_eq!(result, serde_json::json!(["f/system/same_folder_script", "f/system/same_folder_script", "f/system_relative/different_folder_script", "f/system_relative/different_folder_script"])); - - }, port).await; + assert_eq!( + result, + serde_json::json!([ + "f/system/same_folder_script", + "f/system/same_folder_script", + "f/system_relative/different_folder_script", + "f/system_relative/different_folder_script" + ]) + ); + }, + port, + ) + .await; } - -async fn run_preview_relative_imports(db: &Pool, script_content: String, language: ScriptLang) { +async fn run_preview_relative_imports( + db: &Pool, + script_content: String, + language: ScriptLang, +) { initialize_tracing().await; let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); let mut completed = listen_for_completed_jobs(&db).await; let db2 = db.clone(); - in_test_worker(&db, async move { - let job = RunJob::from(JobPayload::Code(RawCode { - hash: None, - content: script_content, - path: Some("f/system/test_import".to_string()), - language, - lock: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, - cache_ttl: None, - dedicated_worker: None - })).push(&db2).await; + in_test_worker( + &db, + async move { + let job = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content: script_content, + path: Some("f/system/test_import".to_string()), + language, + lock: None, + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + })) + .push(&db2) + .await; + completed.next().await; // completed job - completed.next().await; // completed job + let result = completed_job(job, &db2).await.json_result().unwrap(); - - let result = completed_job(job, &db2).await.json_result().unwrap(); - - - assert_eq!(result, serde_json::json!(["f/system/same_folder_script", "f/system/same_folder_script", "f/system_relative/different_folder_script", "f/system_relative/different_folder_script"])); - - }, port).await; + assert_eq!( + result, + serde_json::json!([ + "f/system/same_folder_script", + "f/system/same_folder_script", + "f/system_relative/different_folder_script", + "f/system_relative/different_folder_script" + ]) + ); + }, + port, + ) + .await; } - #[sqlx::test(fixtures("base", "relative_bun"))] async fn test_relative_imports_bun(db: Pool) { let content = r#" @@ -3269,29 +3342,28 @@ import { main as test4 } from "../system_relative/different_folder_script.ts"; export async function main() { return [test1(), test2(), test3(), test4()]; } -"#.to_string(); +"# + .to_string(); run_deployed_relative_imports(&db, content.clone(), ScriptLang::Bun).await; run_preview_relative_imports(&db, content, ScriptLang::Bun).await; } - #[sqlx::test(fixtures("base", "relative_bun"))] async fn test_nested_imports_bun(db: Pool) { - let content = r#" import { main as test } from "/f/system_relative/nested_script.ts"; export async function main() { return test(); } -"#.to_string(); +"# + .to_string(); run_deployed_relative_imports(&db, content.clone(), ScriptLang::Bun).await; run_preview_relative_imports(&db, content, ScriptLang::Bun).await; } - #[sqlx::test(fixtures("base", "relative_deno"))] async fn test_relative_imports_deno(db: Pool) { let content = r#" @@ -3303,30 +3375,28 @@ import { main as test4 } from "../system_relative/different_folder_script.ts"; export async function main() { return [test1(), test2(), test3(), test4()]; } -"#.to_string(); +"# + .to_string(); run_deployed_relative_imports(&db, content.clone(), ScriptLang::Deno).await; run_preview_relative_imports(&db, content, ScriptLang::Deno).await; - } - #[sqlx::test(fixtures("base", "relative_deno"))] async fn test_nested_imports_deno(db: Pool) { - let content = r#" import { main as test } from "/f/system_relative/nested_script.ts"; export async function main() { return test(); } -"#.to_string(); +"# + .to_string(); run_deployed_relative_imports(&db, content.clone(), ScriptLang::Deno).await; run_preview_relative_imports(&db, content, ScriptLang::Deno).await; } - #[sqlx::test(fixtures("base", "relative_python"))] async fn test_relative_imports_python(db: Pool) { let content = r#" @@ -3337,24 +3407,23 @@ from ..system_relative.different_folder_script import main as test4 def main(): return [test1(), test2(), test3(), test4()] -"#.to_string(); +"# + .to_string(); run_deployed_relative_imports(&db, content.clone(), ScriptLang::Python3).await; run_preview_relative_imports(&db, content, ScriptLang::Python3).await; - } - #[sqlx::test(fixtures("base", "relative_python"))] async fn test_nested_imports_python(db: Pool) { - let content = r#" from f.system_relative.nested_script import main as test def main(): return test() -"#.to_string(); +"# + .to_string(); run_deployed_relative_imports(&db, content.clone(), ScriptLang::Python3).await; run_preview_relative_imports(&db, content, ScriptLang::Python3).await; diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index d0d4e6ca32..ed2551dcda 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -86,6 +86,7 @@ datafusion = { workspace = true, optional = true} object_store = { workspace = true, optional = true} openidconnect = { workspace = true, optional = true} url = { workspace = true, optional = true} +jsonwebtoken = { workspace = true } pin-project.workspace = true crc.workspace = true diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index f23ecb35d5..8601483fe6 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -648,6 +648,7 @@ async fn create_app( None, None, None, + Some(&authed.clone().into()), ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -938,6 +939,7 @@ async fn update_app( None, None, None, + Some(&authed.clone().into()), ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1150,6 +1152,7 @@ async fn execute_component( None, None, None, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/ee.rs b/backend/windmill-api/src/ee.rs index a9d170a3ac..eb4ff9d11d 100644 --- a/backend/windmill-api/src/ee.rs +++ b/backend/windmill-api/src/ee.rs @@ -1,6 +1,15 @@ +#[cfg(feature = "enterprise")] +use crate::db::ApiAuthed; use anyhow::anyhow; pub async fn validate_license_key(_license_key: String) -> anyhow::Result { // Implementation is not open source Err(anyhow!("License can't be validated in Windmill CE")) } + +#[cfg(feature = "enterprise")] +pub async fn jwt_ext_auth(_w_id: Option<&String>, _token: &str) -> Option<(ApiAuthed, usize)> { + // Implementation is not open source + + None +} diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 522176a6ee..1d666a786a 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -393,6 +393,7 @@ async fn create_flow( None, None, None, + Some(&authed.clone().into()), ) .await?; @@ -539,7 +540,7 @@ async fn update_flow( clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, &schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule, None).await?; } } @@ -609,6 +610,7 @@ async fn update_flow( None, None, None, + Some(&authed.clone().into()), ) .await?; sqlx::query!( diff --git a/backend/windmill-api/src/folders.rs b/backend/windmill-api/src/folders.rs index fc671b0c88..6b9f905ca4 100644 --- a/backend/windmill-api/src/folders.rs +++ b/backend/windmill-api/src/folders.rs @@ -603,32 +603,6 @@ async fn add_owner( Ok(format!("Added {} to folder {}", owner, name)) } -pub async fn get_folders_for_user( - w_id: &str, - username: &str, - groups: &[String], - db: &DB, -) -> Result> { - let mut perms = groups - .into_iter() - .map(|x| format!("g/{}", x)) - .collect::>(); - perms.insert(0, format!("u/{}", username)); - let folders = sqlx::query!( - "SELECT name, (EXISTS (SELECT 1 FROM (SELECT key, value FROM jsonb_each_text(extra_perms) WHERE key = ANY($1)) t WHERE value::boolean IS true)) as write, $1 && owners::text[] as owner FROM folder - WHERE extra_perms ?| $1 AND workspace_id = $2", - &perms[..], - w_id, - ) - .fetch_all(db) - .await? - .into_iter() - .map(|x| (x.name, x.write.unwrap_or(false), x.owner.unwrap_or(false))) - .collect(); - - Ok(folders) -} - async fn remove_owner( authed: ApiAuthed, Extension(user_db): Extension, diff --git a/backend/windmill-api/src/groups.rs b/backend/windmill-api/src/groups.rs index 3d36cb34c5..dc545ab2e8 100644 --- a/backend/windmill-api/src/groups.rs +++ b/backend/windmill-api/src/groups.rs @@ -7,7 +7,7 @@ */ use crate::db::ApiAuthed; -use crate::{db::DB, users::get_groups_for_user, utils::require_super_admin}; +use crate::{db::DB, utils::require_super_admin}; use axum::{ extract::{Extension, Path, Query}, @@ -17,11 +17,12 @@ use axum::{ use windmill_audit::audit_ee::audit_log; use windmill_audit::ActionKind; use windmill_common::worker::CLOUD_HOSTED; -use windmill_common::{db::UserDB, users::username_to_permissioned_as}; use windmill_common::{ + auth::get_groups_for_user, error::{Error, JsonResult, Result}, utils::{not_found_if_none, paginate, Pagination}, }; +use windmill_common::{db::UserDB, users::username_to_permissioned_as}; use serde::{Deserialize, Serialize}; use sqlx::{query_scalar, FromRow, Postgres, Transaction}; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index a0c658bc2c..daba418caf 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2622,6 +2622,7 @@ pub async fn run_flow_by_path( None, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -2711,6 +2712,7 @@ pub async fn restart_flow( None, None, completed_job.priority, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -2763,6 +2765,7 @@ pub async fn run_script_by_path( timeout, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -2839,6 +2842,7 @@ pub async fn run_workflow_as_code( timeout, None, None, + Some(&authed.clone().into()), ) .await?; sqlx::query!( @@ -3137,6 +3141,7 @@ pub async fn run_wait_result_job_by_path_get( timeout, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3256,6 +3261,7 @@ async fn run_wait_result_script_by_path_internal( timeout, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3336,6 +3342,7 @@ pub async fn run_wait_result_script_by_hash( timeout, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3418,6 +3425,7 @@ async fn run_wait_result_flow_by_path_internal( None, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3485,6 +3493,7 @@ async fn run_preview_script( run_query.timeout, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3566,6 +3575,7 @@ async fn run_bundle_preview_script( run_query.timeout, None, None, + Some(&authed.clone().into()), ) .await?; job_id = Some(uuid); @@ -3704,6 +3714,7 @@ async fn run_dependencies_job( None, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3758,6 +3769,7 @@ async fn run_flow_dependencies_job( None, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -3864,6 +3876,7 @@ async fn add_batch_jobs( None, None, None, + Some(&authed.clone().into()), ) .await?; tx = PushIsolationLevel::Transaction(ntx); @@ -3993,6 +4006,7 @@ async fn run_preview_flow_job( None, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; @@ -4069,6 +4083,7 @@ pub async fn run_job_by_hash( timeout, None, None, + Some(&authed.clone().into()), ) .await?; tx.commit().await?; diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index c7ab2cbcc5..a62926c9e9 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -219,7 +219,7 @@ async fn create_schedule( .await?; if ns.enabled.unwrap_or(true) { - tx = push_scheduled_job(&db, tx, &schedule).await? + tx = push_scheduled_job(&db, tx, &schedule, Some(&authed.clone().into())).await? } tx.commit().await?; @@ -299,7 +299,7 @@ async fn edit_schedule( .await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, &schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule, None).await?; } tx.commit().await?; @@ -506,7 +506,7 @@ pub async fn set_enabled( .await?; if payload.enabled { - tx = push_scheduled_job(&db, tx, &schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule, None).await?; } tx.commit().await?; @@ -554,7 +554,7 @@ pub async fn set_enabled( // .await?; // if payload.enabled { -// tx = push_scheduled_job(&db, tx, &schedule).await?; +// tx = push_scheduled_job(&db, tx, &schedule, None).await?; // } // tx.commit().await?; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index a7407c6830..8e70e7f280 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -648,7 +648,7 @@ async fn create_script_internal<'c>( clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?; if schedule.enabled { - tx = push_scheduled_job(&db, tx, &schedule).await?; + tx = push_scheduled_job(&db, tx, &schedule, None).await?; } } } else { @@ -751,6 +751,7 @@ async fn create_script_internal<'c>( None, None, None, + Some(&authed.clone().into()), ) .await?; Ok((hash, new_tx)) diff --git a/backend/windmill-api/src/static_assets.rs b/backend/windmill-api/src/static_assets.rs index ec45a32fef..9234661693 100644 --- a/backend/windmill-api/src/static_assets.rs +++ b/backend/windmill-api/src/static_assets.rs @@ -38,10 +38,7 @@ const TWO_HUNDRED: &str = "200.html"; fn serve_path(path: &str) -> Response { if path.starts_with("api/") { - return Response::builder() - .status(404) - .body(Body::empty()) - .unwrap(); + return Response::builder().status(404).body(Body::empty()).unwrap(); } match Asset::get(path) { Some(content) => { @@ -62,10 +59,9 @@ fn serve_path(path: &str) -> Response { res.body(body).unwrap() } - None if path.starts_with("_app/") => Response::builder() - .status(404) - .body(Body::empty()) - .unwrap(), + None if path.starts_with("_app/") => { + Response::builder().status(404).body(Body::empty()).unwrap() + } None => serve_path(TWO_HUNDRED), } } diff --git a/backend/windmill-api/src/utils.rs b/backend/windmill-api/src/utils.rs index dd2de8dc83..306d292ec7 100644 --- a/backend/windmill-api/src/utils.rs +++ b/backend/windmill-api/src/utils.rs @@ -9,20 +9,13 @@ use regex::Regex; use sqlx::{Postgres, Transaction}; use windmill_common::{ + auth::is_super_admin_email, error::{self, Error}, - users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL}, DB, }; pub async fn require_super_admin(db: &DB, email: &str) -> error::Result<()> { - if email == SUPERADMIN_SECRET_EMAIL || email == SUPERADMIN_NOTIFICATION_EMAIL { - return Ok(()); - } - let is_admin = sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email) - .fetch_optional(db) - .await - .map_err(|e| Error::InternalErr(format!("fetching super admin: {e:#}")))? - .unwrap_or(false); + let is_admin = is_super_admin_email(db, email).await?; if !is_admin { Err(Error::NotAuthorized( diff --git a/backend/windmill-common/src/auth.rs b/backend/windmill-common/src/auth.rs new file mode 100644 index 0000000000..29d6e076dd --- /dev/null +++ b/backend/windmill-common/src/auth.rs @@ -0,0 +1,199 @@ +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use tokio::sync::RwLock; + +use crate::{ + db::Authed, + error::{Error, Result}, + users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL}, + DB, +}; + +lazy_static::lazy_static! { + pub static ref JWT_SECRET : Arc> = Arc::new(RwLock::new("".to_string())); +} + +#[derive(Deserialize, Serialize)] +pub struct JWTAuthClaims { + pub email: String, + pub username: String, + pub is_admin: bool, + pub is_operator: bool, + pub groups: Vec, + pub folders: Vec<(String, bool, bool)>, + pub label: Option, + pub workspace_id: String, + pub exp: usize, +} + +#[derive(Deserialize)] +pub struct JobPerms { + pub workspace_id: String, + pub job_id: String, + pub email: String, + pub username: String, + pub is_admin: bool, + pub is_operator: bool, + pub groups: Vec, + pub folders: Vec, + pub created_at: chrono::NaiveDateTime, +} + +impl From for Authed { + fn from(value: JobPerms) -> Self { + Self { + email: value.email, + username: value.username, + is_admin: value.is_admin, + is_operator: value.is_operator, + groups: value.groups, + folders: value + .folders + .into_iter() + .filter_map(|x| serde_json::from_value::<(String, bool, bool)>(x).ok()) + .collect(), + scopes: None, + } + } +} + +pub async fn is_super_admin_email(db: &DB, email: &str) -> Result { + if email == SUPERADMIN_SECRET_EMAIL || email == SUPERADMIN_NOTIFICATION_EMAIL { + return Ok(true); + } + + let is_admin = sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email) + .fetch_optional(db) + .await + .map_err(|e| Error::InternalErr(format!("fetching super admin: {e:#}")))? + .unwrap_or(false); + + Ok(is_admin) +} + +pub fn permissioned_as_to_username(permissioned_as: &str) -> String { + if let Some((prefix, name)) = permissioned_as.split_once('/') { + if prefix == "u" { + name.to_string() + } else { + format!("group-{}", name) + } + } else { + permissioned_as.to_string() + } +} + +pub async fn fetch_authed_from_permissioned_as( + permissioned_as: String, + email: String, + w_id: &str, + db: &DB, +) -> Result { + let super_admin = + permissioned_as == SUPERADMIN_SYNC_EMAIL || is_super_admin_email(db, &email).await?; + if let Some((prefix, name)) = permissioned_as.split_once('/') { + if prefix == "u" { + let (is_admin, is_operator) = if super_admin { + (true, false) + } else { + let r = sqlx::query!( + "SELECT is_admin, operator FROM usr where username = $1 AND \ + workspace_id = $2 AND disabled = false", + name, + &w_id + ) + .fetch_one(db) + .await + .ok(); + if let Some(r) = r { + (r.is_admin, r.operator) + } else { + (false, true) + } + }; + + let groups = get_groups_for_user(w_id, &name, &email, db).await?; + + let folders = get_folders_for_user(w_id, &name, &groups, db).await?; + + Ok(Authed { + email: email, + username: name.to_string(), + is_admin, + is_operator, + groups, + folders, + scopes: None, + }) + } else { + let groups = vec![name.to_string()]; + let folders = get_folders_for_user(&w_id, "", &groups, db).await?; + Ok(Authed { + email: email, + username: format!("group-{name}"), + is_admin: false, + groups, + is_operator: false, + folders, + scopes: None, + }) + } + } else { + let groups = vec![]; + let folders = vec![]; + Ok(Authed { + email: email, + username: permissioned_as, + is_admin: super_admin, + is_operator: true, + groups, + folders, + scopes: None, + }) + } +} + +pub async fn get_folders_for_user( + w_id: &str, + username: &str, + groups: &[String], + db: &DB, +) -> Result> { + let mut perms = groups + .into_iter() + .map(|x| format!("g/{}", x)) + .collect::>(); + perms.insert(0, format!("u/{}", username)); + let folders = sqlx::query!( + "SELECT name, (EXISTS (SELECT 1 FROM (SELECT key, value FROM jsonb_each_text(extra_perms) WHERE key = ANY($1)) t WHERE value::boolean IS true)) as write, $1 && owners::text[] as owner FROM folder + WHERE extra_perms ?| $1 AND workspace_id = $2", + &perms[..], + w_id, + ) + .fetch_all(db) + .await? + .into_iter() + .map(|x| (x.name, x.write.unwrap_or(false), x.owner.unwrap_or(false))) + .collect(); + + Ok(folders) +} + +pub async fn get_groups_for_user( + w_id: &str, + username: &str, + email: &str, + db: &DB, +) -> Result> { + let groups = sqlx::query_scalar!( + "SELECT group_ FROM usr_to_group where usr = $1 AND workspace_id = $2 UNION ALL SELECT igroup FROM email_to_igroup WHERE email = $3", + username, + w_id, + email + ) + .fetch_all(db) + .await? + .into_iter().filter_map(|x| x) + .collect(); + Ok(groups) +} diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 1fcc0d0fdd..2dad18e8b9 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -26,6 +26,7 @@ pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation pub const HUB_BASE_URL_SETTING: &str = "hub_base_url"; pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels"; pub const DEV_INSTANCE_SETTING: &str = "dev_instance"; +pub const JWT_SECRET_SETTING: &str = "jwt_secret"; pub const ENV_SETTINGS: [&str; 50] = [ "DISABLE_NSJAIL", diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index bd25dcd608..eb2cf6eb6b 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -32,6 +32,7 @@ pub mod more_serde; pub mod oauth2; pub mod s3_helpers; +pub mod auth; pub mod schedule; pub mod scripts; pub mod server; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 38584161e4..a782cd3437 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -45,6 +45,7 @@ use windmill_audit::ActionKind; #[cfg(not(feature = "enterprise"))] use windmill_common::worker::PriorityTags; use windmill_common::{ + auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, db::{Authed, UserDB}, error::{self, to_anyhow, Error}, flow_status::{ @@ -111,6 +112,7 @@ lazy_static::lazy_static! { .build().unwrap(); + pub static ref JOB_TOKEN: Option = std::env::var("JOB_TOKEN").ok(); } #[cfg(feature = "cloud")] @@ -808,6 +810,12 @@ pub async fn add_completed_job< tracing::debug!("decremented concurrency counter"); } + if JOB_TOKEN.is_none() { + sqlx::query!("DELETE FROM job_perms WHERE job_id = $1", job_id) + .execute(&mut tx) + .await?; + } + tx.commit().await?; tracing::info!( %job_id, @@ -1015,6 +1023,7 @@ pub async fn add_completed_job< queued_job.timeout, None, queued_job.priority, + None, ) .await?; if let Err(e) = tx.commit().await { @@ -1220,7 +1229,7 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon if schedule.enabled && script_path == schedule.script_path { let push_next_job_future = async { let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into(); - tx = push_scheduled_job(db, tx, &schedule).await?; + tx = push_scheduled_job(db, tx, &schedule, None).await?; tx.commit().await?; Ok::<(), Error>(()) }; @@ -1504,6 +1513,7 @@ pub async fn push_error_handler< None, None, priority, + None, ) .await?; tx.commit().await?; @@ -1613,6 +1623,7 @@ async fn handle_recovered_schedule< None, None, None, + None, ) .await?; tracing::info!( @@ -2813,6 +2824,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( custom_timeout: Option, flow_step_id: Option, _priority_override: Option, + authed: Option<&Authed>, ) -> Result<(Uuid, QueueTransaction<'c, R>), Error> { #[cfg(feature = "cloud")] if *CLOUD_HOSTED { @@ -2874,7 +2886,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( && email != SCHEDULE_RECOVERY_HANDLER_USER_EMAIL && email != "worker@windmill.dev" && email != SUPERADMIN_SECRET_EMAIL - && email != SUPERADMIN_SYNC_EMAIL + && permissioned_as != SUPERADMIN_SYNC_EMAIL && email != SUPERADMIN_NOTIFICATION_EMAIL { let user_usage = if let Some(user_usage) = user_usage { @@ -3633,9 +3645,59 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( QUEUE_PUSH_COUNT.inc(); } + let job_authed = match authed { + Some(authed) + if authed.email == email + && authed.username == permissioned_as_to_username(&permissioned_as) => + { + authed.clone() + } + _ => { + if authed.is_some() { + tracing::warn!("Authed passed to push is not the same as permissioned_as, refetching direclty permissions for job {job_id}...") + } + fetch_authed_from_permissioned_as( + permissioned_as.clone(), + email.to_string(), + workspace_id, + _db, + ) + .await + .map_err(|e| { + Error::InternalErr(format!( + "Could not get permissions directly for job {job_id}: {e:#}" + )) + })? + } + }; + + let folders = job_authed + .folders + .iter() + .filter_map(|x| serde_json::to_value(x).ok()) + .collect::>(); + + if JOB_TOKEN.is_none() { + if let Err(err) = sqlx::query!("INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) + values ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8", + job_id, + job_authed.email, + job_authed.username, + job_authed.is_admin, + job_authed.is_operator, + folders.as_slice(), + job_authed.groups.as_slice(), + workspace_id, + ).execute(&mut tx).await { + tracing::error!("Could not insert job_perms for job {job_id}: {err:#}"); + } + } + { let uuid_string = job_id.to_string(); let uuid_str = uuid_string.as_str(); + let mut hm = HashMap::from([("uuid", uuid_str), ("permissioned_as", &permissioned_as)]); let s: String; diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 04227bcc52..7968af0bb5 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -12,6 +12,7 @@ use crate::QueueTransaction; use sqlx::{query_scalar, Postgres, Transaction}; use std::collections::HashMap; use std::str::FromStr; +use windmill_common::db::Authed; use windmill_common::flows::Retry; use windmill_common::jobs::JobPayload; use windmill_common::schedule::schedule_to_user; @@ -27,6 +28,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( db: &DB, mut tx: QueueTransaction<'c, R>, schedule: &Schedule, + authed: Option<&Authed>, ) -> Result> { let sched = cron::Schedule::from_str(schedule.schedule.as_ref()) .map_err(|e| error::Error::BadRequest(e.to_string()))?; @@ -205,6 +207,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( timeout, None, None, + authed, ) .await?; Ok(tx) // TODO: Bubble up pushed UUID from here diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index a8c4b7919c..f4fda2a201 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -11,7 +11,7 @@ path = "src/lib.rs" [features] default = [] prometheus = ["dep:prometheus", "windmill-common/prometheus"] -enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] +enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:pem", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] benchmark = ["windmill-queue/benchmark"] flamegraph = [] parquet = ["windmill-common/parquet", "dep:object_store"] @@ -70,7 +70,7 @@ mysql_async.workspace = true base64.workspace = true gcp_auth = { workspace = true, optional = true } rust_decimal.workspace = true -jsonwebtoken = { workspace = true, optional = true } +jsonwebtoken.workspace = true sha2.workspace = true pem = { workspace = true, optional = true } urlencoding.workspace = true diff --git a/backend/windmill-worker/src/config.rs b/backend/windmill-worker/src/config.rs index e69de29bb2..8b13789179 100644 --- a/backend/windmill-worker/src/config.rs +++ b/backend/windmill-worker/src/config.rs @@ -0,0 +1 @@ + diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index ee5b89c220..d6d56968de 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -58,8 +58,9 @@ pub async fn do_mysql( ) .await?; - let as_raw = serde_json::from_value(val) - .map_err(|e| Error::InternalErr(format!("Error while parsing inline resource: {e:#}")))?; + let as_raw = serde_json::from_value(val).map_err(|e| { + Error::InternalErr(format!("Error while parsing inline resource: {e:#}")) + })?; Some(as_raw) } else { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 4fbdcfe7b3..b376494f9c 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -6,7 +6,10 @@ * LICENSE-AGPL for a copy of the license. */ -use windmill_common::worker::{get_windmill_memory_usage, get_worker_memory_usage, TMP_DIR}; +use windmill_common::{ + auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET}, + worker::{get_windmill_memory_usage, get_worker_memory_usage, TMP_DIR}, +}; use anyhow::Result; use const_format::concatcp; @@ -41,8 +44,8 @@ use windmill_common::{ get_latest_deployed_hash_for_path, jobs::{JobKind, QueuedJob}, scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang, PREVIEW_IS_CODEBASE_HASH}, - users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL}, - utils::{rd_string, StripPath}, + users::SUPERADMIN_SECRET_EMAIL, + utils::StripPath, worker::{to_raw_value, update_ping, CLOUD_HOSTED, NO_LOGS, WORKER_CONFIG, WORKER_GROUP}, DB, IS_READY, }; @@ -164,32 +167,59 @@ pub async fn create_token_for_owner( return Ok(token.clone()); } - let token: String = rd_string(32); - let is_super_admin = - sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email) - .fetch_optional(db) - .await? - .unwrap_or(false) - || email == SUPERADMIN_SECRET_EMAIL - || email == SUPERADMIN_NOTIFICATION_EMAIL - || owner == SUPERADMIN_SYNC_EMAIL; + let jwt_secret = JWT_SECRET.read().await; - sqlx::query_scalar!( - "INSERT INTO token - (workspace_id, token, owner, label, expiration, super_admin, email, job) - VALUES ($1, $2, $3, $4, now() + ($5 || ' seconds')::interval, $6, $7, $8)", - &w_id, - token, - owner, - label, - expires_in.to_string(), - is_super_admin, - email, - job_id + if jwt_secret.is_empty() { + return Err(Error::InternalErr("No JWT secret found".to_string())); + } + + let job_authed = match sqlx::query_as!( + JobPerms, + "SELECT * FROM job_perms WHERE job_id = $1 AND workspace_id = $2", + job_id, + w_id ) - .execute(db) - .await?; - Ok(token) + .fetch_optional(db) + .await + { + Ok(Some(jp)) => jp.into(), + _ => { + tracing::warn!("Could not get permissions for job {job_id} from job_perms table, getting permissions directly..."); + fetch_authed_from_permissioned_as(owner.to_string(), email.to_string(), w_id, db) + .await + .map_err(|e| { + Error::InternalErr(format!( + "Could not get permissions directly for job {job_id}: {e:#}" + )) + })? + } + }; + + let payload = JWTAuthClaims { + email: job_authed.email, + username: job_authed.username, + is_admin: job_authed.is_admin, + is_operator: job_authed.is_operator, + groups: job_authed.groups, + folders: job_authed.folders, + label: Some(label.to_string()), + workspace_id: w_id.to_string(), + exp: (chrono::Utc::now() + chrono::Duration::seconds(expires_in as i64)).timestamp() + as usize, + }; + + let token = jsonwebtoken::encode( + &jsonwebtoken::Header::new(jsonwebtoken::Algorithm::HS256), + &payload, + &jsonwebtoken::EncodingKey::from_secret(jwt_secret.as_bytes()), + ) + .map_err(|err| { + Error::InternalErr(format!( + "Could not encode JWT token for job {job_id}: {:?}", + err + )) + })?; + Ok(format!("jwt_{}", token)) } pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs"); @@ -1918,6 +1948,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>( None, None, None, + None, ) .await?; inner_tx.commit().await?; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 5a44fb42b5..f37de0b9f3 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -15,7 +15,7 @@ use std::time::Duration; use crate::common::{hash_args, save_in_cache}; use crate::js_eval::{eval_timeout, IdContext}; -use crate::{AuthedClient, PreviousResult, SameWorkerPayload, SendResult, KEEP_JOB_DIR}; +use crate::{AuthedClient, PreviousResult, SameWorkerPayload, SendResult, JOB_TOKEN, KEEP_JOB_DIR}; use anyhow::Context; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; @@ -25,6 +25,8 @@ use sqlx::FromRow; use tokio::sync::mpsc::Sender; use tracing::instrument; use uuid::Uuid; +use windmill_common::auth::JobPerms; +use windmill_common::db::Authed; use windmill_common::flow_status::{ ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, }; @@ -2095,6 +2097,25 @@ async fn push_next_flow_job flow_job.root_job.or_else(|| Some(flow_job.id)) }; + // forward root job permissions to the new job + let job_perms: Option = if JOB_TOKEN.is_none() { + if let Some(root_job) = &flow_job.root_job.or_else(|| Some(flow_job.id)) { + sqlx::query_as!( + JobPerms, + "SELECT * FROM job_perms WHERE job_id = $1 AND workspace_id = $2", + root_job, + flow_job.workspace_id, + ) + .fetch_optional(&mut tx) + .await? + .map(|x| x.into()) + } else { + None + } + } else { + None + }; + let tx2 = PushIsolationLevel::Transaction(tx); let (uuid, mut inner_tx) = push( &db, @@ -2125,6 +2146,7 @@ async fn push_next_flow_job payload_tag.timeout, Some(module.id.clone()), new_job_priority_override, + job_perms.as_ref(), ) .await?; diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 3065274c7d..1f86a7887c 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -441,6 +441,7 @@ async fn trigger_dependents_to_recompute_dependencies< None, None, None, + None, ) .await?; tracing::info!(