feat(auth): watch file user provider

Cargo.toml

@@ -110,6 +110,8 @@ lazy_static = "1.4"
 meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "80b72716dcde47ec4161478416a5c6c21343364d" }
 mockall = "0.11.4"
 moka = "0.12"
+notify = "6.1"
+notify-debouncer-full = "0.3"
 num_cpus = "1.16"
 once_cell = "1.18"
 opentelemetry-proto = { git = "https://github.com/waynexia/opentelemetry-rust.git", rev = "33841b38dda79b15f2024952be5f32533325ca02", features = [

src/auth/Cargo.toml

@@ -16,8 +16,11 @@ api.workspace = true
 async-trait.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
+common-telemetry.workspace = true
 digest = "0.10"
 hex = { version = "0.4" }
+notify.workspace = true
+notify-debouncer-full.workspace = true
 secrecy = { version = "0.8", features = ["serde", "alloc"] }
 sha1 = "0.10"
 snafu.workspace = true

src/auth/src/common.rs

@@ -22,6 +22,9 @@ use snafu::{ensure, OptionExt};
 use crate::error::{IllegalParamSnafu, InvalidConfigSnafu, Result, UserPasswordMismatchSnafu};
 use crate::user_info::DefaultUserInfo;
 use crate::user_provider::static_user_provider::{StaticUserProvider, STATIC_USER_PROVIDER};
+use crate::user_provider::watch_file_user_provider::{
+    WatchFileUserProvider, WATCH_FILE_USER_PROVIDER,
+};
 use crate::{UserInfoRef, UserProviderRef};
 
 pub(crate) const DEFAULT_USERNAME: &str = "greptime";
@@ -43,6 +46,9 @@ pub fn user_provider_from_option(opt: &String) -> Result<UserProviderRef> {
                 StaticUserProvider::new(content).map(|p| Arc::new(p) as UserProviderRef)?;
             Ok(provider)
         }
+        WATCH_FILE_USER_PROVIDER => {
+            WatchFileUserProvider::new(content).map(|p| Arc::new(p) as UserProviderRef)
+        }
         _ => InvalidConfigSnafu {
             value: name.to_string(),
             msg: "Invalid UserProviderOption",

src/auth/src/error.rs

@@ -64,6 +64,12 @@ pub enum Error {
         username: String,
     },
 
+    #[snafu(display("Failed to initialize a watcher for file"))]
+    FileWatch {
+        #[snafu(source)]
+        error: notify::Error,
+    },
+
     #[snafu(display("User is not authorized to perform this action"))]
     PermissionDenied { location: Location },
 }
@@ -73,6 +79,7 @@ impl ErrorExt for Error {
         match self {
             Error::InvalidConfig { .. } => StatusCode::InvalidArguments,
             Error::IllegalParam { .. } => StatusCode::InvalidArguments,
+            Error::FileWatch { .. } => StatusCode::InvalidArguments,
             Error::InternalState { .. } => StatusCode::Unexpected,
             Error::Io { .. } => StatusCode::Internal,
             Error::AuthBackend { .. } => StatusCode::Internal,

src/auth/src/user_provider.rs

@@ -13,10 +13,24 @@
 // limitations under the License.
 
 pub(crate) mod static_user_provider;
+pub(crate) mod watch_file_user_provider;
+
+use std::collections::HashMap;
+use std::fs::File;
+use std::io;
+use std::io::BufRead;
+use std::path::Path;
+
+use secrecy::ExposeSecret;
+use snafu::{ensure, OptionExt, ResultExt};
 
 use crate::common::{Identity, Password};
-use crate::error::Result;
-use crate::UserInfoRef;
+use crate::error::{
+    IllegalParamSnafu, InvalidConfigSnafu, IoSnafu, Result, UnsupportedPasswordTypeSnafu,
+    UserNotFoundSnafu, UserPasswordMismatchSnafu,
+};
+use crate::user_info::DefaultUserInfo;
+use crate::{auth_mysql, UserInfoRef};
 
 #[async_trait::async_trait]
 pub trait UserProvider: Send + Sync {
@@ -44,3 +58,85 @@ pub trait UserProvider: Send + Sync {
         Ok(user_info)
     }
 }
+
+fn load_credential_from_file(filepath: &str) -> Result<HashMap<String, Vec<u8>>> {
+    // check valid path
+    let path = Path::new(filepath);
+    ensure!(
+        path.exists() && path.is_file(),
+        InvalidConfigSnafu {
+            value: filepath.to_string(),
+            msg: "UserProvider file must be a valid file path",
+        }
+    );
+
+    let file = File::open(path).context(IoSnafu)?;
+    let credential = io::BufReader::new(file)
+        .lines()
+        .map_while(std::result::Result::ok)
+        .filter_map(|line| {
+            if let Some((k, v)) = line.split_once('=') {
+                Some((k.to_string(), v.as_bytes().to_vec()))
+            } else {
+                None
+            }
+        })
+        .collect::<HashMap<String, Vec<u8>>>();
+
+    ensure!(
+        !credential.is_empty(),
+        InvalidConfigSnafu {
+            value: filepath.to_string(),
+            msg: "UserProvider's file must contains at least one valid credential",
+        }
+    );
+
+    Ok(credential)
+}
+
+fn authenticate_with_credential(
+    users: &HashMap<String, Vec<u8>>,
+    input_id: Identity<'_>,
+    input_pwd: Password<'_>,
+) -> Result<UserInfoRef> {
+    match input_id {
+        Identity::UserId(username, _) => {
+            ensure!(
+                !username.is_empty(),
+                IllegalParamSnafu {
+                    msg: "blank username"
+                }
+            );
+            let save_pwd = users.get(username).context(UserNotFoundSnafu {
+                username: username.to_string(),
+            })?;
+
+            match input_pwd {
+                Password::PlainText(pwd) => {
+                    ensure!(
+                        !pwd.expose_secret().is_empty(),
+                        IllegalParamSnafu {
+                            msg: "blank password"
+                        }
+                    );
+                    if save_pwd == pwd.expose_secret().as_bytes() {
+                        Ok(DefaultUserInfo::with_name(username))
+                    } else {
+                        UserPasswordMismatchSnafu {
+                            username: username.to_string(),
+                        }
+                        .fail()
+                    }
+                }
+                Password::MysqlNativePassword(auth_data, salt) => {
+                    auth_mysql(auth_data, salt, username, save_pwd)
+                        .map(|_| DefaultUserInfo::with_name(username))
+                }
+                Password::PgMD5(_, _) => UnsupportedPasswordTypeSnafu {
+                    password_type: "pg_md5",
+                }
+                .fail(),
+            }
+        }
+    }
+}

src/auth/src/user_provider/static_user_provider.rs

@@ -13,21 +13,13 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-use std::fs::File;
-use std::io;
-use std::io::BufRead;
-use std::path::Path;
 
 use async_trait::async_trait;
-use secrecy::ExposeSecret;
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::OptionExt;
 
-use crate::error::{
-    IllegalParamSnafu, InvalidConfigSnafu, IoSnafu, Result, UnsupportedPasswordTypeSnafu,
-    UserNotFoundSnafu, UserPasswordMismatchSnafu,
-};
-use crate::user_info::DefaultUserInfo;
-use crate::{auth_mysql, Identity, Password, UserInfoRef, UserProvider};
+use crate::error::{InvalidConfigSnafu, Result};
+use crate::user_provider::{authenticate_with_credential, load_credential_from_file};
+use crate::{Identity, Password, UserInfoRef, UserProvider};
 
 pub(crate) const STATIC_USER_PROVIDER: &str = "static_user_provider";
 
@@ -43,32 +35,8 @@ impl StaticUserProvider {
         })?;
         return match mode {
             "file" => {
-                // check valid path
-                let path = Path::new(content);
-                ensure!(path.exists() && path.is_file(), InvalidConfigSnafu {
-                    value: content.to_string(),
-                    msg: "StaticUserProviderOption file must be a valid file path",
-                });
-
-                let file = File::open(path).context(IoSnafu)?;
-                let credential = io::BufReader::new(file)
-                    .lines()
-                    .map_while(std::result::Result::ok)
-                    .filter_map(|line| {
-                        if let Some((k, v)) = line.split_once('=') {
-                            Some((k.to_string(), v.as_bytes().to_vec()))
-                        } else {
-                            None
-                        }
-                    })
-                    .collect::<HashMap<String, Vec<u8>>>();
-
-                ensure!(!credential.is_empty(), InvalidConfigSnafu {
-                    value: content.to_string(),
-                    msg: "StaticUserProviderOption file must contains at least one valid credential",
-                });
-
-                Ok(StaticUserProvider { users: credential, })
+                let users = load_credential_from_file(content)?;
+                Ok(StaticUserProvider { users })
             }
             "cmd" => content
                 .split(',')
@@ -101,46 +69,7 @@ impl UserProvider for StaticUserProvider {
         input_id: Identity<'_>,
         input_pwd: Password<'_>,
     ) -> Result<UserInfoRef> {
-        match input_id {
-            Identity::UserId(username, _) => {
-                ensure!(
-                    !username.is_empty(),
-                    IllegalParamSnafu {
-                        msg: "blank username"
-                    }
-                );
-                let save_pwd = self.users.get(username).context(UserNotFoundSnafu {
-                    username: username.to_string(),
-                })?;
-
-                match input_pwd {
-                    Password::PlainText(pwd) => {
-                        ensure!(
-                            !pwd.expose_secret().is_empty(),
-                            IllegalParamSnafu {
-                                msg: "blank password"
-                            }
-                        );
-                        return if save_pwd == pwd.expose_secret().as_bytes() {
-                            Ok(DefaultUserInfo::with_name(username))
-                        } else {
-                            UserPasswordMismatchSnafu {
-                                username: username.to_string(),
-                            }
-                            .fail()
-                        };
-                    }
-                    Password::MysqlNativePassword(auth_data, salt) => {
-                        auth_mysql(auth_data, salt, username, save_pwd)
-                            .map(|_| DefaultUserInfo::with_name(username))
-                    }
-                    Password::PgMD5(_, _) => UnsupportedPasswordTypeSnafu {
-                        password_type: "pg_md5",
-                    }
-                    .fail(),
-                }
-            }
-        }
+        authenticate_with_credential(&self.users, input_id, input_pwd)
     }
 
     async fn authorize(