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 <marshawcoco@users.noreply.github.com>

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<Str
     Ok(normalized)
 }
 
-fn table_credential_session_policy(bucket: &str, object_prefix: &str) -> S3Result<Policy> {
+fn table_credential_catalog_resource(entry: &crate::table_catalog::TableEntry) -> S3Result<String> {
+    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<Policy> {
+    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();

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<T>(req: &mut S3Request<T>, 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<T>(req: &mut S3Request<T>, 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<T>(req: &mut S3Request<T>, 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<AdminAction> {
+    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<crate::table_catalog::EcStoreTableCatalogStore<ECStore>> {
+    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<Option<crate::table_catalog::TableDataPlaneResource>> {
+    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<String, Vec<String>>,
+    claims: &HashMap<String, serde_json::Value>,
+) -> 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(

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<T> = Result<T, TableCatalogStoreError>;
 
+fn normalize_warehouse_object_prefix(object_prefix: &str) -> TableCatalogStoreResult<String> {
+    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<String> {
+    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<String> {
+    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<Option<String>> {
+    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<S>(
+    store: &S,
+    bucket: &str,
+    object: &str,
+) -> TableCatalogStoreResult<Option<TableDataPlaneResource>>
+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<TableDataPlaneResource> = 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<Option<TableBucketEntry>>;
@@ -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();

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
 

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]:

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"),
             ],
         )