Add feature flag to disallow custom S3 endpoints (#2442)

* Add new realm-level flag: `ALLOW_SETTING_S3_ENDPOINTS` (default: true)

* Enforce in `PolarisServiceImpl.validateStorageConfig()`

Fixes #2436

Author: dimas-b

CHANGELOG.md

@@ -51,6 +51,8 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### New Features
 
 - Added Catalog configuration for S3 and STS endpoints. This also allows using non-AWS S3 implementations.
+  The realm-level feature flag `ALLOW_SETTING_S3_ENDPOINTS` (default: true) may be used to disable this
+  functionality.
 
 - The `IMPLICIT` authentication type enables users to create federated catalogs without explicitly
 providing authentication parameters to Polaris. When the authentication type is set to `IMPLICIT`,

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -79,6 +79,16 @@ public static void enforceFeatureEnabledOrThrow(
           .defaultValue(false)
           .buildFeatureConfiguration();
 
+  public static final FeatureConfiguration<Boolean> ALLOW_SETTING_S3_ENDPOINTS =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_SETTING_S3_ENDPOINTS")
+          .description(
+              "If set to true (default), Polaris will permit S3 storage configurations to have custom endpoints.\n"
+                  + "If set to false, Polaris will not accept catalog create and update requests that contain \n"
+                  + "S3 endpoint properties.")
+          .defaultValue(true)
+          .buildFeatureConfiguration();
+
   @SuppressWarnings("deprecation")
   public static final FeatureConfiguration<Boolean> ALLOW_TABLE_LOCATION_OVERLAP =
       PolarisConfiguration.<Boolean>builder()

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.rest.responses.ErrorResponse;
 import org.apache.polaris.core.admin.model.AddGrantRequest;
 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.CatalogGrant;
 import org.apache.polaris.core.admin.model.CatalogRole;
@@ -180,6 +181,16 @@ private void validateStorageConfig(StorageConfigInfo storageConfigInfo) {
       throw new IllegalArgumentException(
           "Unsupported storage type: " + storageConfigInfo.getStorageType());
     }
+
+    if (!realmConfig.getConfig(FeatureConfiguration.ALLOW_SETTING_S3_ENDPOINTS)) {
+      if (storageConfigInfo instanceof AwsStorageConfigInfo s3Config) {
+        if (s3Config.getEndpoint() != null
+            || s3Config.getStsEndpoint() != null
+            || s3Config.getEndpointInternal() != null) {
+          throw new IllegalArgumentException("Explicitly setting S3 endpoints is not allowed.");
+        }
+      }
+    }
   }
 
   private void validateExternalCatalog(Catalog catalog) {

runtime/service/src/test/java/org/apache/polaris/service/admin/ManagementServiceTest.java

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Supplier;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
@@ -68,6 +69,7 @@ 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))
             .build();
   }
 
@@ -97,6 +99,51 @@ public void testCreateCatalogWithDisallowedStorageConfig() {
         .hasMessage("Unsupported storage type: FILE");
   }
 
+  @Test
+  public void testCreateCatalogWithDisallowedS3Endpoints() {
+    AwsStorageConfigInfo.Builder storageConfig =
+        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"));
+    String catalogName = "test-catalog";
+    Supplier<Catalog> catalog =
+        () ->
+            PolarisCatalog.builder()
+                .setType(Catalog.TypeEnum.INTERNAL)
+                .setName(catalogName)
+                .setProperties(new CatalogProperties("s3://bucket/path/to/data"))
+                .setStorageConfigInfo(storageConfig.build())
+                .build();
+    Supplier<Response> createCatalog =
+        () ->
+            services
+                .catalogsApi()
+                .createCatalog(
+                    new CreateCatalogRequest(catalog.get()),
+                    services.realmContext(),
+                    services.securityContext());
+
+    storageConfig.setEndpoint("http://example.com");
+    assertThatThrownBy(createCatalog::get)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Explicitly setting S3 endpoints is not allowed.");
+
+    storageConfig.setEndpoint(null);
+    storageConfig.setStsEndpoint("http://example.com");
+    assertThatThrownBy(createCatalog::get)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Explicitly setting S3 endpoints is not allowed.");
+
+    storageConfig.setStsEndpoint(null);
+    storageConfig.setEndpointInternal("http://example.com");
+    assertThatThrownBy(createCatalog::get)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Explicitly setting S3 endpoints is not allowed.");
+  }
+
   @Test
   public void testUpdateCatalogWithDisallowedStorageConfig() {
     AwsStorageConfigInfo awsConfigModel =
@@ -162,6 +209,23 @@ public void testUpdateCatalogWithDisallowedStorageConfig() {
                         services.securityContext()))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Unsupported storage type: FILE");
+
+    UpdateCatalogRequest update2 =
+        new UpdateCatalogRequest(
+            fetchedCatalog.getEntityVersion(),
+            Map.of(),
+            AwsStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.S3)
+                .setRoleArn("arn:aws:iam::123456789012:role/my-role")
+                .setEndpoint("http://example.com")
+                .build());
+    assertThatThrownBy(
+            () ->
+                services
+                    .catalogsApi()
+                    .updateCatalog(
+                        catalogName, update2, services.realmContext(), services.securityContext()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Explicitly setting S3 endpoints is not allowed.");
   }
 
   private PolarisMetaStoreManager setupMetaStoreManager() {