[Catalog Federation] Add feature flag to disallow setting sub-RBAC for federated catalog at catalog level (#2696)

HonahX · web-flow · commit aeae51f728c1 · 2025-09-29T21:16:17.000-05:00
In #2688 (comment), we've identified that configuring polaris.config.enable-sub-catalog-rbac-for-federated-catalogs at catalog level should not be allowed in all cases, especially when the owner is not the same subject as the catalog user or admin. This PR add a feature flag, ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS to allow owner to disable catalog level setting polaris.config.enable-sub-catalog-rbac-for-federated-catalogs
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -40,6 +40,9 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### New Features
 
 - Added a Management API endpoint to reset principal credentials, controlled by the `ENABLE_CREDENTIAL_RESET` (default: true) feature flag.
+- The `ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS` was added to support sub-catalog (initially namespace and table) RBAC for federated catalogs.
+  The setting can be configured on a per-catalog basis by setting the catalog property: `polaris.config.enable-sub-catalog-rbac-for-federated-catalogs`.
+  The realm-level feature flag `ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS` (default: true) controls whether this functionality can be enabled or modified at the catalog level.
 
 - Added support for S3-compatible storage that does not have STS (use `stsUavailable: true` in catalog storage configuration)
 
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java b/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java
@@ -386,4 +386,15 @@ public static void enforceFeatureEnabledOrThrow(
                   + "Defaults to enabled, but service providers may want to disable it.")
           .defaultValue(true)
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Boolean>
+      ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS =
+          PolarisConfiguration.<Boolean>builder()
+              .key("ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS")
+              .description(
+                  "If set to true (default), Polaris will allow setting or changing "
+                      + "catalog property polaris.config.enable-sub-catalog-rbac-for-federated-catalogs."
+                      + "If set to false, Polaris will disallow setting or changing the above catalog property")
+              .defaultValue(true)
+              .buildFeatureConfiguration();
 }
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java b/runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java
@@ -25,6 +25,7 @@
 import jakarta.ws.rs.core.SecurityContext;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.responses.ErrorResponse;
@@ -124,6 +125,7 @@ public Response createCatalog(
     Catalog catalog = request.getCatalog();
     validateStorageConfig(catalog.getStorageConfigInfo());
     validateExternalCatalog(catalog);
+    validateCatalogProperties(catalog.getProperties());
     Catalog newCatalog = CatalogEntity.of(adminService.createCatalog(request)).asCatalog();
     LOGGER.info("Created new catalog {}", newCatalog);
     return Response.status(Response.Status.CREATED).entity(newCatalog).build();
@@ -180,6 +182,23 @@ private void validateExternalCatalog(Catalog catalog) {
     }
   }
 
+  private void validateCatalogProperties(Map<String, String> catalogProperties) {
+    if (catalogProperties != null) {
+      if (!realmConfig.getConfig(
+              FeatureConfiguration.ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS)
+          && catalogProperties.containsKey(
+              FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS
+                  .catalogConfig())) {
+
+        throw new IllegalArgumentException(
+            String.format(
+                "Explicitly setting %s is not allowed because %s is set to false.",
+                FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS.catalogConfig(),
+                FeatureConfiguration.ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS.key()));
+      }
+    }
+  }
+
   private void validateConnectionConfigInfo(ConnectionConfigInfo connectionConfigInfo) {
 
     String connectionType = connectionConfigInfo.getConnectionType().name();
@@ -231,6 +250,7 @@ public Response updateCatalog(
     if (updateRequest.getStorageConfigInfo() != null) {
       validateStorageConfig(updateRequest.getStorageConfigInfo());
     }
+    validateCatalogProperties(updateRequest.getProperties());
     return Response.ok(adminService.updateCatalog(catalogName, updateRequest).asCatalog()).build();
   }
 
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java b/runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java
@@ -31,11 +31,16 @@
 import java.util.function.Supplier;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
 import org.apache.polaris.core.admin.model.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.ExternalCatalog;
 import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
+import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
 import org.apache.polaris.core.admin.model.PolarisCatalog;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.admin.model.UpdateCatalogRequest;
@@ -66,8 +71,16 @@ public class ManagementServiceTest {
   public void setup() {
     services =
         TestServices.builder()
-            .config(Map.of("SUPPORTED_CATALOG_STORAGE_TYPES", List.of("S3", "GCS", "AZURE")))
-            .config(Map.of("ALLOW_SETTING_S3_ENDPOINTS", Boolean.FALSE))
+            .config(
+                Map.of(
+                    "SUPPORTED_CATALOG_STORAGE_TYPES",
+                    List.of("S3", "GCS", "AZURE"),
+                    "ALLOW_SETTING_S3_ENDPOINTS",
+                    Boolean.FALSE,
+                    "ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS",
+                    Boolean.FALSE,
+                    "ENABLE_CATALOG_FEDERATION",
+                    Boolean.TRUE))
             .build();
   }
 
@@ -232,6 +245,129 @@ public void testUpdateCatalogWithDisallowedStorageConfig() {
         .hasMessage("Explicitly setting S3 endpoints is not allowed.");
   }
 
+  @Test
+  public void testCreateCatalogWithDisallowedConfigs() {
+    AwsStorageConfigInfo awsConfigModel =
+        AwsStorageConfigInfo.builder()
+            .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+            .setExternalId("externalId")
+            .setUserArn("userArn")
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+            .setAllowedLocations(List.of("s3://my-old-bucket/path/to/data"))
+            .build();
+    ConnectionConfigInfo connectionConfigInfo =
+        IcebergRestConnectionConfigInfo.builder(
+                ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST)
+            .setUri("https://myorg-my_account.snowflakecomputing.com/polaris/api/catalog")
+            .setRemoteCatalogName("my-remote-catalog")
+            .setAuthenticationParameters(
+                OAuthClientCredentialsParameters.builder(
+                        AuthenticationParameters.AuthenticationTypeEnum.OAUTH)
+                    .setClientId("my-client-id")
+                    .setClientSecret("my-client-secret")
+                    .setScopes(List.of("PRINCIPAL_ROLE:ALL"))
+                    .build())
+            .build();
+    String catalogName = "mycatalog";
+    CatalogProperties catalogProperties =
+        CatalogProperties.builder("s3://bucket/path/to/data")
+            .addProperty("polaris.config.enable-sub-catalog-rbac-for-federated-catalogs", "true")
+            .build();
+    Catalog catalog =
+        ExternalCatalog.builder()
+            .setType(Catalog.TypeEnum.EXTERNAL)
+            .setName(catalogName)
+            .setProperties(catalogProperties)
+            .setStorageConfigInfo(awsConfigModel)
+            .setConnectionConfigInfo(connectionConfigInfo)
+            .build();
+    Supplier<Response> createCatalog =
+        () ->
+            services
+                .catalogsApi()
+                .createCatalog(
+                    new CreateCatalogRequest(catalog),
+                    services.realmContext(),
+                    services.securityContext());
+    assertThatThrownBy(createCatalog::get)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(
+            "Explicitly setting polaris.config.enable-sub-catalog-rbac-for-federated-catalogs is not allowed because ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS is set to false.");
+  }
+
+  @Test
+  public void testUpdateCatalogWithDisallowedConfigs() {
+    AwsStorageConfigInfo awsConfigModel =
+        AwsStorageConfigInfo.builder()
+            .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+            .setExternalId("externalId")
+            .setUserArn("userArn")
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+            .setAllowedLocations(List.of("s3://my-old-bucket/path/to/data"))
+            .build();
+    ConnectionConfigInfo connectionConfigInfo =
+        IcebergRestConnectionConfigInfo.builder(
+                ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST)
+            .setUri("https://myorg-my_account.snowflakecomputing.com/polaris/api/catalog")
+            .setRemoteCatalogName("my-remote-catalog")
+            .setAuthenticationParameters(
+                OAuthClientCredentialsParameters.builder(
+                        AuthenticationParameters.AuthenticationTypeEnum.OAUTH)
+                    .setClientId("my-client-id")
+                    .setClientSecret("my-client-secret")
+                    .setScopes(List.of("PRINCIPAL_ROLE:ALL"))
+                    .build())
+            .build();
+    String catalogName = "mycatalog";
+    CatalogProperties catalogProperties =
+        CatalogProperties.builder("s3://bucket/path/to/data").build();
+    Catalog catalog =
+        ExternalCatalog.builder()
+            .setType(Catalog.TypeEnum.EXTERNAL)
+            .setName(catalogName)
+            .setProperties(catalogProperties)
+            .setStorageConfigInfo(awsConfigModel)
+            .setConnectionConfigInfo(connectionConfigInfo)
+            .build();
+    try (Response response =
+        services
+            .catalogsApi()
+            .createCatalog(
+                new CreateCatalogRequest(catalog),
+                services.realmContext(),
+                services.securityContext())) {
+      assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
+    }
+    Catalog fetchedCatalog;
+    try (Response response =
+        services
+            .catalogsApi()
+            .getCatalog(catalogName, services.realmContext(), services.securityContext())) {
+      assertThat(response).returns(Response.Status.OK.getStatusCode(), Response::getStatus);
+      fetchedCatalog = (Catalog) response.getEntity();
+
+      assertThat(fetchedCatalog.getName()).isEqualTo(catalogName);
+      assertThat(fetchedCatalog.getProperties().toMap())
+          .isEqualTo(Map.of("default-base-location", "s3://bucket/path/to/data"));
+      assertThat(fetchedCatalog.getEntityVersion()).isGreaterThan(0);
+    }
+
+    UpdateCatalogRequest update =
+        UpdateCatalogRequest.builder()
+            .setProperties(
+                Map.of("polaris.config.enable-sub-catalog-rbac-for-federated-catalogs", "true"))
+            .build();
+    assertThatThrownBy(
+            () ->
+                services
+                    .catalogsApi()
+                    .updateCatalog(
+                        catalogName, update, services.realmContext(), services.securityContext()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(
+            "Explicitly setting polaris.config.enable-sub-catalog-rbac-for-federated-catalogs is not allowed because ALLOW_SETTING_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS is set to false.");
+  }
+
   private PolarisAdminService setupPolarisAdminService(
       PolarisMetaStoreManager metaStoreManager, PolarisCallContext callContext) {
     return new PolarisAdminService(
