diff --git a/crates/paimon/src/api/api_request.rs b/crates/paimon/src/api/api_request.rs index 48646757a..8c721fdf5 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 { + 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 9d68cf3d4..ce62a392c 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_page_token: Option, +} + +impl ListPermissionsResponse { + pub fn new(permissions: Vec, next_page_token: Option) -> 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 000000000..188ba370f --- /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) -> 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) -> Option { + 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 { + 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(&self, serializer: S) -> std::result::Result { + serializer.serialize_str(self.as_str()) + } +} + +impl<'de> Deserialize<'de> for ResourceType { + fn deserialize>(deserializer: D) -> std::result::Result { + 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + table: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + function: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + view: Option, +} + +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) -> Self { + Self::unchecked( + ResourceType::Database, + Some(database.into()), + None, + None, + None, + ) + } + + pub fn database_all(database: impl Into) -> Self { + Self::unchecked( + ResourceType::DatabaseAll, + Some(database.into()), + None, + None, + None, + ) + } + + pub fn table(database: impl Into, table: impl Into) -> 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, table: impl Into) -> Self { + Self::unchecked( + ResourceType::Column, + Some(database.into()), + Some(table.into()), + None, + None, + ) + } + + pub fn function(database: impl Into, function: impl Into) -> Self { + Self::unchecked( + ResourceType::Function, + Some(database.into()), + None, + Some(function.into()), + None, + ) + } + + pub fn view(database: impl Into, view: impl Into) -> 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::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.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, + table: Option, + function: Option, + view: Option, + ) -> 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 { + 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 { + 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>, + #[serde(default, skip_serializing_if = "Option::is_none")] + excluded_column_names: Option>, +} + +impl PermissionColumns { + pub fn names(column_names: Vec) -> Result { + Self::check(&column_names, "columnNames")?; + Ok(Self { + column_names: Some(column_names), + excluded_column_names: None, + }) + } + + pub fn excluded(excluded_column_names: Vec) -> Result { + 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + expire_time: Option, +} + +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, + expire_time: Option<&str>, + ) -> Result { + 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.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, + pub access: Option, + /// Page size, `1..=MAX_PAGE_SIZE`. + pub max_results: Option, + /// Opaque continuation token from the previous page. + pub page_token: Option, +} + +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> { + 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::(r#""database_all""#).unwrap(), + ResourceType::DatabaseAll + ); + assert_eq!("view".parse::().unwrap(), ResourceType::View); + assert!("SCHEMA".parse::().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| 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| 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, + 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| 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::(r#"{}"#).unwrap(); + let both = serde_json::from_str::( + 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::(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::(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 181d0fa24..7f765cf73 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 013519562..b1652172f 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 8cd943c92..da1f72b7a 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> { + 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 09f9691ea..41ec6759f 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> { + 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 b5b3bf56b..815720256 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, list_partitions_error_status: Option, + permissions: Vec, + list_permissions_queries: Vec>, + grant_permission_bodies: Vec, + revoke_permission_bodies: Vec, + grant_permission_error_status: Option, /// ECS metadata role name (for token loader testing) ecs_role_name: Option, /// ECS metadata token (for token loader testing) @@ -111,22 +117,22 @@ fn partition_from_spec(spec: HashMap) -> Partition { } } -fn paginate_names( - names: Vec, +fn paginate( + items: Vec, params: &HashMap, page_size: Option, -) -> (Vec, Option) { +) -> (Vec, Option) { let Some(page_size) = page_size else { - return (names, None); + return (items, None); }; let offset = params .get("pageToken") .and_then(|token| token.parse::().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>, + Extension(state): Extension>, + ) -> 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::().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 = 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>, + Json(body): Json, + ) -> 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>, + Json(body): Json, + ) -> 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>, @@ -1548,6 +1664,30 @@ impl RESTServer { .unwrap_or_default() } + /// Every query string received by `GET /permissions`. + pub fn list_permissions_queries(&self) -> Vec> { + self.inner.lock().unwrap().list_permissions_queries.clone() + } + + /// Raw JSON bodies received by `POST /permissions/grant`. + pub fn grant_permission_bodies(&self) -> Vec { + self.inner.lock().unwrap().grant_permission_bodies.clone() + } + + /// Raw JSON bodies received by `POST /permissions/revoke`. + pub fn revoke_permission_bodies(&self) -> Vec { + self.inner.lock().unwrap().revoke_permission_bodies.clone() + } + + pub fn permissions(&self) -> Vec { + 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) { + 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 086c33013..f8ef468b2 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 63729cc59..691c11593 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()); +}