Support S3 storage that does not have STS

CHANGELOG.md

@@ -41,6 +41,8 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 - Added a Management API endpoint to reset principal credentials, controlled by the `ENABLE_CREDENTIAL_RESET` (default: true) feature flag.
 
+- Added support for S3-compatible storage that does not have STS (use `stsUavailable: true` in catalog storage configuration)
+
 ### Changes
 
 * The following APIs will now return the newly-created objects as part of the successful 201 response: createCatalog, createPrincipalRole, createCatalogRole. 

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

@@ -155,6 +155,7 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
             .setEndpoint(awsConfig.getEndpoint())
             .setStsEndpoint(awsConfig.getStsEndpoint())
             .setPathStyleAccess(awsConfig.getPathStyleAccess())
+            .setStsUnavailable(awsConfig.getStsUnavailable())
             .setEndpointInternal(awsConfig.getEndpointInternal())
             .build();
       }
@@ -299,6 +300,7 @@ public Builder setStorageConfigurationInfo(
                     .endpoint(awsConfigModel.getEndpoint())
                     .stsEndpoint(awsConfigModel.getStsEndpoint())
                     .pathStyleAccess(awsConfigModel.getPathStyleAccess())
+                    .stsUnavailable(awsConfigModel.getStsUnavailable())
                     .endpointInternal(awsConfigModel.getEndpointInternal())
                     .build();
             config = awsConfig;

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

@@ -79,43 +79,46 @@ public AccessConfig getSubscopedCreds(
     int storageCredentialDurationSeconds =
         realmConfig.getConfig(STORAGE_CREDENTIAL_DURATION_SECONDS);
     AwsStorageConfigurationInfo storageConfig = config();
-    AssumeRoleRequest.Builder request =
-        AssumeRoleRequest.builder()
-            .externalId(storageConfig.getExternalId())
-            .roleArn(storageConfig.getRoleARN())
-            .roleSessionName("PolarisAwsCredentialsStorageIntegration")
-            .policy(
-                policyString(
-                        storageConfig.getAwsPartition(),
-                        allowListOperation,
-                        allowedReadLocations,
-                        allowedWriteLocations)
-                    .toJson())
-            .durationSeconds(storageCredentialDurationSeconds);
-    credentialsProvider.ifPresent(
-        cp -> request.overrideConfiguration(b -> b.credentialsProvider(cp)));
-
     String region = storageConfig.getRegion();
-    @SuppressWarnings("resource")
-    // Note: stsClientProvider returns "thin" clients that do not need closing
-    StsClient stsClient =
-        stsClientProvider.stsClient(StsDestination.of(storageConfig.getStsEndpointUri(), region));
-
-    AssumeRoleResponse response = stsClient.assumeRole(request.build());
     AccessConfig.Builder accessConfig = AccessConfig.builder();
-    accessConfig.put(StorageAccessProperty.AWS_KEY_ID, response.credentials().accessKeyId());
-    accessConfig.put(
-        StorageAccessProperty.AWS_SECRET_KEY, response.credentials().secretAccessKey());
-    accessConfig.put(StorageAccessProperty.AWS_TOKEN, response.credentials().sessionToken());
-    Optional.ofNullable(response.credentials().expiration())
-        .ifPresent(
-            i -> {
-              accessConfig.put(
-                  StorageAccessProperty.EXPIRATION_TIME, String.valueOf(i.toEpochMilli()));
-              accessConfig.put(
-                  StorageAccessProperty.AWS_SESSION_TOKEN_EXPIRES_AT_MS,
-                  String.valueOf(i.toEpochMilli()));
-            });
+
+    if (shouldUseSts(storageConfig)) {
+      AssumeRoleRequest.Builder request =
+          AssumeRoleRequest.builder()
+              .externalId(storageConfig.getExternalId())
+              .roleArn(storageConfig.getRoleARN())
+              .roleSessionName("PolarisAwsCredentialsStorageIntegration")
+              .policy(
+                  policyString(
+                          storageConfig.getAwsPartition(),
+                          allowListOperation,
+                          allowedReadLocations,
+                          allowedWriteLocations)
+                      .toJson())
+              .durationSeconds(storageCredentialDurationSeconds);
+      credentialsProvider.ifPresent(
+          cp -> request.overrideConfiguration(b -> b.credentialsProvider(cp)));
+
+      @SuppressWarnings("resource")
+      // Note: stsClientProvider returns "thin" clients that do not need closing
+      StsClient stsClient =
+          stsClientProvider.stsClient(StsDestination.of(storageConfig.getStsEndpointUri(), region));
+
+      AssumeRoleResponse response = stsClient.assumeRole(request.build());
+      accessConfig.put(StorageAccessProperty.AWS_KEY_ID, response.credentials().accessKeyId());
+      accessConfig.put(
+          StorageAccessProperty.AWS_SECRET_KEY, response.credentials().secretAccessKey());
+      accessConfig.put(StorageAccessProperty.AWS_TOKEN, response.credentials().sessionToken());
+      Optional.ofNullable(response.credentials().expiration())
+          .ifPresent(
+              i -> {
+                accessConfig.put(
+                    StorageAccessProperty.EXPIRATION_TIME, String.valueOf(i.toEpochMilli()));
+                accessConfig.put(
+                    StorageAccessProperty.AWS_SESSION_TOKEN_EXPIRES_AT_MS,
+                    String.valueOf(i.toEpochMilli()));
+              });
+    }
 
     if (region != null) {
       accessConfig.put(StorageAccessProperty.CLIENT_REGION, region);
@@ -149,6 +152,10 @@ public AccessConfig getSubscopedCreds(
     return accessConfig.build();
   }
 
+  private boolean shouldUseSts(AwsStorageConfigurationInfo storageConfig) {
+    return !Boolean.TRUE.equals(storageConfig.getStsUnavailable());
+  }
+
   /**
    * 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

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

@@ -98,6 +98,12 @@ public URI getInternalEndpointUri() {
   /** Flag indicating whether path-style bucket access should be forced in S3 clients. */
   public abstract @Nullable Boolean getPathStyleAccess();
 
+  /**
+   * Flag indicating whether STS 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 getStsUnavailable();
+
   /** Endpoint URI for STS API calls */
   @Nullable
   public abstract String getStsEndpoint();

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

@@ -118,6 +118,14 @@ public void testPathStyleAccess() {
     assertThat(newBuilder().pathStyleAccess(true).build().getPathStyleAccess()).isTrue();
   }
 
+  @Test
+  public void testStsUnavailable() {
+    assertThat(newBuilder().build().getStsUnavailable()).isNull();
+    assertThat(newBuilder().stsUnavailable(null).build().getStsUnavailable()).isNull();
+    assertThat(newBuilder().stsUnavailable(false).build().getStsUnavailable()).isFalse();
+    assertThat(newBuilder().stsUnavailable(true).build().getStsUnavailable()).isTrue();
+  }
+
   @Test
   public void testRoleArnParsing() {
     AwsStorageConfigurationInfo awsConfig =

runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogMinIOSpecialIT.java

@@ -19,12 +19,15 @@
 package org.apache.polaris.service.it;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.iceberg.CatalogProperties.TABLE_DEFAULT_PREFIX;
 import static org.apache.iceberg.aws.AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT;
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.ACCESS_KEY_ID;
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.ENDPOINT;
 import static org.apache.iceberg.aws.s3.S3FileIOProperties.SECRET_ACCESS_KEY;
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.apache.polaris.core.storage.StorageAccessProperty.AWS_KEY_ID;
+import static org.apache.polaris.core.storage.StorageAccessProperty.AWS_SECRET_KEY;
 import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
 import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -165,20 +168,24 @@ private RESTCatalog createCatalog(
       Optional<String> endpoint,
       Optional<String> stsEndpoint,
       boolean pathStyleAccess,
-      Optional<AccessDelegationMode> delegationMode) {
-    return createCatalog(endpoint, stsEndpoint, pathStyleAccess, Optional.empty(), delegationMode);
+      Optional<AccessDelegationMode> delegationMode,
+      boolean stsEnabled) {
+    return createCatalog(
+        endpoint, stsEndpoint, pathStyleAccess, Optional.empty(), delegationMode, stsEnabled);
   }
 
   private RESTCatalog createCatalog(
       Optional<String> endpoint,
       Optional<String> stsEndpoint,
       boolean pathStyleAccess,
       Optional<String> endpointInternal,
-      Optional<AccessDelegationMode> delegationMode) {
+      Optional<AccessDelegationMode> delegationMode,
+      boolean stsEnabled) {
     AwsStorageConfigInfo.Builder storageConfig =
         AwsStorageConfigInfo.builder()
             .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
             .setPathStyleAccess(pathStyleAccess)
+            .setStsUnavailable(!stsEnabled)
             .setAllowedLocations(List.of(storageBase.toString()));
 
     endpoint.ifPresent(storageConfig::setEndpoint);
@@ -187,6 +194,12 @@ private RESTCatalog createCatalog(
 
     CatalogProperties.Builder catalogProps =
         CatalogProperties.builder(storageBase.toASCIIString() + "/" + catalogName);
+    if (!stsEnabled) {
+      catalogProps.addProperty(
+          TABLE_DEFAULT_PREFIX + AWS_KEY_ID.getPropertyName(), MINIO_ACCESS_KEY);
+      catalogProps.addProperty(
+          TABLE_DEFAULT_PREFIX + AWS_SECRET_KEY.getPropertyName(), MINIO_SECRET_KEY);
+    }
     Catalog catalog =
         PolarisCatalog.builder()
             .setType(Catalog.TypeEnum.INTERNAL)
@@ -227,9 +240,12 @@ public void cleanUp() {
   }
 
   @ParameterizedTest
-  @ValueSource(booleans = {true, false})
-  public void testCreateTable(boolean pathStyle) throws IOException {
-    LoadTableResponse response = doTestCreateTable(pathStyle, Optional.empty());
+  @CsvSource("true,  true,")
+  @CsvSource("false, true,")
+  @CsvSource("true,  false,")
+  @CsvSource("false, false,")
+  public void testCreateTable(boolean pathStyle, boolean stsEnabled) throws IOException {
+    LoadTableResponse response = doTestCreateTable(pathStyle, Optional.empty(), stsEnabled);
     assertThat(response.config()).doesNotContainKey(SECRET_ACCESS_KEY);
     assertThat(response.config()).doesNotContainKey(ACCESS_KEY_ID);
     assertThat(response.config()).doesNotContainKey(REFRESH_CREDENTIALS_ENDPOINT);
@@ -239,18 +255,19 @@ public void testCreateTable(boolean pathStyle) throws IOException {
   @ParameterizedTest
   @ValueSource(booleans = {true, false})
   public void testCreateTableVendedCredentials(boolean pathStyle) throws IOException {
-    LoadTableResponse response = doTestCreateTable(pathStyle, Optional.of(VENDED_CREDENTIALS));
+    LoadTableResponse response =
+        doTestCreateTable(pathStyle, Optional.of(VENDED_CREDENTIALS), true);
     assertThat(response.config())
         .containsEntry(
             REFRESH_CREDENTIALS_ENDPOINT,
             "v1/" + catalogName + "/namespaces/test-ns/tables/t1/credentials");
     assertThat(response.credentials()).hasSize(1);
   }
 
-  private LoadTableResponse doTestCreateTable(boolean pathStyle, Optional<AccessDelegationMode> dm)
-      throws IOException {
+  private LoadTableResponse doTestCreateTable(
+      boolean pathStyle, Optional<AccessDelegationMode> dm, boolean stsEnabled) throws IOException {
     try (RESTCatalog restCatalog =
-        createCatalog(Optional.of(endpoint), Optional.empty(), pathStyle, dm)) {
+        createCatalog(Optional.of(endpoint), Optional.empty(), pathStyle, dm, stsEnabled)) {
       LoadTableResponse loadTableResponse = doTestCreateTable(restCatalog, dm);
       if (pathStyle) {
         assertThat(loadTableResponse.config())
@@ -268,7 +285,8 @@ public void testInternalEndpoints() throws IOException {
             Optional.of(endpoint),
             false,
             Optional.of(endpoint),
-            Optional.empty())) {
+            Optional.empty(),
+            true)) {
       StorageConfigInfo storageConfig =
           managementApi.getCatalog(catalogName).getStorageConfigInfo();
       assertThat((AwsStorageConfigInfo) storageConfig)
@@ -319,18 +337,22 @@ public LoadTableResponse doTestCreateTable(
   }
 
   @ParameterizedTest
-  @CsvSource("true,")
-  @CsvSource("false,")
-  @CsvSource("true,VENDED_CREDENTIALS")
-  @CsvSource("false,VENDED_CREDENTIALS")
-  public void testAppendFiles(boolean pathStyle, AccessDelegationMode delegationMode)
+  @CsvSource("true,  true,")
+  @CsvSource("false, true,")
+  @CsvSource("true,  false,")
+  @CsvSource("false, false,")
+  @CsvSource("true,  true,  VENDED_CREDENTIALS")
+  @CsvSource("false, true,  VENDED_CREDENTIALS")
+  public void testAppendFiles(
+      boolean pathStyle, boolean stsEnabled, AccessDelegationMode delegationMode)
       throws IOException {
     try (RESTCatalog restCatalog =
         createCatalog(
             Optional.of(endpoint),
             Optional.of(endpoint),
             pathStyle,
-            Optional.ofNullable(delegationMode))) {
+            Optional.ofNullable(delegationMode),
+            stsEnabled)) {
       catalogApi.createNamespace(catalogName, "test-ns");
       TableIdentifier id = TableIdentifier.of("test-ns", "t1");
       Table table = restCatalog.createTable(id, SCHEMA);
@@ -344,7 +366,8 @@ public void testAppendFiles(boolean pathStyle, AccessDelegationMode delegationMo
               table
                   .locationProvider()
                   .newDataLocation(
-                      String.format("test-file-%s-%s.txt", pathStyle, delegationMode)));
+                      String.format(
+                          "test-file-%s-%s-%s.txt", pathStyle, delegationMode, stsEnabled)));
       OutputFile f1 = io.newOutputFile(loc.toString());
       try (PositionOutputStream os = f1.create()) {
         os.write("Hello World".getBytes(UTF_8));

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

@@ -160,6 +160,10 @@ private void validateStorageConfig(StorageConfigInfo storageConfigInfo) {
             || s3Config.getEndpointInternal() != null) {
           throw new IllegalArgumentException("Explicitly setting S3 endpoints is not allowed.");
         }
+
+        if (s3Config.getStsUnavailable() != null) {
+          throw new IllegalArgumentException("Explicitly disabling STS is not allowed.");
+        }
       }
     }
   }

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

@@ -140,6 +140,12 @@ public void testCreateCatalogWithDisallowedS3Endpoints() {
     assertThatThrownBy(createCatalog::get)
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Explicitly setting S3 endpoints is not allowed.");
+
+    storageConfig.setEndpointInternal(null);
+    storageConfig.setStsUnavailable(false);
+    assertThatThrownBy(createCatalog::get)
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Explicitly disabling STS is not allowed.");
   }
 
   @Test

spec/polaris-management-service.yml

@@ -1119,6 +1119,13 @@ components:
                 endpoint for STS requests made by the Polaris Server (optional). If not set, defaults to
                 'endpointInternal' (which in turn defaults to `endpoint`).
               example: "https://sts.example.com:1234"
+            stsUnavailable:
+              type: boolean
+              description: >-
+                if set to `true`, instructs Polaris Servers to avoid using the STS endpoints when obtaining credentials
+                for accessing data and metadata files within the related catalog. Setting this property to `true`
+                effectively disables vending storage credentials to clients. This setting is intended for configuring
+                catalogs with S3-compatible storage implementations that do not support STS.
             endpointInternal:
               type: string
               description: >-