fix(tables): reject reserved warehouse locations (#7671)

Co-authored-by: cxymds <cxymds@gmail.com>

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -2515,6 +2515,11 @@ fn normalize_table_credential_object_prefix(object_prefix: &str) -> S3Result<Str
     if object_prefix.is_empty() {
         return Err(s3_error!(InvalidRequest, "table credential scope prefix is empty"));
     }
+    if crate::table_catalog::is_reserved_table_object_key(object_prefix) {
+        return Err(S3Error::from(ApiError::invalid_request(
+            "table credential scope overlaps the reserved table catalog prefix",
+        )));
+    }
     if object_prefix.contains('\\') {
         return Err(s3_error!(
             InvalidRequest,

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -10618,6 +10618,10 @@ fn table_credential_scope_rejects_cross_bucket_or_unsafe_prefix() {
     entry.warehouse_location = "s3://warehouse/tables/../table-id".to_string();
     assert!(table_credential_scope(&entry).is_err());
 
+    let mut entry = table_entry_for_credentials();
+    entry.warehouse_location = "s3://warehouse/.rustfs-table".to_string();
+    assert!(table_credential_scope(&entry).is_err());
+
     let mut entry = table_entry_for_credentials();
     entry.metadata_location = "s3://other/.rustfs-table/metadata/00001.metadata.json".to_string();
     assert!(table_credential_scope(&entry).is_err());

rustfs/src/table_catalog/iceberg/validation.rs

@@ -69,6 +69,11 @@ fn warehouse_object_prefix_from_location(
             "table warehouse location must be inside the table bucket".to_string(),
         ));
     }
+    if is_reserved_table_object_key(object_prefix.strip_suffix('/').unwrap_or(object_prefix)) {
+        return Err(TableCatalogStoreError::Invalid(
+            "table warehouse location overlaps the reserved table catalog prefix".to_string(),
+        ));
+    }
     normalize_warehouse_object_prefix(object_prefix, max_prefix_depth)
 }
 

rustfs/src/table_catalog/tests.rs

@@ -5579,6 +5579,30 @@ async fn object_table_catalog_store_rejects_invalid_table_warehouse_location() {
     ));
 }
 
+#[test]
+fn warehouse_locations_reject_the_reserved_catalog_prefix() {
+    for location in [
+        "s3://analytics/.rustfs-table",
+        "s3://analytics/.rustfs-table/",
+        "s3://analytics/.rustfs-table/warehouses/default",
+    ] {
+        let table_error = validate_table_warehouse_location("analytics", location).unwrap_err();
+        assert!(matches!(
+            table_error,
+            TableCatalogStoreError::Invalid(message) if message.contains("reserved table catalog prefix")
+        ));
+
+        let view_error = validate_view_warehouse_location("analytics", location).unwrap_err();
+        assert!(matches!(
+            view_error,
+            TableCatalogStoreError::Invalid(message) if message.contains("reserved table catalog prefix")
+        ));
+    }
+
+    assert!(validate_table_warehouse_location("analytics", "s3://analytics/.rustfs-table-other/table-id").is_ok());
+    assert!(validate_table_warehouse_location("analytics", "s3://analytics/user/.rustfs-table/table-id").is_ok());
+}
+
 #[tokio::test]
 async fn object_table_catalog_store_rejects_deep_table_warehouse_location() {
     let backend = TestCatalogObjectBackend::default();