diff --git a/backend/Cargo.lock b/backend/Cargo.lock index b0436e1745..d3a16e51c2 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -6298,6 +6298,17 @@ dependencies = [ "tokio-postgres 0.7.12", ] +[[package]] +name = "postgres-native-tls" +version = "0.5.0" +source = "git+https://github.com/imor/rust-postgres?rev=20265ef38e32a06f76b6f9b678e2077fc2211f6b#20265ef38e32a06f76b6f9b678e2077fc2211f6b" +dependencies = [ + "native-tls", + "tokio", + "tokio-native-tls", + "tokio-postgres 0.7.11", +] + [[package]] name = "postgres-protocol" version = "0.6.7" @@ -10936,6 +10947,7 @@ dependencies = [ "openssl", "pg_escape", "pin-project", + "postgres-native-tls 0.5.0 (git+https://github.com/imor/rust-postgres?rev=20265ef38e32a06f76b6f9b678e2077fc2211f6b)", "prometheus", "quick_cache", "rand 0.9.0", @@ -11395,7 +11407,7 @@ dependencies = [ "opentelemetry", "oracle", "pem 3.0.4", - "postgres-native-tls", + "postgres-native-tls 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)", "prometheus", "rand 0.9.0", "regex", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 2278dea11e..76c9b304e8 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -265,6 +265,7 @@ convert_case = "0.6.0" getrandom = "0.2" tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1", "with-bit-vec-0_6"]} rust-postgres = { package = "tokio-postgres", git = "https://github.com/imor/rust-postgres", rev = "20265ef38e32a06f76b6f9b678e2077fc2211f6b"} +rust-postgres-native-tls = { package = "postgres-native-tls", git = "https://github.com/imor/rust-postgres", features = ["runtime"], rev = "20265ef38e32a06f76b6f9b678e2077fc2211f6b" } bit-vec = "=0.6.3" mappable-rc = "^0" mysql_async = { version = "*", default-features = false, features = ["minimal", "default", "native-tls-tls", "rust_decimal"]} diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 2033132b47..cbecf865de 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -28,7 +28,7 @@ zip = ["dep:async_zip"] oauth2 = ["dep:async-oauth2"] http_trigger = ["dep:matchit"] static_frontend = ["dep:rust-embed"] -postgres_trigger = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal"] +postgres_trigger = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal", "dep:rust-postgres-native-tls"] [dependencies] windmill-queue.workspace = true @@ -116,4 +116,5 @@ rust-postgres = { workspace = true, optional = true } pg_escape = { workspace = true, optional = true } byteorder = { workspace = true, optional = true } thiserror = { workspace = true, optional = true } -rust_decimal = { workspace = true, optional = true } \ No newline at end of file +rust_decimal = { workspace = true, optional = true } +rust-postgres-native-tls = { workspace = true, optional = true} \ No newline at end of file diff --git a/backend/windmill-api/src/postgres_triggers/handler.rs b/backend/windmill-api/src/postgres_triggers/handler.rs index db8d7398cf..5bcfb69acc 100644 --- a/backend/windmill-api/src/postgres_triggers/handler.rs +++ b/backend/windmill-api/src/postgres_triggers/handler.rs @@ -46,6 +46,7 @@ pub struct Database { pub host: String, pub port: u16, pub dbname: String, + #[serde(default)] pub sslmode: String, pub root_certificate_pem: String, } diff --git a/backend/windmill-api/src/postgres_triggers/trigger.rs b/backend/windmill-api/src/postgres_triggers/trigger.rs index bd40f41c85..ea0f9a5dff 100644 --- a/backend/windmill-api/src/postgres_triggers/trigger.rs +++ b/backend/windmill-api/src/postgres_triggers/trigger.rs @@ -16,9 +16,11 @@ use crate::{ use bytes::{BufMut, Bytes, BytesMut}; use chrono::TimeZone; use futures::{pin_mut, SinkExt, StreamExt}; +use native_tls::TlsConnector; use pg_escape::{quote_identifier, quote_literal}; use rand::seq::SliceRandom; -use rust_postgres::{Client, Config, CopyBothDuplex, NoTls, SimpleQueryMessage}; +use rust_postgres::{config::SslMode, Client, Config, CopyBothDuplex, SimpleQueryMessage}; +use rust_postgres_native_tls::MakeTlsConnector; use windmill_common::{ db::UserDB, utils::report_critical_error, worker::to_raw_value, INSTANCE_NAME, }; @@ -60,28 +62,50 @@ impl RowExist for Vec { #[derive(thiserror::Error, Debug)] enum Error { #[error("Error from database: {0}")] - Postgres(rust_postgres::Error), + Postgres(#[from] rust_postgres::Error), #[error("Error : {0}")] - Common(windmill_common::error::Error), + Common(#[from] windmill_common::error::Error), + #[error("Tls Error: {0}")] + Tls(#[from] native_tls::Error), } pub struct PostgresSimpleClient(Client); impl PostgresSimpleClient { async fn new(database: &Database) -> Result { + let ssl_mode = match database.sslmode.as_ref() { + "disable" => SslMode::Disable, + "" | "prefer" | "allow" => SslMode::Prefer, + "require" => SslMode::Require, + "verify-ca" => SslMode::VerifyCa, + "verify-full" => SslMode::VerifyFull, + ssl_mode => { + return Err(Error::Common(windmill_common::error::Error::BadRequest( + format!("Invalid ssl mode for postgres: {}, please put a valid ssl_mode among the following avalible ssl mode: ['disable', 'allow', 'prefer', 'verify-ca', 'verify-full']", ssl_mode), + ))) + } + }; + let mut config = Config::new(); config .dbname(&database.dbname) .host(&database.host) .port(database.port) .user(&database.user) + .ssl_mode(ssl_mode) .replication_mode(rust_postgres::config::ReplicationMode::Logical); if !database.password.is_empty() { config.password(&database.password); } - let (client, connection) = config.connect(NoTls).await.map_err(Error::Postgres)?; + if !database.root_certificate_pem.is_empty() { + config.ssl_root_cert(database.root_certificate_pem.as_bytes()); + } + + let connector = MakeTlsConnector::new(TlsConnector::new()?); + + let (client, connection) = config.connect(connector).await?; tokio::spawn(async move { if let Err(e) = connection.await { tracing::debug!("{:#?}", e); @@ -111,8 +135,7 @@ impl PostgresSimpleClient { Ok(( self.0 .copy_both_simple::(query.as_str()) - .await - .map_err(Error::Postgres)?, + .await?, LogicalReplicationSettings::new(false), )) } @@ -250,8 +273,7 @@ async fn listen_to_transactions( &db, None, ) - .await - .map_err(Error::Common)?; + .await?; let database = get_database_resource( authed, @@ -260,8 +282,7 @@ async fn listen_to_transactions( &postgres_trigger.postgres_resource_path, &postgres_trigger.workspace_id, ) - .await - .map_err(Error::Common)?; + .await?; let client = PostgresSimpleClient::new(&database).await?; @@ -274,7 +295,6 @@ async fn listen_to_transactions( Ok::<_, Error>((logical_replication_stream, logical_replication_settings)) }; - tokio::select! { biased; _ = killpill_rx.recv() => { diff --git a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte index 6c0934817d..14310d1e9d 100644 --- a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte @@ -55,8 +55,11 @@ let selectedTable: 'all' | 'specific' = 'specific' let tab: 'advanced' | 'basic' let config: { isLogical: boolean; show: boolean } = { isLogical: false, show: false } + let loadingConfiguration = false $: table_to_track = selectedTable === 'all' ? [] : relations - + $: if (postgres_resource_path === undefined) { + config.show = false + } async function createPublication() { try { const message = await PostgresTriggerService.createPostgresPublication({ @@ -107,10 +110,12 @@ config.show = false selectedPublicationAction = selectedPublicationAction selectedSlotAction = selectedSlotAction + relations = [] + transaction_to_track = [] tab = 'basic' await loadTrigger() } catch (err) { - sendUserToast(`Could not load postgres trigger: ${err}`, true) + sendUserToast(`Could not load postgres trigger: ${err.body}`, true) } finally { drawerLoading = false } @@ -253,12 +258,12 @@ } const checkDatabaseConfiguration = async () => { + if (emptyString(postgres_resource_path)) { + sendUserToast('You must first pick a database resource', true) + return + } try { - if (emptyString(postgres_resource_path)) { - sendUserToast('You must first pick a database resource', true) - return - } - + loadingConfiguration = true config.isLogical = await PostgresTriggerService.isValidPostgresConfiguration({ workspace: $workspaceStore!, path: postgres_resource_path @@ -267,6 +272,7 @@ } catch (error) { sendUserToast(error.body, true) } + loadingConfiguration = false } @@ -316,7 +322,10 @@ {/if} {#if drawerLoading} - +
+ +

Loading...

+
{:else}
@@ -350,7 +359,11 @@
{#if postgres_resource_path} -