fix: add privilege check for drop user

src/frontend/src/handler/drop_database.rs

@@ -30,9 +30,10 @@ pub async fn handle_drop_database(
     let catalog_reader = session.env().catalog_reader();
     let database_name = Binder::resolve_database_name(database_name)?;
     if session.database() == database_name {
-        return Err(
-            ErrorCode::InternalError("cannot drop the currently open database".to_owned()).into(),
-        );
+        return Err(ErrorCode::PermissionDenied(
+            "cannot drop the currently open database".to_owned(),
+        )
+        .into());
     }
     if mode.is_some() {
         return Err(ErrorCode::BindError("Drop database not support drop mode".to_owned()).into());

src/frontend/src/handler/drop_user.rs

@@ -34,12 +34,42 @@ pub async fn handle_drop_user(
 
     let user_name = Binder::resolve_user_name(user_name)?;
     let user_info_reader = session.env().user_info_reader();
-    let user_id = user_info_reader
+    let user_info = user_info_reader
         .read_guard()
         .get_user_by_name(&user_name)
-        .map(|u| u.id);
-    match user_id {
-        Some(user_id) => {
+        .map(|u| (u.id, u.is_super));
+    match user_info {
+        Some((user_id, is_super)) => {
+            if session.user_id() == user_id {
+                return Err(ErrorCode::PermissionDenied(
+                    "current user cannot be dropped".to_owned(),
+                )
+                .into());
+            }
+            if let Some(current_user) = user_info_reader
+                .read_guard()
+                .get_user_by_name(&session.user_name())
+            {
+                if !current_user.is_super {
+                    if is_super {
+                        return Err(ErrorCode::PermissionDenied(
+                            "must be superuser to drop superusers".to_owned(),
+                        )
+                        .into());
+                    }
+                    if !current_user.can_create_user {
+                        return Err(ErrorCode::PermissionDenied(
+                            "permission denied to drop user".to_owned(),
+                        )
+                        .into());
+                    }
+                }
+            } else {
+                return Err(
+                    ErrorCode::PermissionDenied("Session user is invalid".to_owned()).into(),
+                );
+            }
+
             let user_info_writer = session.user_info_writer()?;
             user_info_writer.drop_user(user_id).await?;
         }