feat(table-catalog): refine table catalog permissions (#3283)

* feat(table-catalog): refine table catalog permissions

* fix(policy): scope table admin resources

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: cxymds <Cxymds@qq.com>

crates/policy/src/policy/action.rs

@@ -563,6 +563,31 @@ pub enum AdminAction {
 }
 
 impl AdminAction {
+    pub(crate) fn is_table_resource_scoped(&self) -> bool {
+        matches!(
+            self,
+            AdminAction::GetTableBucketAction
+                | AdminAction::SetTableBucketAction
+                | AdminAction::GetTableNamespaceAction
+                | AdminAction::SetTableNamespaceAction
+                | AdminAction::UpdateTableNamespacePropertiesAction
+                | AdminAction::DeleteTableNamespaceAction
+                | AdminAction::GetTableAction
+                | AdminAction::SetTableAction
+                | AdminAction::CreateTableAction
+                | AdminAction::RegisterTableAction
+                | AdminAction::CommitTableAction
+                | AdminAction::DeleteTableAction
+                | AdminAction::GetTableLifecycleAction
+                | AdminAction::SetTableLifecycleAction
+                | AdminAction::RunTableMaintenanceAction
+                | AdminAction::GetTableMetadataLocationAction
+                | AdminAction::SetTableMetadataLocationAction
+                | AdminAction::GetTableMetadataAction
+                | AdminAction::SetTableMetadataAction
+        )
+    }
+
     // IsValid - checks if action is valid or not.
     pub fn is_valid(&self) -> bool {
         matches!(

crates/policy/src/policy/policy.rs

@@ -1530,6 +1530,161 @@ mod test {
         );
     }
 
+    #[tokio::test]
+    async fn test_table_admin_action_with_resource_is_limited_to_bucket() -> Result<()> {
+        use crate::policy::action::{Action, AdminAction};
+
+        let data = r#"
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["admin:GetTableMetadata"],
+      "Resource": ["arn:aws:s3:::warehouse-a"]
+    }
+  ]
+}
+"#;
+
+        let policy = Policy::parse_config(data.as_bytes())?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let groups = None;
+
+        let matching_args = Args {
+            account: "testuser",
+            groups: &groups,
+            action: Action::AdminAction(AdminAction::GetTableMetadataAction),
+            bucket: "warehouse-a",
+            conditions: &conditions,
+            is_owner: false,
+            object: "",
+            claims: &claims,
+            deny_only: false,
+        };
+        assert!(
+            policy.is_allowed(&matching_args).await,
+            "table admin action should allow the explicitly granted warehouse bucket"
+        );
+
+        let mismatched_args = Args {
+            account: "testuser",
+            groups: &groups,
+            action: Action::AdminAction(AdminAction::GetTableMetadataAction),
+            bucket: "warehouse-b",
+            conditions: &conditions,
+            is_owner: false,
+            object: "",
+            claims: &claims,
+            deny_only: false,
+        };
+        assert!(
+            !policy.is_allowed(&mismatched_args).await,
+            "table admin action must not ignore Resource when the request targets a different warehouse bucket"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_table_admin_action_with_not_resource_excludes_bucket() -> Result<()> {
+        use crate::policy::action::{Action, AdminAction};
+
+        let data = r#"
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["admin:GetTableMetadata"],
+      "NotResource": ["arn:aws:s3:::warehouse-b"]
+    }
+  ]
+}
+"#;
+
+        let policy = Policy::parse_config(data.as_bytes())?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let groups = None;
+
+        let allowed_args = Args {
+            account: "testuser",
+            groups: &groups,
+            action: Action::AdminAction(AdminAction::GetTableMetadataAction),
+            bucket: "warehouse-a",
+            conditions: &conditions,
+            is_owner: false,
+            object: "",
+            claims: &claims,
+            deny_only: false,
+        };
+        assert!(
+            policy.is_allowed(&allowed_args).await,
+            "table admin NotResource should allow a warehouse outside the excluded bucket"
+        );
+
+        let excluded_args = Args {
+            account: "testuser",
+            groups: &groups,
+            action: Action::AdminAction(AdminAction::GetTableMetadataAction),
+            bucket: "warehouse-b",
+            conditions: &conditions,
+            is_owner: false,
+            object: "",
+            claims: &claims,
+            deny_only: false,
+        };
+        assert!(
+            !policy.is_allowed(&excluded_args).await,
+            "table admin NotResource should deny the excluded warehouse bucket"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_non_table_admin_action_keeps_unscoped_resource_behavior() -> Result<()> {
+        use crate::policy::action::{Action, AdminAction};
+
+        let data = r#"
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["admin:ServerInfo"],
+      "Resource": ["arn:aws:s3:::warehouse-a"]
+    }
+  ]
+}
+"#;
+
+        let policy = Policy::parse_config(data.as_bytes())?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let groups = None;
+
+        let args = Args {
+            account: "testuser",
+            groups: &groups,
+            action: Action::AdminAction(AdminAction::ServerInfoAdminAction),
+            bucket: "warehouse-b",
+            conditions: &conditions,
+            is_owner: false,
+            object: "",
+            claims: &claims,
+            deny_only: false,
+        };
+        assert!(
+            policy.is_allowed(&args).await,
+            "existing non-table admin actions should preserve resource-independent evaluation"
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_sts_statement_without_resource_is_valid() {
         let data = r#"

crates/policy/src/policy/statement.rs

@@ -89,6 +89,18 @@ enum ActionFamily {
 }
 
 impl Statement {
+    fn skips_resource_match_for_args(&self, args: &Args<'_>) -> bool {
+        if self.is_sts() {
+            return true;
+        }
+
+        if !self.is_admin() {
+            return false;
+        }
+
+        !matches!(args.action, Action::AdminAction(action) if action.is_table_resource_scoped())
+    }
+
     fn is_kms(&self) -> bool {
         for act in self.actions.iter() {
             if matches!(act, Action::KmsAction(_)) {
@@ -188,8 +200,7 @@ impl Statement {
                 .resources
                 .is_match_with_resolver(&resource, args.conditions, Some(resolver))
                 .await
-            && !self.is_admin()
-            && !self.is_sts()
+            && !self.skips_resource_match_for_args(args)
         {
             return false;
         }
@@ -199,8 +210,7 @@ impl Statement {
                 .not_resources
                 .is_match_with_resolver(&resource, args.conditions, Some(resolver))
                 .await
-            && !self.is_admin()
-            && !self.is_sts()
+            && !self.skips_resource_match_for_args(args)
         {
             return false;
         }