fix: sanitize_connection_string (#6012)

src/cmd/src/metasrv.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::time::Duration;
 
 use async_trait::async_trait;
@@ -131,7 +132,7 @@ impl SubCommand {
     }
 }
 
-#[derive(Debug, Default, Parser)]
+#[derive(Default, Parser)]
 pub struct StartCommand {
     /// The address to bind the gRPC server.
     #[clap(long, alias = "bind-addr")]
@@ -171,6 +172,27 @@ pub struct StartCommand {
     backend: Option<BackendImpl>,
 }
 
+impl fmt::Debug for StartCommand {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("StartCommand")
+            .field("rpc_bind_addr", &self.rpc_bind_addr)
+            .field("rpc_server_addr", &self.rpc_server_addr)
+            .field("store_addrs", &self.sanitize_store_addrs())
+            .field("config_file", &self.config_file)
+            .field("selector", &self.selector)
+            .field("use_memory_store", &self.use_memory_store)
+            .field("enable_region_failover", &self.enable_region_failover)
+            .field("http_addr", &self.http_addr)
+            .field("http_timeout", &self.http_timeout)
+            .field("env_prefix", &self.env_prefix)
+            .field("data_home", &self.data_home)
+            .field("store_key_prefix", &self.store_key_prefix)
+            .field("max_txn_ops", &self.max_txn_ops)
+            .field("backend", &self.backend)
+            .finish()
+    }
+}
+
 impl StartCommand {
     pub fn load_options(&self, global_options: &GlobalOptions) -> Result<MetasrvOptions> {
         let mut opts = MetasrvOptions::load_layered_options(
@@ -184,6 +206,15 @@ impl StartCommand {
         Ok(opts)
     }
 
+    fn sanitize_store_addrs(&self) -> Option<Vec<String>> {
+        self.store_addrs.as_ref().map(|addrs| {
+            addrs
+                .iter()
+                .map(|addr| common_meta::kv_backend::util::sanitize_connection_string(addr))
+                .collect()
+        })
+    }
+
     // The precedence order is: cli > config file > environment variables > default values.
     fn merge_with_cli_options(
         &self,

src/common/meta/src/kv_backend.rs

@@ -35,7 +35,7 @@ pub mod memory;
 pub mod rds;
 pub mod test;
 pub mod txn;
-
+pub mod util;
 pub type KvBackendRef<E = Error> = Arc<dyn KvBackend<Error = E> + Send + Sync>;
 
 #[async_trait]

src/common/meta/src/kv_backend/util.rs

@@ -0,0 +1,85 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// Removes sensitive information like passwords from connection strings.
+///
+/// This function sanitizes connection strings by removing credentials:
+/// - For URL format (mysql://user:password@host:port/db): Removes everything before '@'
+/// - For parameter format (host=localhost password=secret): Removes the password parameter
+/// - For URL format without credentials (mysql://host:port/db): Removes the protocol prefix
+///
+/// # Arguments
+///
+/// * `conn_str` - The connection string to sanitize
+///
+/// # Returns
+///
+/// A sanitized version of the connection string with sensitive information removed
+pub fn sanitize_connection_string(conn_str: &str) -> String {
+    // Case 1: URL format with credentials (mysql://user:password@host:port/db)
+    // Extract everything after the '@' symbol
+    if let Some(at_pos) = conn_str.find('@') {
+        return conn_str[at_pos + 1..].to_string();
+    }
+
+    // Case 2: Parameter format with password (host=localhost password=secret dbname=mydb)
+    // Filter out any parameter that starts with "password="
+    if conn_str.contains("password=") {
+        return conn_str
+            .split_whitespace()
+            .filter(|param| !param.starts_with("password="))
+            .collect::<Vec<_>>()
+            .join(" ");
+    }
+
+    // Case 3: URL format without credentials (mysql://host:port/db)
+    // Extract everything after the protocol prefix
+    if let Some(host_part) = conn_str.split("://").nth(1) {
+        return host_part.to_string();
+    }
+
+    // Case 4: Already sanitized or unknown format
+    // Return as is
+    conn_str.to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_sanitize_connection_string() {
+        // Test URL format with username/password
+        let conn_str = "mysql://user:password123@localhost:3306/db";
+        assert_eq!(sanitize_connection_string(conn_str), "localhost:3306/db");
+
+        // Test URL format without credentials
+        let conn_str = "mysql://localhost:3306/db";
+        assert_eq!(sanitize_connection_string(conn_str), "localhost:3306/db");
+
+        // Test parameter format with password
+        let conn_str = "host=localhost port=5432 user=postgres password=secret dbname=mydb";
+        assert_eq!(
+            sanitize_connection_string(conn_str),
+            "host=localhost port=5432 user=postgres dbname=mydb"
+        );
+
+        // Test parameter format without password
+        let conn_str = "host=localhost port=5432 user=postgres dbname=mydb";
+        assert_eq!(
+            sanitize_connection_string(conn_str),
+            "host=localhost port=5432 user=postgres dbname=mydb"
+        );
+    }
+}

src/meta-srv/src/metasrv.rs

@@ -14,7 +14,7 @@
 
 pub mod builder;
 
-use std::fmt::Display;
+use std::fmt::{self, Display};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
@@ -97,7 +97,7 @@ pub enum BackendImpl {
     MysqlStore,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, PartialEq, Serialize, Deserialize)]
 #[serde(default)]
 pub struct MetasrvOptions {
     /// The address the server listens on.
@@ -167,6 +167,47 @@ pub struct MetasrvOptions {
     pub node_max_idle_time: Duration,
 }
 
+impl fmt::Debug for MetasrvOptions {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut debug_struct = f.debug_struct("MetasrvOptions");
+        debug_struct
+            .field("bind_addr", &self.bind_addr)
+            .field("server_addr", &self.server_addr)
+            .field("store_addrs", &self.sanitize_store_addrs())
+            .field("selector", &self.selector)
+            .field("use_memory_store", &self.use_memory_store)
+            .field("enable_region_failover", &self.enable_region_failover)
+            .field(
+                "allow_region_failover_on_local_wal",
+                &self.allow_region_failover_on_local_wal,
+            )
+            .field("http", &self.http)
+            .field("logging", &self.logging)
+            .field("procedure", &self.procedure)
+            .field("failure_detector", &self.failure_detector)
+            .field("datanode", &self.datanode)
+            .field("enable_telemetry", &self.enable_telemetry)
+            .field("data_home", &self.data_home)
+            .field("wal", &self.wal)
+            .field("export_metrics", &self.export_metrics)
+            .field("store_key_prefix", &self.store_key_prefix)
+            .field("max_txn_ops", &self.max_txn_ops)
+            .field("flush_stats_factor", &self.flush_stats_factor)
+            .field("tracing", &self.tracing)
+            .field("backend", &self.backend);
+
+        #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
+        debug_struct.field("meta_table_name", &self.meta_table_name);
+
+        #[cfg(feature = "pg_kvbackend")]
+        debug_struct.field("meta_election_lock_id", &self.meta_election_lock_id);
+
+        debug_struct
+            .field("node_max_idle_time", &self.node_max_idle_time)
+            .finish()
+    }
+}
+
 const DEFAULT_METASRV_ADDR_PORT: &str = "3002";
 
 impl Default for MetasrvOptions {
@@ -250,6 +291,13 @@ impl MetasrvOptions {
             common_telemetry::debug!("detect local IP is not supported on Android");
         }
     }
+
+    fn sanitize_store_addrs(&self) -> Vec<String> {
+        self.store_addrs
+            .iter()
+            .map(|addr| common_meta::kv_backend::util::sanitize_connection_string(addr))
+            .collect()
+    }
 }
 
 pub struct MetasrvInfo {