From 71e915a86c4f0661bad7a0f977b015045f739d8a Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 17 Feb 2024 16:46:02 +0100 Subject: [PATCH] feat: scim token and saml metadata setting in UI directly --- backend/ee-repo-ref.txt | 2 +- backend/src/ee.rs | 54 +- backend/src/main.rs | 14 +- backend/src/monitor.rs | 21 +- backend/windmill-api/openapi.yaml | 21 + backend/windmill-api/src/ee.rs | 41 +- backend/windmill-api/src/job_helpers_ee.rs | 1454 ++++++++++++++++- backend/windmill-api/src/lib.rs | 3 + backend/windmill-api/src/oauth2_ee.rs | 1256 +++++++++++++- backend/windmill-api/src/saml_ee.rs | 189 ++- backend/windmill-api/src/scim_ee.rs | 718 +++++++- backend/windmill-api/src/stripe_ee.rs | 303 +++- backend/windmill-audit/src/audit_ee.rs | 121 +- backend/windmill-common/src/ee.rs | 11 +- .../windmill-common/src/global_settings.rs | 2 +- backend/windmill-git-sync/src/git_sync_ee.rs | 411 ++++- .../lib/components/InstanceSettings.svelte | 13 + .../src/lib/components/instanceSettings.ts | 20 + 18 files changed, 4529 insertions(+), 125 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 01edafce8b..96c76565f2 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -8ed0d6aa6218ec4f9011a60db701295e5c3a1e4a \ No newline at end of file +23eaf9ef7fd5b2d115c0db2aa12a145e35b16f33 \ No newline at end of file diff --git a/backend/src/ee.rs b/backend/src/ee.rs index ef944b984b..ec4a8a70e7 100644 --- a/backend/src/ee.rs +++ b/backend/src/ee.rs @@ -1,16 +1,50 @@ -use anyhow::anyhow; #[cfg(feature = "enterprise")] -use windmill_common::error::{Error, Result}; +use windmill_common::error; -pub async fn set_license_key(_license_key: String) -> anyhow::Result<()> { - // Implementation is not open source - Err(anyhow!("License cannot be set in Windmill CE")) +pub async fn set_license_key(license_key: String) -> anyhow::Result<()> { + use windmill_api::ee::validate_license_key; + use windmill_common::ee::{LICENSE_KEY, LICENSE_KEY_ID, LICENSE_KEY_VALID}; + + let id = validate_license_key(license_key.clone()).await?; + { + let mut l = LICENSE_KEY_ID.write().await; + *l = id.to_string() + } + + { + let mut l = LICENSE_KEY.write().await; + *l = license_key + } + { + let mut l = LICENSE_KEY_VALID.write().await; + *l = true + } + + Ok(()) } #[cfg(feature = "enterprise")] -pub async fn verify_license_key() -> Result<()> { - // Implementation is not open source - Err(Error::InternalErr( - "License always invalid in Windmill CE".to_string(), - )) +pub async fn verify_license_key() -> error::Result<()> { + use windmill_common::ee::{LICENSE_KEY, LICENSE_KEY_VALID}; + use windmill_common::error::to_anyhow; + + let expiry_nb = LICENSE_KEY + .read() + .await + .clone() + .split(".") + .nth(1) + .unwrap_or_else(|| "") + .parse::() + .map_err(to_anyhow)?; + if expiry_nb < chrono::Utc::now().timestamp() as u64 { + tracing::error!( + "License key expired: {} < {}", + expiry_nb, + chrono::Utc::now().timestamp() as u64 + ); + let mut l = LICENSE_KEY_VALID.write().await; + *l = false; + }; + Ok(()) } diff --git a/backend/src/main.rs b/backend/src/main.rs index 4616156b78..feee88adf3 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -28,7 +28,7 @@ use windmill_common::{ JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, + RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, stats::schedule_stats, utils::{rd_string, Mode}, @@ -47,7 +47,8 @@ use crate::monitor::{ monitor_db, monitor_pool, reload_base_url_setting, reload_bunfig_install_scopes_setting, reload_extra_pip_index_url_setting, reload_job_default_timeout_setting, reload_license_key, reload_npm_config_registry_setting, reload_pip_index_url_setting, - reload_retention_period_setting, reload_server_config, reload_worker_config, + reload_retention_period_setting, reload_scim_token_setting, reload_server_config, + reload_worker_config, }; const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); @@ -399,6 +400,9 @@ Windmill Community Edition {GIT_VERSION} JOB_DEFAULT_TIMEOUT_SECS_SETTING => { reload_job_default_timeout_setting(&db).await }, + SCIM_TOKEN_SETTING => { + reload_scim_token_setting(&db).await + }, EXTRA_PIP_INDEX_URL_SETTING => { reload_extra_pip_index_url_setting(&db).await }, @@ -439,6 +443,12 @@ Windmill Community Edition {GIT_VERSION} } } }, + SAML_METADATA_SETTING => { + tracing::info!("SAML metadata change detected, killing server expecting to be restarted"); + if let Err(e) = tx.send(()) { + tracing::error!(error = %e, "Could not send killpill to server"); + } + }, DISABLE_STATS_SETTING => {}, a @_ => { tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 61bf89ea3f..45bfdcd67e 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -16,7 +16,7 @@ use tokio::{ use uuid::Uuid; use windmill_api::{ oauth2_ee::{build_oauth_clients, OAuthClient}, - DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, + DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN, }; use windmill_common::{ error, @@ -26,7 +26,7 @@ use windmill_common::{ JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, + RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, jobs::{JobKind, QueuedJob}, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, @@ -125,6 +125,8 @@ pub async fn initial_load( reload_server_config(&db).await; reload_retention_period_setting(&db).await; reload_request_size(&db).await; + reload_saml_metadata_setting(&db).await; + reload_scim_token_setting(&db).await; } #[cfg(feature = "enterprise")] @@ -317,6 +319,21 @@ pub async fn delete_expired_items(db: &DB) -> () { } } +pub async fn reload_scim_token_setting(db: &DB) { + reload_option_setting_with_tracing(db, SCIM_TOKEN_SETTING, "SCIM_TOKEN", SCIM_TOKEN.clone()) + .await; +} + +pub async fn reload_saml_metadata_setting(db: &DB) { + reload_option_setting_with_tracing( + db, + SAML_METADATA_SETTING, + "SAML_METADATA", + SAML_METADATA.clone(), + ) + .await; +} + pub async fn reload_extra_pip_index_url_setting(db: &DB) { reload_option_setting_with_tracing( db, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 75390f8834..2512d68670 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -664,6 +664,27 @@ paths: schema: type: string + /saml/test_metadata: + post: + summary: test metadata + operationId: testMetadata + tags: + - setting + requestBody: + description: test metadata + required: true + content: + application/json: + schema: + type: string + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + /users/email: get: summary: get current user email (if logged in) diff --git a/backend/windmill-api/src/ee.rs b/backend/windmill-api/src/ee.rs index a9d170a3ac..8eda5eeafd 100644 --- a/backend/windmill-api/src/ee.rs +++ b/backend/windmill-api/src/ee.rs @@ -1,6 +1,41 @@ use anyhow::anyhow; +use base64::Engine; +use rsa::{pkcs8::DecodePublicKey, signature::Verifier}; +use sha2::Sha256; -pub async fn validate_license_key(_license_key: String) -> anyhow::Result { - // Implementation is not open source - Err(anyhow!("License can't be validated in Windmill CE")) +pub async fn validate_license_key(license_key: String) -> anyhow::Result { + let mut splitted_lk = license_key.split("."); + if splitted_lk.clone().count() != 3 { + return Err(anyhow!( + "license_key can be splitted with 2 . (..)" + )); + } + + let id = splitted_lk.next().unwrap(); + let expiry = splitted_lk.next().unwrap(); + let signature_b64 = splitted_lk.next().unwrap(); + + const PUBLIC_KEY: &str = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDgVShzcLSPiOi+8ET8fggob1kmi47/cE12JaidPkwfGnScZItghkqtiLsct0U4kJhlp5gO89DYTBmIKadvxwY7kMsLlZzmi2emVH7c27cByGASY8QmWDNdG4Ggy/NDflGGBdAtN6gHawZAg4zHv3qpbPQGHH1/6sXIohcXhOnouwIDAQAB"; + let pub_key = rsa::RsaPublicKey::from_public_key_der( + &base64::engine::general_purpose::STANDARD.decode(PUBLIC_KEY)?, + )?; + let signature = base64::engine::general_purpose::STANDARD.decode(signature_b64)?; + rsa::pss::VerifyingKey::::new(pub_key) + .verify( + &format!("{id}{expiry}").as_bytes(), + &rsa::pss::Signature::from(signature), + ) + .map_err(|_| anyhow::anyhow!("Invalid license key".to_string()))?; + + let expiry_nb = expiry.parse::()?; + if expiry_nb < chrono::Utc::now().timestamp() as u64 { + tracing::error!( + "License key expired: {} < {}", + expiry_nb, + chrono::Utc::now().timestamp() as u64 + ); + return Err(anyhow!("License key expired".to_string())); + }; + + Ok(id.to_string()) } diff --git a/backend/windmill-api/src/job_helpers_ee.rs b/backend/windmill-api/src/job_helpers_ee.rs index 61a946bb84..83993542ec 100644 --- a/backend/windmill-api/src/job_helpers_ee.rs +++ b/backend/windmill-api/src/job_helpers_ee.rs @@ -1,5 +1,1457 @@ -use axum::Router; +use std::io::{self}; +use std::ops::Range; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; +use std::{cmp, future}; + +use crate::{db::DB, resources::get_resource_value_interpolated_internal, users::Tokened}; +use axum::body::StreamBody; +use axum::extract::{BodyStream, DefaultBodyLimit}; +use axum::headers::HeaderMap; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::{ + extract::{Path, Query}, + routing::{delete, get, post}, + Extension, Json, Router, +}; +use futures::{StreamExt, TryStreamExt}; +use hyper::http; +use object_store::azure::AzureConfigKey; +use object_store::{ClientConfigKey, ObjectStore}; +use polars::{ + io::{ + cloud::{AmazonS3ConfigKey, CloudOptions}, + SerReader, + }, + lazy::{ + dsl::col, + frame::{LazyFrame, ScanArgsParquet}, + }, + prelude::CsvReader, +}; +use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; +use tokio::io::{copy, AsyncWriteExt}; +use tokio_util::io::StreamReader; +use tower_http::cors::{Any, CorsLayer}; +use windmill_common::error::{Error, JsonResult}; +use windmill_common::s3_helpers::{AzureBlobResource, ObjectStoreResource}; +use windmill_common::worker::to_raw_value; +use windmill_common::{ + db::UserDB, + error, + s3_helpers::{build_object_store_client, render_endpoint, LargeFileStorage, S3Resource}, +}; + +use crate::db::ApiAuthed; pub fn workspaced_service() -> Router { + let cors = CorsLayer::new() + .allow_methods([http::Method::GET, http::Method::POST]) + .allow_headers([http::header::CONTENT_TYPE, http::header::AUTHORIZATION]) + .allow_origin(Any); + Router::new() + .route( + "/duckdb_connection_settings", + post(duckdb_connection_settings).layer(cors.clone()), + ) + .route( + "/v2/duckdb_connection_settings", + post(duckdb_connection_settings_v2).layer(cors.clone()), + ) + .route( + "/polars_connection_settings", + post(polars_connection_settings).layer(cors.clone()), + ) + .route( + "/v2/polars_connection_settings", + post(polars_connection_settings_v2).layer(cors.clone()), + ) + .route( + "/v2/s3_resource_info", + post(s3_resource_info).layer(cors.clone()), + ) + .route("/test_connection", get(test_connection).layer(cors.clone())) + .route( + "/list_stored_files", + get(list_stored_files).layer(cors.clone()), + ) + .route( + "/load_file_metadata", + get(load_file_metadata).layer(cors.clone()), + ) + .route( + "/load_file_preview", + get(load_file_preview).layer(cors.clone()), + ) + .route( + "/load_parquet_preview/*path", + get(load_parquet_preview).layer(cors.clone()), + ) + .route( + "/delete_s3_file", + delete(delete_s3_file).layer(cors.clone()), + ) + .route("/move_s3_file", get(move_s3_file).layer(cors.clone())) + .route("/upload_s3_file", post(upload_s3_file).layer(cors.clone())) + .route( + "/download_s3_file", + get(download_s3_file).layer(cors.clone()), + ) + .route( + "/multipart_upload_s3_file", + post(multipart_upload_s3_file).layer(cors.clone()), + ) + .route( + "/multipart_download_s3_file", + post(multipart_download_s3_file).layer(cors.clone()), + ) + .layer(DefaultBodyLimit::max(100 * 1024 * 1024)) // necessary for multipart upload +} + +#[derive(Deserialize)] +struct DuckdbConnectionSettingsQuery { + s3_resource: S3Resource, +} +#[derive(Serialize)] +struct DuckdbConnectionSettingsResponse { + connection_settings_str: String, +} + +async fn duckdb_connection_settings( + Path(_w_id): Path, + Json(query): Json, +) -> JsonResult { + let mut duckdb_settings: String = String::new(); + + let s3_resource = query.s3_resource; + duckdb_settings.push_str("SET home_directory='./';\n"); // TODO: make this configurable maybe, or point to a temporary folder + duckdb_settings.push_str("INSTALL 'httpfs';\n"); + if s3_resource.path_style { + duckdb_settings.push_str("SET s3_url_style='path';\n"); + } + duckdb_settings.push_str(format!("SET s3_region='{}';\n", s3_resource.region).as_str()); + duckdb_settings.push_str(format!("SET s3_endpoint='{}';\n", s3_resource.endpoint).as_str()); + if !s3_resource.use_ssl { + duckdb_settings.push_str("SET s3_use_ssl=0;\n"); // default is true for DuckDB + } + if let Some(access_key_id) = s3_resource.access_key { + duckdb_settings.push_str(format!("SET s3_access_key_id='{}';\n", access_key_id).as_str()); + } + if let Some(secret_access_key) = s3_resource.secret_key { + duckdb_settings + .push_str(format!("SET s3_secret_access_key='{}';\n", secret_access_key).as_str()); + } + + let response = DuckdbConnectionSettingsResponse { + connection_settings_str: duckdb_settings, + }; + return Ok(Json(response)); +} + +#[derive(Deserialize)] +struct DuckdbConnectionSettingsQueryV2 { + s3_resource_path: Option, +} + +async fn duckdb_connection_settings_v2( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Json(query): Json, +) -> JsonResult { + let object_store_resource_opt = match query.s3_resource_path { + Some(s3_resource_path) => { + get_s3_resource( + &authed, + &db, + Some(user_db), + &token, + &w_id, + StorageResourceType::S3, // for now we only support S3 for duckdb + s3_resource_path.as_str(), + ) + .await? + } + None => { + let (_, s3_resource_opt) = + get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?; + s3_resource_opt + } + }; + let object_store_resource = object_store_resource_opt.ok_or(Error::NotFound( + "No datasets storage resource defined at the workspace level".to_string(), + ))?; + match object_store_resource { + ObjectStoreResource::S3Resource(s3_resource) => { + duckdb_connection_settings( + Path(w_id), + Json(DuckdbConnectionSettingsQuery { s3_resource }), + ) + .await + } + ObjectStoreResource::AzureBlobResource(_) => Err(Error::BadConfig( + "DuckDB only works with an S3 storage, Azure Blob is not supported yet".to_string(), + )), + } +} + +#[derive(Deserialize)] +struct PolarsConnectionSettingsQuery { + s3_resource: S3Resource, +} + +#[derive(Serialize)] +struct PolarsConnectionSettings { + pub region_name: String, +} + +async fn polars_connection_settings( + Path(_w_id): Path, + Json(query): Json, +) -> JsonResult { + let s3_resource = query.s3_resource; + + let response = S3fsArgs { + endpoint_url: render_endpoint(s3_resource.endpoint, s3_resource.use_ssl, s3_resource.port), + key: s3_resource.access_key, + secret: s3_resource.secret_key, + use_ssl: s3_resource.use_ssl, + cache_regions: false, + client_kwargs: PolarsConnectionSettings { + region_name: s3_resource.region, + }, + }; + return Ok(Json(response)); +} + +#[derive(Deserialize)] +struct PolarsConnectionSettingsQueryV2 { + s3_resource_path: Option, +} + +#[derive(Serialize)] +struct PolarsConnectionSettingsResponse { + s3fs_args: S3fsArgs, + storage_options: PolarsStorageOptions, +} + +#[derive(Serialize)] +struct S3fsArgs { + endpoint_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + secret: Option, + use_ssl: bool, + cache_regions: bool, + client_kwargs: PolarsConnectionSettings, +} + +#[derive(Serialize)] +struct PolarsStorageOptions { + aws_endpoint_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + aws_access_key_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + aws_secret_access_key: Option, + aws_region: String, + aws_allow_http: String, +} + +async fn polars_connection_settings_v2( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Json(query): Json, +) -> JsonResult { + let object_store_resource_opt = match query.s3_resource_path { + Some(s3_resource_path) => { + get_s3_resource( + &authed, + &db, + Some(user_db), + &token, + &w_id, + StorageResourceType::S3, // for now we only support S3 for polars + s3_resource_path.as_str(), + ) + .await? + } + None => { + let (_, s3_resource_opt) = + get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?; + s3_resource_opt + } + }; + let object_store_resource = object_store_resource_opt.ok_or(Error::NotFound( + "No datasets storage resource defined at the workspace level".to_string(), + ))?; + + let s3_resource = match object_store_resource { + ObjectStoreResource::S3Resource(s3_resource) => Ok(s3_resource), + ObjectStoreResource::AzureBlobResource(_) => Err(Error::BadConfig( + "Polars only works with an S3 storage, Azure Blob is not supported yet".to_string(), + )), + }?; + + let s3fs = polars_connection_settings( + Path(w_id), + Json(PolarsConnectionSettingsQuery { + s3_resource: s3_resource.clone(), + }), + ) + .await? + .0; + let response = PolarsConnectionSettingsResponse { + s3fs_args: s3fs, + storage_options: PolarsStorageOptions { + aws_endpoint_url: render_endpoint( + s3_resource.endpoint, + s3_resource.use_ssl, + s3_resource.port, + ), + aws_access_key_id: s3_resource.access_key, + aws_secret_access_key: s3_resource.secret_key, + aws_region: s3_resource.region, + aws_allow_http: (!s3_resource.use_ssl).to_string(), + }, + }; + return Ok(Json(response)); +} + +#[derive(Deserialize)] +struct S3ResourceInfoQuery { + s3_resource_path: Option, +} + +async fn s3_resource_info( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Json(query): Json, +) -> JsonResult { + let object_store_resource_opt = match query.s3_resource_path { + Some(s3_resource_path) => { + get_s3_resource( + &authed, + &db, + Some(user_db), + &token, + &w_id, + StorageResourceType::S3, + s3_resource_path.as_str(), + ) + .await? + } + None => { + let (_, s3_resource_opt) = + get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?; + s3_resource_opt + } + }; + let object_store_resource = object_store_resource_opt.ok_or(Error::NotFound( + "No datasets storage resource defined at the workspace level".to_string(), + ))?; + match object_store_resource { + ObjectStoreResource::S3Resource(s3_resource) => Ok(Json(s3_resource)), + ObjectStoreResource::AzureBlobResource(_) => Err(Error::BadConfig("Requested S3 resource info but the resource path pointed to an Azure Blob resource type".to_string())) + } +} + +#[derive(Serialize, Deserialize, Clone)] +struct WindmillLargeFile { + s3: String, +} + +async fn test_connection( + authed: ApiAuthed, + Extension(db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, +) -> JsonResult<()> { + let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + if s3_resource_opt.is_none() { + return Err(Error::NotFound( + "No datasets storage resource defined at the workspace level".to_string(), + )); + } + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + s3_client + .list(None) + .next() + .await + .transpose() + .map_err(|err| { + tracing::error!("Error testing connection to S3 bucket: {:?}", err); + Error::InternalErr(format!("Error testing connection: {}", err.to_string())) + })?; + return Ok(Json(())); +} + +#[derive(Deserialize)] +struct ListStoredFilesQuery { + pub max_keys: usize, + pub marker: Option, + pub prefix: Option, +} + +#[derive(Serialize)] +struct ListStoredDatasetsResponse { + pub restricted_access: Option, + windmill_large_files: Vec, + pub next_marker: Option, +} + +async fn list_stored_files( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, +) -> JsonResult { + let (public_resource, s3_resource_opt) = + get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?; + if !public_resource.unwrap_or(false) && s3_resource_opt.is_none() { + return Ok(Json(ListStoredDatasetsResponse { + windmill_large_files: vec![], + next_marker: None, + restricted_access: Some(true), + })); + } + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + let prefix = query.prefix.map(object_store::path::Path::from); + let object_stream = if let Some(marker) = query.marker { + s3_client.list_with_offset(prefix.as_ref(), &object_store::path::Path::from(marker)) + } else { + s3_client.list(prefix.as_ref()) + }; + + let stored_datasets = object_stream + .take((query.max_keys) as usize) + .map(|obj| { + obj.map(|x| WindmillLargeFile { + s3: x.location.to_string(), + }) + }) + .try_collect::>() + .await + .map_err(|e| Error::InternalErr(format!("Error listring files: {e}",)))?; + + let next_marker = if stored_datasets.len() > query.max_keys as usize { + Some((stored_datasets.last().unwrap().s3).clone()) + } else { + None + }; + + // TODO: ideally do it on the size hint, if reliable + #[cfg(not(feature = "enterprise"))] + if stored_datasets.len() > 20 { + return Err(Error::ExecutionErr( + "The workspace s3 bucket contains more than 20 files. Consider upgrading to Windmill Enterprise Edition to continue to use this feature." + .to_string(), + )); + } + + return Ok(Json(ListStoredDatasetsResponse { + windmill_large_files: stored_datasets, + next_marker, + restricted_access: Some(false), + })); +} + +#[derive(Deserialize)] +struct LoadFileMetadataQuery { + pub file_key: String, +} + +#[derive(Serialize)] +struct LoadFileMetadataResponse { + pub mime_type: Option, + pub size_in_bytes: Option, + pub last_modified: Option>, + pub expires: Option>, + pub version_id: Option, +} + +#[derive(Deserialize)] +struct LoadFilePreviewQuery { + pub file_key: String, + + // The two options below are requested from s3 with an additional query is not set + pub file_size_in_bytes: Option, + pub file_mime_type: Option, + + // For CSV files, the separator needs to be specified + pub csv_separator: Option, // defaults to ',' + pub csv_has_header: Option, // defaults to true + + // Specify the content length to be read. Both will be taken into account when reading files, except for: + // - CSVs: only the length will be taken into account, a CSV file larger than this will be truncated. + // Note that truncated CSV files might not be valid CSV files anymore, and therefore the preview might fail + // - Parquet files: Parquet files are lazy-loaded. Therefore none of those params will be taken into account + pub read_bytes_from: usize, + pub read_bytes_length: usize, +} + +#[derive(Serialize)] +struct LoadFilePreviewResponse { + pub content: Option, + pub content_type: WindmillContentType, + pub msg: Option, +} + +#[derive(Serialize)] +enum WindmillContentType { + RawText, + Csv, + Parquet, + Unknown, +} + +async fn load_file_metadata( + authed: ApiAuthed, + Extension(db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, +) -> JsonResult { + let file_key = query.file_key.clone(); + let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + let path = object_store::path::Path::from(file_key.as_str()); + let s3_object_metadata = s3_client.head(&path).await.map_err(|err| { + tracing::error!("Error loading file metadata: {:?}", err); + Error::InternalErr(format!("Error loading file metadata: {}", err.to_string())) + })?; + + let response = LoadFileMetadataResponse { + mime_type: None, + size_in_bytes: Some(s3_object_metadata.size as i64), + last_modified: Some(s3_object_metadata.last_modified.to_utc()), + expires: None, + version_id: s3_object_metadata.version, + }; + return Ok(Json(response)); +} + +#[derive(Deserialize)] +struct LoadParquetQuery { + limit: Option, + offset: Option, + sort_col: Option, + sort_desc: Option, + search_col: Option, + search_term: Option, +} + +async fn load_parquet_preview( + authed: ApiAuthed, + Extension(db): Extension, + Tokened { token }: Tokened, + Path((w_id, file_key)): Path<(String, String)>, + Query(query): Query, +) -> JsonResult> { + let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + + return read_s3_parquet_chunk( + &s3_resource, + &file_key, + query.limit, + query.offset, + query + .sort_col + .map(|v| (v.to_string(), query.sort_desc.unwrap_or(false))), + query + .search_col + .map(|v| (v.to_string(), query.search_term.unwrap_or_default())), + ) + .await + .map(Json); +} + +async fn load_file_preview( + authed: ApiAuthed, + Extension(db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, +) -> JsonResult { + // query validation + if query.read_bytes_length > 8 * 1024 * 1024 { + return Err(Error::BadRequest( + "Cannot load file bigger than 8MB".to_string(), + )); + } + + let file_key = query.file_key.clone(); + let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + // if content length is provided in the request, use it, otherwise get it from s3 + let s3_object_content_length = + if query.file_size_in_bytes.is_some() || query.file_mime_type.is_some() { + query.file_size_in_bytes + } else { + let path = object_store::path::Path::from(file_key.clone()); + let s3_object_metadata = s3_client.head(&path).await.map_err(|err| { + tracing::error!("Error loading file preview: {:?}", err); + Error::InternalErr(format!("Error loading file preview: {}", err.to_string())) + })?; + Some(s3_object_metadata.size) + }; + + let file_chunk_length = if s3_object_content_length.is_some() { + cmp::min( + query.read_bytes_length, + s3_object_content_length.unwrap() - query.read_bytes_from, + ) + } else { + query.read_bytes_length + }; + let content_type: WindmillContentType; + + let lowercased_file_key = file_key.clone().to_lowercase(); + let content_preview = + if lowercased_file_key.ends_with(".csv") || lowercased_file_key.ends_with("tsv") { + content_type = WindmillContentType::Csv; + csv_file_preview_with_fallback( + s3_client, + &file_key, + file_chunk_length, + query.csv_separator, + query.csv_has_header, + ) + .await + } else if lowercased_file_key.ends_with(".json") + || lowercased_file_key.ends_with(".yaml") + || lowercased_file_key.ends_with(".yml") + || lowercased_file_key.ends_with(".xml") + || lowercased_file_key.ends_with(".txt") + || lowercased_file_key.ends_with(".log") + { + content_type = WindmillContentType::RawText; + read_s3_text_object_head( + s3_client, + &file_key, + query.read_bytes_from, + file_chunk_length, + ) + .await + } else if lowercased_file_key.ends_with(".parquet") { + content_type = WindmillContentType::Parquet; + read_s3_parquet_object_head(&s3_resource, &file_key).await + } else { + content_type = WindmillContentType::Unknown; + Err(Error::ExecutionErr( + "Preview is not available. Content type is unknown or not supported".to_string(), + )) + }; + let response: LoadFilePreviewResponse = match content_preview { + Ok(content) => LoadFilePreviewResponse { + content_type, + content: Some(content), + msg: None, + }, + + Err(err) => LoadFilePreviewResponse { + content_type, + content: None, + msg: Some(err.to_string()), + }, + }; + return Ok(Json(response)); +} + +#[derive(Deserialize)] +struct DeleteS3FileQuery { + pub file_key: String, +} + +async fn delete_s3_file( + authed: ApiAuthed, + Extension(db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, +) -> JsonResult<()> { + let file_key = query.file_key.clone(); + let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + let path = object_store::path::Path::from(file_key.as_str()); + + s3_client.delete(&path).await.map_err(|err| { + tracing::error!("Error deleting file: {:?}", err); + Error::InternalErr(format!("Error deleting file: {}", err.to_string())) + })?; + return Ok(Json(())); +} + +#[derive(Deserialize)] +struct MoveS3FileQuery { + pub src_file_key: String, + pub dest_file_key: String, +} + +async fn move_s3_file( + authed: ApiAuthed, + Extension(db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, +) -> JsonResult<()> { + let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + let source_path = object_store::path::Path::from(query.src_file_key); + let dest_path = object_store::path::Path::from(query.dest_file_key); + + s3_client + .copy(&source_path, &dest_path) + .await + .map_err(|err| { + tracing::error!("Error copying object: {:?}", err); + Error::InternalErr(format!("Error copying object: {}", err.to_string())) + })?; + + s3_client.delete(&source_path).await.map_err(|err| { + tracing::error!("Error removing object post copy: {:?}", err); + Error::InternalErr(format!( + "Error removing object post copy: {}", + err.to_string() + )) + })?; + return Ok(Json(())); +} + +#[derive(Deserialize)] +struct DownloadFileQuery { + pub file_key: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub resource_type: Option, + pub s3_resource_path: Option, +} + +async fn multipart_download_s3_file( + _authed: ApiAuthed, + Extension(_db): Extension, + Extension(_user_db): Extension, + Tokened { token: _token }: Tokened, + Path(_w_id): Path, + Json(_value): Json, +) -> JsonResult<()> { + return Err(Error::BadRequest("This endpoint has been replaced by /download_s3_file. Please upgrade your client SDK to the latest version and migrate your code to the latest endpoints.".to_string())); +} + +async fn download_s3_file( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, +) -> error::Result { + let s3_resource_opt = match query.s3_resource_path.clone() { + Some(s3_resource_path) => { + get_s3_resource( + &authed, + &db, + Some(user_db), + &token, + &w_id, + query.resource_type.unwrap_or(StorageResourceType::S3), + s3_resource_path.as_str(), + ) + .await? + } + None => { + let (_, s3_resource_opt) = + get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + s3_resource_opt + } + }; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + let path = object_store::path::Path::from(query.file_key); + let s3_object = s3_client.get(&path).await.map_err(|err| { + tracing::warn!("Error retrieving file from S3: {:?}", err); + Error::InternalErr(format!("Error retrieving file: {}", err.to_string())) + })?; + let body_stream = StreamBody::new(s3_object.into_stream()); + let mut headers = HeaderMap::new(); + headers.insert("content-type", "application/octet-stream".parse().unwrap()); + return Ok((StatusCode::OK, headers, body_stream).into_response()); +} + +#[derive(Deserialize)] +struct UploadFileQuery { + pub file_key: Option, // if none, the file will be placed in windmill_uploads/ with a random name. + pub file_extension: Option, // preferred extension for the file in case a random name has to be generated + #[serde(skip_serializing_if = "Option::is_none")] + pub resource_type: Option, + pub s3_resource_path: Option, // custom S3 resource to use for this upload. It None, the workspace S3 resource will be used +} + +#[derive(Deserialize, Serialize, Clone)] +struct UploadFilePart { + pub part_number: u16, + pub tag: String, +} + +#[derive(Serialize)] +struct UploadFileResponse { + pub file_key: String, +} + +use std::io::Result; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; + +use pin_project::pin_project; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::time::{interval, Interval}; + +#[pin_project] +pub struct ProgressReadAdapter { + #[pin] + inner: R, + interval: Interval, + interval_bytes: usize, + total_bytes: usize, +} + +impl ProgressReadAdapter { + pub fn new(inner: R) -> Self { + Self { + inner, + interval: interval(Duration::from_millis(100)), + interval_bytes: 0, + total_bytes: 0, + } + } +} + +impl AsyncRead for ProgressReadAdapter { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.project(); + + let before = buf.filled().len(); + let result = this.inner.poll_read(cx, buf); + let after = buf.filled().len(); + + *this.interval_bytes += after - before; + match this.interval.poll_tick(cx) { + Poll::Pending => {} + Poll::Ready(_) => { + *this.total_bytes += *this.interval_bytes; + tracing::info!("uploaded {} bytes for s3 upload", *this.total_bytes); + *this.interval_bytes = 0; + } + }; + + #[cfg(not(feature = "enterprise"))] + if *this.total_bytes > 50 * 1024 * 1024 { + let err = error::Error::BadRequest( + "Uploading files bigger than 50Mb is only permitted in Windmill EE".to_string(), + ); + return Poll::Ready(Err(io::Error::new(io::ErrorKind::ConnectionAborted, err))); + } + result + } +} + +async fn multipart_upload_s3_file( + _authed: ApiAuthed, + Extension(_db): Extension, + Extension(_user_db): Extension, + Tokened { token: _token }: Tokened, + Path(_w_id): Path, + Json(_value): Json, +) -> JsonResult<()> { + return Err(Error::BadRequest("This endpoint has been replaced by /upload_s3_file. Please upgrade your client SDK to the latest version to migrate your code.".to_string())); +} + +async fn upload_s3_file( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Tokened { token }: Tokened, + Path(w_id): Path, + Query(query): Query, + body: BodyStream, +) -> JsonResult { + let file_key = match query.file_key.clone() { + Some(fk) => fk, + None => { + // for now, we place all files into `windmill_uploads` folder with a random name + // TODO: make the folder configurable via the workspace settings + format!( + "windmill_uploads/upload_{}_{}.{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + rand::random::(), + query.file_extension.clone().unwrap_or("file".to_string()) + ) + .to_string() + } + }; + + let s3_resource_opt = match query.s3_resource_path.clone() { + Some(s3_resource_path) => { + get_s3_resource( + &authed, + &db, + Some(user_db), + &token, + &w_id, + query.resource_type.unwrap_or(StorageResourceType::S3), + s3_resource_path.as_str(), + ) + .await? + } + None => { + let (_, s3_resource_opt) = + get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; + s3_resource_opt + } + }; + + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( + "No files storage resource defined at the workspace level".to_string(), + ))?; + let s3_client = build_object_store_client(&s3_resource)?; + + let body_with_io_error = body.map_err(|err| io::Error::new(io::ErrorKind::Other, err)); + let body_reader = StreamReader::new(body_with_io_error); + futures::pin_mut!(body_reader); + + let path = object_store::path::Path::from(file_key.clone()); + let (multipart_id, mut parts_writer) = s3_client.put_multipart(&path).await.map_err(|err| { + tracing::error!("Error initializing multipart upload: {:?}", err); + Error::InternalErr(format!( + "Error initializing multipart upload: {}", + err.to_string() + )) + })?; + let mut progressed_body_reader = ProgressReadAdapter::new(&mut body_reader); + + copy(&mut progressed_body_reader, &mut parts_writer) + .await + .map_err(|err| { + let _ = s3_client.abort_multipart(&path, &multipart_id); + tracing::error!("Error forwarding stream to object writer: {:?}", err); + Error::InternalErr(format!("Error copying stream: {}", err.to_string())) + })?; + + parts_writer.flush().await.map_err(|err| { + let _ = s3_client.abort_multipart(&path, &multipart_id); + tracing::error!("Error flushing multipart writer: {:?}", err); + Error::InternalErr(format!( + "Error flushing multipart writer: {}", + err.to_string() + )) + })?; + parts_writer.shutdown().await.map_err(|err| { + let _ = s3_client.abort_multipart(&path, &multipart_id); + tracing::error!("Error finishing multipart upload: {:?}", err); + Error::InternalErr(format!( + "Error finishing multipart upload: {}", + err.to_string() + )) + })?; + return Ok(Json(UploadFileResponse { file_key })); +} + +#[derive(Deserialize)] +pub struct S3Object { + pub s3: String, +} + +async fn get_workspace_s3_resource<'c>( + authed: &ApiAuthed, + db: &DB, + user_db: Option, + token: &str, + w_id: &str, +) -> error::Result<(Option, Option)> { + let raw_lfs_opt = sqlx::query_scalar!( + "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", + w_id + ) + .fetch_optional(db) + .await? + .flatten(); + + if raw_lfs_opt.is_none() { + return Ok((None, None)); + } + + let large_file_storage = serde_json::from_value::( + raw_lfs_opt.unwrap(), + ) + .map_err(|err| { + tracing::error!( + "Value stored in large_file_storage column is invalid and could not be deserialized: {}", + err + ); + Error::InternalErr( + "Could not deserialize LargeFileStorage value found in database".to_string(), + ) + })?; + let (resource_type, public_resource, resource_path) = match large_file_storage { + LargeFileStorage::S3Storage(s3_lfs) => ( + StorageResourceType::S3, + s3_lfs.public_resource, + s3_lfs.s3_resource_path, + ), + LargeFileStorage::AzureBlobStorage(azure_lfs) => ( + StorageResourceType::AzureBlob, + azure_lfs.public_resource, + azure_lfs.azure_blob_resource_path, + ), + }; + + // if the resource is declared public, we replace user_db with None such that the resource info will be + // retrieved using `db` (and ACLs will be bypassed) + let effective_user_db = if user_db.is_some() && public_resource.unwrap_or(false) { + None + } else { + user_db + }; + + let stripped_resource_path = resource_path + .strip_prefix("$res:") + .unwrap_or(resource_path.as_str()); + let s3_resource = match get_s3_resource( + authed, + db, + effective_user_db, + token, + w_id, + resource_type, + stripped_resource_path, + ) + .await + { + Ok(s3_resource) => Ok(s3_resource), + Err(Error::NotAuthorized(_)) if !public_resource.unwrap_or(false) => Ok(None), + Err(err) => Err(err), + }; + return s3_resource.map(|res| (public_resource, res)); +} + +async fn get_s3_resource<'c>( + authed: &ApiAuthed, + db: &DB, + user_db: Option, + token: &str, + w_id: &str, + resource_type: StorageResourceType, + resource_path: &str, +) -> error::Result> { + let s3_resource_value_raw = get_resource_value_interpolated_internal( + authed, + user_db, + db, + w_id, + resource_path, + None, + token, + ) + .await?; + + if s3_resource_value_raw.is_none() { + return Err(Error::NotFound("Resource not found".to_string())); + } + + let object_store_resource = match resource_type { + StorageResourceType::S3 => { + let s3_resource = serde_json::from_value::(s3_resource_value_raw.unwrap()) + .map_err(|err| { + tracing::error!("Error deserializing S3 resource: {:?}", err); + Error::InternalErr(format!("Error reading s3 resource: {}", err.to_string())) + })?; + ObjectStoreResource::S3Resource(s3_resource) + } + StorageResourceType::AzureBlob => { + let azure_blob_resource = + serde_json::from_value::(s3_resource_value_raw.unwrap()) + .map_err(|err| { + tracing::error!("Error deserializing S3 resource: {:?}", err); + Error::InternalErr(format!( + "Error reading s3 resource: {}", + err.to_string() + )) + })?; + ObjectStoreResource::AzureBlobResource(azure_blob_resource) + } + }; + return Ok(Some(object_store_resource)); +} + +#[derive(Deserialize)] +pub enum StorageResourceType { + S3, + AzureBlob, +} + +fn build_polars_s3_config(object_store_resource_ref: &ObjectStoreResource) -> CloudOptions { + match object_store_resource_ref { + ObjectStoreResource::S3Resource(s3_resource_ref) => { + let s3_resource = s3_resource_ref.to_owned(); + let mut s3_configs: Vec<(AmazonS3ConfigKey, String)> = vec![ + (AmazonS3ConfigKey::Region, s3_resource.region), + (AmazonS3ConfigKey::Bucket, s3_resource.bucket), + ( + AmazonS3ConfigKey::Endpoint, + render_endpoint(s3_resource.endpoint, s3_resource.use_ssl, s3_resource.port), + ), + ( + AmazonS3ConfigKey::Client(ClientConfigKey::AllowHttp), + (!s3_resource.use_ssl).to_string(), + ), + ( + AmazonS3ConfigKey::VirtualHostedStyleRequest, + (!s3_resource.path_style).to_string(), + ), + ]; + if let Some(access_key) = s3_resource.access_key { + if access_key != "" { + s3_configs.push((AmazonS3ConfigKey::AccessKeyId, access_key)); + } + } + if let Some(secret_key) = s3_resource.secret_key { + if secret_key != "" { + s3_configs.push((AmazonS3ConfigKey::SecretAccessKey, secret_key)); + } + } + CloudOptions::default().with_aws(s3_configs) + } + ObjectStoreResource::AzureBlobResource(azure_blob_resource) => { + let azure_blob_resource = azure_blob_resource.to_owned(); + let mut azure_blob_configs: Vec<(AzureConfigKey, String)> = vec![ + ( + AzureConfigKey::AccountName, + azure_blob_resource.account_name, + ), + ( + AzureConfigKey::ContainerName, + azure_blob_resource.container_name, + ), + ( + AzureConfigKey::Client(ClientConfigKey::AllowHttp), + (!azure_blob_resource.use_ssl).to_string(), + ), + ]; + if let Some(endpoint) = azure_blob_resource.endpoint { + if endpoint != "" { + azure_blob_configs.push(( + AzureConfigKey::Endpoint, + render_endpoint(endpoint, azure_blob_resource.use_ssl, None), + )) + } + } + if let Some(access_key) = azure_blob_resource.access_key { + if access_key != "" { + azure_blob_configs.push((AzureConfigKey::AccessKey, access_key)); + } + } + CloudOptions::default().with_azure(azure_blob_configs) + } + } +} + +async fn read_object_chunk( + s3_client: Arc, + file_key: &str, + from_byte: usize, + length: usize, +) -> error::Result> { + let path = object_store::path::Path::from(file_key); + let s3_object = s3_client + .get_range( + &path, + Range { + start: from_byte, + end: from_byte + length, + }, + ) + .await + .map_err(|err| { + tracing::warn!("Error fetching text file from S3: {:?}", err); + Error::InternalErr(err.to_string()) + })?; + + return Ok(s3_object.to_vec()); +} + +async fn read_s3_text_object_head( + s3_client: Arc, + file_key: &str, + from_byte: usize, + length: usize, +) -> error::Result { + let payload = read_object_chunk(s3_client, file_key, from_byte, length).await?; + let file_header_str = String::from_utf8(payload).map_err(|err| { + tracing::warn!( + "Encoding of file {} unsupported. Error was: {:?}", + file_key, + err + ); + Error::InternalErr("File encoding is not supported".to_string()) + })?; + return Ok(file_header_str); +} + +async fn read_s3_parquet_object_head( + object_store_resource_ref: &ObjectStoreResource, + file_key: &str, +) -> error::Result { + let polars_cloud_config = build_polars_s3_config(object_store_resource_ref); + + let args: ScanArgsParquet = ScanArgsParquet { + n_rows: Some(1), + cache: false, + parallel: polars::io::parquet::ParallelStrategy::Auto, + rechunk: false, + row_count: None, + low_memory: false, + use_statistics: false, + hive_partitioning: false, + cloud_options: Some(polars_cloud_config), + }; + + let file_key_prefixed = match object_store_resource_ref { + ObjectStoreResource::S3Resource(s3_resource) => { + format!("s3://{}/{}", s3_resource.bucket, file_key).to_string() + } + ObjectStoreResource::AzureBlobResource(azure_blob_resource) => { + format!("az://{}/{}", azure_blob_resource.container_name, file_key).to_string() + } + }; + let polars_df_result = tokio::task::spawn_blocking(move || { + let lzdf_result = LazyFrame::scan_parquet(file_key_prefixed, args); + match lzdf_result { + Err(err) => { + tracing::warn!("Error fetching parquet file from S3: {:?}", err); + return Err(Error::InternalErr(err.to_string())); + } + Ok(lzdf) => { + let df = lzdf + .select(&[col("*")]) + .limit(10) // for now read only first 10 lines + .collect() + .map_err(|err| { + Error::InternalErr(format!("Error querying dataset: {}", err.to_string())) + })?; + return Ok(format!("{:?}", df).to_string()); + } + } + }) + .await + .map_err(|err| Error::InternalErr(format!("Error reading dataset: {}", err.to_string())))?; + + return polars_df_result; +} + +async fn read_s3_parquet_chunk( + object_store_resource_ref: &ObjectStoreResource, + file_key: &str, + limit: Option, + offset: Option, + sort: Option<(String, bool)>, + search: Option<(String, String)>, +) -> error::Result> { + let s3_cloud_config = build_polars_s3_config(object_store_resource_ref); + + let args: ScanArgsParquet = ScanArgsParquet { + n_rows: None, + cache: false, + parallel: polars::io::parquet::ParallelStrategy::Auto, + rechunk: false, + row_count: None, + low_memory: false, + use_statistics: false, + hive_partitioning: false, + cloud_options: Some(s3_cloud_config), + }; + + let file_key_prefixed = match object_store_resource_ref { + ObjectStoreResource::S3Resource(s3_resource) => { + format!("s3://{}/{}", s3_resource.bucket, file_key).to_string() + } + ObjectStoreResource::AzureBlobResource(azure_blob_resource) => { + format!("az://{}/{}", azure_blob_resource.container_name, file_key).to_string() + } + }; + return tokio::task::spawn_blocking(move || { + let lzdf_result = LazyFrame::scan_parquet(file_key_prefixed, args); + match lzdf_result { + Err(err) => { + tracing::warn!("Error fetching parquet file from S3: {:?}", err); + return Err(Error::InternalErr(err.to_string())); + } + Ok(lzdf) => { + let df = lzdf + .select(&[col("*")]) + .slice(offset.unwrap_or(0), limit.unwrap_or(100)); + + let df = if let Some(sort) = sort { + df.sort( + &sort.0, + SortOptions { + descending: sort.1, + nulls_last: false, + multithreaded: false, + maintain_order: false, + }, + ) + } else { + df + }; + use polars::prelude::*; + let df = if let Some(search) = search { + df.filter(col(&search.0).str().contains_literal(lit(search.1.clone()))) + } else { + df + }; + let df = df.collect().map_err(|err| { + Error::InternalErr(format!("Error collecting dataset: {}", err.to_string())) + })?; + return Ok(to_raw_value(&df)); + } + } + }) + .await + .map_err(|err| { + tracing::error!("Error reading dataset: {:?}", err); + Error::InternalErr(format!("Error reading dataset: {}", err.to_string())) + })?; +} +async fn csv_file_preview_with_fallback( + object_store_client: Arc, + file_key: &str, + length: usize, + separator: Option, + has_header: Option, +) -> error::Result { + match read_s3_csv_object_head( + object_store_client.clone(), + &file_key, + length, + separator, + has_header, + ) + .await + { + Ok(csv_preview) => Ok(csv_preview), + Err(_) => { + // fallback to default text file preview is the CSV could not be parsed. It's a text file after all + let raw_text = + read_s3_text_object_head(object_store_client.clone(), &file_key, 0, length).await?; + return Ok(raw_text); + } + } +} + +async fn read_s3_csv_object_head( + object_store_client: Arc, + file_key: &str, + length: usize, + separator: Option, + has_header: Option, +) -> error::Result { + let separator_final = match separator { + Some(separator_char) if separator_char == "\\t" => Ok("\t".as_bytes()[0]), + Some(separator_char) if separator_char.len() != 1 => Err(Error::BadRequest( + "Separator must be a single character".to_string(), + )), + Some(separator_char) => Ok(separator_char.as_bytes()[0]), + None => Ok(",".as_bytes()[0]), // polars uses the comma as default, doing the same here + }?; + + let path = object_store::path::Path::from(file_key); + let stored_object = object_store_client + .get(&path) + .await + .map_err(|err| { + tracing::error!("Error fetching CSV file: {:?}", err); + Error::InternalErr(format!("Error fetching CSV file: {}", err.to_string())) + })? + .into_stream(); + + // TODO: polars does not seem to support lazy csv reader, unfortunately. We can implement it ourselves if needed + // Right now it's fine b/c we limit the download from AWS to 32MB. We should recommend users to use parquet + // for larger files + let file_content_bytes = stored_object + .take(length as usize) + .filter(|obj| future::ready(obj.is_ok())) + .map(|obj| obj.unwrap()) + .collect::>() + .await; + + let cursor = std::io::Cursor::new(file_content_bytes.concat()); + + let csv_df = CsvReader::new(cursor) + .with_n_rows(Some(10)) // for now read only first 10 lines + .with_separator(separator_final) + .has_header(has_header.unwrap_or(true)) + .finish() + .map_err(|err| { + tracing::error!("Error reading CSV file: {:?}", err); + Error::InternalErr(format!("Error reading CSV file: {}", err.to_string())) + })?; + + return Ok(format!("{:?}", csv_df).to_string()); } diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index b06453f548..000a710b42 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -89,6 +89,9 @@ lazy_static::lazy_static! { pub static ref REQUEST_SIZE_LIMIT: Arc> = Arc::new(RwLock::new(DEFAULT_BODY_LIMIT)); + pub static ref SCIM_TOKEN: Arc>> = Arc::new(RwLock::new(None)); + pub static ref SAML_METADATA: Arc>> = Arc::new(RwLock::new(None)); + pub static ref COOKIE_DOMAIN: Option = std::env::var("COOKIE_DOMAIN").ok(); diff --git a/backend/windmill-api/src/oauth2_ee.rs b/backend/windmill-api/src/oauth2_ee.rs index 032e0c9a58..70168074b1 100644 --- a/backend/windmill-api/src/oauth2_ee.rs +++ b/backend/windmill-api/src/oauth2_ee.rs @@ -6,28 +6,70 @@ * LICENSE-AGPL for a copy of the license. */ +use std::sync::Arc; use std::{collections::HashMap, fmt::Debug}; +use anyhow::Context; use axum::body::StreamBody; +use axum::extract::FromRequestParts; +use axum::http::request::Parts; use axum::response::IntoResponse; -use axum::{routing::get, Json, Router}; +use axum::{ + async_trait, + body::Bytes, + extract::{Extension, Path, Query}, + response::Redirect, + routing::{get, post}, + Json, Router, +}; +use base64::Engine; use hmac::Mac; use hyper::{HeaderMap, StatusCode}; +use itertools::Itertools; use oauth2::{Client as OClient, *}; -use serde::{Deserialize, Serialize}; +use reqwest::Client; +use serde::{de::DeserializeOwned, Deserialize, Serialize}; use sqlx::{Postgres, Transaction}; +use tower_cookies::{Cookie, Cookies}; +use windmill_audit::audit_ee::audit_log; +use windmill_audit::ActionKind; +use windmill_common::db::UserDB; +use windmill_common::jobs::JobPayload; use windmill_common::more_serde::maybe_number_opt; +use windmill_common::oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH; +use windmill_common::users::{truncate_token, username_to_permissioned_as}; +use windmill_common::utils::{not_found_if_none, now_from_db}; +use windmill_common::variables::build_crypt; -use crate::{HTTP_CLIENT, OAUTH_CLIENTS}; -use windmill_common::error::{self, to_anyhow}; +#[cfg(feature = "enterprise")] +use windmill_common::ee::{get_license_plan, LicensePlan}; + +use crate::db::ApiAuthed; +use crate::saml_ee::{generate_redirect_url, ServiceProviderExt}; +use crate::users::{add_to_demo_if_exists, LoginUserInfo}; +use crate::webhook_util::WebhookShared; +use crate::workspaces::invite_user_to_all_auto_invite_worspaces; +use crate::{db::DB, variables::encrypt, workspaces::WorkspaceSettings}; +use crate::{BASE_URL, HTTP_CLIENT, IS_SECURE, OAUTH_CLIENTS, SLACK_SIGNING_SECRET}; +use windmill_common::error::{self, to_anyhow, Error}; use windmill_common::oauth2::*; -use crate::db::DB; -use std::str; +use windmill_queue::PushIsolationLevel; + +use std::{fs, str}; pub fn global_service() -> Router { Router::new() + .route("/login/:client", get(login)) + .route("/login_callback/:client", post(login_callback)) + .route("/connect/:client", get(connect)) + .route("/connect_callback/:client", post(connect_callback)) + .route("/connect_slack", get(connect_slack)) + .route( + "/slack_command", + post(slack_command).route_layer(axum::middleware::from_extractor::()), + ) .route("/list_supabase", get(list_supabase)) .route("/list_logins", get(list_logins)) .route("/list_connects", get(list_connects)) @@ -35,27 +77,48 @@ pub fn global_service() -> Router { pub fn workspaced_service() -> Router { Router::new() + .route("/disconnect/:id", post(disconnect)) + .route("/disconnect_slack", post(disconnect_slack)) + .route("/create_account", post(create_account)) + .route("/delete_account/:id", post(delete_account)) + .route("/refresh_token/:id", post(refresh_token)) + .route("/connect_slack_callback", post(connect_slack_callback)) } #[derive(Serialize)] #[serde(tag = "type")] pub enum InstanceEvent { - UserAdded { email: String }, + UserSignupOAuth { + email: String, + }, + UserAdded { + email: String, + }, // UserDeleted { email: String }, // UserDeletedWorkspace { workspace: String, email: String }, - UserAddedWorkspace { workspace: String, email: String }, - UserInvitedWorkspace { workspace: String, email: String }, - UserJoinedWorkspace { workspace: String, email: String, username: String }, + UserAddedWorkspace { + workspace: String, + email: String, + }, + UserInvitedWorkspace { + workspace: String, + email: String, + }, + UserJoinedWorkspace { + workspace: String, + email: String, + username: String, + }, } #[derive(Debug, Clone)] pub struct ClientWithScopes { - _client: OClient, + client: OClient, scopes: Vec, extra_params: Option>, - _extra_params_callback: Option>, - _allowed_domains: Option>, - _userinfo_url: Option, + extra_params_callback: Option>, + allowed_domains: Option>, + userinfo_url: Option, } pub type BasicClientsMap = HashMap; @@ -88,15 +151,224 @@ pub struct AllClients { } pub fn build_oauth_clients( - _base_url: &str, - _oauths_from_config: Option>, + base_url: &str, + oauths_from_config: Option>, ) -> anyhow::Result { - // Implementation is not open source - return Ok(AllClients { - logins: HashMap::default(), - connects: HashMap::default(), - slack: None, - }); + let connect_configs = serde_json::from_str::>(include_str!( + "../../oauth_connect.json" + ))?; + let login_configs = serde_json::from_str::>(include_str!( + "../../oauth_login.json" + ))?; + + let oauths = if let Some(oauths) = oauths_from_config { + oauths + } else { + let path = "./oauth.json"; + let content: String = if let Ok(e) = std::env::var("OAUTH_JSON_AS_BASE64") { + str::from_utf8( + &base64::engine::general_purpose::STANDARD + .decode(e) + .map_err(to_anyhow)?, + )? + .to_string() + } else if std::path::Path::new(path).exists() { + fs::read_to_string(path).map_err(to_anyhow)? + } else { + tracing::warn!("oauth.json not found, no OAuth clients loaded"); + return Ok(AllClients { + logins: HashMap::new(), + connects: HashMap::new(), + slack: None, + }); + }; + + if content.is_empty() { + tracing::warn!("oauth.json is empty, no OAuth clients loaded"); + return Ok(AllClients { + logins: HashMap::new(), + connects: HashMap::new(), + slack: None, + }); + }; + match serde_json::from_str::>(&content) { + Ok(clients) => clients, + Err(e) => { + tracing::error!("deserializing oauth.json: {e}"); + HashMap::new() + } + } + .into_iter() + .collect() + }; + + tracing::info!("OAuth loaded clients: {}", oauths.keys().join(", ")); + + let logins = login_configs + .into_iter() + .filter_map(|x| oauths.get(&x.0).map(|c| (x.0, (c, x.1)))) + .chain(oauths.iter().filter_map(|x| { + x.1.login_config + .as_ref() + .map(|c| (x.0.clone(), (x.1, c.clone()))) + })) + .filter_map(|(k, (client_params, config))| { + let named_client = build_basic_client( + k.clone(), + config.clone(), + client_params.clone(), + true, + base_url, + None, + ); + named_client + .map(|named_client| { + ( + named_client.0, + ClientWithScopes { + client: named_client.1, + scopes: config.scopes.unwrap_or(vec![]), + extra_params: config.extra_params, + extra_params_callback: config.extra_params_callback, + allowed_domains: client_params.allowed_domains.clone(), + userinfo_url: config.userinfo_url, + }, + ) + }) + .map_err(|e| { + tracing::error!("Error building oauth client {k}: {e}"); + e + }) + .ok() + }) + .collect(); + + let connects = connect_configs + .into_iter() + .filter_map(|x| oauths.get(&x.0).map(|c| (x.0, (c, x.1)))) + .chain(oauths.iter().filter_map(|x| { + x.1.connect_config + .as_ref() + .map(|c| (x.0.clone(), (x.1, c.clone()))) + })) + .filter_map(|(k, (client_params, config))| { + let named_client = build_basic_client( + k.clone(), + config.clone(), + client_params.clone(), + false, + base_url, + if k == "supabase_wizard" { + Some(format!("{base_url}/oauth/callback_supabase")) + } else { + None + }, + ); + named_client + .map(|named_client| { + ( + named_client.0, + ClientWithScopes { + client: named_client.1, + scopes: config.scopes.unwrap_or(vec![]), + extra_params: config.extra_params, + extra_params_callback: config.extra_params_callback, + allowed_domains: None, + userinfo_url: None, + }, + ) + }) + .map_err(|e| { + tracing::error!("Error building oauth client {k}: {e}"); + e + }) + .ok() + }) + .collect(); + + let slack = oauths + .get("slack") + .map(|v| { + build_basic_client( + "slack".to_string(), + OAuthConfig { + auth_url: "https://slack.com/oauth/authorize".to_string(), + token_url: "https://slack.com/api/oauth.access".to_string(), + userinfo_url: None, + scopes: None, + extra_params: None, + extra_params_callback: None, + req_body_auth: None, + }, + v.clone(), + false, + base_url, + Some(format!("{base_url}/oauth/callback_slack")), + ) + .map(|x| x.1) + .map_err(|e| { + tracing::error!("Error building oauth slack client: {e}"); + e + }) + .ok() + }) + .flatten(); + let all_clients = AllClients { + logins, + connects, + slack, + }; + tracing::debug!("Final oauth config: {all_clients:#?}"); + Ok(all_clients) +} + +use anyhow::anyhow; +pub fn build_basic_client( + name: String, + config: OAuthConfig, + client_params: OAuthClient, + login: bool, + base_url: &str, + override_callback: Option, +) -> error::Result<(String, OClient)> { + let auth_url = Url::parse(&config.auth_url) + .map_err(|e| anyhow!("Invalid authorization endpoint URL: {e}"))?; + let token_url = + Url::parse(&config.token_url).map_err(|e| anyhow!("Invalid token endpoint URL: {e}"))?; + + let redirect_url = if login { + format!("{base_url}/user/login_callback/{name}") + } else if let Some(callback) = override_callback { + callback + } else { + format!("{base_url}/oauth/callback/{name}") + }; + + let mut client = OClient::new(client_params.id, auth_url, token_url); + if config.req_body_auth.unwrap_or(false) { + client.set_auth_type(AuthType::RequestBody); + } + client.set_client_secret(client_params.secret.clone()); + client.set_redirect_url( + Url::parse(&redirect_url).map_err(|e| anyhow!("Invalid redirect URL: {e}"))?, + ); + + // Set up the config for the Github OAuth2 process. + Ok((name.to_string(), client)) +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct SlackTokenResponse { + access_token: AccessToken, + team_id: String, + team_name: String, + #[serde(rename = "scope")] + #[serde(deserialize_with = "helpers::deserialize_space_delimited_vec")] + #[serde(serialize_with = "helpers::serialize_space_delimited_vec")] + #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default)] + scopes: Option>, + bot: SlackBotToken, } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -112,14 +384,120 @@ pub struct TokenResponse { scope: Option>, } +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct SlackBotToken { + bot_access_token: String, +} + +async fn connect( + Path(client_name): Path, + Query(query): Query>, + cookies: Cookies, +) -> error::Result { + let mut query = query.clone(); + let connects = &OAUTH_CLIENTS.read().await.connects; + let scopes = query + .get("scopes") + .map(|x| x.split('+').map(|x| x.to_owned()).collect()); + query.remove("scopes"); + let extra_params = if query.is_empty() { + None + } else { + Some(query.clone()) + }; + oauth_redirect( + connects, + client_name, + cookies, + scopes, + extra_params, + IS_SECURE.read().await.clone(), + ) +} + +#[derive(Deserialize)] +struct CreateAccount { + client: String, + refresh_token: Option, + expires_in: i64, +} + +async fn create_account( + authed: ApiAuthed, + Extension(user_db): Extension, + Path(w_id): Path, + Json(payload): Json, +) -> error::Result { + let mut tx = user_db.begin(&authed).await?; + + let id = sqlx::query_scalar!( + "INSERT INTO account (workspace_id, client, expires_at, refresh_token) VALUES ($1, \ + $2, now() + ($3 || ' seconds')::interval, $4) RETURNING id", + w_id, + payload.client, + payload.expires_in.to_string(), + payload.refresh_token + ) + .fetch_one(&mut *tx) + .await + .map_err(|e| Error::InternalErr(format!("creating account in {w_id}: {e}")))?; + tx.commit().await?; + Ok(id.to_string()) +} + +async fn delete_account( + authed: ApiAuthed, + Extension(user_db): Extension, + Query((w_id, id)): Query<(String, i32)>, +) -> error::Result { + let mut tx = user_db.begin(&authed).await?; + + let exists = sqlx::query!( + "DELETE FROM account WHERE workspace_id = $1 AND id = $2 RETURNING id", + w_id, + id, + ) + .fetch_optional(&mut *tx) + .await?; + + let id_str = id.to_string(); + not_found_if_none(exists, "Account", &id_str)?; + + audit_log( + &mut *tx, + &authed.username, + "account.delete", + ActionKind::Delete, + &w_id, + Some(&id_str), + None, + ) + .await?; + tx.commit().await?; + Ok(format!("Deleted account id {id}")) +} + #[derive(Serialize)] struct Logins { oauth: Vec, saml: Option, } -async fn list_logins() -> error::JsonResult { - // Implementation is not open source - return Ok(Json(Logins { oauth: vec![], saml: None })); +async fn list_logins( + Extension(sso): Extension>, +) -> error::JsonResult { + let saml_redirect_opt = generate_redirect_url(sso) + .await + .map_err(|e| Error::InternalErr(e.to_string()))?; + Ok(Json(Logins { + oauth: OAUTH_CLIENTS + .read() + .await + .logins + .keys() + .map(|x| x.to_owned()) + .collect::>(), + saml: saml_redirect_opt, + })) } #[derive(Serialize)] @@ -144,16 +522,708 @@ async fn list_connects() -> error::JsonResult> )) } -pub async fn _refresh_token<'c>( - _tx: Transaction<'c, Postgres>, - _path: &str, - _w_id: &str, - _id: i32, +async fn connect_slack(cookies: Cookies) -> error::Result { + let mut client = OAUTH_CLIENTS + .read() + .await + .slack + .as_ref() + .ok_or_else(|| { + error::Error::BadRequest( + "slack client not setup. See: https://www.windmill.dev/docs/misc/setup_oauth#slack" + .to_string(), + ) + })? + .to_owned(); + let state = State::new_random(); + + client.add_scope("bot"); + client.add_scope("commands"); + let url = client.authorize_url(&state); + + set_cookie(&state, cookies, IS_SECURE.read().await.clone()); + Ok(Redirect::to(url.as_str())) +} + +async fn disconnect( + authed: ApiAuthed, + Path((w_id, id)): Path<(String, i32)>, + Extension(user_db): Extension, ) -> error::Result { - // Implementation is not open source - Err(error::Error::BadRequest( - "Not implemented in Windmill's Open Source repository".to_string(), - )) + let mut tx = user_db.begin(&authed).await?; + + sqlx::query!( + "DELETE FROM account WHERE id = $1 AND workspace_id = $2", + id, + w_id + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + + Ok(format!("account {id} disconnected")) +} + +async fn disconnect_slack( + authed: ApiAuthed, + Path(w_id): Path, + Extension(user_db): Extension, +) -> error::Result { + let mut tx = user_db.begin(&authed).await?; + + sqlx::query!( + "UPDATE workspace_settings + SET slack_team_id = null, slack_name = null WHERE workspace_id = $1", + &w_id + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + + Ok(format!("slack disconnected")) +} + +async fn login(Path(client_name): Path, cookies: Cookies) -> error::Result { + let clients = &OAUTH_CLIENTS.read().await.logins; + oauth_redirect( + clients, + client_name, + cookies, + None, + None, + IS_SECURE.read().await.clone(), + ) +} + +#[derive(Deserialize)] +struct VariablePath { + path: String, +} +async fn refresh_token( + authed: ApiAuthed, + Path((w_id, id)): Path<(String, i32)>, + Extension(user_db): Extension, + Json(VariablePath { path }): Json, +) -> error::Result { + let tx = user_db.begin(&authed).await?; + + _refresh_token(tx, &path, &w_id, id).await?; + + Ok(format!("Token at path {path} refreshed")) +} + +pub async fn _refresh_token<'c>( + mut tx: Transaction<'c, Postgres>, + path: &str, + w_id: &str, + id: i32, +) -> error::Result { + let account = sqlx::query!( + "SELECT client, refresh_token FROM account WHERE workspace_id = $1 AND id = $2", + w_id, + id, + ) + .fetch_optional(&mut *tx) + .await?; + let account = not_found_if_none(account, "Account", &id.to_string())?; + let client = (&OAUTH_CLIENTS + .read() + .await + .connects + .get(&account.client) + .ok_or_else(|| error::Error::BadRequest("invalid client".to_string()))? + .client) + .to_owned(); + + let token = _exchange_token(client, &account.refresh_token).await; + + if let Err(token_err) = token { + sqlx::query!( + "UPDATE account SET refresh_error = $1 WHERE workspace_id = $2 AND id = $3", + token_err.alt(), + w_id, + id, + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + return Err(error::Error::BadRequest(format!( + "Error refreshing token: {}", + token_err.alt() + ))); + }; + + let token = token.unwrap(); + + let expires_at = now_from_db(&mut *tx).await? + + chrono::Duration::seconds( + token + .expires_in + .ok_or_else(|| Error::InternalErr("expires_in exepcted and not found".to_string()))? + .try_into() + .unwrap(), + ); + sqlx::query!( + "UPDATE account SET refresh_token = $1, expires_at = $2 WHERE workspace_id = $3 AND id = \ + $4", + token + .refresh_token + .map(|x| x.to_string()) + .unwrap_or(account.refresh_token), + expires_at, + w_id, + id + ) + .execute(&mut *tx) + .await?; + + let token_str = token.access_token.to_string(); + let mc = build_crypt(&mut tx, &w_id).await?; + let encrypted_token = encrypt(&mc, token_str.as_str()); + + sqlx::query!( + "UPDATE variable SET value = $1 WHERE workspace_id = $2 AND path = $3", + encrypted_token, + w_id, + path + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(token_str) +} + +async fn _exchange_token(client: OClient, refresh_token: &str) -> Result { + let token_json = client + .exchange_refresh_token(&RefreshToken::from(refresh_token)) + .with_client(&HTTP_CLIENT) + .execute::() + .await + .map_err(to_anyhow)?; + let token = serde_json::from_value::(token_json.clone()).map_err(|e| { + Error::BadConfig(format!( + "Error deserializing response as a new token: {e}\nresponse:{token_json}" + )) + })?; + Ok(token) +} + +#[derive(Deserialize)] +pub struct OAuthCallback { + code: String, + state: String, +} + +async fn connect_callback( + cookies: Cookies, + Path(client_name): Path, + Json(callback): Json, +) -> error::JsonResult { + let connects = &OAUTH_CLIENTS.read().await.connects; + let client_w_scopes = connects + .get(&client_name) + .ok_or_else(|| error::Error::BadRequest("invalid client".to_string()))?; + let client = client_w_scopes.client.to_owned(); + let extra_params = client_w_scopes.extra_params_callback.clone(); + let token_response = + exchange_code::(callback, &cookies, client, &HTTP_CLIENT, extra_params) + .await?; + + Ok(Json(token_response)) +} + +async fn connect_slack_callback( + Path(w_id): Path, + authed: ApiAuthed, + cookies: Cookies, + Extension(user_db): Extension, + Json(callback): Json, +) -> error::Result { + let client = OAUTH_CLIENTS + .read() + .await + .slack + .as_ref() + .ok_or_else(|| { + error::Error::BadRequest( + "slack client not setup. See: https://www.windmill.dev/docs/misc/setup_oauth#slack" + .to_string(), + ) + })? + .to_owned(); + let token = + exchange_code::(callback, &cookies, client, &HTTP_CLIENT, None).await?; + + let mut tx = user_db.begin(&authed).await?; + + sqlx::query!( + "INSERT INTO workspace_settings + (workspace_id, slack_team_id, slack_name, slack_email) + VALUES ($1, $2, $3, $4) ON CONFLICT (workspace_id) DO UPDATE SET slack_team_id = $2, \ + slack_name = $3, slack_email = $4", + &w_id, + token.team_id, + token.team_name, + authed.email + ) + .execute(&mut *tx) + .await?; + sqlx::query_as!( + Group, + "INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + w_id, + "slack", + "The group slack commands act on behalf of", + serde_json::json!({username_to_permissioned_as(&authed.username): true}) + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "INSERT INTO folder + (workspace_id, name, display_name, owners, extra_perms) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING", + &w_id, + "slack_bot", + "Slack bot", + &["g/slack".to_string()], + serde_json::json!({"g/slack": true, "g/error_handler": false}) + ) + .execute(&mut *tx) + .await?; + + let mc = build_crypt(&mut tx, &w_id).await?; + let value = encrypt(&mc, &token.bot.bot_access_token); + sqlx::query!( + "INSERT INTO variable + (workspace_id, path, value, is_secret, description, account, is_oauth) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (workspace_id, path) DO UPDATE SET value = $3", + &w_id, + WORKSPACE_SLACK_BOT_TOKEN_PATH, + value, + true, + "The slack bot token to act on behalf of the installed app of the connected workspace", + None::, + true, + ) + .execute(&mut *tx) + .await?; + + sqlx::query!( + "INSERT INTO resource + (workspace_id, path, value, description, resource_type) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT (workspace_id, path) DO UPDATE SET value = $3", + w_id, + WORKSPACE_SLACK_BOT_TOKEN_PATH, + serde_json::json!({ "token": format!("$var:{WORKSPACE_SLACK_BOT_TOKEN_PATH}") }), + "The slack bot token to act on behalf of the installed app of the connected workspace", + "slack", + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok("slack workspace connected".to_string()) +} + +#[derive(Deserialize, Debug)] +pub struct SlackCommand { + team_id: String, + user_name: String, + text: String, + response_url: String, +} + +#[derive(Clone, Debug)] +pub struct SlackSig { + sig: String, + ts: String, +} + +#[async_trait] +impl FromRequestParts for SlackSig +where + S: Send + Sync, +{ + type Rejection = (StatusCode, String); + + async fn from_request_parts( + parts: &mut Parts, + _state: &S, + ) -> std::result::Result { + let hm = &parts.headers; + Ok(Self { + sig: hm + .get("X-Slack-Signature") + .map(|x| x.to_str().unwrap_or("")) + .unwrap_or("") + .to_string(), + ts: hm + .get("X-Slack-Request-Timestamp") + .map(|x| x.to_str().unwrap_or("")) + .unwrap_or("") + .to_string(), + }) + } +} + +async fn slack_command( + SlackSig { sig, ts }: SlackSig, + Extension(db): Extension, + Extension(rsmq): Extension>, + body: Bytes, +) -> error::Result { + let form: SlackCommand = serde_urlencoded::from_bytes(&body) + .map_err(|_| error::Error::BadRequest("invalid payload".to_string()))?; + + let body = String::from_utf8_lossy(&body); + if let Some(sv) = SLACK_SIGNING_SECRET.as_ref() { + if sv.verify(&ts, &body, &sig).ok().is_none() { + return Err(error::Error::BadRequest("verification failed".to_owned())); + } + } + + let settings = sqlx::query_as!( + WorkspaceSettings, + "SELECT * FROM workspace_settings WHERE slack_team_id = $1", + form.team_id, + ) + .fetch_optional(&db) + .await?; + + if let Some(settings) = settings { + if let Some(path) = &settings.slack_command_script { + let (payload, tag) = if let Some(path) = path.strip_prefix("flow/") { + ( + JobPayload::Flow { + path: path.to_string(), + dedicated_worker: None, + }, + None, + ) + } else { + let path = path.strip_prefix("script/").unwrap_or_else(|| path); + let ( + script_hash, + tag, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + priority, + _delete_after_use, + _timeout, + ) = windmill_common::get_latest_deployed_hash_for_path( + &db, + &settings.workspace_id, + path, + ) + .await?; + ( + JobPayload::ScriptHash { + hash: script_hash, + path: path.to_owned(), + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + language, + dedicated_worker, + priority, + }, + tag, + ) + }; + let mut map = serde_json::Map::new(); + map.insert("text".to_string(), serde_json::Value::String(form.text)); + map.insert( + "response_url".to_string(), + serde_json::Value::String(form.response_url), + ); + let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq); + + let (uuid, tx) = windmill_queue::push( + &db, + tx, + &settings.workspace_id, + payload, + sqlx::types::Json(map), + &form.user_name, + &settings.slack_email, + "g/slack".to_string(), + None, + None, + None, + None, + None, + false, + false, + None, + true, + tag, + None, + None, + None, + ) + .await?; + let url = BASE_URL.read().await.clone(); + tx.commit().await?; + return Ok(format!( + "Job launched. See details at {url}/run/{uuid}?workspace={}", + &settings.workspace_id + )); + } + } + + return Ok(format!( + "workspace not properly configured (did you set the script to trigger in the settings?)" + )); +} + +fn transform_name_to_email(x: String) -> String { + let r = x.replace(' ', "_"); + if r.contains('@') { + return r; + } else { + return format!("{r}@windmill.dev"); + } +} + +#[allow(non_snake_case)] +async fn login_callback( + Path(client_name): Path, + cookies: Cookies, + Extension(db): Extension, + Extension(webhook): Extension, + Json(callback): Json, +) -> error::Result { + let client_w_config = { + let clients = OAUTH_CLIENTS.read().await.logins.clone(); + clients + .get(&client_name) + .ok_or_else(|| error::Error::BadRequest("invalid client".to_string()))? + .clone() + }; + let client = client_w_config.client.to_owned(); + let token_res = + exchange_code::(callback, &cookies, client, &HTTP_CLIENT, None).await; + + if let Ok(token) = token_res { + let token = &token.access_token.to_string(); + + let userinfo_url = client_w_config.userinfo_url.as_ref().ok_or_else(|| { + Error::BadConfig(format!("Missing userinfo_url in client {client_name}")) + })?; + let user = http_get_user_info::(&HTTP_CLIENT, userinfo_url, token).await?; + let email = match client_name.as_str() { + "github" => http_get_user_info::>( + &HTTP_CLIENT, + "https://api.github.com/user/emails", + token, + ) + .await? + .iter() + .find(|x| x.primary && x.verified) + .ok_or(error::Error::BadRequest(format!( + "user does not have any primary and verified address" + )))? + .email + .to_string(), + _ => user + .email + .clone() + .or(user.name.clone().map(transform_name_to_email)) + .ok_or_else(|| { + error::Error::BadRequest( + "email address not fetchable from user info".to_string(), + ) + })?, + } + .to_lowercase(); + + if let Some(domains) = &client_w_config.allowed_domains { + if !domains.iter().any(|d| email.ends_with(d)) { + return Err(error::Error::BadRequest(format!( + "domain is not in the list of allowed domains: {email}, allowed: {domains:#?}", + ))); + } + } + + login_externally( + db, + &email, + client_name, + cookies, + Some(token.to_string()), + Some(user), + ) + .await?; + + webhook.send_instance_event(InstanceEvent::UserSignupOAuth { + email: email.clone(), + }); + + Ok("Successfully logged in".to_string()) + } else { + Err(error::Error::BadRequest(format!( + "failed to exchange code: {:?}", + token_res.err().unwrap() + ))) + } +} + +pub async fn login_externally( + db: DB, + email: &String, + client_name: String, + cookies: Cookies, + token: Option, + user: Option, +) -> error::Result<()> { + let mut tx = db.begin().await?; + let login: Option<(String, String, bool)> = + sqlx::query_as("SELECT email, login_type, super_admin FROM password WHERE email = $1") + .bind(email) + .fetch_optional(&mut *tx) + .await?; + let require_existing_user = + REQUIRE_PREEXISTING_USER_FOR_OAUTH.load(std::sync::atomic::Ordering::Relaxed); + + if let Some((email, login_type, super_admin)) = login { + let login_type = serde_json::json!(login_type); + if require_existing_user || login_type == client_name { + crate::users::create_session_token(&email, super_admin, &mut tx, cookies).await?; + } else { + return Err(error::Error::BadRequest(format!( + "an user with the email associated to this login exists but with a different \ + login type {login_type}" + ))); + } + if token.is_some() { + audit_log( + &mut *tx, + &email, + "oauth.login", + ActionKind::Create, + "global", + Some(&truncate_token(&token.unwrap())[..]), + None, + ) + .await?; + } else { + audit_log( + &mut *tx, + &email, + "oauth.login", + ActionKind::Create, + "global", + None, + None, + ) + .await?; + }; + } else { + if require_existing_user { + return Err(error::Error::BadRequest(format!( + "no user with the email associated to this login exists (windmill is set to only \ + allow oauth logins for existing users)" + ))); + } + + let mut name = user.clone().and_then(|x| x.name); + if (name.is_none() || name == Some(String::new())) && user.is_some() { + name = user.clone().unwrap().displayName; + } + + check_nb_of_user(&db).await?; + + sqlx::query(&format!( + "INSERT INTO password (email, name, company, login_type, verified) VALUES ($1, \ + $2, $3, '{}', true)", + &client_name + )) + .bind(email) + .bind(&name) + .bind(user.map(|x| x.company)) + .execute(&mut *tx) + .await?; + tx.commit().await?; + invite_user_to_all_auto_invite_worspaces(&db, email).await?; + tx = db.begin().await?; + crate::users::create_session_token(email, false, &mut tx, cookies).await?; + audit_log( + &mut *tx, + email, + "oauth.signup", + ActionKind::Create, + "global", + Some(email), + Some([("method", &client_name[..])].into()), + ) + .await?; + + tx = add_to_demo_if_exists(tx, email).await?; + } + tx.commit().await?; + Ok(()) +} + +pub async fn check_nb_of_user(db: &DB) -> error::Result<()> { + #[cfg(feature = "enterprise")] + if matches!(get_license_plan().await, LicensePlan::Enterprise) { + return Ok(()); + } + let nb_users_sso = + sqlx::query_scalar!("SELECT COUNT(*) FROM password WHERE login_type != 'password'",) + .fetch_one(db) + .await?; + if nb_users_sso.unwrap_or(0) >= 10 { + return Err(Error::BadRequest( + "You have reached the maximum number of oauth users accounts (10) without an enterprise license" + .to_string(), + )); + } + + let nb_users = sqlx::query_scalar!("SELECT COUNT(*) FROM password",) + .fetch_one(db) + .await?; + if nb_users.unwrap_or(0) >= 50 { + return Err(Error::BadRequest( + "You have reached the maximum number of accounts (50) without an enterprise license" + .to_string(), + )); + } + return Ok(()); +} + +async fn exchange_code( + callback: OAuthCallback, + cookies: &Cookies, + client: OClient, + http_client: &Client, + extra_params: Option>, +) -> error::Result { + let csrf_state = cookies + .get("csrf") + .map(|x| x.value().to_string()) + .unwrap_or("".to_string()); + if callback.state != csrf_state { + return Err(error::Error::BadRequest("csrf did not match".to_string())); + } + + let mut token_url = client.exchange_code(callback.code); + + if let Some(extra_params) = extra_params { + for (key, value) in extra_params { + token_url = token_url.param(key, value) + } + } + + token_url + .with_client(http_client) + .execute::() + .await + .map_err(|e| error::Error::InternalErr(format!("{:?}", e))) } async fn list_supabase(headers: HeaderMap) -> impl IntoResponse { @@ -174,39 +1244,115 @@ async fn list_supabase(headers: HeaderMap) -> impl IntoResponse { Ok((status_code, StreamBody::new(stream))) as error::Result<(StatusCode, StreamBody<_>)> } -pub async fn check_nb_of_user(db: &DB) -> error::Result<()> { - let nb_users_sso = - sqlx::query_scalar!("SELECT COUNT(*) FROM password WHERE login_type != 'password'",) - .fetch_one(db) - .await?; - if nb_users_sso.unwrap_or(0) >= 10 { - return Err(error::Error::BadRequest( - "You have reached the maximum number of oauth users accounts (10) without an enterprise license" - .to_string(), - )); +#[derive(Deserialize)] +pub struct GHEmailInfo { + email: String, + verified: bool, + primary: bool, +} + +async fn http_get_user_info( + http_client: &Client, + url: &str, + token: &str, +) -> error::Result { + let res = http_client + .get(url) + .bearer_auth(token) + .send() + .await + .map_err(to_anyhow) + .context("failed to fetch user info")?; + if !res.status().is_success() { + tracing::debug!( + "The bearer token of the failed oauth user info exchange is: {}", + token + ); + return Err(error::Error::BadConfig(format!( + "The user info endpoint responded with non 200: {}\n{}\n{}", + res.status(), + res.headers() + .iter() + .map(|x| format!("{}: {}", x.0.as_str(), x.1.to_str().unwrap_or_default())) + .collect::>() + .join("\n"), + res.text().await.unwrap_or_default(), + ))); + } + Ok(res + .json::() + .await + .map_err(to_anyhow) + .context("failed to decode json from user info")?) +} + +fn oauth_redirect( + clients: &HashMap, + client_name: String, + cookies: Cookies, + scopes: Option>, + extra_params: Option>, + is_secure: bool, +) -> error::Result { + let client_w_scopes = clients + .get(&client_name) + .ok_or_else(|| error::Error::BadRequest("client not found".to_string()))?; + let state = State::new_random(); + let mut client = client_w_scopes.client.clone(); + let scopes_iter = if let Some(scopes) = scopes { + scopes + } else { + client_w_scopes.scopes.clone() + }; + + for scope in scopes_iter.iter() { + client.add_scope(scope); } - let nb_users = sqlx::query_scalar!("SELECT COUNT(*) FROM password",) - .fetch_one(db) - .await?; - if nb_users.unwrap_or(0) >= 50 { - return Err(error::Error::BadRequest( - "You have reached the maximum number of accounts (50) without an enterprise license" - .to_string(), - )); + let mut auth_url = client.authorize_url(&state); + + if let Some(extra_params) = extra_params { + let mut query_string = auth_url.query_pairs_mut(); + for (key, value) in extra_params { + query_string.append_pair(&key, &value); + } } - return Ok(()); + + set_cookie(&state, cookies, is_secure); + Ok(Redirect::to(auth_url.as_str())) +} + +fn set_cookie(state: &State, cookies: Cookies, is_secure: bool) { + let csrf = state.to_base64(); + let mut cookie = Cookie::new("csrf", csrf); + cookie.set_secure(is_secure); + cookie.set_same_site(Some(cookie::SameSite::Lax)); + cookie.set_http_only(true); + cookie.set_path("/"); + cookies.add(cookie); } #[derive(Clone, Debug)] pub struct SlackVerifier { - _mac: HmacSha256, + mac: HmacSha256, } impl SlackVerifier { pub fn new>(secret: S) -> anyhow::Result { HmacSha256::new_from_slice(secret.as_ref()) - .map(|mac| SlackVerifier { _mac: mac }) + .map(|mac| SlackVerifier { mac }) .map_err(|_| anyhow::anyhow!("invalid secret")) } + + pub fn verify(&self, ts: &str, body: &str, exp_sig: &str) -> anyhow::Result<()> { + let basestring = format!("v0:{}:{}", ts, body); + let mut mac = self.mac.clone(); + + mac.update(basestring.as_bytes()); + let sig = format!("v0={}", hex::encode(mac.finalize().into_bytes())); + if sig != exp_sig { + Err(anyhow::anyhow!("signature mismatch"))?; + } + Ok(()) + } } diff --git a/backend/windmill-api/src/saml_ee.rs b/backend/windmill-api/src/saml_ee.rs index b3f1d4653c..783b80a5ad 100644 --- a/backend/windmill-api/src/saml_ee.rs +++ b/backend/windmill-api/src/saml_ee.rs @@ -1,25 +1,190 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2023 - * This file and its contents are licensed under the AGPLv3 License. - * Please see the included NOTICE for copyright information and - * LICENSE-AGPL for a copy of the license. - */ #![allow(non_snake_case)] +#[cfg(feature = "enterprise_saml")] +use axum::response::Redirect; +use axum::Json; use axum::{routing::post, Router}; +#[cfg(feature = "enterprise_saml")] +use axum::{Extension, Form}; +use std::sync::Arc; +#[cfg(feature = "enterprise_saml")] +use samael::metadata::{ContactPerson, ContactType, EntityDescriptor}; +#[cfg(feature = "enterprise_saml")] +use samael::service_provider::{ServiceProvider, ServiceProviderBuilder}; + +#[cfg(feature = "enterprise_saml")] +use serde::Deserialize; +#[cfg(feature = "enterprise_saml")] +use tower_cookies::Cookies; +#[cfg(feature = "enterprise_saml")] +use windmill_common::error::{Error, Result}; + +#[cfg(feature = "enterprise_saml")] +use crate::db::DB; +#[cfg(feature = "enterprise_saml")] +use crate::oauth2_ee::login_externally; +#[cfg(feature = "enterprise_saml")] +use crate::BASE_URL; + +#[cfg(feature = "enterprise_saml")] +use windmill_common::ee::{get_license_plan, LicensePlan}; + +#[cfg(feature = "enterprise_saml")] +#[derive(Clone)] +pub struct ServiceProviderExt(pub Option); + +#[cfg(not(feature = "enterprise_saml"))] pub struct ServiceProviderExt(); +#[cfg(not(feature = "enterprise_saml"))] pub async fn build_sp_extension() -> anyhow::Result { return Ok(ServiceProviderExt()); } -pub fn global_service() -> Router { - Router::new().route("/acs", post(acs)) +#[cfg(feature = "enterprise_saml")] +pub async fn build_sp_extension() -> anyhow::Result { + use crate::SAML_METADATA; + + if !matches!(get_license_plan().await, LicensePlan::Enterprise) { + return Ok(ServiceProviderExt(None)); + } + if let Some(url_metadata) = SAML_METADATA.read().await.clone() { + let idp_metadata: EntityDescriptor = load_metadata(url_metadata.as_str()).await?; + + // let pub_key = openssl::x509::X509::from_pem("")?; + // let private_key = openssl::rsa::Rsa::private_key_from_pem("")?; + + let acs_url = format!("{}/api/saml/acs", BASE_URL.read().await.clone()); + let sp = ServiceProviderBuilder::default() + .entity_id("windmill".to_string()) + // .key(private_key) + // .certificate(pub_key) + .allow_idp_initiated(true) + .contact_person(ContactPerson { + sur_name: Some("Ruben Fiszel ".to_string()), + contact_type: Some(ContactType::Technical.value().to_string()), + ..ContactPerson::default() + }) + .idp_metadata(idp_metadata) + .acs_url(acs_url.clone()) + .build()?; + + tracing::info!("SAML Configured - ACS url is {}", acs_url); + Ok(ServiceProviderExt(Some(sp))) + } else { + Ok(ServiceProviderExt(None)) + } } -pub async fn acs() -> String { - // Implementation is not open source as it is a Windmill Enterprise Edition feature - "SAML available only in enterprise version".to_string() +#[cfg(not(feature = "enterprise_saml"))] +pub async fn generate_redirect_url( + _service_provider: Arc, +) -> anyhow::Result> { + return Ok(None); +} + +#[cfg(feature = "enterprise_saml")] +pub async fn generate_redirect_url( + service_provider: Arc, +) -> anyhow::Result> { + if let Some(sp) = &service_provider.0 { + let url = sp + .idp_metadata + .idp_sso_descriptors + .clone() + .unwrap_or_default() + .get(0) + .and_then(|x| x.single_sign_on_services.get(0).map(|x| x.location.clone())); + + let authn_req = sp + .make_authentication_request(url.unwrap_or_default().as_str()) + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + let redirect_url = authn_req + .redirect(BASE_URL.read().await.clone().as_str()) + .map_err(|e| anyhow::anyhow!(e.to_string()))? + .map(|u| u.to_string()); + + tracing::debug!( + "SAML Configured, sso login link at: {:?}", + redirect_url.clone() + ); + Ok(redirect_url) + } else { + Ok(None) + } +} + +#[cfg(feature = "enterprise_saml")] +async fn load_metadata(saml_metadata: &str) -> anyhow::Result { + let resp = if saml_metadata.starts_with("http") { + reqwest::get(saml_metadata).await?.text().await? + } else { + saml_metadata.to_string() + }; + let idp_metadata: EntityDescriptor = samael::metadata::de::from_str(&resp).map_err(|e| { + Error::BadRequest(format!( + "SAML metadata could not be decoded: {e}. Content was: {resp}" + )) + })?; + return Ok(idp_metadata); +} +#[cfg(feature = "enterprise_saml")] +pub async fn test_metadata(Json(saml_metadata): Json) -> Result { + let idp_metadata = load_metadata(&saml_metadata).await?; + Ok(format!("{:?}", idp_metadata)) +} + +#[cfg(feature = "enterprise_saml")] +pub fn global_service() -> Router { + Router::new() + .route("/acs", post(acs)) + .route("/test_metadata", post(test_metadata)) +} + +#[cfg(not(feature = "enterprise_saml"))] +pub fn global_service() -> Router { + Router::new() +} + +#[cfg(feature = "enterprise_saml")] +#[derive(Deserialize)] +pub struct SamlForm { + pub SAMLResponse: Option, +} + +#[cfg(feature = "enterprise_saml")] +pub async fn acs( + Extension(db): Extension, + cookies: Cookies, + Extension(se): Extension>, + Form(s): Form, +) -> Result { + if matches!(get_license_plan().await, LicensePlan::Pro) { + return Err(Error::BadRequest( + "SAML not available in the pro plan".to_string(), + )); + }; + if let Some(sp_m) = &se.0 { + let sp = sp_m.clone(); + if let Some(encoded_resp) = s.SAMLResponse { + tracing::info!("{:?}", encoded_resp); + let t = sp.parse_base64_response(&encoded_resp, None).map_err(|e| { + Error::BadRequest(format!("Error parsing acs request as base64: {e:?}")) + })?; + + if let Some(email) = t.subject.and_then(|x| x.name_id.map(|x| x.value)) { + login_externally(db, &email, "saml".to_string(), cookies, None, None).await?; + Ok(Redirect::to("/")) + } else { + Err(Error::BadRequest( + "email not found in saml response".to_string(), + )) + } + } else { + Err(Error::BadRequest("SAMLResponse not found".to_string())) + } + } else { + Err(Error::BadConfig("SAML not configured".to_string())) + } } diff --git a/backend/windmill-api/src/scim_ee.rs b/backend/windmill-api/src/scim_ee.rs index 8552405296..edae483faa 100644 --- a/backend/windmill-api/src/scim_ee.rs +++ b/backend/windmill-api/src/scim_ee.rs @@ -1,3 +1,5 @@ +#![allow(non_snake_case)] + /* * Author: Ruben Fiszel * Copyright: Windmill Labs, Inc 2023 @@ -6,17 +8,721 @@ * LICENSE-AGPL for a copy of the license. */ -use axum::{middleware::Next, response::Response, routing::get, Router}; -use hyper::Request; +use axum::{ + extract::Query, + middleware::Next, + response::{IntoResponse, Response}, + routing::get, + Extension, Router, +}; +use bytes::{BufMut, BytesMut}; +use hyper::{header, http::HeaderValue, Request, StatusCode}; +use mime_guess::mime; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use sql_builder::SqlBuilder; +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; + +use crate::{db::DB, SCIM_TOKEN}; + +#[cfg(feature = "enterprise")] pub fn global_service() -> Router { - Router::new().route("/ee", get(ee)) + Router::new() + .route("/Users", get(get_users).post(create_user)) + .route("/Groups", get(get_groups).post(create_group)) + .route( + "/Groups/:id", + get(get_group) + .put(update_group) + .patch(update_group) + .delete(delete_group), + ) + .route( + "/Users/:username", + get(get_user) + .put(update_user) + .patch(update_user) + .delete(delete_user), + ) } -pub async fn ee() -> String { - return "Enterprise Edition".to_string(); +#[cfg(not(feature = "enterprise"))] +pub fn global_service() -> Router { + Router::new().route("/Users", get(get_users)) } +#[derive(Debug, Clone, Copy, Default)] +pub struct JsonScim(pub T); + pub async fn has_scim_token(request: Request, next: Next) -> Response { - return next.run(request).await; + let header = request.headers().get("Authorization"); + if let Some(header) = header { + if let Ok(header) = header.to_str() { + if header.starts_with("Bearer ") { + let token = header.trim_start_matches("Bearer "); + if let Some(scim_token) = SCIM_TOKEN.read().await.clone() { + if token == scim_token { + return next.run(request).await; + } + } + } + } + } + return ( + StatusCode::UNAUTHORIZED, + [( + header::CONTENT_TYPE, + HeaderValue::from_static(mime::TEXT_PLAIN_UTF_8.as_ref()), + )], + "Unauthorized", + ) + .into_response(); +} + +impl IntoResponse for JsonScim +where + T: Serialize, +{ + fn into_response(self) -> Response { + // Use a small initial capacity of 128 bytes like serde_json::to_vec + // https://docs.rs/serde_json/1.0.82/src/serde_json/ser.rs.html#2189 + let mut buf = BytesMut::with_capacity(128).writer(); + match serde_json::to_writer(&mut buf, &self.0) { + Ok(()) => ( + [( + header::CONTENT_TYPE, + HeaderValue::from_static("application/scim+json"), + )], + buf.into_inner().freeze(), + ) + .into_response(), + Err(err) => ( + StatusCode::INTERNAL_SERVER_ERROR, + [( + header::CONTENT_TYPE, + HeaderValue::from_static(mime::TEXT_PLAIN_UTF_8.as_ref()), + )], + err.to_string(), + ) + .into_response(), + } + } +} + +#[derive(Serialize, Debug)] +struct User { + id: String, + userName: String, + active: bool, +} +pub fn resource_response(schema: &str, resources: Vec) -> JsonScim +where + S: Serialize, +{ + return JsonScim(json!({ + "schemas": [schema], + "totalResults": resources.len(), + "Resources": resources, + "startIndex": 1, + "itemsPerPage": 100, + })); +} + +#[derive(Deserialize)] +pub struct ScimQuery { + startIndex: Option, + count: Option, + filter: Option, +} + +pub async fn get_users( + Extension(db): Extension, + Query(query): Query, +) -> Result> { + let mut sqlb = SqlBuilder::select_from("password") + .fields(&["email"]) + .limit(query.count.unwrap_or(100000)) + .offset(query.startIndex.map(|x| x - 1).unwrap_or(0)) + .clone(); + + tracing::info!("SCIM filter: {:?}", query.filter); + + if let Some(filter) = query.filter { + let filter = filter + .replace("userName", "email") + .replace("eq", "=") + .replace("\"", "'"); + sqlb.and_where(&filter); + } + + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let users = sqlx::query_scalar(&sql) + .fetch_all(&db) + .await? + .into_iter() + .map(|x: String| User { + id: x.clone(), + userName: x, + active: true, + }) + .collect(); + tracing::info!("SCIM users: {:?}", users); + Ok(resource_response( + "urn:ietf:params:scim:api:messages:2.0:ListResponse", + users, + )) +} + +#[cfg(feature = "enterprise")] +#[derive(Deserialize, Debug)] +pub struct CreateUser { + userName: String, +} +#[cfg(feature = "enterprise")] +pub async fn create_user( + Extension(db): Extension, + Json(body): Json, +) -> Result> { + use crate::workspaces::invite_user_to_all_auto_invite_worspaces; + + tracing::info!("SCIM creating user: {:?}", body); + let email = body.userName.clone(); + sqlx::query!( + "INSERT INTO password (email, login_type, verified) VALUES ($1, 'saml', true) ON CONFLICT DO NOTHING", + &email, + ).execute(&db).await?; + invite_user_to_all_auto_invite_worspaces(&db, &email).await?; + Ok(JsonScim(json!({ + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"], + "id": body.userName, + "userName": body.userName, + "active": true + }))) +} + +#[cfg(feature = "enterprise")] +pub async fn get_user( + Extension(db): Extension, + Path(username): Path, +) -> Result> { + // This query looks dumb but it's a way to validate the user exists and it sets the stage if we want to support more fields in the future + let username_from_db = + sqlx::query_scalar!("SELECT email FROM password WHERE email = $1", username) + .fetch_optional(&db) + .await?; + let username_from_db = not_found_if_none(username_from_db, "User", username)?; + Ok(JsonScim(json!({ + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"], + "id": username_from_db, + "userName": username_from_db, + "active": true + }))) +} + +#[cfg(feature = "enterprise")] +pub async fn delete_user(Extension(db): Extension, Path(username): Path) -> Result<()> { + sqlx::query!("DELETE FROM email_to_igroup WHERE email = $1", username) + .execute(&db) + .await?; + sqlx::query!("DELETE FROM password WHERE email = $1", username) + .execute(&db) + .await?; + Ok(()) +} + +#[cfg(feature = "enterprise")] +#[derive(Deserialize, Debug)] +pub struct UpdateUser { + pub schemas: Vec, + pub Operations: Option>, +} + +#[cfg(feature = "enterprise")] +pub async fn update_user( + Extension(db): Extension, + Path(username): Path, + Json(body): Json, +) -> Result> { + tracing::info!("SCIM updating user: {:?}", body); + let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; + if body.schemas.len() == 1 { + let schema = body.schemas.get(0).unwrap(); + let mut new_username = username.clone(); + if schema == "urn:ietf:params:scim:schemas:core:2.0:User" { + // Since we only care about the username at this point, all we can do here is check that the user currently exists + let username_from_db = + sqlx::query_scalar!("SELECT email FROM password WHERE email = $1", username) + .fetch_optional(&mut *tx) + .await?; + tx.commit().await?; + let username_from_db = not_found_if_none(username_from_db, "User", username)?; + Ok(JsonScim(json!({ + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"], + "id": username_from_db, + "userName": username_from_db, + "active": true + }))) + } else if schema == "urn:ietf:params:scim:api:messages:2.0:PatchOp" { + for operation in body.Operations.unwrap_or_default() { + match operation.op.as_str() { + "Replace" => match operation.path.as_str() { + "userName" => { + if let Some(username_raw) = operation.value { + let new_username_tmp = serde_json::from_value::( + username_raw, + ) + .map_err(|err| { + Error::InternalErr(format!("Error parsing user name: {}", err)) + })?; + // TODO: that's fishy, maybe we shouldn't allow updating the email + // I think it makes sense in a world where you rely on external IDs, but since it's not + // our case, it makes things a bit borderline + sqlx::query!( + "UPDATE password SET email = $1 WHERE email = $2", + new_username_tmp, + username + ) + .execute(&mut *tx) + .await?; + sqlx::query!( + "UPDATE email_to_igroup SET email = $1 WHERE email = $2", + new_username_tmp, + username + ) + .execute(&mut *tx) + .await?; + new_username = new_username_tmp + } + } + unknown => { + tracing::info!("Unsupported patch operation on user: {} with path {}. It will be ignored", new_username, unknown) + } + }, + "Add" => match operation.path.as_str() { + unknown => { + tracing::info!("Unsupported patch-add operation on user: {} with path {}. It will be ignored", new_username, unknown) + } + }, + "Remove" => match operation.path.as_str() { + unknown => { + tracing::info!("Unsupported patch-add operation on user: {} with path {}. It will be ignored", new_username, unknown) + } + }, + unknown => { + return Err(Error::BadRequest( + format!("Invalid operation: {}", unknown).to_string(), + )) + } + } + } + tx.commit().await?; + Ok(JsonScim(json!({ + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"], + "id": new_username, + "userName": new_username, + "active": true + }))) + } else { + Err(Error::BadRequest("Invalid schemas".to_string())) + } + } else { + Err(Error::BadRequest("Invalid schemas".to_string())) + } +} + +#[cfg(feature = "enterprise")] +pub async fn get_groups( + Extension(db): Extension, + Query(query): Query, +) -> Result> { + let mut sqlb = SqlBuilder::select_from("instance_group") + .fields(&[ + "name", + "external_id", + "scim_display_name", + "array_remove(array_agg(email_to_igroup.email), null) as emails", + ]) + .left() + .join("email_to_igroup") + .on("instance_group.name = email_to_igroup.igroup") + .group_by("name, external_id") + .limit(query.count.unwrap_or(100000)) + .offset(query.startIndex.map(|x| x - 1).unwrap_or(0)) + .clone(); + + if let Some(filter) = query.filter { + let filter = filter + .replace("displayName", "scim_display_name") + .replace("eq", "=") + .replace("\"", "'"); + sqlb.and_where(&filter); + } + + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let groups = sqlx::query_as::<_, Group>(&sql) + .fetch_all(&db) + .await? + .into_iter() + .map(|x| group_response(x).0) + .collect(); + Ok(resource_response( + "urn:ietf:params:scim:api:messages:2.0:ListResponse", + groups, + )) +} + +// { +// "schemas": [], +// "id": "abf4dd94-a4c0-4f67-89c9-76b03340cb9b", +// "displayName": "Test SCIMv2", +// "members": [], +// "meta": { +// "resourceType": "Group" +// } +// } + +#[cfg(feature = "enterprise")] +#[derive(Serialize, sqlx::FromRow)] +pub struct Group { + name: String, + emails: Option>, + external_id: Option, + scim_display_name: Option, +} +#[cfg(feature = "enterprise")] +fn group_response(group: Group) -> JsonScim { + let external_id = group + .external_id + .unwrap_or_else(|| convert_name(&group.name)); + let json = json!({ + "schemas": ["urn:ietf:params:scim:schemas:core:2.0:Group"], + "displayName": group.scim_display_name.unwrap_or_default(), + "id": external_id, + "objectId": external_id, + "members": group.emails.unwrap_or_default().into_iter().map(|x| json!({"value": x, "display": x})).collect::>(), + "meta": { + "resourceType": "Group" + } + }); + tracing::info!("SCIM group: {:?}", json); + return JsonScim(json); +} + +#[cfg(feature = "enterprise")] +pub async fn get_group( + Extension(db): Extension, + Path(id): Path, +) -> Result> { + let group= sqlx::query_as!( + Group, + "SELECT name, external_id, scim_display_name, array_remove(array_agg(email_to_igroup.email), null) as emails FROM email_to_igroup RIGHT JOIN instance_group ON instance_group.name = email_to_igroup.igroup WHERE external_id = $1 group by name, external_id", + id + ) + .fetch_optional(&db) + .await?; + let group = not_found_if_none(group, "Group", id)?; + Ok(group_response(group)) +} + +// { +// "schemas": ["urn:ietf:params:scim:schemas:core:2.0:Group"], +// "displayName": "Test SCIMv2", +// "members": [] +// } + +#[cfg(feature = "enterprise")] +#[derive(Deserialize, Debug)] +pub struct CreateGroup { + pub displayName: String, + pub members: Vec, +} + +#[cfg(feature = "enterprise")] +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Member { + pub value: String, + pub display: String, +} + +#[cfg(feature = "enterprise")] +pub async fn create_group( + Extension(db): Extension, + Json(body): Json, +) -> Result> { + use uuid::Uuid; + + tracing::info!("SCIM creating group: {:?}", body); + let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; + let id = Uuid::new_v4().to_string(); + let scim_display_name = Some(body.displayName.clone()); + let name = convert_name(&body.displayName.clone()); + sqlx::query!( + "INSERT INTO instance_group (name, scim_display_name, external_id) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", + name, + body.displayName, + id, + ) + .execute(&mut *tx) + .await?; + tracing::info!( + "SCIM created group: {} with external_id: {} (display name: {})", + name, + id, + body.displayName + ); + for member in &body.members { + sqlx::query!( + "INSERT INTO email_to_igroup (email, igroup) VALUES ($1, $2) ON CONFLICT DO NOTHING", + convert_name(&member.display), + name, + ) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(group_response(Group { + external_id: Some(id), + name, + scim_display_name, + emails: Some( + body.members + .clone() + .into_iter() + .map(|x| x.display.clone()) + .collect(), + ), + })) +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Operation { + pub op: String, + pub path: String, + pub value: Option, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct UpdateMembersOperationValue { + pub value: String, +} + +#[cfg(feature = "enterprise")] +#[derive(Deserialize, Debug)] +pub struct UpdateGroup { + pub schemas: Vec, + pub displayName: Option, + pub members: Option>, + pub Operations: Option>, +} + +#[cfg(feature = "enterprise")] +pub async fn update_group( + Extension(db): Extension, + Path(id): Path, + Json(body): Json, +) -> Result> { + tracing::info!("SCIM updating group: {:?}", body); + let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; + if body.schemas.len() == 1 { + let schema = body.schemas.get(0).unwrap(); + if schema == "urn:ietf:params:scim:schemas:core:2.0:Group" { + let group= sqlx::query_as!( + Group, + "SELECT name, external_id, scim_display_name, array_remove(array_agg(email_to_igroup.email), null) as emails FROM email_to_igroup RIGHT JOIN instance_group ON instance_group.name = email_to_igroup.igroup WHERE external_id = $1 GROUP BY name", + id + ) + .fetch_optional(&db) + .await?; + let mut group = not_found_if_none(group, "Group", id.clone())?; + + sqlx::query!("DELETE FROM email_to_igroup WHERE igroup = $1", group.name) + .execute(&mut *tx) + .await?; + + let new_name = if let Some(name) = body.displayName.clone() { + let new_name = convert_name(name.as_str()); + sqlx::query!( + "UPDATE instance_group SET scim_display_name = $1, name = $2 where external_id = $3", + name, + new_name, + id + ) + .execute(&mut *tx) + .await?; + group.scim_display_name = Some(name); + new_name + } else { + group.name.clone() + }; + + if let Some(members) = body.members.clone() { + let mut emails = vec![]; + for m in members { + emails.push(m.display.clone()); + sqlx::query!( + "INSERT INTO email_to_igroup (email, igroup) VALUES ($1, $2) ON CONFLICT DO NOTHING", + m.display, + new_name, + ) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + group.emails = Some(emails); + Ok(group_response(group)) + } else { + Err(Error::BadRequest("expected members".to_string())) + } + } else if schema == "urn:ietf:params:scim:api:messages:2.0:PatchOp" { + let group_name_opt = + sqlx::query_scalar!("SELECT name FROM instance_group WHERE external_id = $1", id) + .fetch_optional(&db) + .await?; + let group_name = not_found_if_none(group_name_opt, "Group", id.clone())?; + let mut new_external_id = id; + for operation in body.Operations.unwrap_or_default() { + match operation.op.as_str() { + "Replace" => match operation.path.as_str() { + "displayName" => { + if let Some(name_raw) = operation.value { + let name = + serde_json::from_value::(name_raw).map_err(|err| { + Error::InternalErr(format!( + "Error parsing group name: {}", + err + )) + })?; + let new_name = convert_name(name.as_str()); + sqlx::query!( + "UPDATE instance_group SET scim_display_name = $1, name = $2 where external_id = $3", + name, + new_name, + new_external_id + ) + .execute(&mut *tx) + .await?; + } + } + unknown => { + tracing::info!("Unsupported patch operation on group: {} with path {}. It will be ignored", new_external_id, unknown) + } + }, + "Add" => match operation.path.as_str() { + "members" => { + if let Some(value) = operation.value { + let parsed_ops = serde_json::from_value::< + Vec, + >(value) + .map_err(|err| { + Error::InternalErr(format!("Error parsing operation: {}", err)) + })?; + for op in parsed_ops { + sqlx::query!( + "INSERT INTO email_to_igroup (email, igroup) VALUES ($1, $2) ON CONFLICT DO NOTHING", + op.value, + group_name, + ) + .execute(&mut *tx) + .await?; + } + } + } + "externalId" => { + if let Some(external_id_raw) = operation.value { + let external_id = serde_json::from_value::(external_id_raw) + .map_err(|err| { + Error::InternalErr(format!( + "Error parsing group external ID: {}", + err + )) + })?; + sqlx::query!( + "UPDATE instance_group SET external_id = $1 WHERE name = $2", + external_id, + group_name + ) + .execute(&mut *tx) + .await?; + new_external_id = external_id + } + } + unknown => { + tracing::info!("Unsupported patch-add operation on group: {} with path {}. It will be ignored", new_external_id, unknown) + } + }, + "Remove" => match operation.path.as_str() { + "members" => { + if let Some(value) = operation.value { + let parsed_ops = serde_json::from_value::< + Vec, + >(value) + .map_err(|err| { + Error::InternalErr(format!("Error parsing operation: {}", err)) + })?; + for op in parsed_ops { + sqlx::query!( + "DELETE FROM email_to_igroup WHERE email = $1 AND igroup = $2", + op.value, + group_name + ) + .execute(&mut *tx) + .await?; + } + } + } + unknown => { + tracing::info!("Unsupported patch-delete operation on group: {} with path {}. It will be ignored", new_external_id, unknown) + } + }, + unknown => { + return Err(Error::BadRequest( + format!("Invalid operation: {}", unknown).to_string(), + )) + } + } + } + tx.commit().await?; + let group= sqlx::query_as!( + Group, + "SELECT name, external_id, scim_display_name, array_remove(array_agg(email_to_igroup.email), null) as emails FROM email_to_igroup RIGHT JOIN instance_group ON instance_group.name = email_to_igroup.igroup WHERE external_id = $1 GROUP BY name", + new_external_id + ) + .fetch_optional(&db) + .await?; + let group = not_found_if_none(group, "Group", new_external_id.clone())?; + Ok(group_response(group)) + } else { + Err(Error::BadRequest("Invalid schemas".to_string())) + } + } else { + Err(Error::BadRequest("Invalid schemas".to_string())) + } +} + +#[cfg(feature = "enterprise")] +pub async fn delete_group(Extension(db): Extension, Path(id): Path) -> Result<()> { + tracing::info!("SCIM delete group: {:?}", id); + let group = sqlx::query_scalar!("SELECT name FROM instance_group WHERE external_id = $1", id) + .fetch_optional(&db) + .await?; + let group = not_found_if_none(group, "Group", id.clone())?; + + sqlx::query!("DELETE FROM email_to_igroup WHERE igroup = $1", group) + .execute(&db) + .await?; + sqlx::query!("DELETE FROM instance_group WHERE name = $1", group) + .execute(&db) + .await?; + Ok(()) +} + +#[cfg(feature = "enterprise")] +fn convert_name(name: &str) -> String { + name.replace(" ", "_").to_lowercase() } diff --git a/backend/windmill-api/src/stripe_ee.rs b/backend/windmill-api/src/stripe_ee.rs index 1aea2ecd2d..1de72ccf8c 100644 --- a/backend/windmill-api/src/stripe_ee.rs +++ b/backend/windmill-api/src/stripe_ee.rs @@ -1,7 +1,306 @@ #[cfg(feature = "stripe")] -use axum::Router; +use axum::response::Redirect; +#[cfg(feature = "stripe")] +use axum::{routing::get, Router}; +#[cfg(feature = "stripe")] +use chrono::{Datelike, TimeZone, Timelike}; +#[cfg(feature = "stripe")] +use std::str::FromStr; +#[cfg(feature = "stripe")] +use stripe::CustomerId; + +#[cfg(feature = "stripe")] +use crate::db::{ApiAuthed, DB}; + +#[cfg(feature = "stripe")] +use crate::BASE_URL; +#[cfg(feature = "stripe")] +use axum::{ + extract::{Extension, Path, Query}, + Json, +}; +#[cfg(feature = "stripe")] +use chrono::Utc; + +#[cfg(feature = "stripe")] +use windmill_common::{ + error::{to_anyhow, Error, JsonResult, Result}, + utils::require_admin, +}; + +#[cfg(feature = "stripe")] +use serde::{Deserialize, Serialize}; + +#[cfg(feature = "stripe")] +#[cfg(feature = "enterprise")] +lazy_static::lazy_static! { + pub static ref STRIPE_KEY: Option = std::env::var("STRIPE_KEY").ok(); +} #[cfg(feature = "stripe")] pub fn add_stripe_routes(router: Router) -> Router { - return router; + if STRIPE_KEY.is_none() { + return router; + } else { + tracing::info!("stripe enabled"); + + return router + .route("/premium_info", get(premium_info)) + .route("/checkout", get(stripe_checkout)) + .route("/billing_portal", get(stripe_portal)); + } } + +#[cfg(feature = "stripe")] +async fn premium_info( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, +) -> JsonResult { + require_admin(authed.is_admin, &authed.username)?; + let mut tx = db.begin().await?; + let row = sqlx::query!( + r#"SELECT premium, usage.usage as "usage?", workspace_settings.customer_id, workspace_settings.plan FROM workspace LEFT JOIN workspace_settings ON workspace_settings.workspace_id = $1 LEFT JOIN usage ON usage.id = $1 AND month_ = EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date) AND usage.is_workspace IS true WHERE workspace.id = $1"#, + &w_id + ) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + let result = PremiumWorkspaceInfo { premium: row.premium, usage: row.usage, seats: None }; + #[cfg(feature = "enterprise")] + let mut result = result; + #[cfg(feature = "enterprise")] + if row.premium && row.plan == Some("team".to_string()) { + let customer_id = row.customer_id.ok_or(Error::InternalErr(format!( + "no customer id for workspace {}", + w_id + )))?; + let client = stripe::Client::new( + STRIPE_KEY + .clone() + .ok_or(Error::InternalErr(format!("stripe key not set")))?, + ); + let customer_id = CustomerId::from_str(&customer_id).map_err(to_anyhow)?; + let subscriptions = stripe::Subscription::list( + &client, + &stripe::ListSubscriptions { + customer: Some(customer_id.clone()), + limit: Some(1), + ..Default::default() + }, + ) + .await + .map_err(to_anyhow)?; + if subscriptions.data.len() > 1 { + return Err(Error::InternalErr(format!("multiple subscriptions for customer {}, please contact us at ccontact@windmill.dev", customer_id))); + } + let subscription = subscriptions.data.get(0).ok_or_else(|| { + Error::InternalErr(format!("no subscription for customer {}", customer_id)) + })?; + result.seats = subscription + .items + .data + .iter() + .filter_map(|item| { + item.price + .clone() + .map(|p| { + p.metadata.map(|m| { + m.get("plan") + .filter(|plan| plan == &"team") + .map(|_| item.quantity.map(|x| x as i32)) + }) + }) + .flatten() + .flatten() + .flatten() + }) + .collect::>() + .get(0) + .copied(); + } + + Ok(Json(result)) +} + +#[cfg(feature = "stripe")] +#[derive(Deserialize)] +struct PlanQuery { + plan: String, + seats: Option, +} + +#[cfg(feature = "stripe")] +async fn stripe_checkout( + authed: ApiAuthed, + Path(w_id): Path, + Query(plan): Query, + Extension(db): Extension, +) -> Result { + // #[cfg(feature = "enterprise")] + { + require_admin(authed.is_admin, &authed.username)?; + let client = stripe::Client::new( + STRIPE_KEY + .clone() + .ok_or(Error::InternalErr(format!("stripe key not set")))?, + ); + let base_url = BASE_URL.read().await.clone(); + let success_rd = format!("{}/workspace_settings/checkout?success=true", base_url); + let failure_rd = format!("{}/workspace_settings/checkout?success=false", base_url); + let checkout_session = { + let mut params = stripe::CreateCheckoutSession::new(&success_rd); + params.mode = Some(stripe::CheckoutSessionMode::Subscription); + params.cancel_url = Some(&failure_rd); + params.line_items = match plan.plan.as_str() { + "team" => Some(vec![stripe::CreateCheckoutSessionLineItems { + quantity: Some(plan.seats.unwrap_or(1) as u64), + price: Some("price_1NCNOgGU3NdFi9eLuG4fZuEP".to_string()), + adjustable_quantity: Some( + stripe::CreateCheckoutSessionLineItemsAdjustableQuantity { + enabled: true, + minimum: Some(1), + ..Default::default() + }, + ), + ..Default::default() + }]), + _ => Err(Error::BadRequest("invalid plan".to_string()))?, + }; + params.client_reference_id = Some(&w_id); + let customer_id = sqlx::query_scalar!( + "SELECT customer_id FROM workspace_settings WHERE workspace_id = $1", + &w_id + ) + .fetch_one(&db) + .await?; + + match customer_id { + Some(customer_id) => { + params.customer = Some(CustomerId::from_str(&customer_id).map_err(to_anyhow)?) + } + _ => params.customer_email = Some(&authed.email), + } + + let now = Utc::now(); + params.subscription_data = Some(stripe::CreateCheckoutSessionSubscriptionData { + metadata: { + let mut map = std::collections::HashMap::new(); + map.insert("workspace_id".to_string(), w_id.clone()); + Some(map) + }, + billing_cycle_anchor: if now.day() == 1 && now.hour() < 12 { + // no need to prorate so close to the billing cycle renew date + None + } else { + // first of the next month (and possibly next year) at noon UTC + let date = if now.month() == 12 { + Utc.with_ymd_and_hms(now.year() + 1, 1, 1, 12, 0, 0) + .single() + .unwrap() + } else { + Utc.with_ymd_and_hms(now.year(), now.month() + 1, 1, 12, 0, 0) + .single() + .unwrap() + }; + Some(date.timestamp()) + }, + ..Default::default() + }); + + stripe::CheckoutSession::create(&client, params) + .await + .map_err(to_anyhow)? + }; + let uri = checkout_session + .url + .ok_or_else(|| Error::InternalErr(format!("stripe checkout redirect issue")))?; + Ok(Redirect::to(&uri)) + } +} + +#[cfg(feature = "stripe")] +async fn stripe_portal( + authed: ApiAuthed, + Path(w_id): Path, + Extension(db): Extension, +) -> Result { + require_admin(authed.is_admin, &authed.username)?; + let customer_id = sqlx::query_scalar!( + "SELECT customer_id FROM workspace_settings WHERE workspace_id = $1", + w_id + ) + .fetch_one(&db) + .await? + .ok_or_else(|| Error::InternalErr(format!("no customer id for workspace {}", w_id)))?; + let client = stripe::Client::new( + STRIPE_KEY + .clone() + .ok_or(Error::InternalErr(format!("stripe key not set")))?, + ); + let success_rd = format!( + "{}/workspace_settings?tab=premium", + BASE_URL.read().await.clone() + ); + let portal_session = { + let customer_id = CustomerId::from_str(&customer_id).unwrap(); + let mut params = stripe::CreateBillingPortalSession::new(customer_id); + params.return_url = Some(&success_rd); + stripe::BillingPortalSession::create(&client, params) + .await + .map_err(to_anyhow)? + }; + Ok(Redirect::to(&portal_session.url)) +} + +#[cfg(feature = "stripe")] +#[derive(Serialize)] +pub struct PremiumWorkspaceInfo { + pub premium: bool, + pub usage: Option, + pub seats: Option, +} + +// async fn stripe_usage( +// authed: ApiAuthed, +// Path(w_id): Path, +// Extension(db): Extension, +// Extension(base_url): Extension>, +// ) -> Result { +// require_admin(authed.is_admin, &authed.username)?; +// let customer_id = sqlx::query_scalar!( +// "SELECT customer_id FROM workspace_settings WHERE workspace_id = $1", +// w_id +// ) +// .fetch_one(&db) +// .await? +// .ok_or_else(|| Error::InternalErr(format!("no customer id for workspace {}", w_id)))?; +// let client = stripe::Client::new(std::env::var("STRIPE_KEY").expect("STRIPE_KEY")); +// let success_rd = format!("{}/workspace_settings?tab=premium", base_url.0); +// let portal_session = { +// let customer_id = CustomerId::from_str(&customer_id).unwrap(); +// let subscriptions = stripe::Subscription::list( +// &client, +// stripe::ListSubscriptions { customer: Some(customer_id), ..Default::default() }, +// ) +// .await +// .map_err(to_anyhow)? +// .data[0]; +// let getUsage = +// stripe::SubscriptionItem::list( +// &client, +// stripe::ListSubscriptionItems { +// subscription: subscription.id, +// ..Default::default() +// }, +// ) +// .await +// .map_err(to_anyhow) +// }; +// let mut params = stripe::ListSubscriptionItems::new(customer_id); +// params.return_url = Some(&success_rd); +// stripe::BillingPortalSession::create(&client, params) +// .await +// .map_err(to_anyhow)? +// }; +// } diff --git a/backend/windmill-audit/src/audit_ee.rs b/backend/windmill-audit/src/audit_ee.rs index 5bb52e7b3f..872c882971 100644 --- a/backend/windmill-audit/src/audit_ee.rs +++ b/backend/windmill-audit/src/audit_ee.rs @@ -5,6 +5,8 @@ * Please see the included NOTICE for copyright information and * LICENSE-AGPL for a copy of the license. */ +use sql_builder::prelude::*; + use std::collections::HashMap; use windmill_common::{ @@ -12,37 +14,124 @@ use windmill_common::{ utils::Pagination, }; +#[cfg(feature = "enterprise")] +use windmill_common::ee::{get_license_plan, LicensePlan}; + use crate::{ActionKind, AuditLog, ListAuditLogQuery}; +use sql_builder::SqlBuilder; use sqlx::{Postgres, Transaction}; #[tracing::instrument(level = "trace", skip_all)] pub async fn audit_log<'c, E: sqlx::Executor<'c, Database = Postgres>>( - _db: E, - _username: &str, + db: E, + username: &str, mut _operation: &str, - _action_kind: ActionKind, - _w_id: &str, + action_kind: ActionKind, + w_id: &str, mut _resource: Option<&str>, _parameters: Option>, ) -> Result<()> { - // Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature + #[cfg(feature = "enterprise")] + let p_json = match get_license_plan().await { + LicensePlan::Enterprise => serde_json::to_value(&_parameters).unwrap(), + LicensePlan::Pro => serde_json::json!({"redacted": "-"}), + }; + + #[cfg(not(feature = "enterprise"))] + let p_json: serde_json::Value = serde_json::json!({"redacted": "-"}); + + #[cfg(feature = "enterprise")] + if matches!(get_license_plan().await, LicensePlan::Pro) { + _resource = Some("EE only"); + _operation = "redacted"; + } + + #[cfg(not(feature = "enterprise"))] + { + _resource = Some("EE only"); + _operation = "redacted"; + } + + tracing::info!( + operation = _operation, + action_kind = ?action_kind, + resource = _resource, + parameters = %p_json, + workspace_id = w_id, + username = username, + ); + + sqlx::query( + "INSERT INTO audit + (workspace_id, username, operation, action_kind, resource, parameters) + VALUES ($1, $2, $3, $4, $5, $6)", + ) + .bind(w_id) + .bind(username) + .bind(_operation) + .bind(action_kind) + .bind(_resource) + .bind(p_json) + .execute(db) + .await?; + Ok(()) } pub async fn list_audit( - _tx: Transaction<'_, Postgres>, - _w_id: String, - _pagination: Pagination, - _lq: ListAuditLogQuery, + mut tx: Transaction<'_, sqlx::Postgres>, + w_id: String, + pagination: Pagination, + lq: ListAuditLogQuery, ) -> Result> { - // Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature - return Ok(vec![]); + let (per_page, offset) = windmill_common::utils::paginate(pagination); + + let mut sqlb = SqlBuilder::select_from("audit") + .field("*") + .order_by("id", true) + .and_where_eq("workspace_id", "?".bind(&w_id)) + .offset(offset) + .limit(per_page) + .clone(); + + if let Some(u) = &lq.username { + sqlb.and_where_eq("username", "?".bind(u)); + } + if let Some(o) = &lq.operation { + sqlb.and_where_eq("operation", "?".bind(o)); + } + if let Some(ak) = &lq.action_kind { + sqlb.and_where_eq("action_kind", "?".bind(ak)); + } + if let Some(r) = &lq.resource { + sqlb.and_where_eq("resource", "?".bind(r)); + } + if let Some(b) = &lq.before { + sqlb.and_where_le("timestamp", format!("to_timestamp({})", b.timestamp())); + } + if let Some(a) = &lq.after { + sqlb.and_where_gt("timestamp", format!("to_timestamp({})", a.timestamp())); + } + + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let rows = sqlx::query_as::<_, AuditLog>(&sql) + .fetch_all(&mut *tx) + .await?; + Ok(rows) } -pub async fn get_audit(tx: Transaction<'_, Postgres>, _id: i32, _w_id: &str) -> Result { - // Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature +pub async fn get_audit( + mut tx: Transaction<'_, sqlx::Postgres>, + id: i32, + w_id: &str, +) -> Result { + let audit_o = + sqlx::query_as::<_, AuditLog>("SELECT * FROM audit WHERE id = $1 AND workspace_id = $2") + .bind(id) + .bind(w_id) + .fetch_optional(&mut *tx) + .await?; tx.commit().await?; - Err(Error::NotFound( - "Audit log not not available in Windmill Community edition".to_string(), - )) + let audit = windmill_common::utils::not_found_if_none(audit_o, "AuditLog", &id.to_string())?; + Ok(audit) } diff --git a/backend/windmill-common/src/ee.rs b/backend/windmill-common/src/ee.rs index c942447bbc..d70982d8bd 100644 --- a/backend/windmill-common/src/ee.rs +++ b/backend/windmill-common/src/ee.rs @@ -1,4 +1,3 @@ -use crate::ee::LicensePlan::Community; use std::sync::Arc; use tokio::sync::RwLock; @@ -9,12 +8,16 @@ lazy_static::lazy_static! { } pub enum LicensePlan { - Community, Pro, Enterprise, } pub async fn get_license_plan() -> LicensePlan { - // Implementation is not open source - return Community; + let id = LICENSE_KEY_ID.read().await.clone(); + + if id.ends_with("_pro") { + LicensePlan::Pro + } else { + LicensePlan::Enterprise + } } diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index d5bd88d0ed..2570c92bd6 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -13,7 +13,7 @@ pub const BUNFIG_INSTALL_SCOPES_SETTING: &str = "bunfig_install_scopes"; pub const EXTRA_PIP_INDEX_URL_SETTING: &str = "pip_extra_index_url"; pub const PIP_INDEX_URL_SETTING: &str = "pip_index_url"; pub const SCIM_TOKEN_SETTING: &str = "scim_token"; -pub const SAML_METADATA_SETTING: &str = "saml_token"; +pub const SAML_METADATA_SETTING: &str = "saml_metadata"; pub const UNIQUE_ID_SETTING: &str = "uid"; pub const DISABLE_STATS_SETTING: &str = "disable_stats"; diff --git a/backend/windmill-git-sync/src/git_sync_ee.rs b/backend/windmill-git-sync/src/git_sync_ee.rs index 4be2858c27..8158c026f4 100644 --- a/backend/windmill-git-sync/src/git_sync_ee.rs +++ b/backend/windmill-git-sync/src/git_sync_ee.rs @@ -1,17 +1,408 @@ -use windmill_common::error::Result; +use std::collections::HashMap; + +use uuid::Uuid; +use windmill_common::users::SUPERADMIN_SYNC_EMAIL; +use windmill_common::workspaces::{ObjectType, WorkspaceGitSyncSettings}; + +use regex::Regex; +use serde_json::json; +use windmill_common::error::{Error, Result}; +use windmill_common::jobs::JobPayload; +use windmill_queue::PushIsolationLevel; use crate::{DeployedObject, DB}; +#[cfg(not(feature = "enterprise"))] pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send + Clone + 'c>( - _email: &str, - _created_by: &str, - _db: &DB, - _w_id: &str, - _obj: DeployedObject, - _deployment_message: Option, - _rsmq: Option, - _skip_db_insert: bool, + email: &str, + created_by: &str, + db: &DB, + w_id: &str, + obj: DeployedObject, + deployment_message: Option, + rsmq: Option, + skip_db_insert: bool, ) -> Result<()> { - // Git sync is an enterprise feature and not part of the open-source version + Ok(()) +} + +#[cfg(feature = "enterprise")] +pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send + Clone + 'c>( + email: &str, + created_by: &str, + db: &DB, + w_id: &str, + obj: DeployedObject, + deployment_message: Option, + rsmq: Option, + skip_db_insert: bool, +) -> Result<()> { + let workspace_git_repo_setting = sqlx::query_scalar!( + "SELECT git_sync FROM workspace_settings WHERE workspace_id = $1", + w_id + ) + .fetch_one(db) + .await + .map_err(|err| { + Error::BadRequest(format!( + "No workspace settings found for workspace ID: {} - Error: {}", + w_id, + err.to_string() + )) + })?; + + let workspace_git_sync_settings = workspace_git_repo_setting + .map(|conf| serde_json::from_value::(conf).ok()) + .flatten() + .unwrap_or_default(); + + let obj_path = if one_regexp_match( + obj.get_path(), + workspace_git_sync_settings.include_path.clone(), + ) { + Some(obj.get_path()) + } else { + None + }; + let obj_parent_path = if obj + .get_parent_path() + .map(|p| one_regexp_match(p.as_str(), workspace_git_sync_settings.include_path)) + .unwrap_or(false) + { + obj.get_parent_path() + } else { + None + }; + + let skip_git_sync = if obj_path.is_none() && obj_parent_path.is_none() { + tracing::debug!( + "Ignoring {} from git sync as it's in a private user folder", + obj.get_path() + ); + true + } else { + false + }; + tracing::debug!("Skipping git sync for {:?} -> {}", obj_path, skip_git_sync); + + let mut git_sync_job_uuids: Vec = vec![]; + if !skip_git_sync { + for workspace_git_repo in workspace_git_sync_settings.repositories { + // check that this type of object should indeed be synced + let skip_object_type = match obj { + DeployedObject::Script { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Script) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::Script) + } + DeployedObject::Flow { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Flow) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::Flow) + } + DeployedObject::App { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::App) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::App) + } + DeployedObject::Folder { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Folder) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::Folder) + } + DeployedObject::Resource { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Resource) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::Resource) + } + DeployedObject::Variable { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Variable) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::Variable) + } + DeployedObject::ResourceType { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::ResourceType) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::ResourceType) + } + DeployedObject::Schedule { .. } => { + !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Schedule) + || workspace_git_repo + .exclude_types_override + .unwrap_or_default() + .iter() + .any(|element| *element == ObjectType::Schedule) + } + }; + + if skip_object_type { + tracing::debug!("Skipping sync due to objec type filter"); + continue; + } + + // if yes, sync it + let mut args: HashMap = HashMap::new(); + args.insert( + "repo_url_resource_path".to_string(), + json!(workspace_git_repo + .git_repo_resource_path + .strip_prefix("$res:")), + ); + + if let Some(path) = obj_path { + args.insert("path".to_string(), json!(path)); + } + if let Some(parent_path) = obj_parent_path.clone() { + args.insert("parent_path".to_string(), json!(parent_path)); + } + args.insert("workspace_id".to_string(), json!(w_id)); + + let message = match obj.clone() { + DeployedObject::Script { path, .. } => { + args.insert("path_type".to_string(), json!("script")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Script '{}' deployed", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::Flow { path, .. } => { + args.insert("path_type".to_string(), json!("flow")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Flow '{}' deployed", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::App { path, .. } => { + args.insert("path_type".to_string(), json!("app")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("App '{}' deployed", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::Folder { path, .. } => { + args.insert("path_type".to_string(), json!("folder")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Folder '{}' updated", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::Resource { path, .. } => { + args.insert("path_type".to_string(), json!("resource")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Resource '{}' updated", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::Variable { path, .. } => { + args.insert("path_type".to_string(), json!("variable")); + let skip_secret = !workspace_git_sync_settings + .include_type + .iter() + .any(|element| *element == ObjectType::Secret); + args.insert("skip_secret".to_string(), json!(skip_secret)); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Variable '{}' updated", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::ResourceType { path, .. } => { + args.insert("path_type".to_string(), json!("resourcetype")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Resource Type '{}' updated", path) + } else { + deployment_message.clone().unwrap() + } + } + DeployedObject::Schedule { path, .. } => { + args.insert("path_type".to_string(), json!("schedule")); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Schedule '{}' updated", path) + } else { + deployment_message.clone().unwrap() + } + } + }; + + args.insert("commit_msg".to_string(), json!(message)); + args.insert( + "use_individual_branch".to_string(), + json!(workspace_git_repo.use_individual_branch.unwrap_or(false)), + ); + + let tx: PushIsolationLevel<'_, R> = + PushIsolationLevel::IsolatedRoot(db.clone(), rsmq.clone()); + + let (job_uuid, new_tx) = windmill_queue::push( + db, + tx, + &w_id, + JobPayload::DeploymentCallback { + path: workspace_git_repo.script_path.clone(), + }, + args, + created_by, + &email, + SUPERADMIN_SYNC_EMAIL.to_string(), + None, + None, + None, + None, + None, + false, + false, + None, + true, + None, + None, + None, + None, + ) + .await?; + new_tx.commit().await?; + + git_sync_job_uuids.push(job_uuid); + } + } + + // We're not persisting the default commit msg as it's pretty useless. We will persist the ones manually set by users + if !skip_db_insert && (deployment_message.is_some() || git_sync_job_uuids.len() > 0) { + // if the git sync job hasn't been triggered, and there is not custom deployment message, there's not point adding an entry to the table + match obj.clone() { + DeployedObject::Script { path, hash, .. } => { + sqlx::query!( + "INSERT INTO deployment_metadata (workspace_id, path, script_hash, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)", + w_id, path, hash.0, &git_sync_job_uuids, deployment_message, + ).execute(db) + .await?; + } + DeployedObject::Flow { path, .. } => { + sqlx::query!( + "INSERT INTO deployment_metadata (workspace_id, path, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4) ON CONFLICT (workspace_id, path) WHERE script_hash IS NULL AND app_version IS NULL DO UPDATE SET callback_job_ids = $3, deployment_msg = $4", + w_id, path, &git_sync_job_uuids, deployment_message, + ).execute(db) + .await?; + } + DeployedObject::App { path, version, .. } => { + sqlx::query!( + "INSERT INTO deployment_metadata (workspace_id, path, app_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)", + w_id, path, version, &git_sync_job_uuids, deployment_message, + ).execute(db) + .await?; + } + DeployedObject::Folder { .. } => (), + DeployedObject::Resource { .. } => (), + DeployedObject::Variable { .. } => (), + DeployedObject::ResourceType { .. } => (), + DeployedObject::Schedule { .. } => (), + } + } + return Ok(()); } + +fn one_regexp_match(path: &str, regexps: Vec) -> bool { + let has_match = regexps + .iter() + .map(|regexp| transform_regexp(regexp.to_owned())) + .filter(|regexp| regexp.is_some()) + .map(|regexp| regexp.unwrap()) + .any(|regexp| regexp.is_match(path)); + tracing::debug!( + "{} matches the set of regexps {:?} -> {}", + path, + regexps, + has_match + ); + return has_match; +} + +fn transform_regexp(user_regexp: String) -> Option { + // this is annoying b/c we want to replace ** with [a-zA-Z0-9_.*/]* AND THEN * with [a-zA-Z0-9_.*]* - but b/c the + // first replacement string contains itself '*', we can't do it naively. So, the hack is to use a different + // character in the replacement string, instead of '*' we use here '%', and finally we replace all '%' with '*' + let mut regexp = user_regexp + .replace("**", "[a-zA-Z0-9_.%/]%") + .replace("*", "[a-zA-Z0-9_.%]%"); + regexp = regexp.replace("%", "*"); + // Then we add ^ at the beginning and '$' at the end to match the entire string, not just a substring + regexp = if regexp.starts_with("^") { + regexp + } else { + format!("^{}", regexp) + }; + regexp = if regexp.ends_with("$") { + regexp + } else { + format!("{}$", regexp) + }; + let compiled_regexp = Regex::new(regexp.as_str()).ok(); + tracing::debug!("Compiled regexp: {:?}", compiled_regexp); + return compiled_regexp; +} diff --git a/frontend/src/lib/components/InstanceSettings.svelte b/frontend/src/lib/components/InstanceSettings.svelte index b988cc16cf..fc0d9cb253 100644 --- a/frontend/src/lib/components/InstanceSettings.svelte +++ b/frontend/src/lib/components/InstanceSettings.svelte @@ -436,6 +436,19 @@ placeholder={setting.placeholder} bind:value={values[setting.key]} /> + {#if setting.key == 'saml_metadata'} +
+ +
+ {/if} {:else if setting.fieldType == 'license_key'}