[FEATURE REQUEST #32] On-Premises S3 / S3 Compatible...

.gitignore

@@ -20,6 +20,7 @@
 regtests/derby.log
 regtests/metastore_db
 regtests/output/
+regtests/minio/miniodata/*
 
 # Python stuff
 /poetry.lock
@@ -58,6 +59,9 @@ gradle/wrapper/gradle-wrapper-*.sha256
 *.ipr
 *.iws
 
+# VScode
+.vscode
+
 # Gradle
 /.gradle
 /build-logic/.gradle

build.gradle.kts

@@ -104,6 +104,9 @@ tasks.named<RatTask>("rat").configure {
   excludes.add("regtests/metastore_db/**")
   excludes.add("regtests/client/python/.openapi-generator/**")
   excludes.add("regtests/output/**")
+  excludes.add("regtests/minio/miniodata/**")
+  excludes.add("regtests/minio/**/*.crt")
+  excludes.add("regtests/minio/**/*.key")
 
   excludes.add("**/*.ipynb")
   excludes.add("**/*.iml")

helm/polaris/values.yaml

@@ -294,6 +294,7 @@ polarisServerConfig:
     ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING: false
     SUPPORTED_CATALOG_STORAGE_TYPES:
       - S3
+      - S3_COMPATIBLE
       - GCS
       - AZURE
       - FILE

polaris-core/src/main/java/org/apache/polaris/core/PolarisConfiguration.java

@@ -177,6 +177,7 @@ public static <T> Builder<T> builder() {
           .defaultValue(
               List.of(
                   StorageConfigInfo.StorageTypeEnum.S3.name(),
+                  StorageConfigInfo.StorageTypeEnum.S3_COMPATIBLE.name(),
                   StorageConfigInfo.StorageTypeEnum.AZURE.name(),
                   StorageConfigInfo.StorageTypeEnum.GCS.name(),
                   StorageConfigInfo.StorageTypeEnum.FILE.name()))

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

@@ -38,12 +38,14 @@
 import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
 import org.apache.polaris.core.admin.model.GcpStorageConfigInfo;
 import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.S3CompatibleStorageConfigInfo;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.storage.FileStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
 import org.apache.polaris.core.storage.azure.AzureStorageConfigurationInfo;
 import org.apache.polaris.core.storage.gcp.GcpStorageConfigurationInfo;
+import org.apache.polaris.core.storage.s3compatible.S3CompatibleStorageConfigurationInfo;
 
 /**
  * Catalog specific subclass of the {@link PolarisEntity} that handles conversion from the {@link
@@ -141,6 +143,25 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
             .setRegion(awsConfig.getRegion())
             .build();
       }
+      if (configInfo instanceof S3CompatibleStorageConfigurationInfo) {
+        S3CompatibleStorageConfigurationInfo s3Config =
+            (S3CompatibleStorageConfigurationInfo) configInfo;
+        return S3CompatibleStorageConfigInfo.builder()
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.S3_COMPATIBLE)
+            .setS3Endpoint(s3Config.getS3Endpoint())
+            .setS3PathStyleAccess(s3Config.getS3PathStyleAccess())
+            .setSkipCredentialSubscopingIndirection(
+                s3Config.getSkipCredentialSubscopingIndirection())
+            .setAllowedLocations(s3Config.getAllowedLocations())
+            .setS3CredentialsCatalogAccessKeyId(s3Config.getS3CredentialsCatalogAccessKeyId())
+            .setS3CredentialsCatalogSecretAccessKey(
+                s3Config.getS3CredentialsCatalogSecretAccessKey())
+            .setS3CredentialsClientAccessKeyId(s3Config.getS3CredentialsClientSecretAccessKey())
+            .setS3CredentialsClientSecretAccessKey(s3Config.getS3CredentialsClientAccessKeyId())
+            .setS3Region(s3Config.getS3Region())
+            .setS3RoleArn(s3Config.getS3RoleArn())
+            .build();
+      }
       if (configInfo instanceof AzureStorageConfigurationInfo) {
         AzureStorageConfigurationInfo azureConfig = (AzureStorageConfigurationInfo) configInfo;
         return AzureStorageConfigInfo.builder()
@@ -250,6 +271,24 @@ public Builder setStorageConfigurationInfo(
             awsConfig.validateArn(awsConfigModel.getRoleArn());
             config = awsConfig;
             break;
+
+          case S3_COMPATIBLE:
+            S3CompatibleStorageConfigInfo s3ConfigModel =
+                (S3CompatibleStorageConfigInfo) storageConfigModel;
+            config =
+                new S3CompatibleStorageConfigurationInfo(
+                    PolarisStorageConfigurationInfo.StorageType.S3_COMPATIBLE,
+                    s3ConfigModel.getS3Endpoint(),
+                    s3ConfigModel.getS3CredentialsCatalogAccessKeyId(),
+                    s3ConfigModel.getS3CredentialsCatalogSecretAccessKey(),
+                    s3ConfigModel.getSkipCredentialSubscopingIndirection(),
+                    s3ConfigModel.getS3CredentialsClientAccessKeyId(),
+                    s3ConfigModel.getS3CredentialsClientSecretAccessKey(),
+                    s3ConfigModel.getS3PathStyleAccess(),
+                    s3ConfigModel.getS3Region(),
+                    s3ConfigModel.getS3RoleArn(),
+                    new ArrayList<>(allowedLocations));
+            break;
           case AZURE:
             AzureStorageConfigInfo azureConfigModel = (AzureStorageConfigInfo) storageConfigModel;
             AzureStorageConfigurationInfo azureconfigInfo =

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisCredentialProperty.java

@@ -23,6 +23,8 @@ public enum PolarisCredentialProperty {
   AWS_KEY_ID(String.class, "s3.access-key-id", "the aws access key id"),
   AWS_SECRET_KEY(String.class, "s3.secret-access-key", "the aws access key secret"),
   AWS_TOKEN(String.class, "s3.session-token", "the aws scoped access token"),
+  AWS_ENDPOINT(String.class, "s3.endpoint", "the aws s3 endpoint"),
+  AWS_PATH_STYLE_ACCESS(Boolean.class, "s3.path-style-access", "the aws s3 path style access"),
   CLIENT_REGION(
       String.class, "client.region", "region to configure client for making requests to AWS"),
 

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfo.java

@@ -47,6 +47,7 @@
 import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
 import org.apache.polaris.core.storage.azure.AzureStorageConfigurationInfo;
 import org.apache.polaris.core.storage.gcp.GcpStorageConfigurationInfo;
+import org.apache.polaris.core.storage.s3compatible.S3CompatibleStorageConfigurationInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +63,7 @@
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
 @JsonSubTypes({
   @JsonSubTypes.Type(value = AwsStorageConfigurationInfo.class),
+  @JsonSubTypes.Type(value = S3CompatibleStorageConfigurationInfo.class),
   @JsonSubTypes.Type(value = AzureStorageConfigurationInfo.class),
   @JsonSubTypes.Type(value = GcpStorageConfigurationInfo.class),
   @JsonSubTypes.Type(value = FileStorageConfigurationInfo.class),
@@ -241,6 +243,7 @@ public void validateMaxAllowedLocations(int maxAllowedLocations) {
   /** Polaris' storage type, each has a fixed prefix for its location */
   public enum StorageType {
     S3("s3://"),
+    S3_COMPATIBLE("s3://"),
     AZURE(List.of("abfs://", "wasb://", "abfss://", "wasbs://")),
     GCS("gs://"),
     FILE("file://"),

polaris-core/src/main/java/org/apache/polaris/core/storage/s3compatible/S3CompatibleCredentialsStorageIntegration.java

@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.storage.s3compatible;
+
+import jakarta.annotation.Nonnull;
+import java.net.URI;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.storage.InMemoryStorageIntegration;
+import org.apache.polaris.core.storage.PolarisCredentialProperty;
+import org.apache.polaris.core.storage.StorageUtil;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.policybuilder.iam.IamConditionOperator;
+import software.amazon.awssdk.policybuilder.iam.IamEffect;
+import software.amazon.awssdk.policybuilder.iam.IamPolicy;
+import software.amazon.awssdk.policybuilder.iam.IamResource;
+import software.amazon.awssdk.policybuilder.iam.IamStatement;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleResponse;
+
+/** S3 compatible implementation of PolarisStorageIntegration */
+public class S3CompatibleCredentialsStorageIntegration
+    extends InMemoryStorageIntegration<S3CompatibleStorageConfigurationInfo> {
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(S3CompatibleCredentialsStorageIntegration.class);
+  private StsClient stsClient;
+  private String caI, caS; // catalogAccessKeyId & catalogSecretAccessKey
+  private String clI, clS; // clientAccessKeyId & clientSecretAccessKey
+
+  public S3CompatibleCredentialsStorageIntegration() {
+    super(S3CompatibleCredentialsStorageIntegration.class.getName());
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
+      @NotNull PolarisDiagnostics diagnostics,
+      @NotNull S3CompatibleStorageConfigurationInfo storageConfig,
+      boolean allowListOperation,
+      @NotNull Set<String> allowedReadLocations,
+      @NotNull Set<String> allowedWriteLocations) {
+
+    caI = System.getenv(storageConfig.getS3CredentialsCatalogAccessKeyId());
+    caS = System.getenv(storageConfig.getS3CredentialsCatalogSecretAccessKey());
+
+    EnumMap<PolarisCredentialProperty, String> propertiesMap =
+        new EnumMap<>(PolarisCredentialProperty.class);
+    propertiesMap.put(PolarisCredentialProperty.AWS_ENDPOINT, storageConfig.getS3Endpoint());
+    propertiesMap.put(
+        PolarisCredentialProperty.AWS_PATH_STYLE_ACCESS,
+        storageConfig.getS3PathStyleAccess().toString());
+    if (storageConfig.getS3Region() != null) {
+      propertiesMap.put(PolarisCredentialProperty.CLIENT_REGION, storageConfig.getS3Region());
+    }
+
+    if (storageConfig.getSkipCredentialSubscopingIndirection() == true) {
+      LOGGER.debug("S3Compatible - skipCredentialSubscopingIndirection !");
+      clI = System.getenv(storageConfig.getS3CredentialsClientAccessKeyId());
+      clS = System.getenv(storageConfig.getS3CredentialsClientSecretAccessKey());
+      if (clI != null && clS != null) {
+        propertiesMap.put(PolarisCredentialProperty.AWS_KEY_ID, clI);
+        propertiesMap.put(PolarisCredentialProperty.AWS_SECRET_KEY, clS);
+      } else {
+        propertiesMap.put(PolarisCredentialProperty.AWS_KEY_ID, caI);
+        propertiesMap.put(PolarisCredentialProperty.AWS_SECRET_KEY, caS);
+      }
+    } else {
+      LOGGER.debug("S3Compatible - assumeRole !");
+      createStsClient(storageConfig);
+      AssumeRoleResponse response =
+          stsClient.assumeRole(
+              AssumeRoleRequest.builder()
+                  .roleSessionName("PolarisCredentialsSTS")
+                  .roleArn(
+                      (storageConfig.getS3RoleArn() == null) ? "" : storageConfig.getS3RoleArn())
+                  .policy(
+                      policyString(allowListOperation, allowedReadLocations, allowedWriteLocations)
+                          .toJson())
+                  .build());
+
+      propertiesMap.put(PolarisCredentialProperty.AWS_KEY_ID, response.credentials().accessKeyId());
+      propertiesMap.put(
+          PolarisCredentialProperty.AWS_SECRET_KEY, response.credentials().secretAccessKey());
+      propertiesMap.put(PolarisCredentialProperty.AWS_TOKEN, response.credentials().sessionToken());
+      LOGGER.debug(
+          "S3Compatible - assumeRole - Token Expiration at : {}",
+          response.credentials().expiration().toString());
+    }
+
+    return propertiesMap;
+  }
+
+  public void createStsClient(S3CompatibleStorageConfigurationInfo storageConfig) {
+    LOGGER.debug("S3Compatible - createStsClient()");
+    try {
+      StsClientBuilder stsBuilder = software.amazon.awssdk.services.sts.StsClient.builder();
+      stsBuilder.endpointOverride(URI.create(storageConfig.getS3Endpoint()));
+      if (caI != null && caS != null) {
+        // else default provider build credentials from profile or standard AWS env var
+        stsBuilder.credentialsProvider(
+            StaticCredentialsProvider.create(AwsBasicCredentials.create(caI, caS)));
+      }
+      this.stsClient = stsBuilder.build();
+      LOGGER.debug("S3Compatible - stsClient successfully built");
+
+    } catch (Exception e) {
+      System.err.println("S3Compatible - stsClient - build failure : " + e.getMessage());
+    }
+  }
+
+  /*
+   * function from AwsCredentialsStorageIntegration but without roleArn parameter
+   */
+  private IamPolicy policyString(
+      boolean allowList, Set<String> readLocations, Set<String> writeLocations) {
+    IamPolicy.Builder policyBuilder = IamPolicy.builder();
+    IamStatement.Builder allowGetObjectStatementBuilder =
+        IamStatement.builder()
+            .effect(IamEffect.ALLOW)
+            .addAction("s3:GetObject")
+            .addAction("s3:GetObjectVersion");
+    Map<String, IamStatement.Builder> bucketListStatementBuilder = new HashMap<>();
+    Map<String, IamStatement.Builder> bucketGetLocationStatementBuilder = new HashMap<>();
+
+    String arnPrefix = "arn:aws:s3:::";
+    Stream.concat(readLocations.stream(), writeLocations.stream())
+        .distinct()
+        .forEach(
+            location -> {
+              URI uri = URI.create(location);
+              allowGetObjectStatementBuilder.addResource(
+                  IamResource.create(
+                      arnPrefix + StorageUtil.concatFilePrefixes(parseS3Path(uri), "*", "/")));
+              final var bucket = arnPrefix + StorageUtil.getBucket(uri);
+              if (allowList) {
+                bucketListStatementBuilder
+                    .computeIfAbsent(
+                        bucket,
+                        (String key) ->
+                            IamStatement.builder()
+                                .effect(IamEffect.ALLOW)
+                                .addAction("s3:ListBucket")
+                                .addResource(key))
+                    .addCondition(
+                        IamConditionOperator.STRING_LIKE,
+                        "s3:prefix",
+                        StorageUtil.concatFilePrefixes(trimLeadingSlash(uri.getPath()), "*", "/"));
+              }
+              bucketGetLocationStatementBuilder.computeIfAbsent(
+                  bucket,
+                  key ->
+                      IamStatement.builder()
+                          .effect(IamEffect.ALLOW)
+                          .addAction("s3:GetBucketLocation")
+                          .addResource(key));
+            });
+
+    if (!writeLocations.isEmpty()) {
+      IamStatement.Builder allowPutObjectStatementBuilder =
+          IamStatement.builder()
+              .effect(IamEffect.ALLOW)
+              .addAction("s3:PutObject")
+              .addAction("s3:DeleteObject");
+      writeLocations.forEach(
+          location -> {
+            URI uri = URI.create(location);
+            allowPutObjectStatementBuilder.addResource(
+                IamResource.create(
+                    arnPrefix + StorageUtil.concatFilePrefixes(parseS3Path(uri), "*", "/")));
+          });
+      policyBuilder.addStatement(allowPutObjectStatementBuilder.build());
+    }
+    if (!bucketListStatementBuilder.isEmpty()) {
+      bucketListStatementBuilder
+          .values()
+          .forEach(statementBuilder -> policyBuilder.addStatement(statementBuilder.build()));
+    } else if (allowList) {
+      // add list privilege with 0 resources
+      policyBuilder.addStatement(
+          IamStatement.builder().effect(IamEffect.ALLOW).addAction("s3:ListBucket").build());
+    }
+
+    bucketGetLocationStatementBuilder
+        .values()
+        .forEach(statementBuilder -> policyBuilder.addStatement(statementBuilder.build()));
+    return policyBuilder.addStatement(allowGetObjectStatementBuilder.build()).build();
+  }
+
+  /* function from AwsCredentialsStorageIntegration */
+  private static @Nonnull String parseS3Path(URI uri) {
+    String bucket = StorageUtil.getBucket(uri);
+    String path = trimLeadingSlash(uri.getPath());
+    return String.join("/", bucket, path);
+  }
+
+  /* function from AwsCredentialsStorageIntegration */
+  private static @Nonnull String trimLeadingSlash(String path) {
+    if (path.startsWith("/")) {
+      path = path.substring(1);
+    }
+    return path;
+  }
+}