fix(query): use db no need to check privileges just need check grant …

…objects (#12092)

* fix(query): use db no need to check privileges just need check grant objects

* optimize code

src/query/service/src/interpreters/access/privilege_access.rs

@@ -15,11 +15,13 @@
 use std::sync::Arc;
 
 use common_catalog::table_context::TableContext;
+use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_app::principal::GrantObject;
 use common_meta_app::principal::UserPrivilegeType;
 use common_sql::plans::CopyPlan;
 use common_sql::plans::RewriteKind;
+use common_users::RoleCacheManager;
 
 use crate::interpreters::access::AccessChecker;
 use crate::sessions::QueryContext;
@@ -100,13 +102,37 @@ impl AccessChecker for PrivilegeAccess {
                     .await?;
             }
             Plan::UseDatabase(plan) => {
-                let catalog = self.ctx.get_current_catalog();
-                session
-                    .validate_privilege(
-                        &GrantObject::Database(catalog, plan.database.clone()),
-                        vec![UserPrivilegeType::Select],
-                    )
+                // Use db is special. Should not check the privilege.
+                // Just need to check user grant objects contain the db that be used.
+                let database = &plan.database;
+                let user = self.ctx.get_current_user()?;
+                let (identity, grant_set) = (user.identity().to_string(), user.grants);
+                let can_use = RoleCacheManager::instance()
+                    .find_related_roles(&self.ctx.get_tenant(), &grant_set.roles())
                     .await?
+                    .into_iter()
+                    .map(|role| role.grants)
+                    .fold(grant_set, |a, b| a | b)
+                    .entries()
+                    .iter()
+                    .any(|e| {
+                        let object = e.object();
+                        match object {
+                            GrantObject::Global => true,
+                            GrantObject::Database(_, ldb) | GrantObject::Table(_, ldb, _) => {
+                                ldb == database
+                            }
+                        }
+                    });
+
+                return if can_use {
+                    Ok(())
+                } else {
+                    Err(ErrorCode::PermissionDenied(format!(
+                        "Permission denied, user {} don't have privilege for database {}",
+                        identity, database
+                    )))
+                };
             }
 
             // Virtual Column.

src/query/sql/src/planner/binder/ddl/column.rs

@@ -142,54 +142,47 @@ pub(crate) async fn generate_unique_object(
     grant_set: UserGrantSet,
 ) -> Result<(HashSet<String>, bool)> {
     let mut unique_object: HashSet<String> = HashSet::new();
-    let mut has_object_priv = false;
-    let _objects = RoleCacheManager::instance()
+    let has_object_priv = RoleCacheManager::instance()
         .find_related_roles(tenant, &grant_set.roles())
         .await?
         .into_iter()
         .map(|role| role.grants)
         .fold(grant_set, |a, b| a | b)
         .entries()
         .iter()
-        .map(|e| {
+        .any(|e| {
             let object = e.object();
             match object {
-                GrantObject::Global => {
-                    has_object_priv = true;
-                }
+                GrantObject::Global => true,
                 GrantObject::Database(_, ldb) => {
                     if let Some(database) = &database {
                         // show columns from table from db
                         // show tables from db
-                        if ldb == database {
-                            has_object_priv = true;
-                        }
+                        ldb == database
                     } else {
-                        unique_object.insert(format!("'{ldb}'"));
+                        // show databases
+                        unique_object.insert(format!("'{}'", ldb));
+                        false
                     }
                 }
                 GrantObject::Table(_, ldb, ltab) => match (&database, &table) {
                     // show columns from tab from db;
-                    (Some(database), Some(table)) => {
-                        if ldb == database && ltab == table {
-                            has_object_priv = true;
-                        }
-                    }
+                    (Some(database), Some(table)) => ldb == database && ltab == table,
                     // show tables from db;
                     (Some(database), None) => {
                         if ldb == database {
-                            unique_object.insert(format!("'{ltab}'"));
+                            unique_object.insert(format!("'{}'", ltab));
                         }
+                        false
                     }
                     // show databases
                     (None, None) => {
-                        unique_object.insert(format!("'{ldb}'"));
+                        unique_object.insert(format!("'{}'", ldb));
+                        false
                     }
                     _ => unreachable!(),
                 },
             }
-        })
-        .collect::<Vec<_>>();
-
+        });
     Ok((unique_object, has_object_priv))
 }

src/query/sql/src/planner/binder/ddl/database.rs

@@ -30,6 +30,7 @@ use common_expression::DataField;
 use common_expression::DataSchemaRefExt;
 use common_meta_app::schema::DatabaseMeta;
 use common_meta_app::share::ShareNameIdent;
+use itertools::Itertools;
 use tracing::debug;
 
 use crate::binder::ddl::column::generate_unique_object;
@@ -69,15 +70,7 @@ impl Binder {
         let mut select_builder = if has_object_priv {
             SelectBuilder::from("system.databases")
         } else {
-            let mut in_list = "".to_string();
-            let last = unique_dbs.len() - 1;
-            for (i, db) in unique_dbs.iter().enumerate() {
-                if i == last {
-                    in_list += db;
-                    break;
-                }
-                in_list = in_list + db + ",";
-            }
+            let in_list = unique_dbs.iter().join(",");
             SelectBuilder::from(&format!(
                 "(select * from system.databases where name in ({in_list}))"
             ))

src/query/sql/src/planner/binder/ddl/table.rs

@@ -69,6 +69,7 @@ use common_meta_app::storage::StorageParams;
 use common_storage::DataOperator;
 use common_storages_view::view_table::QUERY;
 use common_storages_view::view_table::VIEW_ENGINE;
+use itertools::Itertools;
 use storages_common_table_meta::table::is_reserved_opt_key;
 use storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
 use storages_common_table_meta::table::OPT_KEY_STORAGE_FORMAT;
@@ -160,15 +161,7 @@ impl Binder {
         let mut select_builder = if has_object_priv {
             SelectBuilder::from(target_sys_tab)
         } else {
-            let mut in_list = "".to_string();
-            let last = unique_tables.len() - 1;
-            for (i, tables) in unique_tables.iter().enumerate() {
-                if i == last {
-                    in_list += tables;
-                    break;
-                }
-                in_list = in_list + tables + ",";
-            }
+            let in_list = unique_tables.iter().join(",");
             need_filter = false;
             // Need filter database = 'db', ensure will execute optimizer find_eq_filter
             SelectBuilder::from(&format!(

tests/suites/0_stateless/20+_others/20_0012_privilege_access.result

@@ -30,11 +30,15 @@ ERROR 1105 (HY000) at line 1: PermissionDenied. Code: 1063, Text = Permission de
 ERROR 1105 (HY000) at line 1: PermissionDenied. Code: 1063, Text = Permission denied, privilege [Select] is required on 'default'.'system'.'fuse_block' for user 'test-user'@'%' with role public.
 1
 GRANT SELECT ON 'default'.'default'.* TO 'a'@'%'
+GRANT SELECT ON 'default'.'grant_db'.'t' TO 'a'@'%'
 GRANT SELECT ON 'default'.'system'.'one' TO 'a'@'%'
 default
+grant_db
 system
 test_t
 one
+t
 dummy	TINYINT UNSIGNED	NO		NULL	NULL
+c1	INT	NO		NULL	NULL
 ERROR 1105 (HY000) at line 1: PermissionDenied. Code: 1063, Text = Permission denied, user 'a'@'%' don't have privilege for database information_schema.
 ERROR 1105 (HY000) at line 1: PermissionDenied. Code: 1063, Text = Permission denied, user 'a'@'%' don't have privilege for table system.tables.

tests/suites/0_stateless/20+_others/20_0012_privilege_access.sh

@@ -127,15 +127,23 @@ echo -e "[mysql]\nhost=${QUERY_MYSQL_HANDLER_HOST}\nuser=a\npassword=${TEST_USER
 echo "drop user if exists a" |  $MYSQL_CLIENT_CONNECT
 echo "create user a identified by '$TEST_USER_PASSWORD'" |  $MYSQL_CLIENT_CONNECT
 echo "drop database if exists nogrant" |  $MYSQL_CLIENT_CONNECT
+echo "drop database if exists grant_db" |  $MYSQL_CLIENT_CONNECT
+echo "create database grant_db" |  $MYSQL_CLIENT_CONNECT
+echo "create table grant_db.t(c1 int)" |  $MYSQL_CLIENT_CONNECT
 echo "create database nogrant" |  $MYSQL_CLIENT_CONNECT
 echo "grant select on default.* to a" |  $MYSQL_CLIENT_CONNECT
+echo "grant select on grant_db.t to a" |  $MYSQL_CLIENT_CONNECT
 echo "drop table if exists default.test_t" |  $MYSQL_CLIENT_CONNECT
 echo "create table default.test_t(id int)" |  $MYSQL_CLIENT_CONNECT
 echo "show grants for a" |  $MYSQL_CLIENT_CONNECT
 echo "show databases" | $USER_A_CONNECT
 echo "show tables" | $USER_A_CONNECT
 echo "show tables from system" | $USER_A_CONNECT
+echo "show tables from grant_db" | $USER_A_CONNECT
+echo "use system" | $USER_A_CONNECT
+echo "use grant_db" | $USER_A_CONNECT
 echo "show columns from one from system" | $USER_A_CONNECT
+echo "show columns from t from grant_db" | $USER_A_CONNECT
 
 ### will return err
 echo "show tables from information_schema" | $USER_A_CONNECT
@@ -144,6 +152,7 @@ echo "show columns from tables from system" | $USER_A_CONNECT
 ## Drop user
 echo "drop user a" | $MYSQL_CLIENT_CONNECT
 echo "drop database nogrant" |  $MYSQL_CLIENT_CONNECT
+echo "drop database grant_db" |  $MYSQL_CLIENT_CONNECT
 rm -rf password.out