(feat) Add support for no KMS with s3-compatible backend

CHANGELOG.md

@@ -67,6 +67,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
   the [hierarchical namespace](https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-namespace)
   feature is enabled in Azure.
 - Relaxed `client_id`, `client_secret` regex/pattern validation on reset endpoint call
+- Added support for S3-compatible storage that does not have KMS (use `kmsUavailable: true` in catalog storage configuration)
 
 ### Changes
 

client/python/apache_polaris/cli/command/__init__.py

@@ -77,6 +77,7 @@ def options_get(key: str, f: Callable[[Any], Any] = lambda x: x) -> Any:
                 endpoint_internal=options_get(Arguments.ENDPOINT_INTERNAL),
                 sts_endpoint=options_get(Arguments.STS_ENDPOINT),
                 sts_unavailable=options_get(Arguments.STS_UNAVAILABLE),
+                kms_unavailable=options_get(Arguments.KMS_UNAVAILABLE),
                 path_style_access=options_get(Arguments.PATH_STYLE_ACCESS),
                 current_kms_key=options_get(Arguments.KMS_KEY_CURRENT),
                 allowed_kms_keys=options_get(Arguments.KMS_KEY_ALLOWED),

client/python/apache_polaris/cli/command/catalogs.py

@@ -93,6 +93,7 @@ class CatalogsCommand(Command):
     endpoint_internal: str
     sts_endpoint: str
     sts_unavailable: bool
+    kms_unavailable: bool
     path_style_access: bool
     current_kms_key: str
     allowed_kms_keys: List[str]
@@ -189,7 +190,8 @@ def validate(self) -> None:
                     f" {Argument.to_flag_name(Arguments.KMS_KEY_CURRENT)},"
                     f" {Argument.to_flag_name(Arguments.KMS_KEY_ALLOWED)},"
                     f" {Argument.to_flag_name(Arguments.STS_ENDPOINT)},"
-                    f" {Argument.to_flag_name(Arguments.STS_UNAVAILABLE)}, and"
+                    f" {Argument.to_flag_name(Arguments.STS_UNAVAILABLE)},"
+                    f" {Argument.to_flag_name(Arguments.KMS_UNAVAILABLE)}, and"
                     f" {Argument.to_flag_name(Arguments.PATH_STYLE_ACCESS)}"
                 )
         elif self.storage_type == StorageType.AZURE.value:
@@ -260,6 +262,7 @@ def _build_storage_config_info(self) -> Optional[StorageConfigInfo]:
                 endpoint_internal=self.endpoint_internal,
                 sts_endpoint=self.sts_endpoint,
                 sts_unavailable=self.sts_unavailable,
+                kms_unavailable=self.kms_unavailable,
                 path_style_access=self.path_style_access,
                 current_kms_key=self.current_kms_key,
                 allowed_kms_keys=self.allowed_kms_keys,

client/python/apache_polaris/cli/constants.py

@@ -180,6 +180,7 @@ class Arguments:
     ENDPOINT_INTERNAL = "endpoint_internal"
     STS_ENDPOINT = "sts_endpoint"
     STS_UNAVAILABLE = "no_sts"
+    KMS_UNAVAILABLE = "no_kms"
     KMS_KEY_CURRENT = "current_kms_key"
     KMS_KEY_ALLOWED = "allowed_kms_key"
     PATH_STYLE_ACCESS = "path_style_access"
@@ -258,6 +259,7 @@ class Create:
                 "(Only for S3) The STS endpoint to use when connecting to STS"
             )
             STS_UNAVAILABLE = "(Only for S3) Indicates that Polaris should not use STS (e.g. if STS is not available)"
+            KMS_UNAVAILABLE = "(Only for S3) Indicates that Polaris should not use KMS (e.g. if KMS is not available)"
             PATH_STYLE_ACCESS = "(Only for S3) Whether to use path-style-access for S3"
             KMS_KEY_CURRENT = "(Only for AWS S3) The AWS KMS key ARN to be used for encrypting new S3 data"
             KMS_KEY_ALLOWED = "(Only for AWS S3) AWS KMS key ARN(s) that this catalog and its clients are allowed to use for reading S3 data (zero or more)"

client/python/apache_polaris/cli/options/option_tree.py

@@ -220,6 +220,11 @@ def get_tree() -> List[Option]:
                                 bool,
                                 Hints.Catalogs.Create.STS_UNAVAILABLE,
                             ),
+                            Argument(
+                                Arguments.KMS_UNAVAILABLE,
+                                bool,
+                                Hints.Catalogs.Create.KMS_UNAVAILABLE,
+                            ),
                             Argument(
                                 Arguments.PATH_STYLE_ACCESS,
                                 bool,

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -171,6 +171,7 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
             .setPathStyleAccess(awsConfig.getPathStyleAccess())
             .setStsUnavailable(awsConfig.getStsUnavailable())
             .setEndpointInternal(awsConfig.getEndpointInternal())
+            .setKmsUnavailable(awsConfig.getKmsUnavailable())
             .build();
       }
       if (configInfo instanceof AzureStorageConfigurationInfo) {
@@ -320,6 +321,7 @@ public Builder setStorageConfigurationInfo(
                     .pathStyleAccess(awsConfigModel.getPathStyleAccess())
                     .stsUnavailable(awsConfigModel.getStsUnavailable())
                     .endpointInternal(awsConfigModel.getEndpointInternal())
+                    .kmsUnavailable(awsConfigModel.getKmsUnavailable())
                     .build();
             config = awsConfig;
             break;

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java

@@ -199,6 +199,10 @@ private boolean shouldUseSts(AwsStorageConfigurationInfo storageConfig) {
     return !Boolean.TRUE.equals(storageConfig.getStsUnavailable());
   }
 
+  private boolean shouldUseKms(AwsStorageConfigurationInfo storageConfig) {
+    return !Boolean.TRUE.equals(storageConfig.getKmsUnavailable());
+  }
+
   /**
    * generate an IamPolicy from the input readLocations and writeLocations, optionally with list
    * support. Credentials will be scoped to exactly the resources provided. If read and write
@@ -271,9 +275,13 @@ private IamPolicy policyString(
                     arnPrefix + StorageUtil.concatFilePrefixes(parseS3Path(uri), "*", "/")));
           });
       policyBuilder.addStatement(allowPutObjectStatementBuilder.build());
-      addKmsKeyPolicy(currentKmsKey, allowedKmsKeys, policyBuilder, true, region, accountId);
+      if (shouldUseKms(storageConfigurationInfo)) {
+        addKmsKeyPolicy(currentKmsKey, allowedKmsKeys, policyBuilder, true, region, accountId);
+      }
     } else {
-      addKmsKeyPolicy(currentKmsKey, allowedKmsKeys, policyBuilder, false, region, accountId);
+      if (shouldUseKms(storageConfigurationInfo)) {
+        addKmsKeyPolicy(currentKmsKey, allowedKmsKeys, policyBuilder, false, region, accountId);
+      }
     }
     if (!bucketListStatementBuilder.isEmpty()) {
       bucketListStatementBuilder

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java

@@ -113,6 +113,12 @@ public URI getInternalEndpointUri() {
    */
   public abstract @Nullable Boolean getStsUnavailable();
 
+  /**
+   * Flag indicating whether KMS is available or not. It is modeled in the negative to simplify
+   * support for unset values ({@code null} being interpreted as {@code false}).
+   */
+  public abstract @Nullable Boolean getKmsUnavailable();
+
   /** Endpoint URI for STS API calls */
   @Nullable
   public abstract String getStsEndpoint();

polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java

@@ -126,6 +126,14 @@ public void testStsUnavailable() {
     assertThat(newBuilder().stsUnavailable(true).build().getStsUnavailable()).isTrue();
   }
 
+  @Test
+  public void testKmsUnavailable() {
+    assertThat(newBuilder().build().getKmsUnavailable()).isNull();
+    assertThat(newBuilder().kmsUnavailable(null).build().getKmsUnavailable()).isNull();
+    assertThat(newBuilder().kmsUnavailable(false).build().getKmsUnavailable()).isFalse();
+    assertThat(newBuilder().kmsUnavailable(true).build().getKmsUnavailable()).isTrue();
+  }
+
   @Test
   public void testRoleArnParsing() {
     AwsStorageConfigurationInfo awsConfig =

polaris-core/src/test/java/org/apache/polaris/service/storage/aws/AwsCredentialsStorageIntegrationTest.java

@@ -751,6 +751,82 @@ public void testNoClientRegion(String awsPartition) {
     ;
   }
 
+  @Test
+  public void testKmsUnavailablePolicyLogic() {
+    StsClient stsClient = Mockito.mock(StsClient.class);
+    String roleARN = "arn:aws:iam::012345678901:role/jdoe";
+    String externalId = "externalId";
+    String bucket = "bucket";
+    String warehouseKeyPrefix = "path/to/warehouse";
+    String region = "us-east-2";
+    String currentKmsKey = "arn:aws:kms:us-east-1:012345678901:key/current-key";
+    // Test with kmsUnavailable=true and write permissions - should NOT add KMS policies
+    Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
+        .thenAnswer(
+            invocation -> {
+              AssumeRoleRequest request = invocation.getArgument(0);
+              IamPolicy policy = IamPolicy.fromJson(request.policy());
+              assertThat(policy.statements())
+                  .noneMatch(
+                      stmt ->
+                          stmt.actions().stream()
+                              .anyMatch(action -> action.value().startsWith("kms:")));
+              return ASSUME_ROLE_RESPONSE;
+            });
+
+    new AwsCredentialsStorageIntegration(
+            AwsStorageConfigurationInfo.builder()
+                .addAllowedLocation(s3Path(bucket, warehouseKeyPrefix))
+                .roleARN(roleARN)
+                .externalId(externalId)
+                .region(region)
+                .currentKmsKey(currentKmsKey)
+                .kmsUnavailable(true)
+                .build(),
+            stsClient)
+        .getSubscopedCreds(
+            EMPTY_REALM_CONFIG,
+            true,
+            Set.of(s3Path(bucket, warehouseKeyPrefix + "/table")),
+            Set.of(s3Path(bucket, warehouseKeyPrefix + "/table")),
+            POLARIS_PRINCIPAL,
+            Optional.empty(),
+            CredentialVendingContext.empty());
+
+    // Test with kmsUnavailable=true and read-only permissions - should NOT add KMS policies
+    Mockito.reset(stsClient);
+    Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
+        .thenAnswer(
+            invocation -> {
+              AssumeRoleRequest request = invocation.getArgument(0);
+              IamPolicy policy = IamPolicy.fromJson(request.policy());
+              assertThat(policy.statements())
+                  .noneMatch(
+                      stmt ->
+                          stmt.actions().stream()
+                              .anyMatch(action -> action.value().startsWith("kms:")));
+              return ASSUME_ROLE_RESPONSE;
+            });
+
+    new AwsCredentialsStorageIntegration(
+            AwsStorageConfigurationInfo.builder()
+                .addAllowedLocation(s3Path(bucket, warehouseKeyPrefix))
+                .roleARN(roleARN)
+                .externalId(externalId)
+                .region(region)
+                .kmsUnavailable(true)
+                .build(),
+            stsClient)
+        .getSubscopedCreds(
+            EMPTY_REALM_CONFIG,
+            true,
+            Set.of(s3Path(bucket, warehouseKeyPrefix + "/table")),
+            Set.of(),
+            POLARIS_PRINCIPAL,
+            Optional.empty(),
+            CredentialVendingContext.empty());
+  }
+
   @Test
   public void testKmsKeyPolicyLogic() {
     StsClient stsClient = Mockito.mock(StsClient.class);

site/content/in-dev/unreleased/command-line-interface.md

@@ -148,6 +148,7 @@ options:
       --endpoint-internal  (Only for S3) The S3 endpoint used by Polaris to use when connecting to S3, if different from the one that clients use
       --sts-endpoint  (Only for S3) The STS endpoint to use when connecting to STS
       --no-sts  (Only for S3) Indicates that Polaris should not use STS (e.g. if STS is not available)
+      --no-kms  (Only for S3) Indicates that Polaris should not use KMS (e.g. if KMS is not available)
       --path-style-access  (Only for S3) Whether to use path-style-access for S3
       --allowed-location  An allowed location for files tracked by the catalog. Multiple locations can be provided by specifying this option more than once.
       --role-arn  (Only for AWS S3) A role ARN to use when connecting to S3

spec/polaris-management-service.yml

@@ -1149,6 +1149,10 @@ components:
                 Whether S3 requests to files in this catalog should use 'path-style addressing for buckets'.
               example: true
               default: false
+            kmsUnavailable:
+              type: boolean
+              description: >-
+                if set to `true`, instructs Polaris Servers to avoid adding KMS key policies.
 
     AzureStorageConfigInfo:
       type: object