feat: scim token and saml metadata setting in UI directly

This commit is contained in:
Ruben Fiszel
2024-02-17 16:46:02 +01:00
parent 1de8479259
commit 71e915a86c
18 changed files with 4529 additions and 125 deletions
+1 -1
View File
@@ -1 +1 @@
8ed0d6aa6218ec4f9011a60db701295e5c3a1e4a
23eaf9ef7fd5b2d115c0db2aa12a145e35b16f33
+44 -10
View File
@@ -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::<u64>()
.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(())
}
+12 -2
View File
@@ -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);
+19 -2
View File
@@ -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,
+21
View File
@@ -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)
+38 -3
View File
@@ -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<String> {
// 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<String> {
let mut splitted_lk = license_key.split(".");
if splitted_lk.clone().count() != 3 {
return Err(anyhow!(
"license_key can be splitted with 2 . (<client id>.<expiry>.<signature>)"
));
}
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::<Sha256>::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::<u64>()?;
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())
}
File diff suppressed because it is too large Load Diff
+3
View File
@@ -89,6 +89,9 @@ lazy_static::lazy_static! {
pub static ref REQUEST_SIZE_LIMIT: Arc<RwLock<usize>> = Arc::new(RwLock::new(DEFAULT_BODY_LIMIT));
pub static ref SCIM_TOKEN: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
pub static ref SAML_METADATA: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
pub static ref COOKIE_DOMAIN: Option<String> = std::env::var("COOKIE_DOMAIN").ok();
File diff suppressed because it is too large Load Diff
+177 -12
View File
@@ -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<ServiceProvider>);
#[cfg(not(feature = "enterprise_saml"))]
pub struct ServiceProviderExt();
#[cfg(not(feature = "enterprise_saml"))]
pub async fn build_sp_extension() -> anyhow::Result<ServiceProviderExt> {
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<ServiceProviderExt> {
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 <ruben@windmill.dev>".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<ServiceProviderExt>,
) -> anyhow::Result<Option<String>> {
return Ok(None);
}
#[cfg(feature = "enterprise_saml")]
pub async fn generate_redirect_url(
service_provider: Arc<ServiceProviderExt>,
) -> anyhow::Result<Option<String>> {
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<EntityDescriptor> {
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<String>) -> Result<String> {
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<String>,
}
#[cfg(feature = "enterprise_saml")]
pub async fn acs(
Extension(db): Extension<DB>,
cookies: Cookies,
Extension(se): Extension<Arc<ServiceProviderExt>>,
Form(s): Form<SamlForm>,
) -> Result<Redirect> {
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()))
}
}
+712 -6
View File
@@ -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<T>(pub T);
pub async fn has_scim_token<B>(request: Request<B>, next: Next<B>) -> 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<T> IntoResponse for JsonScim<T>
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<S>(schema: &str, resources: Vec<S>) -> JsonScim<serde_json::Value>
where
S: Serialize,
{
return JsonScim(json!({
"schemas": [schema],
"totalResults": resources.len(),
"Resources": resources,
"startIndex": 1,
"itemsPerPage": 100,
}));
}
#[derive(Deserialize)]
pub struct ScimQuery {
startIndex: Option<u32>,
count: Option<u32>,
filter: Option<String>,
}
pub async fn get_users(
Extension(db): Extension<DB>,
Query(query): Query<ScimQuery>,
) -> Result<JsonScim<serde_json::Value>> {
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<DB>,
Json(body): Json<CreateUser>,
) -> Result<JsonScim<serde_json::Value>> {
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<DB>,
Path(username): Path<String>,
) -> Result<JsonScim<serde_json::Value>> {
// 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<DB>, Path(username): Path<String>) -> 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<String>,
pub Operations: Option<Vec<Operation>>,
}
#[cfg(feature = "enterprise")]
pub async fn update_user(
Extension(db): Extension<DB>,
Path(username): Path<String>,
Json(body): Json<UpdateUser>,
) -> Result<JsonScim<serde_json::Value>> {
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::<String>(
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<DB>,
Query(query): Query<ScimQuery>,
) -> Result<JsonScim<serde_json::Value>> {
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<Vec<String>>,
external_id: Option<String>,
scim_display_name: Option<String>,
}
#[cfg(feature = "enterprise")]
fn group_response(group: Group) -> JsonScim<serde_json::Value> {
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::<Vec<serde_json::Value>>(),
"meta": {
"resourceType": "Group"
}
});
tracing::info!("SCIM group: {:?}", json);
return JsonScim(json);
}
#[cfg(feature = "enterprise")]
pub async fn get_group(
Extension(db): Extension<DB>,
Path(id): Path<String>,
) -> Result<JsonScim<serde_json::Value>> {
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<Member>,
}
#[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<DB>,
Json(body): Json<CreateGroup>,
) -> Result<JsonScim<serde_json::Value>> {
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<serde_json::Value>,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct UpdateMembersOperationValue {
pub value: String,
}
#[cfg(feature = "enterprise")]
#[derive(Deserialize, Debug)]
pub struct UpdateGroup {
pub schemas: Vec<String>,
pub displayName: Option<String>,
pub members: Option<Vec<Member>>,
pub Operations: Option<Vec<Operation>>,
}
#[cfg(feature = "enterprise")]
pub async fn update_group(
Extension(db): Extension<DB>,
Path(id): Path<String>,
Json(body): Json<UpdateGroup>,
) -> Result<JsonScim<serde_json::Value>> {
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::<String>(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<UpdateMembersOperationValue>,
>(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::<String>(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<UpdateMembersOperationValue>,
>(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<DB>, Path(id): Path<String>) -> 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()
}
+301 -2
View File
@@ -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<String> = 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<DB>,
Path(w_id): Path<String>,
) -> JsonResult<PremiumWorkspaceInfo> {
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::<Vec<_>>()
.get(0)
.copied();
}
Ok(Json(result))
}
#[cfg(feature = "stripe")]
#[derive(Deserialize)]
struct PlanQuery {
plan: String,
seats: Option<i32>,
}
#[cfg(feature = "stripe")]
async fn stripe_checkout(
authed: ApiAuthed,
Path(w_id): Path<String>,
Query(plan): Query<PlanQuery>,
Extension(db): Extension<DB>,
) -> Result<Redirect> {
// #[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<String>,
Extension(db): Extension<DB>,
) -> Result<Redirect> {
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<i32>,
pub seats: Option<i32>,
}
// async fn stripe_usage(
// authed: ApiAuthed,
// Path(w_id): Path<String>,
// Extension(db): Extension<DB>,
// Extension(base_url): Extension<Arc<BaseUrl>>,
// ) -> Result<Redirect> {
// 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)?
// };
// }
+105 -16
View File
@@ -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<HashMap<&str, &str>>,
) -> 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<Vec<AuditLog>> {
// 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<AuditLog> {
// 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<AuditLog> {
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)
}
+7 -4
View File
@@ -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
}
}
@@ -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";
+401 -10
View File
@@ -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<String>,
_rsmq: Option<R>,
_skip_db_insert: bool,
email: &str,
created_by: &str,
db: &DB,
w_id: &str,
obj: DeployedObject,
deployment_message: Option<String>,
rsmq: Option<R>,
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<String>,
rsmq: Option<R>,
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::<WorkspaceGitSyncSettings>(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<Uuid> = 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<String, serde_json::Value> = 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<String>) -> 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<Regex> {
// 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;
}
@@ -436,6 +436,19 @@
placeholder={setting.placeholder}
bind:value={values[setting.key]}
/>
{#if setting.key == 'saml_metadata'}
<div class="flex mt-2">
<Button
on:click={async (e) => {
const res = await SettingService.testMetadata({
requestBody: values[setting.key]
})
sendUserToast(`Metadata valid, see console for full content`)
console.log(`Metadata content:`, res)
}}>Test content/url</Button
>
</div>
{/if}
{:else if setting.fieldType == 'license_key'}
<div class="flex justify-between gap-2">
<textarea
@@ -185,6 +185,26 @@ export const settings: Record<string, Setting[]> = {
storage: 'config'
}
],
'SCIM/SAML': [
{
label: 'SCIM Token',
description: 'Token used to authenticate requests from the IdP',
key: 'scim_token',
fieldType: 'text',
placeholder: 'mytoken',
storage: 'setting',
ee_only: ''
},
{
label: 'SAML metadata',
description: 'XML metadata url OR content for the SAML IdP',
key: 'saml_metadata',
fieldType: 'textarea',
placeholder: 'https://dev-2578259.okta.com/app/exkaell8gidiiUWrg5d7/sso/saml/metadata ',
storage: 'setting',
ee_only: ''
}
],
Debug: [
{
label: 'Keep Job Directories',