Merge pull request apache#4 from snowflakedb/dhuo-merge-pinnacle-main

SNOW-1528940 Fix usability issues related to the pentest of invalid role ARN

polaris/polaris-core/src/main/java/io/polaris/core/storage/cache/StorageCredentialCache.java

@@ -13,6 +13,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import org.apache.iceberg.exceptions.UnprocessableEntityException;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.VisibleForTesting;
 import org.slf4j.Logger;
@@ -124,9 +125,10 @@ public Map<String, String> getOrGenerateSubScopeCreds(
           LOGGER
               .atDebug()
               .addKeyValue("errorMessage", scopedCredentialsResult.getExtraInformation())
-              .log("fail to getSubscopedCreds");
-          throw new RuntimeException(
-              "Fail to getSubscopedCreds: " + scopedCredentialsResult.getExtraInformation());
+              .log("Failed to get subscoped credentials");
+          throw new UnprocessableEntityException(
+              "Failed to get subscoped credentials: "
+                  + scopedCredentialsResult.getExtraInformation());
         };
     return cache.get(key, loader).convertToMapOfString();
   }

polaris/polaris-service/src/main/java/io/polaris/service/admin/PolarisAdminService.java

@@ -40,7 +40,9 @@
 import io.polaris.core.persistence.resolver.PolarisResolutionManifest;
 import io.polaris.core.persistence.resolver.ResolverPath;
 import io.polaris.core.persistence.resolver.ResolverStatus;
-import jakarta.ws.rs.BadRequestException;
+import io.polaris.core.storage.PolarisStorageConfigurationInfo;
+import io.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import io.polaris.core.storage.azure.AzureStorageConfigurationInfo;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -50,6 +52,7 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.BadRequestException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -513,6 +516,69 @@ public void deleteCatalog(String name) {
         .orElseThrow(() -> new NotFoundException("Catalog %s not found", name));
   }
 
+  /**
+   * Helper to validate business logic of what is allowed to be updated or throw a
+   * BadRequestException.
+   */
+  private void validateUpdateCatalogDiffOrThrow(
+      CatalogEntity currentEntity, CatalogEntity newEntity) {
+    // TODO: Expand the set of validations if there are other fields for other cloud providers
+    // that we can't successfully apply changes to.
+    PolarisStorageConfigurationInfo currentStorageConfig =
+        currentEntity.getStorageConfigurationInfo();
+    PolarisStorageConfigurationInfo newStorageConfig = newEntity.getStorageConfigurationInfo();
+
+    if (currentStorageConfig == null && newStorageConfig == null) {
+      return;
+    }
+
+    if (!currentStorageConfig.getClass().equals(newStorageConfig.getClass())) {
+      throw new BadRequestException(
+          "Cannot modify storage type of storage config from %s to %s",
+          currentStorageConfig, newStorageConfig);
+    }
+
+    if (currentStorageConfig instanceof AwsStorageConfigurationInfo
+        && newStorageConfig instanceof AwsStorageConfigurationInfo) {
+      AwsStorageConfigurationInfo currentAwsConfig =
+          (AwsStorageConfigurationInfo) currentStorageConfig;
+      AwsStorageConfigurationInfo newAwsConfig = (AwsStorageConfigurationInfo) newStorageConfig;
+
+      if ((currentAwsConfig.getRoleARN() != null
+              && !currentAwsConfig.getRoleARN().equals(newAwsConfig.getRoleARN()))
+          || (newAwsConfig.getRoleARN() != null
+              && !newAwsConfig.getRoleARN().equals(currentAwsConfig.getRoleARN()))) {
+        throw new BadRequestException(
+            "Cannot modify Role ARN in storage config from %s to %s",
+            currentStorageConfig, newStorageConfig);
+      }
+
+      if ((currentAwsConfig.getExternalId() != null
+              && !currentAwsConfig.getExternalId().equals(newAwsConfig.getExternalId()))
+          || (newAwsConfig.getExternalId() != null
+              && !newAwsConfig.getExternalId().equals(currentAwsConfig.getExternalId()))) {
+        throw new BadRequestException(
+            "Cannot modify ExternalId in storage config from %s to %s",
+            currentStorageConfig, newStorageConfig);
+      }
+    } else if (currentStorageConfig instanceof AzureStorageConfigurationInfo
+        && newStorageConfig instanceof AzureStorageConfigurationInfo) {
+      AzureStorageConfigurationInfo currentAzureConfig =
+          (AzureStorageConfigurationInfo) currentStorageConfig;
+      AzureStorageConfigurationInfo newAzureConfig =
+          (AzureStorageConfigurationInfo) newStorageConfig;
+
+      if ((currentAzureConfig.getTenantId() != null
+              && !currentAzureConfig.getTenantId().equals(newAzureConfig.getTenantId()))
+          || (newAzureConfig.getTenantId() != null
+              && !newAzureConfig.getTenantId().equals(currentAzureConfig.getTenantId()))) {
+        throw new BadRequestException(
+            "Cannot modify TenantId in storage config from %s to %s",
+            currentStorageConfig, newStorageConfig);
+      }
+    }
+  }
+
   public @NotNull CatalogEntity updateCatalog(String name, UpdateCatalogRequest updateRequest) {
     PolarisAuthorizableOperation op = PolarisAuthorizableOperation.UPDATE_CATALOG;
     authorizeBasicTopLevelEntityOperationOrThrow(op, name, PolarisEntityType.CATALOG);
@@ -539,6 +605,9 @@ public void deleteCatalog(String name) {
           updateRequest.getStorageConfigInfo(), defaultBaseLocation);
     }
     CatalogEntity updatedEntity = updateBuilder.build();
+
+    validateUpdateCatalogDiffOrThrow(currentCatalogEntity, updatedEntity);
+
     CatalogEntity returnedEntity =
         Optional.ofNullable(
                 CatalogEntity.of(

polaris/polaris-service/src/main/java/io/polaris/service/catalog/BasePolarisCatalog.java

@@ -58,6 +58,7 @@
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.exceptions.UnprocessableEntityException;
 import org.apache.iceberg.io.CloseableGroup;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.view.BaseMetastoreViewCatalog;
@@ -85,7 +86,11 @@ public boolean test(Exception ex) {
           // Default arguments from BaseMetastoreTableOperation only stop retries on
           // NotFoundException. We should more carefully identify the set of retriable
           // and non-retriable exceptions here.
-          return !(ex instanceof NotFoundException) && !(ex instanceof IllegalArgumentException);
+          return !(ex instanceof NotFoundException)
+              && !(ex instanceof IllegalArgumentException)
+              && !(ex instanceof AlreadyExistsException)
+              && !(ex instanceof ForbiddenException)
+              && !(ex instanceof UnprocessableEntityException);
         }
       };
   public static final String CLEANUP_ON_NAMESPACE_DROP = "CLEANUP_ON_NAMESPACE_DROP";

polaris/polaris-service/src/test/java/io/polaris/service/admin/PolarisServiceImplIntegrationTest.java

@@ -12,6 +12,7 @@
 import io.dropwizard.testing.junit5.DropwizardAppExtension;
 import io.dropwizard.testing.junit5.DropwizardExtensionsSupport;
 import io.polaris.core.admin.model.AwsStorageConfigInfo;
+import io.polaris.core.admin.model.AzureStorageConfigInfo;
 import io.polaris.core.admin.model.Catalog;
 import io.polaris.core.admin.model.CatalogProperties;
 import io.polaris.core.admin.model.CatalogRole;
@@ -494,14 +495,94 @@ public void serialization() throws JsonProcessingException {
         .containsEntry("default-base-location", "s3://my-bucket/path/to/data");
   }
 
+  @Test
+  public void testCreateAndUpdateAzureCatalog() {
+    StorageConfigInfo storageConfig =
+        new AzureStorageConfigInfo("azure:tenantid:12345", StorageConfigInfo.StorageTypeEnum.AZURE);
+    Catalog catalog =
+        PolarisCatalog.builder()
+            .setType(Catalog.TypeEnum.INTERNAL)
+            .setName("myazurecatalog")
+            .setStorageConfigInfo(storageConfig)
+            .setProperties(new CatalogProperties("abfss://container1@acct1.dfs.core.windows.net/"))
+            .build();
+
+    // 200 Successful create
+    try (Response response =
+        newRequest("http://localhost:%d/api/management/v1/catalogs")
+            .post(Entity.json(new CreateCatalogRequest(catalog)))) {
+      assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
+    }
+
+    // 200 successful GET after creation
+    Catalog fetchedCatalog = null;
+    try (Response response =
+        newRequest("http://localhost:%d/api/management/v1/catalogs/myazurecatalog").get()) {
+      assertThat(response).returns(200, Response::getStatus);
+      fetchedCatalog = response.readEntity(Catalog.class);
+
+      assertThat(fetchedCatalog.getName()).isEqualTo("myazurecatalog");
+      assertThat(fetchedCatalog.getProperties().toMap())
+          .isEqualTo(
+              Map.of("default-base-location", "abfss://container1@acct1.dfs.core.windows.net/"));
+      assertThat(fetchedCatalog.getEntityVersion()).isGreaterThan(0);
+    }
+
+    StorageConfigInfo modifiedStorageConfig =
+        new AzureStorageConfigInfo("azure:tenantid:22222", StorageConfigInfo.StorageTypeEnum.AZURE);
+    UpdateCatalogRequest badUpdateRequest =
+        new UpdateCatalogRequest(
+            fetchedCatalog.getEntityVersion(),
+            Map.of("default-base-location", "abfss://newcontainer@acct1.dfs.core.windows.net/"),
+            modifiedStorageConfig);
+    try (Response response =
+        newRequest("http://localhost:%d/api/management/v1/catalogs/myazurecatalog")
+            .put(Entity.json(badUpdateRequest))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+      ErrorResponse error = response.readEntity(ErrorResponse.class);
+      assertThat(error)
+          .isNotNull()
+          .extracting(ErrorResponse::message)
+          .asString()
+          .startsWith("Cannot modify");
+    }
+
+    UpdateCatalogRequest updateRequest =
+        new UpdateCatalogRequest(
+            fetchedCatalog.getEntityVersion(),
+            Map.of("default-base-location", "abfss://newcontainer@acct1.dfs.core.windows.net/"),
+            storageConfig);
+
+    // 200 successful update
+    try (Response response =
+        newRequest("http://localhost:%d/api/management/v1/catalogs/myazurecatalog")
+            .put(Entity.json(updateRequest))) {
+      assertThat(response).returns(200, Response::getStatus);
+      fetchedCatalog = response.readEntity(Catalog.class);
+
+      assertThat(fetchedCatalog.getProperties().toMap())
+          .isEqualTo(
+              Map.of("default-base-location", "abfss://newcontainer@acct1.dfs.core.windows.net/"));
+    }
+
+    // 204 Successful delete
+    try (Response response =
+        newRequest("http://localhost:%d/api/management/v1/catalogs/myazurecatalog").delete()) {
+      assertThat(response).returns(204, Response::getStatus);
+    }
+  }
+
   @Test
   public void testCreateListUpdateAndDeleteCatalog() {
+    StorageConfigInfo storageConfig =
+        new AwsStorageConfigInfo(
+            "arn:aws:iam::12345:role/role1", StorageConfigInfo.StorageTypeEnum.S3);
     Catalog catalog =
         PolarisCatalog.builder()
             .setType(Catalog.TypeEnum.INTERNAL)
             .setName("mycatalog")
-            .setStorageConfigInfo(
-                new AwsStorageConfigInfo("aws:role:arn", StorageConfigInfo.StorageTypeEnum.S3))
+            .setStorageConfigInfo(storageConfig)
             .setProperties(new CatalogProperties("s3://bucket1/"))
             .build();
 
@@ -542,11 +623,33 @@ public void testCreateListUpdateAndDeleteCatalog() {
           .satisfiesExactly(cat -> assertThat(cat).returns("mycatalog", Catalog::getName));
     }
 
+    // Reject update of fields that can't be currently updated
+    StorageConfigInfo modifiedStorageConfig =
+        new AwsStorageConfigInfo(
+            "arn:aws:iam::22222:role/newrole", StorageConfigInfo.StorageTypeEnum.S3);
+    UpdateCatalogRequest badUpdateRequest =
+        new UpdateCatalogRequest(
+            fetchedCatalog.getEntityVersion(),
+            Map.of("default-base-location", "s3://newbucket/"),
+            modifiedStorageConfig);
+    try (Response response =
+        newRequest("http://localhost:%d/api/management/v1/catalogs/mycatalog")
+            .put(Entity.json(badUpdateRequest))) {
+      assertThat(response)
+          .returns(Response.Status.BAD_REQUEST.getStatusCode(), Response::getStatus);
+      ErrorResponse error = response.readEntity(ErrorResponse.class);
+      assertThat(error)
+          .isNotNull()
+          .extracting(ErrorResponse::message)
+          .asString()
+          .startsWith("Cannot modify");
+    }
+
     UpdateCatalogRequest updateRequest =
         new UpdateCatalogRequest(
             fetchedCatalog.getEntityVersion(),
             Map.of("default-base-location", "s3://newbucket/"),
-            null);
+            storageConfig);
 
     // 200 successful update
     try (Response response =