feat(rest): add permission management APIs

crates/paimon/src/api/api_request.rs

@@ -22,6 +22,7 @@
 use serde::{Deserialize, Deserializer, Serialize};
 use std::collections::HashMap;
 
+use crate::api::management::{PermissionAccess, PermissionAssignment, PermissionResource};
 use crate::{
     catalog::{Function, FunctionDefinition, Identifier, ViewSchema},
     spec::{DataField, PartitionStatistics, Schema, SchemaChange},
@@ -311,6 +312,28 @@ impl AuthTableQueryRequest {
     }
 }
 
+/// Body of `POST {prefix}/permissions/revoke`: the assignment identity only. Revoking a
+/// `COLUMN` assignment removes its whole column range, so no `columns` travel here.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RevokePermissionRequest {
+    pub resource: PermissionResource,
+    pub access: String,
+    pub principal: String,
+}
+
+impl RevokePermissionRequest {
+    pub fn new(resource: PermissionResource, access: &str, principal: &str) -> crate::Result<Self> {
+        let access = PermissionAccess::canonicalize_for(resource.resource_type(), access)?;
+        PermissionAssignment::validate_principal(principal)?;
+        Ok(Self {
+            resource: resource.canonicalized()?,
+            access,
+            principal: principal.to_string(),
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -349,6 +372,36 @@ mod tests {
         assert_eq!(serde_json::to_string(&req).unwrap(), "{}");
     }
 
+    #[test]
+    fn test_revoke_permission_request_canonicalizes_access_and_carries_only_the_identity() {
+        let request = RevokePermissionRequest::new(
+            PermissionResource::table("sales", "orders"),
+            "select",
+            "analyst",
+        )
+        .unwrap();
+        assert_eq!(
+            serde_json::to_string(&request).unwrap(),
+            r#"{"resource":{"type":"TABLE","database":"sales","table":"orders"},"access":"SELECT","principal":"analyst"}"#
+        );
+        let blank_view: PermissionResource = serde_json::from_str(
+            r#"{"type":"TABLE","database":"sales","table":"orders","view":""}"#,
+        )
+        .unwrap();
+        let request = RevokePermissionRequest::new(blank_view, "select", "analyst").unwrap();
+        assert_eq!(
+            serde_json::to_string(&request).unwrap(),
+            r#"{"resource":{"type":"TABLE","database":"sales","table":"orders"},"access":"SELECT","principal":"analyst"}"#
+        );
+        let error =
+            RevokePermissionRequest::new(PermissionResource::catalog(), "select", "analyst")
+                .unwrap_err();
+        assert!(
+            error.to_string().contains("not valid for CATALOG"),
+            "{error}"
+        );
+    }
+
     #[test]
     fn test_create_partitions_request_serialization() {
         let req = CreatePartitionsRequest::new(

crates/paimon/src/api/api_response.rs

@@ -22,6 +22,7 @@
 use serde::{Deserialize, Deserializer, Serialize};
 use std::collections::HashMap;
 
+use crate::api::management::PermissionAssignment;
 use crate::catalog::{Function, FunctionDefinition, ViewSchema};
 use crate::spec::{DataField, Schema};
 
@@ -493,6 +494,25 @@ impl AuthTableQueryResponse {
     }
 }
 
+/// Response of `GET {prefix}/permissions`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListPermissionsResponse {
+    #[serde(default)]
+    pub permissions: Vec<PermissionAssignment>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_page_token: Option<String>,
+}
+
+impl ListPermissionsResponse {
+    pub fn new(permissions: Vec<PermissionAssignment>, next_page_token: Option<String>) -> Self {
+        Self {
+            permissions,
+            next_page_token,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -633,6 +653,24 @@ mod tests {
         assert!(blank.is_unrestricted());
     }
 
+    #[test]
+    fn test_list_permissions_response_deserialization() {
+        let response: ListPermissionsResponse = serde_json::from_str(
+            r#"{"permissions":[{"resource":{"type":"TABLE","database":"sales","table":"orders"},"access":"SELECT","principal":"analyst"}],"nextPageToken":"next"}"#,
+        )
+        .unwrap();
+        assert_eq!(response.permissions.len(), 1);
+        assert_eq!(response.permissions[0].principal(), "analyst");
+        assert_eq!(response.next_page_token.as_deref(), Some("next"));
+        let last: ListPermissionsResponse = serde_json::from_str("{}").unwrap();
+        assert!(last.permissions.is_empty());
+        assert_eq!(last.next_page_token, None);
+        assert_eq!(
+            serde_json::to_string(&ListPermissionsResponse::new(vec![], None)).unwrap(),
+            r#"{"permissions":[]}"#
+        );
+    }
+
     #[test]
     fn test_error_response_serialization() {
         let resp = ErrorResponse::new(