Add TLS support to PostgreSQL storage backend

Use native-tls and postgres-native-tls to support encrypted
connections, following the same pattern as the VSS server.

A new PostgresTlsConfig struct controls TLS behavior:
- None: plaintext (existing behavior)
- Some with certificate_pem: None: TLS using system root CAs
- Some with certificate_pem: Some(pem): TLS with a custom CA

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: benthecarman

Cargo.toml

@@ -25,7 +25,7 @@ panic = 'abort'     # Abort on panic
 
 [features]
 default = []
-postgres = ["dep:tokio-postgres"]
+postgres = ["dep:tokio-postgres", "dep:native-tls", "dep:postgres-native-tls"]
 
 [dependencies]
 #lightning = { version = "0.2.0", features = ["std"] }
@@ -78,6 +78,8 @@ log = { version = "0.4.22", default-features = false, features = ["std"]}
 
 async-trait = { version = "0.1", default-features = false }
 tokio-postgres = { version = "0.7", default-features = false, features = ["runtime"], optional = true }
+native-tls = { version = "0.2", default-features = false, optional = true }
+postgres-native-tls = { version = "0.5", default-features = false, features = ["runtime"], optional = true }
 vss-client = { package = "vss-client-ng", version = "0.5" }
 prost = { version = "0.11.6", default-features = false}
 #bitcoin-payment-instructions = { version = "0.6" }

src/builder.rs

@@ -638,15 +638,22 @@ impl NodeBuilder {
 	/// if it doesn't already exist.
 	///
 	/// The given `kv_table_name` will be used or default to
-	/// [`DEFAULT_KV_TABLE_NAME`](crate::io::postgres_store::DEFAULT_KV_TABLE_NAME).
+	/// [`DEFAULT_KV_TABLE_NAME`](io::postgres_store::DEFAULT_KV_TABLE_NAME).
+	///
+	/// If `tls_config` is `Some`, TLS will be used for database connections. A custom CA
+	/// certificate can be provided via
+	/// [`PostgresTlsConfig::certificate_pem`](io::postgres_store::PostgresTlsConfig::certificate_pem),
+	/// otherwise the system's default root certificates are used. If `tls_config` is `None`,
+	/// connections will be unencrypted.
 	///
 	/// [PostgreSQL]: https://www.postgresql.org
 	#[cfg(feature = "postgres")]
 	pub fn build_with_postgres_store(
 		&self, node_entropy: NodeEntropy, connection_string: String, kv_table_name: Option<String>,
+		tls_config: Option<io::postgres_store::PostgresTlsConfig>,
 	) -> Result<Node, BuildError> {
 		let kv_store =
-			crate::io::postgres_store::PostgresStore::new(connection_string, kv_table_name)
+			io::postgres_store::PostgresStore::new(connection_string, kv_table_name, tls_config)
 				.map_err(|_| BuildError::KVStoreSetupFailed)?;
 		self.build_with_store(node_entropy, kv_store)
 	}
@@ -1116,12 +1123,12 @@ impl ArcedNodeBuilder {
 	#[cfg(feature = "postgres")]
 	pub fn build_with_postgres_store(
 		&self, node_entropy: Arc<NodeEntropy>, connection_string: String,
-		kv_table_name: Option<String>,
+		kv_table_name: Option<String>, tls_config: Option<io::postgres_store::PostgresTlsConfig>,
 	) -> Result<Arc<Node>, BuildError> {
 		self.inner
 			.read()
 			.unwrap()
-			.build_with_postgres_store(*node_entropy, connection_string, kv_table_name)
+			.build_with_postgres_store(*node_entropy, connection_string, kv_table_name, tls_config)
 			.map(Arc::new)
 	}
 

src/io/postgres_store/mod.rs

@@ -16,7 +16,9 @@ use lightning::util::persist::{
 	KVStore, KVStoreSync, PageToken, PaginatedKVStore, PaginatedKVStoreSync, PaginatedListResponse,
 };
 use lightning_types::string::PrintableString;
-use tokio_postgres::{connect, Client, Config, Error as PgError, NoTls};
+use native_tls::TlsConnector;
+use postgres_native_tls::MakeTlsConnector;
+use tokio_postgres::{Client, Config, Error as PgError, NoTls};
 
 use crate::io::utils::check_namespace_key_validity;
 
@@ -65,7 +67,17 @@ impl PostgresStore {
 	/// if it doesn't already exist.
 	///
 	/// The given `kv_table_name` will be used or default to [`DEFAULT_KV_TABLE_NAME`].
-	pub fn new(connection_string: String, kv_table_name: Option<String>) -> io::Result<Self> {
+	///
+	/// If `tls_config` is `Some`, TLS will be used for database connections. A custom CA
+	/// certificate can be provided via [`PostgresTlsConfig::certificate_pem`], otherwise the
+	/// system's default root certificates are used. If `tls_config` is `None`, connections
+	/// will be unencrypted.
+	pub fn new(
+		connection_string: String, kv_table_name: Option<String>,
+		tls_config: Option<PostgresTlsConfig>,
+	) -> io::Result<Self> {
+		let tls = Self::build_tls_connector(tls_config)?;
+
 		let internal_runtime = tokio::runtime::Builder::new_multi_thread()
 			.enable_all()
 			.thread_name_fn(|| {
@@ -79,15 +91,41 @@ impl PostgresStore {
 			.unwrap();
 
 		let inner = tokio::task::block_in_place(|| {
-			internal_runtime
-				.block_on(async { PostgresStoreInner::new(connection_string, kv_table_name).await })
+			internal_runtime.block_on(async {
+				PostgresStoreInner::new(connection_string, kv_table_name, tls).await
+			})
 		})?;
 
 		let inner = Arc::new(inner);
 		let next_write_version = AtomicU64::new(1);
 		Ok(Self { inner, next_write_version, internal_runtime: Some(internal_runtime) })
 	}
 
+	fn build_tls_connector(tls_config: Option<PostgresTlsConfig>) -> io::Result<PgTlsConnector> {
+		match tls_config {
+			Some(config) => {
+				let mut builder = TlsConnector::builder();
+				if let Some(pem) = config.certificate_pem {
+					let crt = native_tls::Certificate::from_pem(pem.as_bytes()).map_err(|e| {
+						io::Error::new(
+							io::ErrorKind::InvalidInput,
+							format!("Failed to parse PEM certificate: {e}"),
+						)
+					})?;
+					builder.add_root_certificate(crt);
+				}
+				let connector = builder.build().map_err(|e| {
+					io::Error::new(
+						io::ErrorKind::Other,
+						format!("Failed to build TLS connector: {e}"),
+					)
+				})?;
+				Ok(PgTlsConnector::NativeTls(MakeTlsConnector::new(connector)))
+			},
+			None => Ok(PgTlsConnector::Plain),
+		}
+	}
+
 	fn build_locking_key(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> String {
@@ -309,14 +347,17 @@ impl PaginatedKVStore for PostgresStore {
 
 struct PostgresStoreInner {
 	client: tokio::sync::Mutex<Client>,
-	connection_string: String,
+	config: Config,
 	kv_table_name: String,
+	tls: PgTlsConnector,
 	write_version_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<u64>>>>,
 	next_sort_order: AtomicI64,
 }
 
 impl PostgresStoreInner {
-	async fn new(connection_string: String, kv_table_name: Option<String>) -> io::Result<Self> {
+	async fn new(
+		connection_string: String, kv_table_name: Option<String>, tls: PgTlsConnector,
+	) -> io::Result<Self> {
 		let kv_table_name = kv_table_name.unwrap_or(DEFAULT_KV_TABLE_NAME.to_string());
 
 		// If a dbname is specified in the connection string, ensure the database exists
@@ -327,10 +368,10 @@ impl PostgresStoreInner {
 		})?;
 
 		if let Some(db_name) = config.get_dbname() {
-			Self::create_database_if_not_exists(&connection_string, db_name).await?;
+			Self::create_database_if_not_exists(&config, db_name, &tls).await?;
 		}
 
-		let client = Self::make_connection(&connection_string).await?;
+		let client = Self::make_config_connection(&config, &tls).await?;
 
 		// Create the KV data table if it doesn't exist.
 		let sql = format!(
@@ -399,29 +440,17 @@ impl PostgresStoreInner {
 
 		let client = tokio::sync::Mutex::new(client);
 		let write_version_locks = Mutex::new(HashMap::new());
-		Ok(Self { client, connection_string, kv_table_name, write_version_locks, next_sort_order })
+		Ok(Self { client, config, kv_table_name, tls, write_version_locks, next_sort_order })
 	}
 
 	async fn create_database_if_not_exists(
-		connection_string: &str, db_name: &str,
+		config: &Config, db_name: &str, tls: &PgTlsConnector,
 	) -> io::Result<()> {
 		// Connect without a dbname (to the default database) so we can create the target.
-		let mut config: Config = connection_string.parse().map_err(|e: PgError| {
-			let msg = format!("Failed to parse PostgreSQL connection string: {e}");
-			io::Error::new(io::ErrorKind::InvalidInput, msg)
-		})?;
+		let mut config = config.clone();
 		config.dbname("postgres");
 
-		let (client, connection) = config.connect(NoTls).await.map_err(|e| {
-			let msg = format!("Failed to connect to PostgreSQL: {e}");
-			io::Error::new(io::ErrorKind::Other, msg)
-		})?;
-
-		tokio::spawn(async move {
-			if let Err(e) = connection.await {
-				log::error!("PostgreSQL connection error: {e}");
-			}
-		});
+		let client = Self::make_config_connection(&config, tls).await?;
 
 		let row = client
 			.query_opt("SELECT 1 FROM pg_database WHERE datname = $1", &[&db_name])
@@ -443,27 +472,41 @@ impl PostgresStoreInner {
 		Ok(())
 	}
 
-	async fn make_connection(connection_string: &str) -> io::Result<Client> {
-		let (client, connection) = connect(connection_string, NoTls).await.map_err(|e| {
+	async fn make_config_connection(config: &Config, tls: &PgTlsConnector) -> io::Result<Client> {
+		let err_map = |e| {
 			let msg = format!("Failed to connect to PostgreSQL: {e}");
 			io::Error::new(io::ErrorKind::Other, msg)
-		})?;
-
-		tokio::spawn(async move {
-			if let Err(e) = connection.await {
-				log::error!("PostgreSQL connection error: {e}");
-			}
-		});
+		};
 
-		Ok(client)
+		match tls {
+			PgTlsConnector::Plain => {
+				let (client, connection) = config.connect(NoTls).await.map_err(err_map)?;
+				tokio::spawn(async move {
+					if let Err(e) = connection.await {
+						log::error!("PostgreSQL connection error: {e}");
+					}
+				});
+				Ok(client)
+			},
+			PgTlsConnector::NativeTls(tls_connector) => {
+				let (client, connection) =
+					config.connect(tls_connector.clone()).await.map_err(err_map)?;
+				tokio::spawn(async move {
+					if let Err(e) = connection.await {
+						log::error!("PostgreSQL connection error: {e}");
+					}
+				});
+				Ok(client)
+			},
+		}
 	}
 
 	async fn ensure_connected(
 		&self, client: &mut tokio::sync::MutexGuard<'_, Client>,
 	) -> io::Result<()> {
 		if client.is_closed() || client.check_connection().await.is_err() {
 			log::debug!("Reconnecting to PostgreSQL database");
-			let new_client = Self::make_connection(&self.connection_string).await?;
+			let new_client = Self::make_config_connection(&self.config, &self.tls).await?;
 			**client = new_client;
 		}
 		Ok(())
@@ -750,6 +793,19 @@ impl PostgresStoreInner {
 	}
 }
 
+/// TLS configuration for PostgreSQL connections.
+#[derive(Debug, Clone)]
+pub struct PostgresTlsConfig {
+	/// PEM-encoded CA certificate. If `None`, the system's default root certificates are used.
+	pub certificate_pem: Option<String>,
+}
+
+#[derive(Clone)]
+enum PgTlsConnector {
+	Plain,
+	NativeTls(MakeTlsConnector),
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -761,7 +817,7 @@ mod tests {
 	}
 
 	fn create_test_store(table_name: &str) -> PostgresStore {
-		PostgresStore::new(test_connection_string(), Some(table_name.to_string())).unwrap()
+		PostgresStore::new(test_connection_string(), Some(table_name.to_string()), None).unwrap()
 	}
 
 	fn cleanup_store(store: &PostgresStore) {
@@ -1092,4 +1148,25 @@ mod tests {
 			cleanup_store(&store);
 		}
 	}
+
+	#[test]
+	fn test_tls_config_none_builds_plain_connector() {
+		let connector = PostgresStore::build_tls_connector(None).unwrap();
+		assert!(matches!(connector, PgTlsConnector::Plain));
+	}
+
+	#[test]
+	fn test_tls_config_system_certs_builds_native_tls_connector() {
+		let config = Some(PostgresTlsConfig { certificate_pem: None });
+		let connector = PostgresStore::build_tls_connector(config).unwrap();
+		assert!(matches!(connector, PgTlsConnector::NativeTls(_)));
+	}
+
+	#[test]
+	fn test_tls_config_invalid_pem_returns_error() {
+		let config =
+			Some(PostgresTlsConfig { certificate_pem: Some("not-a-valid-pem".to_string()) });
+		let result = PostgresStore::build_tls_connector(config);
+		assert!(result.is_err());
+	}
 }

tests/integration_tests_postgres.rs

@@ -31,6 +31,7 @@ async fn channel_full_cycle_with_postgres_store() {
 			config_a.node_entropy,
 			test_connection_string(),
 			Some("channel_cycle_a".to_string()),
+			None,
 		)
 		.unwrap();
 	node_a.start().unwrap();
@@ -44,6 +45,7 @@ async fn channel_full_cycle_with_postgres_store() {
 			config_b.node_entropy,
 			test_connection_string(),
 			Some("channel_cycle_b".to_string()),
+			None,
 		)
 		.unwrap();
 	node_b.start().unwrap();
@@ -82,6 +84,7 @@ async fn postgres_node_restart() {
 				node_entropy,
 				connection_string.clone(),
 				Some("restart_test".to_string()),
+				None,
 			)
 			.unwrap();
 
@@ -115,6 +118,7 @@ async fn postgres_node_restart() {
 			node_entropy,
 			connection_string.clone(),
 			Some("restart_test".to_string()),
+			None,
 		)
 		.unwrap();