From 77aaf2724cda66041ed266027e8bc59ea2a68188 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 18 Feb 2024 14:03:00 +0100 Subject: [PATCH] chore: remove warnings --- backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 8 +- .../windmill-api/src/concurrency_groups.rs | 17 +- backend/windmill-api/src/db.rs | 7 +- backend/windmill-api/src/flows.rs | 23 +- backend/windmill-api/src/jobs.rs | 82 +++-- backend/windmill-api/src/lib.rs | 6 +- backend/windmill-api/src/oidc.rs | 305 ------------------ backend/windmill-api/src/oidc_ee.rs | 17 + backend/windmill-api/src/scripts.rs | 23 +- backend/windmill-api/src/workspaces.rs | 42 ++- backend/windmill-queue/src/jobs.rs | 10 +- backend/windmill-worker/src/common.rs | 29 +- .../windmill-worker/src/python_executor.rs | 6 +- backend/windmill-worker/src/worker.rs | 21 +- backend/windmill-worker/src/worker_flow.rs | 8 +- 16 files changed, 206 insertions(+), 400 deletions(-) delete mode 100644 backend/windmill-api/src/oidc.rs create mode 100644 backend/windmill-api/src/oidc_ee.rs diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index d745e03cdf..036bd91159 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -045dc822a95f2230e4ae97ac492df0dd9f797af6 \ No newline at end of file +3de554fc89db8f9c6cc544c6a2ba004e65a7c133 \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 0e6e72094a..b2a776e7fe 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -33,8 +33,12 @@ use windmill_common::{ stats::schedule_stats, utils::{rd_string, Mode}, worker::{reload_custom_tags_setting, WORKER_GROUP}, - DB, METRICS_ADDR, METRICS_ENABLED, + DB, METRICS_ENABLED, }; + +#[cfg(feature = "enterprise")] +use windmill_common::METRICS_ADDR; + use windmill_worker::{ BUN_CACHE_DIR, BUN_TMP_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM, DENO_TMP_CACHE_DIR, DENO_TMP_CACHE_DIR_DEPS, DENO_TMP_CACHE_DIR_NPM, GO_BIN_CACHE_DIR, @@ -140,7 +144,7 @@ async fn windmill_main() -> anyhow::Result<()> { { panic!("Agent mode is only available in the EE, ignoring..."); } - + #[cfg(feature = "enterprise")] Mode::Agent } else { if &x != "standalone" { diff --git a/backend/windmill-api/src/concurrency_groups.rs b/backend/windmill-api/src/concurrency_groups.rs index d66da5417f..836aed26d3 100644 --- a/backend/windmill-api/src/concurrency_groups.rs +++ b/backend/windmill-api/src/concurrency_groups.rs @@ -1,12 +1,23 @@ +#[cfg(feature = "enterprise")] use crate::db::{ApiAuthed, DB}; -use std::collections::HashMap; - +#[cfg(feature = "enterprise")] use axum::extract::Path; +#[cfg(feature = "enterprise")] use axum::routing::{delete, get}; -use axum::{Extension, Json, Router}; +#[cfg(feature = "enterprise")] +use axum::{Extension, Json}; + +use axum::Router; + +#[cfg(feature = "enterprise")] use polars::prelude::IntoVec; +#[cfg(feature = "enterprise")] use serde::Serialize; +#[cfg(feature = "enterprise")] +use std::collections::HashMap; +#[cfg(feature = "enterprise")] use windmill_common::error::Error::{InternalErr, PermissionDenied}; +#[cfg(feature = "enterprise")] use windmill_common::error::JsonResult; #[cfg(feature = "enterprise")] diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 504b1de0cb..670affc92b 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -6,7 +6,10 @@ * LICENSE-AGPL for a copy of the license. */ -use sqlx::{migrate::Migrate, pool::PoolConnection, Executor, Pool, Postgres}; +#[cfg(feature = "enterprise")] +use sqlx::Executor; + +use sqlx::{migrate::Migrate, pool::PoolConnection, Pool, Postgres}; use windmill_common::{ db::{Authable, Authed}, error::Error, @@ -103,6 +106,7 @@ pub async fn migrate(db: &DB) -> Result<(), Error> { Err(err) => Err(err), }?; + #[cfg(feature = "enterprise")] if let Err(e) = windmill_migrations(&mut custom_migrator, db).await { tracing::error!("Could not apply windmill custom migrations: {e}") } @@ -110,6 +114,7 @@ pub async fn migrate(db: &DB) -> Result<(), Error> { Ok(()) } +#[cfg(feature = "enterprise")] async fn windmill_migrations(migrator: &mut CustomMigrator, db: &DB) -> Result<(), Error> { if std::env::var("MIGRATION_NO_BYPASSRLS").is_ok() { #[cfg(feature = "enterprise")] diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 7f23abb0d5..b2878f0666 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -221,20 +221,27 @@ pub async fn get_hub_flow_by_id( pub struct ToggleWorkspaceErrorHandler { pub muted: Option, } + +#[cfg(not(feature = "enterprise"))] +async fn toggle_workspace_error_handler( + _authed: ApiAuthed, + Extension(_user_db): Extension, + Path((_w_id, _path)): Path<(String, StripPath)>, + Json(_req): Json, +) -> Result { + return Err(Error::BadRequest( + "Muting the error handler for certain flow is only available in enterprise version" + .to_string(), + )); +} + +#[cfg(feature = "enterprise")] async fn toggle_workspace_error_handler( authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, Json(req): Json, ) -> Result { - #[cfg(not(feature = "enterprise"))] - { - return Err(Error::BadRequest( - "Muting the error handler for certain flow is only available in enterprise version" - .to_string(), - )); - } - let mut tx = user_db.begin(&authed).await?; let error_handler_maybe: Option = sqlx::query_scalar!( diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index eb6a01071c..d9556b3c7c 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1422,9 +1422,9 @@ pub async fn get_suspended_job_flow( } fn conditionally_require_authed_user( - authed: Option, + _authed: Option, flow_status: FlowStatus, - trigger_email: &str, + _trigger_email: &str, ) -> error::Result<()> { let approval_conditions_opt = flow_status.approval_conditions; @@ -1434,37 +1434,41 @@ fn conditionally_require_authed_user( let approval_conditions = approval_conditions_opt.unwrap(); if approval_conditions.user_auth_required { - #[cfg(not(feature = "enterprise"))] - return Err(Error::BadRequest( - "Approvals for logged in users is an enterprise only feature".to_string(), - )); - #[cfg(feature = "enterprise")] { - if authed.is_none() { - return Err(Error::NotAuthorized( - "Only logged in users can approve this flow step".to_string(), - )); - } + #[cfg(not(feature = "enterprise"))] + return Err(Error::BadRequest( + "Approvals for logged in users is an enterprise only feature".to_string(), + )); - let authed = authed.unwrap(); - if !authed.is_admin { - if approval_conditions.self_approval_disabled && authed.email.eq(trigger_email) { - return Err(Error::PermissionDenied( - "Self-approval is disabled for this flow step".to_string(), + #[cfg(feature = "enterprise")] + { + if _authed.is_none() { + return Err(Error::NotAuthorized( + "Only logged in users can approve this flow step".to_string(), )); } - if !approval_conditions.user_groups_required.is_empty() { - #[cfg(feature = "enterprise")] + let authed = _authed.unwrap(); + if !authed.is_admin { + if approval_conditions.self_approval_disabled && authed.email.eq(_trigger_email) { - for required_group in approval_conditions.user_groups_required.iter() { - if authed.groups.contains(&required_group) { - return Ok(()); + return Err(Error::PermissionDenied( + "Self-approval is disabled for this flow step".to_string(), + )); + } + + if !approval_conditions.user_groups_required.is_empty() { + #[cfg(feature = "enterprise")] + { + for required_group in approval_conditions.user_groups_required.iter() { + if authed.groups.contains(&required_group) { + return Ok(()); + } } - } - let error_msg = format!("Only users from one of the following groups are allowed to approve this workflow: {}", + let error_msg = format!("Only users from one of the following groups are allowed to approve this workflow: {}", approval_conditions.user_groups_required.join(", ")); - return Err(Error::PermissionDenied(error_msg)); + return Err(Error::PermissionDenied(error_msg)); + } } } } @@ -1962,6 +1966,26 @@ pub async fn run_flow_by_path( Ok((StatusCode::CREATED, uuid.to_string())) } +#[cfg(not(feature = "enterprise"))] +pub async fn restart_flow( + _authed: ApiAuthed, + Extension(_db): Extension, + Extension(_user_db): Extension, + Extension(_rsmq): Extension>, + Path((_w_id, _job_id, _step_id, _branch_or_iteration_n)): Path<( + String, + Uuid, + String, + Option, + )>, + Query(_run_query): Query, +) -> error::Result<(StatusCode, String)> { + return Err(Error::BadRequest( + "Restarting a flow is a feature only available in enterprise version".to_string(), + )); +} + +#[cfg(feature = "enterprise")] pub async fn restart_flow( authed: ApiAuthed, Extension(db): Extension, @@ -1975,14 +1999,6 @@ pub async fn restart_flow( )>, Query(run_query): Query, ) -> error::Result<(StatusCode, String)> { - #[cfg(not(feature = "enterprise"))] - { - return Err(Error::BadRequest( - "Restarting a flow is a feature only available in enterprise version".to_string(), - )); - } - - #[cfg(feature = "enterprise")] check_license_key_valid().await?; let completed_job = sqlx::query_as::<_, CompletedJob>( diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 000a710b42..f1e6412f76 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -61,7 +61,7 @@ pub mod job_helpers_ee; pub mod job_metrics; pub mod jobs; pub mod oauth2_ee; -mod oidc; +mod oidc_ee; mod openai; mod raw_apps; mod resources; @@ -209,7 +209,7 @@ pub async fn run_server( ) .nest("/variables", variables::workspaced_service()) .nest("/workspaces", workspaces::workspaced_service()) - .nest("/oidc", oidc::workspaced_service()), + .nest("/oidc", oidc_ee::workspaced_service()), ) .nest("/workspaces", workspaces::global_service()) .nest( @@ -232,7 +232,7 @@ pub async fn run_server( .route_layer(from_extractor::()) .route_layer(from_extractor::()) .nest("/jobs", jobs::global_root_service()) - .nest("/oidc", oidc::global_service()) + .nest("/oidc", oidc_ee::global_service()) .nest( "/saml", saml_ee::global_service().layer(Extension(Arc::clone(&sp_extension))), diff --git a/backend/windmill-api/src/oidc.rs b/backend/windmill-api/src/oidc.rs deleted file mode 100644 index 4a03200b90..0000000000 --- a/backend/windmill-api/src/oidc.rs +++ /dev/null @@ -1,305 +0,0 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2023 - * This file and its contents are licensed under the AGPLv3 License. - * Please see the included NOTICE for copyright information and - * LICENSE-AGPL for a copy of the license. - */ - -use std::process::Command; - -use anyhow; - -#[cfg(feature = "enterprise")] -use openidconnect::{ - core::{ - CoreClaimName, CoreJsonWebKeySet, CoreJwsSigningAlgorithm, CoreProviderMetadata, - CoreResponseType, CoreRsaPrivateSigningKey, CoreSubjectIdentifierType, - }, - AdditionalClaims, AuthUrl, EmptyAdditionalProviderMetadata, IssuerUrl, JsonWebKeyId, - JsonWebKeySetUrl, ResponseTypes, -}; - -#[cfg(feature = "enterprise")] -impl AdditionalClaims for JobClaim {} - -use crate::db::DB; -use axum::extract::Path; -use axum::routing::{get, post}; -use axum::Extension; -use axum::{Json, Router}; -use serde::{Deserialize, Serialize}; - -#[cfg(feature = "enterprise")] -pub fn global_service() -> Router { - Router::new() - .route( - "/.well-known/openid-configuration", - get(openid_configuration), - ) - .route("/jwks", get(jwks)) -} -#[cfg(not(feature = "enterprise"))] -pub fn global_service() -> Router { - Router::new() -} - -#[cfg(not(feature = "enterprise"))] -pub fn workspaced_service() -> Router { - Router::new() -} - -#[cfg(feature = "enterprise")] -pub fn workspaced_service() -> Router { - Router::new().route("/token/:audience", post(gen_token)) -} - -#[cfg(feature = "enterprise")] -#[derive(Debug, Clone, serde::Serialize)] -struct Keys { - private_key: String, -} - -#[cfg(feature = "enterprise")] -async fn gen_pems(db: &DB) -> anyhow::Result { - let private_key_cmd = Command::new("openssl") - .arg("genrsa") - .arg("--traditional") - .arg("2048") - .output() - .expect("failed to execute process"); - - let private_key = String::from_utf8(private_key_cmd.stdout).unwrap(); - - tracing::debug!("Generated private key: {}", private_key); - let keys = Keys { private_key }; - - sqlx::query!( - "INSERT INTO global_settings (name, value) VALUES ('rsa_keys', $1)", - serde_json::to_value(&keys).unwrap() - ) - .execute(db) - .await?; - - Ok(keys) -} - -#[cfg(feature = "enterprise")] -async fn get_private_key(db: &DB) -> anyhow::Result { - let key = sqlx::query_scalar!( - "SELECT value->>'private_key' FROM global_settings WHERE name = 'rsa_keys'", - ) - .fetch_optional(db) - .await? - .flatten(); - - if let Some(key) = key { - return Ok(key); - } else { - let keys = gen_pems(db).await?; - return Ok(keys.private_key); - } -} - -#[cfg(feature = "enterprise")] -pub async fn jwks( - Extension(db): Extension, -) -> windmill_common::error::JsonResult { - use openidconnect::PrivateSigningKey; - - let private_key = get_private_key(&db).await?; - let jwks = CoreJsonWebKeySet::new(vec![CoreRsaPrivateSigningKey::from_pem( - &private_key, - Some(JsonWebKeyId::new("windmill".to_string())), - ) - .map_err(|e| anyhow::anyhow!("Failed to parse PEM: {}", e))? - .as_verification_key()]); - - Ok(Json(jwks)) -} - -#[cfg(feature = "enterprise")] -pub async fn openid_configuration() -> windmill_common::error::JsonResult { - use windmill_common::BASE_URL; - - let base_url = BASE_URL.read().await.clone(); - return get_provider_metadata(base_url) - .map(Json) - .map_err(|e| e.into()); -} - -#[cfg(feature = "enterprise")] -pub fn get_provider_metadata(base_url: String) -> anyhow::Result { - let provider_metadata = CoreProviderMetadata::new( - IssuerUrl::new(format!("{base_url}/api/oidc/"))?, - AuthUrl::new(format!("{base_url}/api/oidc/"))?, - JsonWebKeySetUrl::new(format!("{base_url}/api/oidc/jwks"))?, - vec![ - // Optional: support the implicit flow. - ResponseTypes::new(vec![CoreResponseType::Token, CoreResponseType::IdToken]), // Other flows including hybrid flows may also be specified here. - ], - vec![CoreSubjectIdentifierType::Public], - vec![CoreJwsSigningAlgorithm::RsaSsaPssSha256], - EmptyAdditionalProviderMetadata {}, - ) - // Recommended: specify the supported ID token claims. - .set_claims_supported(Some(vec![ - // Providers may also define an enum instead of using CoreClaimName. - CoreClaimName::new("sub".to_string()), - CoreClaimName::new("aud".to_string()), - CoreClaimName::new("email".to_string()), - CoreClaimName::new("email_verified".to_string()), - CoreClaimName::new("exp".to_string()), - CoreClaimName::new("iat".to_string()), - CoreClaimName::new("iss".to_string()), - CoreClaimName::new("job_id".to_string()), - CoreClaimName::new("path".to_string()), - CoreClaimName::new("flow_path".to_string()), - CoreClaimName::new("groups".to_string()), - CoreClaimName::new("username".to_string()), - CoreClaimName::new("workspace".to_string()), - ])); - return Ok(provider_metadata); -} - -#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)] -struct JobClaim { - job_id: String, - path: Option, - flow_path: Option, - groups: Vec, - username: String, - email: String, - workspace: String, -} - -use crate::db::ApiAuthed; -use crate::users::Tokened; - -#[cfg(feature = "enterprise")] -pub async fn gen_token( - authed: ApiAuthed, - token: Tokened, - Extension(db): Extension, - Path((w_id, audience)): Path<(String, String)>, -) -> windmill_common::error::Result { - use chrono::{Duration, Utc}; - use openidconnect::{ - core::{CoreGenderClaim, CoreJsonWebKeyType, CoreJweContentEncryptionAlgorithm}, - Audience, EndUserEmail, IdToken, IdTokenClaims, StandardClaims, SubjectIdentifier, - }; - use windmill_queue::get_queued_job; - - use crate::users::get_groups_for_user; - - let private_key = get_private_key(&db).await?; - - let username = authed.username; - let email = authed.email; - - let job_id = { - let job = sqlx::query_scalar!("SELECT job FROM token WHERE token = $1", token.token) - .fetch_optional(&db) - .await? - .flatten(); - if job.is_none() { - return Err(anyhow::anyhow!("Token not found").into()); - } else { - job.unwrap() - } - }; - let job = get_queued_job(job_id, &w_id, &db).await?; - - let job = job.ok_or_else(|| anyhow::anyhow!("Queued job {} not found", job_id))?; - let issue_url = format!("{}/api/oidc/", crate::BASE_URL.read().await.clone()); - let flow_path = if let Some(uuid) = job.parent_job { - sqlx::query_scalar!("SELECT script_path FROM queue WHERE id = $1", uuid) - .fetch_optional(&db) - .await? - .flatten() - } else { - None - }; - - let groups = get_groups_for_user(&w_id, &username, &email, &db) - .await - .ok() - .unwrap_or_default(); - - let id_token = IdToken::< - JobClaim, - CoreGenderClaim, - CoreJweContentEncryptionAlgorithm, - CoreJwsSigningAlgorithm, - CoreJsonWebKeyType, - >::new( - IdTokenClaims::::new( - // Specify the issuer URL for the OpenID Connect Provider. - IssuerUrl::new(issue_url) - .map_err(|e| anyhow::anyhow!("Failed to generate IssueUrl: {}", e))?, - // The audience is usually a single entry with the client ID of the client for whom - // the ID token is intended. This is a required claim. - vec![Audience::new(audience)], - // The ID token expiration is usually much shorter than that of the access or refresh - // tokens issued to clients. - Utc::now() + Duration::hours(48), - // The issue time is usually the current time. - Utc::now(), - // Set the standard claims defined by the OpenID Connect Core spec. - StandardClaims::new( - // Stable subject identifiers are recommended in place of e-mail addresses or other - // potentially unstable identifiers. This is the only required claim. - SubjectIdentifier::new(format!( - "{}::{}::{}::{}", - email, - job.script_path - .clone() - .unwrap_or_else(|| "no_path".to_string()), - flow_path.clone().unwrap_or_else(|| "no_flow".to_string()), - w_id - )), - ) - // Optional: specify the user's e-mail address. This should only be provided if the - // client has been granted the 'profile' or 'email' scopes. - .set_email(Some(EndUserEmail::new(job.email.clone()))) - // Optional: specify whether the provider has verified the user's e-mail address. - .set_email_verified(Some(true)), - // OpenID Connect Providers may supply custom claims by providing a struct that - // implements the AdditionalClaims trait. This requires manually using the - // generic IdTokenClaims struct rather than the CoreIdTokenClaims type alias, - // however. - JobClaim { - job_id: job_id.to_string(), - path: job.script_path, - flow_path, - username: job.created_by, - email: job.email, - workspace: job.workspace_id, - groups, - }, - ), - // The private key used for signing the ID token. For confidential clients (those able - // to maintain a client secret), a CoreHmacKey can also be used, in conjunction - // with one of the CoreJwsSigningAlgorithm::HmacSha* signing algorithms. When using an - // HMAC-based signing algorithm, the UTF-8 representation of the client secret should - // be used as the HMAC key. - &CoreRsaPrivateSigningKey::from_pem( - &private_key, - Some(JsonWebKeyId::new("windmill".to_string())), - ) - .map_err(|e| anyhow::anyhow!("Invalid private key: {}", e))?, - // Uses the RS256 signature algorithm. This crate supports any RS*, PS*, or HS* - // signature algorithm. - CoreJwsSigningAlgorithm::RsaSsaPkcs1V15Sha256, - // When returning the ID token alongside an access token (e.g., in the Authorization Code - // flow), it is recommended to pass the access token here to set the `at_hash` claim - // automatically. - None, - // When returning the ID token alongside an authorization code (e.g., in the implicit - // flow), it is recommended to pass the authorization code here to set the `c_hash` claim - // automatically. - None, - ) - .map_err(|e| anyhow::anyhow!("Failed to generate token: {}", e))?; - Ok(id_token.to_string()) -} diff --git a/backend/windmill-api/src/oidc_ee.rs b/backend/windmill-api/src/oidc_ee.rs new file mode 100644 index 0000000000..248b990f54 --- /dev/null +++ b/backend/windmill-api/src/oidc_ee.rs @@ -0,0 +1,17 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2023 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +use axum::Router; + +pub fn global_service() -> Router { + Router::new() +} + +pub fn workspaced_service() -> Router { + Router::new() +} diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 195c46a9d2..2b421dfe3b 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -812,20 +812,27 @@ async fn list_paths( pub struct ToggleWorkspaceErrorHandler { pub muted: Option, } + +#[cfg(not(feature = "enterprise"))] +async fn toggle_workspace_error_handler( + _authed: ApiAuthed, + Extension(_user_db): Extension, + Path((_w_id, _path)): Path<(String, StripPath)>, + Json(_req): Json, +) -> Result { + return Err(Error::BadRequest( + "Muting the error handler for certain script is only available in enterprise version" + .to_string(), + )); +} + +#[cfg(feature = "enterprise")] async fn toggle_workspace_error_handler( authed: ApiAuthed, Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, Json(req): Json, ) -> Result { - #[cfg(not(feature = "enterprise"))] - { - return Err(Error::BadRequest( - "Muting the error handler for certain script is only available in enterprise version" - .to_string(), - )); - } - let mut tx = user_db.begin(&authed).await?; let error_handler_maybe: Option = sqlx::query_scalar!( diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index ea3e33ecce..bf9429b705 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -881,10 +881,23 @@ async fn edit_large_file_storage_config( } #[derive(Deserialize)] -struct EditGitSyncConfig { - git_sync_settings: Option, +pub struct EditGitSyncConfig { + pub git_sync_settings: Option, } +#[cfg(not(feature = "enterprise"))] +async fn edit_git_sync_config( + _authed: ApiAuthed, + Extension(_db): Extension, + Path(_w_id): Path, + Json(_new_config): Json, +) -> Result { + return Err(Error::BadRequest( + "Git sync is only available on Windmill Enterprise Edition".to_string(), + )); +} + +#[cfg(feature = "enterprise")] async fn edit_git_sync_config( authed: ApiAuthed, Extension(db): Extension, @@ -892,13 +905,6 @@ async fn edit_git_sync_config( ApiAuthed { is_admin, username, .. }: ApiAuthed, Json(new_config): Json, ) -> Result { - #[cfg(not(feature = "enterprise"))] - { - return Err(Error::BadRequest( - "Git sync is only available on Windmill Enterprise Edition".to_string(), - )); - } - require_admin(is_admin, &username)?; let mut tx = db.begin().await?; @@ -940,10 +946,24 @@ async fn edit_git_sync_config( } #[derive(Deserialize)] -struct EditDefaultApp { - default_app_path: Option, +pub struct EditDefaultApp { + pub default_app_path: Option, } +#[cfg(not(feature = "enterprise"))] +async fn edit_default_app( + _authed: ApiAuthed, + Extension(_db): Extension, + Path(_w_id): Path, + Json(_new_config): Json, +) -> Result { + return Err(Error::BadRequest( + "Setting a workspace default app is only available on Windmill Enterprise Edition" + .to_string(), + )); +} + +#[cfg(feature = "enterprise")] async fn edit_default_app( authed: ApiAuthed, Extension(db): Extension, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index a25b8e11a1..1ad6fd2f9a 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2413,7 +2413,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection mut tag: Option, custom_timeout: Option, flow_step_id: Option, - priority_override: Option, + _priority_override: Option, ) -> Result<(Uuid, QueueTransaction<'c, R>), Error> { #[cfg(feature = "enterprise")] if *CLOUD_HOSTED { @@ -2527,7 +2527,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection concurrency_time_window_s, cache_ttl, dedicated_worker, - low_level_priority, + _low_level_priority, ) = match job_payload { JobPayload::ScriptHash { hash, @@ -2918,11 +2918,11 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection final_priority = if *CLOUD_HOSTED { // for cloud hosted instance, priority queues is disabled None - } else if priority_override.is_some() { - priority_override + } else if _priority_override.is_some() { + _priority_override } else { // else it takes the priority defined at the script/flow level, if it's a script or flow - low_level_priority + _low_level_priority }; // else it remains empty, i.e. no priority } diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index f0d11f817e..746ea9fe58 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -40,7 +40,10 @@ use std::{ use tracing::{trace_span, Instrument}; use uuid::Uuid; -use windmill_common::{job_metrics, variables, DB}; +use windmill_common::{variables, DB}; + +#[cfg(feature = "enterprise")] +use windmill_common::job_metrics; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, @@ -496,6 +499,8 @@ pub async fn handle_child( interval.set_missed_tick_behavior(MissedTickBehavior::Skip); let mut i = 0; + + #[cfg(feature = "enterprise")] let mut memory_metric_id: Result = Err(Error::NotFound("not yet initialized".to_string())); @@ -795,19 +800,19 @@ pub async fn start_child_process(mut cmd: Command, executable: &str) -> Result, - w_id: &str, - job_id: Uuid, + _db: &Pool, + _w_id: &str, + _job_id: Uuid, custom_timeout_secs: Option, ) -> (Duration, Option) { let mut warn_msg: Option = None; #[cfg(feature = "enterprise")] let cloud_premium_workspace = *CLOUD_HOSTED - && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", w_id) - .fetch_one(db) + && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id) + .fetch_one(_db) .await .map_err(|e| { - tracing::error!(%e, "error getting premium workspace for job {job_id}: {e}"); + tracing::error!(%e, "error getting premium workspace for job {_job_id}: {e}"); }) .unwrap_or(false); #[cfg(not(feature = "enterprise"))] @@ -1078,13 +1083,9 @@ pub async fn get_cached_resource_value_if_valid( } for (s3_file_key, s3_file_etag) in s3_etags { if let Some(object_store_resource) = object_store_resource_opt.clone() { - let etag = get_etag_or_empty( - &object_store_resource, - S3Object { - s3: s3_file_key.clone(), - }, - ) - .await; + let etag = + get_etag_or_empty(&object_store_resource, S3Object { s3: s3_file_key.clone() }) + .await; if etag.is_none() || etag.clone().unwrap() != s3_file_etag { tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag); return None; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 4e720e7734..2542c48b93 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -16,10 +16,13 @@ use windmill_common::{ error::{self, Error}, jobs::QueuedJob, utils::calculate_hash, - variables::get_secret_value_as_admin, worker::WORKER_CONFIG, DB, }; + +#[cfg(feature = "enterprise")] +use windmill_common::variables::get_secret_value_as_admin; + use windmill_queue::CanceledBy; lazy_static::lazy_static! { @@ -569,6 +572,7 @@ if args["{name}"] is None: )) } +#[cfg(feature = "enterprise")] async fn replace_pip_secret( db: &DB, w_id: &str, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 26cf657595..d9fe1a5579 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1822,14 +1822,33 @@ async fn spawn_dedicated_workers_for_flow( workers } -enum SpawnWorker { +pub enum SpawnWorker { Script { path: String, hash: Option }, RawScript { path: String, content: String, lock: Option, lang: ScriptLang }, } +#[cfg(not(feature = "enterprise"))] +async fn spawn_dedicated_worker( + _sw: SpawnWorker, + _w_id: &str, + killpill_tx: tokio::sync::broadcast::Sender<()>, + _killpill_rx: &tokio::sync::broadcast::Receiver<()>, + _db: &Pool, + _worker_dir: &str, + _base_internal_url: &str, + _worker_name: &str, + _job_completed_tx: &JobCompletedSender, + _node_id: Option, +) -> Option { + tracing::error!("Dedicated worker is an enterprise feature"); + killpill_tx.send(()).expect("send"); + return None; +} + // spawn one dedicated worker and return the key, the channel sender and the join handle // note that for it will return none for language that do not support dedicated workers // note that go using cache binary does not need dedicated workers so all languages are supported +#[cfg(feature = "enterprise")] async fn spawn_dedicated_worker( sw: SpawnWorker, w_id: &str, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 6c0606a620..294746762c 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -180,7 +180,7 @@ pub async fn update_flow_status_after_job_completion_internal< skip_if_stop_early, nresult, is_failure_step, - cleanup_module, + _cleanup_module, ) = { // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}"); @@ -590,16 +590,16 @@ pub async fn update_flow_status_after_job_completion_internal< #[cfg(feature = "enterprise")] if flow_job.parent_job.is_none() { // run the cleanup step only when the root job is complete - if !cleanup_module.flow_jobs_to_clean.is_empty() { + if !_cleanup_module.flow_jobs_to_clean.is_empty() { tracing::debug!( "Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}", - cleanup_module.flow_jobs_to_clean + _cleanup_module.flow_jobs_to_clean ); sqlx::query!( "UPDATE completed_job SET logs = '##DELETED##', args = '{}'::jsonb, result = '{}'::jsonb WHERE id = ANY($1)", - &cleanup_module.flow_jobs_to_clean, + &_cleanup_module.flow_jobs_to_clean, ) .execute(db) .await?;