feat(db): support TLS client certificates

crates/dbx-core/Cargo.toml

@@ -29,6 +29,7 @@ mysql_async = { version = "0.36", default-features = false, features = ["default
 sqlparser = "0.62.0"
 redis = { version = "0.32.2", features = ["tokio-comp", "tls-rustls", "tokio-rustls-comp", "sentinel"] }
 rustls = { version = "0.23", features = ["aws-lc-rs"] }
+rustls-pemfile = "2.2"
 duckdb = "1.3.2"
 tiberius = { version = "0.12.3", default-features = false, features = ["tds73", "chrono", "rust_decimal", "rustls", "sql-browser-tokio"] }
 reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls", "socks"] }

crates/dbx-core/src/db/mysql.rs

@@ -2,6 +2,7 @@ use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
 use futures::StreamExt;
 use mysql_async::consts::ColumnType;
 use mysql_async::prelude::*;
+use percent_encoding::percent_decode_str;
 use rust_decimal::Decimal;
 use std::borrow::Cow;
 use std::collections::HashSet;
@@ -14,6 +15,8 @@ use crate::types::{
     ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectInfo, QueryResult, TableInfo, TriggerInfo,
 };
 
+use super::file_validator::validate_file_path;
+
 pub type MySqlPool = mysql_async::Pool;
 
 fn quote_value(s: &str) -> String {
@@ -280,8 +283,17 @@ pub async fn connect_with_ca_cert(url: &str, ca_cert_path: Option<&str>) -> Resu
     result.map(|_| pool)
 }
 
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+struct MySqlTlsFiles {
+    sslcert: Option<String>,
+    sslkey: Option<String>,
+}
+
 fn create_pool(url: &str, ca_cert_path: Option<&str>) -> Result<MySqlPool, String> {
-    let opts = mysql_async::Opts::from_url(&mysql_async_url(url)).map_err(|e| format!("Invalid MySQL URL: {e}"))?;
+    let tls_url = mysql_tls_url(url)?;
+    let opts =
+        mysql_async::Opts::from_url(&mysql_async_url(&tls_url.url)).map_err(|e| format!("Invalid MySQL URL: {e}"))?;
+    let base_ssl_opts = opts.ssl_opts().cloned();
     let pool_opts = mysql_async::PoolOpts::new()
         .with_constraints(mysql_async::PoolConstraints::new(1, 1).unwrap())
         .with_inactive_connection_ttl(Duration::from_secs(300));
@@ -290,17 +302,109 @@ fn create_pool(url: &str, ca_cert_path: Option<&str>) -> Result<MySqlPool, Strin
         .prefer_socket(false)
         .pool_opts(Some(pool_opts))
         .setup(mysql_setup_queries(url));
-    if let Some(ca_cert_path) =
-        ca_cert_path.map(str::trim).filter(|path| !path.is_empty()).filter(|_| mysql_url_requires_ssl(url))
-    {
-        let ssl_opts = mysql_async::SslOpts::default()
-            .with_root_certs(vec![PathBuf::from(ca_cert_path).into()])
-            .with_danger_skip_domain_validation(true);
+    if let Some(ssl_opts) = mysql_ssl_opts(base_ssl_opts, url, ca_cert_path, &tls_url.files)? {
         builder = builder.ssl_opts(ssl_opts);
     }
     Ok(MySqlPool::new(builder))
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct MySqlTlsUrl {
+    url: String,
+    files: MySqlTlsFiles,
+}
+
+fn mysql_tls_url(url: &str) -> Result<MySqlTlsUrl, String> {
+    let Some(query_start) = url.find('?') else {
+        return Ok(MySqlTlsUrl { url: url.to_string(), files: MySqlTlsFiles::default() });
+    };
+
+    let prefix = &url[..query_start];
+    let suffix = &url[query_start + 1..];
+    let (query_string, fragment) = suffix.split_once('#').map_or((suffix, ""), |(query, fragment)| (query, fragment));
+    let mut files = MySqlTlsFiles::default();
+    let mut kept_params = Vec::new();
+
+    for param in query_string.split('&') {
+        if param.is_empty() {
+            continue;
+        }
+
+        let Some((key, value)) = param.split_once('=') else {
+            kept_params.push(param.to_string());
+            continue;
+        };
+
+        if mysql_tls_file_param_is(key, "cert") || mysql_tls_file_param_is(key, "key") {
+            let decoded = percent_decode_str(value)
+                .decode_utf8()
+                .map_err(|_| format!("Invalid URL encoding in {key}"))?
+                .into_owned();
+            validate_file_path(&decoded, |_| false).map_err(|e| format!("{key}: {e}"))?;
+
+            if mysql_tls_file_param_is(key, "cert") {
+                files.sslcert = Some(decoded);
+            } else {
+                files.sslkey = Some(decoded);
+            }
+        } else {
+            kept_params.push(param.to_string());
+        }
+    }
+
+    let mut sanitized_url = prefix.to_string();
+    if !kept_params.is_empty() {
+        sanitized_url.push('?');
+        sanitized_url.push_str(&kept_params.join("&"));
+    }
+    if !fragment.is_empty() {
+        sanitized_url.push('#');
+        sanitized_url.push_str(fragment);
+    }
+
+    Ok(MySqlTlsUrl { url: sanitized_url, files })
+}
+
+fn mysql_tls_file_param_is(key: &str, target: &str) -> bool {
+    let normalized = key.to_ascii_lowercase().replace(['-', '_'], "");
+    normalized == format!("ssl{target}")
+}
+
+fn mysql_ssl_opts(
+    base_ssl_opts: Option<mysql_async::SslOpts>,
+    url: &str,
+    ca_cert_path: Option<&str>,
+    files: &MySqlTlsFiles,
+) -> Result<Option<mysql_async::SslOpts>, String> {
+    let ca_cert_path = ca_cert_path.map(str::trim).filter(|path| !path.is_empty());
+    let has_client_identity = files.sslcert.as_deref().is_some() || files.sslkey.as_deref().is_some();
+    if !mysql_url_requires_ssl(url) && !has_client_identity {
+        return Ok(None);
+    }
+
+    let mut ssl_opts = base_ssl_opts.unwrap_or_default();
+    if let Some(ca_cert_path) = ca_cert_path.filter(|_| mysql_url_requires_ssl(url) || has_client_identity) {
+        ssl_opts = ssl_opts.with_root_certs(vec![PathBuf::from(ca_cert_path).into()]);
+        if !mysql_url_verifies_identity(url) {
+            ssl_opts = ssl_opts.with_danger_skip_domain_validation(true);
+        }
+    }
+
+    match (files.sslcert.as_deref(), files.sslkey.as_deref()) {
+        (Some(cert_path), Some(key_path)) => {
+            ssl_opts = ssl_opts.with_client_identity(Some(mysql_async::ClientIdentity::new(
+                PathBuf::from(cert_path).into(),
+                PathBuf::from(key_path).into(),
+            )));
+        }
+        (Some(_), None) => return Err("MySQL ssl-cert requires ssl-key".to_string()),
+        (None, Some(_)) => return Err("MySQL ssl-key requires ssl-cert".to_string()),
+        (None, None) => {}
+    }
+
+    Ok(Some(ssl_opts))
+}
+
 fn mysql_setup_queries(url: &str) -> Vec<String> {
     let charset = mysql_connection_charset(url).unwrap_or("utf8mb4");
     vec![format!("SET NAMES {charset}")]
@@ -424,6 +528,8 @@ fn mysql_url_requires_ssl(url: &str) -> bool {
         let key = key.trim();
         let value = value.trim();
         (key.eq_ignore_ascii_case("require_ssl") && value.eq_ignore_ascii_case("true"))
+            || mysql_tls_file_param_is(key, "cert")
+            || mysql_tls_file_param_is(key, "key")
             || ((key.eq_ignore_ascii_case("ssl-mode") || key.eq_ignore_ascii_case("sslmode"))
                 && matches!(
                     value.to_ascii_lowercase().replace('-', "_").as_str(),
@@ -432,6 +538,22 @@ fn mysql_url_requires_ssl(url: &str) -> bool {
     })
 }
 
+fn mysql_url_verifies_identity(url: &str) -> bool {
+    let Some((_, query)) = url.split_once('?') else {
+        return false;
+    };
+    query.split('&').any(|segment| {
+        let Some((key, value)) = segment.split_once('=') else {
+            return false;
+        };
+        let key = key.trim();
+        let value = value.trim();
+        (key.eq_ignore_ascii_case("verify_identity") && value.eq_ignore_ascii_case("true"))
+            || ((key.eq_ignore_ascii_case("ssl-mode") || key.eq_ignore_ascii_case("sslmode"))
+                && matches!(value.to_ascii_lowercase().replace('-', "_").as_str(), "verify_identity"))
+    })
+}
+
 fn mysql_async_url(url: &str) -> Cow<'_, str> {
     let Some((base, query)) = url.split_once('?') else {
         return Cow::Borrowed(url);
@@ -1082,6 +1204,43 @@ mod tests {
         assert!(mysql_error_should_retry_without_ssl(error));
     }
 
+    #[test]
+    fn mysql_tls_url_strips_client_identity_params_before_driver_parse() {
+        let dir = std::env::temp_dir();
+        let cert = dir.join(format!("dbx-mysql-client-cert-{}.pem", std::process::id()));
+        let key = dir.join(format!("dbx-mysql-client-key-{}.pem", std::process::id()));
+        std::fs::write(&cert, "not a real cert").unwrap();
+        std::fs::write(&key, "not a real key").unwrap();
+
+        let url = format!(
+            "mysql://root:secret@localhost/test?require_ssl=true&ssl-cert={}&ssl-key={}&charset=utf8mb4",
+            cert.display(),
+            key.display()
+        );
+        let parsed = mysql_tls_url(&url).unwrap();
+
+        assert_eq!(parsed.url, "mysql://root:secret@localhost/test?require_ssl=true&charset=utf8mb4");
+        assert_eq!(parsed.files.sslcert.as_deref(), Some(cert.to_str().unwrap()));
+        assert_eq!(parsed.files.sslkey.as_deref(), Some(key.to_str().unwrap()));
+        mysql_async::Opts::from_url(&mysql_async_url(&parsed.url)).unwrap();
+
+        let _ = std::fs::remove_file(cert);
+        let _ = std::fs::remove_file(key);
+    }
+
+    #[test]
+    fn mysql_tls_rejects_unpaired_client_cert_and_key() {
+        let files = MySqlTlsFiles { sslcert: Some("/tmp/client.crt".to_string()), sslkey: None };
+
+        let error = mysql_ssl_opts(None, "mysql://root@localhost/db?require_ssl=true", None, &files).unwrap_err();
+        assert!(error.contains("ssl-key"));
+    }
+
+    #[test]
+    fn mysql_tls_client_identity_requires_ssl() {
+        assert!(mysql_url_requires_ssl("mysql://root@localhost/db?ssl-cert=/tmp/client.crt&ssl-key=/tmp/client.key"));
+    }
+
     #[test]
     fn mysql_unknown_error_can_retry_with_text_protocol() {
         let error = "error returned from database: 1105 (HY000): Unknown error";

crates/dbx-core/src/db/postgres.rs

@@ -4,8 +4,12 @@ use futures::{SinkExt, StreamExt};
 use percent_encoding::percent_decode_str;
 use rust_decimal::Decimal;
 use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
+use rustls::client::verify_server_cert_signed_by_trust_anchor;
 use rustls::crypto::{verify_tls12_signature, verify_tls13_signature, CryptoProvider};
-use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
+use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime};
+use rustls::server::ParsedCertificate;
+use std::fs::File;
+use std::io::BufReader;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Instant;
@@ -319,17 +323,22 @@ async fn execute_select_query(
 }
 
 pub async fn connect(url: &str) -> Result<Pool, String> {
-    validate_postgres_ssl_paths(url)?;
+    let postgres_url = postgres_connection_url(url)?;
     let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
 
     let tz = iana_time_zone::get_timezone().unwrap_or_else(|_| "UTC".to_string());
 
     super::with_connection_timeout("PostgreSQL", super::connection_timeout(), async {
-        let pg_config =
-            tokio_postgres::Config::from_str(url).map_err(|e| format!("Invalid PostgreSQL connection URL: {e}"))?;
+        let pg_config = tokio_postgres::Config::from_str(&postgres_url.url)
+            .map_err(|e| format!("Invalid PostgreSQL connection URL: {e}"))?;
 
         let mgr_config = ManagerConfig { recycling_method: RecyclingMethod::Fast };
-        let tls_config = postgres_tls_config(&pg_config);
+        let tls_config = postgres_tls_config(
+            &pg_config,
+            &postgres_url.ssl_files,
+            postgres_url.accepts_invalid_certs,
+            postgres_url.verifies_hostname,
+        )?;
         let mgr = deadpool_postgres::Manager::from_config(
             pg_config.clone(),
             tokio_postgres_rustls::MakeRustlsConnect::new(tls_config),
@@ -357,18 +366,182 @@ pub async fn connect(url: &str) -> Result<Pool, String> {
     .await
 }
 
-fn postgres_tls_config(pg_config: &tokio_postgres::Config) -> rustls::ClientConfig {
-    if postgres_sslmode_accepts_invalid_certs(pg_config.get_ssl_mode()) {
-        let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
-        return rustls::ClientConfig::builder()
-            .dangerous()
-            .with_custom_certificate_verifier(Arc::new(NoPostgresCertVerification { provider }))
-            .with_no_client_auth();
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+struct PostgresSslFiles {
+    sslcert: Option<String>,
+    sslkey: Option<String>,
+    sslrootcert: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct PostgresConnectionUrl {
+    url: String,
+    ssl_files: PostgresSslFiles,
+    accepts_invalid_certs: bool,
+    verifies_hostname: bool,
+}
+
+fn postgres_connection_url(url: &str) -> Result<PostgresConnectionUrl, String> {
+    let Some(query_start) = url.find('?') else {
+        let pg_config =
+            tokio_postgres::Config::from_str(url).map_err(|e| format!("Invalid PostgreSQL connection URL: {e}"))?;
+        return Ok(PostgresConnectionUrl {
+            url: url.to_string(),
+            ssl_files: PostgresSslFiles::default(),
+            accepts_invalid_certs: postgres_sslmode_accepts_invalid_certs(pg_config.get_ssl_mode()),
+            verifies_hostname: false,
+        });
+    };
+
+    let prefix = &url[..query_start];
+    let suffix = &url[query_start + 1..];
+    let (query_string, fragment) = suffix.split_once('#').map_or((suffix, ""), |(query, fragment)| (query, fragment));
+    let mut ssl_files = PostgresSslFiles::default();
+    let mut kept_params = Vec::new();
+    let mut accepts_invalid_certs = true;
+    let mut verifies_hostname = false;
+
+    for param in query_string.split('&') {
+        if param.is_empty() {
+            continue;
+        }
+
+        let Some((key, value)) = param.split_once('=') else {
+            kept_params.push(param.to_string());
+            continue;
+        };
+
+        if key.eq_ignore_ascii_case("sslcert")
+            || key.eq_ignore_ascii_case("sslkey")
+            || key.eq_ignore_ascii_case("sslrootcert")
+        {
+            let decoded = percent_decode_str(value)
+                .decode_utf8()
+                .map_err(|_| format!("Invalid URL encoding in {key}"))?
+                .into_owned();
+            validate_file_path(&decoded, |_| false).map_err(|e| format!("{key}: {e}"))?;
+
+            if key.eq_ignore_ascii_case("sslcert") {
+                ssl_files.sslcert = Some(decoded);
+            } else if key.eq_ignore_ascii_case("sslkey") {
+                ssl_files.sslkey = Some(decoded);
+            } else {
+                ssl_files.sslrootcert = Some(decoded);
+            }
+        } else if key.eq_ignore_ascii_case("sslmode") {
+            match value.to_ascii_lowercase().as_str() {
+                "verify-ca" => {
+                    accepts_invalid_certs = false;
+                    kept_params.push("sslmode=require".to_string());
+                }
+                "verify-full" | "verify_identity" | "verify-identity" => {
+                    accepts_invalid_certs = false;
+                    verifies_hostname = true;
+                    kept_params.push("sslmode=require".to_string());
+                }
+                "disable" => {
+                    accepts_invalid_certs = false;
+                    kept_params.push(param.to_string());
+                }
+                "prefer" | "require" => {
+                    accepts_invalid_certs = true;
+                    kept_params.push(param.to_string());
+                }
+                _ => kept_params.push(param.to_string()),
+            }
+        } else {
+            kept_params.push(param.to_string());
+        }
     }
 
+    let mut sanitized_url = prefix.to_string();
+    if !kept_params.is_empty() {
+        sanitized_url.push('?');
+        sanitized_url.push_str(&kept_params.join("&"));
+    }
+    if !fragment.is_empty() {
+        sanitized_url.push('#');
+        sanitized_url.push_str(fragment);
+    }
+
+    Ok(PostgresConnectionUrl { url: sanitized_url, ssl_files, accepts_invalid_certs, verifies_hostname })
+}
+
+fn postgres_tls_config(
+    pg_config: &tokio_postgres::Config,
+    ssl_files: &PostgresSslFiles,
+    accepts_invalid_certs: bool,
+    verifies_hostname: bool,
+) -> Result<rustls::ClientConfig, String> {
+    if pg_config.get_ssl_mode() != SslMode::Disable && accepts_invalid_certs {
+        let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
+        let builder = rustls::ClientConfig::builder()
+            .dangerous()
+            .with_custom_certificate_verifier(Arc::new(NoPostgresCertVerification { provider }));
+        return postgres_tls_client_auth(builder, ssl_files);
+    }
+
+    let root_store = postgres_root_cert_store(ssl_files)?;
+    let builder = if verifies_hostname {
+        rustls::ClientConfig::builder().with_root_certificates(root_store)
+    } else {
+        let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
+        rustls::ClientConfig::builder().dangerous().with_custom_certificate_verifier(Arc::new(
+            PostgresCaOnlyCertVerification { provider, roots: Arc::new(root_store) },
+        ))
+    };
+    postgres_tls_client_auth(builder, ssl_files)
+}
+
+fn postgres_root_cert_store(ssl_files: &PostgresSslFiles) -> Result<rustls::RootCertStore, String> {
     let mut root_store = rustls::RootCertStore::empty();
-    root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
-    rustls::ClientConfig::builder().with_root_certificates(root_store).with_no_client_auth()
+    if let Some(path) = ssl_files.sslrootcert.as_deref() {
+        let certs = read_postgres_pem_certs("sslrootcert", path)?;
+        let (valid_count, _) = root_store.add_parsable_certificates(certs);
+        if valid_count == 0 {
+            return Err(format!("sslrootcert: no valid CA certificates found in {path}"));
+        }
+    } else {
+        root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
+    }
+    Ok(root_store)
+}
+
+fn postgres_tls_client_auth(
+    builder: rustls::ConfigBuilder<rustls::ClientConfig, rustls::client::WantsClientCert>,
+    ssl_files: &PostgresSslFiles,
+) -> Result<rustls::ClientConfig, String> {
+    match (ssl_files.sslcert.as_deref(), ssl_files.sslkey.as_deref()) {
+        (Some(cert_path), Some(key_path)) => {
+            let certs = read_postgres_pem_certs("sslcert", cert_path)?;
+            if certs.is_empty() {
+                return Err(format!("sslcert: no certificates found in {cert_path}"));
+            }
+            let private_key = read_postgres_private_key(key_path)?;
+            builder
+                .with_client_auth_cert(certs, private_key)
+                .map_err(|e| format!("PostgreSQL client certificate/key mismatch or invalid key: {e}"))
+        }
+        (Some(_), None) => Err("PostgreSQL sslcert requires sslkey".to_string()),
+        (None, Some(_)) => Err("PostgreSQL sslkey requires sslcert".to_string()),
+        (None, None) => Ok(builder.with_no_client_auth()),
+    }
+}
+
+fn read_postgres_pem_certs(label: &str, path: &str) -> Result<Vec<CertificateDer<'static>>, String> {
+    let file = File::open(path).map_err(|e| format!("{label}: failed to open {path}: {e}"))?;
+    let mut reader = BufReader::new(file);
+    rustls_pemfile::certs(&mut reader)
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|e| format!("{label}: failed to read PEM certificates from {path}: {e}"))
+}
+
+fn read_postgres_private_key(path: &str) -> Result<PrivateKeyDer<'static>, String> {
+    let file = File::open(path).map_err(|e| format!("sslkey: failed to open {path}: {e}"))?;
+    let mut reader = BufReader::new(file);
+    rustls_pemfile::private_key(&mut reader)
+        .map_err(|e| format!("sslkey: failed to read PEM private key from {path}: {e}"))?
+        .ok_or_else(|| format!("sslkey: no private key found in {path}"))
 }
 
 fn postgres_sslmode_accepts_invalid_certs(ssl_mode: SslMode) -> bool {
@@ -415,6 +588,55 @@ impl ServerCertVerifier for NoPostgresCertVerification {
     }
 }
 
+#[derive(Debug)]
+struct PostgresCaOnlyCertVerification {
+    provider: Arc<CryptoProvider>,
+    roots: Arc<rustls::RootCertStore>,
+}
+
+impl ServerCertVerifier for PostgresCaOnlyCertVerification {
+    fn verify_server_cert(
+        &self,
+        end_entity: &CertificateDer<'_>,
+        intermediates: &[CertificateDer<'_>],
+        _server_name: &ServerName<'_>,
+        _ocsp_response: &[u8],
+        now: UnixTime,
+    ) -> Result<ServerCertVerified, rustls::Error> {
+        let cert = ParsedCertificate::try_from(end_entity)?;
+        verify_server_cert_signed_by_trust_anchor(
+            &cert,
+            &self.roots,
+            intermediates,
+            now,
+            self.provider.signature_verification_algorithms.all,
+        )?;
+        Ok(ServerCertVerified::assertion())
+    }
+
+    fn verify_tls12_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &rustls::DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, rustls::Error> {
+        verify_tls12_signature(message, cert, dss, &self.provider.signature_verification_algorithms)
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &rustls::DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, rustls::Error> {
+        verify_tls13_signature(message, cert, dss, &self.provider.signature_verification_algorithms)
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
+        self.provider.signature_verification_algorithms.supported_schemes()
+    }
+}
+
 /// Check whether the user's connection URL already specifies a timezone via
 /// the `options` parameter so we don't overwrite it with the local timezone.
 fn pg_url_has_timezone_setting(url: &str) -> bool {
@@ -431,27 +653,9 @@ fn pg_url_has_timezone_setting(url: &str) -> bool {
     false
 }
 
+#[cfg(test)]
 fn validate_postgres_ssl_paths(url: &str) -> Result<(), String> {
-    if let Some(query_start) = url.find('?') {
-        let query_string = &url[query_start + 1..];
-
-        for param in query_string.split('&') {
-            if let Some((key, value)) = param.split_once('=') {
-                match key {
-                    "sslcert" | "sslkey" | "sslrootcert" => {
-                        let decoded = percent_decode_str(value)
-                            .decode_utf8()
-                            .map_err(|_| format!("Invalid URL encoding in {key}"))?;
-
-                        validate_file_path(&decoded, |_| false).map_err(|e| format!("{key}: {e}"))?;
-                    }
-                    _ => {}
-                }
-            }
-        }
-    }
-
-    Ok(())
+    postgres_connection_url(url).map(|_| ())
 }
 
 pub async fn list_databases(pool: &Pool) -> Result<Vec<DatabaseInfo>, String> {
@@ -1023,6 +1227,60 @@ mod tests {
         assert!(result.is_err());
     }
 
+    #[test]
+    fn postgres_connection_url_strips_ssl_file_params_before_driver_parse() {
+        let dir = std::env::temp_dir();
+        let cert = dir.join(format!("dbx-postgres-cert-{}.pem", std::process::id()));
+        let key = dir.join(format!("dbx-postgres-key-{}.pem", std::process::id()));
+        let root = dir.join(format!("dbx-postgres-root-{}.pem", std::process::id()));
+        std::fs::write(&cert, "not a real cert").unwrap();
+        std::fs::write(&key, "not a real key").unwrap();
+        std::fs::write(&root, "not a real root").unwrap();
+
+        let url = format!(
+            "postgres://localhost/db?sslmode=verify-full&sslcert={}&sslkey={}&sslrootcert={}&application_name=dbx",
+            cert.display(),
+            key.display(),
+            root.display()
+        );
+        let parsed = postgres_connection_url(&url).unwrap();
+
+        assert_eq!(parsed.url, "postgres://localhost/db?sslmode=require&application_name=dbx");
+        assert_eq!(parsed.ssl_files.sslcert.as_deref(), Some(cert.to_str().unwrap()));
+        assert_eq!(parsed.ssl_files.sslkey.as_deref(), Some(key.to_str().unwrap()));
+        assert_eq!(parsed.ssl_files.sslrootcert.as_deref(), Some(root.to_str().unwrap()));
+        assert!(!parsed.accepts_invalid_certs);
+        assert!(parsed.verifies_hostname);
+        tokio_postgres::Config::from_str(&parsed.url).unwrap();
+
+        let _ = std::fs::remove_file(cert);
+        let _ = std::fs::remove_file(key);
+        let _ = std::fs::remove_file(root);
+    }
+
+    #[test]
+    fn postgres_connection_url_keeps_verify_ca_ca_only_semantics() {
+        let parsed = postgres_connection_url("postgres://localhost/db?sslmode=verify-ca").unwrap();
+
+        assert_eq!(parsed.url, "postgres://localhost/db?sslmode=require");
+        assert!(!parsed.accepts_invalid_certs);
+        assert!(!parsed.verifies_hostname);
+    }
+
+    #[test]
+    fn postgres_tls_rejects_unpaired_client_cert_and_key() {
+        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+        let pg_config = tokio_postgres::Config::from_str("postgres://localhost/db?sslmode=require").unwrap();
+        let ssl_files =
+            PostgresSslFiles { sslcert: Some("/tmp/client.crt".to_string()), sslkey: None, sslrootcert: None };
+
+        let error = match postgres_tls_config(&pg_config, &ssl_files, true, false) {
+            Ok(_) => panic!("expected missing sslkey to fail"),
+            Err(error) => error,
+        };
+        assert!(error.contains("sslkey"));
+    }
+
     #[test]
     fn postgres_tls_accepts_invalid_certs_for_require_sslmode() {
         let pg_config = tokio_postgres::Config::from_str("postgres://localhost/db?sslmode=require").unwrap();

crates/dbx-core/tests/live_mtls.rs

@@ -0,0 +1,24 @@
+#[tokio::test]
+#[ignore = "requires the remote DBX mTLS smoke-test containers and local client certificates"]
+async fn live_mysql_mtls_connection_succeeds() {
+    let url = std::env::var("DBX_LIVE_MYSQL_MTLS_URL").expect("DBX_LIVE_MYSQL_MTLS_URL");
+    let ca = std::env::var("DBX_LIVE_MYSQL_MTLS_CA").ok();
+
+    let pool = dbx_core::db::mysql::connect_with_ca_cert(&url, ca.as_deref()).await.unwrap();
+    let result =
+        dbx_core::db::mysql::execute_query(&pool, "SELECT label FROM mtls_smoke WHERE id = 1", false).await.unwrap();
+
+    assert_eq!(result.rows[0][0], serde_json::json!("mysql mtls ok"));
+}
+
+#[tokio::test]
+#[ignore = "requires the remote DBX mTLS smoke-test containers and local client certificates"]
+async fn live_postgres_mtls_connection_succeeds() {
+    let url = std::env::var("DBX_LIVE_POSTGRES_MTLS_URL").expect("DBX_LIVE_POSTGRES_MTLS_URL");
+
+    let pool = dbx_core::db::postgres::connect(&url).await.unwrap();
+    let result =
+        dbx_core::db::postgres::execute_query(&pool, "SELECT label FROM mtls_smoke WHERE id = 1").await.unwrap();
+
+    assert_eq!(result.rows[0][0], serde_json::json!("postgres mtls ok"));
+}