diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index bba0cb2ffc..ce1e681c92 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -11,6 +11,7 @@ use crate::{ users::{check_scopes, require_owner_of_path, Authed, OptAuthed}, utils::require_super_admin, variables::get_workspace_key, + workers::{CUSTOM_TAGS, CUSTOM_TAGS_PER_WORKSPACE}, BASE_URL, }; use anyhow::Context; @@ -1403,6 +1404,33 @@ fn add_raw_string( return args; } +fn check_tag_available_for_workspace(w_id: &str, tag: &Option) -> error::Result<()> { + if let Some(tag) = tag { + if tag == "" { + return Ok(()); + } + let custom_tags_per_w = &*CUSTOM_TAGS_PER_WORKSPACE; + if custom_tags_per_w.0.contains(&tag.to_string()) { + Ok(()) + } else if custom_tags_per_w.1.contains_key(tag) + && custom_tags_per_w + .1 + .get(tag) + .unwrap() + .contains(&w_id.to_string()) + { + Ok(()) + } else { + return Err(error::Error::BadRequest(format!( + "Tag {tag} cannot be used on workspace {w_id}: (CUSTOM_TAGS: {:?})", + *CUSTOM_TAGS + ))); + } + } else { + Ok(()) + } +} + pub async fn run_flow_by_path( authed: Authed, Extension(user_db): Extension, @@ -1424,6 +1452,7 @@ pub async fn run_flow_by_path( .fetch_optional(&mut tx) .await? .flatten(); + check_tag_available_for_workspace(&w_id, &tag)?; let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); @@ -1469,6 +1498,7 @@ pub async fn run_job_by_path( let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); + check_tag_available_for_workspace(&w_id, &tag)?; let (uuid, tx) = push( tx, @@ -1669,6 +1699,7 @@ pub async fn run_wait_result_job_by_path_get( let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); let (job_payload, tag) = script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?; + check_tag_available_for_workspace(&w_id, &tag)?; let (uuid, tx) = push( tx, @@ -1798,6 +1829,7 @@ async fn run_wait_result_script_by_path_internal( let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); + check_tag_available_for_workspace(&w_id, &tag)?; let (uuid, tx) = push( tx, @@ -1851,6 +1883,7 @@ pub async fn run_wait_result_script_by_hash( let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); + check_tag_available_for_workspace(&w_id, &tag)?; let (uuid, tx) = push( tx, @@ -1951,6 +1984,15 @@ async fn run_wait_result_flow_by_path_internal( let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); + let tag = sqlx::query_scalar!( + "SELECT tag from flow WHERE path = $1 and workspace_id = $2", + flow_path, + w_id + ) + .fetch_optional(&mut tx) + .await? + .flatten(); + check_tag_available_for_workspace(&w_id, &tag)?; let (uuid, tx) = push( tx, @@ -1969,7 +2011,7 @@ async fn run_wait_result_flow_by_path_internal( false, None, !run_query.invisible_to_owner.unwrap_or(false), - None, + tag, ) .await?; tx.commit().await?; @@ -2002,6 +2044,7 @@ async fn run_preview_job( let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default()); + check_tag_available_for_workspace(&w_id, &preview.tag)?; let (uuid, tx) = push( tx, @@ -2096,6 +2139,7 @@ async fn run_preview_flow_job( let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default()); + check_tag_available_for_workspace(&w_id, &raw_flow.tag)?; let (uuid, tx) = push( tx, @@ -2140,6 +2184,7 @@ pub async fn run_job_by_hash( let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let args = add_raw_string(raw_string, args); + check_tag_available_for_workspace(&w_id, &tag)?; let (uuid, tx) = push( tx, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index a34c270e75..dff743fcf6 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -9,6 +9,7 @@ use crate::oauth2::AllClients; use crate::saml::{SamlSsoLogin, ServiceProviderExt}; use crate::scim::has_scim_token; +use crate::tracing_init::MyOnFailure; use crate::{ db::UserDB, oauth2::{build_oauth_clients, SlackVerifier}, @@ -16,6 +17,7 @@ use crate::{ users::{Authed, OptAuthed}, webhook_util::WebhookShared, }; +use anyhow::Context; use argon2::Argon2; use axum::extract::DefaultBodyLimit; use axum::{middleware::from_extractor, routing::get, Extension, Router}; @@ -160,7 +162,8 @@ pub async fn run_server( TraceLayer::new_for_http() .on_response(MyOnResponse {}) .make_span_with(MyMakeSpan {}) - .on_request(()), + .on_request(()) + .on_failure(MyOnFailure {}), ) .layer(Extension(db.clone())) .layer(Extension(rsmq)) @@ -296,7 +299,8 @@ async fn is_up_to_date() -> Result { let version = HTTP_CLIENT .get("https://api.github.com/repos/windmill-labs/windmill/releases/latest") .send() - .await? + .await + .context("Impossible to reach api.github")? .json::() .await? .get("tag_name") diff --git a/backend/windmill-api/src/scim.rs b/backend/windmill-api/src/scim.rs index d3c725eb1f..a1285d3aae 100644 --- a/backend/windmill-api/src/scim.rs +++ b/backend/windmill-api/src/scim.rs @@ -1,3 +1,5 @@ +#![allow(non_snake_case)] + /* * Author: Ruben Fiszel * Copyright: Windmill Labs, Inc 2023 @@ -7,11 +9,11 @@ */ use axum::{ - extract::{Path, Query}, + extract::Query, middleware::Next, response::{IntoResponse, Response}, - routing::{get, post}, - Extension, Json, Router, + routing::get, + Extension, Router, }; use bytes::{BufMut, BytesMut}; use hyper::{header, http::HeaderValue, Request, StatusCode}; @@ -19,10 +21,16 @@ use mime_guess::mime; use serde::{Deserialize, Serialize}; use serde_json::json; use sql_builder::SqlBuilder; -use windmill_common::{ - error::{Error, Result}, - utils::not_found_if_none, -}; +use windmill_common::error::{Error, Result}; + +#[cfg(feature = "enterprise")] +use axum::{extract::Path, Json}; + +#[cfg(feature = "enterprise")] +use windmill_common::utils::not_found_if_none; + +#[cfg(feature = "enterprise")] +use axum::routing::post; use crate::db::DB; @@ -78,8 +86,6 @@ pub async fn has_scim_token(request: Request, next: Next) -> Response { .into_response(); } -pub type JsonScimResult = std::result::Result, Error>; - impl IntoResponse for JsonScim where T: Serialize, @@ -170,11 +176,12 @@ pub async fn get_users( )) } +#[cfg(feature = "enterprise")] #[derive(Deserialize, Debug)] pub struct CreateUser { userName: String, } -// #[cfg(feature = "enterprise")] +#[cfg(feature = "enterprise")] pub async fn create_user( Extension(db): Extension, Json(body): Json, @@ -382,6 +389,7 @@ pub async fn update_group( } } +#[cfg(feature = "enterprise")] pub async fn delete_group(Extension(db): Extension, Path(id): Path) -> Result<()> { tracing::info!("SCIM delete group: {:?}", id); sqlx::query!("DELETE FROM email_to_igroup WHERE igroup = $1", id) @@ -393,6 +401,7 @@ pub async fn delete_group(Extension(db): Extension, Path(id): Path) Ok(()) } +#[cfg(feature = "enterprise")] fn convert_name(name: &str) -> String { name.replace(" ", "_").to_lowercase() } diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index a7d177d8a1..72ca4901db 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -45,13 +45,6 @@ use windmill_common::{ }; use windmill_queue::{self, schedule::push_scheduled_job, QueueTransaction}; -lazy_static::lazy_static! { - pub static ref CUSTOM_TAGS: Vec = std::env::var("CUSTOM_TAGS") - .ok() - .map(|x| x.split(',').map(|x| x.to_string()).filter(|x| !x.is_empty()).collect::>()).unwrap_or_default(); - -} - const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20; #[derive(Serialize, sqlx::FromRow)] diff --git a/backend/windmill-api/src/tracing_init.rs b/backend/windmill-api/src/tracing_init.rs index 951408c10f..b30fa84499 100644 --- a/backend/windmill-api/src/tracing_init.rs +++ b/backend/windmill-api/src/tracing_init.rs @@ -8,7 +8,7 @@ use ::tracing::{field, Span}; use hyper::Response; -use tower_http::trace::{MakeSpan, OnResponse}; +use tower_http::trace::{MakeSpan, OnFailure, OnResponse}; #[derive(Clone)] pub struct MyOnResponse {} @@ -28,6 +28,14 @@ impl OnResponse for MyOnResponse { } } +#[derive(Clone)] +pub struct MyOnFailure {} + +impl OnFailure for MyOnFailure { + fn on_failure(&mut self, _b: B, _latency: std::time::Duration, _span: &tracing::Span) { + // tracing::error!(latency = latency.as_millis(), "response") + } +} #[derive(Clone)] pub struct MyMakeSpan {} diff --git a/backend/windmill-api/src/workers.rs b/backend/windmill-api/src/workers.rs index d2b0261dd4..4ab10815ea 100644 --- a/backend/windmill-api/src/workers.rs +++ b/backend/windmill-api/src/workers.rs @@ -13,6 +13,8 @@ use axum::{ Json, Router, }; +use itertools::Itertools; +use regex::Regex; use serde::{Deserialize, Serialize}; use sqlx::FromRow; use windmill_common::{ @@ -20,6 +22,7 @@ use windmill_common::{ utils::{paginate, Pagination}, }; +use std::collections::HashMap; #[cfg(feature = "benchmark")] use std::sync::atomic::Ordering; #[cfg(feature = "benchmark")] @@ -45,6 +48,31 @@ lazy_static::lazy_static! { .ok() .map(|x| x.split(',').map(|x| x.to_string()).collect::>()).unwrap_or_default(); + pub static ref CUSTOM_TAGS_PER_WORKSPACE: (Vec, HashMap>) = process_custom_tags(std::env::var("CUSTOM_TAGS") + .ok()); + + pub static ref ALL_TAGS: Vec = [CUSTOM_TAGS_PER_WORKSPACE.0.clone(), CUSTOM_TAGS_PER_WORKSPACE.1.keys().map(|x| x.to_string()).collect_vec()].concat(); + +} + +fn process_custom_tags(o: Option) -> (Vec, HashMap>) { + let regex = Regex::new(r"^(\w+)\(((?:\w+)\+?)+\)$").unwrap(); + if let Some(s) = o { + let mut global = vec![]; + let mut specific: HashMap> = HashMap::new(); + for e in s.split(",") { + if let Some(cap) = regex.captures(e) { + let tag = cap.get(1).unwrap().as_str().to_string(); + let workspaces = cap.get(2).unwrap().as_str().split("+"); + specific.insert(tag, workspaces.map(|x| x.to_string()).collect_vec()); + } else { + global.push(e.to_string()); + } + } + (global, specific) + } else { + (vec![], HashMap::new()) + } } #[derive(FromRow, Serialize, Deserialize)] @@ -91,5 +119,5 @@ async fn toggle(Query(query): Query) -> JsonResult { } async fn get_custom_tags() -> Json> { - Json(CUSTOM_TAGS.clone()) + Json(ALL_TAGS.clone()) } diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index 7fa890f5be..3f1c8ef56a 100644 --- a/backend/windmill-common/src/error.rs +++ b/backend/windmill-common/src/error.rs @@ -115,11 +115,13 @@ pub struct AppError(anyhow::Error); // Tell axum how to convert `AppError` into a response. impl IntoResponse for AppError { fn into_response(self) -> Response { - ( - StatusCode::INTERNAL_SERVER_ERROR, - format!("Something went wrong: {}", self.0), - ) - .into_response() + let body = body::boxed(body::Full::from(self.0.to_string())); + tracing::error!(error = self.0.to_string()); + axum::response::Response::builder() + .header("Content-Type", "text/plain") + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(body) + .unwrap() } } diff --git a/backend/windmill-queue/src/queue_transaction.rs b/backend/windmill-queue/src/queue_transaction.rs index 01bd021807..5dc137b0dd 100644 --- a/backend/windmill-queue/src/queue_transaction.rs +++ b/backend/windmill-queue/src/queue_transaction.rs @@ -1,4 +1,4 @@ -use std::{fmt::{Debug}}; +use std::fmt::Debug; use futures_core::{future::BoxFuture, stream::BoxStream}; use rsmq_async::{RedisBytes, RsmqConnection};