feat: supports permission mode for static user provider (#7017)

* feat: supports permission mode for static user provider

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore: style

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: comment

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

src/auth/src/lib.rs

@@ -25,7 +25,7 @@ pub use common::{
     HashedPassword, Identity, Password, auth_mysql, static_user_provider_from_option,
     user_provider_from_option, userinfo_by_name,
 };
-pub use permission::{PermissionChecker, PermissionReq, PermissionResp};
+pub use permission::{DefaultPermissionChecker, PermissionChecker, PermissionReq, PermissionResp};
 pub use user_info::UserInfo;
 pub use user_provider::UserProvider;
 pub use user_provider::static_user_provider::StaticUserProvider;

src/auth/src/permission.rs

@@ -13,12 +13,15 @@
 // limitations under the License.
 
 use std::fmt::Debug;
+use std::sync::Arc;
 
 use api::v1::greptime_request::Request;
+use common_telemetry::debug;
 use sql::statements::statement::Statement;
 
 use crate::error::{PermissionDeniedSnafu, Result};
-use crate::{PermissionCheckerRef, UserInfoRef};
+use crate::user_info::DefaultUserInfo;
+use crate::{PermissionCheckerRef, UserInfo, UserInfoRef};
 
 #[derive(Debug, Clone)]
 pub enum PermissionReq<'a> {
@@ -35,6 +38,32 @@ pub enum PermissionReq<'a> {
     BulkInsert,
 }
 
+impl<'a> PermissionReq<'a> {
+    /// Returns true if the permission request is for read operations.
+    pub fn is_readonly(&self) -> bool {
+        match self {
+            PermissionReq::GrpcRequest(Request::Query(_))
+            | PermissionReq::PromQuery
+            | PermissionReq::LogQuery
+            | PermissionReq::PromStoreRead => true,
+            PermissionReq::SqlStatement(stmt) => stmt.is_readonly(),
+
+            PermissionReq::GrpcRequest(_)
+            | PermissionReq::Opentsdb
+            | PermissionReq::LineProtocol
+            | PermissionReq::PromStoreWrite
+            | PermissionReq::Otlp
+            | PermissionReq::LogWrite
+            | PermissionReq::BulkInsert => false,
+        }
+    }
+
+    /// Returns true if the permission request is for write operations.
+    pub fn is_write(&self) -> bool {
+        !self.is_readonly()
+    }
+}
+
 #[derive(Debug)]
 pub enum PermissionResp {
     Allow,
@@ -65,3 +94,106 @@ impl PermissionChecker for Option<&PermissionCheckerRef> {
         }
     }
 }
+
+/// The default permission checker implementation.
+/// It checks the permission mode of [DefaultUserInfo].
+pub struct DefaultPermissionChecker;
+
+impl DefaultPermissionChecker {
+    /// Returns a new [PermissionCheckerRef] instance.
+    pub fn arc() -> PermissionCheckerRef {
+        Arc::new(DefaultPermissionChecker)
+    }
+}
+
+impl PermissionChecker for DefaultPermissionChecker {
+    fn check_permission(
+        &self,
+        user_info: UserInfoRef,
+        req: PermissionReq,
+    ) -> Result<PermissionResp> {
+        if let Some(default_user) = user_info.as_any().downcast_ref::<DefaultUserInfo>() {
+            let permission_mode = default_user.permission_mode();
+
+            if req.is_readonly() && !permission_mode.can_read() {
+                debug!(
+                    "Permission denied: read operation not allowed, user = {}, permission = {}",
+                    default_user.username(),
+                    permission_mode.as_str()
+                );
+                return Ok(PermissionResp::Reject);
+            }
+
+            if req.is_write() && !permission_mode.can_write() {
+                debug!(
+                    "Permission denied: write operation not allowed, user = {}, permission = {}",
+                    default_user.username(),
+                    permission_mode.as_str()
+                );
+                return Ok(PermissionResp::Reject);
+            }
+        }
+
+        // default allow all
+        Ok(PermissionResp::Allow)
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::user_info::PermissionMode;
+
+    #[test]
+    fn test_default_permission_checker_allow_all_operations() {
+        let checker = DefaultPermissionChecker;
+        let user_info =
+            DefaultUserInfo::with_name_and_permission("test_user", PermissionMode::ReadWrite);
+
+        let read_req = PermissionReq::PromQuery;
+        let write_req = PermissionReq::PromStoreWrite;
+
+        let read_result = checker
+            .check_permission(user_info.clone(), read_req)
+            .unwrap();
+        let write_result = checker.check_permission(user_info, write_req).unwrap();
+
+        assert!(matches!(read_result, PermissionResp::Allow));
+        assert!(matches!(write_result, PermissionResp::Allow));
+    }
+
+    #[test]
+    fn test_default_permission_checker_readonly_user() {
+        let checker = DefaultPermissionChecker;
+        let user_info =
+            DefaultUserInfo::with_name_and_permission("readonly_user", PermissionMode::ReadOnly);
+
+        let read_req = PermissionReq::PromQuery;
+        let write_req = PermissionReq::PromStoreWrite;
+
+        let read_result = checker
+            .check_permission(user_info.clone(), read_req)
+            .unwrap();
+        let write_result = checker.check_permission(user_info, write_req).unwrap();
+
+        assert!(matches!(read_result, PermissionResp::Allow));
+        assert!(matches!(write_result, PermissionResp::Reject));
+    }
+
+    #[test]
+    fn test_default_permission_checker_writeonly_user() {
+        let checker = DefaultPermissionChecker;
+        let user_info =
+            DefaultUserInfo::with_name_and_permission("writeonly_user", PermissionMode::WriteOnly);
+
+        let read_req = PermissionReq::LogQuery;
+        let write_req = PermissionReq::LogWrite;
+
+        let read_result = checker
+            .check_permission(user_info.clone(), read_req)
+            .unwrap();
+        let write_result = checker.check_permission(user_info, write_req).unwrap();
+
+        assert!(matches!(read_result, PermissionResp::Reject));
+        assert!(matches!(write_result, PermissionResp::Allow));
+    }
+}

src/auth/src/user_info.rs

@@ -23,17 +23,86 @@ pub trait UserInfo: Debug + Sync + Send {
     fn username(&self) -> &str;
 }
 
+/// The user permission mode
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PermissionMode {
+    #[default]
+    ReadWrite,
+    ReadOnly,
+    WriteOnly,
+}
+
+impl PermissionMode {
+    /// Parse permission mode from string.
+    /// Supported values are:
+    /// - "rw", "readwrite", "read_write" => ReadWrite
+    /// - "ro", "readonly", "read_only" => ReadOnly
+    /// - "wo", "writeonly", "write_only" => WriteOnly
+    ///     Returns None if the input string is not a valid permission mode.
+    pub fn from_str(s: &str) -> Self {
+        match s.to_lowercase().as_str() {
+            "readwrite" | "read_write" | "rw" => PermissionMode::ReadWrite,
+            "readonly" | "read_only" | "ro" => PermissionMode::ReadOnly,
+            "writeonly" | "write_only" | "wo" => PermissionMode::WriteOnly,
+            _ => PermissionMode::ReadWrite,
+        }
+    }
+
+    /// Convert permission mode to string.
+    /// - ReadWrite => "rw"
+    /// - ReadOnly => "ro"
+    /// - WriteOnly => "wo"
+    ///     The returned string is a static string slice.
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            PermissionMode::ReadWrite => "rw",
+            PermissionMode::ReadOnly => "ro",
+            PermissionMode::WriteOnly => "wo",
+        }
+    }
+
+    /// Returns true if the permission mode allows read operations.
+    pub fn can_read(&self) -> bool {
+        matches!(self, PermissionMode::ReadWrite | PermissionMode::ReadOnly)
+    }
+
+    /// Returns true if the permission mode allows write operations.
+    pub fn can_write(&self) -> bool {
+        matches!(self, PermissionMode::ReadWrite | PermissionMode::WriteOnly)
+    }
+}
+
+impl std::fmt::Display for PermissionMode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.as_str())
+    }
+}
+
 #[derive(Debug)]
 pub(crate) struct DefaultUserInfo {
     username: String,
+    permission_mode: PermissionMode,
 }
 
 impl DefaultUserInfo {
     pub(crate) fn with_name(username: impl Into<String>) -> UserInfoRef {
+        Self::with_name_and_permission(username, PermissionMode::default())
+    }
+
+    /// Create a UserInfo with specified permission mode.
+    pub(crate) fn with_name_and_permission(
+        username: impl Into<String>,
+        permission_mode: PermissionMode,
+    ) -> UserInfoRef {
         Arc::new(Self {
             username: username.into(),
+            permission_mode,
         })
     }
+
+    pub(crate) fn permission_mode(&self) -> &PermissionMode {
+        &self.permission_mode
+    }
 }
 
 impl UserInfo for DefaultUserInfo {
@@ -45,3 +114,120 @@ impl UserInfo for DefaultUserInfo {
         self.username.as_str()
     }
 }
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_permission_mode_from_str() {
+        // Test ReadWrite variants
+        assert_eq!(
+            PermissionMode::from_str("readwrite"),
+            PermissionMode::ReadWrite
+        );
+        assert_eq!(
+            PermissionMode::from_str("read_write"),
+            PermissionMode::ReadWrite
+        );
+        assert_eq!(PermissionMode::from_str("rw"), PermissionMode::ReadWrite);
+        assert_eq!(
+            PermissionMode::from_str("ReadWrite"),
+            PermissionMode::ReadWrite
+        );
+        assert_eq!(PermissionMode::from_str("RW"), PermissionMode::ReadWrite);
+
+        // Test ReadOnly variants
+        assert_eq!(
+            PermissionMode::from_str("readonly"),
+            PermissionMode::ReadOnly
+        );
+        assert_eq!(
+            PermissionMode::from_str("read_only"),
+            PermissionMode::ReadOnly
+        );
+        assert_eq!(PermissionMode::from_str("ro"), PermissionMode::ReadOnly);
+        assert_eq!(
+            PermissionMode::from_str("ReadOnly"),
+            PermissionMode::ReadOnly
+        );
+        assert_eq!(PermissionMode::from_str("RO"), PermissionMode::ReadOnly);
+
+        // Test WriteOnly variants
+        assert_eq!(
+            PermissionMode::from_str("writeonly"),
+            PermissionMode::WriteOnly
+        );
+        assert_eq!(
+            PermissionMode::from_str("write_only"),
+            PermissionMode::WriteOnly
+        );
+        assert_eq!(PermissionMode::from_str("wo"), PermissionMode::WriteOnly);
+        assert_eq!(
+            PermissionMode::from_str("WriteOnly"),
+            PermissionMode::WriteOnly
+        );
+        assert_eq!(PermissionMode::from_str("WO"), PermissionMode::WriteOnly);
+
+        // Test invalid inputs default to ReadWrite
+        assert_eq!(
+            PermissionMode::from_str("invalid"),
+            PermissionMode::ReadWrite
+        );
+        assert_eq!(PermissionMode::from_str(""), PermissionMode::ReadWrite);
+        assert_eq!(PermissionMode::from_str("xyz"), PermissionMode::ReadWrite);
+    }
+
+    #[test]
+    fn test_permission_mode_as_str() {
+        assert_eq!(PermissionMode::ReadWrite.as_str(), "rw");
+        assert_eq!(PermissionMode::ReadOnly.as_str(), "ro");
+        assert_eq!(PermissionMode::WriteOnly.as_str(), "wo");
+    }
+
+    #[test]
+    fn test_permission_mode_default() {
+        assert_eq!(PermissionMode::default(), PermissionMode::ReadWrite);
+    }
+
+    #[test]
+    fn test_permission_mode_round_trip() {
+        let modes = [
+            PermissionMode::ReadWrite,
+            PermissionMode::ReadOnly,
+            PermissionMode::WriteOnly,
+        ];
+
+        for mode in modes {
+            let str_repr = mode.as_str();
+            let parsed = PermissionMode::from_str(str_repr);
+            assert_eq!(mode, parsed);
+        }
+    }
+
+    #[test]
+    fn test_default_user_info_with_name() {
+        let user_info = DefaultUserInfo::with_name("test_user");
+        assert_eq!(user_info.username(), "test_user");
+    }
+
+    #[test]
+    fn test_default_user_info_with_name_and_permission() {
+        let user_info =
+            DefaultUserInfo::with_name_and_permission("test_user", PermissionMode::ReadOnly);
+        assert_eq!(user_info.username(), "test_user");
+
+        // Cast to DefaultUserInfo to access permission_mode
+        let default_user = user_info
+            .as_any()
+            .downcast_ref::<DefaultUserInfo>()
+            .unwrap();
+        assert_eq!(default_user.permission_mode, PermissionMode::ReadOnly);
+    }
+
+    #[test]
+    fn test_user_info_as_any() {
+        let user_info = DefaultUserInfo::with_name("test_user");
+        let any_ref = user_info.as_any();
+        assert!(any_ref.downcast_ref::<DefaultUserInfo>().is_some());
+    }
+}

src/auth/src/user_provider.rs

@@ -29,7 +29,7 @@ use crate::error::{
     IllegalParamSnafu, InvalidConfigSnafu, IoSnafu, Result, UnsupportedPasswordTypeSnafu,
     UserNotFoundSnafu, UserPasswordMismatchSnafu,
 };
-use crate::user_info::DefaultUserInfo;
+use crate::user_info::{DefaultUserInfo, PermissionMode};
 use crate::{UserInfoRef, auth_mysql};
 
 #[async_trait::async_trait]
@@ -64,7 +64,11 @@ pub trait UserProvider: Send + Sync {
     }
 }
 
-fn load_credential_from_file(filepath: &str) -> Result<Option<HashMap<String, Vec<u8>>>> {
+/// Type alias for user info map
+/// Key is username, value is (password, permission_mode)
+pub type UserInfoMap = HashMap<String, (Vec<u8>, PermissionMode)>;
+
+fn load_credential_from_file(filepath: &str) -> Result<Option<UserInfoMap>> {
     // check valid path
     let path = Path::new(filepath);
     if !path.exists() {
@@ -83,13 +87,19 @@ fn load_credential_from_file(filepath: &str) -> Result<Option<HashMap<String, Ve
         .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
+            // The line format is:
+            // - `username=password` - Basic user with default permissions
+            // - `username:permission_mode=password` - User with specific permission mode
+            // - Lines starting with '#' are treated as comments and ignored
+            // - Empty lines are ignored
+            let line = line.trim();
+            if line.is_empty() || line.starts_with('#') {
+                return None;
             }
+
+            parse_credential_line(line)
         })
-        .collect::<HashMap<String, Vec<u8>>>();
+        .collect::<HashMap<String, _>>();
 
     ensure!(
         !credential.is_empty(),
@@ -102,8 +112,28 @@ fn load_credential_from_file(filepath: &str) -> Result<Option<HashMap<String, Ve
     Ok(Some(credential))
 }
 
+/// Parse a line of credential in the format of `username=password` or `username:permission_mode=password`
+pub(crate) fn parse_credential_line(line: &str) -> Option<(String, (Vec<u8>, PermissionMode))> {
+    let parts = line.split('=').collect::<Vec<&str>>();
+    if parts.len() != 2 {
+        return None;
+    }
+
+    let (username_part, password) = (parts[0], parts[1]);
+    let (username, permission_mode) = if let Some((user, perm)) = username_part.split_once(':') {
+        (user, PermissionMode::from_str(perm))
+    } else {
+        (username_part, PermissionMode::default())
+    };
+
+    Some((
+        username.to_string(),
+        (password.as_bytes().to_vec(), permission_mode),
+    ))
+}
+
 fn authenticate_with_credential(
-    users: &HashMap<String, Vec<u8>>,
+    users: &UserInfoMap,
     input_id: Identity<'_>,
     input_pwd: Password<'_>,
 ) -> Result<UserInfoRef> {
@@ -115,7 +145,7 @@ fn authenticate_with_credential(
                     msg: "blank username"
                 }
             );
-            let save_pwd = users.get(username).context(UserNotFoundSnafu {
+            let (save_pwd, permission_mode) = users.get(username).context(UserNotFoundSnafu {
                 username: username.to_string(),
             })?;
 
@@ -128,7 +158,10 @@ fn authenticate_with_credential(
                         }
                     );
                     if save_pwd == pwd.expose_secret().as_bytes() {
-                        Ok(DefaultUserInfo::with_name(username))
+                        Ok(DefaultUserInfo::with_name_and_permission(
+                            username,
+                            *permission_mode,
+                        ))
                     } else {
                         UserPasswordMismatchSnafu {
                             username: username.to_string(),
@@ -137,8 +170,9 @@ fn authenticate_with_credential(
                     }
                 }
                 Password::MysqlNativePassword(auth_data, salt) => {
-                    auth_mysql(auth_data, salt, username, save_pwd)
-                        .map(|_| DefaultUserInfo::with_name(username))
+                    auth_mysql(auth_data, salt, username, save_pwd).map(|_| {
+                        DefaultUserInfo::with_name_and_permission(username, *permission_mode)
+                    })
                 }
                 Password::PgMD5(_, _) => UnsupportedPasswordTypeSnafu {
                     password_type: "pg_md5",
@@ -148,3 +182,108 @@ fn authenticate_with_credential(
         }
     }
 }
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_credential_line() {
+        // Basic username=password format
+        let result = parse_credential_line("admin=password123");
+        assert_eq!(
+            result,
+            Some((
+                "admin".to_string(),
+                ("password123".as_bytes().to_vec(), PermissionMode::default())
+            ))
+        );
+
+        // Username with permission mode
+        let result = parse_credential_line("user:ReadOnly=secret");
+        assert_eq!(
+            result,
+            Some((
+                "user".to_string(),
+                ("secret".as_bytes().to_vec(), PermissionMode::ReadOnly)
+            ))
+        );
+        let result = parse_credential_line("user:ro=secret");
+        assert_eq!(
+            result,
+            Some((
+                "user".to_string(),
+                ("secret".as_bytes().to_vec(), PermissionMode::ReadOnly)
+            ))
+        );
+        // Username with WriteOnly permission mode
+        let result = parse_credential_line("writer:WriteOnly=mypass");
+        assert_eq!(
+            result,
+            Some((
+                "writer".to_string(),
+                ("mypass".as_bytes().to_vec(), PermissionMode::WriteOnly)
+            ))
+        );
+
+        // Username with 'wo' as WriteOnly permission shorthand
+        let result = parse_credential_line("writer:wo=mypass");
+        assert_eq!(
+            result,
+            Some((
+                "writer".to_string(),
+                ("mypass".as_bytes().to_vec(), PermissionMode::WriteOnly)
+            ))
+        );
+
+        // Username with complex password containing special characters
+        let result = parse_credential_line("admin:rw=p@ssw0rd!123");
+        assert_eq!(
+            result,
+            Some((
+                "admin".to_string(),
+                (
+                    "p@ssw0rd!123".as_bytes().to_vec(),
+                    PermissionMode::ReadWrite
+                )
+            ))
+        );
+
+        // Username with spaces should be preserved
+        let result = parse_credential_line("user name:WriteOnly=password");
+        assert_eq!(
+            result,
+            Some((
+                "user name".to_string(),
+                ("password".as_bytes().to_vec(), PermissionMode::WriteOnly)
+            ))
+        );
+
+        // Invalid format - no equals sign
+        let result = parse_credential_line("invalid_line");
+        assert_eq!(result, None);
+
+        // Invalid format - multiple equals signs
+        let result = parse_credential_line("user=pass=word");
+        assert_eq!(result, None);
+
+        // Empty password
+        let result = parse_credential_line("user=");
+        assert_eq!(
+            result,
+            Some((
+                "user".to_string(),
+                ("".as_bytes().to_vec(), PermissionMode::default())
+            ))
+        );
+
+        // Empty username
+        let result = parse_credential_line("=password");
+        assert_eq!(
+            result,
+            Some((
+                "".to_string(),
+                ("password".as_bytes().to_vec(), PermissionMode::default())
+            ))
+        );
+    }
+}

src/auth/src/user_provider/static_user_provider.rs

@@ -12,19 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
-
 use async_trait::async_trait;
 use snafu::{OptionExt, ResultExt};
 
 use crate::error::{FromUtf8Snafu, InvalidConfigSnafu, Result};
-use crate::user_provider::{authenticate_with_credential, load_credential_from_file};
+use crate::user_provider::{
+    UserInfoMap, authenticate_with_credential, load_credential_from_file, parse_credential_line,
+};
 use crate::{Identity, Password, UserInfoRef, UserProvider};
 
 pub(crate) const STATIC_USER_PROVIDER: &str = "static_user_provider";
 
 pub struct StaticUserProvider {
-    users: HashMap<String, Vec<u8>>,
+    users: UserInfoMap,
 }
 
 impl StaticUserProvider {
@@ -45,13 +45,12 @@ impl StaticUserProvider {
             "cmd" => content
                 .split(',')
                 .map(|kv| {
-                    let (k, v) = kv.split_once('=').context(InvalidConfigSnafu {
+                   parse_credential_line(kv).context(InvalidConfigSnafu {
                         value: kv.to_string(),
                         msg: "StaticUserProviderOption cmd values must be in format `user=pwd[,user=pwd]`",
-                    })?;
-                    Ok((k.to_string(), v.as_bytes().to_vec()))
+                    })
                 })
-                .collect::<Result<HashMap<String, Vec<u8>>>>()
+                .collect::<Result<UserInfoMap>>()
                 .map(|users| StaticUserProvider { users }),
             _ => InvalidConfigSnafu {
                 value: mode.to_string(),
@@ -69,7 +68,7 @@ impl StaticUserProvider {
             msg: "Expect at least one pair of username and password",
         })?;
         let username = kv.0;
-        let pwd = String::from_utf8(kv.1.clone()).context(FromUtf8Snafu)?;
+        let pwd = String::from_utf8(kv.1.0.clone()).context(FromUtf8Snafu)?;
         Ok((username.clone(), pwd))
     }
 }

src/auth/src/user_provider/watch_file_user_provider.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
 use std::path::Path;
 use std::sync::mpsc::channel;
 use std::sync::{Arc, Mutex};
@@ -24,12 +23,12 @@ use snafu::{ResultExt, ensure};
 
 use crate::error::{FileWatchSnafu, InvalidConfigSnafu, Result};
 use crate::user_info::DefaultUserInfo;
-use crate::user_provider::{authenticate_with_credential, load_credential_from_file};
+use crate::user_provider::{UserInfoMap, authenticate_with_credential, load_credential_from_file};
 use crate::{Identity, Password, UserInfoRef, UserProvider};
 
 pub(crate) const WATCH_FILE_USER_PROVIDER: &str = "watch_file_user_provider";
 
-type WatchedCredentialRef = Arc<Mutex<Option<HashMap<String, Vec<u8>>>>>;
+type WatchedCredentialRef = Arc<Mutex<Option<UserInfoMap>>>;
 
 /// A user provider that reads user credential from a file and watches the file for changes.
 ///

src/plugins/src/frontend.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use auth::UserProviderRef;
+use auth::{DefaultPermissionChecker, PermissionCheckerRef, UserProviderRef};
 use common_base::Plugins;
 use frontend::error::{IllegalAuthConfigSnafu, Result};
 use frontend::frontend::FrontendOptions;
@@ -29,6 +29,9 @@ pub async fn setup_frontend_plugins(
     if let Some(user_provider) = fe_opts.user_provider.as_ref() {
         let provider =
             auth::user_provider_from_option(user_provider).context(IllegalAuthConfigSnafu)?;
+        let permission_checker = DefaultPermissionChecker::arc();
+
+        plugins.insert::<PermissionCheckerRef>(permission_checker);
         plugins.insert::<UserProviderRef>(provider);
     }
     Ok(())

tests-integration/src/standalone.rs

@@ -109,7 +109,6 @@ impl GreptimeDbStandaloneBuilder {
         }
     }
 
-    #[cfg(test)]
     #[must_use]
     pub fn with_plugin(self, plugin: Plugins) -> Self {
         Self {

tests-integration/src/test_util.rs

@@ -17,9 +17,10 @@ use std::fmt::Display;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use auth::UserProviderRef;
+use auth::{DefaultPermissionChecker, PermissionCheckerRef, UserProviderRef};
 use axum::Router;
 use catalog::kvbackend::KvBackendCatalogManager;
+use common_base::Plugins;
 use common_config::Configurable;
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::schema_name::SchemaNameKey;
@@ -373,6 +374,18 @@ async fn setup_standalone_instance(
         .await
 }
 
+async fn setup_standalone_instance_with_plugins(
+    test_name: &str,
+    store_type: StorageType,
+    plugins: Plugins,
+) -> GreptimeDbStandalone {
+    GreptimeDbStandaloneBuilder::new(test_name)
+        .with_default_store_type(store_type)
+        .with_plugin(plugins)
+        .build()
+        .await
+}
+
 pub async fn setup_test_http_app(store_type: StorageType, name: &str) -> (Router, TestGuard) {
     let instance = setup_standalone_instance(name, store_type).await;
 
@@ -406,7 +419,13 @@ pub async fn setup_test_http_app_with_frontend_and_user_provider(
     name: &str,
     user_provider: Option<UserProviderRef>,
 ) -> (Router, TestGuard) {
-    let instance = setup_standalone_instance(name, store_type).await;
+    let plugins = Plugins::new();
+    if let Some(user_provider) = user_provider.clone() {
+        plugins.insert::<UserProviderRef>(user_provider.clone());
+        plugins.insert::<PermissionCheckerRef>(DefaultPermissionChecker::arc());
+    }
+
+    let instance = setup_standalone_instance_with_plugins(name, store_type, plugins).await;
 
     create_test_table(instance.fe_instance(), "demo").await;
 
@@ -587,7 +606,13 @@ pub async fn setup_mysql_server_with_user_provider(
     name: &str,
     user_provider: Option<UserProviderRef>,
 ) -> (TestGuard, Arc<Box<dyn Server>>) {
-    let instance = setup_standalone_instance(name, store_type).await;
+    let plugins = Plugins::new();
+    if let Some(user_provider) = user_provider.clone() {
+        plugins.insert::<UserProviderRef>(user_provider.clone());
+        plugins.insert::<PermissionCheckerRef>(DefaultPermissionChecker::arc());
+    }
+
+    let instance = setup_standalone_instance_with_plugins(name, store_type, plugins).await;
 
     let runtime = RuntimeBuilder::default()
         .worker_threads(2)

tests-integration/tests/http.rs

@@ -150,7 +150,7 @@ pub async fn test_http_auth(store_type: StorageType) {
     common_telemetry::init_default_ut_logging();
 
     let user_provider = user_provider_from_option(
-        &"static_user_provider:cmd:greptime_user=greptime_pwd".to_string(),
+        &"static_user_provider:cmd:greptime_user=greptime_pwd,readonly_user:ro=readonly_pwd,writeonly_user:wo=writeonly_pwd".to_string(),
     )
     .unwrap();
 
@@ -187,6 +187,65 @@ pub async fn test_http_auth(store_type: StorageType) {
         .send()
         .await;
     assert_eq!(res.status(), StatusCode::OK);
+
+    // 4. readonly user cannot write
+    let res = client
+        .get("/v1/sql?db=public&sql=show tables;")
+        .header(
+            "Authorization",
+            "basic cmVhZG9ubHlfdXNlcjpyZWFkb25seV9wd2Q=",
+        )
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let res = client
+        .get("/v1/sql?db=public&sql=create table auth_test(ts timestamp time index);")
+        .header(
+            "Authorization",
+            "basic cmVhZG9ubHlfdXNlcjpyZWFkb25seV9wd2Q=",
+        )
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::FORBIDDEN);
+
+    // 5. writeonly user cannot read
+    let res = client
+        .get("/v1/sql?db=public&sql=show tables;")
+        .header(
+            "Authorization",
+            "basic d3JpdGVvbmx5X3VzZXI6d3JpdGVvbmx5X3B3ZA==",
+        )
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::FORBIDDEN);
+    let res = client
+        .get("/v1/sql?db=public&sql=create table auth_test(ts timestamp time index);")
+        .header(
+            "Authorization",
+            "basic d3JpdGVvbmx5X3VzZXI6d3JpdGVvbmx5X3B3ZA==",
+        )
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let res = client
+        .get("/v1/sql?db=public&sql=insert into auth_test values(1);")
+        .header(
+            "Authorization",
+            "basic d3JpdGVvbmx5X3VzZXI6d3JpdGVvbmx5X3B3ZA==",
+        )
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let res = client
+        .get("/v1/sql?db=public&sql=select * from auth_test;")
+        .header(
+            "Authorization",
+            "basic d3JpdGVvbmx5X3VzZXI6d3JpdGVvbmx5X3B3ZA==",
+        )
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::FORBIDDEN);
+
     guard.remove_all().await;
 }
 

tests-integration/tests/sql.rs

@@ -88,7 +88,7 @@ macro_rules! sql_tests {
 
 pub async fn test_mysql_auth(store_type: StorageType) {
     let user_provider = user_provider_from_option(
-        &"static_user_provider:cmd:greptime_user=greptime_pwd".to_string(),
+        &"static_user_provider:cmd:greptime_user=greptime_pwd,readonly_user:ro=readonly_pwd,writeonly_user:wo=writeonly_pwd".to_string(),
     )
     .unwrap();
 
@@ -140,6 +140,46 @@ pub async fn test_mysql_auth(store_type: StorageType) {
 
     assert!(conn_re.is_ok());
 
+    // 4. readonly user
+    let conn_re = MySqlPoolOptions::new()
+        .max_connections(2)
+        .connect(&format!("mysql://readonly_user:readonly_pwd@{addr}/public"))
+        .await;
+    assert!(conn_re.is_ok());
+    let pool = conn_re.unwrap();
+    let _ = pool.execute("SELECT 1").await.unwrap();
+    let err = pool
+        .execute("CREATE TABLE test (ts timestamp time index)")
+        .await
+        .unwrap_err();
+    assert!(
+        err.to_string()
+            .contains("(PermissionDenied): User is not authorized to perform this action"),
+        "{}",
+        err.to_string()
+    );
+
+    // 5. writeonly user
+    let conn_re = MySqlPoolOptions::new()
+        .max_connections(2)
+        .connect(&format!(
+            "mysql://writeonly_user:writeonly_pwd@{addr}/public"
+        ))
+        .await;
+    assert!(conn_re.is_ok());
+    let pool = conn_re.unwrap();
+    let _ = pool
+        .execute("CREATE TABLE test (ts timestamp time index)")
+        .await
+        .unwrap();
+    let err = pool.execute("SHOW TABLES").await.unwrap_err();
+    assert!(
+        err.to_string()
+            .contains("(PermissionDenied): User is not authorized to perform this action"),
+        "{}",
+        err.to_string()
+    );
+
     let _ = fe_mysql_server.shutdown().await;
     guard.remove_all().await;
 }