This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 6311a012 feat(rest): add permission management APIs (#820)
6311a012 is described below
commit 6311a012b6436ff91cfc6730bfa8f2a5974bc694
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Sep 14 18:06:42 2026 +0800
feat(rest): add permission management APIs (#820)
---
crates/paimon/src/api/api_request.rs | 53 +
crates/paimon/src/api/api_response.rs | 38 +
crates/paimon/src/api/management.rs | 1298 ++++++++++++++++++++++++
crates/paimon/src/api/mod.rs | 13 +-
crates/paimon/src/api/resource_paths.rs | 24 +
crates/paimon/src/api/rest_api.rs | 48 +-
crates/paimon/src/catalog/rest/rest_catalog.rs | 28 +
crates/paimon/tests/mock_server.rs | 176 +++-
crates/paimon/tests/rest_api_test.rs | 265 ++++-
crates/paimon/tests/rest_catalog_test.rs | 28 +
10 files changed, 1953 insertions(+), 18 deletions(-)
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index 48646757..8c721fdf 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/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(
diff --git a/crates/paimon/src/api/api_response.rs
b/crates/paimon/src/api/api_response.rs
index 9d68cf3d..ce62a392 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/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(
diff --git a/crates/paimon/src/api/management.rs
b/crates/paimon/src/api/management.rs
new file mode 100644
index 00000000..188ba370
--- /dev/null
+++ b/crates/paimon/src/api/management.rs
@@ -0,0 +1,1298 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! REST management API model: permission assignments and (later) data
policies.
+//!
+//! Mirrors Java `org.apache.paimon.management`, whose request-side
constructors store a
+//! corrected value rather than only checking it; `Deserialize` does neither,
so a server may
+//! list values a client could not have sent. The send paths therefore run
`canonicalized` on
+//! the types whose Java constructor corrects, `validate` on the types whose
Java constructor
+//! only checks, and send what comes back.
+
+use std::fmt;
+use std::str::FromStr;
+
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+use crate::api::rest_error::RestError;
+use crate::{Error, Result};
+
+/// Client-side validation failures map to this: a real server answers 400 for
the same input.
+pub(crate) fn bad_request(message: impl Into<String>) -> Error {
+ Error::RestApi {
+ source: RestError::BadRequest {
+ message: message.into(),
+ },
+ }
+}
+
+/// Java's `String.trim()` strips only characters `<= U+0020`, so a value is
blank exactly when
+/// every character is one of those; `str::trim` would also strip the rest of
Unicode whitespace.
+pub(crate) fn is_blank(value: &str) -> bool {
+ value.chars().all(|ch| ch <= ' ')
+}
+
+/// Java's `String.length()` counts UTF-16 code units, and the wire limits are
written against it.
+pub(crate) fn utf16_len(value: &str) -> usize {
+ value.encode_utf16().count()
+}
+
+fn blank_to_none(value: Option<String>) -> Option<String> {
+ value.filter(|value| !is_blank(value))
+}
+
+/// The kind of object a permission is attached to (Java `ResourceType`).
+///
+/// `CATALOG_ALL` / `DATABASE_ALL` scope over a catalog's or database's
descendants, now and
+/// later; every other type names one exact object.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum ResourceType {
+ Catalog,
+ CatalogAll,
+ Database,
+ DatabaseAll,
+ Table,
+ Column,
+ View,
+ Function,
+}
+
+impl ResourceType {
+ pub const VALUES: [ResourceType; 8] = [
+ ResourceType::Catalog,
+ ResourceType::CatalogAll,
+ ResourceType::Database,
+ ResourceType::DatabaseAll,
+ ResourceType::Table,
+ ResourceType::Column,
+ ResourceType::View,
+ ResourceType::Function,
+ ];
+
+ pub fn as_str(&self) -> &'static str {
+ match self {
+ ResourceType::Catalog => "CATALOG",
+ ResourceType::CatalogAll => "CATALOG_ALL",
+ ResourceType::Database => "DATABASE",
+ ResourceType::DatabaseAll => "DATABASE_ALL",
+ ResourceType::Table => "TABLE",
+ ResourceType::Column => "COLUMN",
+ ResourceType::View => "VIEW",
+ ResourceType::Function => "FUNCTION",
+ }
+ }
+}
+
+impl fmt::Display for ResourceType {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.write_str(self.as_str())
+ }
+}
+
+impl FromStr for ResourceType {
+ type Err = Error;
+
+ /// Case-insensitive, like Java `ResourceType.fromString`.
+ fn from_str(value: &str) -> Result<Self> {
+ let upper = value.to_uppercase();
+ Self::VALUES
+ .into_iter()
+ .find(|resource_type| resource_type.as_str() == upper)
+ .ok_or_else(|| bad_request(format!("Unknown resource type
'{value}'.")))
+ }
+}
+
+impl Serialize for ResourceType {
+ fn serialize<S: Serializer>(&self, serializer: S) ->
std::result::Result<S::Ok, S::Error> {
+ serializer.serialize_str(self.as_str())
+ }
+}
+
+impl<'de> Deserialize<'de> for ResourceType {
+ fn deserialize<D: Deserializer<'de>>(deserializer: D) ->
std::result::Result<Self, D::Error> {
+ String::deserialize(deserializer)?
+ .parse()
+ .map_err(serde::de::Error::custom)
+ }
+}
+
+/// A permission target: a resource type and the locators it needs (Java
`PermissionResource`).
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PermissionResource {
+ #[serde(rename = "type")]
+ resource_type: ResourceType,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ database: Option<String>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ table: Option<String>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ function: Option<String>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ view: Option<String>,
+}
+
+impl PermissionResource {
+ pub fn catalog() -> Self {
+ Self::unchecked(ResourceType::Catalog, None, None, None, None)
+ }
+
+ pub fn catalog_all() -> Self {
+ Self::unchecked(ResourceType::CatalogAll, None, None, None, None)
+ }
+
+ pub fn database(database: impl Into<String>) -> Self {
+ Self::unchecked(
+ ResourceType::Database,
+ Some(database.into()),
+ None,
+ None,
+ None,
+ )
+ }
+
+ pub fn database_all(database: impl Into<String>) -> Self {
+ Self::unchecked(
+ ResourceType::DatabaseAll,
+ Some(database.into()),
+ None,
+ None,
+ None,
+ )
+ }
+
+ pub fn table(database: impl Into<String>, table: impl Into<String>) ->
Self {
+ Self::unchecked(
+ ResourceType::Table,
+ Some(database.into()),
+ Some(table.into()),
+ None,
+ None,
+ )
+ }
+
+ /// The columns of one table; the column range travels in the assignment.
+ pub fn column(database: impl Into<String>, table: impl Into<String>) ->
Self {
+ Self::unchecked(
+ ResourceType::Column,
+ Some(database.into()),
+ Some(table.into()),
+ None,
+ None,
+ )
+ }
+
+ pub fn function(database: impl Into<String>, function: impl Into<String>)
-> Self {
+ Self::unchecked(
+ ResourceType::Function,
+ Some(database.into()),
+ None,
+ Some(function.into()),
+ None,
+ )
+ }
+
+ pub fn view(database: impl Into<String>, view: impl Into<String>) -> Self {
+ Self::unchecked(
+ ResourceType::View,
+ Some(database.into()),
+ None,
+ None,
+ Some(view.into()),
+ )
+ }
+
+ pub fn new(
+ resource_type: ResourceType,
+ database: Option<&str>,
+ table: Option<&str>,
+ function: Option<&str>,
+ view: Option<&str>,
+ ) -> Result<Self> {
+ Self::unchecked(
+ resource_type,
+ database.map(str::to_string),
+ table.map(str::to_string),
+ function.map(str::to_string),
+ view.map(str::to_string),
+ )
+ .canonicalized()
+ }
+
+ /// Blank locators count as absent (Java `blankToNull`); what remains must
be exactly the
+ /// locators the type needs.
+ pub fn canonicalized(mut self) -> Result<Self> {
+ self.database = blank_to_none(self.database);
+ self.table = blank_to_none(self.table);
+ self.function = blank_to_none(self.function);
+ self.view = blank_to_none(self.view);
+ self.check_locators()?;
+ Ok(self)
+ }
+
+ fn check_locators(&self) -> Result<()> {
+ let resource_type = self.resource_type;
+ let (needs_database, needs_table, needs_function, needs_view) = match
resource_type {
+ ResourceType::Catalog | ResourceType::CatalogAll => (false, false,
false, false),
+ ResourceType::Database | ResourceType::DatabaseAll => (true,
false, false, false),
+ ResourceType::Table | ResourceType::Column => (true, true, false,
false),
+ ResourceType::Function => (true, false, true, false),
+ ResourceType::View => (true, false, false, true),
+ };
+ for (present, needed, name) in [
+ (self.database.is_some(), needs_database, "database"),
+ (self.table.is_some(), needs_table, "table"),
+ (self.function.is_some(), needs_function, "function"),
+ (self.view.is_some(), needs_view, "view"),
+ ] {
+ if needed && !present {
+ return Err(bad_request(format!(
+ "{name} is required for {resource_type} resource."
+ )));
+ }
+ if !needed && present {
+ return Err(bad_request(format!(
+ "{resource_type} resource cannot contain {name}."
+ )));
+ }
+ }
+ Ok(())
+ }
+
+ fn unchecked(
+ resource_type: ResourceType,
+ database: Option<String>,
+ table: Option<String>,
+ function: Option<String>,
+ view: Option<String>,
+ ) -> Self {
+ Self {
+ resource_type,
+ database,
+ table,
+ function,
+ view,
+ }
+ }
+
+ pub fn resource_type(&self) -> ResourceType {
+ self.resource_type
+ }
+
+ /// The locator getters carry a `_name` suffix because the plain names
belong to the
+ /// constructors above; Rust allows only one inherent item per name.
+ pub fn database_name(&self) -> Option<&str> {
+ self.database.as_deref()
+ }
+
+ pub fn table_name(&self) -> Option<&str> {
+ self.table.as_deref()
+ }
+
+ pub fn function_name(&self) -> Option<&str> {
+ self.function.as_deref()
+ }
+
+ pub fn view_name(&self) -> Option<&str> {
+ self.view.as_deref()
+ }
+}
+
+/// The built-in access names and their canonical (upper-case) form (Java
`PermissionAccess`).
+pub struct PermissionAccess;
+
+impl PermissionAccess {
+ /// Maximum access-name length, in UTF-16 code units.
+ pub const MAX_LENGTH: usize = 32;
+ pub const ALL: &'static str = "ALL";
+ pub const CREATEDATABASE: &'static str = "CREATEDATABASE";
+ pub const DESCRIBE: &'static str = "DESCRIBE";
+ pub const ALTER: &'static str = "ALTER";
+ pub const DROP: &'static str = "DROP";
+ pub const CREATETABLE: &'static str = "CREATETABLE";
+ pub const CREATEFUNCTION: &'static str = "CREATEFUNCTION";
+ pub const CREATEVIEW: &'static str = "CREATEVIEW";
+ pub const LIST: &'static str = "LIST";
+ pub const SELECT: &'static str = "SELECT";
+ pub const UPDATE: &'static str = "UPDATE";
+ pub const GRANT: &'static str = "GRANT";
+
+ pub fn built_ins(resource_type: ResourceType) -> &'static [&'static str] {
+ match resource_type {
+ ResourceType::Catalog => &[
+ Self::ALL,
+ Self::ALTER,
+ Self::DROP,
+ Self::GRANT,
+ Self::CREATEDATABASE,
+ ],
+ ResourceType::CatalogAll => &[
+ Self::ALL,
+ Self::DESCRIBE,
+ Self::ALTER,
+ Self::DROP,
+ Self::GRANT,
+ Self::CREATETABLE,
+ Self::CREATEVIEW,
+ Self::CREATEFUNCTION,
+ Self::LIST,
+ Self::SELECT,
+ Self::UPDATE,
+ ],
+ ResourceType::Database => &[
+ Self::ALL,
+ Self::DESCRIBE,
+ Self::ALTER,
+ Self::DROP,
+ Self::GRANT,
+ Self::CREATETABLE,
+ Self::CREATEVIEW,
+ Self::CREATEFUNCTION,
+ Self::LIST,
+ ],
+ ResourceType::DatabaseAll | ResourceType::Table => &[
+ Self::ALL,
+ Self::SELECT,
+ Self::UPDATE,
+ Self::ALTER,
+ Self::DROP,
+ Self::GRANT,
+ ],
+ ResourceType::Column => &[Self::SELECT],
+ ResourceType::View | ResourceType::Function => &[
+ Self::ALL,
+ Self::SELECT,
+ Self::ALTER,
+ Self::DROP,
+ Self::GRANT,
+ ],
+ }
+ }
+
+ /// Upper-cases `access`; the length is checked again afterwards because
upper-casing can
+ /// grow a string ("ß" -> "SS").
+ pub fn canonicalize(access: &str) -> Result<String> {
+ if is_blank(access) {
+ return Err(bad_request("access cannot be empty."));
+ }
+ if utf16_len(access) > Self::MAX_LENGTH {
+ return Err(bad_request(format!(
+ "access must contain at most {} characters.",
+ Self::MAX_LENGTH
+ )));
+ }
+ let canonical = access.to_uppercase();
+ if utf16_len(&canonical) > Self::MAX_LENGTH {
+ return Err(bad_request(format!(
+ "access must contain at most {} characters after
canonicalization.",
+ Self::MAX_LENGTH
+ )));
+ }
+ let known = ResourceType::VALUES
+ .into_iter()
+ .any(|resource_type|
Self::built_ins(resource_type).contains(&canonical.as_str()));
+ if !known {
+ return Err(bad_request(format!("Unknown access '{canonical}'.")));
+ }
+ Ok(canonical)
+ }
+
+ pub fn canonicalize_for(resource_type: ResourceType, access: &str) ->
Result<String> {
+ let canonical = Self::canonicalize(access)?;
+ if !Self::built_ins(resource_type).contains(&canonical.as_str()) {
+ return Err(bad_request(format!(
+ "Access '{canonical}' is not valid for {resource_type}."
+ )));
+ }
+ Ok(canonical)
+ }
+}
+
+/// The column range of a `COLUMN` assignment: an allowlist (`columnNames`) or
a denylist
+/// (`excludedColumnNames`), never both (Java `PermissionColumns`). Names are
top-level fields;
+/// the allowlist denies columns added later, the denylist allows them.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PermissionColumns {
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ column_names: Option<Vec<String>>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ excluded_column_names: Option<Vec<String>>,
+}
+
+impl PermissionColumns {
+ pub fn names(column_names: Vec<String>) -> Result<Self> {
+ Self::check(&column_names, "columnNames")?;
+ Ok(Self {
+ column_names: Some(column_names),
+ excluded_column_names: None,
+ })
+ }
+
+ pub fn excluded(excluded_column_names: Vec<String>) -> Result<Self> {
+ Self::check(&excluded_column_names, "excludedColumnNames")?;
+ Ok(Self {
+ column_names: None,
+ excluded_column_names: Some(excluded_column_names),
+ })
+ }
+
+ /// The constructors' rules: exactly one list, non-empty, no blank or
duplicate name.
+ pub fn validate(&self) -> Result<()> {
+ match (&self.column_names, &self.excluded_column_names) {
+ (Some(columns), None) => Self::check(columns, "columnNames"),
+ (None, Some(columns)) => Self::check(columns,
"excludedColumnNames"),
+ _ => Err(bad_request(
+ "columns must contain exactly one of columnNames or
excludedColumnNames.",
+ )),
+ }
+ }
+
+ pub fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub fn excluded_column_names(&self) -> Option<&[String]> {
+ self.excluded_column_names.as_deref()
+ }
+
+ fn check(columns: &[String], field: &str) -> Result<()> {
+ if columns.is_empty() {
+ return Err(bad_request(format!("{field} cannot be empty.")));
+ }
+ if columns.iter().any(|column| is_blank(column)) {
+ return Err(bad_request(format!(
+ "{field} cannot contain an empty column name."
+ )));
+ }
+ let unique: std::collections::HashSet<&String> =
columns.iter().collect();
+ if unique.len() != columns.len() {
+ return Err(bad_request(format!(
+ "{field} cannot contain duplicate column names."
+ )));
+ }
+ Ok(())
+ }
+}
+
+/// One access granted to one principal on one resource (Java
`PermissionAssignment`).
+///
+/// The identity is `(resource, access, principal)`; `expire_time` is an
exclusive upper bound
+/// on the server clock.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PermissionAssignment {
+ resource: PermissionResource,
+ access: String,
+ principal: String,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ columns: Option<PermissionColumns>,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ expire_time: Option<String>,
+}
+
+impl PermissionAssignment {
+ /// Maximum principal length, in UTF-16 code units.
+ pub const MAX_PRINCIPAL_LENGTH: usize = 128;
+
+ pub fn new(
+ resource: PermissionResource,
+ access: &str,
+ principal: &str,
+ columns: Option<PermissionColumns>,
+ expire_time: Option<&str>,
+ ) -> Result<Self> {
+ Self {
+ resource,
+ access: access.to_string(),
+ principal: principal.to_string(),
+ columns,
+ expire_time: expire_time.map(str::to_string),
+ }
+ .canonicalized()
+ }
+
+ pub fn canonicalized(mut self) -> Result<Self> {
+ self.access =
+ PermissionAccess::canonicalize_for(self.resource.resource_type(),
&self.access)?;
+ Self::validate_principal(&self.principal)?;
+ self.resource = self.resource.canonicalized()?;
+ match (
+ self.resource.resource_type() == ResourceType::Column,
+ self.columns.is_some(),
+ ) {
+ (true, false) => return Err(bad_request("columns is required for
COLUMN resource.")),
+ (false, true) => return Err(bad_request("columns is only valid for
COLUMN resource.")),
+ _ => {}
+ }
+ if let Some(columns) = &self.columns {
+ columns.validate()?;
+ }
+ if let Some(expire_time) = &self.expire_time {
+ validate_expire_time(expire_time)?;
+ }
+ Ok(self)
+ }
+
+ /// Principals are opaque, server-defined strings: only blankness and
length are checked.
+ pub fn validate_principal(principal: &str) -> Result<()> {
+ if is_blank(principal) {
+ return Err(bad_request("principal cannot be empty."));
+ }
+ if utf16_len(principal) > Self::MAX_PRINCIPAL_LENGTH {
+ return Err(bad_request(format!(
+ "principal must contain at most {} characters.",
+ Self::MAX_PRINCIPAL_LENGTH
+ )));
+ }
+ Ok(())
+ }
+
+ pub fn resource(&self) -> &PermissionResource {
+ &self.resource
+ }
+
+ pub fn access(&self) -> &str {
+ &self.access
+ }
+
+ pub fn principal(&self) -> &str {
+ &self.principal
+ }
+
+ pub fn columns(&self) -> Option<&PermissionColumns> {
+ self.columns.as_ref()
+ }
+
+ pub fn expire_time(&self) -> Option<&str> {
+ self.expire_time.as_deref()
+ }
+}
+
+fn validate_expire_time(expire_time: &str) -> Result<()> {
+ // Java parses with `Instant.parse` (ISO_INSTANT, case-insensitive), which
takes the `Z`
+ // offset and no other, so an offset RFC-3339 allows here the server would
still refuse.
+ let instant = chrono::DateTime::parse_from_rfc3339(expire_time)
+ .ok()
+ .filter(|_| expire_time.ends_with(['Z', 'z']))
+ .ok_or_else(|| bad_request("expireTime must be an ISO-8601 UTC
instant."))?;
+ if instant.timestamp_subsec_nanos() % 1_000_000 != 0 {
+ return Err(bad_request(
+ "expireTime must have at most millisecond precision.",
+ ));
+ }
+ Ok(())
+}
+
+pub(crate) fn validate_max_results(max_results: u32) -> Result<()> {
+ if max_results == 0 || max_results > ListPermissionsRequest::MAX_PAGE_SIZE
{
+ return Err(bad_request(format!(
+ "maxResults must be between 1 and {}.",
+ ListPermissionsRequest::MAX_PAGE_SIZE
+ )));
+ }
+ Ok(())
+}
+
+/// Filters for `GET {prefix}/permissions` (Java `ListPermissionsRequest`).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ListPermissionsRequest {
+ pub resource: PermissionResource,
+ pub principal: Option<String>,
+ pub access: Option<String>,
+ /// Page size, `1..=MAX_PAGE_SIZE`.
+ pub max_results: Option<u32>,
+ /// Opaque continuation token from the previous page.
+ pub page_token: Option<String>,
+}
+
+impl ListPermissionsRequest {
+ /// Largest page a server has to honour.
+ pub const MAX_PAGE_SIZE: u32 = 1000;
+
+ pub fn new(resource: PermissionResource) -> Self {
+ Self {
+ resource,
+ principal: None,
+ access: None,
+ max_results: None,
+ page_token: None,
+ }
+ }
+
+ /// The query string as `(name, value)` pairs, in the order Java sends
them.
+ pub fn query_params(&self) -> Result<Vec<(&'static str, String)>> {
+ let resource = self.resource.clone().canonicalized()?;
+ let resource_type = resource.resource_type();
+ let mut params = vec![("resourceType", resource_type.to_string())];
+ for (name, value) in [
+ ("database", resource.database_name()),
+ ("table", resource.table_name()),
+ ("function", resource.function_name()),
+ ("view", resource.view_name()),
+ ] {
+ if let Some(value) = value {
+ params.push((name, value.to_string()));
+ }
+ }
+ if let Some(principal) = self.principal.as_deref().filter(|value|
!is_blank(value)) {
+ PermissionAssignment::validate_principal(principal)?;
+ params.push(("principal", principal.to_string()));
+ }
+ if let Some(access) = self.access.as_deref().filter(|value|
!is_blank(value)) {
+ params.push((
+ "access",
+ PermissionAccess::canonicalize_for(resource_type, access)?,
+ ));
+ }
+ if let Some(max_results) = self.max_results {
+ validate_max_results(max_results)?;
+ params.push(("maxResults", max_results.to_string()));
+ }
+ if let Some(page_token) = self.page_token.as_deref().filter(|value|
!value.is_empty()) {
+ params.push(("pageToken", page_token.to_string()));
+ }
+ Ok(params)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn table_resource() -> PermissionResource {
+ PermissionResource::table("sales", "orders")
+ }
+
+ #[test]
+ fn
test_resource_type_wire_names_are_upper_snake_and_read_case_insensitively() {
+ assert_eq!(
+ serde_json::to_string(&ResourceType::CatalogAll).unwrap(),
+ r#""CATALOG_ALL""#
+ );
+ assert_eq!(
+ serde_json::from_str::<ResourceType>(r#""database_all""#).unwrap(),
+ ResourceType::DatabaseAll
+ );
+ assert_eq!("view".parse::<ResourceType>().unwrap(),
ResourceType::View);
+ assert!("SCHEMA".parse::<ResourceType>().is_err());
+ assert_eq!(ResourceType::Function.to_string(), "FUNCTION");
+ }
+
+ #[test]
+ fn test_resource_serialization_omits_absent_locators() {
+ assert_eq!(
+ serde_json::to_string(&PermissionResource::catalog()).unwrap(),
+ r#"{"type":"CATALOG"}"#
+ );
+ assert_eq!(
+ serde_json::to_string(&PermissionResource::catalog_all()).unwrap(),
+ r#"{"type":"CATALOG_ALL"}"#
+ );
+ assert_eq!(
+
serde_json::to_string(&PermissionResource::database_all("sales")).unwrap(),
+ r#"{"type":"DATABASE_ALL","database":"sales"}"#
+ );
+ assert_eq!(
+ serde_json::to_string(&table_resource()).unwrap(),
+ r#"{"type":"TABLE","database":"sales","table":"orders"}"#
+ );
+ assert_eq!(
+ serde_json::to_string(&PermissionResource::function("sales",
"calculate_tax")).unwrap(),
+
r#"{"type":"FUNCTION","database":"sales","function":"calculate_tax"}"#
+ );
+ assert_eq!(
+ serde_json::to_string(&PermissionResource::view("sales",
"daily_orders")).unwrap(),
+ r#"{"type":"VIEW","database":"sales","view":"daily_orders"}"#
+ );
+ let parsed: PermissionResource = serde_json::from_str(
+
r#"{"type":"table","database":"sales","table":"orders","extra":1}"#,
+ )
+ .unwrap();
+ assert_eq!(parsed, table_resource());
+ }
+
+ #[test]
+ fn
test_resource_new_canonicalizes_blank_locators_and_enforces_the_type_rules() {
+ assert_eq!(
+ PermissionResource::new(ResourceType::Catalog, Some(""), Some("
"), None, None)
+ .unwrap(),
+ PermissionResource::catalog()
+ );
+ assert_eq!(
+ PermissionResource::new(
+ ResourceType::Function,
+ Some("sales"),
+ None,
+ Some("calculate_tax"),
+ None
+ )
+ .unwrap(),
+ PermissionResource::function("sales", "calculate_tax")
+ );
+ let message = |result: Result<PermissionResource>|
result.unwrap_err().to_string();
+ assert!(message(PermissionResource::new(
+ ResourceType::Catalog,
+ Some("sales"),
+ None,
+ None,
+ None
+ ))
+ .contains("CATALOG resource cannot contain database"));
+ assert!(message(PermissionResource::new(
+ ResourceType::Database,
+ None,
+ None,
+ None,
+ None
+ ))
+ .contains("database is required for DATABASE"));
+ assert!(message(PermissionResource::new(
+ ResourceType::Table,
+ Some("sales"),
+ None,
+ None,
+ None
+ ))
+ .contains("table is required for TABLE"));
+ assert!(message(PermissionResource::new(
+ ResourceType::Column,
+ Some("sales"),
+ Some("orders"),
+ Some("f"),
+ None
+ ))
+ .contains("COLUMN resource cannot contain function"));
+ assert!(message(PermissionResource::new(
+ ResourceType::Function,
+ Some("sales"),
+ None,
+ None,
+ Some("v")
+ ))
+ .contains("function is required for FUNCTION"));
+ assert!(message(PermissionResource::new(
+ ResourceType::View,
+ Some("sales"),
+ Some("orders"),
+ None,
+ Some("v")
+ ))
+ .contains("VIEW resource cannot contain table"));
+ }
+
+ #[test]
+ fn test_access_built_ins_match_java() {
+ let sorted = |resource_type| {
+ let mut values =
PermissionAccess::built_ins(resource_type).to_vec();
+ values.sort_unstable();
+ values
+ };
+ assert_eq!(
+ sorted(ResourceType::Catalog),
+ ["ALL", "ALTER", "CREATEDATABASE", "DROP", "GRANT"]
+ );
+ assert_eq!(
+ sorted(ResourceType::CatalogAll),
+ [
+ "ALL",
+ "ALTER",
+ "CREATEFUNCTION",
+ "CREATETABLE",
+ "CREATEVIEW",
+ "DESCRIBE",
+ "DROP",
+ "GRANT",
+ "LIST",
+ "SELECT",
+ "UPDATE"
+ ]
+ );
+ assert_eq!(
+ sorted(ResourceType::Database),
+ [
+ "ALL",
+ "ALTER",
+ "CREATEFUNCTION",
+ "CREATETABLE",
+ "CREATEVIEW",
+ "DESCRIBE",
+ "DROP",
+ "GRANT",
+ "LIST"
+ ]
+ );
+ assert_eq!(
+ sorted(ResourceType::DatabaseAll),
+ ["ALL", "ALTER", "DROP", "GRANT", "SELECT", "UPDATE"]
+ );
+ assert_eq!(
+ sorted(ResourceType::Table),
+ ["ALL", "ALTER", "DROP", "GRANT", "SELECT", "UPDATE"]
+ );
+ assert_eq!(
+ sorted(ResourceType::View),
+ ["ALL", "ALTER", "DROP", "GRANT", "SELECT"]
+ );
+ assert_eq!(
+ sorted(ResourceType::Function),
+ ["ALL", "ALTER", "DROP", "GRANT", "SELECT"]
+ );
+ assert_eq!(sorted(ResourceType::Column), ["SELECT"]);
+ }
+
+ #[test]
+ fn test_access_canonicalization() {
+ assert_eq!(
+ PermissionAccess::canonicalize("createdatabase").unwrap(),
+ "CREATEDATABASE"
+ );
+ assert_eq!(
+ PermissionAccess::canonicalize_for(ResourceType::Database,
"createview").unwrap(),
+ "CREATEVIEW"
+ );
+ let message = |result: Result<String>| result.unwrap_err().to_string();
+ assert!(message(PermissionAccess::canonicalize(" ")).contains("access
cannot be empty"));
+ assert!(message(PermissionAccess::canonicalize_for(
+ ResourceType::Catalog,
+ "SELECT"
+ ))
+ .contains("Access 'SELECT' is not valid for CATALOG"));
+ assert!(message(PermissionAccess::canonicalize_for(
+ ResourceType::CatalogAll,
+ "CREATEDATABASE"
+ ))
+ .contains("not valid for CATALOG_ALL"));
+ assert!(message(PermissionAccess::canonicalize_for(
+ ResourceType::DatabaseAll,
+ "LIST"
+ ))
+ .contains("not valid for DATABASE_ALL"));
+ for access in [
+ "USE_CATALOG",
+ "CREATE_DATABASE",
+ "USE_DATABASE",
+ "CREATE_TABLE",
+ "CREATE_VIEW",
+ "CREATE_FUNCTION",
+ "INSERT",
+ "DELETE",
+ "EXECUTE",
+ "MANAGE_PERMISSIONS",
+ "vendor.example/read_sensitive",
+ ] {
+ assert!(
+
message(PermissionAccess::canonicalize(access)).contains("Unknown access"),
+ "{access}"
+ );
+ }
+
assert!(message(PermissionAccess::canonicalize(&"A".repeat(33))).contains("32"));
+ // 18 code units before upper-casing, 34 after.
+ assert!(message(PermissionAccess::canonicalize(&format!(
+ "a/{}",
+ "ß".repeat(16)
+ )))
+ .contains("after canonicalization"));
+ }
+
+ const ASSIGNMENT_JSON: &str =
r#"{"resource":{"type":"TABLE","database":"sales","table":"orders"},"access":"SELECT","principal":"analyst","expireTime":"2027-01-01T00:00:00Z"}"#;
+ const COLUMN_ASSIGNMENT_JSON: &str =
r#"{"resource":{"type":"COLUMN","database":"sales","table":"orders"},"access":"SELECT","principal":"analyst","columns":{"columnNames":["id","region"]}}"#;
+
+ #[test]
+ fn test_assignment_round_trips_java_wire_json() {
+ let assignment: PermissionAssignment =
serde_json::from_str(ASSIGNMENT_JSON).unwrap();
+ assert_eq!(assignment.resource(), &table_resource());
+ assert_eq!(assignment.access(), "SELECT");
+ assert_eq!(assignment.principal(), "analyst");
+ assert_eq!(assignment.expire_time(), Some("2027-01-01T00:00:00Z"));
+ assert_eq!(assignment.columns(), None);
+ assert_eq!(serde_json::to_string(&assignment).unwrap(),
ASSIGNMENT_JSON);
+ let lower: PermissionAssignment =
+ serde_json::from_str(&ASSIGNMENT_JSON.replace("\"TABLE\"",
"\"table\"")).unwrap();
+ assert_eq!(lower, assignment);
+ let built = PermissionAssignment::new(
+ table_resource(),
+ "select",
+ "analyst",
+ None,
+ Some("2027-01-01T00:00:00Z"),
+ )
+ .unwrap();
+ assert_eq!(built, assignment);
+ }
+
+ #[test]
+ fn test_column_assignment_round_trips_java_wire_json() {
+ let assignment: PermissionAssignment =
+ serde_json::from_str(COLUMN_ASSIGNMENT_JSON).unwrap();
+ assert_eq!(assignment.resource().resource_type(),
ResourceType::Column);
+ let columns = assignment.columns().unwrap();
+ assert_eq!(
+ columns.column_names(),
+ Some(&["id".to_string(), "region".to_string()][..])
+ );
+ assert_eq!(columns.excluded_column_names(), None);
+ assert_eq!(
+ serde_json::to_string(&assignment).unwrap(),
+ COLUMN_ASSIGNMENT_JSON
+ );
+ let excluded =
PermissionColumns::excluded(vec!["email".to_string()]).unwrap();
+ assert_eq!(
+ serde_json::to_string(&excluded).unwrap(),
+ r#"{"excludedColumnNames":["email"]}"#
+ );
+ }
+
+ #[test]
+ fn test_responses_are_not_validated_but_requests_are() {
+ let precise =
+ ASSIGNMENT_JSON.replace("2027-01-01T00:00:00Z",
"2027-01-01T00:00:00.123456Z");
+ let listed: PermissionAssignment =
serde_json::from_str(&precise).unwrap();
+ assert_eq!(listed.expire_time(), Some("2027-01-01T00:00:00.123456Z"));
+ let message = |expire_time: &str| {
+ PermissionAssignment::new(
+ table_resource(),
+ "SELECT",
+ "analyst",
+ None,
+ Some(expire_time),
+ )
+ .unwrap_err()
+ .to_string()
+ };
+
assert!(message("2027-01-01T00:00:00.123456Z").contains("millisecond"));
+
assert!(message("2027-01-01T00:00:00.000001Z").contains("millisecond"));
+ assert!(message("tomorrow").contains("ISO-8601"));
+ assert!(message("2027-01-01T00:00:00+08:00").contains("ISO-8601"));
+ for expire_time in ["2027-01-01T00:00:00Z",
"2027-01-01T00:00:00.123Z"] {
+ assert!(
+ PermissionAssignment::new(
+ table_resource(),
+ "SELECT",
+ "analyst",
+ None,
+ Some(expire_time),
+ )
+ .is_ok(),
+ "{expire_time}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_assignment_validation_matches_java() {
+ let included =
+ PermissionColumns::names(vec!["id".to_string(),
"region".to_string()]).unwrap();
+ let column_resource = PermissionResource::column("sales", "orders");
+ assert_eq!(
+ PermissionAssignment::new(
+ column_resource.clone(),
+ "select",
+ "analyst",
+ Some(included.clone()),
+ None
+ )
+ .unwrap()
+ .columns(),
+ Some(&included)
+ );
+ assert_eq!(
+ PermissionAssignment::new(
+ PermissionResource::catalog(),
+ "createdatabase",
+ "analyst",
+ None,
+ None
+ )
+ .unwrap()
+ .access(),
+ "CREATEDATABASE"
+ );
+ let rejects = |resource: PermissionResource,
+ access: &str,
+ principal: &str,
+ columns: Option<PermissionColumns>,
+ needle: &str| {
+ let message = PermissionAssignment::new(resource, access,
principal, columns, None)
+ .unwrap_err()
+ .to_string();
+ assert!(
+ message.contains(needle),
+ "{message:?} should mention {needle:?}"
+ );
+ };
+ rejects(
+ PermissionResource::catalog(),
+ "SELECT",
+ "analyst",
+ None,
+ "not valid for CATALOG",
+ );
+ rejects(
+ PermissionResource::database("sales"),
+ "SELECT",
+ "analyst",
+ None,
+ "not valid for DATABASE",
+ );
+ rejects(
+ table_resource(),
+ "CREATEVIEW",
+ "analyst",
+ None,
+ "not valid for TABLE",
+ );
+ rejects(
+ PermissionResource::function("sales", "calculate_tax"),
+ "UPDATE",
+ "analyst",
+ None,
+ "not valid for FUNCTION",
+ );
+ rejects(
+ column_resource.clone(),
+ "UPDATE",
+ "analyst",
+ Some(included.clone()),
+ "not valid for COLUMN",
+ );
+ rejects(
+ column_resource,
+ "SELECT",
+ "analyst",
+ None,
+ "columns is required",
+ );
+ rejects(
+ table_resource(),
+ "SELECT",
+ "analyst",
+ Some(included),
+ "only valid for COLUMN",
+ );
+ rejects(
+ table_resource(),
+ "SELECT",
+ " ",
+ None,
+ "principal cannot be empty",
+ );
+ rejects(table_resource(), "SELECT", &"p".repeat(129), None, "128");
+
assert!(PermissionAssignment::validate_principal(&"p".repeat(128)).is_ok());
+ }
+
+ #[test]
+ fn test_column_range_rules() {
+ let message = |result: Result<PermissionColumns>|
result.unwrap_err().to_string();
+
assert!(message(PermissionColumns::names(vec![])).contains("columnNames cannot
be empty"));
+ assert!(message(PermissionColumns::excluded(vec![]))
+ .contains("excludedColumnNames cannot be empty"));
+ assert!(message(PermissionColumns::names(vec![
+ "id".to_string(),
+ "id".to_string()
+ ]))
+ .contains("duplicate"));
+ assert!(message(PermissionColumns::excluded(vec![" ".to_string()]))
+ .contains("empty column name"));
+ }
+
+ #[test]
+ fn
test_assignment_rejects_a_column_range_no_constructor_could_have_built() {
+ let neither =
serde_json::from_str::<PermissionColumns>(r#"{}"#).unwrap();
+ let both = serde_json::from_str::<PermissionColumns>(
+ r#"{"columnNames":["id"],"excludedColumnNames":["email"]}"#,
+ )
+ .unwrap();
+ let assign = |columns: PermissionColumns| {
+ PermissionAssignment::new(
+ PermissionResource::column("sales", "orders"),
+ "SELECT",
+ "analyst",
+ Some(columns),
+ None,
+ )
+ };
+ for columns in [neither, both] {
+ let message = assign(columns).unwrap_err().to_string();
+ assert!(message.contains("exactly one"), "{message:?}");
+ }
+
assert!(assign(PermissionColumns::names(vec!["id".to_string()]).unwrap()).is_ok());
+ }
+
+ #[test]
+ fn test_send_paths_reject_values_no_constructor_could_have_built() {
+ let blank_table = PermissionResource::table("sales", "");
+ let message = |error: Error| error.to_string();
+ assert!(message(blank_table.clone().canonicalized().unwrap_err())
+ .contains("table is required for TABLE"));
+ assert!(message(
+ PermissionResource::database(" ")
+ .canonicalized()
+ .unwrap_err()
+ )
+ .contains("database is required for DATABASE"));
+ assert!(table_resource().canonicalized().is_ok());
+ assert!(ListPermissionsRequest::new(blank_table.clone())
+ .query_params()
+ .unwrap_err()
+ .to_string()
+ .contains("table is required for TABLE"));
+ assert!(
+ PermissionAssignment::new(blank_table, "SELECT", "analyst", None,
None)
+ .unwrap_err()
+ .to_string()
+ .contains("table is required for TABLE")
+ );
+ let empty =
serde_json::from_str::<PermissionColumns>(r#"{"columnNames":[]}"#).unwrap();
+ assert!(message(empty.validate().unwrap_err()).contains("columnNames
cannot be empty"));
+ let rejected = PermissionAssignment::new(
+ PermissionResource::column("sales", "orders"),
+ "SELECT",
+ "analyst",
+ Some(empty),
+ None,
+ )
+ .unwrap_err()
+ .to_string();
+ assert!(
+ rejected.contains("columnNames cannot be empty"),
+ "{rejected:?}"
+ );
+ for json in [
+ r#"{"columnNames":[" "]}"#,
+ r#"{"excludedColumnNames":["id","id"]}"#,
+ ] {
+ let columns =
serde_json::from_str::<PermissionColumns>(json).unwrap();
+ assert!(columns.validate().is_err(), "{json}");
+ }
+ }
+
+ #[test]
+ fn test_send_paths_canonicalize_values_no_constructor_could_have_built() {
+ let blank_view: PermissionResource = serde_json::from_str(
+
r#"{"type":"TABLE","database":"sales","table":"orders","view":""}"#,
+ )
+ .unwrap();
+ let assignment =
+ PermissionAssignment::new(blank_view.clone(), "select", "analyst",
None, None).unwrap();
+ assert_eq!(
+ serde_json::to_string(&assignment).unwrap(),
+
r#"{"resource":{"type":"TABLE","database":"sales","table":"orders"},"access":"SELECT","principal":"analyst"}"#
+ );
+ assert_eq!(
+ ListPermissionsRequest::new(blank_view)
+ .query_params()
+ .unwrap(),
+ vec![
+ ("resourceType", "TABLE".to_string()),
+ ("database", "sales".to_string()),
+ ("table", "orders".to_string()),
+ ]
+ );
+ }
+
+ #[test]
+ fn test_list_permissions_request_query_params() {
+ let mut request =
ListPermissionsRequest::new(PermissionResource::database("sales"));
+ request.access = Some("createview".to_string());
+ request.principal = Some("analyst".to_string());
+ request.max_results = Some(25);
+ request.page_token = Some("start".to_string());
+ assert_eq!(
+ request.query_params().unwrap(),
+ vec![
+ ("resourceType", "DATABASE".to_string()),
+ ("database", "sales".to_string()),
+ ("principal", "analyst".to_string()),
+ ("access", "CREATEVIEW".to_string()),
+ ("maxResults", "25".to_string()),
+ ("pageToken", "start".to_string()),
+ ]
+ );
+ let mut request =
+ ListPermissionsRequest::new(PermissionResource::function("sales",
"calculate_tax"));
+ request.principal = Some(" ".to_string());
+ request.page_token = Some(" \t".to_string());
+ assert_eq!(
+ request.query_params().unwrap(),
+ vec![
+ ("resourceType", "FUNCTION".to_string()),
+ ("database", "sales".to_string()),
+ ("function", "calculate_tax".to_string()),
+ ("pageToken", " \t".to_string()),
+ ]
+ );
+ assert_eq!(
+ ListPermissionsRequest::new(PermissionResource::view("sales",
"daily_orders"))
+ .query_params()
+ .unwrap(),
+ vec![
+ ("resourceType", "VIEW".to_string()),
+ ("database", "sales".to_string()),
+ ("view", "daily_orders".to_string()),
+ ]
+ );
+ let mut request =
ListPermissionsRequest::new(PermissionResource::database("sales"));
+ request.access = Some("SELECT".to_string());
+ assert!(request
+ .query_params()
+ .unwrap_err()
+ .to_string()
+ .contains("not valid for DATABASE"));
+ let mut request =
ListPermissionsRequest::new(PermissionResource::catalog());
+ request.principal = Some("p".repeat(129));
+ assert!(request
+ .query_params()
+ .unwrap_err()
+ .to_string()
+ .contains("128"));
+ for max_results in [0, 1001] {
+ let mut request =
ListPermissionsRequest::new(PermissionResource::catalog());
+ request.max_results = Some(max_results);
+ assert!(
+ request
+ .query_params()
+ .unwrap_err()
+ .to_string()
+ .contains("1000"),
+ "{max_results}"
+ );
+ }
+ }
+
+ /// Java's `String.trim` strips only characters `<= U+0020`, so an em
space or a
+ /// non-breaking space is a legal principal, not a blank one.
+ #[test]
+ fn test_is_blank_matches_java_trim_semantics() {
+ for blank in ["", " ", "\t", "\n", " \r\n\t "] {
+ assert!(is_blank(blank), "{blank:?}");
+ }
+ for present in ["\u{2003}", "\u{00a0}", "a", " a "] {
+ assert!(!is_blank(present), "{present:?}");
+ }
+ assert!(PermissionAssignment::validate_principal("\u{2003}").is_ok());
+ let mut request =
ListPermissionsRequest::new(PermissionResource::catalog());
+ request.principal = Some("\u{2003}".to_string());
+ assert_eq!(
+ request.query_params().unwrap(),
+ vec![
+ ("resourceType", "CATALOG".to_string()),
+ ("principal", "\u{2003}".to_string()),
+ ]
+ );
+ }
+}
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index 181d0fa2..7f765cf7 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -21,6 +21,7 @@
pub mod api_request;
pub mod auth;
+pub mod management;
pub mod resource_paths;
pub mod rest_api;
pub mod rest_client;
@@ -34,15 +35,21 @@ pub use api_request::{
AlterDatabaseRequest, AlterTableRequest, AuthTableQueryRequest,
CreateDatabaseRequest,
CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateViewRequest,
DropPartitionsRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- RenameTableRequest,
+ RenameTableRequest, RevokePermissionRequest,
};
// Re-export response types
pub use api_response::{
AuditRESTResponse, AuthTableQueryResponse, ConfigResponse, ErrorResponse,
GetDatabaseResponse,
GetFunctionResponse, GetTableResponse, GetTableTokenResponse,
GetViewResponse,
- ListDatabasesResponse, ListFunctionsResponse, ListPartitionsResponse,
ListTablesResponse,
- ListViewsResponse, PagedList,
+ ListDatabasesResponse, ListFunctionsResponse, ListPartitionsResponse,
ListPermissionsResponse,
+ ListTablesResponse, ListViewsResponse, PagedList,
+};
+
+// Re-export management types
+pub use management::{
+ ListPermissionsRequest, PermissionAccess, PermissionAssignment,
PermissionColumns,
+ PermissionResource, ResourceType,
};
// Re-export error types
diff --git a/crates/paimon/src/api/resource_paths.rs
b/crates/paimon/src/api/resource_paths.rs
index 01351956..b1652172 100644
--- a/crates/paimon/src/api/resource_paths.rs
+++ b/crates/paimon/src/api/resource_paths.rs
@@ -35,6 +35,7 @@ impl ResourcePaths {
const PARTITIONS: &'static str = "partitions";
const VIEWS: &'static str = "views";
const FUNCTIONS: &'static str = "functions";
+ const PERMISSIONS: &'static str = "permissions";
/// Create a new ResourcePaths with the given prefix.
pub fn new(prefix: &str) -> Self {
@@ -243,6 +244,21 @@ impl ResourcePaths {
self.partitions(database_name, table_name)
)
}
+
+ /// Get the permission collection of the catalog (`{base}/permissions`).
+ pub fn permissions(&self) -> String {
+ format!("{}/{}", self.base_path, Self::PERMISSIONS)
+ }
+
+ /// Get the action endpoint that grants or replaces one permission
assignment.
+ pub fn grant_permission(&self) -> String {
+ format!("{}/grant", self.permissions())
+ }
+
+ /// Get the action endpoint that revokes one permission assignment.
+ pub fn revoke_permission(&self) -> String {
+ format!("{}/revoke", self.permissions())
+ }
}
#[cfg(test)]
@@ -344,4 +360,12 @@ mod tests {
"/v1/catalog/databases/analytics+db/tables/user+events/partitions/drop"
);
}
+
+ #[test]
+ fn test_permission_paths_hang_off_the_catalog_prefix() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(paths.permissions(), "/v1/catalog/permissions");
+ assert_eq!(paths.grant_permission(), "/v1/catalog/permissions/grant");
+ assert_eq!(paths.revoke_permission(),
"/v1/catalog/permissions/revoke");
+ }
}
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 8cd943c9..da1f72b7 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -32,14 +32,16 @@ use super::api_request::{
AlterDatabaseRequest, AlterTableRequest, AuthTableQueryRequest,
CreateDatabaseRequest,
CreateFunctionRequest, CreatePartitionsRequest, CreateTableRequest,
CreateViewRequest,
DropPartitionsRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- RenameTableRequest,
+ RenameTableRequest, RevokePermissionRequest,
};
use super::api_response::{
AuthTableQueryResponse, ConfigResponse, GetDatabaseResponse,
GetFunctionResponse,
GetTableResponse, GetViewResponse, ListDatabasesResponse,
ListFunctionsResponse,
- ListPartitionsResponse, ListTablesResponse, ListViewsResponse, PagedList,
+ ListPartitionsResponse, ListPermissionsResponse, ListTablesResponse,
ListViewsResponse,
+ PagedList,
};
use super::auth::{AuthProviderFactory, RESTAuthFunction};
+use super::management::{ListPermissionsRequest, PermissionAssignment,
PermissionResource};
use super::resource_paths::ResourcePaths;
use super::rest_util::RESTUtil;
@@ -846,6 +848,48 @@ impl RESTApi {
self.client.post(&path, &request).await
}
+ // ==================== Permission Management ====================
+ //
+ // Experimental REST management API (Java `RESTPermissionManagement`).
+
+ /// List the direct assignments on one exact resource or scope: the server
synthesizes none
+ /// of those inherited through `CATALOG_ALL` / `DATABASE_ALL`, and may
still list expired ones.
+ pub async fn list_permissions_paged(
+ &self,
+ request: &ListPermissionsRequest,
+ ) -> Result<PagedList<PermissionAssignment>> {
+ let params = request.query_params()?;
+ let response: ListPermissionsResponse = self
+ .client
+ .get(&self.resource_paths.permissions(), Some(¶ms))
+ .await?;
+ Ok(PagedList::new(
+ response.permissions,
+ response.next_page_token,
+ ))
+ }
+
+ /// Grant an assignment, replacing the expiry and column range of an
identical one.
+ pub async fn grant_permission(&self, assignment: &PermissionAssignment) ->
Result<()> {
+ let assignment = assignment.clone().canonicalized()?;
+ let path = self.resource_paths.grant_permission();
+ let _resp: serde_json::Value = self.client.post(&path,
&assignment).await?;
+ Ok(())
+ }
+
+ /// Revoke an assignment by identity; revoking an absent one succeeds.
+ pub async fn revoke_permission(
+ &self,
+ resource: &PermissionResource,
+ access: &str,
+ principal: &str,
+ ) -> Result<()> {
+ let request = RevokePermissionRequest::new(resource.clone(), access,
principal)?;
+ let path = self.resource_paths.revoke_permission();
+ let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
+ Ok(())
+ }
+
// ==================== Commit Operations ====================
/// Commit a snapshot for a table.
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 09f9691e..41ec6759 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -25,6 +25,7 @@ use std::sync::Arc;
use async_trait::async_trait;
+use crate::api::management::{ListPermissionsRequest, PermissionAssignment,
PermissionResource};
use crate::api::rest_api::RESTApi;
use crate::api::rest_error::RestError;
use crate::api::PagedList;
@@ -126,6 +127,33 @@ impl RESTCatalog {
.list_databases_paged(max_results, page_token,
database_name_pattern)
.await
}
+
+ // ======================= permission management ==========================
+ //
+ // Like Java's `RESTCatalog.permissionManagement()`, these live on the
REST catalog only
+ // and are deliberately not part of the `Catalog` trait.
+
+ pub async fn list_permissions_paged(
+ &self,
+ request: &ListPermissionsRequest,
+ ) -> Result<PagedList<PermissionAssignment>> {
+ self.api.list_permissions_paged(request).await
+ }
+
+ pub async fn grant_permission(&self, assignment: &PermissionAssignment) ->
Result<()> {
+ self.api.grant_permission(assignment).await
+ }
+
+ pub async fn revoke_permission(
+ &self,
+ resource: &PermissionResource,
+ access: &str,
+ principal: &str,
+ ) -> Result<()> {
+ self.api
+ .revoke_permission(resource, access, principal)
+ .await
+ }
}
// ============================================================================
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index b5b3bf56..81572025 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -38,8 +38,9 @@ use paimon::api::{
CreateFunctionRequest, CreatePartitionsRequest, CreateViewRequest,
DropPartitionsRequest,
ErrorResponse, GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
ListDatabasesResponse, ListFunctionsResponse,
ListPartitionsByFilterRequest,
- ListPartitionsByNamesRequest, ListPartitionsResponse, ListTablesResponse,
ListViewsResponse,
- RenameTableRequest, ResourcePaths,
+ ListPartitionsByNamesRequest, ListPartitionsResponse,
ListPermissionsResponse,
+ ListTablesResponse, ListViewsResponse, PermissionAssignment,
PermissionResource,
+ RenameTableRequest, ResourcePaths, ResourceType, RevokePermissionRequest,
};
use paimon::catalog::{Function, Identifier};
use paimon::spec::Partition;
@@ -70,6 +71,11 @@ struct MockState {
drop_partitions_calls: Vec<(String, String, DropPartitionsRequest)>,
create_partitions_error_status: Option<StatusCode>,
list_partitions_error_status: Option<StatusCode>,
+ permissions: Vec<PermissionAssignment>,
+ list_permissions_queries: Vec<HashMap<String, String>>,
+ grant_permission_bodies: Vec<serde_json::Value>,
+ revoke_permission_bodies: Vec<serde_json::Value>,
+ grant_permission_error_status: Option<StatusCode>,
/// ECS metadata role name (for token loader testing)
ecs_role_name: Option<String>,
/// ECS metadata token (for token loader testing)
@@ -111,22 +117,22 @@ fn partition_from_spec(spec: HashMap<String, String>) ->
Partition {
}
}
-fn paginate_names(
- names: Vec<String>,
+fn paginate<T: Clone>(
+ items: Vec<T>,
params: &HashMap<String, String>,
page_size: Option<usize>,
-) -> (Vec<String>, Option<String>) {
+) -> (Vec<T>, Option<String>) {
let Some(page_size) = page_size else {
- return (names, None);
+ return (items, None);
};
let offset = params
.get("pageToken")
.and_then(|token| token.parse::<usize>().ok())
.unwrap_or(0)
- .min(names.len());
- let end = (offset + page_size).min(names.len());
- let next_page_token = (end < names.len()).then(|| end.to_string());
- (names[offset..end].to_vec(), next_page_token)
+ .min(items.len());
+ let end = (offset + page_size).min(items.len());
+ let next_page_token = (end < items.len()).then(|| end.to_string());
+ (items[offset..end].to_vec(), next_page_token)
}
#[derive(Clone)]
@@ -488,7 +494,7 @@ impl RESTServer {
.filter_map(|key|
key.strip_prefix(&prefix).map(ToString::to_string))
.collect();
views.sort();
- let (views, next_page_token) = paginate_names(views, ¶ms,
s.list_page_size);
+ let (views, next_page_token) = paginate(views, ¶ms,
s.list_page_size);
(
StatusCode::OK,
Json(ListViewsResponse::new(views, next_page_token)),
@@ -594,7 +600,7 @@ impl RESTServer {
.filter_map(|key|
key.strip_prefix(&prefix).map(ToString::to_string))
.collect();
functions.sort();
- let (functions, next_page_token) = paginate_names(functions, ¶ms,
s.list_page_size);
+ let (functions, next_page_token) = paginate(functions, ¶ms,
s.list_page_size);
(
StatusCode::OK,
Json(ListFunctionsResponse::new(functions, next_page_token)),
@@ -1156,6 +1162,116 @@ impl RESTServer {
.into_response()
}
+ // ==================== Permission management ====================
+
+ fn same_assignment_identity(left: &PermissionAssignment, right:
&PermissionAssignment) -> bool {
+ left.resource() == right.resource()
+ && left.access() == right.access()
+ && left.principal() == right.principal()
+ }
+
+ fn bad_request(message: String) -> axum::response::Response {
+ let error = ErrorResponse::new(None, None, Some(message), Some(400));
+ (StatusCode::BAD_REQUEST, Json(error)).into_response()
+ }
+
+ /// Handle GET {prefix}/permissions - direct assignments on the exact
resource in the query.
+ pub async fn list_permissions(
+ Query(params): Query<HashMap<String, String>>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ inner.list_permissions_queries.push(params.clone());
+ let Some(resource_type) = params
+ .get("resourceType")
+ .and_then(|value| value.parse::<ResourceType>().ok())
+ else {
+ return Self::bad_request("resourceType is required".to_string());
+ };
+ let locator = |name: &str| params.get(name).map(String::as_str);
+ let resource = match PermissionResource::new(
+ resource_type,
+ locator("database"),
+ locator("table"),
+ locator("function"),
+ locator("view"),
+ ) {
+ Ok(resource) => resource,
+ Err(error) => return Self::bad_request(error.to_string()),
+ };
+ let matching: Vec<PermissionAssignment> = inner
+ .permissions
+ .iter()
+ .filter(|assignment| assignment.resource() == &resource)
+ .filter(|assignment| {
+ params
+ .get("principal")
+ .is_none_or(|p| p == assignment.principal())
+ })
+ .filter(|assignment| {
+ params
+ .get("access")
+ .is_none_or(|a| a == assignment.access())
+ })
+ .cloned()
+ .collect();
+ let page_size = params
+ .get("maxResults")
+ .and_then(|value| value.parse().ok());
+ let (permissions, next_page_token) = paginate(matching, ¶ms,
page_size);
+ (
+ StatusCode::OK,
+ Json(ListPermissionsResponse::new(permissions, next_page_token)),
+ )
+ .into_response()
+ }
+
+ /// Handle POST {prefix}/permissions/grant - upsert by (resource, access,
principal).
+ pub async fn grant_permission(
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(body): Json<serde_json::Value>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ inner.grant_permission_bodies.push(body.clone());
+ if let Some(status) = inner.grant_permission_error_status {
+ let error = ErrorResponse::new(
+ None,
+ None,
+ Some("forbidden".to_string()),
+ Some(status.as_u16() as i32),
+ );
+ return (status, Json(error)).into_response();
+ }
+ let assignment: PermissionAssignment = match
serde_json::from_value(body) {
+ Ok(assignment) => assignment,
+ Err(error) => return Self::bad_request(error.to_string()),
+ };
+ inner
+ .permissions
+ .retain(|existing| !Self::same_assignment_identity(existing,
&assignment));
+ inner.permissions.push(assignment);
+ StatusCode::OK.into_response()
+ }
+
+ /// Handle POST {prefix}/permissions/revoke - idempotent removal by
identity.
+ pub async fn revoke_permission(
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(body): Json<serde_json::Value>,
+ ) -> impl IntoResponse {
+ let mut inner = state.inner.lock().unwrap();
+ inner.revoke_permission_bodies.push(body.clone());
+ let request: RevokePermissionRequest = match
serde_json::from_value(body) {
+ Ok(request) => request,
+ Err(error) => return Self::bad_request(error.to_string()),
+ };
+ inner.permissions.retain(|existing| {
+ !(existing.resource() == &request.resource
+ && existing.access() == request.access
+ && existing.principal() == request.principal)
+ });
+ StatusCode::OK.into_response()
+ }
+
/// Handle POST /rename-table - rename a table.
pub async fn rename_table(
Extension(state): Extension<Arc<RESTServer>>,
@@ -1548,6 +1664,30 @@ impl RESTServer {
.unwrap_or_default()
}
+ /// Every query string received by `GET /permissions`.
+ pub fn list_permissions_queries(&self) -> Vec<HashMap<String, String>> {
+ self.inner.lock().unwrap().list_permissions_queries.clone()
+ }
+
+ /// Raw JSON bodies received by `POST /permissions/grant`.
+ pub fn grant_permission_bodies(&self) -> Vec<serde_json::Value> {
+ self.inner.lock().unwrap().grant_permission_bodies.clone()
+ }
+
+ /// Raw JSON bodies received by `POST /permissions/revoke`.
+ pub fn revoke_permission_bodies(&self) -> Vec<serde_json::Value> {
+ self.inner.lock().unwrap().revoke_permission_bodies.clone()
+ }
+
+ pub fn permissions(&self) -> Vec<PermissionAssignment> {
+ self.inner.lock().unwrap().permissions.clone()
+ }
+
+ /// Make every grant fail with `status` (e.g. 403) instead of storing it.
+ pub fn set_grant_permission_error_status(&self, status:
Option<StatusCode>) {
+ self.inner.lock().unwrap().grant_permission_error_status = status;
+ }
+
/// Return all create-partitions calls received by the server.
pub fn create_partitions_calls(&self) -> Vec<(String, String,
CreatePartitionsRequest)> {
self.inner.lock().unwrap().create_partitions_calls.clone()
@@ -1711,6 +1851,18 @@ pub async fn start_mock_server(
&format!("{prefix}/tables/rename"),
post(RESTServer::rename_table),
)
+ .route(
+ &format!("{prefix}/permissions"),
+ get(RESTServer::list_permissions),
+ )
+ .route(
+ &format!("{prefix}/permissions/grant"),
+ post(RESTServer::grant_permission),
+ )
+ .route(
+ &format!("{prefix}/permissions/revoke"),
+ post(RESTServer::revoke_permission),
+ )
// ECS metadata endpoints (for token loader testing)
.route(
"/ram/security-credentials/",
diff --git a/crates/paimon/tests/rest_api_test.rs
b/crates/paimon/tests/rest_api_test.rs
index 086c3301..f8ef468b 100644
--- a/crates/paimon/tests/rest_api_test.rs
+++ b/crates/paimon/tests/rest_api_test.rs
@@ -22,9 +22,13 @@
use std::collections::HashMap;
+use axum::http::StatusCode;
use paimon::api::auth::{DLFECSTokenLoader, DLFToken, DLFTokenLoader};
use paimon::api::rest_api::RESTApi;
-use paimon::api::{ConfigResponse, CreatePartitionsRequest,
DropPartitionsRequest};
+use paimon::api::{
+ ConfigResponse, CreatePartitionsRequest, DropPartitionsRequest,
ListPermissionsRequest,
+ PermissionAssignment, PermissionColumns, PermissionResource, RestError,
+};
use paimon::catalog::{Function, FunctionDefinition, Identifier, ViewSchema};
use paimon::common::Options;
use paimon::spec::DataField;
@@ -868,6 +872,265 @@ async fn
test_list_partitions_rejects_repeated_page_token() {
);
}
+// ==================== Permission Management Tests ====================
+
+fn orders_table() -> PermissionResource {
+ PermissionResource::table("sales", "orders")
+}
+
+fn assignment(access: &str, principal: &str) -> PermissionAssignment {
+ PermissionAssignment::new(orders_table(), access, principal, None,
None).unwrap()
+}
+
+#[tokio::test]
+async fn test_list_permissions_sends_every_filter_and_parses_the_page() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ ctx.api
+ .grant_permission(&assignment("SELECT", "analyst"))
+ .await
+ .unwrap();
+ ctx.api
+ .grant_permission(&assignment("UPDATE", "writer"))
+ .await
+ .unwrap();
+
+ let mut request = ListPermissionsRequest::new(orders_table());
+ request.principal = Some("analyst".to_string());
+ request.access = Some("select".to_string());
+ request.max_results = Some(25);
+ request.page_token = Some("0".to_string());
+ let page = ctx.api.list_permissions_paged(&request).await.unwrap();
+
+ assert_eq!(page.elements, vec![assignment("SELECT", "analyst")]);
+ assert_eq!(page.next_page_token, None);
+ assert_eq!(
+ ctx.server.list_permissions_queries(),
+ vec![HashMap::from([
+ ("resourceType".to_string(), "TABLE".to_string()),
+ ("database".to_string(), "sales".to_string()),
+ ("table".to_string(), "orders".to_string()),
+ ("principal".to_string(), "analyst".to_string()),
+ ("access".to_string(), "SELECT".to_string()),
+ ("maxResults".to_string(), "25".to_string()),
+ ("pageToken".to_string(), "0".to_string()),
+ ])]
+ );
+}
+
+#[tokio::test]
+async fn test_list_permissions_pages_with_the_server_token() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ for principal in ["a", "b", "c"] {
+ ctx.api
+ .grant_permission(&assignment("SELECT", principal))
+ .await
+ .unwrap();
+ }
+ let mut request = ListPermissionsRequest::new(orders_table());
+ request.max_results = Some(2);
+ let first = ctx.api.list_permissions_paged(&request).await.unwrap();
+ assert_eq!(first.elements.len(), 2);
+ let token = first.next_page_token.expect("a second page");
+ request.page_token = Some(token);
+ let second = ctx.api.list_permissions_paged(&request).await.unwrap();
+ assert_eq!(second.elements, vec![assignment("SELECT", "c")]);
+ assert_eq!(second.next_page_token, None);
+}
+
+#[tokio::test]
+async fn test_grant_and_revoke_post_the_java_wire_shapes() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let granted = PermissionAssignment::new(
+ orders_table(),
+ "select",
+ "analyst",
+ None,
+ Some("2027-01-01T00:00:00Z"),
+ )
+ .unwrap();
+ ctx.api.grant_permission(&granted).await.unwrap();
+ // Granting the same identity again replaces the expiry instead of adding
a row.
+ ctx.api
+ .grant_permission(&assignment("SELECT", "analyst"))
+ .await
+ .unwrap();
+ assert_eq!(
+ ctx.server.permissions(),
+ vec![assignment("SELECT", "analyst")]
+ );
+
+ ctx.api
+ .revoke_permission(&orders_table(), "select", "analyst")
+ .await
+ .unwrap();
+ assert!(ctx.server.permissions().is_empty());
+
+ assert_eq!(
+ ctx.server.grant_permission_bodies()[0],
+ json!({
+ "resource": {"type": "TABLE", "database": "sales", "table":
"orders"},
+ "access": "SELECT",
+ "principal": "analyst",
+ "expireTime": "2027-01-01T00:00:00Z"
+ })
+ );
+ assert_eq!(
+ ctx.server.revoke_permission_bodies(),
+ vec![json!({
+ "resource": {"type": "TABLE", "database": "sales", "table":
"orders"},
+ "access": "SELECT",
+ "principal": "analyst"
+ })]
+ );
+}
+
+#[tokio::test]
+async fn test_column_grant_carries_the_range_but_revoke_only_the_identity() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let columns = PermissionColumns::names(vec!["id".to_string(),
"region".to_string()]).unwrap();
+ let granted = PermissionAssignment::new(
+ PermissionResource::column("sales", "orders"),
+ "SELECT",
+ "analyst",
+ Some(columns),
+ None,
+ )
+ .unwrap();
+ ctx.api.grant_permission(&granted).await.unwrap();
+ ctx.api
+ .revoke_permission(granted.resource(), granted.access(),
granted.principal())
+ .await
+ .unwrap();
+
+ let grant = &ctx.server.grant_permission_bodies()[0];
+ assert_eq!(grant["resource"]["type"], "COLUMN");
+ assert_eq!(grant["columns"], json!({"columnNames": ["id", "region"]}));
+ let revoke = &ctx.server.revoke_permission_bodies()[0];
+ assert_eq!(revoke["resource"]["type"], "COLUMN");
+ assert!(revoke.get("columns").is_none());
+ assert!(revoke.get("expireTime").is_none());
+}
+
+#[tokio::test]
+async fn test_forbidden_grant_surfaces_the_rest_error() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ ctx.server
+ .set_grant_permission_error_status(Some(StatusCode::FORBIDDEN));
+ let error = ctx
+ .api
+ .grant_permission(&assignment("SELECT", "denied"))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(
+ error,
+ paimon::Error::RestApi {
+ source: RestError::Forbidden { .. }
+ }
+ ),
+ "{error:?}"
+ );
+}
+
+#[tokio::test]
+async fn test_revoking_an_absent_assignment_is_idempotent() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ for _ in 0..2 {
+ ctx.api
+ .revoke_permission(&orders_table(), "SELECT", "missing")
+ .await
+ .unwrap();
+ }
+ assert_eq!(ctx.server.revoke_permission_bodies().len(), 2);
+}
+
+#[tokio::test]
+async fn test_invalid_permission_requests_never_reach_the_server() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let error = ctx
+ .api
+ .revoke_permission(&PermissionResource::catalog(), "SELECT", "analyst")
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("not valid for CATALOG"),
+ "{error}"
+ );
+ let mut request = ListPermissionsRequest::new(orders_table());
+ request.max_results = Some(1001);
+ let error = ctx.api.list_permissions_paged(&request).await.unwrap_err();
+ assert!(error.to_string().contains("1000"), "{error}");
+ assert!(ctx.server.revoke_permission_bodies().is_empty());
+ assert!(ctx.server.list_permissions_queries().is_empty());
+}
+
+#[tokio::test]
+async fn test_grant_sends_the_canonical_form_of_a_listed_assignment() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let listed: PermissionAssignment = serde_json::from_value(json!({
+ "resource": {"type": "TABLE", "database": "sales", "table": "orders",
"view": ""},
+ "access": "select",
+ "principal": "analyst",
+ }))
+ .unwrap();
+ ctx.api.grant_permission(&listed).await.unwrap();
+ assert_eq!(
+ ctx.server.grant_permission_bodies(),
+ vec![json!({
+ "resource": {"type": "TABLE", "database": "sales", "table":
"orders"},
+ "access": "SELECT",
+ "principal": "analyst"
+ })]
+ );
+
+ let mut request = ListPermissionsRequest::new(listed.resource().clone());
+ request.access = Some("select".to_string());
+ ctx.api.list_permissions_paged(&request).await.unwrap();
+ assert_eq!(
+ ctx.server.list_permissions_queries(),
+ vec![HashMap::from([
+ ("resourceType".to_string(), "TABLE".to_string()),
+ ("database".to_string(), "sales".to_string()),
+ ("table".to_string(), "orders".to_string()),
+ ("access".to_string(), "SELECT".to_string()),
+ ])]
+ );
+}
+
+#[tokio::test]
+async fn
test_a_resource_no_constructor_could_have_built_never_reaches_the_server() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let blank = PermissionResource::table("sales", "");
+ let assignment: PermissionAssignment = serde_json::from_value(json!({
+ "resource": {"type": "TABLE", "database": "sales", "table": ""},
+ "access": "SELECT",
+ "principal": "analyst",
+ }))
+ .unwrap();
+ let rejected = |error: paimon::Error| {
+ assert!(
+ error.to_string().contains("table is required for TABLE"),
+ "{error}"
+ );
+ };
+ rejected(ctx.api.grant_permission(&assignment).await.unwrap_err());
+ rejected(
+ ctx.api
+ .revoke_permission(&blank, "SELECT", "analyst")
+ .await
+ .unwrap_err(),
+ );
+ rejected(
+ ctx.api
+ .list_permissions_paged(&ListPermissionsRequest::new(blank))
+ .await
+ .unwrap_err(),
+ );
+ assert!(ctx.server.grant_permission_bodies().is_empty());
+ assert!(ctx.server.revoke_permission_bodies().is_empty());
+ assert!(ctx.server.list_permissions_queries().is_empty());
+}
+
// ==================== Rename Table Tests ====================
#[tokio::test]
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 63729cc5..691c1159 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -28,6 +28,7 @@ use arrow_schema::{DataType as ArrowDataType, Field as
ArrowField, Schema as Arr
use axum::http::StatusCode;
use futures::TryStreamExt;
use paimon::api::ConfigResponse;
+use paimon::api::{ListPermissionsRequest, PermissionAssignment,
PermissionResource};
use paimon::catalog::{Catalog, Function, FunctionDefinition, Identifier,
RESTCatalog, ViewSchema};
use paimon::common::Options;
use paimon::spec::{
@@ -2460,3 +2461,30 @@ async fn test_load_table_rejects_unknown_declared_type()
{
"{err:?}"
);
}
+
+#[tokio::test]
+async fn test_rest_catalog_manages_permissions_end_to_end() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let resource = PermissionResource::table("default", "orders");
+ let assignment =
+ PermissionAssignment::new(resource.clone(), "select", "analyst", None,
None).unwrap();
+
+ ctx.catalog.grant_permission(&assignment).await.unwrap();
+ let page = ctx
+ .catalog
+ .list_permissions_paged(&ListPermissionsRequest::new(resource.clone()))
+ .await
+ .unwrap();
+ assert_eq!(page.elements, vec![assignment]);
+
+ ctx.catalog
+ .revoke_permission(&resource, "SELECT", "analyst")
+ .await
+ .unwrap();
+ let page = ctx
+ .catalog
+ .list_permissions_paged(&ListPermissionsRequest::new(resource))
+ .await
+ .unwrap();
+ assert!(page.elements.is_empty());
+}
