From 9372ee70329d08ea7aafae8df84f6b1ecb5bd686 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sun, 14 Jun 2026 18:13:02 +0800 Subject: [PATCH] feat(table-catalog): bridge table data-plane policy (#3436) * feat(table-catalog): bridge table data-plane policy * test(table-catalog): harden vended credential smoke * test(table-catalog): cover data-plane policy denials * fix(table-catalog): protect relocated warehouse scope * fix(table-catalog): skip invalid warehouse entries --------- Co-authored-by: Henry Guo --- rustfs/src/admin/handlers/table_catalog.rs | 98 +++++- rustfs/src/storage/access.rs | 173 +++++++++- rustfs/src/table_catalog.rs | 308 +++++++++++++++++- scripts/table-catalog/README.md | 13 +- scripts/table-catalog/pyiceberg_smoke.py | 45 ++- scripts/table-catalog/test_pyiceberg_smoke.py | 40 ++- 6 files changed, 651 insertions(+), 26 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 44429ac63..51259cada 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -389,7 +389,7 @@ impl TableCredentialIssuer for IamTableCredentialIssuer { )); } - let policy = table_credential_session_policy(&request.entry.table_bucket, &request.object_prefix)?; + let policy = table_credential_session_policy(request.entry, &request.object_prefix)?; let policy_buf = serde_json::to_vec(&policy) .map_err(|err| s3_error!(InternalError, "failed to serialize table credential session policy: {}", err))?; let expiration = OffsetDateTime::now_utc().saturating_add(Duration::seconds(self.ttl_seconds)); @@ -990,8 +990,18 @@ fn normalize_table_credential_object_prefix(object_prefix: &str) -> S3Result S3Result { +fn table_credential_catalog_resource(entry: &crate::table_catalog::TableEntry) -> S3Result { + let namespace = crate::table_catalog::Namespace::parse(&entry.namespace) + .map_err(|err| s3_error!(InvalidRequest, "invalid table credential namespace: {}", err))?; + let table = crate::table_catalog::IdentifierSegment::parse(&entry.table) + .map_err(|err| s3_error!(InvalidRequest, "invalid table credential table name: {}", err))?; + Ok(format!("namespaces/{}/tables/{}", namespace.storage_id(), table.as_str())) +} + +fn table_credential_session_policy(entry: &crate::table_catalog::TableEntry, object_prefix: &str) -> S3Result { + let bucket = &entry.table_bucket; let object_prefix = normalize_table_credential_object_prefix(object_prefix)?; + let catalog_resource = table_credential_catalog_resource(entry)?; let policy = serde_json::json!({ "Version": "2012-10-17", "Statement": [ @@ -1016,6 +1026,16 @@ fn table_credential_session_policy(bucket: &str, object_prefix: &str) -> S3Resul "Resource": [ format!("arn:aws:s3:::{bucket}") ] + }, + { + "Effect": "Allow", + "Action": [ + "admin:GetTableMetadata", + "admin:SetTableMetadata" + ], + "Resource": [ + format!("arn:aws:s3:::{bucket}/{catalog_resource}") + ] } ] }); @@ -4313,8 +4333,8 @@ mod tests { #[tokio::test] async fn table_credential_session_policy_is_limited_to_table_prefix() { - let policy = - table_credential_session_policy("warehouse", "tables/table-id/").expect("table credential policy should build"); + let policy = table_credential_session_policy(&table_entry_for_credentials(), "tables/table-id/") + .expect("table credential policy should build"); let groups = None; let conditions = std::collections::HashMap::new(); let claims = std::collections::HashMap::new(); @@ -4349,6 +4369,21 @@ mod tests { }) .await ); + assert!( + policy + .is_allowed(&rustfs_policy::policy::Args { + account: "temporary-access-key", + groups: &groups, + action: Action::AdminAction(rustfs_policy::policy::action::AdminAction::SetTableMetadataAction), + bucket: "warehouse", + conditions: &conditions, + is_owner: false, + object: "namespaces/analytics/tables/events", + claims: &claims, + deny_only: false, + }) + .await + ); assert!( !policy .is_allowed(&rustfs_policy::policy::Args { @@ -4381,6 +4416,61 @@ mod tests { ); } + #[tokio::test] + async fn table_credential_session_policy_includes_table_resource_actions() { + let policy = table_credential_session_policy(&table_entry_for_credentials(), "tables/table-id/") + .expect("table credential policy should build"); + let groups = None; + let conditions = std::collections::HashMap::new(); + let claims = std::collections::HashMap::new(); + + assert!( + policy + .is_allowed(&rustfs_policy::policy::Args { + account: "temporary-access-key", + groups: &groups, + action: Action::AdminAction(rustfs_policy::policy::action::AdminAction::GetTableMetadataAction), + bucket: "warehouse", + conditions: &conditions, + is_owner: false, + object: "namespaces/analytics/tables/events", + claims: &claims, + deny_only: false, + }) + .await + ); + assert!( + !policy + .is_allowed(&rustfs_policy::policy::Args { + account: "temporary-access-key", + groups: &groups, + action: Action::AdminAction(rustfs_policy::policy::action::AdminAction::SetTableMetadataAction), + bucket: "warehouse", + conditions: &conditions, + is_owner: false, + object: "namespaces/analytics/tables/orders", + claims: &claims, + deny_only: false, + }) + .await + ); + assert!( + !policy + .is_allowed(&rustfs_policy::policy::Args { + account: "temporary-access-key", + groups: &groups, + action: Action::AdminAction(rustfs_policy::policy::action::AdminAction::CreateTableAction), + bucket: "warehouse", + conditions: &conditions, + is_owner: false, + object: "namespaces/analytics/tables/events", + claims: &claims, + deny_only: false, + }) + .await + ); + } + #[test] fn table_credential_scope_rejects_cross_bucket_or_unsafe_prefix() { let mut entry = table_entry_for_credentials(); diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index e0c10e068..b2b49d197 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -23,9 +23,10 @@ use rustfs_ecstore::bucket::metadata_sys; use rustfs_ecstore::bucket::policy_sys::PolicySys; use rustfs_ecstore::error::{StorageError, is_err_bucket_not_found}; use rustfs_ecstore::new_object_layer_fn; +use rustfs_ecstore::store::ECStore; use rustfs_ecstore::store_api::BucketOperations; use rustfs_iam::error::Error as IamError; -use rustfs_policy::policy::action::{Action, S3Action}; +use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use rustfs_policy::policy::{ Args, BucketPolicy, BucketPolicyArgs, bucket_policy_needs_existing_object_tag_for_args, bucket_policy_uses_existing_object_tag_conditions, @@ -501,6 +502,8 @@ pub async fn authorize_request(req: &mut S3Request, action: Action) -> S3R }; if iam_allowed { + authorize_table_data_plane_if_needed(action, bucket.as_str(), object.as_str(), cred, is_owner, &conditions, claims) + .await?; return Ok(()); } @@ -516,6 +519,8 @@ pub async fn authorize_request(req: &mut S3Request, action: Action) -> S3R .await; if policy_allowed_fallback { + authorize_table_data_plane_if_needed(action, bucket.as_str(), object.as_str(), cred, is_owner, &conditions, claims) + .await?; return Ok(()); } @@ -664,6 +669,7 @@ pub async fn authorize_request(req: &mut S3Request, action: Action) -> S3R .await; if policy_allowed { + deny_anonymous_table_data_plane_if_needed(action, bucket.as_str(), object.as_str()).await?; // RestrictPublicBuckets: when true, deny public access even if bucket policy allows it. match metadata_sys::get_public_access_block_config(bucket_name).await { Ok((config, _)) => { @@ -739,6 +745,146 @@ fn list_parts_authorize_action() -> Action { Action::S3Action(S3Action::ListMultipartUploadPartsAction) } +fn table_data_plane_admin_action(action: Action) -> Option { + match action { + Action::S3Action( + S3Action::GetObjectAction + | S3Action::GetObjectAclAction + | S3Action::GetObjectVersionAction + | S3Action::GetObjectAttributesAction + | S3Action::GetObjectVersionAttributesAction + | S3Action::GetObjectTaggingAction + | S3Action::GetObjectVersionTaggingAction + | S3Action::GetObjectRetentionAction + | S3Action::GetObjectLegalHoldAction + | S3Action::GetObjectVersionForReplicationAction, + ) => Some(AdminAction::GetTableMetadataAction), + Action::S3Action( + S3Action::PutObjectAction + | S3Action::PutObjectAclAction + | S3Action::DeleteObjectAction + | S3Action::DeleteObjectVersionAction + | S3Action::PutObjectTaggingAction + | S3Action::PutObjectVersionTaggingAction + | S3Action::DeleteObjectTaggingAction + | S3Action::DeleteObjectVersionTaggingAction + | S3Action::PutObjectRetentionAction + | S3Action::PutObjectLegalHoldAction + | S3Action::BypassGovernanceRetentionAction + | S3Action::AbortMultipartUploadAction + | S3Action::ListMultipartUploadPartsAction + | S3Action::RestoreObjectAction + | S3Action::ReplicateObjectAction + | S3Action::ReplicateDeleteAction + | S3Action::ReplicateTagsAction + | S3Action::PutObjectFanOutAction, + ) => Some(AdminAction::SetTableMetadataAction), + _ => None, + } +} + +fn table_catalog_store_for_data_plane() -> S3Result> { + let store = new_object_layer_fn().ok_or_else(|| s3_error!(InternalError, "object store not initialized"))?; + let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store); + Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend)) +} + +async fn table_data_plane_resource_for_request( + bucket: &str, + object: &str, +) -> S3Result> { + if bucket.is_empty() || object.is_empty() { + return Ok(None); + } + + match metadata_sys::get(bucket).await { + Ok(metadata) if metadata.table_bucket_enabled() => {} + Ok(_) | Err(StorageError::ConfigNotFound) => return Ok(None), + Err(err) if is_err_bucket_not_found(&err) => return Ok(None), + Err(err) => { + tracing::warn!( + bucket = %bucket, + error = %err, + "failed to load bucket metadata while authorizing table data-plane access" + ); + return Err(s3_error!(AccessDenied, "Access Denied")); + } + } + + let store = table_catalog_store_for_data_plane()?; + crate::table_catalog::table_data_plane_resource_for_object(&store, bucket, object) + .await + .map_err(|err| { + tracing::warn!( + bucket = %bucket, + object = %object, + error = %err, + "failed to resolve table data-plane resource" + ); + s3_error!(AccessDenied, "Access Denied") + }) +} + +async fn authorize_table_data_plane_if_needed( + action: Action, + bucket: &str, + object: &str, + cred: &rustfs_credentials::Credentials, + is_owner: bool, + conditions: &HashMap>, + claims: &HashMap, +) -> S3Result<()> { + let Some(admin_action) = table_data_plane_admin_action(action) else { + return Ok(()); + }; + let Some(resource) = table_data_plane_resource_for_request(bucket, object).await? else { + return Ok(()); + }; + let Ok(iam_store) = rustfs_iam::get() else { + return Err(s3_error!(InternalError, "iam not init")); + }; + + let resource_object = resource.catalog_resource_object(); + let allowed = iam_store + .is_allowed(&Args { + account: &cred.access_key, + groups: &cred.groups, + action: Action::AdminAction(admin_action), + bucket: resource.table_bucket.as_str(), + conditions, + is_owner, + object: resource_object.as_str(), + claims, + deny_only: false, + }) + .await; + if allowed { + return Ok(()); + } + + tracing::debug!( + bucket = %bucket, + object = %object, + table_bucket = %resource.table_bucket, + table_namespace = %resource.namespace, + table = %resource.table, + table_id = %resource.table_id, + ?admin_action, + "table data-plane access denied by table resource policy" + ); + Err(s3_error!(AccessDenied, "Access Denied")) +} + +async fn deny_anonymous_table_data_plane_if_needed(action: Action, bucket: &str, object: &str) -> S3Result<()> { + if table_data_plane_admin_action(action).is_none() { + return Ok(()); + } + if table_data_plane_resource_for_request(bucket, object).await?.is_some() { + return Err(s3_error!(AccessDenied, "Access Denied")); + } + Ok(()) +} + fn validate_post_object_success_controls(input: &PostObjectInput) -> S3Result<()> { if let Some(status) = input.success_action_status && !matches!(status, 200 | 201 | 204) @@ -1954,6 +2100,31 @@ mod tests { assert_eq!(list_parts_authorize_action(), Action::S3Action(S3Action::ListMultipartUploadPartsAction)); } + #[test] + fn table_data_plane_admin_action_maps_s3_object_operations() { + assert_eq!( + table_data_plane_admin_action(Action::S3Action(S3Action::GetObjectAction)), + Some(rustfs_policy::policy::action::AdminAction::GetTableMetadataAction) + ); + assert_eq!( + table_data_plane_admin_action(Action::S3Action(S3Action::PutObjectAction)), + Some(rustfs_policy::policy::action::AdminAction::SetTableMetadataAction) + ); + assert_eq!( + table_data_plane_admin_action(Action::S3Action(S3Action::PutObjectTaggingAction)), + Some(rustfs_policy::policy::action::AdminAction::SetTableMetadataAction) + ); + assert_eq!( + table_data_plane_admin_action(Action::S3Action(S3Action::DeleteObjectAction)), + Some(rustfs_policy::policy::action::AdminAction::SetTableMetadataAction) + ); + assert_eq!( + table_data_plane_admin_action(Action::S3Action(S3Action::GetObjectTaggingAction)), + Some(rustfs_policy::policy::action::AdminAction::GetTableMetadataAction) + ); + assert_eq!(table_data_plane_admin_action(Action::S3Action(S3Action::ListBucketAction)), None); + } + #[test] fn legal_hold_write_requested_is_true_when_status_present() { assert!(legal_hold_write_requested(Some(&ObjectLockLegalHoldStatus::from_static( diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index a057922a9..e55568afc 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -302,6 +302,24 @@ pub(crate) struct TableCatalogExport { pub table: TableEntry, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct TableDataPlaneResource { + pub table_bucket: String, + pub namespace: String, + pub table: String, + pub table_id: String, + pub warehouse_object_prefix: String, +} + +impl TableDataPlaneResource { + pub(crate) fn catalog_resource_object(&self) -> String { + let namespace = Namespace::parse(&self.namespace) + .map(|namespace| namespace.storage_id()) + .unwrap_or_else(|_| self.namespace.clone()); + format!("{NAMESPACE_ROOT}/{namespace}/{TABLE_ROOT}/{}", self.table) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableMetadataPointerStatus { @@ -341,6 +359,118 @@ impl std::error::Error for TableCatalogStoreError {} pub(crate) type TableCatalogStoreResult = Result; +fn normalize_warehouse_object_prefix(object_prefix: &str) -> TableCatalogStoreResult { + let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix); + if object_prefix.is_empty() { + return Err(TableCatalogStoreError::Invalid( + "table warehouse location must include an object prefix".to_string(), + )); + } + if object_prefix.contains('\\') { + return Err(TableCatalogStoreError::Invalid( + "table warehouse location contains an invalid path separator".to_string(), + )); + } + if object_prefix + .split('/') + .any(|segment| segment.is_empty() || segment == "." || segment == "..") + { + return Err(TableCatalogStoreError::Invalid( + "table warehouse location contains an invalid path segment".to_string(), + )); + } + + let mut normalized = object_prefix.to_string(); + normalized.push('/'); + Ok(normalized) +} + +fn table_warehouse_object_prefix_from_location(table_bucket: &str, warehouse_location: &str) -> TableCatalogStoreResult { + let location = warehouse_location + .strip_prefix("s3://") + .ok_or_else(|| TableCatalogStoreError::Invalid("table warehouse location must be an s3 URI".to_string()))?; + let (bucket, object_prefix) = location + .split_once('/') + .ok_or_else(|| TableCatalogStoreError::Invalid("table warehouse location must include an object prefix".to_string()))?; + if bucket != table_bucket { + return Err(TableCatalogStoreError::Invalid( + "table warehouse location must be inside the table bucket".to_string(), + )); + } + normalize_warehouse_object_prefix(object_prefix) +} + +fn table_warehouse_object_prefix(entry: &TableEntry) -> TableCatalogStoreResult { + table_warehouse_object_prefix_from_location(&entry.table_bucket, &entry.warehouse_location) +} + +fn metadata_warehouse_location( + table_bucket: &str, + metadata_location: &str, + metadata_object: &TableCatalogObject, +) -> TableCatalogStoreResult> { + let metadata: serde_json::Value = serde_json::from_slice(&metadata_object.data) + .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to parse new metadata {metadata_location}: {err}")))?; + let Some(location) = metadata.get("location").and_then(serde_json::Value::as_str) else { + return Ok(None); + }; + table_warehouse_object_prefix_from_location(table_bucket, location)?; + Ok(Some(location.to_string())) +} + +pub(crate) async fn table_data_plane_resource_for_object( + store: &S, + bucket: &str, + object: &str, +) -> TableCatalogStoreResult> +where + S: TableCatalogStore + ?Sized, +{ + if bucket.is_empty() || object.is_empty() { + return Ok(None); + } + + let Some(table_bucket) = store.get_table_bucket(bucket).await? else { + return Ok(None); + }; + if table_bucket.state != TableCatalogEntryState::Active { + return Ok(None); + } + + let mut matched: Option = None; + for namespace in store.list_namespaces(bucket).await? { + if namespace.state != TableCatalogEntryState::Active { + continue; + } + for table in store.list_tables(bucket, &namespace.namespace).await? { + if table.state != TableCatalogEntryState::Active { + continue; + } + let Ok(warehouse_object_prefix) = table_warehouse_object_prefix(&table) else { + continue; + }; + if !object.starts_with(&warehouse_object_prefix) { + continue; + } + if matched + .as_ref() + .is_some_and(|current| current.warehouse_object_prefix.len() >= warehouse_object_prefix.len()) + { + continue; + } + matched = Some(TableDataPlaneResource { + table_bucket: table.table_bucket, + namespace: table.namespace, + table: table.table, + table_id: table.table_id, + warehouse_object_prefix, + }); + } + } + + Ok(matched) +} + #[async_trait::async_trait] pub(crate) trait TableCatalogStore: Send + Sync { async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult>; @@ -1353,16 +1483,18 @@ where "new metadata location must be inside the table metadata directory".to_string(), )); } - if !self + let Some(new_metadata_object) = self .backend - .object_exists(&request.table_bucket, &request.new_metadata_location) + .read_object(&request.table_bucket, &request.new_metadata_location) .await? - { + else { return Err(TableCatalogStoreError::NotFound(format!( "new metadata object {}", request.new_metadata_location ))); - } + }; + let next_warehouse_location = + metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &new_metadata_object)?; let has_existing_commit = existing_commit.is_some(); let mut staged_commit_log = existing_commit.unwrap_or_else(|| CommitLogEntry { @@ -1385,6 +1517,9 @@ where let mut next = current.clone(); next.metadata_location = staged_commit_log.new_metadata_location.clone(); + if let Some(warehouse_location) = next_warehouse_location { + next.warehouse_location = warehouse_location; + } next.version_token = staged_commit_log.new_version_token.clone(); next.generation = current.generation.saturating_add(1); @@ -2788,6 +2923,114 @@ mod tests { assert_eq!(report.cleanup_candidate_locations, vec![v1, v2]); } + #[tokio::test] + async fn table_data_plane_resource_resolves_registered_warehouse_prefix() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; + + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") + .await + .expect("data-plane resource lookup should succeed") + .expect("object should resolve to the registered table"); + + assert_eq!(resource.table_bucket, bucket); + assert_eq!(resource.namespace, "sales"); + assert_eq!(resource.table, "orders"); + assert_eq!(resource.table_id, "table-id"); + assert_eq!(resource.warehouse_object_prefix, "tables/table-id/"); + assert_eq!(resource.catalog_resource_object(), "namespaces/sales/tables/orders"); + } + + #[tokio::test] + async fn table_data_plane_resource_does_not_match_sibling_prefix() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; + + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id-other/data/part-00001.parquet") + .await + .expect("data-plane resource lookup should succeed"); + + assert!(resource.is_none()); + } + + #[tokio::test] + async fn table_data_plane_resource_prefers_longest_registered_warehouse_prefix() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let parent_table = IdentifierSegment::parse("orders").unwrap(); + let child_table = IdentifierSegment::parse("orders_child").unwrap(); + let current = default_table_metadata_file_path(&namespace, &parent_table, "00001.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &parent_table, current.clone()).await; + let mut child_entry = test_table_entry(bucket, &namespace, &child_table, current); + child_entry.table_id = "table-id-child".to_string(); + child_entry.warehouse_location = format!("s3://{bucket}/tables/table-id/child"); + store.create_table(child_entry).await.unwrap(); + + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/data/part-00001.parquet") + .await + .expect("data-plane resource lookup should succeed") + .expect("object should resolve to the child table"); + + assert_eq!(resource.table, "orders_child"); + assert_eq!(resource.table_id, "table-id-child"); + assert_eq!(resource.warehouse_object_prefix, "tables/table-id/child/"); + assert_eq!(resource.catalog_resource_object(), "namespaces/sales/tables/orders_child"); + } + + #[tokio::test] + async fn table_data_plane_resource_skips_invalid_warehouse_locations() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let invalid_table = IdentifierSegment::parse("bad_orders").unwrap(); + let valid_table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &valid_table, "00001.metadata.json"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + + let mut invalid_entry = test_table_entry(bucket, &namespace, &invalid_table, current.clone()); + invalid_entry.table_id = "bad-table-id".to_string(); + invalid_entry.warehouse_location = format!("s3://{bucket}/"); + store.create_table(invalid_entry).await.unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &valid_table, current)) + .await + .unwrap(); + + let unrelated = table_data_plane_resource_for_object(&store, bucket, "ordinary/object.parquet") + .await + .expect("invalid table warehouse location should not deny unrelated object lookup"); + assert!(unrelated.is_none()); + + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") + .await + .expect("invalid table warehouse location should not block a later valid match") + .expect("valid table warehouse object should resolve to the table"); + assert_eq!(resource.table, "orders"); + assert_eq!(resource.table_id, "table-id"); + assert_eq!(resource.warehouse_object_prefix, "tables/table-id/"); + } + #[tokio::test] async fn maintenance_dry_run_reports_job_context_and_deletable_candidates() { let backend = TestCatalogObjectBackend::default(); @@ -3387,6 +3630,63 @@ mod tests { ); } + #[tokio::test] + async fn object_table_catalog_store_syncs_warehouse_location_from_committed_metadata() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let new_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .unwrap(); + backend + .seed_object( + bucket, + &new_metadata, + serde_json::to_vec(&serde_json::json!({ + "location": "s3://analytics/tables/relocated-table-id", + "table-uuid": "table-uuid" + })) + .unwrap(), + ) + .await; + + let result = store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: None, + operation: "set-location".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata, + new_metadata_location: new_metadata, + requirements: vec![serde_json::json!({"type": "assert-table-uuid", "uuid": "table-uuid"})], + writer: Some("pyiceberg/test".to_string()), + }) + .await + .unwrap(); + + assert_eq!(result.table.warehouse_location, "s3://analytics/tables/relocated-table-id"); + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/relocated-table-id/data/part-00001.parquet") + .await + .expect("data-plane resource lookup should succeed") + .expect("relocated table warehouse object should resolve to the table"); + assert_eq!(resource.table, "orders"); + assert_eq!(resource.warehouse_object_prefix, "tables/relocated-table-id/"); + } + #[tokio::test] async fn object_table_catalog_store_does_not_advance_table_when_idempotency_staging_fails() { let backend = TestCatalogObjectBackend::default(); diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 2fe42ae69..fd7718eca 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -94,12 +94,15 @@ table-scoped S3 access key, secret key, and session token before append, reload, and scan operations. Before the PyIceberg append, the profile also checks that the returned -credential prefix exactly matches the created table warehouse location, then -runs a direct S3 data-plane scope probe with the returned temporary credentials: +credential prefix exactly matches the created table warehouse location after +canonical S3 URI normalization, including percent-decoding equivalent path +encodings. It then runs a direct S3 data-plane scope probe with the returned +temporary credentials: -- `PutObject`, `HeadObject`, and `DeleteObject` must work inside the returned - table warehouse prefix. -- `PutObject` to the same bucket outside that prefix must be rejected. +- `PutObject`, `HeadObject`, `GetObject`, and `DeleteObject` must work inside + the returned table warehouse prefix. +- `PutObject` and `GetObject` to the same bucket outside that prefix must be + rejected. ## Machine-Readable Inventories diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index e382802ba..9e0eca5d5 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -431,12 +431,15 @@ def s3_scope_from_uri(s3_uri: str, description: str) -> tuple[str, str]: parsed = urllib.parse.urlparse(s3_uri) if parsed.scheme != "s3" or not parsed.netloc: raise RuntimeError(f"{description} must be an s3 URI") - object_prefix = parsed.path.lstrip("/") + bucket = urllib.parse.unquote(parsed.netloc) + object_prefix = urllib.parse.unquote(parsed.path.lstrip("/")).rstrip("/") if not object_prefix: raise RuntimeError(f"{description} must include an object prefix") - if not object_prefix.endswith("/"): - object_prefix = f"{object_prefix}/" - return parsed.netloc, object_prefix + if "\\" in object_prefix: + raise RuntimeError(f"{description} contains an invalid path separator") + if any(segment in {"", ".", ".."} for segment in object_prefix.split("/")): + raise RuntimeError(f"{description} contains an invalid path segment") + return bucket, f"{object_prefix}/" def s3_scope_from_credential(storage_credential: StorageCredential) -> tuple[str, str]: @@ -520,22 +523,44 @@ def verify_vended_credential_data_plane_scope( client.put_object(Bucket=bucket, Key=inside_key, Body=b"rustfs table credential scope probe\n") put_inside = True client.head_object(Bucket=bucket, Key=inside_key) + client.get_object(Bucket=bucket, Key=inside_key) finally: if put_inside: client.delete_object(Bucket=bucket, Key=inside_key) + put_denied = False try: client.put_object(Bucket=bucket, Key=denied_key, Body=b"rustfs denied table credential scope probe\n") except deps.botocore_client_error as error: if is_access_denied_error(error): - return - raise RuntimeError(f"scope-denied data-plane probe failed with unexpected S3 error: {error}") from error + put_denied = True + else: + raise RuntimeError(f"scope-denied write probe failed with unexpected S3 error: {error}") from error + if not put_denied: + try: + configured_s3_client(args, deps).delete_object(Bucket=bucket, Key=denied_key) + except Exception as cleanup_error: + print(f"warning: failed to clean unexpected denied-scope probe object {denied_key}: {cleanup_error}", file=sys.stderr) + raise RuntimeError("vended table credentials unexpectedly wrote outside the table warehouse prefix") + admin_client = configured_s3_client(args, deps) + seeded_denied_key = False try: - configured_s3_client(args, deps).delete_object(Bucket=bucket, Key=denied_key) - except Exception as cleanup_error: - print(f"warning: failed to clean unexpected denied-scope probe object {denied_key}: {cleanup_error}", file=sys.stderr) - raise RuntimeError("vended table credentials unexpectedly wrote outside the table warehouse prefix") + admin_client.put_object(Bucket=bucket, Key=denied_key, Body=b"rustfs denied read table credential scope probe\n") + seeded_denied_key = True + try: + client.get_object(Bucket=bucket, Key=denied_key) + except deps.botocore_client_error as error: + if is_access_denied_error(error): + return + raise RuntimeError(f"scope-denied read probe failed with unexpected S3 error: {error}") from error + raise RuntimeError("vended table credentials unexpectedly read outside the table warehouse prefix") + finally: + if seeded_denied_key: + try: + admin_client.delete_object(Bucket=bucket, Key=denied_key) + except Exception as cleanup_error: + print(f"warning: failed to clean denied-scope read probe object {denied_key}: {cleanup_error}", file=sys.stderr) def catalog_properties(args: argparse.Namespace, storage_credential: StorageCredential | None = None) -> dict[str, str]: diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 2aa11aa26..9b31ef3f9 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -149,6 +149,15 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertEqual(bucket, "lake") self.assertEqual(key_prefix, "tables/table-id/") + def test_s3_scope_from_uri_decodes_equivalent_prefix_encoding(self) -> None: + bucket, key_prefix = pyiceberg_smoke.s3_scope_from_uri( + "s3://lake/tables/table%2Did/", + "storage credential prefix", + ) + + self.assertEqual(bucket, "lake") + self.assertEqual(key_prefix, "tables/table-id/") + def test_storage_credential_prefix_rejects_bucket_scope(self) -> None: credential = pyiceberg_smoke.StorageCredential( prefix="s3://lake", @@ -220,6 +229,22 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): def head_object(self, *, Bucket: str, Key: str) -> None: self.calls.append(("head", Key)) + def get_object(self, *, Bucket: str, Key: str) -> dict[str, bytes]: + self.calls.append(("get", Key)) + if Key == "outside/probe": + raise FakeClientError(403, "AccessDenied") + return {"Body": b"rustfs table credential scope probe\n"} + + def delete_object(self, *, Bucket: str, Key: str) -> None: + self.calls.append(("delete", Key)) + + class FakeAdminS3Client: + def __init__(self) -> None: + self.calls: list[tuple[str, str]] = [] + + def put_object(self, *, Bucket: str, Key: str, Body: bytes) -> None: + self.calls.append(("put", Key)) + def delete_object(self, *, Bucket: str, Key: str) -> None: self.calls.append(("delete", Key)) @@ -233,19 +258,30 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): }, ) fake_client = FakeS3Client() + fake_admin_client = FakeAdminS3Client() deps = mock.Mock(botocore_client_error=FakeClientError) with mock.patch.object(pyiceberg_smoke, "vended_s3_client", return_value=fake_client): - with mock.patch.object(pyiceberg_smoke, "scope_probe_keys", return_value=("tables/table-id/probe", "outside/probe")): - pyiceberg_smoke.verify_vended_credential_data_plane_scope(args, deps, credential, "s3://lake/tables/table-id") + with mock.patch.object(pyiceberg_smoke, "configured_s3_client", return_value=fake_admin_client): + with mock.patch.object(pyiceberg_smoke, "scope_probe_keys", return_value=("tables/table-id/probe", "outside/probe")): + pyiceberg_smoke.verify_vended_credential_data_plane_scope(args, deps, credential, "s3://lake/tables/table-id") self.assertEqual( fake_client.calls, [ ("put", "tables/table-id/probe"), ("head", "tables/table-id/probe"), + ("get", "tables/table-id/probe"), ("delete", "tables/table-id/probe"), ("put", "outside/probe"), + ("get", "outside/probe"), + ], + ) + self.assertEqual( + fake_admin_client.calls, + [ + ("put", "outside/probe"), + ("delete", "outside/probe"), ], )