feat(postgres, trigger): support ssl (#5149)

* fix

* feat: support-ssl

* Update backend/windmill-api/src/postgres_triggers/trigger.rs

Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>

* fix: match error

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>

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",

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"]}

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 }
+rust_decimal = { workspace = true, optional = true }
+rust-postgres-native-tls = { workspace = true, optional = true}

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,
 }

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<SimpleQueryMessage> {
 #[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<Self, Error> {
+        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::<bytes::Bytes>(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() => {

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
 	}
 </script>
 
@@ -316,7 +322,10 @@
 			{/if}
 		</svelte:fragment>
 		{#if drawerLoading}
-			<Loader2 class="animate-spin" />
+			<div class="flex flex-col items-center justify-center h-full w-full">
+				<Loader2 size="50" class="animate-spin" />
+				<p>Loading...</p>
+			</div>
 		{:else}
 			<div class="flex flex-col gap-5">
 				<Alert title="Info" type="info">
@@ -350,7 +359,11 @@
 					<div class="flex flex-col mb-2 gap-3">
 						<ResourcePicker bind:value={postgres_resource_path} resourceType={'postgresql'} />
 						{#if postgres_resource_path}
-							<Button on:click={checkDatabaseConfiguration} color="gray" size="sm"
+							<Button
+								loading={loadingConfiguration}
+								on:click={checkDatabaseConfiguration}
+								color="gray"
+								size="sm"
 								>Check Database Configuration
 								<Tooltip>
 									<p class="text-sm">

frontend/src/lib/components/triggers/postgres/PostgresTriggersPanel.svelte

@@ -3,7 +3,7 @@
 	import { PostgresTriggerService, type PostgresTrigger } from '$lib/gen'
 	import { UnplugIcon } from 'lucide-svelte'
 
-	import { canWrite } from '$lib/utils'
+	import { canWrite, sendUserToast } from '$lib/utils'
 	import { getContext } from 'svelte'
 	import { isCloudHosted } from '$lib/cloud'
 	import { Alert, Button, Skeleton } from '$lib/components/common'
@@ -33,8 +33,8 @@
 				return { canWrite: canWrite(x.path, x.extra_perms!, $userStore), ...x }
 			})
 			$triggersCount = { ...($triggersCount ?? {}), postgres_count: databaseTriggers?.length }
-		} catch (e) {
-			console.error('impossible to load Postgres triggers', e)
+		} catch (err) {
+			sendUserToast(`Could not load postgres triggers ${err.body}`, true)
 		}
 	}
 </script>