From 2b8af5ddd48543bf785078a2972b2e54f3fff452 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 6 Feb 2026 13:56:30 +0000 Subject: [PATCH] with cc --- ...-is-being-continued-from-a-previous-co.txt | 1998 +++++++++++++++++ backend/windmill-api-auth/src/lib.rs | 2 + backend/windmill-api-auth/src/permissions.rs | 156 ++ backend/windmill-api-auth/src/scopes.rs | 32 +- backend/windmill-api-auth/src/tokens.rs | 97 + backend/windmill-api/src/args.rs | 678 ------ backend/windmill-api/src/capture.rs | 73 +- backend/windmill-api/src/folders.rs | 20 +- backend/windmill-api/src/job_ops_impl.rs | 48 +- backend/windmill-api/src/jobs.rs | 157 +- backend/windmill-api/src/lib.rs | 2 +- backend/windmill-api/src/users.rs | 215 +- backend/windmill-common/src/ee_oss.rs | 12 + backend/windmill-common/src/jobs.rs | 44 + backend/windmill-triggers/src/flow_ext.rs | 4 +- backend/windmill-triggers/src/jobs_ext.rs | 106 +- backend/windmill-triggers/src/lib.rs | 151 -- backend/windmill-triggers/src/script_ext.rs | 4 +- backend/windmill-triggers/src/user_ext.rs | 89 +- 19 files changed, 2411 insertions(+), 1477 deletions(-) create mode 100644 backend/2026-02-06-this-session-is-being-continued-from-a-previous-co.txt create mode 100644 backend/windmill-api-auth/src/permissions.rs create mode 100644 backend/windmill-api-auth/src/tokens.rs delete mode 100644 backend/windmill-api/src/args.rs diff --git a/backend/2026-02-06-this-session-is-being-continued-from-a-previous-co.txt b/backend/2026-02-06-this-session-is-being-continued-from-a-previous-co.txt new file mode 100644 index 0000000000..16ffc5d3ba --- /dev/null +++ b/backend/2026-02-06-this-session-is-being-continued-from-a-previous-co.txt @@ -0,0 +1,1998 @@ + + ▐▛███▜▌ Claude Code v2.1.34 +▝▜█████▛▘ Opus 4.6 · Claude Max + ▘▘ ▝▝ ~/windmill/backend + +✻ Conversation compacted (ctrl+o for history) + + +● Let me check the current state and continue with the Tier 1-4 deduplication + plan. + +● Bash(cd /home/rfiszel/windmill && git log --oneline -5) + ⎿  5aaecf0da3 refactor: complete trigger extraction by removing originals + from windmill-api + 9b165361d9 with ee ref + … +5 lines (ctrl+o to expand) + ⎿  Shell cwd was reset to /home/rfiszel/windmill/backend + +● Good, I'm on branch feat/extract-windmill-triggers-and-api-auth. Let me read + the key source files I need to modify, then implement all tiers. + +● Read 3 files (ctrl+o to expand) + +● Now let me read the windmill-api duplicates and the other files I need to + modify. + +● Searched for 4 patterns (ctrl+o to expand) + +● Let me find all usages of the duplicated functions across both crates to plan + the replacements. + +● Searched for 10 patterns (ctrl+o to expand) + +● Let me also check the impl RunJobQuery methods in windmill-api that I'll need + to handle. + +● Read 7 files (ctrl+o to expand) + +● Now let me find the windmill-api duplicates of permission helpers and where + they're used. + +● Searched for 4 patterns (ctrl+o to expand) + +● Let me check the windmill-api/src/capture.rs duplicate and where + NewToken/create_token_internal are used in windmill-api. + +● Searched for 3 patterns (ctrl+o to expand) + +● Let me read the windmill-api version of require_is_writer to confirm it's + identical to require_is_writer_internal. + +● Read 1 file (ctrl+o to expand) + +● Now let me check the folders.rs for require_is_owner and is_owner (the + equivalents of require_is_folder_owner and is_folder_owner in + windmill-triggers). + +● Searched for 5 patterns, read 2 files (ctrl+o to expand) + +● Now I have the complete picture. Let me also check the + windmill-api-auth/src/scopes.rs to see what's already there and the full + windmill-api-auth structure. + +● Searched for 2 patterns, read 3 files (ctrl+o to expand) + +● Now I have the full picture. Let me implement all 4 tiers systematically. + Starting with Tier 1. + + Tier 1: Move shared logic to windmill-common + + First, add RunJobQuery and delete_job_metadata_after_use to + windmill-common/src/jobs.rs, and check_license_key_valid to + windmill-common/src/ee_oss.rs. + +● Update(windmill-common/src/jobs.rs) + ⎿  Added 3 lines + 10 use tokio::io::AsyncReadExt; + 11 use uuid::Uuid; + 12 + 13 +use crate::error::{Error, Result}; + 14 +use crate::DB; + 15 + + 16 pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE"; + 17 pub const LARGE_LOG_THRESHOLD_SIZE: usize = 9000; + 18 + +● Wait, those imports already exist in the use crate::{...} block below. Let me + revert that. + +● Update(windmill-common/src/jobs.rs) + ⎿  Removed 3 lines + 10 use tokio::io::AsyncReadExt; + 11 use uuid::Uuid; + 12 + 13 -use crate::error::{Error, Result}; + 14 -use crate::DB; + 15 - + 13 pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE"; + 14 pub const LARGE_LOG_THRESHOLD_SIZE: usize = 9000; + 15 + +● Now add RunJobQuery and delete_job_metadata_after_use at the end of + windmill-common/src/jobs.rs. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-common/src/jobs.rs) + ⎿  Added 44 lines + 917 // The server cannot call the worker functions directly becaus + e they are independent crates + 918 pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell = + 919 OnceCell::new(); + 920 + + 921 +#[derive(Debug, Deserialize, Clone, Default)] + 922 +pub struct RunJobQuery { + 923 + pub scheduled_for: Option>, + 924 + pub scheduled_in_secs: Option, + 925 + pub parent_job: Option, + 926 + pub root_job: Option, + 927 + pub invisible_to_owner: Option, + 928 + pub queue_limit: Option, + 929 + pub payload: Option, + 930 + pub job_id: Option, + 931 + pub tag: Option, + 932 + pub timeout: Option, + 933 + pub cache_ttl: Option, + 934 + pub cache_ignore_s3_path: Option, + 935 + pub skip_preprocessor: Option, + 936 + pub poll_delay_ms: Option, + 937 + pub memory_id: Option, + 938 + pub trigger_external_id: Option, + 939 + pub service_name: Option, + 940 + pub suspended_mode: Option, + 941 +} + 942 + + 943 +pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: + +Uuid) -> Result<(), Error> { + 944 + sqlx::query!( + 945 + "UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1", + 946 + job_uuid, + 947 + ) + 948 + .execute(db) + 949 + .await?; + 950 + sqlx::query!( + 951 + "UPDATE v2_job_completed SET result = '{}'::jsonb WHER + +E id = $1", + 952 + job_uuid, + 953 + ) + 954 + .execute(db) + 955 + .await?; + 956 + sqlx::query!( + 957 + "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id + + = $1", + 958 + job_uuid, + 959 + ) + 960 + .execute(db) + 961 + .await?; + 962 + Ok(()) + 963 +} + +● Now add check_license_key_valid to windmill-common/src/ee_oss.rs. + +● Update(windmill-common/src/ee_oss.rs) + ⎿  Added 12 lines + 122 ) { + 123 // Implementation is not open source + 124 } + 125 + + 126 +#[cfg(feature = "enterprise")] + 127 +pub async fn check_license_key_valid() -> error::Result<()> { + 128 + let valid = *LICENSE_KEY_VALID.read().await; + 129 + if !valid { + 130 + return Err(error::Error::BadRequest( + 131 + "License key is not valid. Go to your superadmin s + +ettings to update your license key." + 132 + .to_string(), + 133 + )); + 134 + } + 135 + Ok(()) + 136 +} + +● Wait, the check_license_key_valid should be gated on just enterprise (not + !private) since the original code was gated only on enterprise and accessed + LICENSE_KEY_VALID which is available in both private and non-private builds. + Let me fix the cfg guard. + +● Update(windmill-common/src/ee_oss.rs) + ⎿  Added 1 line, removed 1 line + 123 // Implementation is not open source + 124 } + 125 + 126 -#[cfg(feature = "enterprise")] + 126 +#[cfg(all(feature = "enterprise", not(feature = "private")))] + 127 pub async fn check_license_key_valid() -> error::Result<()> { + 128 let valid = *LICENSE_KEY_VALID.read().await; + 129 if !valid { + +● Now update windmill-triggers/src/jobs_ext.rs: remove the duplicated + definitions and re-export from windmill_common. + +● Update(windmill-triggers/src/jobs_ext.rs) + ⎿  Added 6 lines, removed 59 lines + 13 //! implementation at startup via `set_ops()`. + 14 + 15 use axum::response::Response; + 16 -use serde::Deserialize; + 16 use serde_json::value::RawValue; + 17 use sqlx::Postgres; + 18 use std::sync::Arc; + ... + 20 use windmill_api_auth::{ApiAuthed, OptTokened}; + 21 use windmill_common::{ + 22 db::UserDB, + 24 - error::{self, Error}, + 23 + error, + 24 jobs::check_tag_available_for_workspace_internal, + 25 triggers::TriggerMetadata, + 26 utils::StripPath, + ... + 28 }; + 29 use windmill_queue::PushArgsOwned; + 30 + 32 -#[derive(Debug, Deserialize, Clone, Default)] + 33 -pub struct RunJobQuery { + 34 - pub scheduled_for: Option>, + 35 - pub scheduled_in_secs: Option, + 36 - pub parent_job: Option, + 37 - pub root_job: Option, + 38 - pub invisible_to_owner: Option, + 39 - pub queue_limit: Option, + 40 - pub payload: Option, + 41 - pub job_id: Option, + 42 - pub tag: Option, + 43 - pub timeout: Option, + 44 - pub cache_ttl: Option, + 45 - pub cache_ignore_s3_path: Option, + 46 - pub skip_preprocessor: Option, + 47 - pub poll_delay_ms: Option, + 48 - pub memory_id: Option, + 49 - pub trigger_external_id: Option, + 50 - pub service_name: Option, + 51 - pub suspended_mode: Option, + 52 -} + 31 +// Re-export shared types/functions from windmill-common + 32 +pub use windmill_common::jobs::{delete_job_metadata_after_use, + +RunJobQuery}; + 33 + 34 +#[cfg(feature = "enterprise")] + 35 +pub use windmill_common::ee_oss::check_license_key_valid; + 36 + + 37 pub fn get_scope_tags(authed: &ApiAuthed) -> Option> + { + 38 authed.scopes.as_ref()?.iter().find_map(|s| { + 39 if s.starts_with("if_jobs:filter_tags:") { + ... + 62 } + 63 } + 64 + 82 -pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: + -Uuid) -> Result<(), Error> { + 83 - sqlx::query!( + 84 - "UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1", + 85 - job_uuid, + 86 - ) + 87 - .execute(db) + 88 - .await?; + 89 - sqlx::query!( + 90 - "UPDATE v2_job_completed SET result = '{}'::jsonb WHER + -E id = $1", + 91 - job_uuid, + 92 - ) + 93 - .execute(db) + 94 - .await?; + 95 - sqlx::query!( + 96 - "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id + - = $1", + 97 - job_uuid, + 98 - ) + 99 - .execute(db) + 100 - .await?; + 101 - Ok(()) + 102 -} + 103 - + 104 -#[cfg(feature = "enterprise")] + 105 -pub async fn check_license_key_valid() -> error::Result<()> { + 106 - use windmill_common::ee_oss::LICENSE_KEY_VALID; + 107 - + 108 - let valid = *LICENSE_KEY_VALID.read().await; + 109 - if !valid { + 110 - return Err(Error::BadRequest( + 111 - "License key is not valid. Go to your superadmin s + -ettings to update your license key." + 112 - .to_string(), + 113 - )); + 114 - } + 115 - Ok(()) + 116 -} + 117 - + 65 /// Trait for complex job operations that stay in windmill-api + . + 66 /// windmill-api provides the implementation at startup via `s + et_ops()`. + 67 #[axum::async_trait] + +● Also remove the dead JobUpdate struct from jobs_ext.rs (Tier 4 item, doing it + now since we're editing this file). + +● Update(windmill-triggers/src/jobs_ext.rs) + ⎿  Removed 15 lines + 183 ) -> error::Result<(Option, Option)>; + 184 } + 185 + 186 -// Re-export types used by SSE streaming + 187 -#[derive(Debug, Clone, serde::Serialize)] + 188 -pub struct JobUpdate { + 189 - #[serde(skip_serializing_if = "Option::is_none")] + 190 - pub running: Option, + 191 - #[serde(skip_serializing_if = "Option::is_none")] + 192 - pub completed: Option, + 193 - #[serde(skip_serializing_if = "Option::is_none")] + 194 - pub new_logs: Option, + 195 - #[serde(skip_serializing_if = "Option::is_none")] + 196 - pub log_offset: Option, + 197 - #[serde(skip_serializing_if = "Option::is_none")] + 198 - pub mem_peak: Option, + 199 -} + 200 - + 186 #[derive(Debug, serde::Serialize)] + 187 pub enum JobUpdateSSEStream { + 188 Update(serde_json::Value), + +● Now update windmill-api/src/jobs.rs. First, replace the RunJobQuery struct + with an import and convert the impl methods to free functions. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/jobs.rs) + ⎿  Added 27 lines, removed 49 lines + 1775 pub args: Option, + 1776 } + 1777 + 1778 -#[derive(Debug, Deserialize, Clone, Default)] + 1779 -pub struct RunJobQuery { + 1780 - pub scheduled_for: Option>, + 1781 - pub scheduled_in_secs: Option, + 1782 - pub parent_job: Option, + 1783 - pub root_job: Option, + 1784 - pub invisible_to_owner: Option, + 1785 - pub queue_limit: Option, + 1786 - pub payload: Option, + 1787 - pub job_id: Option, + 1788 - pub tag: Option, + 1789 - pub timeout: Option, + 1790 - pub cache_ttl: Option, + 1791 - pub cache_ignore_s3_path: Option, + 1792 - pub skip_preprocessor: Option, + 1793 - pub poll_delay_ms: Option, + 1794 - pub memory_id: Option, + 1795 - pub trigger_external_id: Option, + 1796 - pub service_name: Option, + 1797 - pub suspended_mode: Option, + 1798 -} + 1778 +pub use windmill_common::jobs::RunJobQuery; + 1779 + 1800 -impl RunJobQuery { + 1801 - async fn get_scheduled_for<'c>( + 1802 - &self, + 1803 - db: &DB, + 1804 - ) -> error::Result>> + - { + 1805 - if let Some(scheduled_for) = self.scheduled_for { + 1806 - Ok(Some(scheduled_for)) + 1807 - } else if let Some(scheduled_in_secs) = self.schedule + -d_in_secs { + 1808 - let now = now_from_db(db).await?; + 1809 - Ok(Some( + 1810 - now + chrono::Duration::try_seconds(scheduled + -_in_secs).unwrap_or_default(), + 1811 - )) + 1812 - } else { + 1813 - Ok(None) + 1814 - } + 1780 +async fn run_query_get_scheduled_for( + 1781 + run_query: &RunJobQuery, + 1782 + db: &DB, + 1783 +) -> error::Result>> { + 1784 + if let Some(scheduled_for) = run_query.scheduled_for { + 1785 + Ok(Some(scheduled_for)) + 1786 + } else if let Some(scheduled_in_secs) = run_query.schedul + +ed_in_secs { + 1787 + let now = now_from_db(db).await?; + 1788 + Ok(Some( + 1789 + now + chrono::Duration::try_seconds(scheduled_in_ + +secs).unwrap_or_default(), + 1790 + )) + 1791 + } else { + 1792 + Ok(None) + 1793 } + 1794 +} + 1795 + 1817 - fn payload_as_args(&self) -> error::Result>> { + 1818 - let payload_r = self.payload.clone().map(decode_paylo + -ad).map(|x| { + 1819 - x.map_err(|e| { + 1820 - error::Error::internal_err(format!("Impossibl + -e to decode query payload: {e:#?}")) + 1821 - }) + 1822 - }); + 1796 +fn run_query_payload_as_args(run_query: &RunJobQuery) -> erro + +r::Result>> { + 1797 + let payload_r = run_query.payload.clone().map(decode_payl + +oad).map(|x| { + 1798 + x.map_err(|e| { + 1799 + error::Error::internal_err(format!("Impossible to + + decode query payload: {e:#?}")) + 1800 + }) + 1801 + }); + 1802 + 1824 - let payload_as_args = if let Some(payload) = payload_ + -r { + 1825 - payload? + 1826 - } else { + 1827 - HashMap::new() + 1828 - }; + 1803 + let payload_as_args = if let Some(payload) = payload_r { + 1804 + payload? + 1805 + } else { + 1806 + HashMap::new() + 1807 + }; + 1808 + 1830 - Ok(payload_as_args) + 1831 - } + 1809 + Ok(payload_as_args) + 1810 } + 1811 + 1812 #[derive(Deserialize, Clone)] + +● Now update all call sites for get_scheduled_for and payload_as_args in + jobs.rs. + +● Update(windmill-api/src/jobs.rs) + ⎿  Added 9 lines, removed 9 lines + 4301 let tag = run_query.tag.clone().or(tag); + 4302 + 4303 check_tag_available_for_workspace(&db, &w_id, &tag, &auth + ed).await?; + 4304 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 4304 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 4305 + 4306 let return_tx = tx_o.is_some(); + 4307 + ... + 4605 .map(|json| PushArgs { args: &json.0, extra: None }) + 4606 .unwrap_or_else(|| PushArgs::from(&ehm)); + 4607 + 4608 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 4608 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 4609 + 4610 let tx = PushIsolationLevel::Isolated(user_db, authed.clo + ne().into()); + 4611 + ... + 4753 run_query.skip_preprocessor, + 4754 ) + 4755 .await?; + 4756 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 4756 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 4757 + 4758 let tag = run_query.tag.clone().or(tag); + 4759 check_tag_available_for_workspace(&db, &w_id, &tag, &auth + ed).await?; + ... + 4931 extra.insert(ENTRYPOINT_OVERRIDE.to_string(), to_raw_valu + e(&entrypoint)); + 4932 + 4933 let args = PushArgs { args: &task.args.unwrap_or_else(Has + hMap::new), extra: Some(extra) }; + 4934 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 4934 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 4935 + 4936 let tag = run_query.tag.clone().or(tag).or(Some(job.tag)) + ; + 4937 + ... + 6221 )); + 6222 } + 6223 require_path_read_access_for_preview(&authed, &preview.pa + th)?; + 6224 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 6224 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 6225 let tag = run_query.tag.clone().or(preview.tag.clone()); + 6226 check_tag_available_for_workspace(&db, &w_id, &tag, &auth + ed).await?; + 6227 let tx = PushIsolationLevel::Isolated(user_db.clone(), au + thed.clone().into()); + ... + 6434 .and_then(|s| BundleFormat::from_string(&s)) + 6435 .unwrap_or(BundleFormat::Cjs); + 6436 + 6437 - let scheduled_for = run_query.get_scheduled_for(& + -db).await?; + 6437 + let scheduled_for = run_query_get_scheduled_for(& + +run_query, &db).await?; + 6438 let tag = run_query.tag.clone().or(preview.tag.cl + one()); + 6439 check_tag_available_for_workspace(&db, &w_id, &ta + g, &authed).await?; + 6440 let ltx = PushIsolationLevel::Isolated(user_db.cl + one(), authed.clone().into()); + ... + 7041 )); + 7042 } + 7043 require_path_read_access_for_preview(&authed, &raw_flow.p + ath)?; + 7044 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 7044 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 7045 let tag = run_query.tag.clone().or(raw_flow.tag.clone()); + 7046 check_tag_available_for_workspace(&db, &w_id, &tag, &auth + ed).await?; + 7047 let tx = PushIsolationLevel::Isolated(user_db.clone(), au + thed.clone().into()); + ... + 7241 } + 7242 } + 7243 + 7244 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 7244 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 7245 let tx = PushIsolationLevel::Isolated(user_db.clone(), au + thed.clone().into()); + 7246 + 7247 let (uuid, tx) = push( + ... + 7365 cache_ttl = Some(run_query_cache_ttl); + 7366 cache_ignore_s3_path = run_query.cache_ignore_s3_path + ; + 7367 } + 7368 - let scheduled_for = run_query.get_scheduled_for(&db).awai + -t?; + 7368 + let scheduled_for = run_query_get_scheduled_for(&run_quer + +y, &db).await?; + 7369 let tag = run_query.tag.clone().or(tag); + 7370 + 7371 check_tag_available_for_workspace(&db, &w_id, &tag, &auth + ed).await?; + +● Update(windmill-api/src/jobs.rs) + ⎿  Added 4 lines, removed 4 lines + 5453 return Ok(Json(serde_json::json!("")).into_response() + ); + 5454 } + 5455 + 5456 - let payload_as_args = run_query.payload_as_args()?; + 5456 + let payload_as_args = run_query_payload_as_args(&run_quer + +y)?; + 5457 + 5458 let mut args = args.process_args(&authed, &db, &w_id, Non + e).await?; + 5459 args.body = args::Body::HashMap(payload_as_args); + ... + 5558 if method == http::Method::HEAD { + 5559 return Ok(Json(serde_json::json!("")).into_response() + ); + 5560 } + 5561 - let payload_as_args = run_query.payload_as_args()?; + 5561 + let payload_as_args = run_query_payload_as_args(&run_quer + +y)?; + 5562 + 5563 let mut args = args.process_args(&authed, &db, &w_id, Non + e).await?; + 5564 args.body = args::Body::HashMap(payload_as_args); + ... + 5936 is_get: bool, + 5937 ) -> error::Result { + 5938 let args = if is_get { + 5939 - let payload_as_args = run_query.payload_as_args()?; + 5939 + let payload_as_args = run_query_payload_as_args(&run_ + +query)?; + 5940 + 5941 let mut args = args.process_args(&authed, &db, &w_id, + None).await?; + 5942 args.body = args::Body::HashMap(payload_as_args); + ... + 6118 return Ok(Json(serde_json::json!("")).into_response() + ); + 6119 } + 6120 + 6121 - let payload_as_args = run_query.payload_as_args()?; + 6121 + let payload_as_args = run_query_payload_as_args(&run_quer + +y)?; + 6122 + 6123 let mut args = args.process_args(&authed, &db, &w_id, Non + e).await?; + 6124 args.body = args::Body::HashMap(payload_as_args); + +● Now replace delete_job_metadata_after_use and check_license_key_valid in + windmill-api/src/jobs.rs with imports from windmill_common. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/jobs.rs) + ⎿  Added 1 line, removed 21 lines + 5302 result_to_response(result, success) + 5303 } + 5304 + 5305 -pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: + - Uuid) -> Result<(), Error> { + 5306 - sqlx::query!( + 5307 - "UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1", + 5308 - job_uuid, + 5309 - ) + 5310 - .execute(db) + 5311 - .await?; + 5312 - sqlx::query!( + 5313 - "UPDATE v2_job_completed SET result = '{}'::jsonb WHE + -RE id = $1", + 5314 - job_uuid, + 5315 - ) + 5316 - .execute(db) + 5317 - .await?; + 5318 - sqlx::query!( + 5319 - "UPDATE job_logs SET logs = '##DELETED##' WHERE job_i + -d = $1", + 5320 - job_uuid, + 5321 - ) + 5322 - .execute(db) + 5323 - .await?; + 5324 - Ok(()) + 5325 -} + 5305 +pub use windmill_common::jobs::delete_job_metadata_after_use; + 5306 + 5307 pub async fn check_queue_too_long(db: &DB, queue_limit: Optio + n) -> error::Result<()> { + 5308 if let Some(limit) = queue_limit { + +● Now replace check_license_key_valid and check_tag_available_for_workspace in + jobs.rs. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/jobs.rs) + ⎿  Added 2 lines, removed 25 lines + 3847 return args; + 3848 } + 3849 + 3850 -pub async fn check_tag_available_for_workspace( + 3851 - db: &DB, + 3852 - w_id: &str, + 3853 - tag: &Option, + 3854 - authed: &ApiAuthed, + 3855 -) -> error::Result<()> { + 3856 - if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty( + -)) { + 3857 - let tags = get_scope_tags(authed); + 3858 - check_tag_available_for_workspace_internal(&db, w_id, + - tag, &authed.email, tags).await + 3859 - } else { + 3860 - Ok(()) + 3861 - } + 3862 -} + 3850 +pub use windmill_triggers::jobs_ext::check_tag_available_for_ + +workspace; + 3851 + 3852 #[cfg(feature = "enterprise")] + 3865 -pub async fn check_license_key_valid() -> error::Result<()> { + 3866 - use windmill_common::ee_oss::LICENSE_KEY_VALID; + 3853 +pub use windmill_common::ee_oss::check_license_key_valid; + 3854 + 3868 - let valid = *LICENSE_KEY_VALID.read().await; + 3869 - if !valid { + 3870 - return Err(Error::BadRequest( + 3871 - "License key is not valid. Go to your superadmin + -settings to update your license key." + 3872 - .to_string(), + 3873 - )); + 3874 - } + 3875 - Ok(()) + 3876 -} + 3877 - + 3855 use windmill_common::flows::InputTransform; + 3856 + 3857 #[derive(Deserialize)] + +● Now update the job_ops_impl.rs to remove the RunJobQuery field-by-field + conversion since both crates now use the same type. + +● Read 1 file (ctrl+o to expand) + +● Now simplify job_ops_impl.rs by removing the RunJobQuery conversion (both + types are now the same windmill_common::jobs::RunJobQuery): + +● Update(windmill-api/src/job_ops_impl.rs) + ⎿  Added 4 lines, removed 44 lines + 23 utils::StripPath, + 24 DB, + 25 }; + 26 +use windmill_common::jobs::RunJobQuery; + 27 use windmill_queue::PushArgsOwned; + 27 -use windmill_triggers::jobs_ext::{JobOps, JobUpdateSSEStream, R + -unJobQuery}; + 28 +use windmill_triggers::jobs_ext::{JobOps, JobUpdateSSEStream}; + 29 + 30 pub struct JobOpsImpl; + 31 + ... + 47 Option, + 48 Option>, + 49 )> { + 49 - // Convert RunJobQuery from windmill-triggers to the wi + -ndmill-api version + 50 - let api_run_query = crate::jobs::RunJobQuery { + 51 - scheduled_for: run_query.scheduled_for, + 52 - scheduled_in_secs: run_query.scheduled_in_secs, + 53 - parent_job: run_query.parent_job, + 54 - root_job: run_query.root_job, + 55 - invisible_to_owner: run_query.invisible_to_owner, + 56 - queue_limit: run_query.queue_limit, + 57 - payload: run_query.payload, + 58 - job_id: run_query.job_id, + 59 - tag: run_query.tag, + 60 - timeout: run_query.timeout, + 61 - cache_ttl: run_query.cache_ttl, + 62 - cache_ignore_s3_path: run_query.cache_ignore_s3_pat + -h, + 63 - skip_preprocessor: run_query.skip_preprocessor, + 64 - poll_delay_ms: run_query.poll_delay_ms, + 65 - memory_id: run_query.memory_id, + 66 - trigger_external_id: run_query.trigger_external_id, + 67 - service_name: run_query.service_name, + 68 - suspended_mode: run_query.suspended_mode, + 69 - }; + 50 crate::jobs::push_script_job_by_path_into_queue( + 71 - authed, db, tx_o, user_db, w_id, script_path, api_r + -un_query, args, trigger, + 51 + authed, db, tx_o, user_db, w_id, script_path, run_q + +uery, args, trigger, + 52 ) + 53 .await + 54 } + ... + 69 Option, + 70 Option>, + 71 )> { + 92 - let api_run_query = crate::jobs::RunJobQuery { + 93 - scheduled_for: run_query.scheduled_for, + 94 - scheduled_in_secs: run_query.scheduled_in_secs, + 95 - parent_job: run_query.parent_job, + 96 - root_job: run_query.root_job, + 97 - invisible_to_owner: run_query.invisible_to_owner, + 98 - queue_limit: run_query.queue_limit, + 99 - payload: run_query.payload, + 100 - job_id: run_query.job_id, + 101 - tag: run_query.tag, + 102 - timeout: run_query.timeout, + 103 - cache_ttl: run_query.cache_ttl, + 104 - cache_ignore_s3_path: run_query.cache_ignore_s3_pa + -th, + 105 - skip_preprocessor: run_query.skip_preprocessor, + 106 - poll_delay_ms: run_query.poll_delay_ms, + 107 - memory_id: run_query.memory_id, + 108 - trigger_external_id: run_query.trigger_external_id + -, + 109 - service_name: run_query.service_name, + 110 - suspended_mode: run_query.suspended_mode, + 111 - }; + 72 crate::jobs::push_flow_job_by_path_into_queue( + 113 - authed, db, tx_o, user_db, w_id, flow_path, api_ru + -n_query, args, trigger, + 73 + authed, db, tx_o, user_db, w_id, flow_path, run_qu + +ery, args, trigger, + 74 ) + 75 .await + 76 } + +● Now make insert_capture_payload single-sourced in windmill-triggers. Replace + the windmill-api/src/capture.rs duplicate with an import. + +● Searched for 1 pattern, read 1 file (ctrl+o to expand) + +● clear_captures_history is only used within insert_capture_payload. Both are + duplicated in capture_ext.rs. I can replace both with a re-export from + windmill-triggers. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/capture.rs) + ⎿  Added 1 line, removed 67 lines + 802 )) + 803 } + 804 + 805 -async fn clear_captures_history(db: &DB, w_id: &str) -> Result + -<()> { + 806 - if *CLOUD_HOSTED { + 807 - /* Retain only KEEP_LAST most recent captures in this + -workspace. */ + 808 - sqlx::query!( + 809 - r#" + 810 - DELETE FROM + 811 - capture + 812 - WHERE + 813 - workspace_id = $1 + 814 - AND created_at <= ( + 815 - SELECT + 816 - created_at + 817 - FROM + 818 - capture + 819 - WHERE + 820 - workspace_id = $1 + 821 - ORDER BY + 822 - created_at DESC + 823 - OFFSET $2 + 824 - LIMIT 1 + 825 - ) + 826 - "#, + 827 - &w_id, + 828 - KEEP_LAST, + 829 - ) + 830 - .execute(db) + 831 - .await?; + 832 - } + 833 - Ok(()) + 834 -} + 805 +pub use windmill_triggers::capture_ext::insert_capture_payload + +; + 806 + 836 -pub async fn insert_capture_payload( + 837 - db: &DB, + 838 - w_id: &str, + 839 - path: &str, + 840 - is_flow: bool, + 841 - trigger_kind: &TriggerKind, + 842 - main_args: PushArgsOwned, + 843 - preprocessor_args: PushArgsOwned, + 844 - owner: &str, + 845 -) -> Result<()> { + 846 - sqlx::query!( + 847 - r#" + 848 - INSERT INTO + 849 - capture ( + 850 - workspace_id, path, is_flow, trigger_kind, main_ar + -gs, preprocessor_args, created_by + 851 - ) + 852 - VALUES ( + 853 - $1, $2, $3, $4, $5, $6, $7 + 854 - ) + 855 - "#, + 856 - &w_id, + 857 - path, + 858 - is_flow, + 859 - trigger_kind as &TriggerKind, + 860 - SqlxJson(PushArgs { args: &main_args.args, extra: main + -_args.extra }) as SqlxJson, + 861 - SqlxJson(PushArgs { args: &preprocessor_args.args, ext + -ra: preprocessor_args.extra }) + 862 - as SqlxJson, + 863 - owner, + 864 - ) + 865 - .execute(db) + 866 - .await?; + 867 - + 868 - clear_captures_history(db, &w_id).await?; + 869 - + 870 - Ok(()) + 871 -} + 872 - + 807 async fn webhook_payload( + 808 Extension(db): Extension, + 809 Path((w_id, runnable_kind, path)): Path<(String, RunnableK + ind, StripPath)>, + +● Good. Now let me check if there are unused imports in capture.rs that were + only used by the deleted functions. + +● Searched for 4 patterns (ctrl+o to expand) + +● to_raw_value is used elsewhere, just CLOUD_HOSTED is dead. Let me clean up the + dead code. + +● Update(windmill-api/src/capture.rs) + ⎿  Added 1 line, removed 1 line + 82 error::{JsonResult, Result}, + 83 triggers::{RunnableFormat, RunnableFormatVersion, TriggerKi + nd}, + 84 utils::{not_found_if_none, paginate, Pagination, RunnableKi + nd, StripPath}, + 85 - worker::{to_raw_value, CLOUD_HOSTED}, + 85 + worker::to_raw_value, + 86 }; + 87 + 88 use windmill_queue::{PushArgs, PushArgsOwned}; + +● Searched for 1 pattern (ctrl+o to expand) + +● Update(windmill-api/src/capture.rs) + ⎿  Added 1 line, removed 1 line + 87 + 88 use windmill_queue::{PushArgs, PushArgsOwned}; + 89 + 90 -const KEEP_LAST: i64 = 20; + 90 + 91 + + 92 pub fn workspaced_service() -> Router { + 93 Router::new() + 94 .route("/set_config", post(set_config)) + +● Now let me also check if PushArgs, PushArgsOwned, TriggerKind, SqlxJson + imports are still needed in capture.rs. + +● Searched for 1 pattern (ctrl+o to expand) + +● Still used elsewhere. Good. Now let me try a quick build to check for obvious + issues before continuing to Tier 2. + +● Bash(cargo check -p windmill-common --no-default-features 2>&1 | tail -20) + ⎿   Checking hstr v0.2.17 + Checking swc_allocator v0.1.10 + Checking swc_atoms v0.6.7 + … +20 lines (ctrl+o to expand) + +● windmill-common compiles. Now check windmill-triggers. + +● Bash(cargo check -p windmill-triggers --no-default-features 2>&1 | tail -20) + ⎿ Checking windmill-common v1.628.2 (/home/rfiszel/windmill/backend/w + indmill-common) + Checking windmill-audit v1.628.2 (/home/rfiszel/windmill/backend/wi + … +10 lines (ctrl+o to expand) + + + +● Now check windmill-api. + +● Bash(cargo check -p windmill-api --no-default-features 2>&1 | tail -40) + ⎿ Checking windmill-common v1.628.2 (/home/rfiszel/windmill/backend/w + indmill-common) + Checking windmill-audit v1.628.2 (/home/rfiszel/windmill/backend/wi + … +37 lines (ctrl+o to expand) + + + +● Only 2 warnings about unused imports. Let me fix those. + +● Update(windmill-api/src/capture.rs) + ⎿  Added 1 line, removed 1 line + 85 worker::to_raw_value, + 86 }; + 87 + 88 -use windmill_queue::{PushArgs, PushArgsOwned}; + 88 +use windmill_queue::PushArgsOwned; + 89 + 90 + 91 + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/jobs.rs) + ⎿  Added 2 lines, removed 2 lines + 35 use windmill_common::flow_conversations::add_message_to_convers + ation_tx; + 36 use windmill_common::flow_status::{JobResult, RestartedFrom}; + 37 use windmill_common::jobs::{ + 38 - check_tag_available_for_workspace_internal, format_complete + -d_job_result, format_result, + 39 - DynamicInput, JobTriggerKind, RunInlinePreviewScriptFnParam + -s, ENTRYPOINT_OVERRIDE, + 38 + format_completed_job_result, format_result, DynamicInput, J + +obTriggerKind, + 39 + RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE, + 40 }; + 41 use windmill_common::runnable_settings::{ + 42 ConcurrencySettings, ConcurrencySettingsWithCustom, Debounc + ingSettings, RunnableSettings, + +● Tier 1 is done. Let me update the task and move to Tier 2. + +● Tier 2: Move permission/token/scope helpers to windmill-api-auth + + I need to: + 1. Create windmill-api-auth/src/permissions.rs with permission helpers + 2. Create windmill-api-auth/src/tokens.rs with NewToken + + create_token_internal + 3. Add get_scope_tags + check_tag_available_for_workspace to + windmill-api-auth/src/scopes.rs + 4. Update windmill-triggers/src/lib.rs to remove duplicated permission helpers + and import from windmill-api-auth + 5. Update windmill-triggers/src/user_ext.rs (to be deleted after moving to + windmill-api-auth) + 6. Update windmill-api callers + + First, let me create the permissions module. + +● Write(windmill-api-auth/src/permissions.rs) + ⎿  Wrote 157 lines to windmill-api-auth/src/permissions.rs + 1 /* + 2 * Author: Windmill Labs, Inc + 3 * Copyright: Windmill Labs, Inc 2024 + 4 * This file and its contents are licensed under the AGPLv3 Lice + nse. + 5 * Please see the included NOTICE for copyright information and + 6 * LICENSE-AGPL for a copy of the license. + 7 */ + 8 + 9 use windmill_common::error::{Error, Result}; + 10 use windmill_common::DB; + … +147 lines (ctrl+o to expand) + +● Now create the tokens module. + +● Write(windmill-api-auth/src/tokens.rs) + ⎿  Wrote 98 lines to windmill-api-auth/src/tokens.rs + 1 /* + 2 * Author: Windmill Labs, Inc + 3 * Copyright: Windmill Labs, Inc 2024 + 4 * This file and its contents are licensed under the AGPLv3 Lice + nse. + 5 * Please see the included NOTICE for copyright information and + 6 * LICENSE-AGPL for a copy of the license. + 7 */ + 8 + 9 use sqlx::PgConnection; + 10 use tracing::Instrument; + … +88 lines (ctrl+o to expand) + +● Now add get_scope_tags and check_tag_available_for_workspace to scopes.rs. Let + me read the end of scopes.rs first. + +● Read 2 files (ctrl+o to expand) + +● Update(windmill-api-auth/src/scopes.rs) + ⎿  Added 3 lines, removed 1 line + 9 use itertools::Itertools; + 10 use serde::{Deserialize, Serialize}; + 11 use std::collections::HashSet; + 12 -use windmill_common::error::{Error, Result}; + 12 +use windmill_common::error::{self, Error, Result}; + 13 +use windmill_common::jobs::check_tag_available_for_workspace_in + +ternal; + 14 +use windmill_common::DB; + 15 + 16 use crate::ApiAuthed; + 17 + +● Now add the functions at the end of the file (before the test module). + +● Read 1 file (ctrl+o to expand) + +● I need to find the right place to add the functions - before the #[cfg(test)] + module. + +● Searched for 1 pattern, read 1 file (ctrl+o to expand) + +● Update(windmill-api-auth/src/scopes.rs) + ⎿  Added 28 lines + 704 Ok(()) + 705 } + 706 + 707 +pub fn get_scope_tags(authed: &ApiAuthed) -> Option> + + { + 708 + authed.scopes.as_ref()?.iter().find_map(|s| { + 709 + if s.starts_with("if_jobs:filter_tags:") { + 710 + Some( + 711 + s.trim_start_matches("if_jobs:filter_tags:") + 712 + .split(",") + 713 + .collect::>(), + 714 + ) + 715 + } else { + 716 + None + 717 + } + 718 + }) + 719 +} + 720 + + 721 +pub async fn check_tag_available_for_workspace( + 722 + db: &DB, + 723 + w_id: &str, + 724 + tag: &Option, + 725 + authed: &ApiAuthed, + 726 +) -> error::Result<()> { + 727 + if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty() + +) { + 728 + let tags = get_scope_tags(authed); + 729 + check_tag_available_for_workspace_internal(db, w_id, t + +ag, &authed.email, tags).await + 730 + } else { + 731 + Ok(()) + 732 + } + 733 +} + 734 + + 735 #[cfg(test)] + 736 mod tests { + 737 use super::*; + +● Now update windmill-api-auth/src/lib.rs to export the new modules. + +● Update(windmill-api-auth/src/lib.rs) + ⎿  Added 2 lines + 7 */ + 8 + 9 mod auth; + 10 +pub mod permissions; + 11 pub mod scopes; + 12 +pub mod tokens; + 13 mod types; + 14 + 15 pub use auth::*; + +● Now update windmill-triggers/src/lib.rs to remove the duplicated permission + helpers and import from windmill-api-auth. + +● Update(windmill-triggers/src/lib.rs) + ⎿  Removed 151 lines + 17 pub mod resource_ext; + 18 pub mod script_ext; + 19 pub mod user_ext; + 20 - + 21 -// Shared permission helpers used by flow_ext, script_ext, and + - native_triggers + 22 - + 23 -use windmill_api_auth::ApiAuthed; + 24 -use windmill_common::error::{Error, Result}; + 25 -use windmill_common::DB; + 26 - + 27 -/// Check if the user is an owner of the given path. + 28 -pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) - + -> Result<()> { + 29 - if authed.is_admin { + 30 - return Ok(()); + 31 - } + 32 - if !path.is_empty() { + 33 - let splitted = path.split("/").collect::>(); + 34 - if splitted[0] == "u" { + 35 - if splitted[1] == authed.username { + 36 - Ok(()) + 37 - } else { + 38 - Err(Error::BadRequest(format!( + 39 - "only the owner {} is authorized to perfor + -m this operation", + 40 - splitted[1] + 41 - ))) + 42 - } + 43 - } else if splitted[0] == "f" { + 44 - require_is_folder_owner(authed, splitted[1]) + 45 - } else { + 46 - Err(Error::BadRequest(format!( + 47 - "Not recognized path kind: {}", + 48 - path + 49 - ))) + 50 - } + 51 - } else { + 52 - Err(Error::BadRequest(format!( + 53 - "Cannot be owner of an empty path" + 54 - ))) + 55 - } + 56 -} + 57 - + 58 -fn is_folder_owner( + 59 - ApiAuthed { is_admin, folders, .. }: &ApiAuthed, + 60 - name: &str, + 61 -) -> bool { + 62 - if *is_admin { + 63 - true + 64 - } else { + 65 - folders.into_iter().any(|x| x.0 == name && x.2) + 66 - } + 67 -} + 68 - + 69 -fn require_is_folder_owner(authed: &ApiAuthed, name: &str) -> + -Result<()> { + 70 - if is_folder_owner(authed, name) { + 71 - Ok(()) + 72 - } else { + 73 - Err(Error::NotAuthorized(format!( + 74 - "You are not owner of the folder {}", + 75 - name + 76 - ))) + 77 - } + 78 -} + 79 - + 80 -fn get_perm_in_extra_perms_for_authed( + 81 - v: serde_json::Value, + 82 - authed: &ApiAuthed, + 83 -) -> Option { + 84 - match v { + 85 - serde_json::Value::Object(obj) => { + 86 - let mut keys = vec![format!("u/{}", authed.usernam + -e)]; + 87 - for g in authed.groups.iter() { + 88 - keys.push(format!("g/{}", g)); + 89 - } + 90 - let mut res = None; + 91 - for k in keys { + 92 - if let Some(v) = obj.get(&k) { + 93 - if let Some(v) = v.as_bool() { + 94 - if v { + 95 - return Some(true); + 96 - } + 97 - res = Some(v); + 98 - } + 99 - } + 100 - } + 101 - res + 102 - } + 103 - _ => None, + 104 - } + 105 -} + 106 - + 107 -/// Generic require_is_writer with a configurable SQL query. + 108 -/// Used by flow_ext and script_ext. + 109 -pub async fn require_is_writer_internal( + 110 - authed: &ApiAuthed, + 111 - path: &str, + 112 - w_id: &str, + 113 - db: DB, + 114 - query: &str, + 115 - kind: &str, + 116 -) -> Result<()> { + 117 - if authed.is_admin { + 118 - return Ok(()); + 119 - } + 120 - if !path.is_empty() { + 121 - if require_owner_of_path(authed, path).is_ok() { + 122 - return Ok(()); + 123 - } + 124 - if path.starts_with("f/") && path.split('/').count() > + -= 2 { + 125 - let folder = path.split('/').nth(1).unwrap(); + 126 - let extra_perms = sqlx::query_scalar!( + 127 - "SELECT extra_perms FROM folder WHERE name = $ + -1 AND workspace_id = $2", + 128 - folder, + 129 - w_id + 130 - ) + 131 - .fetch_optional(&db) + 132 - .await?; + 133 - if let Some(perms) = extra_perms { + 134 - let is_folder_writer = + 135 - get_perm_in_extra_perms_for_authed(perms, + -authed).unwrap_or(false); + 136 - if is_folder_writer { + 137 - return Ok(()); + 138 - } + 139 - } + 140 - } + 141 - let extra_perms = sqlx::query_scalar(query) + 142 - .bind(path) + 143 - .bind(w_id) + 144 - .fetch_optional(&db) + 145 - .await?; + 146 - if let Some(perms) = extra_perms { + 147 - let perm = get_perm_in_extra_perms_for_authed(perm + -s, authed); + 148 - match perm { + 149 - Some(true) => Ok(()), + 150 - Some(false) => Err(Error::BadRequest(format!( + 151 - "User {} is not a writer of {kind} path {p + -ath}", + 152 - authed.username + 153 - ))), + 154 - None => Err(Error::BadRequest(format!( + 155 - "User {} has neither read or write permiss + -ion on {kind} {path}", + 156 - authed.username + 157 - ))), + 158 - } + 159 - } else { + 160 - Err(Error::BadRequest(format!( + 161 - "{path} does not exist yet and user {} is not + -an owner of the parent folder", + 162 - authed.username + 163 - ))) + 164 - } + 165 - } else { + 166 - Err(Error::BadRequest(format!( + 167 - "Cannot be writer of an empty path" + 168 - ))) + 169 - } + 170 -} + +● Now update flow_ext.rs and script_ext.rs to use windmill_api_auth::permissions + instead of crate::require_is_writer_internal. + +● Write(windmill-triggers/src/flow_ext.rs) + ⎿  Added 1 line, removed 3 lines + 9 use windmill_api_auth::ApiAuthed; + 10 use windmill_common::{error::Result, DB}; + 11 + 12 -use crate::require_is_writer_internal; + 13 - + 12 pub async fn require_is_writer(authed: &ApiAuthed, path: &str, + w_id: &str, db: DB) -> Result<()> { + 15 - require_is_writer_internal( + 13 + windmill_api_auth::permissions::require_is_writer( + 14 authed, + 15 path, + 16 w_id, + +● Write(windmill-triggers/src/script_ext.rs) + ⎿  Added 1 line, removed 3 lines + 9 use windmill_api_auth::ApiAuthed; + 10 use windmill_common::{error::Result, DB}; + 11 + 12 -use crate::require_is_writer_internal; + 13 - + 12 pub async fn require_is_writer(authed: &ApiAuthed, path: &str, + w_id: &str, db: DB) -> Result<()> { + 15 - require_is_writer_internal( + 13 + windmill_api_auth::permissions::require_is_writer( + 14 authed, + 15 path, + 16 w_id, + +● Now update windmill-triggers/src/user_ext.rs to re-export from + windmill-api-auth instead of having its own copy. + +● Write(windmill-triggers/src/user_ext.rs) + ⎿  Added 1 line, removed 88 lines + 6 * LICENSE-AGPL for a copy of the license. + 7 */ + 8 + 9 -use sqlx::PgConnection; + 10 -use tracing::Instrument; + 11 -use windmill_api_auth::ApiAuthed; + 12 -use windmill_audit::{audit_oss::audit_log, ActionKind}; + 13 -use windmill_common::{ + 14 - error::{Error, Result}, + 15 - utils::rd_string, + 16 - worker::CLOUD_HOSTED, + 17 - DB, + 18 -}; + 19 - + 20 -#[derive(serde::Deserialize)] + 21 -pub struct NewToken { + 22 - pub label: Option, + 23 - pub expiration: Option>, + 24 - pub impersonate_email: Option, + 25 - pub scopes: Option>, + 26 - pub workspace_id: Option, + 27 -} + 28 - + 29 -impl NewToken { + 30 - pub fn new( + 31 - label: Option, + 32 - expiration: Option>, + 33 - impersonate_email: Option, + 34 - scopes: Option>, + 35 - workspace_id: Option, + 36 - ) -> NewToken { + 37 - NewToken { label, expiration, impersonate_email, scopes + -, workspace_id } + 38 - } + 39 -} + 40 - + 41 -pub async fn create_token_internal( + 42 - tx: &mut PgConnection, + 43 - db: &DB, + 44 - authed: &ApiAuthed, + 45 - token_config: NewToken, + 46 -) -> Result { + 47 - let token = rd_string(32); + 48 - + 49 - let is_super_admin = sqlx::query_scalar!( + 50 - "SELECT super_admin FROM password WHERE email = $1", + 51 - authed.email + 52 - ) + 53 - .fetch_optional(&mut *tx) + 54 - .await? + 55 - .unwrap_or(false); + 56 - if *CLOUD_HOSTED { + 57 - let nb_tokens = + 58 - sqlx::query_scalar!("SELECT COUNT(*) FROM token WHE + -RE email = $1", &authed.email) + 59 - .fetch_one(db) + 60 - .await?; + 61 - if nb_tokens.unwrap_or(0) >= 10000 { + 62 - return Err(Error::BadRequest( + 63 - "You have reached the maximum number of tokens + -(10000) on cloud. Contact support@windmill.dev to increase the + -limit" + 64 - .to_string(), + 65 - )); + 66 - } + 67 - } + 68 - sqlx::query!( + 69 - "INSERT INTO token + 70 - (token, email, label, expiration, super_admin, scop + -es, workspace_id) + 71 - VALUES ($1, $2, $3, $4, $5, $6, $7)", + 72 - token, + 73 - authed.email, + 74 - token_config.label, + 75 - token_config.expiration, + 76 - is_super_admin, + 77 - token_config.scopes.as_ref().map(|x| x.as_slice()), + 78 - token_config.workspace_id, + 79 - ) + 80 - .execute(&mut *tx) + 81 - .await?; + 82 - + 83 - audit_log( + 84 - &mut *tx, + 85 - authed, + 86 - "users.token.create", + 87 - ActionKind::Create, + 88 - &"global", + 89 - Some(&token[0..10]), + 90 - None, + 91 - ) + 92 - .instrument(tracing::info_span!("token", email = &authed.em + -ail)) + 93 - .await?; + 94 - + 95 - Ok(token) + 96 -} + 9 +pub use windmill_api_auth::tokens::{create_token_internal, NewT + +oken}; + +● Now update windmill-triggers/src/jobs_ext.rs to re-export scope functions from + windmill-api-auth instead of having its own copy. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-triggers/src/jobs_ext.rs) + ⎿  Added 2 lines, removed 28 lines + 21 use windmill_common::{ + 22 db::UserDB, + 23 error, + 24 - jobs::check_tag_available_for_workspace_internal, + 24 triggers::TriggerMetadata, + 25 utils::StripPath, + 26 DB, + ... + 33 #[cfg(feature = "enterprise")] + 34 pub use windmill_common::ee_oss::check_license_key_valid; + 35 + 37 -pub fn get_scope_tags(authed: &ApiAuthed) -> Option> + -{ + 38 - authed.scopes.as_ref()?.iter().find_map(|s| { + 39 - if s.starts_with("if_jobs:filter_tags:") { + 40 - Some( + 41 - s.trim_start_matches("if_jobs:filter_tags:") + 42 - .split(",") + 43 - .collect::>(), + 44 - ) + 45 - } else { + 46 - None + 47 - } + 48 - }) + 49 -} + 36 +// Re-export scope helpers from windmill-api-auth + 37 +pub use windmill_api_auth::scopes::{check_tag_available_for_wor + +kspace, get_scope_tags}; + 38 + 51 -pub async fn check_tag_available_for_workspace( + 52 - db: &DB, + 53 - w_id: &str, + 54 - tag: &Option, + 55 - authed: &ApiAuthed, + 56 -) -> error::Result<()> { + 57 - if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) + - { + 58 - let tags = get_scope_tags(authed); + 59 - check_tag_available_for_workspace_internal(db, w_id, ta + -g, &authed.email, tags).await + 60 - } else { + 61 - Ok(()) + 62 - } + 63 -} + 64 - + 39 /// Trait for complex job operations that stay in windmill-api. + 40 /// windmill-api provides the implementation at startup via `se + t_ops()`. + 41 #[axum::async_trait] + +● Now update windmill-api/src/users.rs to remove the duplicated functions and + import from windmill-api-auth. First let me check what's used from users.rs + that I'm moving. + +● Searched for 1 pattern, read 1 file (ctrl+o to expand) + +● Now I need to update windmill-api/src/users.rs to replace duplicated functions + with re-exports. Let me replace get_scope_tags, require_owner_of_path, + get_perm_in_extra_perms_for_authed, require_is_writer, NewToken, and + create_token_internal. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/users.rs) + ⎿  Added 1 line, removed 13 lines + 161 } + 162 } + 163 + 164 -pub fn get_scope_tags(authed: &ApiAuthed) -> Option> + - { + 165 - authed.scopes.as_ref()?.iter().find_map(|s| { + 166 - if s.starts_with("if_jobs:filter_tags:") { + 167 - Some( + 168 - s.trim_start_matches("if_jobs:filter_tags:") + 169 - .split(",") + 170 - .collect::>(), + 171 - ) + 172 - } else { + 173 - None + 174 - } + 175 - }) + 176 -} + 164 +pub use windmill_api_auth::scopes::get_scope_tags; + 165 + 166 #[derive(Clone, Debug)] + 167 pub struct OptAuthed(pub Option); + +● Now replace require_owner_of_path, get_perm_in_extra_perms_for_authed, and + require_is_writer. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/users.rs) + ⎿  Added 1 line, removed 29 lines + 840 } + 841 } + 842 + 843 -pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) - + -> Result<()> { + 844 - if authed.is_admin { + 845 - return Ok(()); + 846 - } + 847 - if !path.is_empty() { + 848 - let splitted = path.split("/").collect::>(); + 849 - if splitted[0] == "u" { + 850 - if splitted[1] == authed.username { + 851 - Ok(()) + 852 - } else { + 853 - Err(Error::BadRequest(format!( + 854 - "only the owner {} is authorized to perfor + -m this operation", + 855 - splitted[1] + 856 - ))) + 857 - } + 858 - } else if splitted[0] == "f" { + 859 - crate::folders::require_is_owner(authed, splitted[ + -1]) + 860 - } else { + 861 - Err(Error::BadRequest(format!( + 862 - "Not recognized path kind: {}", + 863 - path + 864 - ))) + 865 - } + 866 - } else { + 867 - Err(Error::BadRequest(format!( + 868 - "Cannot be owner of an empty path" + 869 - ))) + 870 - } + 871 -} + 843 +pub use windmill_api_auth::permissions::require_owner_of_path; + 844 + 845 /// Checks that a user has at least read access to the path fo + r preview jobs. + 846 /// This prevents privilege escalation where a user could run + preview code + +● Now replace get_perm_in_extra_perms_for_authed and require_is_writer. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/users.rs) + ⎿  Added 2 lines, removed 89 lines + 899 } + 900 } + 901 + 902 -pub fn get_perm_in_extra_perms_for_authed( + 903 - v: serde_json::Value, + 904 - authed: &ApiAuthed, + 905 -) -> Option { + 906 - match v { + 907 - serde_json::Value::Object(obj) => { + 908 - let mut keys = vec![format!("u/{}", authed.usernam + -e)]; + 909 - for g in authed.groups.iter() { + 910 - keys.push(format!("g/{}", g)); + 911 - } + 912 - let mut res = None; + 913 - for k in keys { + 914 - if let Some(v) = obj.get(&k) { + 915 - if let Some(v) = v.as_bool() { + 916 - if v { + 917 - return Some(true); + 918 - } + 919 - res = Some(v); + 920 - } + 921 - } + 922 - } + 923 - res + 924 - } + 925 - _ => None, + 926 - } + 927 -} + 928 - + 929 -pub async fn require_is_writer( + 930 - authed: &ApiAuthed, + 931 - path: &str, + 932 - w_id: &str, + 933 - db: DB, + 934 - query: &str, + 935 - kind: &str, + 936 -) -> Result<()> { + 937 - if authed.is_admin { + 938 - return Ok(()); + 939 - } + 940 - if !path.is_empty() { + 941 - if require_owner_of_path(authed, path).is_ok() { + 942 - return Ok(()); + 943 - } + 944 - if path.starts_with("f/") && path.split('/').count() > + -= 2 { + 945 - let folder = path.split('/').nth(1).unwrap(); + 946 - let extra_perms = sqlx::query_scalar!( + 947 - "SELECT extra_perms FROM folder WHERE name = $ + -1 AND workspace_id = $2", + 948 - folder, + 949 - w_id + 950 - ) + 951 - .fetch_optional(&db) + 952 - .await?; + 953 - if let Some(perms) = extra_perms { + 954 - let is_folder_writer = + 955 - get_perm_in_extra_perms_for_authed(perms, + -authed).unwrap_or(false); + 956 - if is_folder_writer { + 957 - return Ok(()); + 958 - } + 959 - } + 960 - } + 961 - let extra_perms = sqlx::query_scalar(query) + 962 - .bind(path) + 963 - .bind(w_id) + 964 - .fetch_optional(&db) + 965 - .await?; + 966 - if let Some(perms) = extra_perms { + 967 - let perm = get_perm_in_extra_perms_for_authed(perm + -s, authed); + 968 - match perm { + 969 - Some(true) => Ok(()), + 970 - Some(false) => Err(Error::BadRequest(format!( + 971 - "User {} is not a writer of {kind} path {p + -ath}", + 972 - authed.username + 973 - ))), + 974 - None => Err(Error::BadRequest(format!( + 975 - "User {} has neither read or write permiss + -ion on {kind} {path}", + 976 - authed.username + 977 - ))), + 978 - } + 979 - } else { + 980 - Err(Error::BadRequest(format!( + 981 - "{path} does not exist yet and user {} is not + -an owner of the parent folder", + 982 - authed.username + 983 - ))) + 984 - } + 985 - } else { + 986 - Err(Error::BadRequest(format!( + 987 - "Cannot be writer of an empty path" + 988 - ))) + 989 - } + 990 -} + 902 +pub use windmill_api_auth::permissions::get_perm_in_extra_perm + +s_for_authed; + 903 +pub use windmill_api_auth::permissions::require_is_writer; + 904 async fn whois( + 905 Extension(db): Extension, + 906 Path((w_id, username)): Path<(String, String)>, + +● Now replace NewToken and create_token_internal in users.rs. + +● Read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/users.rs) + ⎿  Added 1 line, removed 21 lines + 307 pub scopes: Option>, + 308 } + 309 + 310 -#[derive(Deserialize)] + 311 -pub struct NewToken { + 312 - pub label: Option, + 313 - pub expiration: Option>, + 314 - pub impersonate_email: Option, + 315 - pub scopes: Option>, + 316 - pub workspace_id: Option, + 317 -} + 310 +pub use windmill_api_auth::tokens::NewToken; + 311 + 319 -#[cfg(feature = "native_trigger")] + 320 -impl NewToken { + 321 - pub fn new( + 322 - label: Option, + 323 - expiration: Option>, + 324 - impersonate_email: Option, + 325 - scopes: Option>, + 326 - workspace_id: Option, + 327 - ) -> NewToken { + 328 - NewToken { label, expiration, impersonate_email, scope + -s, workspace_id } + 329 - } + 330 -} + 331 - + 312 #[derive(Deserialize)] + 313 pub struct Login { + 314 pub email: String, + +● Now replace create_token_internal. + +● Searched for 1 pattern, read 1 file (ctrl+o to expand) + +● Update(windmill-api/src/users.rs) + ⎿  Added 1 line, removed 56 lines + 1960 Ok(token) + 1961 } + 1962 + 1963 -pub async fn create_token_internal( + 1964 - tx: &mut PgConnection, + 1965 - db: &DB, + 1966 - authed: &ApiAuthed, + 1967 - token_config: NewToken, + 1968 -) -> Result { + 1969 - let token = rd_string(32); + 1963 +pub use windmill_api_auth::tokens::create_token_internal; + 1964 + 1971 - let is_super_admin = sqlx::query_scalar!( + 1972 - "SELECT super_admin FROM password WHERE email = $1", + 1973 - authed.email + 1974 - ) + 1975 - .fetch_optional(&mut *tx) + 1976 - .await? + 1977 - .unwrap_or(false); + 1978 - if *CLOUD_HOSTED { + 1979 - let nb_tokens = + 1980 - sqlx::query_scalar!("SELECT COUNT(*) FROM token W + -HERE email = $1", &authed.email) + 1981 - .fetch_one(db) + 1982 - .await?; + 1983 - if nb_tokens.unwrap_or(0) >= 10000 { + 1984 - return Err(Error::BadRequest( + 1985 - "You have reached the maximum number of token + -s (10000) on cloud. Contact support@windmill.dev to increase + -the limit" + 1986 - .to_string(), + 1987 - )); + 1988 - } + 1989 - } + 1990 - sqlx::query!( + 1991 - "INSERT INTO token + 1992 - (token, email, label, expiration, super_admin, sc + -opes, workspace_id) + 1993 - VALUES ($1, $2, $3, $4, $5, $6, $7)", + 1994 - token, + 1995 - authed.email, + 1996 - token_config.label, + 1997 - token_config.expiration, + 1998 - is_super_admin, + 1999 - token_config.scopes.as_ref().map(|x| x.as_slice()), + 2000 - token_config.workspace_id, + 2001 - ) + 2002 - .execute(&mut *tx) + 2003 - .await?; + 2004 - + 2005 - audit_log( + 2006 - &mut *tx, + 2007 - authed, + 2008 - "users.token.create", + 2009 - ActionKind::Create, + 2010 - &"global", + 2011 - Some(&token[0..10]), + 2012 - None, + 2013 - ) + 2014 - .instrument(tracing::info_span!("token", email = &authed. + -email)) + 2015 - .await?; + 2016 - + 2017 - Ok(token) + 2018 -} + 2019 - + 1965 async fn create_token( + 1966 Extension(db): Extension, + 1967 authed: ApiAuthed, + +● Now I also need to update windmill-api/src/folders.rs to use windmill-api-auth + for is_owner and require_is_owner. + +● Update(windmill-api/src/folders.rs) + ⎿  Added 2 lines, removed 18 lines + 306 Ok(Json(is_owner(&authed, &name))) + 307 } + 308 + 309 -pub fn is_owner(ApiAuthed { is_admin, folders, .. }: &ApiAuthe + -d, name: &str) -> bool { + 310 - if *is_admin { + 311 - true + 312 - } else { + 313 - folders.into_iter().any(|x| x.0 == name && x.2) + 314 - } + 315 -} + 309 +pub use windmill_api_auth::permissions::is_folder_owner as is_ + +owner; + 310 +pub use windmill_api_auth::permissions::require_is_folder_owne + +r as require_is_owner; + 311 + 317 -pub fn require_is_owner(authed: &ApiAuthed, name: &str) -> Res + -ult<()> { + 318 - if is_owner(authed, name) { + 319 - Ok(()) + 320 - } else { + 321 - Err(windmill_common::error::Error::NotAuthorized(forma + -t!( + 322 - "You are not owner of the folder {}", + 323 - name + 324 - ))) + 325 - } + 326 -} + 327 - + 312 async fn update_folder( + 313 authed: ApiAuthed, + 314 Extension(db): Extension, + +● Now let me check compilation. + +● Bash(cargo check -p windmill-api-auth --no-default-features 2>&1 | tail -20) + ⎿ Checking windmill-audit v1.628.2 (/home/rfiszel/windmill/backend/wi + ndmill-audit) + Checking windmill-api-auth v1.628.2 (/home/rfiszel/windmill/backend + … +2 lines (ctrl+o to expand) + + + +● Bash(cargo check -p windmill-triggers --no-default-features 2>&1 | tail -20) + ⎿ Checking windmill-api-auth v1.628.2 (/home/rfiszel/windmill/backend + /windmill-api-auth) + Checking windmill-triggers v1.628.2 (/home/rfiszel/windmill/backend + … +2 lines (ctrl+o to expand) + + + +● Bash(cargo check -p windmill-api --no-default-features 2>&1 | tail -30) + ⎿ Checking windmill-api-auth v1.628.2 (/home/rfiszel/windmill/backend + /windmill-api-auth) + Checking windmill-triggers v1.628.2 (/home/rfiszel/windmill/backend + … +31 lines (ctrl+o to expand) + + + +● Only warnings. Let me fix them. + +● Update(windmill-api/src/capture.rs) + ⎿  Added 1 line, removed 1 line + 85 worker::to_raw_value, + 86 }; + 87 + 88 -use windmill_queue::PushArgsOwned; + 88 +use windmill_queue::PushArgsOwned; // used by webhook_payload a + +nd other capture endpoints + 89 + 90 + 91 + +● Searched for 1 pattern (ctrl+o to expand) + +● It's truly unused now since insert_capture_payload was the only user and is + now re-exported from windmill-triggers. + +● Update(windmill-api/src/capture.rs) + ⎿  Removed 1 line + 85 worker::to_raw_value, + 86 }; + 87 + 88 -use windmill_queue::PushArgsOwned; // used by webhook_payload a + -nd other capture endpoints + 88 + 89 + 90 + +● Searched for 2 patterns (ctrl+o to expand) + +● Good. Now fix the unused import in users.rs. + +● Searched for 1 pattern (ctrl+o to expand) + +● Update(windmill-api/src/users.rs) + ⎿  Added 1 line, removed 1 line + 8 + 9 #![allow(non_snake_case)] + 10 + 11 -use sqlx::{PgConnection, Postgres, Transaction}; + 11 +use sqlx::{Postgres, Transaction}; + 12 + 13 use std::sync::atomic::AtomicBool; + 14 use std::sync::Arc; + +● For the get_perm_in_extra_perms_for_authed warning, it's unused because + nothing in windmill-api's users.rs uses it directly anymore. But it was a pub + function, so other files may use it via + crate::users::get_perm_in_extra_perms_for_authed. Let me check. + +● Searched for 1 pattern (ctrl+o to expand) + +● It's only in the re-export itself, nobody uses it. Let me remove it. + +● Update(windmill-api/src/users.rs) + ⎿  Removed 1 line + 879 } + 880 } + 881 + 882 -pub use windmill_api_auth::permissions::get_perm_in_extra_perm + -s_for_authed; + 882 pub use windmill_api_auth::permissions::require_is_writer; + 883 async fn whois( + 884 Extension(db): Extension, + +● Bash(cargo check -p windmill-api --no-default-features 2>&1 | tail -10) + ⎿ Checking windmill-api v1.628.2 (/home/rfiszel/windmill/backend/wind + mill-api) + Finished `dev` profile [unoptimized + debuginfo] target(s) in 4.54s + + + +● Clean compilation. Tier 2 is done. Let me move to Tier 3. + +● Tier 3: Single-source args_ext.rs + + Delete windmill-api/src/args.rs and make windmill-api import from + windmill_triggers::args_ext. + +● Searched for 2 patterns (ctrl+o to expand) + +● Only capture.rs uses crate::args. Let me check what functions are in + windmill-api/src/args.rs vs windmill-triggers/src/args_ext.rs. + +● Searched for 2 patterns (ctrl+o to expand) + +● Good, same public interface. Now I'll replace the windmill-api/src/args.rs + module with a re-export from windmill-triggers. + +● Read 1 file (ctrl+o to expand) + +● I'll replace pub mod args; with a pub use re-export. + +● Update(windmill-api/src/lib.rs) + ⎿  Added 1 line, removed 1 line + 74 mod agent_workers_oss; + 75 mod ai; + 76 mod apps; + 77 -pub mod args; + 77 +pub use windmill_triggers::args_ext as args; + 78 mod assets; + 79 mod audit; + 80 pub mod auth; + +● Now delete the old args.rs file. + +● Bash(rm /home/rfiszel/windmill/backend/windmill-api/src/args.rs) + ⎿  (No content) + +● Bash(cargo check -p windmill-api --no-default-features 2>&1 | tail -10) + ⎿ Checking windmill-api v1.628.2 (/home/rfiszel/windmill/backend/wind + mill-api) + Finished `dev` profile [unoptimized + debuginfo] target(s) in 14.64 + s + + + +● Tier 3 done. Moving to Tier 4. + +● Tier 4: Delete dead code + + The JobUpdate struct was already removed in the Tier 1 step. Now flow_ext.rs, + script_ext.rs, and user_ext.rs are thin wrappers. They're still used by + native_triggers/handler.rs, so I should keep the convenience wrappers (they're + just 1-line delegations now, not duplicates). Let me verify the current state + of user_ext.rs: + +● Read 1 file (ctrl+o to expand) + +● These are all pure re-exports or thin wrappers now - no duplicated logic. The + goal was "logic should be only once at one place", which is achieved. The ext + files are just re-export shims. I won't delete them since they'd break imports + in the triggers code unnecessarily. \ No newline at end of file diff --git a/backend/windmill-api-auth/src/lib.rs b/backend/windmill-api-auth/src/lib.rs index afb27d5c32..fd2ff0c41b 100644 --- a/backend/windmill-api-auth/src/lib.rs +++ b/backend/windmill-api-auth/src/lib.rs @@ -7,7 +7,9 @@ */ mod auth; +pub mod permissions; pub mod scopes; +pub mod tokens; mod types; pub use auth::*; diff --git a/backend/windmill-api-auth/src/permissions.rs b/backend/windmill-api-auth/src/permissions.rs new file mode 100644 index 0000000000..de397035f2 --- /dev/null +++ b/backend/windmill-api-auth/src/permissions.rs @@ -0,0 +1,156 @@ +/* + * Author: Windmill Labs, Inc + * Copyright: Windmill Labs, Inc 2024 + * 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 windmill_common::error::{Error, Result}; +use windmill_common::DB; + +use crate::ApiAuthed; + +/// Check if the user is an owner of the given path. +pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) -> Result<()> { + if authed.is_admin { + return Ok(()); + } + if !path.is_empty() { + let splitted = path.split("/").collect::>(); + if splitted[0] == "u" { + if splitted[1] == authed.username { + Ok(()) + } else { + Err(Error::BadRequest(format!( + "only the owner {} is authorized to perform this operation", + splitted[1] + ))) + } + } else if splitted[0] == "f" { + require_is_folder_owner(authed, splitted[1]) + } else { + Err(Error::BadRequest(format!( + "Not recognized path kind: {}", + path + ))) + } + } else { + Err(Error::BadRequest(format!( + "Cannot be owner of an empty path" + ))) + } +} + +pub fn is_folder_owner( + ApiAuthed { is_admin, folders, .. }: &ApiAuthed, + name: &str, +) -> bool { + if *is_admin { + true + } else { + folders.into_iter().any(|x| x.0 == name && x.2) + } +} + +pub fn require_is_folder_owner(authed: &ApiAuthed, name: &str) -> Result<()> { + if is_folder_owner(authed, name) { + Ok(()) + } else { + Err(Error::NotAuthorized(format!( + "You are not owner of the folder {}", + name + ))) + } +} + +pub fn get_perm_in_extra_perms_for_authed( + v: serde_json::Value, + authed: &ApiAuthed, +) -> Option { + match v { + serde_json::Value::Object(obj) => { + let mut keys = vec![format!("u/{}", authed.username)]; + for g in authed.groups.iter() { + keys.push(format!("g/{}", g)); + } + let mut res = None; + for k in keys { + if let Some(v) = obj.get(&k) { + if let Some(v) = v.as_bool() { + if v { + return Some(true); + } + res = Some(v); + } + } + } + res + } + _ => None, + } +} + +/// Generic require_is_writer with a configurable SQL query. +pub async fn require_is_writer( + authed: &ApiAuthed, + path: &str, + w_id: &str, + db: DB, + query: &str, + kind: &str, +) -> Result<()> { + if authed.is_admin { + return Ok(()); + } + if !path.is_empty() { + if require_owner_of_path(authed, path).is_ok() { + return Ok(()); + } + if path.starts_with("f/") && path.split('/').count() >= 2 { + let folder = path.split('/').nth(1).unwrap(); + let extra_perms = sqlx::query_scalar!( + "SELECT extra_perms FROM folder WHERE name = $1 AND workspace_id = $2", + folder, + w_id + ) + .fetch_optional(&db) + .await?; + if let Some(perms) = extra_perms { + let is_folder_writer = + get_perm_in_extra_perms_for_authed(perms, authed).unwrap_or(false); + if is_folder_writer { + return Ok(()); + } + } + } + let extra_perms = sqlx::query_scalar(query) + .bind(path) + .bind(w_id) + .fetch_optional(&db) + .await?; + if let Some(perms) = extra_perms { + let perm = get_perm_in_extra_perms_for_authed(perms, authed); + match perm { + Some(true) => Ok(()), + Some(false) => Err(Error::BadRequest(format!( + "User {} is not a writer of {kind} path {path}", + authed.username + ))), + None => Err(Error::BadRequest(format!( + "User {} has neither read or write permission on {kind} {path}", + authed.username + ))), + } + } else { + Err(Error::BadRequest(format!( + "{path} does not exist yet and user {} is not an owner of the parent folder", + authed.username + ))) + } + } else { + Err(Error::BadRequest(format!( + "Cannot be writer of an empty path" + ))) + } +} diff --git a/backend/windmill-api-auth/src/scopes.rs b/backend/windmill-api-auth/src/scopes.rs index 5a8cc8b148..ed8cbe8d08 100644 --- a/backend/windmill-api-auth/src/scopes.rs +++ b/backend/windmill-api-auth/src/scopes.rs @@ -9,7 +9,9 @@ use itertools::Itertools; use serde::{Deserialize, Serialize}; use std::collections::HashSet; -use windmill_common::error::{Error, Result}; +use windmill_common::error::{self, Error, Result}; +use windmill_common::jobs::check_tag_available_for_workspace_internal; +use windmill_common::DB; use crate::ApiAuthed; @@ -702,6 +704,34 @@ where Ok(()) } +pub fn get_scope_tags(authed: &ApiAuthed) -> Option> { + authed.scopes.as_ref()?.iter().find_map(|s| { + if s.starts_with("if_jobs:filter_tags:") { + Some( + s.trim_start_matches("if_jobs:filter_tags:") + .split(",") + .collect::>(), + ) + } else { + None + } + }) +} + +pub async fn check_tag_available_for_workspace( + db: &DB, + w_id: &str, + tag: &Option, + authed: &ApiAuthed, +) -> error::Result<()> { + if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) { + let tags = get_scope_tags(authed); + check_tag_available_for_workspace_internal(db, w_id, tag, &authed.email, tags).await + } else { + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/backend/windmill-api-auth/src/tokens.rs b/backend/windmill-api-auth/src/tokens.rs new file mode 100644 index 0000000000..af0d9d15cb --- /dev/null +++ b/backend/windmill-api-auth/src/tokens.rs @@ -0,0 +1,97 @@ +/* + * Author: Windmill Labs, Inc + * Copyright: Windmill Labs, Inc 2024 + * 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 sqlx::PgConnection; +use tracing::Instrument; +use windmill_audit::{audit_oss::audit_log, ActionKind}; +use windmill_common::{ + error::{Error, Result}, + utils::rd_string, + worker::CLOUD_HOSTED, + DB, +}; + +use crate::ApiAuthed; + +#[derive(serde::Deserialize)] +pub struct NewToken { + pub label: Option, + pub expiration: Option>, + pub impersonate_email: Option, + pub scopes: Option>, + pub workspace_id: Option, +} + +impl NewToken { + pub fn new( + label: Option, + expiration: Option>, + impersonate_email: Option, + scopes: Option>, + workspace_id: Option, + ) -> NewToken { + NewToken { label, expiration, impersonate_email, scopes, workspace_id } + } +} + +pub async fn create_token_internal( + tx: &mut PgConnection, + db: &DB, + authed: &ApiAuthed, + token_config: NewToken, +) -> Result { + let token = rd_string(32); + + let is_super_admin = sqlx::query_scalar!( + "SELECT super_admin FROM password WHERE email = $1", + authed.email + ) + .fetch_optional(&mut *tx) + .await? + .unwrap_or(false); + if *CLOUD_HOSTED { + let nb_tokens = + sqlx::query_scalar!("SELECT COUNT(*) FROM token WHERE email = $1", &authed.email) + .fetch_one(db) + .await?; + if nb_tokens.unwrap_or(0) >= 10000 { + return Err(Error::BadRequest( + "You have reached the maximum number of tokens (10000) on cloud. Contact support@windmill.dev to increase the limit" + .to_string(), + )); + } + } + sqlx::query!( + "INSERT INTO token + (token, email, label, expiration, super_admin, scopes, workspace_id) + VALUES ($1, $2, $3, $4, $5, $6, $7)", + token, + authed.email, + token_config.label, + token_config.expiration, + is_super_admin, + token_config.scopes.as_ref().map(|x| x.as_slice()), + token_config.workspace_id, + ) + .execute(&mut *tx) + .await?; + + audit_log( + &mut *tx, + authed, + "users.token.create", + ActionKind::Create, + &"global", + Some(&token[0..10]), + None, + ) + .instrument(tracing::info_span!("token", email = &authed.email)) + .await?; + + Ok(token) +} diff --git a/backend/windmill-api/src/args.rs b/backend/windmill-api/src/args.rs deleted file mode 100644 index bcbfba03d0..0000000000 --- a/backend/windmill-api/src/args.rs +++ /dev/null @@ -1,678 +0,0 @@ -use std::collections::HashMap; - -use axum::{ - extract::{FromRequest, Multipart, Query, Request}, - http::HeaderMap, - response::{IntoResponse, Response}, -}; -use bytes::Bytes; -use http::{header::CONTENT_TYPE, StatusCode}; -use serde::{Deserialize, Serialize}; -use serde_json::value::RawValue; -use sqlx::types::JsonRawValue; -use windmill_common::{ - error::Error, - triggers::{RunnableFormat, RunnableFormatVersion, TriggerKind}, - worker::to_raw_value, - DB, -}; -use windmill_queue::PushArgsOwned; - -use crate::{ - db::ApiAuthed, - triggers::trigger_helpers::{get_runnable_format, RunnableId}, -}; - -#[derive(Debug)] -pub enum RawBody { - Json(String), - CEJson(String), - Text(String), - Xml(String), - UrlEncoded(Bytes), - Multipart(Multipart), - Empty, -} - -#[derive(Debug, Clone, Serialize)] -#[serde(untagged)] -pub enum Body { - HashMap(HashMap>), - NoHashMap(Box), -} - -#[derive(Debug, Clone, Default)] -pub struct WebhookArgsMetadata { - pub raw_string: Option, - pub headers: HeaderMap, - pub query: Option, - pub method: http::Method, - pub query_wrap_body: bool, - pub query_use_raw: bool, - pub query_include_header: Option, - pub query_include_query: Option, -} - -pub struct RawWebhookArgs { - pub body: RawBody, - pub metadata: WebhookArgsMetadata, -} - -#[derive(Debug, Clone)] -pub struct WebhookArgs { - pub body: Body, - pub metadata: WebhookArgsMetadata, -} - -// capture -// - -impl RawWebhookArgs { - #[cfg(not(feature = "parquet"))] - pub async fn process_multipart( - _multipart: Multipart, - _authed: &ApiAuthed, - _db: &DB, - _w_id: &str, - ) -> Result>, Error> { - return Err(Error::BadRequest(format!( - "multipart/form-data requires the parquet feature" - ))); - } - - #[cfg(feature = "parquet")] - async fn process_multipart( - mut multipart: Multipart, - authed: &ApiAuthed, - db: &DB, - w_id: &str, - ) -> Result>, Error> { - use crate::job_helpers_oss::{ - get_random_file_name, get_workspace_s3_resource, upload_file_internal, - }; - use futures::TryStreamExt; - use object_store::{Attribute, Attributes}; - use windmill_common::s3_helpers::build_object_store_client; - - let (_, s3_resource) = get_workspace_s3_resource(authed, db, None, w_id, None).await?; - - if let Some(s3_resource) = s3_resource { - let s3_client = build_object_store_client(&s3_resource).await?; - - let mut body = HashMap::new(); - let mut files = HashMap::new(); - - while let Some(field) = multipart.next_field().await.map_err(|e| { - Error::BadRequest(format!("Error reading multipart field: {}", e.body_text())) - })? { - if let Some(name) = field.name().map(|x| x.to_string()) { - if let Some(content_type) = field.content_type() { - let ext = field - .file_name() - .map(|x| x.split('.').last()) - .flatten() - .map(|x| x.to_string()); - - let file_key = get_random_file_name(ext); - - let options = Attributes::from_iter(vec![ - (Attribute::ContentType, content_type.to_string()), - ( - Attribute::ContentDisposition, - if let Some(filename) = field.file_name() { - format!("inline; filename=\"{}\"", filename) - } else { - "inline".to_string() - }, - ), - ]) - .into(); - - let bytes_stream = field - .into_stream() - .map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err)); - - upload_file_internal(s3_client.clone(), &file_key, bytes_stream, options) - .await?; - - files.entry(name).or_insert(vec![]).push(serde_json::json!({ - "s3": &file_key - })); - } else { - body.insert(name, to_raw_value(&field.text().await.unwrap_or_default())); - } - } - } - - for (k, v) in files { - body.insert(k, to_raw_value(&v)); - } - - Ok(body) - } else { - Err(Error::BadRequest(format!( - "You need to connect your workspace to an S3 bucket to use multipart/form-data" - ))) - } - } - - pub async fn process_args( - self, - authed: &ApiAuthed, - db: &DB, - w_id: &str, - force_use_raw: Option, - ) -> Result { - let use_raw = force_use_raw.unwrap_or(self.metadata.query_use_raw); - - match self.body { - RawBody::Multipart(multipart) => { - let body = Self::process_multipart(multipart, authed, db, w_id).await?; - Ok(WebhookArgs { body: Body::HashMap(body), metadata: self.metadata }) - } - RawBody::Empty => { - let mut metadata = self.metadata; - if use_raw { - metadata.raw_string = Some("".to_string()); - } - Ok(WebhookArgs { body: Body::HashMap(HashMap::new()), metadata }) - } - RawBody::Text(s) | RawBody::Xml(s) => Ok(WebhookArgs { - body: Body::HashMap(HashMap::new()), - metadata: WebhookArgsMetadata { raw_string: Some(s), ..self.metadata }, - }), - RawBody::UrlEncoded(bytes) => { - let mut metadata = self.metadata; - if use_raw { - let raw_string = String::from_utf8(bytes.to_vec()) - .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)))?; - metadata.raw_string = Some(raw_string); - } - let payload: HashMap> = serde_urlencoded::from_bytes(&bytes) - .map_err(|e| Error::BadRequest(format!("invalid urlencoded data: {}", e)))?; - let payload = payload - .into_iter() - .map(|(k, v)| (k, to_raw_value(&v))) - .collect::>(); - - Ok(WebhookArgs { body: Body::HashMap(payload), metadata }) - } - RawBody::Json(s) => WebhookArgs::from_json(self.metadata, use_raw, s).await, - RawBody::CEJson(s) => WebhookArgs::from_ce_json(self.metadata, use_raw, s).await, - } - } - - pub async fn to_main_args( - self, - authed: &ApiAuthed, - db: &DB, - w_id: &str, - ) -> Result { - let args = self.process_args(authed, db, w_id, None).await?; - args.to_main_args() - } - - pub async fn to_args_from_runnable( - self, - authed: &ApiAuthed, - db: &DB, - w_id: &str, - runnable_id: RunnableId, - skip_preprocessor: Option, - ) -> Result { - let args = self.process_args(authed, db, w_id, None).await?; - args.to_args_from_runnable(db, w_id, runnable_id, skip_preprocessor) - .await - } -} - -#[derive(Serialize)] -struct WebhookPreprocessorEvent { - kind: String, - body: Box, - raw_string: Option, - headers: HashMap>, - query: HashMap>, -} - -impl WebhookArgs { - pub fn to_main_args(self) -> Result { - self.to_args_from_format(RunnableFormat { - has_preprocessor: false, - version: RunnableFormatVersion::V2, - }) - } - - pub async fn to_args_from_runnable( - self, - db: &DB, - w_id: &str, - runnable_id: RunnableId, - skip_preprocessor: Option, - ) -> Result { - if skip_preprocessor.unwrap_or(false) { - self.to_main_args() - } else { - let runnable_format = - get_runnable_format(runnable_id, w_id, db, &TriggerKind::Webhook).await?; - - self.to_args_from_format(runnable_format) - } - } - - pub fn to_args_from_format( - self, - runnable_format: RunnableFormat, - ) -> Result { - let headers = build_headers( - &self.metadata.headers, - self.metadata.query_include_header, - runnable_format.has_preprocessor, - ); - - let query = build_query( - self.metadata.query.as_deref(), - self.metadata.query_include_query, - runnable_format.has_preprocessor, - ); - - match runnable_format { - RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V2 } => { - let mut args = HashMap::new(); - - args.insert( - "event".to_string(), - to_raw_value(&WebhookPreprocessorEvent { - kind: "webhook".to_string(), - body: to_raw_value(&self.body), - raw_string: self.metadata.raw_string, - headers, - query, - }), - ); - - Ok(PushArgsOwned { args, extra: None }) - } - RunnableFormat { has_preprocessor, .. } => { - let mut extra = HashMap::new(); - - let WebhookArgsMetadata { query_wrap_body, raw_string, .. } = self.metadata; - - for (k, v) in headers { - extra.insert(k, v); - } - - for (k, v) in query { - extra.insert(k, v); - } - - if let Some(raw_string) = raw_string { - extra.insert("raw_string".to_string(), to_raw_value(&raw_string)); - } - - if has_preprocessor { - // if has preprocessor, it has to be v1 - extra.insert( - "wm_trigger".to_string(), - to_raw_value(&serde_json::json!({ - "kind": "webhook", - })), - ); - } - - let extra = if extra.is_empty() { None } else { Some(extra) }; - - match self.body { - Body::HashMap(mut body) => { - if query_wrap_body { - body = HashMap::from([("body".to_string(), to_raw_value(&body))]); - } - Ok(PushArgsOwned { args: body, extra }) - } - Body::NoHashMap(args) => { - let mut hm = HashMap::new(); - hm.insert("body".to_string(), args); - Ok(PushArgsOwned { args: hm, extra }) - } - } - } - } - } -} - -#[derive(Deserialize)] -pub struct RequestQuery { - pub raw: Option, - pub wrap_body: Option, - pub include_header: Option, - pub include_query: Option, -} - -async fn req_to_string( - req: Request, - _state: &S, -) -> Result { - let bytes = Bytes::from_request(req, _state) - .await - .map_err(IntoResponse::into_response)?; - String::from_utf8(bytes.to_vec()) - .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response()) -} - -pub async fn try_from_request_body( - request: Request, - _state: &S, - is_http_trigger: bool, -) -> Result -where - S: Send + Sync, -{ - let (content_type, metadata) = { - let headers_map = request.headers(); - let content_type_header = headers_map.get(CONTENT_TYPE); - let content_type = content_type_header.and_then(|value| value.to_str().ok()); - let uri = request.uri(); - let request_query = Query::::try_from_uri(uri).unwrap().0; - - let query = uri.query().map(|s| s.to_owned()); - let raw = !is_http_trigger && request_query.raw.unwrap_or(false); - let wrap_body = !is_http_trigger && request_query.wrap_body.unwrap_or(false); - ( - content_type, - WebhookArgsMetadata { - headers: headers_map.clone(), - query, - method: request.method().clone(), - raw_string: None, - query_wrap_body: wrap_body, - query_use_raw: raw, - query_include_header: request_query.include_header, - query_include_query: request_query.include_query, - }, - ) - }; - - let no_content_type = content_type.is_none(); - if no_content_type || content_type.unwrap().starts_with("application/json") { - let bytes = Bytes::from_request(request, _state) - .await - .map_err(IntoResponse::into_response)?; - if no_content_type && bytes.is_empty() { - Ok(RawWebhookArgs { body: RawBody::Empty, metadata }) - } else { - let str = String::from_utf8(bytes.to_vec()) - .map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?; - Ok(RawWebhookArgs { body: RawBody::Json(str), metadata }) - } - } else if content_type - .unwrap() - .starts_with("application/cloudevents+json") - { - let str = req_to_string(request, _state).await?; - - Ok(RawWebhookArgs { body: RawBody::CEJson(str), metadata }) - } else if content_type - .unwrap() - .starts_with("application/cloudevents-batch+json") - { - Err( - Error::BadRequest(format!("Cloud events batching is not supported yet")) - .into_response(), - ) - } else if content_type.unwrap().starts_with("text/plain") { - let str = req_to_string(request, _state).await?; - Ok(RawWebhookArgs { body: RawBody::Text(str), metadata }) - } else if content_type - .unwrap() - .starts_with("application/x-www-form-urlencoded") - { - let bytes = Bytes::from_request(request, _state) - .await - .map_err(IntoResponse::into_response)?; - - Ok(RawWebhookArgs { body: RawBody::UrlEncoded(bytes), metadata }) - } else if content_type.unwrap().starts_with("application/xml") - || content_type.unwrap().starts_with("text/xml") - { - let str = req_to_string(request, _state).await?; - Ok(RawWebhookArgs { body: RawBody::Xml(str), metadata }) - } else if content_type.unwrap().starts_with("multipart/form-data") { - let multipart = Multipart::from_request(request, _state) - .await - .map_err(IntoResponse::into_response)?; - - Ok(RawWebhookArgs { body: RawBody::Multipart(multipart), metadata }) - } else { - Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response()) - } -} - -#[axum::async_trait] -impl FromRequest for RawWebhookArgs -where - S: Send + Sync, -{ - type Rejection = Response; - - async fn from_request(request: Request, _state: &S) -> Result { - let args = try_from_request_body(request, _state, false).await?; - - Ok(args) - } -} - -lazy_static::lazy_static! { - static ref INCLUDE_HEADERS: Vec = std::env::var("INCLUDE_HEADERS") - .ok().map(|x| x - .split(',') - .map(|s| s.to_string()) - .collect()).unwrap_or_default(); -} - -pub fn build_headers( - headers: &HeaderMap, - include_header: Option, - include_all_headers: bool, -) -> HashMap> { - let mut selected_headers = HashMap::new(); - - if include_all_headers { - for (k, v) in headers.iter() { - selected_headers.insert( - k.to_string(), - to_raw_value(&v.to_str().unwrap_or("").to_string()), - ); - } - } else { - let whitelist = include_header - .map(|s| s.split(",").map(|s| s.to_string()).collect::>()) - .unwrap_or_default(); - whitelist - .iter() - .chain(INCLUDE_HEADERS.iter()) - .for_each(|h| { - if let Some(v) = headers.get(h) { - selected_headers.insert( - h.to_string().to_lowercase().replace('-', "_"), - to_raw_value(&v.to_str().unwrap_or("").to_string()), - ); - } - }); - } - selected_headers -} - -pub fn build_query( - query: Option<&str>, - include_query: Option, - include_all_query: bool, -) -> HashMap> { - let Some(query) = query else { - return HashMap::new(); - }; - - if include_all_query { - let queries = - serde_urlencoded::from_str::>(&query).unwrap_or_default(); - queries - .into_iter() - .map(|(k, v)| (k, to_raw_value(&v))) - .collect() - } else { - let parse_query_args = include_query - .map(|s| s.split(",").map(|p| p.to_string()).collect::>()) - .unwrap_or_default(); - let mut args = HashMap::new(); - if !parse_query_args.is_empty() { - let queries = - serde_urlencoded::from_str::>(&query).unwrap_or_default(); - parse_query_args.iter().for_each(|h| { - if let Some(v) = queries.get(h) { - args.insert(h.to_string(), to_raw_value(v)); - } - }); - } - args - } -} - -fn restructure_cloudevents_metadata( - mut p: HashMap>, -) -> Result>, Error> { - let data = p - .remove("data") - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); - let str = data.to_string(); - - let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{'; - - if wrap_body { - let args = serde_json::from_str::>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); - let mut hm = HashMap::new(); - hm.insert("body".to_string(), args); - hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p)); - Ok(hm) - } else { - let mut hm = serde_json::from_str::>>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? - .unwrap_or_else(HashMap::new); - hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p)); - Ok(hm) - } -} - -impl WebhookArgs { - async fn from_json( - mut metadata: WebhookArgsMetadata, - use_raw: bool, - str: String, - ) -> Result { - if use_raw { - metadata.raw_string = Some(str.clone()); - } - - let no_hashmap = str.len() > 0 && str.chars().next().unwrap() != '{'; - - if no_hashmap { - let args = serde_json::from_str::>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); - - Ok(Self { body: Body::NoHashMap(args), metadata }) - } else { - let hm = serde_json::from_str::>>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))? - .unwrap_or_else(HashMap::new); - Ok(Self { body: Body::HashMap(hm), metadata }) - } - } - - async fn from_ce_json( - mut metadata: WebhookArgsMetadata, - use_raw: bool, - str: String, - ) -> Result { - if use_raw { - metadata.raw_string = Some(str.clone()); - } - - let hm = serde_json::from_str::>>(&str) - .map_err(|e| Error::BadRequest(format!("invalid cloudevents+json: {}", e)))?; - let hm = restructure_cloudevents_metadata(hm)?; - Ok(Self { body: Body::HashMap(hm), metadata }) - } -} - -#[cfg(test)] -mod tests { - - use super::*; - - #[tokio::test] - async fn test_cloudevents_json_payload() { - let r1 = r#" - { - "specversion" : "1.0", - "type" : "com.example.someevent", - "source" : "/mycontext", - "subject": null, - "id" : "C234-1234-1234", - "time" : "2018-04-05T17:31:00Z", - "comexampleextension1" : "value", - "comexampleothervalue" : 5, - "datacontenttype" : "application/json", - "data" : { - "appinfoA" : "abc", - "appinfoB" : 123, - "appinfoC" : true - } - } - "#; - let r2 = r#" - { - "specversion" : "1.0", - "type" : "com.example.someevent", - "source" : "/mycontext", - "subject": null, - "id" : "C234-1234-1234", - "time" : "2018-04-05T17:31:00Z", - "comexampleextension1" : "value", - "comexampleothervalue" : 5, - "datacontenttype" : "application/json", - "data" : 1.5 - } - "#; - let metadata = WebhookArgsMetadata::default(); - - let a1 = WebhookArgs::from_ce_json(metadata.clone(), false, r1.to_string()) - .await - .expect("Failed to parse the cloudevent"); - let a2 = WebhookArgs::from_ce_json(metadata.clone(), false, r2.to_string()) - .await - .expect("Failed to parse the cloudevent"); - - match a1.body { - Body::HashMap(body) => { - body.get("WEBHOOK__METADATA__").expect( - "CloudEvents should generate a neighboring `webhook-metadata` field in PushArgs", - ); - } - _ => panic!("Expected a HashMap"), - } - - match a2.body { - Body::HashMap(body) => { - assert_eq!( - body - .get("body") - .expect("Cloud events with a data field with no wrapping curly brackets should be inside of a `body` field in PushArgs") - .to_string(), - "1.5" - ); - } - _ => panic!("Expected a HashMap"), - } - } -} diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index bcc287ec04..46fd6cbd19 100644 --- a/backend/windmill-api/src/capture.rs +++ b/backend/windmill-api/src/capture.rs @@ -82,12 +82,11 @@ use windmill_common::{ error::{JsonResult, Result}, triggers::{RunnableFormat, RunnableFormatVersion, TriggerKind}, utils::{not_found_if_none, paginate, Pagination, RunnableKind, StripPath}, - worker::{to_raw_value, CLOUD_HOSTED}, + worker::to_raw_value, }; -use windmill_queue::{PushArgs, PushArgsOwned}; -const KEEP_LAST: i64 = 20; + pub fn workspaced_service() -> Router { Router::new() @@ -802,73 +801,7 @@ async fn get_capture_trigger_config_and_owner( )) } -async fn clear_captures_history(db: &DB, w_id: &str) -> Result<()> { - if *CLOUD_HOSTED { - /* Retain only KEEP_LAST most recent captures in this workspace. */ - sqlx::query!( - r#" - DELETE FROM - capture - WHERE - workspace_id = $1 - AND created_at <= ( - SELECT - created_at - FROM - capture - WHERE - workspace_id = $1 - ORDER BY - created_at DESC - OFFSET $2 - LIMIT 1 - ) - "#, - &w_id, - KEEP_LAST, - ) - .execute(db) - .await?; - } - Ok(()) -} - -pub async fn insert_capture_payload( - db: &DB, - w_id: &str, - path: &str, - is_flow: bool, - trigger_kind: &TriggerKind, - main_args: PushArgsOwned, - preprocessor_args: PushArgsOwned, - owner: &str, -) -> Result<()> { - sqlx::query!( - r#" - INSERT INTO - capture ( - workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by - ) - VALUES ( - $1, $2, $3, $4, $5, $6, $7 - ) - "#, - &w_id, - path, - is_flow, - trigger_kind as &TriggerKind, - SqlxJson(PushArgs { args: &main_args.args, extra: main_args.extra }) as SqlxJson, - SqlxJson(PushArgs { args: &preprocessor_args.args, extra: preprocessor_args.extra }) - as SqlxJson, - owner, - ) - .execute(db) - .await?; - - clear_captures_history(db, &w_id).await?; - - Ok(()) -} +pub use windmill_triggers::capture_ext::insert_capture_payload; async fn webhook_payload( Extension(db): Extension, diff --git a/backend/windmill-api/src/folders.rs b/backend/windmill-api/src/folders.rs index a73eb99b7e..fd3381c7d7 100644 --- a/backend/windmill-api/src/folders.rs +++ b/backend/windmill-api/src/folders.rs @@ -306,24 +306,8 @@ pub async fn is_owner_api( Ok(Json(is_owner(&authed, &name))) } -pub fn is_owner(ApiAuthed { is_admin, folders, .. }: &ApiAuthed, name: &str) -> bool { - if *is_admin { - true - } else { - folders.into_iter().any(|x| x.0 == name && x.2) - } -} - -pub fn require_is_owner(authed: &ApiAuthed, name: &str) -> Result<()> { - if is_owner(authed, name) { - Ok(()) - } else { - Err(windmill_common::error::Error::NotAuthorized(format!( - "You are not owner of the folder {}", - name - ))) - } -} +pub use windmill_api_auth::permissions::is_folder_owner as is_owner; +pub use windmill_api_auth::permissions::require_is_folder_owner as require_is_owner; async fn update_folder( authed: ApiAuthed, diff --git a/backend/windmill-api/src/job_ops_impl.rs b/backend/windmill-api/src/job_ops_impl.rs index 3979a59229..4ee53776b8 100644 --- a/backend/windmill-api/src/job_ops_impl.rs +++ b/backend/windmill-api/src/job_ops_impl.rs @@ -23,8 +23,9 @@ use windmill_common::{ utils::StripPath, DB, }; +use windmill_common::jobs::RunJobQuery; use windmill_queue::PushArgsOwned; -use windmill_triggers::jobs_ext::{JobOps, JobUpdateSSEStream, RunJobQuery}; +use windmill_triggers::jobs_ext::{JobOps, JobUpdateSSEStream}; pub struct JobOpsImpl; @@ -46,29 +47,8 @@ impl JobOps for JobOpsImpl { Option, Option>, )> { - // Convert RunJobQuery from windmill-triggers to the windmill-api version - let api_run_query = crate::jobs::RunJobQuery { - scheduled_for: run_query.scheduled_for, - scheduled_in_secs: run_query.scheduled_in_secs, - parent_job: run_query.parent_job, - root_job: run_query.root_job, - invisible_to_owner: run_query.invisible_to_owner, - queue_limit: run_query.queue_limit, - payload: run_query.payload, - job_id: run_query.job_id, - tag: run_query.tag, - timeout: run_query.timeout, - cache_ttl: run_query.cache_ttl, - cache_ignore_s3_path: run_query.cache_ignore_s3_path, - skip_preprocessor: run_query.skip_preprocessor, - poll_delay_ms: run_query.poll_delay_ms, - memory_id: run_query.memory_id, - trigger_external_id: run_query.trigger_external_id, - service_name: run_query.service_name, - suspended_mode: run_query.suspended_mode, - }; crate::jobs::push_script_job_by_path_into_queue( - authed, db, tx_o, user_db, w_id, script_path, api_run_query, args, trigger, + authed, db, tx_o, user_db, w_id, script_path, run_query, args, trigger, ) .await } @@ -89,28 +69,8 @@ impl JobOps for JobOpsImpl { Option, Option>, )> { - let api_run_query = crate::jobs::RunJobQuery { - scheduled_for: run_query.scheduled_for, - scheduled_in_secs: run_query.scheduled_in_secs, - parent_job: run_query.parent_job, - root_job: run_query.root_job, - invisible_to_owner: run_query.invisible_to_owner, - queue_limit: run_query.queue_limit, - payload: run_query.payload, - job_id: run_query.job_id, - tag: run_query.tag, - timeout: run_query.timeout, - cache_ttl: run_query.cache_ttl, - cache_ignore_s3_path: run_query.cache_ignore_s3_path, - skip_preprocessor: run_query.skip_preprocessor, - poll_delay_ms: run_query.poll_delay_ms, - memory_id: run_query.memory_id, - trigger_external_id: run_query.trigger_external_id, - service_name: run_query.service_name, - suspended_mode: run_query.suspended_mode, - }; crate::jobs::push_flow_job_by_path_into_queue( - authed, db, tx_o, user_db, w_id, flow_path, api_run_query, args, trigger, + authed, db, tx_o, user_db, w_id, flow_path, run_query, args, trigger, ) .await } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 15adf46caa..2feb6b2858 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -35,8 +35,8 @@ use windmill_common::error::JsonResult; use windmill_common::flow_conversations::add_message_to_conversation_tx; use windmill_common::flow_status::{JobResult, RestartedFrom}; use windmill_common::jobs::{ - check_tag_available_for_workspace_internal, format_completed_job_result, format_result, - DynamicInput, JobTriggerKind, RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE, + format_completed_job_result, format_result, DynamicInput, JobTriggerKind, + RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE, }; use windmill_common::runnable_settings::{ ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettings, @@ -1775,60 +1775,38 @@ pub struct ListableCompletedJob { pub args: Option, } -#[derive(Debug, Deserialize, Clone, Default)] -pub struct RunJobQuery { - pub scheduled_for: Option>, - pub scheduled_in_secs: Option, - pub parent_job: Option, - pub root_job: Option, - pub invisible_to_owner: Option, - pub queue_limit: Option, - pub payload: Option, - pub job_id: Option, - pub tag: Option, - pub timeout: Option, - pub cache_ttl: Option, - pub cache_ignore_s3_path: Option, - pub skip_preprocessor: Option, - pub poll_delay_ms: Option, - pub memory_id: Option, - pub trigger_external_id: Option, - pub service_name: Option, - pub suspended_mode: Option, +pub use windmill_common::jobs::RunJobQuery; + +async fn run_query_get_scheduled_for( + run_query: &RunJobQuery, + db: &DB, +) -> error::Result>> { + if let Some(scheduled_for) = run_query.scheduled_for { + Ok(Some(scheduled_for)) + } else if let Some(scheduled_in_secs) = run_query.scheduled_in_secs { + let now = now_from_db(db).await?; + Ok(Some( + now + chrono::Duration::try_seconds(scheduled_in_secs).unwrap_or_default(), + )) + } else { + Ok(None) + } } -impl RunJobQuery { - async fn get_scheduled_for<'c>( - &self, - db: &DB, - ) -> error::Result>> { - if let Some(scheduled_for) = self.scheduled_for { - Ok(Some(scheduled_for)) - } else if let Some(scheduled_in_secs) = self.scheduled_in_secs { - let now = now_from_db(db).await?; - Ok(Some( - now + chrono::Duration::try_seconds(scheduled_in_secs).unwrap_or_default(), - )) - } else { - Ok(None) - } - } +fn run_query_payload_as_args(run_query: &RunJobQuery) -> error::Result>> { + let payload_r = run_query.payload.clone().map(decode_payload).map(|x| { + x.map_err(|e| { + error::Error::internal_err(format!("Impossible to decode query payload: {e:#?}")) + }) + }); - fn payload_as_args(&self) -> error::Result>> { - let payload_r = self.payload.clone().map(decode_payload).map(|x| { - x.map_err(|e| { - error::Error::internal_err(format!("Impossible to decode query payload: {e:#?}")) - }) - }); + let payload_as_args = if let Some(payload) = payload_r { + payload? + } else { + HashMap::new() + }; - let payload_as_args = if let Some(payload) = payload_r { - payload? - } else { - HashMap::new() - }; - - Ok(payload_as_args) - } + Ok(payload_as_args) } #[derive(Deserialize, Clone)] @@ -3869,33 +3847,10 @@ pub fn add_raw_string( return args; } -pub async fn check_tag_available_for_workspace( - db: &DB, - w_id: &str, - tag: &Option, - authed: &ApiAuthed, -) -> error::Result<()> { - if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) { - let tags = get_scope_tags(authed); - check_tag_available_for_workspace_internal(&db, w_id, tag, &authed.email, tags).await - } else { - Ok(()) - } -} +pub use windmill_triggers::jobs_ext::check_tag_available_for_workspace; #[cfg(feature = "enterprise")] -pub async fn check_license_key_valid() -> error::Result<()> { - use windmill_common::ee_oss::LICENSE_KEY_VALID; - - let valid = *LICENSE_KEY_VALID.read().await; - if !valid { - return Err(Error::BadRequest( - "License key is not valid. Go to your superadmin settings to update your license key." - .to_string(), - )); - } - Ok(()) -} +pub use windmill_common::ee_oss::check_license_key_valid; use windmill_common::flows::InputTransform; @@ -4323,7 +4278,7 @@ pub async fn run_flow<'c>( let tag = run_query.tag.clone().or(tag); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let return_tx = tx_o.is_some(); @@ -4627,7 +4582,7 @@ pub async fn restart_flow( .map(|json| PushArgs { args: &json.0, extra: None }) .unwrap_or_else(|| PushArgs::from(&ehm)); - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into()); @@ -4775,7 +4730,7 @@ pub async fn push_script_job_by_path_into_queue<'c>( run_query.skip_preprocessor, ) .await?; - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tag = run_query.tag.clone().or(tag); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; @@ -4953,7 +4908,7 @@ pub async fn run_workflow_as_code( extra.insert(ENTRYPOINT_OVERRIDE.to_string(), to_raw_value(&entrypoint)); let args = PushArgs { args: &task.args.unwrap_or_else(HashMap::new), extra: Some(extra) }; - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tag = run_query.tag.clone().or(tag).or(Some(job.tag)); @@ -5324,27 +5279,7 @@ pub async fn run_wait_result( result_to_response(result, success) } -pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> { - sqlx::query!( - "UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1", - job_uuid, - ) - .execute(db) - .await?; - sqlx::query!( - "UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = $1", - job_uuid, - ) - .execute(db) - .await?; - sqlx::query!( - "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = $1", - job_uuid, - ) - .execute(db) - .await?; - Ok(()) -} +pub use windmill_common::jobs::delete_job_metadata_after_use; pub async fn check_queue_too_long(db: &DB, queue_limit: Option) -> error::Result<()> { if let Some(limit) = queue_limit { @@ -5475,7 +5410,7 @@ pub async fn run_wait_result_job_by_path_get( return Ok(Json(serde_json::json!("")).into_response()); } - let payload_as_args = run_query.payload_as_args()?; + let payload_as_args = run_query_payload_as_args(&run_query)?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; args.body = args::Body::HashMap(payload_as_args); @@ -5580,7 +5515,7 @@ pub async fn run_wait_result_flow_by_path_get( if method == http::Method::HEAD { return Ok(Json(serde_json::json!("")).into_response()); } - let payload_as_args = run_query.payload_as_args()?; + let payload_as_args = run_query_payload_as_args(&run_query)?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; args.body = args::Body::HashMap(payload_as_args); @@ -5958,7 +5893,7 @@ pub async fn stream_job( is_get: bool, ) -> error::Result { let args = if is_get { - let payload_as_args = run_query.payload_as_args()?; + let payload_as_args = run_query_payload_as_args(&run_query)?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; args.body = args::Body::HashMap(payload_as_args); @@ -6140,7 +6075,7 @@ pub async fn run_wait_result_flow_by_version_get( return Ok(Json(serde_json::json!("")).into_response()); } - let payload_as_args = run_query.payload_as_args()?; + let payload_as_args = run_query_payload_as_args(&run_query)?; let mut args = args.process_args(&authed, &db, &w_id, None).await?; args.body = args::Body::HashMap(payload_as_args); @@ -6243,7 +6178,7 @@ async fn run_preview_script( )); } require_path_read_access_for_preview(&authed, &preview.path)?; - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tag = run_query.tag.clone().or(preview.tag.clone()); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()); @@ -6456,7 +6391,7 @@ async fn run_bundle_preview_script( .and_then(|s| BundleFormat::from_string(&s)) .unwrap_or(BundleFormat::Cjs); - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tag = run_query.tag.clone().or(preview.tag.clone()); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; let ltx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()); @@ -7063,7 +6998,7 @@ async fn run_preview_flow_job( )); } require_path_read_access_for_preview(&authed, &raw_flow.path)?; - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tag = run_query.tag.clone().or(raw_flow.tag.clone()); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()); @@ -7263,7 +7198,7 @@ async fn run_dynamic_select( } } - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()); let (uuid, tx) = push( @@ -7387,7 +7322,7 @@ pub async fn run_job_by_hash_inner( cache_ttl = Some(run_query_cache_ttl); cache_ignore_s3_path = run_query.cache_ignore_s3_path; } - let scheduled_for = run_query.get_scheduled_for(&db).await?; + let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?; let tag = run_query.tag.clone().or(tag); check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?; diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 7ffc40ef5c..35a1e4a10c 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -74,7 +74,7 @@ pub mod agent_workers_ee; mod agent_workers_oss; mod ai; mod apps; -pub mod args; +pub use windmill_triggers::args_ext as args; mod assets; mod audit; pub mod auth; diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 0b9aa5ce24..738527280b 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -8,7 +8,7 @@ #![allow(non_snake_case)] -use sqlx::{PgConnection, Postgres, Transaction}; +use sqlx::{Postgres, Transaction}; use std::sync::atomic::AtomicBool; use std::sync::Arc; @@ -161,19 +161,7 @@ pub async fn maybe_refresh_folders( } } -pub fn get_scope_tags(authed: &ApiAuthed) -> Option> { - authed.scopes.as_ref()?.iter().find_map(|s| { - if s.starts_with("if_jobs:filter_tags:") { - Some( - s.trim_start_matches("if_jobs:filter_tags:") - .split(",") - .collect::>(), - ) - } else { - None - } - }) -} +pub use windmill_api_auth::scopes::get_scope_tags; #[derive(Clone, Debug)] pub struct OptAuthed(pub Option); @@ -319,27 +307,7 @@ pub struct TruncatedToken { pub scopes: Option>, } -#[derive(Deserialize)] -pub struct NewToken { - pub label: Option, - pub expiration: Option>, - pub impersonate_email: Option, - pub scopes: Option>, - pub workspace_id: Option, -} - -#[cfg(feature = "native_trigger")] -impl NewToken { - pub fn new( - label: Option, - expiration: Option>, - impersonate_email: Option, - scopes: Option>, - workspace_id: Option, - ) -> NewToken { - NewToken { label, expiration, impersonate_email, scopes, workspace_id } - } -} +pub use windmill_api_auth::tokens::NewToken; #[derive(Deserialize)] pub struct Login { @@ -852,35 +820,7 @@ pub async fn is_owner_of_path( } } -pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) -> Result<()> { - if authed.is_admin { - return Ok(()); - } - if !path.is_empty() { - let splitted = path.split("/").collect::>(); - if splitted[0] == "u" { - if splitted[1] == authed.username { - Ok(()) - } else { - Err(Error::BadRequest(format!( - "only the owner {} is authorized to perform this operation", - splitted[1] - ))) - } - } else if splitted[0] == "f" { - crate::folders::require_is_owner(authed, splitted[1]) - } else { - Err(Error::BadRequest(format!( - "Not recognized path kind: {}", - path - ))) - } - } else { - Err(Error::BadRequest(format!( - "Cannot be owner of an empty path" - ))) - } -} +pub use windmill_api_auth::permissions::require_owner_of_path; /// Checks that a user has at least read access to the path for preview jobs. /// This prevents privilege escalation where a user could run preview code @@ -939,95 +879,7 @@ pub fn require_path_read_access_for_preview( } } -pub fn get_perm_in_extra_perms_for_authed( - v: serde_json::Value, - authed: &ApiAuthed, -) -> Option { - match v { - serde_json::Value::Object(obj) => { - let mut keys = vec![format!("u/{}", authed.username)]; - for g in authed.groups.iter() { - keys.push(format!("g/{}", g)); - } - let mut res = None; - for k in keys { - if let Some(v) = obj.get(&k) { - if let Some(v) = v.as_bool() { - if v { - return Some(true); - } - res = Some(v); - } - } - } - res - } - _ => None, - } -} - -pub async fn require_is_writer( - authed: &ApiAuthed, - path: &str, - w_id: &str, - db: DB, - query: &str, - kind: &str, -) -> Result<()> { - if authed.is_admin { - return Ok(()); - } - if !path.is_empty() { - if require_owner_of_path(authed, path).is_ok() { - return Ok(()); - } - if path.starts_with("f/") && path.split('/').count() >= 2 { - let folder = path.split('/').nth(1).unwrap(); - let extra_perms = sqlx::query_scalar!( - "SELECT extra_perms FROM folder WHERE name = $1 AND workspace_id = $2", - folder, - w_id - ) - .fetch_optional(&db) - .await?; - if let Some(perms) = extra_perms { - let is_folder_writer = - get_perm_in_extra_perms_for_authed(perms, authed).unwrap_or(false); - if is_folder_writer { - return Ok(()); - } - } - } - let extra_perms = sqlx::query_scalar(query) - .bind(path) - .bind(w_id) - .fetch_optional(&db) - .await?; - if let Some(perms) = extra_perms { - let perm = get_perm_in_extra_perms_for_authed(perms, authed); - match perm { - Some(true) => Ok(()), - Some(false) => Err(Error::BadRequest(format!( - "User {} is not a writer of {kind} path {path}", - authed.username - ))), - None => Err(Error::BadRequest(format!( - "User {} has neither read or write permission on {kind} {path}", - authed.username - ))), - } - } else { - Err(Error::BadRequest(format!( - "{path} does not exist yet and user {} is not an owner of the parent folder", - authed.username - ))) - } - } else { - Err(Error::BadRequest(format!( - "Cannot be writer of an empty path" - ))) - } -} +pub use windmill_api_auth::permissions::require_is_writer; async fn whois( Extension(db): Extension, Path((w_id, username)): Path<(String, String)>, @@ -2107,62 +1959,7 @@ pub async fn create_session_token<'c>( Ok(token) } -pub async fn create_token_internal( - tx: &mut PgConnection, - db: &DB, - authed: &ApiAuthed, - token_config: NewToken, -) -> Result { - let token = rd_string(32); - - let is_super_admin = sqlx::query_scalar!( - "SELECT super_admin FROM password WHERE email = $1", - authed.email - ) - .fetch_optional(&mut *tx) - .await? - .unwrap_or(false); - if *CLOUD_HOSTED { - let nb_tokens = - sqlx::query_scalar!("SELECT COUNT(*) FROM token WHERE email = $1", &authed.email) - .fetch_one(db) - .await?; - if nb_tokens.unwrap_or(0) >= 10000 { - return Err(Error::BadRequest( - "You have reached the maximum number of tokens (10000) on cloud. Contact support@windmill.dev to increase the limit" - .to_string(), - )); - } - } - sqlx::query!( - "INSERT INTO token - (token, email, label, expiration, super_admin, scopes, workspace_id) - VALUES ($1, $2, $3, $4, $5, $6, $7)", - token, - authed.email, - token_config.label, - token_config.expiration, - is_super_admin, - token_config.scopes.as_ref().map(|x| x.as_slice()), - token_config.workspace_id, - ) - .execute(&mut *tx) - .await?; - - audit_log( - &mut *tx, - authed, - "users.token.create", - ActionKind::Create, - &"global", - Some(&token[0..10]), - None, - ) - .instrument(tracing::info_span!("token", email = &authed.email)) - .await?; - - Ok(token) -} +pub use windmill_api_auth::tokens::create_token_internal; async fn create_token( Extension(db): Extension, diff --git a/backend/windmill-common/src/ee_oss.rs b/backend/windmill-common/src/ee_oss.rs index 51b1efd2e2..04ecdd210e 100644 --- a/backend/windmill-common/src/ee_oss.rs +++ b/backend/windmill-common/src/ee_oss.rs @@ -122,3 +122,15 @@ pub async fn low_disk_alerts( ) { // Implementation is not open source } + +#[cfg(all(feature = "enterprise", not(feature = "private")))] +pub async fn check_license_key_valid() -> error::Result<()> { + let valid = *LICENSE_KEY_VALID.read().await; + if !valid { + return Err(error::Error::BadRequest( + "License key is not valid. Go to your superadmin settings to update your license key." + .to_string(), + )); + } + Ok(()) +} diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index b1497c3d3e..fa2dc1813c 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -917,3 +917,47 @@ pub struct WorkerInternalServerInlineUtils { // The server cannot call the worker functions directly because they are independent crates pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell = OnceCell::new(); + +#[derive(Debug, Deserialize, Clone, Default)] +pub struct RunJobQuery { + pub scheduled_for: Option>, + pub scheduled_in_secs: Option, + pub parent_job: Option, + pub root_job: Option, + pub invisible_to_owner: Option, + pub queue_limit: Option, + pub payload: Option, + pub job_id: Option, + pub tag: Option, + pub timeout: Option, + pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, + pub skip_preprocessor: Option, + pub poll_delay_ms: Option, + pub memory_id: Option, + pub trigger_external_id: Option, + pub service_name: Option, + pub suspended_mode: Option, +} + +pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> { + sqlx::query!( + "UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1", + job_uuid, + ) + .execute(db) + .await?; + sqlx::query!( + "UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = $1", + job_uuid, + ) + .execute(db) + .await?; + sqlx::query!( + "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = $1", + job_uuid, + ) + .execute(db) + .await?; + Ok(()) +} diff --git a/backend/windmill-triggers/src/flow_ext.rs b/backend/windmill-triggers/src/flow_ext.rs index 67126913f2..03a8e98c7b 100644 --- a/backend/windmill-triggers/src/flow_ext.rs +++ b/backend/windmill-triggers/src/flow_ext.rs @@ -9,10 +9,8 @@ use windmill_api_auth::ApiAuthed; use windmill_common::{error::Result, DB}; -use crate::require_is_writer_internal; - pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> { - require_is_writer_internal( + windmill_api_auth::permissions::require_is_writer( authed, path, w_id, diff --git a/backend/windmill-triggers/src/jobs_ext.rs b/backend/windmill-triggers/src/jobs_ext.rs index 7fc9b9b380..a6239114f8 100644 --- a/backend/windmill-triggers/src/jobs_ext.rs +++ b/backend/windmill-triggers/src/jobs_ext.rs @@ -13,7 +13,6 @@ //! implementation at startup via `set_ops()`. use axum::response::Response; -use serde::Deserialize; use serde_json::value::RawValue; use sqlx::Postgres; use std::sync::Arc; @@ -21,99 +20,21 @@ use uuid::Uuid; use windmill_api_auth::{ApiAuthed, OptTokened}; use windmill_common::{ db::UserDB, - error::{self, Error}, - jobs::check_tag_available_for_workspace_internal, + error, triggers::TriggerMetadata, utils::StripPath, DB, }; use windmill_queue::PushArgsOwned; -#[derive(Debug, Deserialize, Clone, Default)] -pub struct RunJobQuery { - pub scheduled_for: Option>, - pub scheduled_in_secs: Option, - pub parent_job: Option, - pub root_job: Option, - pub invisible_to_owner: Option, - pub queue_limit: Option, - pub payload: Option, - pub job_id: Option, - pub tag: Option, - pub timeout: Option, - pub cache_ttl: Option, - pub cache_ignore_s3_path: Option, - pub skip_preprocessor: Option, - pub poll_delay_ms: Option, - pub memory_id: Option, - pub trigger_external_id: Option, - pub service_name: Option, - pub suspended_mode: Option, -} - -pub fn get_scope_tags(authed: &ApiAuthed) -> Option> { - authed.scopes.as_ref()?.iter().find_map(|s| { - if s.starts_with("if_jobs:filter_tags:") { - Some( - s.trim_start_matches("if_jobs:filter_tags:") - .split(",") - .collect::>(), - ) - } else { - None - } - }) -} - -pub async fn check_tag_available_for_workspace( - db: &DB, - w_id: &str, - tag: &Option, - authed: &ApiAuthed, -) -> error::Result<()> { - if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) { - let tags = get_scope_tags(authed); - check_tag_available_for_workspace_internal(db, w_id, tag, &authed.email, tags).await - } else { - Ok(()) - } -} - -pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> { - sqlx::query!( - "UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1", - job_uuid, - ) - .execute(db) - .await?; - sqlx::query!( - "UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = $1", - job_uuid, - ) - .execute(db) - .await?; - sqlx::query!( - "UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = $1", - job_uuid, - ) - .execute(db) - .await?; - Ok(()) -} +// Re-export shared types/functions from windmill-common +pub use windmill_common::jobs::{delete_job_metadata_after_use, RunJobQuery}; #[cfg(feature = "enterprise")] -pub async fn check_license_key_valid() -> error::Result<()> { - use windmill_common::ee_oss::LICENSE_KEY_VALID; +pub use windmill_common::ee_oss::check_license_key_valid; - let valid = *LICENSE_KEY_VALID.read().await; - if !valid { - return Err(Error::BadRequest( - "License key is not valid. Go to your superadmin settings to update your license key." - .to_string(), - )); - } - Ok(()) -} +// Re-export scope helpers from windmill-api-auth +pub use windmill_api_auth::scopes::{check_tag_available_for_workspace, get_scope_tags}; /// Trait for complex job operations that stay in windmill-api. /// windmill-api provides the implementation at startup via `set_ops()`. @@ -236,21 +157,6 @@ pub trait JobOps: Send + Sync + 'static { ) -> error::Result<(Option, Option)>; } -// Re-export types used by SSE streaming -#[derive(Debug, Clone, serde::Serialize)] -pub struct JobUpdate { - #[serde(skip_serializing_if = "Option::is_none")] - pub running: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub completed: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub new_logs: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub log_offset: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub mem_peak: Option, -} - #[derive(Debug, serde::Serialize)] pub enum JobUpdateSSEStream { Update(serde_json::Value), diff --git a/backend/windmill-triggers/src/lib.rs b/backend/windmill-triggers/src/lib.rs index b591c7a1e3..d73490e065 100644 --- a/backend/windmill-triggers/src/lib.rs +++ b/backend/windmill-triggers/src/lib.rs @@ -17,154 +17,3 @@ pub mod jobs_ext; pub mod resource_ext; pub mod script_ext; pub mod user_ext; - -// Shared permission helpers used by flow_ext, script_ext, and native_triggers - -use windmill_api_auth::ApiAuthed; -use windmill_common::error::{Error, Result}; -use windmill_common::DB; - -/// Check if the user is an owner of the given path. -pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) -> Result<()> { - if authed.is_admin { - return Ok(()); - } - if !path.is_empty() { - let splitted = path.split("/").collect::>(); - if splitted[0] == "u" { - if splitted[1] == authed.username { - Ok(()) - } else { - Err(Error::BadRequest(format!( - "only the owner {} is authorized to perform this operation", - splitted[1] - ))) - } - } else if splitted[0] == "f" { - require_is_folder_owner(authed, splitted[1]) - } else { - Err(Error::BadRequest(format!( - "Not recognized path kind: {}", - path - ))) - } - } else { - Err(Error::BadRequest(format!( - "Cannot be owner of an empty path" - ))) - } -} - -fn is_folder_owner( - ApiAuthed { is_admin, folders, .. }: &ApiAuthed, - name: &str, -) -> bool { - if *is_admin { - true - } else { - folders.into_iter().any(|x| x.0 == name && x.2) - } -} - -fn require_is_folder_owner(authed: &ApiAuthed, name: &str) -> Result<()> { - if is_folder_owner(authed, name) { - Ok(()) - } else { - Err(Error::NotAuthorized(format!( - "You are not owner of the folder {}", - name - ))) - } -} - -fn get_perm_in_extra_perms_for_authed( - v: serde_json::Value, - authed: &ApiAuthed, -) -> Option { - match v { - serde_json::Value::Object(obj) => { - let mut keys = vec![format!("u/{}", authed.username)]; - for g in authed.groups.iter() { - keys.push(format!("g/{}", g)); - } - let mut res = None; - for k in keys { - if let Some(v) = obj.get(&k) { - if let Some(v) = v.as_bool() { - if v { - return Some(true); - } - res = Some(v); - } - } - } - res - } - _ => None, - } -} - -/// Generic require_is_writer with a configurable SQL query. -/// Used by flow_ext and script_ext. -pub async fn require_is_writer_internal( - authed: &ApiAuthed, - path: &str, - w_id: &str, - db: DB, - query: &str, - kind: &str, -) -> Result<()> { - if authed.is_admin { - return Ok(()); - } - if !path.is_empty() { - if require_owner_of_path(authed, path).is_ok() { - return Ok(()); - } - if path.starts_with("f/") && path.split('/').count() >= 2 { - let folder = path.split('/').nth(1).unwrap(); - let extra_perms = sqlx::query_scalar!( - "SELECT extra_perms FROM folder WHERE name = $1 AND workspace_id = $2", - folder, - w_id - ) - .fetch_optional(&db) - .await?; - if let Some(perms) = extra_perms { - let is_folder_writer = - get_perm_in_extra_perms_for_authed(perms, authed).unwrap_or(false); - if is_folder_writer { - return Ok(()); - } - } - } - let extra_perms = sqlx::query_scalar(query) - .bind(path) - .bind(w_id) - .fetch_optional(&db) - .await?; - if let Some(perms) = extra_perms { - let perm = get_perm_in_extra_perms_for_authed(perms, authed); - match perm { - Some(true) => Ok(()), - Some(false) => Err(Error::BadRequest(format!( - "User {} is not a writer of {kind} path {path}", - authed.username - ))), - None => Err(Error::BadRequest(format!( - "User {} has neither read or write permission on {kind} {path}", - authed.username - ))), - } - } else { - Err(Error::BadRequest(format!( - "{path} does not exist yet and user {} is not an owner of the parent folder", - authed.username - ))) - } - } else { - Err(Error::BadRequest(format!( - "Cannot be writer of an empty path" - ))) - } -} diff --git a/backend/windmill-triggers/src/script_ext.rs b/backend/windmill-triggers/src/script_ext.rs index 04915e340c..498b6f0365 100644 --- a/backend/windmill-triggers/src/script_ext.rs +++ b/backend/windmill-triggers/src/script_ext.rs @@ -9,10 +9,8 @@ use windmill_api_auth::ApiAuthed; use windmill_common::{error::Result, DB}; -use crate::require_is_writer_internal; - pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> { - require_is_writer_internal( + windmill_api_auth::permissions::require_is_writer( authed, path, w_id, diff --git a/backend/windmill-triggers/src/user_ext.rs b/backend/windmill-triggers/src/user_ext.rs index 2596cb2800..4ea57dd9c5 100644 --- a/backend/windmill-triggers/src/user_ext.rs +++ b/backend/windmill-triggers/src/user_ext.rs @@ -6,91 +6,4 @@ * LICENSE-AGPL for a copy of the license. */ -use sqlx::PgConnection; -use tracing::Instrument; -use windmill_api_auth::ApiAuthed; -use windmill_audit::{audit_oss::audit_log, ActionKind}; -use windmill_common::{ - error::{Error, Result}, - utils::rd_string, - worker::CLOUD_HOSTED, - DB, -}; - -#[derive(serde::Deserialize)] -pub struct NewToken { - pub label: Option, - pub expiration: Option>, - pub impersonate_email: Option, - pub scopes: Option>, - pub workspace_id: Option, -} - -impl NewToken { - pub fn new( - label: Option, - expiration: Option>, - impersonate_email: Option, - scopes: Option>, - workspace_id: Option, - ) -> NewToken { - NewToken { label, expiration, impersonate_email, scopes, workspace_id } - } -} - -pub async fn create_token_internal( - tx: &mut PgConnection, - db: &DB, - authed: &ApiAuthed, - token_config: NewToken, -) -> Result { - let token = rd_string(32); - - let is_super_admin = sqlx::query_scalar!( - "SELECT super_admin FROM password WHERE email = $1", - authed.email - ) - .fetch_optional(&mut *tx) - .await? - .unwrap_or(false); - if *CLOUD_HOSTED { - let nb_tokens = - sqlx::query_scalar!("SELECT COUNT(*) FROM token WHERE email = $1", &authed.email) - .fetch_one(db) - .await?; - if nb_tokens.unwrap_or(0) >= 10000 { - return Err(Error::BadRequest( - "You have reached the maximum number of tokens (10000) on cloud. Contact support@windmill.dev to increase the limit" - .to_string(), - )); - } - } - sqlx::query!( - "INSERT INTO token - (token, email, label, expiration, super_admin, scopes, workspace_id) - VALUES ($1, $2, $3, $4, $5, $6, $7)", - token, - authed.email, - token_config.label, - token_config.expiration, - is_super_admin, - token_config.scopes.as_ref().map(|x| x.as_slice()), - token_config.workspace_id, - ) - .execute(&mut *tx) - .await?; - - audit_log( - &mut *tx, - authed, - "users.token.create", - ActionKind::Create, - &"global", - Some(&token[0..10]), - None, - ) - .instrument(tracing::info_span!("token", email = &authed.email)) - .await?; - - Ok(token) -} +pub use windmill_api_auth::tokens::{create_token_internal, NewToken};