feat: add TLS support for mysql backend

scripts/generate_pgsql_certs.sh → scripts/generate_certs.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-CERT_DIR="${1:-$(dirname "$0")/../tests-integration/fixtures/pgsql-certs}"
+CERT_DIR="${1:-$(dirname "$0")/../tests-integration/fixtures/certs}"
 DAYS="${2:-365}"
 
 mkdir -p "${CERT_DIR}"
@@ -10,13 +10,13 @@ cd "${CERT_DIR}"
 echo "Generating CA certificate..."
 openssl req -new -x509 -days "${DAYS}" -nodes -text \
   -out root.crt -keyout root.key \
-  -subj "/CN=PostgresRootCA"
+  -subj "/CN=GreptimeDBRootCA"
 
 
 echo "Generating server certificate..."
 openssl req -new -nodes -text \
   -out server.csr -keyout server.key \
-  -subj "/CN=postgres"
+  -subj "/CN=greptime"
 
 openssl x509 -req -in server.csr -text -days "${DAYS}" \
   -CA root.crt -CAkey root.key -CAcreateserial \
@@ -36,6 +36,6 @@ rm -f *.csr
 
 echo "TLS certificates generated successfully in ${CERT_DIR}"
 
-chmod 600 root.key
-chmod 600 client.key
-chmod 600 server.key
+chmod 644 root.key
+chmod 644 client.key
+chmod 644 server.key

src/cli/src/metadata/common.rs

@@ -19,8 +19,8 @@ use common_error::ext::BoxedError;
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::kv_backend::chroot::ChrootKvBackend;
 use common_meta::kv_backend::etcd::EtcdStore;
-use meta_srv::bootstrap::create_etcd_client_with_tls;
 use meta_srv::metasrv::BackendImpl;
+use meta_srv::utils::etcd::create_etcd_client_with_tls;
 use servers::tls::{TlsMode, TlsOption};
 
 use crate::error::{EmptyStoreAddrsSnafu, UnsupportedMemoryBackendSnafu};
@@ -116,9 +116,13 @@ impl StoreConfig {
                 BackendImpl::PostgresStore => {
                     let table_name = &self.meta_table_name;
                     let tls_config = self.tls_config();
-                    let pool = meta_srv::bootstrap::create_postgres_pool(store_addrs, tls_config)
-                        .await
-                        .map_err(BoxedError::new)?;
+                    let pool = meta_srv::utils::postgres::create_postgres_pool(
+                        store_addrs,
+                        None,
+                        tls_config,
+                    )
+                    .await
+                    .map_err(BoxedError::new)?;
                     let schema_name = self.meta_schema_name.as_deref();
                     Ok(common_meta::kv_backend::rds::PgStore::with_pg_pool(
                         pool,
@@ -132,9 +136,11 @@ impl StoreConfig {
                 #[cfg(feature = "mysql_kvbackend")]
                 BackendImpl::MysqlStore => {
                     let table_name = &self.meta_table_name;
-                    let pool = meta_srv::bootstrap::create_mysql_pool(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
+                    let tls_config = self.tls_config();
+                    let pool =
+                        meta_srv::utils::mysql::create_mysql_pool(store_addrs, tls_config.as_ref())
+                            .await
+                            .map_err(BoxedError::new)?;
                     Ok(common_meta::kv_backend::rds::MySqlStore::with_mysql_pool(
                         pool,
                         table_name,

src/common/meta/src/error.rs

@@ -752,7 +752,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[cfg(feature = "pg_kvbackend")]
     #[snafu(display("Failed to load TLS certificate from path: {}", path))]
     LoadTlsCertificate {
         path: String,
@@ -1181,13 +1180,14 @@ impl ErrorExt for Error {
             | InvalidRole { .. }
             | EmptyDdlTasks { .. } => StatusCode::InvalidArguments,
 
+            LoadTlsCertificate { .. } => StatusCode::Internal,
+
             #[cfg(feature = "pg_kvbackend")]
             PostgresExecution { .. }
             | CreatePostgresPool { .. }
             | GetPostgresConnection { .. }
             | PostgresTransaction { .. }
             | PostgresTlsConfig { .. }
-            | LoadTlsCertificate { .. }
             | InvalidTlsConfig { .. } => StatusCode::Internal,
             #[cfg(feature = "mysql_kvbackend")]
             MySqlExecution { .. } | CreateMySqlPool { .. } | MySqlTransaction { .. } => {

src/common/meta/src/kv_backend/etcd.rs

@@ -13,15 +13,17 @@
 // limitations under the License.
 
 use std::any::Any;
+use std::fs;
 use std::sync::Arc;
 
-use common_telemetry::info;
+use common_telemetry::{debug, info};
 use etcd_client::{
-    Client, DeleteOptions, GetOptions, PutOptions, Txn, TxnOp, TxnOpResponse, TxnResponse,
+    Certificate, Client, DeleteOptions, GetOptions, Identity, PutOptions, TlsOptions, Txn, TxnOp,
+    TxnOpResponse, TxnResponse,
 };
 use snafu::{ResultExt, ensure};
 
-use crate::error::{self, Error, Result};
+use crate::error::{self, Error, LoadTlsCertificateSnafu, Result};
 use crate::kv_backend::txn::{Txn as KvTxn, TxnResponse as KvTxnResponse};
 use crate::kv_backend::{KvBackend, KvBackendRef, TxnService};
 use crate::metrics::METRIC_META_TXN_REQUEST;
@@ -451,8 +453,76 @@ impl TryFrom<DeleteRangeRequest> for Delete {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub enum TlsMode {
+    #[default]
+    Disable,
+    Require,
+}
+
+/// TLS configuration for Etcd connections.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TlsOption {
+    pub mode: TlsMode,
+    pub cert_path: String,
+    pub key_path: String,
+    pub ca_cert_path: String,
+}
+
+/// Creates a Etcd [`TlsOptions`] from a [`TlsOption`].
+///
+/// This function builds the TLS options for etcd client connections based on the provided
+/// [`TlsOption`]. It supports disabling TLS, setting a custom CA certificate, and configuring
+/// client identity for mutual TLS authentication.
+///
+/// Note: All TlsMode variants except [`TlsMode::Disable`] will be treated as enabling TLS.
+pub fn create_etcd_tls_options(tls_config: &TlsOption) -> Result<Option<TlsOptions>> {
+    // If TLS mode is disabled, return None to indicate no TLS configuration.
+    if matches!(tls_config.mode, TlsMode::Disable) {
+        return Ok(None);
+    }
+
+    info!("Creating etcd TLS with mode: {:?}", tls_config.mode);
+    // Start with default TLS options.
+    let mut etcd_tls_opts = TlsOptions::new();
+
+    // If a CA certificate path is provided, load the CA certificate and add it to the options.
+    if !tls_config.ca_cert_path.is_empty() {
+        debug!("Using CA certificate from {}", tls_config.ca_cert_path);
+        let ca_cert_pem = fs::read(&tls_config.ca_cert_path).context(LoadTlsCertificateSnafu {
+            path: &tls_config.ca_cert_path,
+        })?;
+        let ca_cert = Certificate::from_pem(ca_cert_pem);
+        etcd_tls_opts = etcd_tls_opts.ca_certificate(ca_cert);
+    }
+
+    // If both client certificate and key paths are provided, load them and set the client identity.
+    if !tls_config.cert_path.is_empty() && !tls_config.key_path.is_empty() {
+        info!("Loading client certificate for mutual TLS");
+        debug!(
+            "Using client certificate from {} and key from {}",
+            tls_config.cert_path, tls_config.key_path
+        );
+        let cert_pem = fs::read(&tls_config.cert_path).context(LoadTlsCertificateSnafu {
+            path: &tls_config.cert_path,
+        })?;
+        let key_pem = fs::read(&tls_config.key_path).context(LoadTlsCertificateSnafu {
+            path: &tls_config.key_path,
+        })?;
+        let identity = Identity::from_pem(cert_pem, key_pem);
+        etcd_tls_opts = etcd_tls_opts.identity(identity);
+    }
+
+    // Always enable native TLS roots for additional trust anchors.
+    etcd_tls_opts = etcd_tls_opts.with_native_roots();
+
+    Ok(Some(etcd_tls_opts))
+}
+
 #[cfg(test)]
 mod tests {
+    use etcd_client::ConnectOptions;
+
     use super::*;
 
     #[test]
@@ -555,6 +625,8 @@ mod tests {
         test_txn_compare_not_equal, test_txn_one_compare_op, text_txn_multi_compare_op,
         unprepare_kv,
     };
+    use crate::maybe_skip_etcd_tls_integration_test;
+    use crate::test_util::etcd_certs_dir;
 
     async fn build_kv_backend() -> Option<EtcdStore> {
         let endpoints = std::env::var("GT_ETCD_ENDPOINTS").unwrap_or_default();
@@ -654,4 +726,41 @@ mod tests {
             test_txn_compare_not_equal(&kv_backend).await;
         }
     }
+
+    async fn create_etcd_client_with_tls(endpoints: &[String], tls_config: &TlsOption) -> Client {
+        let endpoints = endpoints
+            .iter()
+            .map(|s| s.trim())
+            .filter(|s| !s.is_empty())
+            .collect::<Vec<_>>();
+        let connect_options =
+            ConnectOptions::new().with_tls(create_etcd_tls_options(tls_config).unwrap().unwrap());
+
+        Client::connect(&endpoints, Some(connect_options))
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_create_etcd_client_with_mtls_and_ca() {
+        maybe_skip_etcd_tls_integration_test!();
+        let endpoints = std::env::var("GT_ETCD_TLS_ENDPOINTS")
+            .unwrap()
+            .split(',')
+            .map(|s| s.to_string())
+            .collect::<Vec<_>>();
+
+        let cert_dir = etcd_certs_dir();
+        let tls_config = TlsOption {
+            mode: TlsMode::Require,
+            ca_cert_path: cert_dir.join("ca.crt").to_string_lossy().to_string(),
+            cert_path: cert_dir.join("client.crt").to_string_lossy().to_string(),
+            key_path: cert_dir
+                .join("client-key.pem")
+                .to_string_lossy()
+                .to_string(),
+        };
+        let mut client = create_etcd_client_with_tls(&endpoints, &tls_config).await;
+        let _ = client.get(b"hello", None).await.unwrap();
+    }
 }

src/common/meta/src/kv_backend/rds/mysql.rs

@@ -573,6 +573,7 @@ impl MySqlStore {
 #[cfg(test)]
 mod tests {
     use common_telemetry::init_default_ut_logging;
+    use sqlx::mysql::{MySqlConnectOptions, MySqlSslMode};
 
     use super::*;
     use crate::kv_backend::test::{
@@ -584,6 +585,7 @@ mod tests {
         text_txn_multi_compare_op, unprepare_kv,
     };
     use crate::maybe_skip_mysql_integration_test;
+    use crate::test_util::test_certs_dir;
 
     async fn build_mysql_kv_backend(table_name: &str) -> Option<MySqlStore> {
         init_default_ut_logging();
@@ -711,4 +713,71 @@ mod tests {
         test_txn_compare_less(&kv_backend).await;
         test_txn_compare_not_equal(&kv_backend).await;
     }
+
+    #[tokio::test]
+    async fn test_mysql_with_tls() {
+        common_telemetry::init_default_ut_logging();
+        maybe_skip_mysql_integration_test!();
+        let endpoint = std::env::var("GT_MYSQL_ENDPOINTS").unwrap();
+
+        let opts = endpoint
+            .parse::<MySqlConnectOptions>()
+            .unwrap()
+            .ssl_mode(MySqlSslMode::Required);
+        let pool = MySqlPool::connect_with(opts).await.unwrap();
+        sqlx::query("SELECT 1").execute(&pool).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_mysql_with_mtls() {
+        common_telemetry::init_default_ut_logging();
+        maybe_skip_mysql_integration_test!();
+        let endpoint = std::env::var("GT_MYSQL_ENDPOINTS").unwrap();
+        let certs_dir = test_certs_dir();
+
+        let opts = endpoint
+            .parse::<MySqlConnectOptions>()
+            .unwrap()
+            .ssl_mode(MySqlSslMode::Required)
+            .ssl_client_cert(certs_dir.join("client.crt").to_string_lossy().to_string())
+            .ssl_client_key(certs_dir.join("client.key").to_string_lossy().to_string());
+        let pool = MySqlPool::connect_with(opts).await.unwrap();
+        sqlx::query("SELECT 1").execute(&pool).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_mysql_with_tls_verify_ca() {
+        common_telemetry::init_default_ut_logging();
+        maybe_skip_mysql_integration_test!();
+        let endpoint = std::env::var("GT_MYSQL_ENDPOINTS").unwrap();
+        let certs_dir = test_certs_dir();
+
+        let opts = endpoint
+            .parse::<MySqlConnectOptions>()
+            .unwrap()
+            .ssl_mode(MySqlSslMode::VerifyCa)
+            .ssl_ca(certs_dir.join("root.crt").to_string_lossy().to_string())
+            .ssl_client_cert(certs_dir.join("client.crt").to_string_lossy().to_string())
+            .ssl_client_key(certs_dir.join("client.key").to_string_lossy().to_string());
+        let pool = MySqlPool::connect_with(opts).await.unwrap();
+        sqlx::query("SELECT 1").execute(&pool).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_mysql_with_tls_verify_ident() {
+        common_telemetry::init_default_ut_logging();
+        maybe_skip_mysql_integration_test!();
+        let endpoint = std::env::var("GT_MYSQL_ENDPOINTS").unwrap();
+        let certs_dir = test_certs_dir();
+
+        let opts = endpoint
+            .parse::<MySqlConnectOptions>()
+            .unwrap()
+            .ssl_mode(MySqlSslMode::VerifyIdentity)
+            .ssl_ca(certs_dir.join("root.crt").to_string_lossy().to_string())
+            .ssl_client_cert(certs_dir.join("client.crt").to_string_lossy().to_string())
+            .ssl_client_key(certs_dir.join("client.key").to_string_lossy().to_string());
+        let pool = MySqlPool::connect_with(opts).await.unwrap();
+        sqlx::query("SELECT 1").execute(&pool).await.unwrap();
+    }
 }

src/common/meta/src/kv_backend/rds/postgres.rs

@@ -903,7 +903,7 @@ mod tests {
         test_txn_compare_less, test_txn_compare_not_equal, test_txn_one_compare_op,
         text_txn_multi_compare_op, unprepare_kv,
     };
-    use crate::test_util::pgsql_certs_dir;
+    use crate::test_util::test_certs_dir;
     use crate::{maybe_skip_postgres_integration_test, maybe_skip_postgres15_integration_test};
 
     async fn build_pg_kv_backend(table_name: &str) -> Option<PgStore> {
@@ -1020,7 +1020,7 @@ mod tests {
     async fn test_pg_with_mtls() {
         common_telemetry::init_default_ut_logging();
         maybe_skip_postgres_integration_test!();
-        let certs_dir = pgsql_certs_dir();
+        let certs_dir = test_certs_dir();
         let endpoints = std::env::var("GT_POSTGRES_ENDPOINTS").unwrap();
         let tls_connector = create_postgres_tls_connector(&TlsOption {
             mode: TlsMode::Require,
@@ -1043,7 +1043,7 @@ mod tests {
     async fn test_pg_verify_ca() {
         common_telemetry::init_default_ut_logging();
         maybe_skip_postgres_integration_test!();
-        let certs_dir = pgsql_certs_dir();
+        let certs_dir = test_certs_dir();
         let endpoints = std::env::var("GT_POSTGRES_ENDPOINTS").unwrap();
         let tls_connector = create_postgres_tls_connector(&TlsOption {
             mode: TlsMode::VerifyCa,
@@ -1066,7 +1066,7 @@ mod tests {
     async fn test_pg_verify_full() {
         common_telemetry::init_default_ut_logging();
         maybe_skip_postgres_integration_test!();
-        let certs_dir = pgsql_certs_dir();
+        let certs_dir = test_certs_dir();
         let endpoints = std::env::var("GT_POSTGRES_ENDPOINTS").unwrap();
         let tls_connector = create_postgres_tls_connector(&TlsOption {
             mode: TlsMode::VerifyFull,