fix: add TLS connection settings

crates/dbx-core/src/connection.rs

@@ -175,7 +175,7 @@ impl AppState {
                 PoolKind::Mysql(db::mysql::connect_bare(&url).await?, MysqlMode::Bare)
             }
             DatabaseType::Mysql => {
-                let pool = db::mysql::connect(&url).await?;
+                let pool = db::mysql::connect_with_ca_cert(&url, Some(&db_config.ca_cert_path)).await?;
                 let mode = detect_ob_oracle_mode(&db_config, &pool).await;
                 PoolKind::Mysql(pool, mode)
             }
@@ -611,9 +611,14 @@ fn native_postgres_url_config(config: &ConnectionConfig) -> Option<ConnectionCon
                 let params = normalized.url_params.as_deref().unwrap_or("").trim().trim_start_matches('?');
                 if !params.to_lowercase().contains("sslmode=") {
                     normalized.url_params = Some(if params.is_empty() {
-                        "sslmode=disable".to_string()
+                        if config.ssl {
+                            "sslmode=require".to_string()
+                        } else {
+                            "sslmode=disable".to_string()
+                        }
                     } else {
-                        format!("sslmode=disable&{params}")
+                        let sslmode = if config.ssl { "sslmode=require" } else { "sslmode=disable" };
+                        format!("{sslmode}&{params}")
                     });
                 }
             }
@@ -1228,6 +1233,20 @@ mod tests {
         );
     }
 
+    #[test]
+    fn gaussdb_endpoint_url_uses_require_sslmode_when_tls_enabled() {
+        let mut config = mysql_config(Some("postgres"));
+        config.db_type = DatabaseType::Gaussdb;
+        config.username = "gaussdb".to_string();
+        config.password = "secret".to_string();
+        config.ssl = true;
+
+        assert_eq!(
+            connection_url_for_endpoint(&config, &config.host, config.port),
+            "postgres://gaussdb:secret@127.0.0.1:3306/postgres?sslmode=require"
+        );
+    }
+
     #[test]
     fn gaussdb_endpoint_url_prepends_default_sslmode_to_custom_params() {
         let mut config = mysql_config(Some("postgres"));

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

@@ -5,6 +5,7 @@ use mysql_async::prelude::*;
 use rust_decimal::Decimal;
 use std::borrow::Cow;
 use std::collections::HashSet;
+use std::path::PathBuf;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -172,15 +173,19 @@ fn mysql_value_to_json(row: &mysql_async::Row, idx: usize) -> serde_json::Value
 }
 
 pub async fn connect(url: &str) -> Result<MySqlPool, String> {
+    connect_with_ca_cert(url, None).await
+}
+
+pub async fn connect_with_ca_cert(url: &str, ca_cert_path: Option<&str>) -> Result<MySqlPool, String> {
     let timeout = super::parse_connect_timeout(url);
-    let pool = create_pool(url)?;
+    let pool = create_pool(url, ca_cert_path)?;
     let result = verify_pool_connection(&pool, timeout).await;
 
     if let Err(ref e) = result {
         if mysql_error_should_retry_without_ssl(e) {
             if let Some(fallback_url) = ssl_fallback_url(url) {
                 log::info!("SSL handshake failed, retrying with ssl-mode=disabled");
-                let fallback_pool = create_pool(&fallback_url)?;
+                let fallback_pool = create_pool(&fallback_url, None)?;
                 return match verify_pool_connection(&fallback_pool, timeout).await {
                     Ok(()) => Ok(fallback_pool),
                     Err(e) => Err(e),
@@ -192,16 +197,24 @@ pub async fn connect(url: &str) -> Result<MySqlPool, String> {
     result.map(|_| pool)
 }
 
-fn create_pool(url: &str) -> Result<MySqlPool, 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 pool_opts = mysql_async::PoolOpts::new()
         .with_constraints(mysql_async::PoolConstraints::new(1, 1).unwrap())
         .with_inactive_connection_ttl(Duration::from_secs(300));
-    let builder = mysql_async::OptsBuilder::from_opts(opts)
+    let mut builder = mysql_async::OptsBuilder::from_opts(opts)
         .stmt_cache_size(0)
         .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);
+        builder = builder.ssl_opts(ssl_opts);
+    }
     Ok(MySqlPool::new(builder))
 }
 
@@ -251,6 +264,9 @@ fn mysql_error_should_retry_with_text_protocol(error: &str) -> bool {
 }
 
 fn ssl_fallback_url(url: &str) -> Option<String> {
+    if mysql_url_requires_ssl(url) {
+        return None;
+    }
     if url.contains("ssl-mode=preferred") {
         Some(url.replace("ssl-mode=preferred", "ssl-mode=disabled"))
     } else if !url.contains("ssl-mode=") {
@@ -261,26 +277,69 @@ fn ssl_fallback_url(url: &str) -> Option<String> {
     }
 }
 
+fn mysql_url_requires_ssl(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("require_ssl") && 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(),
+                    "required" | "require" | "verify_ca" | "verify_identity"
+                ))
+    })
+}
+
 fn mysql_async_url(url: &str) -> Cow<'_, str> {
     let Some((base, query)) = url.split_once('?') else {
         return Cow::Borrowed(url);
     };
 
-    let filtered: Vec<&str> = query
-        .split('&')
-        .filter(|segment| {
-            let segment = segment.trim();
-            !segment.is_empty()
-                && !segment.starts_with("ssl-mode=")
-                && !segment.starts_with("charset=")
-                && !segment.starts_with("time_zone=")
-                && !segment.starts_with("time-zone=")
-                && !segment.to_ascii_lowercase().starts_with("connect_timeout=")
-                && !segment.to_ascii_lowercase().starts_with("connecttimeout=")
-        })
-        .collect();
+    let original_count = query.split('&').filter(|segment| !segment.trim().is_empty()).count();
+    let mut filtered: Vec<String> = Vec::new();
+    for segment in query.split('&') {
+        let segment = segment.trim();
+        if segment.is_empty()
+            || segment.starts_with("charset=")
+            || segment.starts_with("time_zone=")
+            || segment.starts_with("time-zone=")
+            || segment.to_ascii_lowercase().starts_with("connect_timeout=")
+            || segment.to_ascii_lowercase().starts_with("connecttimeout=")
+        {
+            continue;
+        }
 
-    if filtered.len() == query.split('&').filter(|segment| !segment.trim().is_empty()).count() {
+        let Some((key, value)) = segment.split_once('=') else {
+            filtered.push(segment.to_string());
+            continue;
+        };
+        if key.eq_ignore_ascii_case("ssl-mode") || key.eq_ignore_ascii_case("sslmode") {
+            match value.to_ascii_lowercase().replace('-', "_").as_str() {
+                "disabled" | "disable" => filtered.push("require_ssl=false".to_string()),
+                "required" | "require" => {
+                    filtered.push("require_ssl=true".to_string());
+                    filtered.push("verify_ca=false".to_string());
+                    filtered.push("verify_identity=false".to_string());
+                }
+                "verify_ca" => {
+                    filtered.push("require_ssl=true".to_string());
+                    filtered.push("verify_identity=false".to_string());
+                }
+                "verify_identity" => filtered.push("require_ssl=true".to_string()),
+                _ => {}
+            }
+            continue;
+        }
+        filtered.push(segment.to_string());
+    }
+
+    if filtered.len() == original_count {
         Cow::Borrowed(url)
     } else if filtered.is_empty() {
         Cow::Owned(base.to_string())
@@ -291,7 +350,7 @@ fn mysql_async_url(url: &str) -> Cow<'_, str> {
 
 pub async fn connect_bare(url: &str) -> Result<MySqlPool, String> {
     let timeout = super::parse_connect_timeout(url);
-    let pool = create_pool(url)?;
+    let pool = create_pool(url, None)?;
     verify_pool_connection(&pool, timeout).await.map(|_| pool)
 }
 
@@ -857,6 +916,22 @@ mod tests {
         assert_eq!(crate::db::parse_connect_timeout(url), default);
     }
 
+    #[test]
+    fn mysql_async_url_translates_standard_required_ssl_mode() {
+        let url = "mysql://host:3306/db?ssl-mode=required&charset=utf8mb4";
+
+        assert_eq!(
+            mysql_async_url(url).as_ref(),
+            "mysql://host:3306/db?require_ssl=true&verify_ca=false&verify_identity=false"
+        );
+    }
+
+    #[test]
+    fn ssl_fallback_does_not_disable_required_tls() {
+        assert_eq!(ssl_fallback_url("mysql://host:3306/db?require_ssl=true&charset=utf8mb4"), None);
+        assert_eq!(ssl_fallback_url("mysql://host:3306/db?ssl-mode=verify_ca&charset=utf8mb4"), None);
+    }
+
     #[test]
     fn mysql_setup_queries_default_to_utf8mb4() {
         assert_eq!(mysql_setup_queries("mysql://host:3306/db"), vec!["SET NAMES utf8mb4"]);

crates/dbx-core/src/models/connection.rs

@@ -494,26 +494,7 @@ impl ConnectionConfig {
             };
         }
         match self.db_type {
-            DatabaseType::Mysql => {
-                let base = "ssl-mode=preferred&charset=utf8mb4";
-                if value.is_empty() {
-                    base.to_string()
-                } else if value.contains("ssl-mode=") {
-                    let v = value.trim_start_matches('?');
-                    if v.contains("charset=") {
-                        v.to_string()
-                    } else {
-                        format!("{v}&charset=utf8mb4")
-                    }
-                } else {
-                    let v = value.trim_start_matches('?');
-                    if v.contains("charset=") {
-                        format!("ssl-mode=preferred&{v}")
-                    } else {
-                        format!("{base}&{v}")
-                    }
-                }
-            }
+            DatabaseType::Mysql => normalize_mysql_url_params(value, self.ssl, self.ca_cert_path.trim().is_empty()),
             DatabaseType::Doris | DatabaseType::StarRocks => {
                 let v = value.trim_start_matches('?');
                 let filtered: Vec<&str> = v
@@ -526,7 +507,7 @@ impl ConnectionConfig {
                     format!("ssl-mode=disabled&{}", filtered.join("&"))
                 }
             }
-            DatabaseType::Postgres | DatabaseType::Redshift => normalize_postgres_url_params(value),
+            DatabaseType::Postgres | DatabaseType::Redshift => normalize_postgres_url_params(value, self.ssl),
             DatabaseType::MongoDb => value.trim_start_matches('?').to_string(),
             _ => value.trim_start_matches('?').to_string(),
         }
@@ -543,12 +524,37 @@ fn url_params_contains_flag(params: Option<&str>, key: &str, expected: &str) ->
     )
 }
 
-fn normalize_postgres_url_params(value: &str) -> String {
+fn normalize_mysql_url_params(value: &str, force_tls: bool, accept_invalid_certs: bool) -> String {
     let value = value.trim_start_matches('?');
-    if value.is_empty() {
-        return String::new();
+    let mut parts: Vec<String> = value.split('&').filter(|part| !part.is_empty()).map(str::to_string).collect();
+
+    if force_tls {
+        parts.retain(|part| !url_param_key_is(part, "ssl-mode") && !url_param_key_is(part, "sslmode"));
+        if !parts.iter().any(|part| url_param_key_is(part, "require_ssl")) {
+            parts.insert(0, "require_ssl=true".to_string());
+        }
+        if accept_invalid_certs && !parts.iter().any(|part| url_param_key_is(part, "verify_ca")) {
+            parts.push("verify_ca=false".to_string());
+        }
+        if !parts.iter().any(|part| url_param_key_is(part, "verify_identity")) {
+            parts.push("verify_identity=false".to_string());
+        }
+    } else if !parts.iter().any(|part| {
+        url_param_key_is(part, "ssl-mode") || url_param_key_is(part, "sslmode") || url_param_key_is(part, "require_ssl")
+    }) {
+        parts.insert(0, "ssl-mode=preferred".to_string());
     }
 
+    if !parts.iter().any(|part| url_param_key_is(part, "charset")) {
+        parts.push("charset=utf8mb4".to_string());
+    }
+
+    parts.join("&")
+}
+
+fn normalize_postgres_url_params(value: &str, force_tls: bool) -> String {
+    let value = value.trim_start_matches('?');
+
     let mut timezone: Option<String> = None;
     let mut parts: Vec<String> = Vec::new();
 
@@ -566,6 +572,9 @@ fn normalize_postgres_url_params(value: &str) -> String {
     }
 
     let Some(timezone) = timezone else {
+        if force_tls && !parts.iter().any(|part| url_param_key_is(part, "sslmode")) {
+            parts.insert(0, "sslmode=require".to_string());
+        }
         return parts.join("&");
     };
 
@@ -585,9 +594,18 @@ fn normalize_postgres_url_params(value: &str) -> String {
         parts.push(format!("options={}", encode_url_part(&timezone_option)));
     }
 
+    if force_tls && !parts.iter().any(|part| url_param_key_is(part, "sslmode")) {
+        parts.insert(0, "sslmode=require".to_string());
+    }
+
     parts.join("&")
 }
 
+fn url_param_key_is(part: &str, expected: &str) -> bool {
+    let key = part.split_once('=').map(|(key, _)| key).unwrap_or(part);
+    percent_decode_str(key).decode_utf8_lossy().eq_ignore_ascii_case(expected)
+}
+
 fn clickhouse_http_url(config: &ConnectionConfig, host: &str, port: u16) -> String {
     let trimmed = host.trim();
     if let Some(rest) = trimmed.strip_prefix("https://") {
@@ -985,6 +1003,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn mysql_tls_switch_requires_ssl_without_strict_certificate_checks_by_default() {
+        let mut config = mysql_config("root", "secret", Some("test"));
+        config.ssl = true;
+
+        assert_eq!(
+            config.connection_url(),
+            "mysql://root:secret@10.1.2.3:2883/test?require_ssl=true&verify_ca=false&verify_identity=false&charset=utf8mb4"
+        );
+    }
+
+    #[test]
+    fn mysql_tls_switch_uses_ca_cert_for_ca_validation() {
+        let mut config = mysql_config("root", "secret", Some("test"));
+        config.ssl = true;
+        config.ca_cert_path = "/tmp/tidb-ca.pem".to_string();
+
+        assert_eq!(
+            config.connection_url(),
+            "mysql://root:secret@10.1.2.3:2883/test?require_ssl=true&verify_identity=false&charset=utf8mb4"
+        );
+    }
+
     #[test]
     fn postgres_url_appends_custom_params() {
         let mut config = mysql_config("postgres", "secret", Some("test"));
@@ -994,6 +1035,15 @@ mod tests {
         assert_eq!(config.connection_url(), "postgres://postgres:secret@10.1.2.3:2883/test?sslmode=disable");
     }
 
+    #[test]
+    fn postgres_tls_switch_adds_require_sslmode() {
+        let mut config = mysql_config("postgres", "secret", Some("test"));
+        config.db_type = DatabaseType::Postgres;
+        config.ssl = true;
+
+        assert_eq!(config.connection_url(), "postgres://postgres:secret@10.1.2.3:2883/test?sslmode=require");
+    }
+
     #[test]
     fn postgres_url_normalizes_timezone_param_into_options() {
         let mut config = mysql_config("postgres", "secret", Some("test"));