From 96e1832d6cbe622a089583a20f5ec326f90dfaac Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Fri, 14 Aug 2026 12:42:56 -0400 Subject: [PATCH] feat(auth): authorize query-auth reads and carry the grant on the split --- crates/paimon/src/api/api_response.rs | 17 +- crates/paimon/src/spec/schema.rs | 29 +- crates/paimon/src/table/format_table_read.rs | 10 +- crates/paimon/src/table/format_table_scan.rs | 16 +- crates/paimon/src/table/incremental_scan.rs | 20 +- crates/paimon/src/table/mod.rs | 125 ++++++ crates/paimon/src/table/query_auth.rs | 320 ++++++++++++++ crates/paimon/src/table/read_builder.rs | 35 +- crates/paimon/src/table/rest_env.rs | 78 +++- crates/paimon/src/table/source.rs | 77 ++++ crates/paimon/src/table/table_read.rs | 371 ++++++++++++++++- crates/paimon/src/table/table_scan.rs | 144 +++++-- crates/paimon/tests/mock_server.rs | 145 ++++++- crates/paimon/tests/rest_catalog_test.rs | 412 +++++++++++++++++++ 14 files changed, 1736 insertions(+), 63 deletions(-) create mode 100644 crates/paimon/src/table/query_auth.rs diff --git a/crates/paimon/src/api/api_response.rs b/crates/paimon/src/api/api_response.rs index 9d68cf3d4..dbafea866 100644 --- a/crates/paimon/src/api/api_response.rs +++ b/crates/paimon/src/api/api_response.rs @@ -476,8 +476,11 @@ pub struct GetTableTokenResponse { /// Response for auth table query: the per-user row filter and column masking the /// client must enforce at read time for a `query-auth.enabled` table. +/// +/// Unknown fields are rejected: an absent one reads as "no rule", so protocol +/// drift would look like an unrestricted grant. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] +#[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct AuthTableQueryResponse { /// JSON-serialized row-filter predicates, ANDed together. Empty/None = no filter. pub filter: Option>, @@ -495,6 +498,18 @@ impl AuthTableQueryResponse { #[cfg(test)] mod tests { + + #[test] + fn test_auth_table_query_response_rejects_unknown_fields() { + let drifted = r#"{"rowFilter":["restricted"]}"#; + assert!( + serde_json::from_str::(drifted).is_err(), + "an auth response this client does not understand must not parse" + ); + assert!(serde_json::from_str::("{}") + .unwrap() + .is_unrestricted()); + } use super::*; #[test] diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 635e188ee..000d6a82c 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -621,22 +621,27 @@ impl TableSchema { } } -/// Reject column names reserved for system use, mirroring Java `SpecialFields`: -/// the five `SYSTEM_FIELD_NAMES` and the `_KEY_` key-field prefix. +/// Whether `name` is one Paimon reserves for a system column. Java +/// `SpecialFields.SYSTEM_FIELD_NAMES` plus the `_KEY_` key-field prefix. +pub(crate) fn is_reserved_system_field_name(name: &str) -> bool { + name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name) +} + +// Java SpecialFields.SYSTEM_FIELD_NAMES. +const SYSTEM_FIELD_NAMES: [&str; 5] = [ + SEQUENCE_NUMBER_FIELD_NAME, + VALUE_KIND_FIELD_NAME, + "_LEVEL", + ROW_KIND_FIELD_NAME, + ROW_ID_FIELD_NAME, +]; +const KEY_FIELD_PREFIX: &str = "_KEY_"; + +/// Reject column names reserved for system use, mirroring Java `SpecialFields`. /// /// A user column colliding with a system field is otherwise excluded from the /// physical read and silently filled with the system value. fn validate_no_reserved_field_names(fields: &[DataField]) -> crate::Result<()> { - // Java SpecialFields.SYSTEM_FIELD_NAMES. - const SYSTEM_FIELD_NAMES: [&str; 5] = [ - SEQUENCE_NUMBER_FIELD_NAME, - VALUE_KIND_FIELD_NAME, - "_LEVEL", - ROW_KIND_FIELD_NAME, - ROW_ID_FIELD_NAME, - ]; - const KEY_FIELD_PREFIX: &str = "_KEY_"; - for field in fields { let name = field.name(); if name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name) { diff --git a/crates/paimon/src/table/format_table_read.rs b/crates/paimon/src/table/format_table_read.rs index 9f0814d6f..3b81012a1 100644 --- a/crates/paimon/src/table/format_table_read.rs +++ b/crates/paimon/src/table/format_table_read.rs @@ -103,7 +103,15 @@ impl<'a> FormatTableRead<'a> { data_splits: &[DataSplit], ) -> crate::Result { let core_options = self.table.schema().core_options(); - core_options.ensure_read_authorized()?; + core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?; + // Sync, so the marker stands in for asking the server. + if core_options.query_auth_enabled() + || data_splits.iter().any(|split| split.query_auth_required()) + { + return Err(super::query_auth::unsupported( + "a format table cannot apply a row filter or column masking", + )); + } // Mapping the conjunct onto the data fields drops it, so the read would // silently ignore the filter. Guard on the read path, not the builder: // `TableRead` is public and can be constructed and filtered directly. diff --git a/crates/paimon/src/table/format_table_scan.rs b/crates/paimon/src/table/format_table_scan.rs index 1638c0a67..f81b83356 100644 --- a/crates/paimon/src/table/format_table_scan.rs +++ b/crates/paimon/src/table/format_table_scan.rs @@ -56,20 +56,28 @@ impl<'a> FormatTableScan<'a> { } pub(crate) async fn plan(&self) -> crate::Result { - self.ensure_query_auth_allowed()?; + self.ensure_query_auth_allowed().await?; self.plan_inner(None).await } pub(crate) async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> { - self.ensure_query_auth_allowed()?; + self.ensure_query_auth_allowed().await?; let mut trace = ScanTrace::default(); let plan = self.plan_inner(Some(&mut trace)).await?; trace.planned_data_file_bytes = plan.planned_data_file_bytes(); Ok((plan, trace)) } - fn ensure_query_auth_allowed(&self) -> crate::Result<()> { - CoreOptions::new(self.table.schema().options()).ensure_read_authorized() + /// Refused outright. Asks the server: the option can be set after a load. + async fn ensure_query_auth_allowed(&self) -> crate::Result<()> { + let core_options = CoreOptions::new(self.table.schema().options()); + core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?; + if self.table.server_query_auth_enabled().await? { + return Err(super::query_auth::unsupported( + "a format table cannot apply a row filter or column masking", + )); + } + Ok(()) } async fn plan_inner(&self, trace: Option<&mut ScanTrace>) -> crate::Result { diff --git a/crates/paimon/src/table/incremental_scan.rs b/crates/paimon/src/table/incremental_scan.rs index 747278836..e9ac52b81 100644 --- a/crates/paimon/src/table/incremental_scan.rs +++ b/crates/paimon/src/table/incremental_scan.rs @@ -149,6 +149,18 @@ impl IncrementalPlan { &self.splits } + /// Whether any underlying split came from a query-auth plan. Unlike + /// [`Self::data_splits`] this sees the diff pairs too. + pub(crate) fn any_query_auth_required(&self) -> bool { + self.splits.iter().any(|split| match split { + IncrementalSplit::Data(split) => split.query_auth_required(), + IncrementalSplit::DiffPair { before, after } => before + .iter() + .chain(after) + .any(DataSplit::query_auth_required), + }) + } + pub fn data_splits(&self) -> Vec { self.splits .iter() @@ -244,7 +256,13 @@ impl<'a> IncrementalScan<'a> { } pub async fn plan(&self) -> crate::Result { - crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; + let core_options = crate::spec::CoreOptions::new(self.table.schema().options()); + core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?; + if self.table.server_query_auth_enabled().await? { + return Err(super::query_auth::unsupported( + "an incremental read cannot apply a row filter or column masking", + )); + } let mode = self.resolve_mode(); self.validate_snapshot_range(mode).await?; if self.start_exclusive == self.end_inclusive { diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 35f1e45a0..66d31ba74 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -81,6 +81,7 @@ mod postpone_fixed_bucket_router; mod postpone_fixed_bucket_write; mod postpone_fixed_bucket_write_builder; mod prepared_files; +mod query_auth; mod read_builder; pub mod referenced_files; pub(crate) mod rest_env; @@ -172,6 +173,9 @@ pub struct Table { schema_manager: SchemaManager, branch: String, branch_reference: bool, + /// Minted only by [`RESTEnv::build_table`], so a handle assembled with the + /// public [`Table::new`] cannot replay a grant. + query_auth_session: Option, rest_env: Option, /// True when this table copy was switched to a historical schema by /// [`Table::copy_with_time_travel`]. Such a copy is read-only. @@ -201,6 +205,7 @@ impl Table { schema_manager, branch, branch_reference: false, + query_auth_session: None, rest_env, time_traveled: false, travel_snapshot: None, @@ -241,6 +246,7 @@ impl Table { schema_manager, branch, branch_reference, + query_auth_session: None, rest_env: None, time_traveled: false, travel_snapshot: None, @@ -315,6 +321,97 @@ impl Table { } } + /// Whether the server says this table is `query-auth.enabled` right now: the + /// handle's schema is a snapshot, and a cached `false` would skip the check. + pub(crate) async fn server_query_auth_enabled(&self) -> Result { + let local = CoreOptions::new(self.schema.options()).query_auth_enabled(); + let Some(rest_env) = &self.rest_env else { + return Ok(local); + }; + // Only ever strengthens: the name can be re-created over this handle's + // files, so the answer may be about a different table. + if local { + return Ok(true); + } + match rest_env.current_table().await?.schema.as_ref() { + Some(schema) => Ok(CoreOptions::new(schema.options()).query_auth_enabled()), + None => Ok(true), + } + } + + /// Whether this user may read this table; `None` when it is not + /// `query-auth.enabled`. The caller stamps the grant onto its splits. + /// + /// `server_query_auth` is the caller's already-fetched + /// [`Self::server_query_auth_enabled`], so planning asks the server once. + pub(crate) async fn authorize_read( + &self, + server_query_auth: bool, + ) -> Result>> { + let local = CoreOptions::new(self.schema.options()); + // Ask the selector too: `copy_with_options` adds one without the flag. + let travels = local.try_time_travel_selector()?.is_some(); + // A `$branch_x` or `$files` handle authorizes against the decorated + // name while its managers read the base table's own files. + let decorated = self.identifier.branch_name()?.is_some() + || self.identifier.system_table_name()?.is_some(); + if (travels || self.time_traveled || self.branch_reference || decorated) + && local.query_auth_enabled() + { + return Err(query_auth::unsupported( + "a time-travelled or branch read authorizes against the table's current schema, \ + which is not the one it reads", + )); + } + + let Some(rest_env) = &self.rest_env else { + // Only a REST catalog can authorize. + return if local.query_auth_enabled() { + Err(query_auth::unsupported( + "it requires a REST catalog to authorize the query", + )) + } else { + Ok(None) + }; + }; + + // No freshness assertion yet — an ordinary table must not inherit one. + if !server_query_auth { + return Ok(None); + } + if travels || self.time_traveled || self.branch_reference || decorated { + return Err(query_auth::unsupported( + "a time-travelled or branch read authorizes against the table's current schema, \ + which is not the one it reads", + )); + } + + // Before any RPC: only the catalog mints a session, so a handle the + // caller assembled stops here whatever name or files it wears. + let session = self.query_auth_session.ok_or_else(|| { + query_auth::unsupported("this table handle was assembled rather than loaded") + })?; + + // Naming a system column here would fail the server's column check. + let response = rest_env + .table_query_auth(&self.branch, self.schema.id(), None) + .await?; + Ok(Some(std::sync::Arc::new(query_auth::QueryAuthGrant::new( + response, session, + )))) + } + + /// Handed out once per catalog-loaded table; wraps only after 2^64 loads. + pub(crate) fn with_query_auth_session(mut self) -> Self { + static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + self.query_auth_session = Some(NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)); + self + } + + pub(crate) fn query_auth_session(&self) -> Option { + self.query_auth_session + } + /// Get the REST environment, if this table was loaded from a REST catalog. pub fn rest_env(&self) -> Option<&RESTEnv> { self.rest_env.as_ref() @@ -414,6 +511,7 @@ impl Table { schema_manager: self.schema_manager.clone(), branch: self.branch.clone(), branch_reference: self.branch_reference, + query_auth_session: self.query_auth_session, rest_env: self.rest_env.clone(), time_traveled: self.time_traveled, travel_snapshot: if selector_changed { @@ -524,6 +622,7 @@ impl Table { schema_manager, branch, branch_reference: true, + query_auth_session: self.query_auth_session, rest_env: self.rest_env.clone(), time_traveled: false, travel_snapshot: None, @@ -558,6 +657,32 @@ pub(crate) fn find_field_id_by_name(fields: &[DataField], name: &str) -> Option< fields.iter().find(|f| f.name() == name).map(|f| f.id()) } +/// A `query-auth.enabled` table wired to its own REST session. +#[cfg(test)] +pub(crate) async fn rest_query_auth_table() -> Table { + use crate::api::rest_api::RESTApi; + use crate::common::{CatalogOptions, Options}; + + let mut options = Options::default(); + options.set(CatalogOptions::URI, "http://127.0.0.1:1"); + options.set("token.provider", "bear"); + options.set("token", "test_token"); + let api = std::sync::Arc::new(RESTApi::new(options.clone(), false).await.unwrap()); + let table = query_auth_table(); + Table { + rest_env: Some(RESTEnv::new( + table.identifier.clone(), + "uuid-1".to_string(), + api, + options, + false, + None, + )), + ..table + } + .with_query_auth_session() +} + /// A minimal table with `query-auth.enabled = true`, for the fail-closed read guard. #[cfg(test)] pub(crate) fn query_auth_table() -> Table { diff --git a/crates/paimon/src/table/query_auth.rs b/crates/paimon/src/table/query_auth.rs new file mode 100644 index 000000000..bc321618e --- /dev/null +++ b/crates/paimon/src/table/query_auth.rs @@ -0,0 +1,320 @@ +// 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. + +//! What the REST server authorized a user to read from one table. + +use crate::api::AuthTableQueryResponse; + +/// The server's answer for one user on one table, kept unparsed. +/// +/// `session` pins it to the handle that asked: `to_arrow` is public and the +/// response names neither table nor principal. Java binds nothing. Routing +/// options are unbound on purpose — sound only while unrestricted grants +/// authorize. +#[derive(Debug, PartialEq)] +pub(crate) struct QueryAuthGrant { + response: AuthTableQueryResponse, + session: u64, +} + +impl QueryAuthGrant { + pub(crate) fn new(response: AuthTableQueryResponse, session: u64) -> Self { + Self { response, session } + } + + /// The only case this client can serve. + pub(crate) fn is_unrestricted(&self) -> bool { + self.response.is_unrestricted() + } + + /// Travelled and branch views read a schema the server did not rule on. + /// Everything else follows from the session, which only the catalog mints. + pub(crate) fn matches_table(&self, table: &super::Table) -> bool { + !table.is_time_traveled() + && !table.is_branch_reference() + && table.query_auth_session() == Some(self.session) + } +} + +/// `value_stats` and `write_cols` are public on every split, and an older file +/// can name a since-dropped column the server never ruled on. Refused rather +/// than scrubbed: rewriting encoded stats is how bounds get mismatched. +pub(crate) async fn reject_unauthorized_stats( + plan: &super::Plan, + current: &crate::spec::TableSchema, + schemas: &super::schema_manager::SchemaManager, +) -> crate::Result<()> { + let refuse = |column: &str| { + Err(unsupported(&format!( + "a data file still carries statistics for '{column}', which the current schema — the \ + one the server authorized — does not have" + ))) + }; + let named = |name: &String| current.fields().iter().any(|f| f.name() == name); + let mut checked = std::collections::HashSet::new(); + for split in plan.splits() { + for file in split.data_files() { + for column in file + .value_stats_cols + .iter() + .chain(file.write_cols.iter()) + .flatten() + { + if !named(column) { + return refuse(column); + } + } + // The file's own schema is the authority: a name can be dropped and + // re-added under a new id, and the lists may be absent entirely. + if file.schema_id == current.id() || !checked.insert(file.schema_id) { + continue; + } + let older = schemas.schema(file.schema_id).await?; + if let Some(gone) = older.fields().iter().find(|f| { + !current + .fields() + .iter() + .any(|c| c.id() == f.id() && c.name() == f.name()) + }) { + return refuse(gone.name()); + } + } + } + Ok(()) +} + +/// A refusal naming the option, so callers never match on prose. +pub(crate) fn unsupported(reason: &str) -> crate::Error { + crate::Error::Unsupported { + message: format!( + "reading a table with 'query-auth.enabled' = true is not supported: {reason}" + ), + } +} + +/// Column permissions cover real schema fields, so the server can neither grant +/// nor refuse `_ROW_ID` and friends. +pub(crate) fn reject_system_columns<'a>( + names: impl IntoIterator, +) -> crate::Result<()> { + for name in names { + if crate::spec::is_reserved_system_field_name(name) { + return Err(unsupported(&format!( + "the system column '{name}' is not one the server can authorize: column \ + permissions are granted over table columns" + ))); + } + } + Ok(()) +} + +/// The read resolves older files by field id, so a non-canonical `(id, name)` +/// pair reads as something no grant covered. System fields have no entry. +pub(crate) fn reject_noncanonical_fields( + read_type: &[crate::spec::DataField], + schema_fields: &[crate::spec::DataField], +) -> crate::Result<()> { + for field in read_type { + if crate::spec::is_reserved_system_field_name(field.name()) { + continue; + } + let canonical = schema_fields + .iter() + .any(|f| f.id() == field.id() && f.name() == field.name()); + if !canonical { + return Err(unsupported(&format!( + "'{}' (field id {}) is not a column of the current schema, which is what the \ + server authorized", + field.name(), + field.id() + ))); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::reject_system_columns; + use crate::table::query_auth_table; + + #[tokio::test] + async fn test_a_grant_is_pinned_to_the_handle_that_obtained_it() { + let a = crate::table::rest_query_auth_table().await; + let b = crate::table::rest_query_auth_table().await; + let grant = super::QueryAuthGrant::new( + crate::api::AuthTableQueryResponse::default(), + a.query_auth_session().unwrap(), + ); + assert!(grant.matches_table(&a)); + assert!( + !grant.matches_table(&b), + "another handle — another principal or another table — must not reuse it" + ); + } + + #[tokio::test] + async fn test_a_time_travel_selector_alone_is_refused() { + for selector in [ + "scan.snapshot-id", + "scan.version", + "scan.tag-name", + "scan.timestamp-millis", + "scan.watermark", + ] { + let table = query_auth_table().copy_with_options(std::collections::HashMap::from([( + selector.to_string(), + "1".to_string(), + )])); + assert!(!table.is_time_traveled(), "{selector} sets no flag"); + let err = table.authorize_read(true).await.unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("time-travelled or branch read")), + "{selector}: {err:?}" + ); + } + } + + #[tokio::test] + async fn test_a_grant_does_not_cross_into_a_travelled_or_branch_view() { + let table = crate::table::rest_query_auth_table().await; + let grant = super::QueryAuthGrant::new( + crate::api::AuthTableQueryResponse::default(), + table.query_auth_session().unwrap(), + ); + assert!(grant.matches_table(&table)); + + let mut travelled = table.copy_with_options(std::collections::HashMap::new()); + travelled.time_traveled = true; + assert!( + !grant.matches_table(&travelled), + "an older schema is not the one the server ruled on" + ); + + let assembled = crate::table::Table::new( + table.file_io().clone(), + table.identifier().clone(), + "/tmp/somewhere-else".to_string(), + table.schema().clone(), + table.rest_env().cloned(), + ); + assert!( + !grant.matches_table(&assembled), + "an assembled handle must not replay a grant" + ); + + let mut branch = table.copy_with_options(std::collections::HashMap::new()); + branch.branch_reference = true; + assert!( + !grant.matches_table(&branch), + "a branch view is refused even when its schema id coincides" + ); + } + + #[tokio::test] + async fn test_stats_for_a_dropped_column_are_refused() { + let table = query_auth_table(); + let file = + |cols: Option>, written: Option>| crate::spec::DataFileMeta { + file_name: "f.parquet".to_string(), + file_size: 1, + row_count: 1, + min_key: Vec::new(), + max_key: Vec::new(), + key_stats: crate::spec::stats::BinaryTableStats::empty(), + value_stats: crate::spec::stats::BinaryTableStats::empty(), + min_sequence_number: 0, + max_sequence_number: 0, + schema_id: table.schema().id(), + level: 0, + extra_files: Vec::new(), + creation_time: None, + delete_row_count: Some(0), + embedded_index: None, + file_source: None, + value_stats_cols: cols.map(|c| c.iter().map(|s| s.to_string()).collect()), + external_path: None, + first_row_id: None, + write_cols: written.map(|c| c.iter().map(|s| s.to_string()).collect()), + column_max_sequence_numbers: None, + }; + let plan_of = |meta| { + crate::table::Plan::new(vec![crate::table::DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(crate::spec::BinaryRowBuilder::new(0).build()) + .with_bucket(0) + .with_bucket_path("p".to_string()) + .with_total_buckets(1) + .with_data_files(vec![meta]) + .with_raw_convertible(false) + .build() + .unwrap()]) + }; + let schemas = table.schema_manager(); + + for meta in [ + file(Some(vec!["id", "gone"]), None), + file(None, Some(vec!["id", "gone"])), + file(Some(vec!["id"]), Some(vec!["id", "gone"])), + ] { + let err = super::reject_unauthorized_stats(&plan_of(meta), table.schema(), schemas) + .await + .unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("statistics for 'gone'")), + "{err:?}" + ); + } + + assert!(super::reject_unauthorized_stats( + &plan_of(file(Some(vec!["id"]), Some(vec!["id"]))), + table.schema(), + schemas + ) + .await + .is_ok()); + } + + #[test] + fn test_a_system_column_read_is_refused() { + let err = reject_system_columns(["id", crate::spec::ROW_ID_FIELD_NAME]).unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("system column '_ROW_ID'")), + "{err:?}" + ); + assert!(reject_system_columns(["id", "name"]).is_ok()); + } + + #[tokio::test] + async fn test_time_travelled_or_branch_read_is_refused() { + let mut travelled = query_auth_table(); + travelled.time_traveled = true; + let err = travelled.authorize_read(true).await.unwrap_err(); + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("time-travelled or branch read")), + "got {err:?}" + ); + + let mut branch = query_auth_table(); + branch.branch_reference = true; + assert!(branch.authorize_read(true).await.is_err()); + } +} diff --git a/crates/paimon/src/table/read_builder.rs b/crates/paimon/src/table/read_builder.rs index ec8ef966e..2e0eb638c 100644 --- a/crates/paimon/src/table/read_builder.rs +++ b/crates/paimon/src/table/read_builder.rs @@ -501,10 +501,12 @@ impl<'a> PaimonReadBuilder<'a> { /// Create a table read for consuming splits (e.g. from a scan plan). pub fn new_read(&self) -> Result> { - // Fail closed at read construction so bindings that short-circuit before - // `to_arrow` (e.g. an empty-splits fast path) can't bypass the guard. - let core_options = self.table.schema.core_options(); - core_options.ensure_read_authorized()?; + // Stays here: a table's declared type is known without a grant. Only + // query-auth moved to `to_arrow`, where the split's grant is visible. + self.table + .schema + .core_options() + .ensure_type_paimon_served(&self.table.identifier().full_name())?; let read_type = match self.resolve_read_type()? { None => self.table.schema.fields().to_vec(), Some(fields) => fields, @@ -946,14 +948,30 @@ mod tests { #[test] fn test_read_fails_closed_when_query_auth_enabled() { let table = query_auth_table(); - // `new_read` fails closed, so bindings that short-circuit before `to_arrow` can't bypass. - let err = table.new_read_builder().new_read().unwrap_err(); + let read = table.new_read_builder().new_read().unwrap(); + let err = ungranted_read_error(&read); assert!( matches!(err, crate::Error::Unsupported { ref message } if message.contains("query-auth.enabled")), - "building a read for a query-auth.enabled table must fail closed" + "reading a query-auth.enabled table without a grant must fail closed" ); } + fn ungranted_read_error(read: &crate::table::TableRead<'_>) -> crate::Error { + let split = DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path("memory:/t/bucket-0".to_string()) + .with_total_buckets(1) + .with_data_files(Vec::new()) + .build() + .unwrap(); + match read.to_arrow(&[split]) { + Ok(_) => panic!("reading without a grant must fail closed"), + Err(err) => err, + } + } + #[test] fn test_dynamic_option_cannot_disable_query_auth() { // Copying the table with the option off must not weaken a stored `true`. @@ -961,7 +979,8 @@ mod tests { "query-auth.enabled".to_string(), "false".to_string(), )])); - let err = table.new_read_builder().new_read().unwrap_err(); + let read = table.new_read_builder().new_read().unwrap(); + let err = ungranted_read_error(&read); assert!( matches!(err, crate::Error::Unsupported { ref message } if message.contains("query-auth.enabled")), "a dynamic override must not disable query-auth" diff --git a/crates/paimon/src/table/rest_env.rs b/crates/paimon/src/table/rest_env.rs index 133acf698..49f9270d7 100644 --- a/crates/paimon/src/table/rest_env.rs +++ b/crates/paimon/src/table/rest_env.rs @@ -19,6 +19,7 @@ use crate::api::rest_api::RESTApi; use crate::api::rest_error::RestError; +use crate::api::GetTableResponse; use crate::catalog::{Identifier, RESTTokenFileIO}; use crate::common::Options; use crate::error::Error; @@ -81,6 +82,80 @@ impl RESTEnv { &self.api } + /// Bracketed by a freshness check: the response names no table, so a drop + /// and re-create in between would let a replacement's grant serve this one. + pub(crate) async fn table_query_auth( + &self, + branch: &str, + schema_id: i64, + select: Option>, + ) -> Result { + self.current_table_checked(schema_id).await?; + let response = self + .api + .auth_table_query(&self.branch_identifier(branch), select) + .await?; + self.current_table_checked(schema_id).await?; + Ok(response) + } + + /// Asserts nothing about identity: an ordinary table must not inherit a + /// freshness restriction. + pub(crate) async fn current_table(&self) -> Result { + self.api.get_table(&self.identifier).await + } + + /// Refused unless the name still resolves to the loaded table — a missing + /// identity too, which checks nothing. + pub(crate) async fn current_table_checked(&self, schema_id: i64) -> Result { + let response = self.current_table().await?; + let name = self.identifier.full_name(); + let drifted = |what: &str, from: String, to: String| crate::Error::DataInvalid { + message: format!( + "table '{name}' now resolves to {what} {to}, not the {from} this handle was \ + loaded with; re-load the table before reading it" + ), + source: None, + }; + match response.id.as_deref() { + Some(uuid) if uuid == self.uuid => {} + Some(uuid) => return Err(drifted("uuid", self.uuid.clone(), uuid.to_string())), + None => { + return Err(drifted( + "uuid", + self.uuid.clone(), + "nothing the server reports".to_string(), + )) + } + } + match response.schema_id { + Some(id) if id == schema_id => Ok(response), + Some(id) => Err(drifted("schema", schema_id.to_string(), id.to_string())), + None => Err(drifted( + "schema", + schema_id.to_string(), + "nothing the server reports".to_string(), + )), + } + } + + /// `db.table$branch_`, as Java names a branch. Only the auth call uses it. + fn branch_identifier(&self, branch: &str) -> Identifier { + if branch == crate::catalog::DEFAULT_MAIN_BRANCH { + return self.identifier.clone(); + } + Identifier::new( + self.identifier.database(), + format!( + "{}{}{}{}", + self.identifier.object(), + crate::catalog::SYSTEM_TABLE_SPLITTER, + crate::catalog::SYSTEM_BRANCH_PREFIX, + branch + ), + ) + } + /// Get the table identifier. pub fn identifier(&self) -> &Identifier { &self.identifier @@ -218,7 +293,8 @@ impl RESTEnv { table_path, table_schema, Some(rest_env), - )) + ) + .with_query_auth_session()) } pub(crate) async fn build_object_table( diff --git a/crates/paimon/src/table/source.rs b/crates/paimon/src/table/source.rs index aaaf66cfc..2ac265fe1 100644 --- a/crates/paimon/src/table/source.rs +++ b/crates/paimon/src/table/source.rs @@ -20,6 +20,7 @@ //! Reference: [org.apache.paimon.table.source](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/). use crate::spec::{BinaryRow, DataFileMeta, DataFileMetaRowLayout}; +use crate::table::query_auth::QueryAuthGrant; use crate::table::stats_filter::group_by_overlapping_row_id; use serde::{Deserialize, Serialize}; use std::sync::Arc; @@ -500,9 +501,33 @@ pub struct DataSplit { /// physical rows are exactly its logical rows (modulo deletion files). /// Mirrors Java `DataSplit#rawConvertible`. raw_convertible: bool, + /// Mirrors Java `QueryAuthSplit`, but is dropped by serialization, so a + /// plan must be read where it was made. + #[serde(skip)] + query_auth_grant: Option>, + /// That this split came from a `query-auth.enabled` table. Unlike the grant + /// it survives serialization, so a round-tripped split fails closed. + #[serde(default)] + query_auth_required: bool, } impl DataSplit { + /// Marks the split as needing authorization whether or not a grant came + /// with it, so a plan that produced none still refuses at the read. + pub(crate) fn with_query_auth_grant(mut self, grant: Option>) -> Self { + self.query_auth_required = true; + self.query_auth_grant = grant; + self + } + + pub(crate) fn query_auth_required(&self) -> bool { + self.query_auth_required + } + + pub(crate) fn query_auth_grant(&self) -> Option<&Arc> { + self.query_auth_grant.as_ref() + } + pub fn snapshot_id(&self) -> i64 { self.snapshot_id } @@ -683,10 +708,23 @@ impl DataSplit { DataSplitBuilder::new() } + /// The Java-compatible frames have no field for the marker, so a reader would + /// rebuild the split without it. Serde keeps it; only these two must refuse. + fn ensure_serializable_without_grant(&self) -> crate::Result<()> { + if self.query_auth_required { + return Err(crate::table::query_auth::unsupported( + "a split of such a table cannot be serialized to the cross-language \ + format, which has no field to carry the authorization with it", + )); + } + Ok(()) + } + /// Serialize the DataSplit fields to Java `DataSplit#serialize` (version 9) binary. /// Byte-compatible with `compatibility/datasplit-v9`. Row ranges are not part of the /// format; `serialize_split_v1` wraps a row-range split as an `IndexedSplit` instead. pub fn serialize(&self) -> crate::Result> { + self.ensure_serializable_without_grant()?; let mut out = Vec::new(); out.extend_from_slice(&SPLIT_MAGIC.to_be_bytes()); out.extend_from_slice(&SPLIT_VERSION.to_be_bytes()); @@ -848,6 +886,7 @@ impl DataSplit { /// `IndexedSplit` (type 3) wrapping the DataSplit body plus the ranges. Byte-compatible with /// `compatibility/split-v1-data` / `split-v1-indexed`. pub fn serialize_split_v1(&self) -> crate::Result> { + self.ensure_serializable_without_grant()?; let mut out = Vec::new(); out.extend_from_slice(&SPLIT_SER_MAGIC.to_be_bytes()); out.extend_from_slice(&SPLIT_SER_VERSION.to_be_bytes()); @@ -1274,6 +1313,8 @@ impl DataSplitBuilder { } } Ok(DataSplit { + query_auth_grant: None, + query_auth_required: false, snapshot_id: self.snapshot_id, partition: Arc::new(partition), bucket: self.bucket, @@ -1311,6 +1352,17 @@ impl Plan { &self.splits } + /// Stamp the grant every split of this plan was authorized under. + pub(crate) fn with_query_auth_grant(mut self, grant: Option>) -> Self { + if grant.is_some() { + self.splits = std::mem::take(&mut self.splits) + .into_iter() + .map(|split| split.with_query_auth_grant(grant.clone())) + .collect(); + } + self + } + /// Sum of data-file bytes referenced by this plan. /// /// Negative file sizes are treated as unknown and do not contribute. The @@ -1984,6 +2036,31 @@ mod tests { } // Same hardening for the IndexedSplit row-ranges count in the SPLIT_V1 frame. + #[test] + fn test_a_marked_split_refuses_the_cross_language_formats() { + let split = DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(crate::spec::BinaryRowBuilder::new(0).build()) + .with_bucket(0) + .with_bucket_path("p".to_string()) + .with_total_buckets(1) + .with_data_files(vec![]) + .with_raw_convertible(false) + .build() + .unwrap() + .with_query_auth_grant(None); + for bytes in [split.serialize(), split.serialize_split_v1()] { + let Err(err) = bytes else { + panic!("the marker has nowhere to go in these formats") + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("query-auth.enabled")), + "{err:?}" + ); + } + } + #[test] fn deserialize_split_v1_rejects_huge_ranges_count_without_aborting() { let split = DataSplitBuilder::new() diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index 99c01f7d9..8750aaf09 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -183,7 +183,7 @@ impl<'a> TableRead<'a> { &self, plan: &IncrementalPlan, ) -> crate::Result { - self.ensure_query_auth_allowed()?; + self.ensure_query_auth_allowed(plan)?; plan.validate()?; match &self.0 { TableReadKind::Paimon(read) => read.to_incremental_arrow(plan), @@ -203,7 +203,7 @@ impl<'a> TableRead<'a> { &self, plan: &IncrementalPlan, ) -> crate::Result { - self.ensure_query_auth_allowed()?; + self.ensure_query_auth_allowed(plan)?; plan.validate()?; match &self.0 { TableReadKind::Paimon(read) => read.to_audit_log_arrow(plan), @@ -213,8 +213,33 @@ impl<'a> TableRead<'a> { } } - fn ensure_query_auth_allowed(&self) -> crate::Result<()> { - CoreOptions::new(self.table().schema().options()).ensure_read_authorized() + /// Sync, so the split's marker stands in for asking the server. + fn ensure_query_auth_allowed(&self, plan: &IncrementalPlan) -> crate::Result<()> { + let core_options = CoreOptions::new(self.table().schema().options()); + core_options.ensure_type_paimon_served(&self.table().identifier().full_name())?; + if core_options.query_auth_enabled() || plan.any_query_auth_required() { + return Err(super::query_auth::unsupported( + "an incremental read cannot apply a row filter or column masking", + )); + } + Ok(()) + } +} + +/// Every leaf's column name. Unlike the index-based walks this sees system +/// columns, whose leaf index is only a placeholder. +fn collect_leaf_column_names(predicate: &Predicate, out: &mut std::collections::HashSet) { + match predicate { + Predicate::Leaf { column, .. } => { + out.insert(column.clone()); + } + Predicate::And(children) | Predicate::Or(children) => { + children + .iter() + .for_each(|child| collect_leaf_column_names(child, out)); + } + Predicate::Not(inner) => collect_leaf_column_names(inner, out), + Predicate::AlwaysTrue | Predicate::AlwaysFalse => {} } } @@ -712,12 +737,68 @@ impl<'a> PaimonTableRead<'a> { reader.read(splits) } + /// Allowed only if the splits carry a grant saying the server imposed + /// nothing. Never fetched here, so a split without one fails closed. + fn ensure_authorized_by_splits( + &self, + core_options: &CoreOptions, + data_splits: &[DataSplit], + ) -> crate::Result<()> { + // Unconditional: unrelated to query-auth. + core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?; + // The handle's flag is a snapshot; the marker survives a round-trip. + let required = core_options.query_auth_enabled() + || data_splits.iter().any(|s| s.query_auth_required()); + if !required { + return Ok(()); + } + // The read's own scope: a caller can plan clean, then read differently. + let mut filter_columns = std::collections::HashSet::new(); + for predicate in &self.data_predicates { + collect_leaf_column_names(predicate, &mut filter_columns); + } + super::query_auth::reject_system_columns( + self.read_type + .iter() + .map(|f| f.name()) + .chain(filter_columns.iter().map(String::as_str)), + )?; + // By id AND name: older files resolve by id, so a dropped field passed + // through the public `with_read_type` returns an uncovered column. + super::query_auth::reject_noncanonical_fields( + &self.read_type, + self.table.schema().fields(), + )?; + // Per split, as Java binds one `QueryAuthSplit` each: lists get + // concatenated and the first grant must not cover the rest. + for split in data_splits { + let Some(grant) = split.query_auth_grant() else { + return Err(super::query_auth::unsupported( + "the split carries no authorization; it was built directly, or serialized, \ + which drops the grant — re-plan the scan", + )); + }; + if !grant.matches_table(self.table) { + return Err(super::query_auth::unsupported( + "the grant was issued for a different table, schema or session; re-plan the \ + scan", + )); + } + if !grant.is_unrestricted() { + return Err(super::query_auth::unsupported( + "this client cannot apply a row filter or column masking, so it refuses \ + rather than return unfiltered rows", + )); + } + } + Ok(()) + } + /// Returns an [`ArrowRecordBatchStream`]. pub fn to_arrow(&self, data_splits: &[DataSplit]) -> crate::Result { let has_primary_keys = !self.table.schema.primary_keys().is_empty(); let core_options = self.table.schema.core_options(); - // Fail closed for a direct `TableRead` (bypassing `ReadBuilder::new_read`). - core_options.ensure_read_authorized()?; + self.ensure_authorized_by_splits(&core_options, data_splits)?; let merge_engine = core_options.merge_engine()?; // Route supported PK merge engines through the split-aware reader. @@ -1583,15 +1664,287 @@ mod tests { )); } + #[test] + fn test_incremental_and_audit_log_reads_refuse_a_query_auth_table() { + let table = query_auth_table(); + let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let plan = IncrementalPlan::new(IncrementalScanMode::Delta, Vec::new()); + for err in [ + read.to_incremental_arrow(&plan).err(), + read.to_audit_log_arrow(&plan).err(), + ] { + let err = err.expect("both must refuse a query-auth.enabled table"); + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("query-auth.enabled")), + "got {err:?}" + ); + } + } + + fn stale_handle(name: &str, options: &[(&str, &str)]) -> Table { + let mut builder = crate::spec::Schema::builder().column( + "id", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ); + for (key, value) in options { + builder = builder.option(*key, *value); + } + Table::new( + FileIOBuilder::new("file").build().unwrap(), + Identifier::new("default", name), + format!("/tmp/test-{name}"), + crate::spec::TableSchema::new(0, &builder.build().unwrap()), + None, + ) + } + + #[test] + fn test_a_marked_split_refuses_the_format_and_incremental_reads() { + let stamped = split_with_grant(None); + + let format = stale_handle( + "fmt", + &[("type", "format-table"), ("file.format", "parquet")], + ); + let read = + TableRead::new_format(&format, format.schema().fields().to_vec(), Vec::new(), None); + let Err(err) = read.to_arrow(std::slice::from_ref(&stamped)) else { + panic!("a marked split must refuse a format read") + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("query-auth.enabled")), + "{err:?}" + ); + + let paimon = stale_handle("inc", &[]); + let read = TableRead::new(&paimon, paimon.schema().fields().to_vec(), Vec::new()); + let plan = IncrementalPlan::new( + IncrementalScanMode::Delta, + vec![IncrementalSplit::Data(stamped)], + ); + let Err(err) = read.to_incremental_arrow(&plan) else { + panic!("a marked split must refuse an incremental read") + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("query-auth.enabled")), + "{err:?}" + ); + } + + fn split_with_grant( + grant: Option, + ) -> crate::table::DataSplit { + crate::table::DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(crate::spec::BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path("memory:/t/bucket-0".to_string()) + .with_total_buckets(1) + .with_data_files(Vec::new()) + .build() + .unwrap() + .with_query_auth_grant(grant.map(std::sync::Arc::new)) + } + + fn grant_for(table: &Table, restricted: bool) -> crate::table::query_auth::QueryAuthGrant { + crate::table::query_auth::QueryAuthGrant::new( + crate::api::AuthTableQueryResponse { + filter: restricted.then(|| vec!["{}".to_string()]), + column_masking: None, + }, + table + .query_auth_session() + .expect("a catalog-loaded table has a session"), + ) + } + + #[tokio::test] + async fn test_one_unrestricted_grant_does_not_cover_the_other_splits() { + let table = crate::table::rest_query_auth_table().await; + let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let allowed = split_with_grant(Some(grant_for(&table, false))); + for other in [ + split_with_grant(Some(grant_for(&table, true))), + split_with_grant(None), + ] { + assert!( + read.to_arrow(&[allowed.clone(), other]).is_err(), + "every split must be authorized on its own" + ); + } + } + + #[test] + fn test_engine_served_table_is_refused_at_the_read_boundary() { + let schema = crate::spec::Schema::builder() + .column( + "id", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .option("type", "iceberg-table") + .build() + .unwrap(); + let table = Table::new( + crate::io::FileIOBuilder::new("file").build().unwrap(), + crate::catalog::Identifier::new("default", "ice_t"), + "/tmp/test-engine-served-read".to_string(), + crate::spec::TableSchema::new(0, &schema), + None, + ); + let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let err = match read.to_arrow(&[split_with_grant(None)]) { + Ok(_) => panic!("an engine-served table must not be read as Paimon"), + Err(err) => err, + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("cannot be served as a Paimon table")), + "got {err:?}" + ); + } + + #[tokio::test] + async fn test_a_row_id_filter_configured_on_the_read_is_refused() { + let table = crate::table::rest_query_auth_table().await; + let row_id = Predicate::Leaf { + index: 0, + column: crate::spec::ROW_ID_FIELD_NAME.to_string(), + data_type: crate::spec::DataType::BigInt(crate::spec::BigIntType::new()), + op: crate::spec::PredicateOperator::GtEq, + literals: vec![crate::spec::Datum::Long(1)], + }; + let read = TableRead::new(&table, table.schema.fields().to_vec(), vec![row_id]); + let split = split_with_grant(Some(grant_for(&table, false))); + let err = match read.to_arrow(&[split]) { + Ok(_) => panic!("a system-column filter must be refused"), + Err(err) => err, + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("system column '_ROW_ID'")), + "got {err:?}" + ); + } + + #[tokio::test] + async fn test_a_field_outside_the_current_schema_is_refused() { + let table = crate::table::rest_query_auth_table().await; + let dropped = crate::spec::DataField::new( + 999, + "dropped".to_string(), + crate::spec::DataType::Int(crate::spec::IntType::new()), + ); + let read = TableRead::new(&table, vec![dropped], Vec::new()); + let split = split_with_grant(Some(grant_for(&table, false))); + let err = match read.to_arrow(&[split]) { + Ok(_) => panic!("a field outside the current schema must be refused"), + Err(err) => err, + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("not a column of the current schema")), + "got {err:?}" + ); + } + + #[test] + fn test_a_serialized_split_still_demands_authorization() { + let stamped = split_with_grant(None); + let bytes = serde_json::to_vec(&stamped).unwrap(); + let restored: crate::table::DataSplit = serde_json::from_slice(&bytes).unwrap(); + assert!( + restored.query_auth_grant().is_none(), + "the grant is dropped" + ); + assert!( + restored.query_auth_required(), + "but the demand for authorization survives" + ); + + let stale = Table::new( + crate::io::FileIOBuilder::new("file").build().unwrap(), + crate::catalog::Identifier::new("default", "stale"), + "/tmp/test-stale-handle".to_string(), + crate::spec::TableSchema::new( + 0, + &crate::spec::Schema::builder() + .column( + "id", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .build() + .unwrap(), + ), + None, + ); + assert!(!stale.schema().core_options().query_auth_enabled()); + let read = TableRead::new(&stale, stale.schema().fields().to_vec(), Vec::new()); + assert!( + read.to_arrow(&[restored]).is_err(), + "a round-tripped split must fail closed even on a handle that predates the option" + ); + } + + #[tokio::test] + async fn test_a_grant_from_another_handle_refuses_the_read() { + let table = crate::table::rest_query_auth_table().await; + let other = crate::table::rest_query_auth_table().await; + let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let err = match read.to_arrow(&[split_with_grant(Some(grant_for(&other, false)))]) { + Ok(_) => panic!("a grant obtained elsewhere must not authorize this read"), + Err(err) => err, + }; + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("different table, schema or session")), + "got {err:?}" + ); + } + + #[tokio::test] + async fn test_restricted_grant_on_a_split_refuses_the_read() { + let table = crate::table::rest_query_auth_table().await; + let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let split = split_with_grant(Some(grant_for(&table, true))); + assert!( + matches!( + read.to_arrow(&[split]), + Err(crate::Error::Unsupported { ref message }) if message.contains("query-auth.enabled") + ), + "a row filter this client cannot apply must refuse the read" + ); + } + + #[tokio::test] + async fn test_unrestricted_grant_on_a_split_allows_the_read() { + let table = crate::table::rest_query_auth_table().await; + let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let split = split_with_grant(Some(grant_for(&table, false))); + assert!( + read.to_arrow(&[split]).is_ok(), + "an unrestricted grant must let the read through" + ); + } + #[test] fn test_direct_table_read_fails_closed_when_query_auth_enabled() { let table = query_auth_table(); - // Bypass `ReadBuilder` by constructing `TableRead` directly; the `to_arrow` guard - // still fails closed. let read = TableRead::new(&table, table.schema.fields().to_vec(), Vec::new()); + let split = crate::table::DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(crate::spec::BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path("memory:/t/bucket-0".to_string()) + .with_total_buckets(1) + .with_data_files(Vec::new()) + .build() + .unwrap(); assert!( matches!( - read.to_arrow(&[]), + read.to_arrow(&[split]), Err(crate::Error::Unsupported { ref message }) if message.contains("query-auth.enabled") ), "directly-constructed read of a query-auth.enabled table must fail closed" diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 601eeae90..a91c5b48c 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -1071,43 +1071,111 @@ impl<'a> PaimonTableScan<'a> { /// for `scan.version`; the strict selectors mirror Java's typed /// `scan.snapshot-id` / `scan.tag-name` handling. pub async fn plan(&self) -> crate::Result { - self.ensure_query_auth_allowed()?; + let grant = self.authorize_query().await?; let data_evolution_read_field_ids = self.projected_read_field_ids()?; - let snapshot = match super::time_travel::resolve_snapshot(self.table).await? { - Some(snapshot) => snapshot, - None => return Ok(Plan::new(Vec::new())), + let plan = match super::time_travel::resolve_snapshot(self.table).await? { + Some(snapshot) => { + self.plan_snapshot(snapshot, data_evolution_read_field_ids.as_ref(), None) + .await? + } + None => Plan::new(Vec::new()), }; - self.plan_snapshot(snapshot, data_evolution_read_field_ids.as_ref(), None) - .await + self.check_planned_files(&plan, grant.is_some()).await?; + Ok(plan.with_query_auth_grant(grant)) } /// Plan the full scan and return metadata-pruning trace counters. pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> { - self.ensure_query_auth_allowed()?; + let grant = self.authorize_query().await?; let mut trace = ScanTrace { limit: self.limit, ..Default::default() }; let data_evolution_read_field_ids = self.projected_read_field_ids()?; - let snapshot = match super::time_travel::resolve_snapshot(self.table).await? { - Some(snapshot) => snapshot, - None => return Ok((Plan::new(Vec::new()), trace)), + let plan = match super::time_travel::resolve_snapshot(self.table).await? { + Some(snapshot) => { + trace.snapshot_id = Some(snapshot.id()); + let plan = self + .plan_snapshot( + snapshot, + data_evolution_read_field_ids.as_ref(), + Some(&mut trace), + ) + .await?; + trace.planned_data_file_bytes = plan.planned_data_file_bytes(); + plan + } + None => Plan::new(Vec::new()), }; - trace.snapshot_id = Some(snapshot.id()); - let plan = self - .plan_snapshot( - snapshot, - data_evolution_read_field_ids.as_ref(), - Some(&mut trace), - ) - .await?; - trace.planned_data_file_bytes = plan.planned_data_file_bytes(); - Ok((plan, trace)) + self.check_planned_files(&plan, grant.is_some()).await?; + Ok((plan.with_query_auth_grant(grant), trace)) + } + + /// The grant was issued before the manifests were read, so the table can have + /// been dropped and re-created at the same path in between; the files just + /// planned would then be the replacement's. Closes that window and refuses a + /// plan whose files carry statistics the current schema no longer covers. + async fn check_planned_files(&self, plan: &Plan, query_auth: bool) -> crate::Result<()> { + if !query_auth { + return Ok(()); + } + if let Some(rest_env) = self.table.rest_env() { + rest_env + .current_table_checked(self.table.schema().id()) + .await?; + } + super::query_auth::reject_unauthorized_stats( + plan, + self.table.schema(), + self.table.schema_manager(), + ) + .await + } + + /// Authorize this scan and return the grant for the caller to stamp. + async fn authorize_query( + &self, + ) -> crate::Result>> { + let core_options = CoreOptions::new(self.table.schema().options()); + // Unconditional: unrelated to query-auth. + core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?; + + // File paths and stats, not table columns: the endpoint cannot rule on them. + let query_auth = self.table.server_query_auth_enabled().await?; + if self.scan_all_files { + return if query_auth { + Err(super::query_auth::unsupported( + "`$files` and friends are file paths and stats, not table columns, so the \ + auth endpoint can never rule on them", + )) + } else { + Ok(None) + }; + } + // A predicate or a row-range slice reads `_ROW_ID` unprojected. + let touches_row_id = self.row_ranges.is_some() + || self + .data_predicates + .iter() + .any(super::row_id_predicate::references_row_id); + if query_auth && touches_row_id { + super::query_auth::reject_system_columns([ROW_ID_FIELD_NAME])?; + } + + let grant = self.table.authorize_read(query_auth).await?; + // A plan carries row counts and bounds that answer COUNT/MIN/MAX without + // reading a row. + if grant.as_ref().is_some_and(|g| !g.is_unrestricted()) { + return Err(super::query_auth::unsupported( + "a plan already carries file paths, row counts and column bounds that a row \ + filter or column masking must not expose", + )); + } + Ok(grant) } - /// Fail closed for a `query-auth.enabled` table: scan planning — including - /// `with_scan_all_files`, which read-facing system tables like `files` use — - /// exposes file paths, row counts, and stats the client can't authorize. + /// Fail closed on planning paths that do not authorize, including + /// `with_scan_all_files`: it exposes stats the client cannot check. fn ensure_query_auth_allowed(&self) -> crate::Result<()> { CoreOptions::new(self.table.schema().options()).ensure_read_authorized() } @@ -2145,6 +2213,36 @@ mod tests { use chrono::{DateTime, Utc}; use std::collections::{HashMap, HashSet}; + #[tokio::test] + async fn test_engine_served_table_is_refused_at_plan() { + let schema = crate::spec::Schema::builder() + .column( + "id", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .option("type", "iceberg-table") + .build() + .unwrap(); + let table = Table::new( + crate::io::FileIOBuilder::new("file").build().unwrap(), + crate::catalog::Identifier::new("default", "ice_t"), + "/tmp/test-engine-served".to_string(), + crate::spec::TableSchema::new(0, &schema), + None, + ); + let err = table + .new_read_builder() + .new_scan() + .plan() + .await + .expect_err("an engine-served table must not plan as Paimon"); + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("cannot be served as a Paimon table")), + "got {err:?}" + ); + } + /// Helper to build a DataFileMeta with data evolution fields. fn make_evo_file( name: &str, diff --git a/crates/paimon/tests/mock_server.rs b/crates/paimon/tests/mock_server.rs index 6c8a0048a..24d4e0c48 100644 --- a/crates/paimon/tests/mock_server.rs +++ b/crates/paimon/tests/mock_server.rs @@ -34,8 +34,8 @@ use std::sync::{Arc, Mutex}; use tokio::task::JoinHandle; use paimon::api::{ - AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, ConfigResponse, - CreateFunctionRequest, CreateViewRequest, ErrorResponse, GetDatabaseResponse, + AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, AuthTableQueryResponse, + ConfigResponse, CreateFunctionRequest, CreateViewRequest, ErrorResponse, GetDatabaseResponse, GetFunctionResponse, GetTableResponse, GetViewResponse, ListDatabasesResponse, ListFunctionsResponse, ListTablesResponse, ListViewsResponse, RenameTableRequest, ResourcePaths, @@ -53,6 +53,10 @@ struct MockState { list_page_size: Option, no_permission_databases: HashSet, no_permission_tables: HashSet, + auth_responses: HashMap, + column_auth: HashMap>, + uuid_after_auth: HashMap, + uuid_after_calls: HashMap, /// ECS metadata role name (for token loader testing) ecs_role_name: Option, /// ECS metadata token (for token loader testing) @@ -82,6 +86,7 @@ pub struct RESTServer { warehouse: String, _data_path: String, config: ConfigResponse, + get_table_calls: Arc, inner: Arc>, resource_paths: ResourcePaths, addr: Option, @@ -118,6 +123,7 @@ impl RESTServer { _data_path, config, warehouse, + get_table_calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)), inner: Arc::new(Mutex::new(MockState { databases, ..Default::default() @@ -669,7 +675,10 @@ impl RESTServer { Path((db, table)): Path<(String, String)>, Extension(state): Extension>, ) -> impl IntoResponse { - let s = state.inner.lock().unwrap(); + state + .get_table_calls + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let mut s = state.inner.lock().unwrap(); let key = format!("{db}.{table}"); if s.no_permission_tables.contains(&key) { @@ -682,6 +691,18 @@ impl RESTServer { return (StatusCode::FORBIDDEN, Json(err)).into_response(); } + if let Some((uuid, remaining)) = s.uuid_after_calls.get_mut(&key) { + if *remaining == 0 { + let uuid = uuid.clone(); + s.uuid_after_calls.remove(&key); + if let Some(t) = s.tables.get_mut(&key) { + t.id = Some(uuid); + } + } else { + *remaining -= 1; + } + } + if let Some(response) = s.tables.get(&key) { return (StatusCode::OK, Json(response.clone())).into_response(); } @@ -705,6 +726,78 @@ impl RESTServer { (StatusCode::NOT_FOUND, Json(err)).into_response() } + pub async fn auth_table_query( + Path((db, table)): Path<(String, String)>, + Extension(state): Extension>, + Json(request): Json, + ) -> impl IntoResponse { + let s = state.inner.lock().unwrap(); + let key = format!("{db}.{table}"); + + // Mirrors the reference server: a null select means the real schema + // fields, and any column outside the grant denies the query. + if let Some(allowed) = s.column_auth.get(&key) { + let requested = request.select.clone().unwrap_or_else(|| { + s.tables + .get(&key) + .and_then(|t| t.schema.as_ref()) + .map(|schema| { + schema + .fields() + .iter() + .map(|f| f.name().to_string()) + .collect() + }) + .unwrap_or_default() + }); + if let Some(denied) = requested.iter().find(|c| !allowed.contains(c)) { + return ( + StatusCode::FORBIDDEN, + Json(ErrorResponse::new( + Some("table".to_string()), + Some(denied.clone()), + Some(format!("no permission for column '{denied}'")), + Some(403), + )), + ) + .into_response(); + } + } + + let response = s.auth_responses.get(&key).cloned().unwrap_or_default(); + drop(s); + let mut s = state.inner.lock().unwrap(); + if let Some(uuid) = s.uuid_after_auth.remove(&key) { + if let Some(existing) = s.tables.get_mut(&key) { + existing.id = Some(uuid); + } + } + (StatusCode::OK, Json(response)).into_response() + } + + pub fn set_table_uuid_after_calls( + &self, + database: &str, + table: &str, + uuid: &str, + after: usize, + ) { + let mut s = self.inner.lock().unwrap(); + s.uuid_after_calls + .insert(format!("{database}.{table}"), (uuid.to_string(), after)); + } + + pub fn set_table_uuid_after_auth(&self, database: &str, table: &str, uuid: &str) { + let mut s = self.inner.lock().unwrap(); + s.uuid_after_auth + .insert(format!("{database}.{table}"), uuid.to_string()); + } + + pub fn set_column_auth(&self, database: &str, table: &str, columns: Vec) { + let mut s = self.inner.lock().unwrap(); + s.column_auth.insert(format!("{database}.{table}"), columns); + } + /// Handle DELETE /databases/:db/tables/:table - drop a table. pub async fn drop_table( Path((db, table)): Path<(String, String)>, @@ -967,6 +1060,48 @@ impl RESTServer { ); } + #[allow(dead_code)] + pub fn get_table_calls(&self) -> usize { + self.get_table_calls + .load(std::sync::atomic::Ordering::Relaxed) + } + + pub fn clear_table_identity(&self, database: &str, table: &str) { + let mut s = self.inner.lock().unwrap(); + if let Some(existing) = s.tables.get_mut(&format!("{database}.{table}")) { + existing.id = None; + existing.schema_id = None; + } + } + + pub fn set_table_uuid(&self, database: &str, table: &str, uuid: &str) { + let mut s = self.inner.lock().unwrap(); + if let Some(existing) = s.tables.get_mut(&format!("{database}.{table}")) { + existing.id = Some(uuid.to_string()); + } + } + + pub fn set_table_schema_id( + &self, + database: &str, + table: &str, + schema: paimon::spec::Schema, + schema_id: i64, + ) { + let mut s = self.inner.lock().unwrap(); + let key = format!("{database}.{table}"); + if let Some(existing) = s.tables.get_mut(&key) { + existing.schema_id = Some(schema_id); + existing.schema = Some(schema); + } + } + + pub fn set_auth_response(&self, database: &str, table: &str, response: AuthTableQueryResponse) { + let mut s = self.inner.lock().unwrap(); + s.auth_responses + .insert(format!("{database}.{table}"), response); + } + /// Add a no-permission table to the server state. pub fn add_no_permission_table(&self, database: &str, table: &str) { let mut s = self.inner.lock().unwrap(); @@ -1105,6 +1240,10 @@ pub async fn start_mock_server( &format!("{prefix}/databases/:db/functions/:function"), get(RESTServer::get_function), ) + .route( + &format!("{prefix}/databases/:db/tables/:table/auth"), + post(RESTServer::auth_table_query), + ) .route( &format!("{prefix}/tables/rename"), post(RESTServer::rename_table), diff --git a/crates/paimon/tests/rest_catalog_test.rs b/crates/paimon/tests/rest_catalog_test.rs index 41052c453..3e70f6516 100644 --- a/crates/paimon/tests/rest_catalog_test.rs +++ b/crates/paimon/tests/rest_catalog_test.rs @@ -1677,3 +1677,415 @@ async fn test_load_table_rejects_unknown_declared_type() { "{err:?}" ); } + +// Skipped on Windows for the same opendal `fs` StripPrefixError as the +// blob-view regression above: it writes through FileSystemCatalog. +#[cfg(not(windows))] +#[tokio::test] +async fn test_query_auth_unrestricted_user_can_read() { + let tmp = tempfile::tempdir().unwrap(); + let warehouse = format!("file://{}", tmp.path().display()); + let mut fs_options = Options::new(); + fs_options.set(CatalogOptions::WAREHOUSE, &warehouse); + let fs_catalog = FileSystemCatalog::new(fs_options).expect("create filesystem catalog"); + fs_catalog + .create_database("default", true, HashMap::new()) + .await + .unwrap(); + let identifier = Identifier::new("default", "guarded"); + let columns = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .build() + .unwrap(); + fs_catalog + .create_table(&identifier, columns, false) + .await + .unwrap(); + let plain = fs_catalog.get_table(&identifier).await.unwrap(); + let batch = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ArrowField::new( + "id", + ArrowDataType::Int32, + false, + )])), + vec![Arc::new(Int32Array::from(vec![1, 2, 3]))], + ) + .unwrap(); + write_batch(&plain, batch, "query-auth-fixture").await; + + let ctx = setup_catalog(vec!["default"]).await; + let guarded_schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("query-auth.enabled", "true") + .build() + .unwrap(); + ctx.server + .add_table_with_schema("default", "guarded", guarded_schema, plain.location()); + + let table = ctx.catalog.get_table(&identifier).await.unwrap(); + let read_builder = table.new_read_builder(); + let plan = read_builder.new_scan().plan().await.unwrap(); + assert!( + !plan.splits().is_empty(), + "the fixture must produce a split, or the read below proves nothing" + ); + + let batches = read_builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .expect("an unrestricted user must be allowed to read") + .try_collect::>() + .await + .expect("and the rows must decode"); + assert_eq!( + batches.iter().map(|b| b.num_rows()).sum::(), + 3, + "every written row must come back" + ); +} + +fn schema_of(columns: &[&str], options: &[(&str, &str)]) -> Schema { + let mut builder = Schema::builder(); + for name in columns { + builder = builder.column(*name, DataType::Int(IntType::new())); + } + for (key, value) in options { + builder = builder.option(*key, *value); + } + builder.build().unwrap() +} + +const GUARDED: &[(&str, &str)] = &[("query-auth.enabled", "true")]; + +struct Guarded { + ctx: TestContext, + table: Table, + identifier: Identifier, + _tmp: tempfile::TempDir, +} + +async fn guarded(name: &str, columns: &[&str]) -> Guarded { + let ctx = setup_catalog(vec!["default"]).await; + let tmp = tempfile::tempdir().unwrap(); + let path = format!("file://{}", tmp.path().display()); + ctx.server + .add_table_with_schema("default", name, schema_of(columns, GUARDED), &path); + let identifier = Identifier::new("default", name); + let table = ctx.catalog.get_table(&identifier).await.unwrap(); + Guarded { + ctx, + table, + identifier, + _tmp: tmp, + } +} + +async fn plan_err(table: &Table, why: &str) -> paimon::Error { + table + .new_read_builder() + .new_scan() + .plan() + .await + .expect_err(why) +} + +#[track_caller] +fn assert_refused(err: paimon::Error) { + assert!( + matches!(err, paimon::Error::Unsupported { ref message } + if message.contains("query-auth.enabled")), + "{err:?}" + ); +} + +#[track_caller] +fn assert_drifted(err: paimon::Error, what: &str) { + assert!( + matches!(err, paimon::Error::DataInvalid { ref message, .. } + if message.contains(what)), + "{err:?}" + ); +} + +fn restricted() -> paimon::api::AuthTableQueryResponse { + paimon::api::AuthTableQueryResponse { + filter: Some(vec!["{\"field\":\"id\"}".to_string()]), + column_masking: None, + } +} + +#[tokio::test] +async fn test_query_auth_restricted_user_is_refused_at_plan_time() { + let g = guarded("restricted", &["id"]).await; + g.ctx + .server + .set_auth_response("default", "restricted", restricted()); + + assert_refused( + plan_err( + &g.table, + "a restricted user must be refused before a plan exists", + ) + .await, + ); +} + +#[tokio::test] +async fn test_query_auth_refuses_a_stale_handle() { + let g = guarded("drifting", &["id"]).await; + g.ctx.server.set_table_schema_id( + "default", + "drifting", + schema_of(&["id", "extra"], GUARDED), + 7, + ); + + assert_drifted( + plan_err(&g.table, "a handle whose schema drifted must be refused").await, + "now resolves to schema", + ); +} + +#[tokio::test] +async fn test_query_auth_refuses_a_recreated_table() { + let g = guarded("recreated", &["id"]).await; + g.ctx + .server + .set_table_uuid("default", "recreated", "uuid-of-the-replacement"); + + assert_drifted( + plan_err( + &g.table, + "a re-created table must not reuse this handle's grant", + ) + .await, + "now resolves to uuid", + ); +} + +#[tokio::test] +async fn test_query_auth_refuses_a_table_recreated_while_planning() { + let g = guarded("planned", &["id"]).await; + g.ctx + .server + .set_table_uuid_after_calls("default", "planned", "uuid-of-the-replacement", 2); + + assert_drifted( + plan_err(&g.table, "the files just planned belong to the replacement").await, + "now resolves to uuid", + ); +} + +#[tokio::test] +async fn test_query_auth_refuses_a_table_recreated_mid_exchange() { + let g = guarded("swapped", &["id"]).await; + g.ctx + .server + .set_table_uuid_after_auth("default", "swapped", "uuid-after-the-exchange"); + + assert_drifted( + plan_err(&g.table, "a table replaced mid-exchange must be refused").await, + "now resolves to uuid", + ); +} + +#[tokio::test] +async fn test_query_auth_refuses_a_server_that_reports_no_identity() { + let g = guarded("anonymous", &["id"]).await; + g.ctx.server.clear_table_identity("default", "anonymous"); + + assert_drifted( + plan_err( + &g.table, + "a check that cannot establish the table has not checked anything", + ) + .await, + "nothing the server reports", + ); +} + +#[tokio::test] +async fn test_query_auth_user_granted_all_business_columns_can_read() { + let g = guarded("granted", &["id", "name"]).await; + g.ctx.server.set_column_auth( + "default", + "granted", + vec!["id".to_string(), "name".to_string()], + ); + + g.table + .new_read_builder() + .new_scan() + .plan() + .await + .expect("a user granted every column must be authorized"); +} + +#[tokio::test] +async fn test_query_auth_enabled_after_a_handle_was_loaded_is_still_enforced() { + let ctx = setup_catalog(vec!["default"]).await; + let tmp = tempfile::tempdir().unwrap(); + let path = format!("file://{}", tmp.path().display()); + ctx.server + .add_table_with_schema("default", "later", schema_of(&["id"], &[]), &path); + let table = ctx + .catalog + .get_table(&Identifier::new("default", "later")) + .await + .unwrap(); + + ctx.server + .set_table_schema_id("default", "later", schema_of(&["id"], GUARDED), 0); + ctx.server + .set_auth_response("default", "later", restricted()); + + assert_refused( + plan_err( + &table, + "a handle loaded before the option was set must still be authorized", + ) + .await, + ); +} + +#[tokio::test] +async fn test_query_auth_is_not_weakened_by_a_table_recreated_under_the_same_name() { + let g = guarded("guarded", &["id"]).await; + g.ctx + .server + .set_table_schema_id("default", "guarded", schema_of(&["id"], &[]), 0); + + let err = g + .table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .expect_err("the answer is now about a different table over the same files"); + assert_refused(err); +} + +#[tokio::test] +async fn test_query_auth_refuses_a_decorated_handle() { + let ctx = setup_catalog(vec!["default"]).await; + let tmp = tempfile::tempdir().unwrap(); + let path = format!("file://{}", tmp.path().display()); + for name in ["guarded$branch_dev", "guarded$files"] { + ctx.server + .add_table_with_schema("default", name, schema_of(&["id"], GUARDED), &path); + let table = ctx + .catalog + .get_table(&Identifier::new("default", name)) + .await + .unwrap(); + assert_refused( + plan_err( + &table, + "the decorated endpoint rules on files this handle does not read", + ) + .await, + ); + } +} + +#[tokio::test] +async fn test_query_auth_refuses_an_assembled_handle() { + let g = guarded("guarded", &["id"]).await; + let elsewhere = tempfile::tempdir().unwrap(); + for (schema, location) in [ + (g.table.schema().clone(), g.table.location().to_string()), + ( + paimon::spec::TableSchema::new( + g.table.schema().id(), + &schema_of(&["id", "dropped"], GUARDED), + ), + g.table.location().to_string(), + ), + ( + g.table.schema().clone(), + format!("file://{}", elsewhere.path().display()), + ), + ] { + let assembled = paimon::table::Table::new( + g.table.file_io().clone(), + g.identifier.clone(), + location, + schema, + g.table.rest_env().cloned(), + ); + assert_refused(plan_err(&assembled, "an assembled handle carries no session").await); + } +} + +#[tokio::test] +async fn test_planning_an_ordinary_rest_table_asks_the_server_once() { + let ctx = setup_catalog(vec!["default"]).await; + let tmp = tempfile::tempdir().unwrap(); + let path = format!("file://{}", tmp.path().display()); + ctx.server + .add_table_with_schema("default", "plain", schema_of(&["id"], &[]), &path); + let table = ctx + .catalog + .get_table(&Identifier::new("default", "plain")) + .await + .unwrap(); + + let before = ctx.server.get_table_calls(); + table.new_read_builder().new_scan().plan().await.unwrap(); + assert_eq!( + ctx.server.get_table_calls() - before, + 1, + "planning must not repeat the query-auth lookup" + ); +} + +#[tokio::test] +async fn test_query_auth_refuses_scan_all_files_and_format_tables() { + let ctx = setup_catalog(vec!["default"]).await; + let tmp = tempfile::tempdir().unwrap(); + let path = format!("file://{}", tmp.path().display()); + ctx.server + .add_table_with_schema("default", "metadata", schema_of(&["id"], &[]), &path); + let table = ctx + .catalog + .get_table(&Identifier::new("default", "metadata")) + .await + .unwrap(); + ctx.server + .set_table_schema_id("default", "metadata", schema_of(&["id"], GUARDED), 0); + + let err = table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .expect_err("file metadata is not something the auth endpoint can rule on"); + assert_refused(err); + + let format = &[("type", "format-table"), ("file.format", "parquet")]; + ctx.server + .add_table_with_schema("default", "fmt", schema_of(&["id"], format), &path); + let fmt = ctx + .catalog + .get_table(&Identifier::new("default", "fmt")) + .await + .unwrap(); + let mut guarded_format = format.to_vec(); + guarded_format.push(("query-auth.enabled", "true")); + ctx.server + .set_table_schema_id("default", "fmt", schema_of(&["id"], &guarded_format), 0); + + assert_refused(plan_err(&fmt, "a format table cannot apply the server's rules").await); + + assert_refused( + table + .new_read_builder() + .new_incremental_scan(paimon::table::IncrementalScanMode::Delta, 0, 1) + .plan() + .await + .expect_err("an incremental read cannot apply the server's rules"), + ); +}