fix: close database ACL gaps in permission checks (#8492)

* fix: database ACL

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: reduce code

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: cr issue

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

src/auth/src/permission.rs

@@ -36,7 +36,11 @@ pub enum PermissionReq<'a> {
     PromStoreRead,
     Otlp,
     LogWrite,
-    BulkInsert,
+    BulkInsert {
+        catalog: &'a str,
+        schema: &'a str,
+        table: &'a str,
+    },
 }
 
 impl<'a> PermissionReq<'a> {
@@ -57,7 +61,7 @@ impl<'a> PermissionReq<'a> {
             | PermissionReq::PromStoreWrite
             | PermissionReq::Otlp
             | PermissionReq::LogWrite
-            | PermissionReq::BulkInsert => false,
+            | PermissionReq::BulkInsert { .. } => false,
         }
     }
 
@@ -79,6 +83,15 @@ pub trait PermissionChecker: Send + Sync {
         user_info: UserInfoRef,
         req: PermissionReq,
     ) -> Result<PermissionResp>;
+
+    fn check_permission_with_context(
+        &self,
+        user_info: UserInfoRef,
+        req: PermissionReq,
+        _current_schema: Option<&str>,
+    ) -> Result<PermissionResp> {
+        self.check_permission(user_info, req)
+    }
 }
 
 impl PermissionChecker for Option<&PermissionCheckerRef> {
@@ -86,13 +99,24 @@ impl PermissionChecker for Option<&PermissionCheckerRef> {
         &self,
         user_info: UserInfoRef,
         req: PermissionReq,
+    ) -> Result<PermissionResp> {
+        self.check_permission_with_context(user_info, req, None)
+    }
+
+    fn check_permission_with_context(
+        &self,
+        user_info: UserInfoRef,
+        req: PermissionReq,
+        current_schema: Option<&str>,
     ) -> Result<PermissionResp> {
         match self {
-            Some(checker) => match checker.check_permission(user_info, req) {
-                Ok(PermissionResp::Reject) => PermissionDeniedSnafu.fail(),
-                Ok(PermissionResp::Allow) => Ok(PermissionResp::Allow),
-                Err(e) => Err(e),
-            },
+            Some(checker) => {
+                match checker.check_permission_with_context(user_info, req, current_schema) {
+                    Ok(PermissionResp::Reject) => PermissionDeniedSnafu.fail(),
+                    Ok(PermissionResp::Allow) => Ok(PermissionResp::Allow),
+                    Err(e) => Err(e),
+                }
+            }
             None => Ok(PermissionResp::Allow),
         }
     }
@@ -209,4 +233,15 @@ mod tests {
 
         assert!(req.is_write());
     }
+
+    #[test]
+    fn test_bulk_insert_is_write_request() {
+        let req = PermissionReq::BulkInsert {
+            catalog: "greptime",
+            schema: "public",
+            table: "metrics",
+        };
+
+        assert!(req.is_write());
+    }
 }