Support S3 storage that does not have STS

dimas-b · dimas-b · commit e9122100e5d8 · 2025-09-24T17:41:00.000-04:00
This change is backward compatible with old catalogs that have storage configuration for S3 systems with STS.

* Add new property to S3 storage config: `stsUnavailable` (defaults to "available").

* Do not call STS when unavailable in `AwsCredentialsStorageIntegration`, but still put other properties (e.g. s3.endpoint) into `AccessConfig`

* Fail create/load table requests if credential vending is requested but STS is not available.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -40,6 +40,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. 
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java b/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;
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java b/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 (storageConfig.shouldUseSts()) {
+      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);
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java
@@ -98,6 +98,18 @@ 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();
+
+  /** Convenience getter for {@link #getStsUnavailable()} that handles defaults. */
+  @JsonIgnore
+  public boolean shouldUseSts() {
+    return !Boolean.TRUE.equals(getStsUnavailable());
+  }
+
   /** Endpoint URI for STS API calls */
   @Nullable
   public abstract String getStsEndpoint();
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java b/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().shouldUseSts()).isTrue();
+    assertThat(newBuilder().stsUnavailable(null).build().shouldUseSts()).isTrue();
+    assertThat(newBuilder().stsUnavailable(false).build().shouldUseSts()).isTrue();
+    assertThat(newBuilder().stsUnavailable(true).build().shouldUseSts()).isFalse();
+  }
+
   @Test
   public void testRoleArnParsing() {
     AwsStorageConfigurationInfo awsConfig =
diff --git a/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogMinIOSpecialIT.java b/runtime/service/src/intTest/java/org/apache/polaris/service/it/RestCatalogMinIOSpecialIT.java
@@ -19,15 +19,19 @@
 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;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.google.common.collect.ImmutableMap;
 import io.quarkus.test.junit.QuarkusIntegrationTest;
@@ -165,20 +169,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 +195,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 +241,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 +256,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 +286,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)
@@ -283,6 +302,62 @@ public void testInternalEndpoints() throws IOException {
     }
   }
 
+  @Test
+  public void testCreateTableFailureWithCredentialVendingWithoutSts() throws IOException {
+    try (RESTCatalog restCatalog =
+        createCatalog(
+            Optional.of(endpoint),
+            Optional.of("http://sts.example.com"), // not called
+            false,
+            Optional.of(VENDED_CREDENTIALS),
+            false)) {
+      StorageConfigInfo storageConfig =
+          managementApi.getCatalog(catalogName).getStorageConfigInfo();
+      assertThat((AwsStorageConfigInfo) storageConfig)
+          .extracting(
+              AwsStorageConfigInfo::getEndpoint,
+              AwsStorageConfigInfo::getStsEndpoint,
+              AwsStorageConfigInfo::getEndpointInternal,
+              AwsStorageConfigInfo::getPathStyleAccess,
+              AwsStorageConfigInfo::getStsUnavailable)
+          .containsExactly(endpoint, "http://sts.example.com", null, false, true);
+
+      catalogApi.createNamespace(catalogName, "test-ns");
+      TableIdentifier id = TableIdentifier.of("test-ns", "t2");
+      // Credential vending is not supported without STS
+      assertThatThrownBy(() -> restCatalog.createTable(id, SCHEMA))
+          .hasMessageContaining("but no credentials are available")
+          .hasMessageContaining(id.toString());
+    }
+  }
+
+  @Test
+  public void testLoadTableFailureWithCredentialVendingWithoutSts() throws IOException {
+    try (RESTCatalog restCatalog =
+        createCatalog(
+            Optional.of(endpoint),
+            Optional.of("http://sts.example.com"), // not called
+            false,
+            Optional.empty(),
+            false)) {
+
+      catalogApi.createNamespace(catalogName, "test-ns");
+      TableIdentifier id = TableIdentifier.of("test-ns", "t3");
+      restCatalog.createTable(id, SCHEMA);
+
+      // Credential vending is not supported without STS
+      assertThatThrownBy(
+              () ->
+                  catalogApi.loadTable(
+                      catalogName,
+                      id,
+                      "ALL",
+                      Map.of("X-Iceberg-Access-Delegation", VENDED_CREDENTIALS.protocolValue())))
+          .hasMessageContaining("but no credentials are available")
+          .hasMessageContaining(id.toString());
+    }
+  }
+
   public LoadTableResponse doTestCreateTable(
       RESTCatalog restCatalog, Optional<AccessDelegationMode> dm) {
     catalogApi.createNamespace(catalogName, "test-ns");
@@ -319,18 +394,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 +423,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));
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java
@@ -804,7 +804,12 @@ private LoadTableResponse.Builder buildLoadTableResponseWithDelegationCredential
           credentialDelegation.getAccessConfig(
               tableIdentifier, tableMetadata, actions, refreshCredentialsEndpoint);
       Map<String, String> credentialConfig = accessConfig.credentials();
-      if (!credentialConfig.isEmpty() && delegationModes.contains(VENDED_CREDENTIALS)) {
+      if (delegationModes.contains(VENDED_CREDENTIALS)) {
+        Preconditions.checkArgument(
+            !credentialConfig.isEmpty(),
+            "Credential vending was requested for table %s, but no credentials are available",
+            tableIdentifier);
+
         responseBuilder.addAllConfig(credentialConfig);
         responseBuilder.addCredential(
             ImmutableCredential.builder()
diff --git a/spec/polaris-management-service.yml b/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: >-
