kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: replace ephemeral tokens by jwt (#3908)

Browse Source 
* feat: replace ephemeral tokens by jwt

* fix: migration + sqlx

* fix: tests, agent, ee ref

* fix: sqlx

* fix: tests

* fix: jwt prefix change + fix tests

* fix: nit

* fix: handle agent mode
This commit is contained in:
HugoCasa
2024-06-14 13:07:35 +02:00
committed by GitHub
parent fa365027a3
commit 93f58056a0
35 changed files with 943 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
backend/.sqlx/query-42cb4bc3abc8cba2193a7fbbae8275c1bc830081a43e3a4a897c41b6b3099b1d.json generated
View File

@@ -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"
}

14
backend/.sqlx/query-64a3b92ce4366518423bc536efe33415b43b325d7380e0c7062c392c7ba51e59.json generated Normal file
View File

@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM job_perms WHERE job_id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "64a3b92ce4366518423bc536efe33415b43b325d7380e0c7062c392c7ba51e59"
}

21
backend/.sqlx/query-8efd06387ded837d7849adafe5bc93acb882ef90fc58b023650c875e0fd17047.json generated Normal file
View File

@@ -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"
}

15
backend/.sqlx/query-a00e61e770e20157bbd9e4cdedf7fb5f9de7c8c9e50282e3ecf2e3ce917ec37a.json generated Normal file
View File

@@ -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"
}

71
backend/.sqlx/query-d2def87d7f7901eebc65082f7df5e0a33e5702b25c3db3affa06155e90480e42.json generated Normal file
View File

@@ -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"
}

1
backend/Cargo.lock generated
View File

@@ -10185,6 +10185,7 @@ dependencies = [
"http 1.1.0",
"hyper 1.3.1",
"itertools 0.13.0",
"jsonwebtoken",
"lazy_static",
"magic-crypt",
"mime_guess",

2
backend/build.rs
View File

@@ -2,4 +2,4 @@
fn main() {
// trigger recompilation when a new migration is added
println!("cargo:rerun-if-changed=migrations");
}
}

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
50a2701b6bf16c55677066ed4aeeaf32cd5a66a8
9f179a95ce20062411cd73dfe5cbc4f22ef522e2

1
backend/migrations/20240613150524_add_job_perms.down.sql Normal file
View File

@@ -0,0 +1 @@
-- Add down migration script here

13
backend/migrations/20240613150524_add_job_perms.up.sql Normal file
View File

@@ -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)
);

23
backend/src/main.rs
View File

@@ -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);
}

124
backend/src/monitor.rs
View File

@@ -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<Postgres>) {
.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<Postgres>) {
// 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<i64>,
memory: Option<i64>,
}
pub async fn save_usage_metrics(db: &Pool<Postgres>) {
@@ -876,7 +894,10 @@ pub async fn save_usage_metrics(db: &Pool<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
.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<Postgres>) {
// 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<Postgres>) {
tx.commit().await.ok();
}
}
pub async fn reload_server_config(db: &Pool<Postgres>) {
@@ -1103,7 +1125,9 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
mpsc::channel::<SameWorkerPayload>(1);
let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::<SendResult>(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<String> {
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::<String>(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(())
}

527
backend/tests/worker.rs
View File

@@ -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<ScriptHash>,
pub language: Option<ScriptLang>,
pub job_kind: JobKind,
}
impl CompletedJob {
@@ -168,6 +164,12 @@ fn find_module_in_vec(modules: Vec<FlowStatusModule>, id: &str) -> Option<FlowSt
modules.into_iter().find(|s| s.id() == id)
}
async fn set_jwt_secret() -> () {
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<Postgres>) {
}))
.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::<Vec<_>>()))
.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::<Vec<_>>()))
.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<Postgres>) {
}))
.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::<Vec<_>>()))
.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::<Vec<_>>()))
.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::<rsmq_async::MultiplexedRsmq>(
&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<Fut: std::future::Future>(
inner: Fut,
port: u16,
) -> <Fut as std::future::Future>::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::<rsmq_async::MultiplexedRsmq>(
&db,
@@ -1028,11 +1056,13 @@ async fn listen_for_uuid_on(
}
async fn completed_job(uuid: Uuid, db: &Pool<Postgres>) -> 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
}))
.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<Postgres>) {
}))
.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<Postgres>) {
.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<Postgres>) {
.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<Postgres>) {
.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<Postgres>) {
}))
.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<Postgres>) {
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::<i32>(first_run_result.json_result().unwrap()).expect("first_run_result was not an int");
let restarted_flow_result_int = serde_json::from_value::<i32>(restarted_flow_result.json_result().unwrap()).expect("restarted_flow_result was not an int");
let first_run_result_int =
serde_json::from_value::<i32>(first_run_result.json_result().unwrap())
.expect("first_run_result was not an int");
let restarted_flow_result_int =
serde_json::from_value::<i32>(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<Postgres>) {
.unwrap();
}
#[cfg(feature = "enterprise")]
#[sqlx::test(fixtures("base", "schedule"))]
async fn test_script_schedule_handlers(db: Pool<Postgres>) {
@@ -2980,7 +3022,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
}
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<Postgres>) {
.await;
}
#[cfg(feature = "enterprise")]
#[sqlx::test(fixtures("base", "schedule"))]
async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
@@ -3126,7 +3167,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
}
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<Postgres>) {
.await;
}
async fn run_deployed_relative_imports(db: &Pool<Postgres>, script_content: String, language: ScriptLang) {
async fn run_deployed_relative_imports(
db: &Pool<Postgres>,
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<Postgres>, 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<Postgres>, script_content: String, language: ScriptLang) {
async fn run_preview_relative_imports(
db: &Pool<Postgres>,
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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;

1
backend/windmill-api/Cargo.toml
View File

@@ -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

3
backend/windmill-api/src/apps.rs
View File

@@ -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?;

9
backend/windmill-api/src/ee.rs
View File

@@ -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<String> {
// 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
}

4
backend/windmill-api/src/flows.rs
View File

@@ -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!(

26
backend/windmill-api/src/folders.rs
View File

@@ -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<Vec<(String, bool, bool)>> {
let mut perms = groups
.into_iter()
.map(|x| format!("g/{}", x))
.collect::<Vec<_>>();
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<UserDB>,

5
backend/windmill-api/src/groups.rs
View File

@@ -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};

15
backend/windmill-api/src/jobs.rs
View File

@@ -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?;

8
backend/windmill-api/src/schedule.rs
View File

@@ -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?;

3
backend/windmill-api/src/scripts.rs
View File

@@ -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))

12
backend/windmill-api/src/static_assets.rs
View File

@@ -38,10 +38,7 @@ const TWO_HUNDRED: &str = "200.html";
fn serve_path(path: &str) -> Response<Body> {
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<Body> {
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),
}
}

11
backend/windmill-api/src/utils.rs
View File

@@ -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(

199
backend/windmill-common/src/auth.rs Normal file
View File

@@ -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<RwLock<String>> = 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<String>,
pub folders: Vec<(String, bool, bool)>,
pub label: Option<String>,
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<String>,
pub folders: Vec<serde_json::Value>,
pub created_at: chrono::NaiveDateTime,
}
impl From<JobPerms> 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<bool> {
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<Authed> {
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<Vec<(String, bool, bool)>> {
let mut perms = groups
.into_iter()
.map(|x| format!("g/{}", x))
.collect::<Vec<_>>();
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<Vec<String>> {
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)
}

1
backend/windmill-common/src/global_settings.rs
View File

@@ -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",

1
backend/windmill-common/src/lib.rs
View File

@@ -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;

66
backend/windmill-queue/src/jobs.rs
View File

@@ -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<String> = 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<i32>,
flow_step_id: Option<String>,
_priority_override: Option<i16>,
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::<Vec<_>>();
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;

3
backend/windmill-queue/src/schedule.rs
View File

@@ -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<QueueTransaction<'c, R>> {
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

4
backend/windmill-worker/Cargo.toml
View File

@@ -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

1
backend/windmill-worker/src/config.rs
View File

@@ -0,0 +1 @@

5
backend/windmill-worker/src/mysql_executor.rs
View File

@@ -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 {

85
backend/windmill-worker/src/worker.rs
View File

@@ -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?;

24
backend/windmill-worker/src/worker_flow.rs
View File

@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
flow_job.root_job.or_else(|| Some(flow_job.id))
};
// forward root job permissions to the new job
let job_perms: Option<Authed> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
payload_tag.timeout,
Some(module.id.clone()),
new_job_priority_override,
job_perms.as_ref(),
)
.await?;

1
backend/windmill-worker/src/worker_lockfiles.rs
View File

@@ -441,6 +441,7 @@ async fn trigger_dependents_to_recompute_dependencies<
None,
None,
None,
None,
)
.await?;
tracing::info!(