feat: prevent sensitive information (key, password, secrets etc.) from being printed in plain (#1501)

* feat: add secret type

* chore: replace key, password, secrets with secret type.

* chore: use secrecy

* chore: remove redundant file

* style: taplo fmt

Cargo.lock

@@ -2488,6 +2488,7 @@ dependencies = [
  "prost",
  "query",
  "regex",
+ "secrecy",
  "serde",
  "serde_json",
  "servers",
@@ -7651,6 +7652,16 @@ version = "4.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b"
 
+[[package]]
+name = "secrecy"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9bd1c54ea06cfd2f6b63219704de0b9b4f72dcc2b8fdef820be6cd799780e91e"
+dependencies = [
+ "serde",
+ "zeroize",
+]
+
 [[package]]
 name = "security-framework"
 version = "2.8.2"
@@ -7873,6 +7884,7 @@ dependencies = [
  "rustls-pemfile",
  "schemars",
  "script",
+ "secrecy",
  "serde",
  "serde_json",
  "session",
@@ -8691,6 +8703,7 @@ dependencies = [
  "once_cell",
  "paste",
  "rand",
+ "secrecy",
  "serde",
  "serde_json",
  "servers",

src/cmd/src/frontend.rs

@@ -366,7 +366,10 @@ mod tests {
 
         let provider = provider.unwrap();
         let result = provider
-            .authenticate(Identity::UserId("test", None), Password::PlainText("test"))
+            .authenticate(
+                Identity::UserId("test", None),
+                Password::PlainText("test".to_string().into()),
+            )
             .await;
         assert!(result.is_ok());
     }

src/cmd/src/standalone.rs

@@ -385,7 +385,10 @@ mod tests {
         assert!(provider.is_some());
         let provider = provider.unwrap();
         let result = provider
-            .authenticate(Identity::UserId("test", None), Password::PlainText("test"))
+            .authenticate(
+                Identity::UserId("test", None),
+                Password::PlainText("test".to_string().into()),
+            )
             .await;
         assert!(result.is_ok());
     }
@@ -414,8 +417,9 @@ mod tests {
             sync_write = false
 
             [storage]
-            type = "File"
-            data_dir = "/tmp/greptimedb/data/"
+            type = "S3"
+            access_key_id = "access_key_id"
+            secret_access_key = "secret_access_key"
 
             [storage.compaction]
             max_inflight_tasks = 3
@@ -481,6 +485,17 @@ mod tests {
         assert!(fe_opts.influxdb_options.as_ref().unwrap().enable);
 
         assert_eq!("/tmp/greptimedb/test/wal", dn_opts.wal.dir);
+        match &dn_opts.storage.store {
+            datanode::datanode::ObjectStoreConfig::S3(s3_config) => {
+                assert_eq!(
+                    "Secret([REDACTED alloc::string::String])".to_string(),
+                    format!("{:?}", s3_config.access_key_id)
+                );
+            }
+            _ => {
+                unreachable!()
+            }
+        }
 
         assert_eq!("debug".to_string(), logging_opts.level);
         assert_eq!("/tmp/greptimedb/test/logs".to_string(), logging_opts.dir);

src/datanode/Cargo.toml

@@ -46,6 +46,7 @@ pin-project = "1.0"
 prost.workspace = true
 query = { path = "../query" }
 regex = "1.6"
+secrecy = { version = "0.8", features = ["serde", "alloc"] }
 serde = "1.0"
 serde_json = "1.0"
 servers = { path = "../servers" }

src/datanode/src/datanode.rs

@@ -19,6 +19,7 @@ use common_base::readable_size::ReadableSize;
 use common_telemetry::info;
 use common_telemetry::logging::LoggingOptions;
 use meta_client::MetaClientOptions;
+use secrecy::SecretString;
 use serde::{Deserialize, Serialize};
 use servers::http::HttpOptions;
 use servers::Mode;
@@ -56,31 +57,64 @@ pub struct FileConfig {
     pub data_dir: String,
 }
 
-#[derive(Debug, Clone, Serialize, Default, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(default)]
 pub struct S3Config {
     pub bucket: String,
     pub root: String,
-    pub access_key_id: String,
-    pub secret_access_key: String,
+    #[serde(skip_serializing)]
+    pub access_key_id: SecretString,
+    #[serde(skip_serializing)]
+    pub secret_access_key: SecretString,
     pub endpoint: Option<String>,
     pub region: Option<String>,
     pub cache_path: Option<String>,
     pub cache_capacity: Option<ReadableSize>,
 }
 
-#[derive(Debug, Clone, Serialize, Default, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(default)]
 pub struct OssConfig {
     pub bucket: String,
     pub root: String,
-    pub access_key_id: String,
-    pub access_key_secret: String,
+    #[serde(skip_serializing)]
+    pub access_key_id: SecretString,
+    #[serde(skip_serializing)]
+    pub access_key_secret: SecretString,
     pub endpoint: String,
     pub cache_path: Option<String>,
     pub cache_capacity: Option<ReadableSize>,
 }
 
+impl Default for S3Config {
+    fn default() -> Self {
+        Self {
+            bucket: String::default(),
+            root: String::default(),
+            access_key_id: SecretString::from(String::default()),
+            secret_access_key: SecretString::from(String::default()),
+            endpoint: Option::default(),
+            region: Option::default(),
+            cache_path: Option::default(),
+            cache_capacity: Option::default(),
+        }
+    }
+}
+
+impl Default for OssConfig {
+    fn default() -> Self {
+        Self {
+            bucket: String::default(),
+            root: String::default(),
+            access_key_id: SecretString::from(String::default()),
+            access_key_secret: SecretString::from(String::default()),
+            endpoint: String::default(),
+            cache_path: Option::default(),
+            cache_capacity: Option::default(),
+        }
+    }
+}
+
 impl Default for ObjectStoreConfig {
     fn default() -> Self {
         ObjectStoreConfig::File(FileConfig {
@@ -315,6 +349,8 @@ impl Datanode {
 
 #[cfg(test)]
 mod tests {
+    use secrecy::ExposeSecret;
+
     use super::*;
 
     #[test]
@@ -323,4 +359,25 @@ mod tests {
         let toml_string = toml::to_string(&opts).unwrap();
         let _parsed: DatanodeOptions = toml::from_str(&toml_string).unwrap();
     }
+
+    #[test]
+    fn test_secstr() {
+        let toml_str = r#"
+            [storage]
+            type = "S3"
+            access_key_id = "access_key_id"
+            secret_access_key = "secret_access_key"
+        "#;
+        let opts: DatanodeOptions = toml::from_str(toml_str).unwrap();
+        match opts.storage.store {
+            ObjectStoreConfig::S3(cfg) => {
+                assert_eq!(
+                    "Secret([REDACTED alloc::string::String])".to_string(),
+                    format!("{:?}", cfg.access_key_id)
+                );
+                assert_eq!("access_key_id", cfg.access_key_id.expose_secret());
+            }
+            _ => unreachable!(),
+        }
+    }
 }

src/datanode/src/instance.rs

@@ -39,6 +39,7 @@ use object_store::layers::{LoggingLayer, MetricsLayer, RetryLayer, TracingLayer}
 use object_store::services::{Fs as FsBuilder, Oss as OSSBuilder, S3 as S3Builder};
 use object_store::{util, ObjectStore, ObjectStoreBuilder};
 use query::query_engine::{QueryEngineFactory, QueryEngineRef};
+use secrecy::ExposeSecret;
 use servers::Mode;
 use session::context::QueryContext;
 use snafu::prelude::*;
@@ -374,8 +375,8 @@ pub(crate) async fn new_oss_object_store(store_config: &ObjectStoreConfig) -> Re
         .root(&root)
         .bucket(&oss_config.bucket)
         .endpoint(&oss_config.endpoint)
-        .access_key_id(&oss_config.access_key_id)
-        .access_key_secret(&oss_config.access_key_secret);
+        .access_key_id(oss_config.access_key_id.expose_secret())
+        .access_key_secret(oss_config.access_key_secret.expose_secret());
 
     let object_store = ObjectStore::new(builder)
         .with_context(|_| error::InitBackendSnafu {
@@ -439,8 +440,8 @@ pub(crate) async fn new_s3_object_store(store_config: &ObjectStoreConfig) -> Res
     builder
         .root(&root)
         .bucket(&s3_config.bucket)
-        .access_key_id(&s3_config.access_key_id)
-        .secret_access_key(&s3_config.secret_access_key);
+        .access_key_id(s3_config.access_key_id.expose_secret())
+        .secret_access_key(s3_config.secret_access_key.expose_secret());
 
     if s3_config.endpoint.is_some() {
         builder.endpoint(s3_config.endpoint.as_ref().unwrap());

src/servers/Cargo.toml

@@ -58,6 +58,7 @@ rustls = "0.21"
 rustls-pemfile = "1.0"
 rust-embed = { version = "6.6", features = ["debug-embed"] }
 schemars = "0.8"
+secrecy = { version = "0.8", features = ["serde", "alloc"] }
 serde.workspace = true
 serde_json = "1.0"
 session = { path = "../session" }

src/servers/src/auth.rs

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use common_error::ext::BoxedError;
 use common_error::prelude::ErrorExt;
 use common_error::status_code::StatusCode;
+use secrecy::SecretString;
 use session::context::UserInfo;
 use snafu::{Location, OptionExt, Snafu};
 
@@ -66,7 +67,7 @@ pub type Salt<'a> = &'a [u8];
 
 /// Authentication information sent by the client.
 pub enum Password<'a> {
-    PlainText(&'a str),
+    PlainText(SecretString),
     MysqlNativePassword(HashedPassword<'a>, Salt<'a>),
     PgMD5(HashedPassword<'a>, Salt<'a>),
 }

src/servers/src/auth/user_provider.rs

@@ -21,6 +21,7 @@ use std::path::Path;
 use async_trait::async_trait;
 use digest;
 use digest::Digest;
+use secrecy::ExposeSecret;
 use session::context::UserInfo;
 use sha1::Sha1;
 use snafu::{ensure, OptionExt, ResultExt};
@@ -120,12 +121,12 @@ impl UserProvider for StaticUserProvider {
                 match input_pwd {
                     Password::PlainText(pwd) => {
                         ensure!(
-                            !pwd.is_empty(),
+                            !pwd.expose_secret().is_empty(),
                             IllegalParamSnafu {
                                 msg: "blank password"
                             }
                         );
-                        return if save_pwd == pwd.as_bytes() {
+                        return if save_pwd == pwd.expose_secret().as_bytes() {
                             Ok(UserInfo::new(username))
                         } else {
                             UserPasswordMismatchSnafu {
@@ -240,7 +241,7 @@ pub mod test {
         let re = provider
             .authenticate(
                 Identity::UserId(username, None),
-                Password::PlainText(password),
+                Password::PlainText(password.to_string().into()),
             )
             .await;
         assert!(re.is_ok());

src/servers/src/grpc/handler.rs

@@ -109,7 +109,7 @@ impl GreptimeRequestHandler {
             AuthScheme::Basic(Basic { username, password }) => user_provider
                 .auth(
                     Identity::UserId(&username, None),
-                    Password::PlainText(&password),
+                    Password::PlainText(password.into()),
                     &query_ctx.current_catalog(),
                     &query_ctx.current_schema(),
                 )

src/servers/src/http/authorize.rs

@@ -21,6 +21,7 @@ use common_telemetry::warn;
 use futures::future::BoxFuture;
 use http_body::Body;
 use metrics::increment_counter;
+use secrecy::SecretString;
 use session::context::UserInfo;
 use snafu::{ensure, OptionExt, ResultExt};
 use tower_http::auth::AsyncAuthorizeRequest;
@@ -112,7 +113,7 @@ where
             match user_provider
                 .auth(
                     Identity::UserId(username.as_str(), None),
-                    crate::auth::Password::PlainText(password.as_str()),
+                    crate::auth::Password::PlainText(password),
                     catalog,
                     schema,
                 )
@@ -172,7 +173,7 @@ fn get_influxdb_credentials<B: Send + Sync + 'static>(
             .split_once(':')
             .context(InvalidAuthorizationHeaderSnafu)?;
 
-        Ok(Some((username.to_string(), password.to_string())))
+        Ok(Some((username.to_string(), password.to_string().into())))
     } else {
         // try v1
         let Some(query_str) = request.uri().query() else { return Ok(None) };
@@ -180,7 +181,7 @@ fn get_influxdb_credentials<B: Send + Sync + 'static>(
         match extract_influxdb_user_from_query(query_str) {
             (None, None) => Ok(None),
             (Some(username), Some(password)) => {
-                Ok(Some((username.to_string(), password.to_string())))
+                Ok(Some((username.to_string(), password.to_string().into())))
             }
             _ => Err(Auth {
                 source: IllegalParam {
@@ -222,7 +223,7 @@ pub enum AuthScheme {
 }
 
 type Username = String;
-type Password = String;
+type Password = SecretString;
 
 impl TryFrom<&str> for AuthScheme {
     type Error = error::Error;
@@ -262,7 +263,7 @@ fn decode_basic(credential: Credential) -> Result<(Username, Password)> {
     let as_utf8 = String::from_utf8(decoded).context(error::InvalidUtf8ValueSnafu)?;
 
     if let Some((user_id, password)) = as_utf8.split_once(':') {
-        return Ok((user_id.to_string(), password.to_string()));
+        return Ok((user_id.to_string(), password.to_string().into()));
     }
 
     InvalidAuthorizationHeaderSnafu {}.fail()
@@ -307,6 +308,8 @@ fn extract_influxdb_user_from_query(query: &str) -> (Option<&str>, Option<&str>)
 mod tests {
     use std::assert_matches::assert_matches;
 
+    use secrecy::ExposeSecret;
+
     use super::*;
 
     #[test]
@@ -339,7 +342,7 @@ mod tests {
         let credential = "dXNlcm5hbWU6cGFzc3dvcmQ=";
         let (username, pwd) = decode_basic(credential).unwrap();
         assert_eq!("username", username);
-        assert_eq!("password", pwd);
+        assert_eq!("password", pwd.expose_secret());
 
         let wrong_credential = "dXNlcm5hbWU6cG Fzc3dvcmQ=";
         let result = decode_basic(wrong_credential);
@@ -354,7 +357,7 @@ mod tests {
 
         let auth_scheme_str = "basic dGVzdDp0ZXN0";
         let scheme: AuthScheme = auth_scheme_str.try_into().unwrap();
-        assert_matches!(scheme, AuthScheme::Basic(username, pwd) if username == "test" && pwd == "test");
+        assert_matches!(scheme, AuthScheme::Basic(username, pwd) if username == "test" && pwd.expose_secret() == "test");
 
         let unsupported = "digest";
         let auth_scheme: Result<AuthScheme> = unsupported.try_into();
@@ -367,7 +370,7 @@ mod tests {
         let req = mock_http_request(Some("Basic dXNlcm5hbWU6cGFzc3dvcmQ="), None).unwrap();
 
         let auth_scheme = auth_header(&req).unwrap();
-        assert_matches!(auth_scheme, AuthScheme::Basic(username, pwd) if username == "username" && pwd == "password");
+        assert_matches!(auth_scheme, AuthScheme::Basic(username, pwd) if username == "username" && pwd.expose_secret() == "password");
 
         let wrong_req = mock_http_request(Some("Basic dXNlcm5hbWU6 cGFzc3dvcmQ="), None).unwrap();
         let res = auth_header(&wrong_req);

src/servers/src/postgres/auth_handler.rs

@@ -92,7 +92,7 @@ impl PgLoginVerifier {
         if let Err(e) = user_provider
             .auth(
                 Identity::UserId(user_name, None),
-                Password::PlainText(password),
+                Password::PlainText(password.to_string().into()),
                 catalog,
                 schema,
             )

src/servers/tests/auth.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use secrecy::ExposeSecret;
 use servers::auth::user_provider::auth_mysql;
 use servers::auth::{
     AccessDeniedSnafu, Identity, Password, UnsupportedPasswordTypeSnafu, UserNotFoundSnafu,
@@ -64,7 +65,7 @@ impl UserProvider for MockUserProvider {
             Identity::UserId(username, _host) => match password {
                 Password::PlainText(password) => {
                     if username == "greptime" {
-                        if password == "greptime" {
+                        if password.expose_secret() == "greptime" {
                             Ok(UserInfo::new("greptime"))
                         } else {
                             UserPasswordMismatchSnafu {
@@ -120,7 +121,7 @@ async fn test_auth_by_plain_text() {
     let auth_result = user_provider
         .authenticate(
             Identity::UserId("greptime", None),
-            Password::PlainText("greptime"),
+            Password::PlainText("greptime".to_string().into()),
         )
         .await;
     assert!(auth_result.is_ok());
@@ -143,7 +144,7 @@ async fn test_auth_by_plain_text() {
     let auth_result = user_provider
         .authenticate(
             Identity::UserId("not_exist_username", None),
-            Password::PlainText("greptime"),
+            Password::PlainText("greptime".to_string().into()),
         )
         .await;
     assert!(auth_result.is_err());
@@ -156,7 +157,7 @@ async fn test_auth_by_plain_text() {
     let auth_result = user_provider
         .authenticate(
             Identity::UserId("greptime", None),
-            Password::PlainText("wrong_password"),
+            Password::PlainText("wrong_password".to_string().into()),
         )
         .await;
     assert!(auth_result.is_err());

tests-integration/Cargo.toml

@@ -28,6 +28,7 @@ mito = { path = "../src/mito", features = ["test"] }
 object-store = { path = "../src/object-store" }
 once_cell = "1.16"
 rand.workspace = true
+secrecy = "0.8"
 serde.workspace = true
 serde_json = "1.0"
 servers = { path = "../src/servers" }

tests-integration/src/test_util.rs

@@ -40,6 +40,7 @@ use object_store::test_util::TempFolder;
 use object_store::ObjectStore;
 use once_cell::sync::OnceCell;
 use rand::Rng;
+use secrecy::ExposeSecret;
 use servers::grpc::GrpcServer;
 use servers::http::{HttpOptions, HttpServerBuilder};
 use servers::metrics_handler::MetricsHandler;
@@ -95,8 +96,8 @@ impl StorageType {
 fn s3_test_config() -> S3Config {
     S3Config {
         root: uuid::Uuid::new_v4().to_string(),
-        access_key_id: env::var("GT_S3_ACCESS_KEY_ID").unwrap(),
-        secret_access_key: env::var("GT_S3_ACCESS_KEY").unwrap(),
+        access_key_id: env::var("GT_S3_ACCESS_KEY_ID").unwrap().into(),
+        secret_access_key: env::var("GT_S3_ACCESS_KEY").unwrap().into(),
         bucket: env::var("GT_S3_BUCKET").unwrap(),
         region: Some(env::var("GT_S3_REGION").unwrap()),
         ..Default::default()
@@ -113,8 +114,8 @@ fn get_test_store_config(
         StorageType::Oss => {
             let oss_config = OssConfig {
                 root: uuid::Uuid::new_v4().to_string(),
-                access_key_id: env::var("GT_OSS_ACCESS_KEY_ID").unwrap(),
-                access_key_secret: env::var("GT_OSS_ACCESS_KEY").unwrap(),
+                access_key_id: env::var("GT_OSS_ACCESS_KEY_ID").unwrap().into(),
+                access_key_secret: env::var("GT_OSS_ACCESS_KEY").unwrap().into(),
                 bucket: env::var("GT_OSS_BUCKET").unwrap(),
                 endpoint: env::var("GT_OSS_ENDPOINT").unwrap(),
                 ..Default::default()
@@ -124,8 +125,8 @@ fn get_test_store_config(
             builder
                 .root(&oss_config.root)
                 .endpoint(&oss_config.endpoint)
-                .access_key_id(&oss_config.access_key_id)
-                .access_key_secret(&oss_config.access_key_secret)
+                .access_key_id(oss_config.access_key_id.expose_secret())
+                .access_key_secret(oss_config.access_key_secret.expose_secret())
                 .bucket(&oss_config.bucket);
 
             let config = ObjectStoreConfig::Oss(oss_config);
@@ -147,8 +148,8 @@ fn get_test_store_config(
             let mut builder = S3::default();
             builder
                 .root(&s3_config.root)
-                .access_key_id(&s3_config.access_key_id)
-                .secret_access_key(&s3_config.secret_access_key)
+                .access_key_id(s3_config.access_key_id.expose_secret())
+                .secret_access_key(s3_config.secret_access_key.expose_secret())
                 .bucket(&s3_config.bucket);
 
             if s3_config.endpoint.is_some() {