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

.gitignore

@@ -23,6 +23,7 @@ regtests/output/
 # This file, if checked in after running for example regtests, contains unmanaged dependencies that eventually
 # cause unnecessary "security alerts" like https://github.com/apache/polaris/pull/718.
 regtests/client/python/poetry.lock
+regtests/minio/miniodata/*
 
 # Python stuff (see note about poetry.lock above as well!)
 /poetry.lock
@@ -64,6 +65,9 @@ gradle/wrapper/gradle-wrapper-*.sha256
 *.ipr
 *.iws
 
+# VScode
+.vscode
+
 # Gradle
 /.gradle
 /build-logic/.gradle

build.gradle.kts

@@ -109,6 +109,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")

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

@@ -127,6 +127,7 @@ protected FeatureConfiguration(
           .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,22 @@ 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())
+            .setS3ProfileName(s3Config.getS3ProfileName())
+            .setS3PathStyleAccess(s3Config.getS3PathStyleAccess())
+            .setAllowedLocations(s3Config.getAllowedLocations())
+            .setS3CredentialsCatalogAccessKeyEnvVar(s3Config.getS3CredentialsCatalogAccessKeyId())
+            .setS3CredentialsCatalogSecretAccessKeyEnvVar(
+                s3Config.getS3CredentialsCatalogSecretAccessKey())
+            .setS3Region(s3Config.getS3Region())
+            .setS3RoleArn(s3Config.getS3RoleArn())
+            .build();
+      }
       if (configInfo instanceof AzureStorageConfigurationInfo) {
         AzureStorageConfigurationInfo azureConfig = (AzureStorageConfigurationInfo) configInfo;
         return AzureStorageConfigInfo.builder()
@@ -250,6 +268,21 @@ public Builder setStorageConfigurationInfo(
             awsConfig.validateArn(awsConfigModel.getRoleArn());
             config = awsConfig;
             break;
+
+          case S3_COMPATIBLE:
+            S3CompatibleStorageConfigInfo s3ConfigModel =
+                (S3CompatibleStorageConfigInfo) storageConfigModel;
+            config =
+                new S3CompatibleStorageConfigurationInfo(
+                    s3ConfigModel.getS3Endpoint(),
+                    s3ConfigModel.getS3ProfileName(),
+                    s3ConfigModel.getS3CredentialsCatalogAccessKeyEnvVar(),
+                    s3ConfigModel.getS3CredentialsCatalogSecretAccessKeyEnvVar(),
+                    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

@@ -27,6 +27,9 @@ public enum PolarisCredentialProperty {
       String.class,
       "s3.session-token-expires-at-ms",
       "the time the aws session token expires, in milliseconds"),
+  AWS_ENDPOINT(String.class, "s3.endpoint", "the aws s3 endpoint"),
+  AWS_PATH_STYLE_ACCESS(
+      Boolean.class, "s3.path-style-access", "whether or not to use 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/StorageUtil.java

@@ -20,6 +20,11 @@
 
 import jakarta.annotation.Nonnull;
 import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Stream;
+import software.amazon.awssdk.policybuilder.iam.*;
 
 public class StorageUtil {
   /**
@@ -62,4 +67,136 @@ public class StorageUtil {
   public static @Nonnull String getBucket(URI uri) {
     return uri.getAuthority();
   }
+
+  /**
+   * Given a path, return it without leading slash
+   *
+   * @param path A path to parse
+   * @return Same path without leading slash
+   */
+  private static @Nonnull String trimLeadingSlash(String path) {
+    if (path.startsWith("/")) {
+      path = path.substring(1);
+    }
+    return path;
+  }
+
+  /**
+   * Given an uri, and format an S3 path
+   *
+   * @param uri A path to parse
+   * @return A bucket and a path joined by slash
+   */
+  private static @Nonnull String parseS3Path(URI uri) {
+    String bucket = getBucket(uri);
+    String path = trimLeadingSlash(uri.getPath());
+    return String.join("/", bucket, path);
+  }
+
+  /**
+   * Given a roleArn, return the prefix
+   *
+   * @param roleArn A roleArn to parse
+   * @return The prefix of the roleArn
+   */
+  private static String getArnPrefixFor(String roleArn) {
+    if (roleArn.contains("aws-cn")) {
+      return "arn:aws-cn:s3:::";
+    } else if (roleArn.contains("aws-us-gov")) {
+      return "arn:aws-us-gov:s3:::";
+    } else {
+      return "arn:aws:s3:::";
+    }
+  }
+
+  /**
+   * 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
+   * locations are empty, a non-empty policy will be generated that grants GetObject and optionally
+   * ListBucket privileges with no resources. This prevents us from sending an empty policy to AWS
+   * and just assuming the role with full privileges.
+   *
+   * @param roleArn A roleArn
+   * @param allowList Allow list or not
+   * @param readLocations A list of input read locations
+   * @param writeLocations A list of input write locations
+   * @return A policy limiting scope access
+   */
+  // TODO - add KMS key access
+  public static IamPolicy policyString(
+      String roleArn, 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 = getArnPrefixFor(roleArn);
+    Stream.concat(readLocations.stream(), writeLocations.stream())
+        .distinct()
+        .forEach(
+            location -> {
+              URI uri = URI.create(location);
+              allowGetObjectStatementBuilder.addResource(
+                  // TODO add support for CN and GOV
+                  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);
+            // TODO add support for CN and GOV
+            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();
+  }
 }