fix(query): account_admin should support grant ownership to new role after drop an created role (#14597)

* fix(query): after drop role account_admin should support grant ownership to new role

1. if has ownership but owner role not exists, the owner role will be
   set to account_admin

2. has_ownership should get ownership and role from meta

* modify api get_ownership

if has ownership but role not exists

this object's owner set to account_admin

* fix err

* refactor has_ownership

* fix build err

* find_object_owner driectly return role_name

* refactor find_object_owner, not get role again

* fix build err

* add table test and add some comment about why not modify get_ownerships

Author: TCeason

src/query/service/src/sessions/session_privilege_mgr.rs

@@ -269,13 +269,10 @@ impl SessionPrivilegeManager for SessionPrivilegeManagerImpl {
     async fn has_ownership(&self, object: &OwnershipObject) -> Result<bool> {
         let role_mgr = RoleCacheManager::instance();
         let tenant = self.session_ctx.get_current_tenant();
-
-        // if the object is not owned by any role, then considered as ACCOUNT_ADMIN, the normal users
-        // can not access it unless ACCOUNT_ADMIN grant relevant privileges to them.
-        let owner_role_name = match role_mgr.find_object_owner(&tenant, object).await? {
-            Some(owner_role) => owner_role.name,
-            None => BUILTIN_ROLE_ACCOUNT_ADMIN.to_string(),
-        };
+        let owner_role_name = role_mgr
+            .find_object_owner(&tenant, object)
+            .await?
+            .unwrap_or_else(|| BUILTIN_ROLE_ACCOUNT_ADMIN.to_string());
 
         let effective_roles = self.get_all_effective_roles().await?;
         let exists = effective_roles.iter().any(|r| r.name == owner_role_name);

src/query/users/src/role_cache_mgr.rs

@@ -116,15 +116,11 @@ impl RoleCacheManager {
         &self,
         tenant: &str,
         object: &OwnershipObject,
-    ) -> Result<Option<RoleInfo>> {
-        let owner = match self.user_manager.get_ownership(tenant, object).await? {
+    ) -> Result<Option<String>> {
+        match self.user_manager.get_ownership(tenant, object).await? {
             None => return Ok(None),
-            Some(owner) => owner,
-        };
-
-        // cache manager would not look into built-in roles.
-        let role = self.user_manager.get_role(tenant, owner.role).await?;
-        Ok(Some(role))
+            Some(owner) => Ok(Some(owner.role)),
+        }
     }
 
     // find_related_roles is called on validating an user's privileges.

src/query/users/src/role_mgr.rs

@@ -165,7 +165,24 @@ impl UserApiProvider {
             .get_ownership(object)
             .await
             .map_err(|e| e.add_message_back("(while get ownership)"))?;
-        Ok(ownership)
+        if let Some(owner) = ownership {
+            // if object has ownership, but the owner role is not exists, set owner role to ACCOUNT_ADMIN,
+            // only account_admin can access this object.
+            // Note: get_ownerships no need to do this check.
+            // Because this can cause system.table queries to slow down
+            // The intention is that the account admin can grant ownership.
+            // So system.tables will display dropped role. It's by design.
+            if !self.exists_role(tenant, owner.role.clone()).await? {
+                Ok(Some(OwnershipInfo {
+                    role: BUILTIN_ROLE_ACCOUNT_ADMIN.to_string(),
+                    object: object.clone(),
+                }))
+            } else {
+                Ok(Some(owner))
+            }
+        } else {
+            Ok(None)
+        }
     }
 
     #[async_backtrace::framed]

tests/suites/0_stateless/18_rbac/18_0002_ownership_cover.result

@@ -37,3 +37,12 @@ t1
 Error: APIError: ResponseError with 1063: Permission denied: User 'a'@'%' does not have the required privileges for database 'db_a'
 t
 t1
+=== fix_issue_14572: test drop role; grant ownership ===
+a	drop_role
+t	drop_role
+a	account_admin
+t	drop_role
+a	drop_role1
+t	drop_role1
+GRANT OWNERSHIP ON 'default'.'a'.* TO ROLE `drop_role1`
+GRANT OWNERSHIP ON 'default'.'a'.'t' TO ROLE `drop_role1`

tests/suites/0_stateless/18_rbac/18_0002_ownership_cover.sh

@@ -126,3 +126,31 @@ echo "drop database db_a;" | $BENDSQL_CLIENT_CONNECT
 echo "drop role if exists role1;" | $BENDSQL_CLIENT_CONNECT
 echo "drop user if exists a;" | $BENDSQL_CLIENT_CONNECT
 echo "drop user if exists b;" | $BENDSQL_CLIENT_CONNECT
+
+echo "=== fix_issue_14572: test drop role; grant ownership ==="
+echo "drop role if exists drop_role;" | $BENDSQL_CLIENT_CONNECT
+echo "drop role if exists drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "drop user if exists u1;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists a;" | $BENDSQL_CLIENT_CONNECT
+echo "create role drop_role;" | $BENDSQL_CLIENT_CONNECT
+echo "create role drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "create user u1 identified by '123' with DEFAULT_ROLE='drop_role'" | $BENDSQL_CLIENT_CONNECT
+echo "grant role drop_role to u1;" | $BENDSQL_CLIENT_CONNECT
+echo "grant create on *.* to u1;" | $BENDSQL_CLIENT_CONNECT
+export USER_U1_CONNECT="bendsql --user=u1 --password=123 --host=${QUERY_MYSQL_HANDLER_HOST} --port ${QUERY_HTTP_HANDLER_PORT}"
+
+echo "create database a" | $USER_U1_CONNECT
+echo "create table a.t(id int)" | $USER_U1_CONNECT
+echo "select name, owner from system.databases where name='a'" | $USER_U1_CONNECT
+echo "select name, owner from system.tables where database='a'" | $USER_U1_CONNECT
+echo "drop role drop_role;" | $BENDSQL_CLIENT_CONNECT
+echo "select name, owner from system.databases where name='a'" | $BENDSQL_CLIENT_CONNECT
+echo "select name, owner from system.tables where database='a'" | $USER_U1_CONNECT
+echo "grant ownership on a.* to role drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "grant ownership on a.t to role drop_role1;" | $BENDSQL_CLIENT_CONNECT
+echo "select name, owner from system.databases where name='a'" | $BENDSQL_CLIENT_CONNECT
+echo "select name, owner from system.tables where database='a'" | $USER_U1_CONNECT
+echo "show grants for role drop_role1" | $BENDSQL_CLIENT_CONNECT
+echo "drop role drop_role1" | $BENDSQL_CLIENT_CONNECT
+echo "drop user u1" | $BENDSQL_CLIENT_CONNECT
+echo "drop database a" | $BENDSQL_CLIENT_CONNECT