Add TLS capability

Cargo.toml

@@ -97,3 +97,6 @@ targets = [
 ]
 # Publish jobs to run in CI
 pr-run-mode = "plan"
+
+[patch.crates-io]
+diesel-async = { git = "https://github.com/weiznich/diesel_async.git", rev = "017ebe2fb7a2709ab5db92148dea5ce812a35e09" }

crates/kitsune-config/src/database.rs

@@ -6,4 +6,5 @@ use smol_str::SmolStr;
 pub struct Configuration {
     pub url: SmolStr,
     pub max_connections: u32,
+    pub use_tls: bool,
 }

crates/kitsune-db/Cargo.toml

@@ -15,17 +15,27 @@ diesel-async = { version = "0.4.1", features = [
 ] }
 diesel_full_text_search = { version = "2.1.1", default-features = false }
 diesel_migrations = "2.1.0"
+futures-util = { version = "0.3.30", default-features = false, features = [
+    "alloc",
+] }
 iso8601-timestamp = { version = "0.2.16", features = ["diesel-pg"] }
 kitsune-blocking = { path = "../kitsune-blocking" }
+kitsune-config = { path = "../kitsune-config" }
 kitsune-language = { path = "../kitsune-language" }
 kitsune-type = { path = "../kitsune-type" }
 miette = "5.10.0"
 num-derive = "0.4.1"
 num-traits = "0.2.17"
+rustls = "0.22.2"
+rustls-native-certs = "0.7.0"
 serde = { version = "1.0.195", features = ["derive"] }
 simd-json = "0.13.8"
 speedy-uuid = { path = "../../lib/speedy-uuid", features = ["diesel"] }
 thiserror = "1.0.56"
+tokio = { version = "1.35.1", features = ["rt"] }
+tokio-postgres = "0.7.10"
+tokio-postgres-rustls = "0.11.1"
+tracing = "0.1.40"
 tracing-log = "0.2.0"
 typed-builder = "0.18.1"
 

crates/kitsune-db/src/lib.rs

@@ -1,10 +1,14 @@
+#[macro_use]
+extern crate tracing;
+
 use diesel::Connection;
 use diesel_async::{
     async_connection_wrapper::AsyncConnectionWrapper,
     pooled_connection::{deadpool::Pool, AsyncDieselConnectionManager},
     AsyncPgConnection,
 };
 use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
+use kitsune_config::database::Configuration as DatabaseConfig;
 use tracing_log::LogTracer;
 
 pub use crate::{
@@ -14,6 +18,7 @@ pub use crate::{
 
 mod error;
 mod pool;
+mod tls;
 
 pub mod activity;
 pub mod function;
@@ -27,11 +32,11 @@ pub mod schema;
 pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!();
 
 /// Connect to the database and run any pending migrations
-pub async fn connect(conn_str: &str, max_pool_size: usize) -> Result<PgPool> {
+pub async fn connect(config: &DatabaseConfig) -> Result<PgPool> {
     LogTracer::init().ok();
 
     kitsune_blocking::io({
-        let conn_str = conn_str.to_string();
+        let conn_str = config.url.clone();
 
         move || {
             let mut migration_conn =
@@ -46,9 +51,14 @@ pub async fn connect(conn_str: &str, max_pool_size: usize) -> Result<PgPool> {
     })
     .await??;
 
-    let config = AsyncDieselConnectionManager::<AsyncPgConnection>::new(conn_str);
-    let pool = Pool::builder(config)
-        .max_size(max_pool_size)
+    let pool_config = if config.use_tls {
+        AsyncDieselConnectionManager::new_with_config(config.url.as_str(), self::tls::pool_config())
+    } else {
+        AsyncDieselConnectionManager::new(config.url.as_str())
+    };
+
+    let pool = Pool::builder(pool_config)
+        .max_size(config.max_connections as usize)
         .build()
         .unwrap();
 

crates/kitsune-db/src/tls.rs

@@ -0,0 +1,38 @@
+use diesel::{ConnectionError, ConnectionResult};
+use diesel_async::{pooled_connection::ManagerConfig, AsyncPgConnection};
+use futures_util::{future::BoxFuture, FutureExt};
+
+pub fn pool_config() -> ManagerConfig<AsyncPgConnection> {
+    let mut config = ManagerConfig::default();
+    config.custom_setup = Box::new(establish_conn);
+    config
+}
+
+fn establish_conn(config: &str) -> BoxFuture<'_, ConnectionResult<AsyncPgConnection>> {
+    async {
+        let rustls_config = rustls::ClientConfig::builder()
+            .with_root_certificates(load_certs())
+            .with_no_client_auth();
+        let tls = tokio_postgres_rustls::MakeRustlsConnect::new(rustls_config);
+        let (client, conn) = tokio_postgres::connect(config, tls)
+            .await
+            .map_err(|err| ConnectionError::BadConnection(err.to_string()))?;
+
+        tokio::spawn(async move {
+            if let Err(err) = conn.await {
+                error!("Database connection error: {err}");
+            }
+        });
+
+        AsyncPgConnection::try_from(client).await
+    }
+    .boxed()
+}
+
+fn load_certs() -> rustls::RootCertStore {
+    let mut roots = rustls::RootCertStore::empty();
+    let certs =
+        rustls_native_certs::load_native_certs().expect("Failed to load native certificates");
+    roots.add_parsable_certificates(certs);
+    roots
+}

crates/kitsune-test/Cargo.toml

@@ -12,6 +12,7 @@ diesel-async = "0.4.1"
 futures-util = "0.3.30"
 http = "1.0.0"
 http-body-util = "0.1.0"
+kitsune-config = { path = "../kitsune-config" }
 kitsune-db = { path = "../kitsune-db" }
 pin-project-lite = "0.2.13"
 redis = "0.24.0"

crates/kitsune-test/src/lib.rs

@@ -4,6 +4,7 @@ use diesel_async::RunQueryDsl;
 use futures_util::Future;
 use http::header::CONTENT_TYPE;
 use http_body_util::Full;
+use kitsune_config::database::Configuration as DatabaseConfig;
 use kitsune_db::PgPool;
 use scoped_futures::ScopedFutureExt;
 use std::{env, error::Error, panic};
@@ -28,9 +29,13 @@ where
     Fut: Future,
 {
     let db_url = env::var("DATABASE_URL").expect("Missing database URL");
-    let pool = kitsune_db::connect(&db_url, 10)
-        .await
-        .expect("Failed to connect to database");
+    let pool = kitsune_db::connect(&DatabaseConfig {
+        url: db_url.into(),
+        max_connections: 10,
+        use_tls: false,
+    })
+    .await
+    .expect("Failed to connect to database");
 
     let out = CatchPanic::new(func(pool.clone())).await;
 

kitsune-cli/Cargo.toml

@@ -17,6 +17,7 @@ diesel = "2.1.4"
 diesel-async = "0.4.1"
 dotenvy = "0.15.7"
 envy = "0.4.2"
+kitsune-config = { path = "../crates/kitsune-config" }
 kitsune-db = { path = "../crates/kitsune-db" }
 miette = { version = "5.10.0", features = ["fancy"] }
 serde = { version = "1.0.195", features = ["derive"] }

kitsune-cli/src/config.rs

@@ -3,4 +3,6 @@ use serde::Deserialize;
 #[derive(Deserialize)]
 pub struct Configuration {
     pub database_url: String,
+    #[serde(default)]
+    pub database_use_tls: bool,
 }

kitsune-cli/src/main.rs

@@ -1,6 +1,7 @@
 use self::{config::Configuration, role::RoleSubcommand};
 use clap::{Parser, Subcommand};
 use diesel_async::scoped_futures::ScopedFutureExt;
+use kitsune_config::database::Configuration as DatabaseConfig;
 use miette::{IntoDiagnostic, Result};
 
 mod config;
@@ -28,7 +29,13 @@ async fn main() -> Result<()> {
     tracing_subscriber::fmt::init();
 
     let config: Configuration = envy::from_env().into_diagnostic()?;
-    let db_conn = kitsune_db::connect(&config.database_url, 1).await?;
+    let db_conn = kitsune_db::connect(&DatabaseConfig {
+        url: config.database_url.into(),
+        max_connections: 1,
+        use_tls: config.database_use_tls,
+    })
+    .await?;
+
     let cmd = App::parse();
 
     db_conn

kitsune-job-runner/src/main.rs

@@ -29,11 +29,7 @@ async fn main() -> miette::Result<()> {
 
     kitsune_observability::initialise(env!("CARGO_PKG_NAME"), &config)?;
 
-    let db_pool = kitsune_db::connect(
-        &config.database.url,
-        config.database.max_connections as usize,
-    )
-    .await?;
+    let db_pool = kitsune_db::connect(&config.database).await?;
     let job_queue = kitsune_job_runner::prepare_job_queue(db_pool.clone(), &config.job_queue)
         .into_diagnostic()?;
 

kitsune/src/main.rs

@@ -25,12 +25,9 @@ async fn boot() -> miette::Result<()> {
     let config = Configuration::load(args.config).await?;
     kitsune_observability::initialise(env!("CARGO_PKG_NAME"), &config)?;
 
-    let conn = kitsune_db::connect(
-        &config.database.url,
-        config.database.max_connections as usize,
-    )
-    .await
-    .wrap_err("Failed to connect to and migrate the database")?;
+    let conn = kitsune_db::connect(&config.database)
+        .await
+        .wrap_err("Failed to connect to and migrate the database")?;
 
     let job_queue = kitsune_job_runner::prepare_job_queue(conn.clone(), &config.job_queue)
         .into_diagnostic()