Enforce that S3 credentials are vended when requested

This is a follow-up change to #2672 striving to improve user-facing error reporting for S3 storage systems without STS.

* Add property to `AccessConfig` to indicate whether the backing storage integration can produce credentials.

* Add a check to `IcebergCatalogHandler` (leading to 400) that storage credentials are vended when requested and the backend is capable of vending credentials in principle.

* Update `PolarisStorageIntegrationProviderImpl` to indicate that FILE storage does not support credential vending (requesitng redential vending with FILE storage does not produce any credentials and does not flag an error, which matches current Polaris behaviour).

* Only those S3 systems where STS is not available (or disabled / not permitted) are affected.

* Other storage integrations are not affected by this PR.

Author: dimas-b

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalMetaStoreManagerImpl.java

@@ -2093,6 +2093,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
     PolarisStorageIntegration<PolarisStorageConfigurationInfo> storageIntegration =
         ms.loadPolarisStorageIntegrationInCurrentTxn(callCtx, reloadedEntity.getEntity());
 
+    storageIntegration.config();
     // cannot be null
     getDiagnostics()
         .checkNotNull(

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

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.Optional;
 import org.apache.polaris.immutables.PolarisImmutable;
+import org.immutables.value.Value;
 
 @PolarisImmutable
 public interface AccessConfig {
@@ -38,6 +39,15 @@ public interface AccessConfig {
 
   Optional<Instant> expiresAt();
 
+  /**
+   * Indicates whether the storage integration subsystem that produced this object is capable of
+   * credential vending in principle.
+   */
+  @Value.Default
+  default boolean supportsCredentialVending() {
+    return true;
+  }
+
   default String get(StorageAccessProperty key) {
     if (key.isCredential()) {
       return credentials().get(key.getPropertyName());
@@ -64,6 +74,9 @@ interface Builder {
     @CanIgnoreReturnValue
     Builder expiresAt(Instant expiresAt);
 
+    @CanIgnoreReturnValue
+    Builder supportsCredentialVending(boolean supportsCredentialVending);
+
     default Builder put(StorageAccessProperty key, String value) {
       if (key.isExpirationTimestamp()) {
         expiresAt(Instant.ofEpochMilli(Long.parseLong(value)));

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

@@ -40,7 +40,7 @@ public PolarisStorageIntegration(T config, String identifierOrId) {
     this.integrationIdentifierOrId = identifierOrId;
   }
 
-  protected T config() {
+  public T config() {
     return config;
   }
 
@@ -108,6 +108,14 @@ public abstract AccessConfig getSubscopedCreds(
           @Nonnull Set<PolarisStorageActions> actions,
           @Nonnull Set<String> locations);
 
+  /**
+   * @param credentialsRequired if {@code true} indicates that the caller requires the returned
+   *     {@link AccessConfig} to have storage credentials; if {@code false} the returned {@link
+   *     AccessConfig} may or may not contain credentials.
+   */
+  public void validateCredentials(
+      @Nonnull AccessConfig accessConfig, boolean credentialsRequired) {}
+
   /**
    * Result of calling {@link #validateAccessToLocations(RealmConfig,
    * PolarisStorageConfigurationInfo, Set, Set)}

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

@@ -31,6 +31,7 @@
 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;
@@ -301,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");

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -844,7 +844,7 @@ public AccessConfig getAccessConfig(
           .atWarn()
           .addKeyValue("tableIdentifier", tableIdentifier)
           .log("Table entity has no storage configuration in its hierarchy");
-      return AccessConfig.builder().build();
+      return AccessConfig.builder().supportsCredentialVending(false).build();
     }
     return FileIOUtil.refreshAccessConfig(
         callContext,

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

@@ -804,13 +804,22 @@ private LoadTableResponse.Builder buildLoadTableResponseWithDelegationCredential
           credentialDelegation.getAccessConfig(
               tableIdentifier, tableMetadata, actions, refreshCredentialsEndpoint);
       Map<String, String> credentialConfig = accessConfig.credentials();
-      if (!credentialConfig.isEmpty() && delegationModes.contains(VENDED_CREDENTIALS)) {
-        responseBuilder.addAllConfig(credentialConfig);
-        responseBuilder.addCredential(
-            ImmutableCredential.builder()
-                .prefix(tableMetadata.location())
-                .config(credentialConfig)
-                .build());
+      if (delegationModes.contains(VENDED_CREDENTIALS)) {
+        if (!credentialConfig.isEmpty()) {
+          responseBuilder.addAllConfig(credentialConfig);
+          responseBuilder.addCredential(
+              ImmutableCredential.builder()
+                  .prefix(tableMetadata.location())
+                  .config(credentialConfig)
+                  .build());
+        } else {
+          Boolean skipCredIndirection =
+              realmConfig.getConfig(FeatureConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION);
+          Preconditions.checkArgument(
+              !accessConfig.supportsCredentialVending() || skipCredIndirection,
+              "Credential vending was requested for table %s, but no credentials are available",
+              tableIdentifier);
+        }
       }
       responseBuilder.addAllConfig(accessConfig.extraProperties());
     }

runtime/service/src/main/java/org/apache/polaris/service/storage/PolarisStorageIntegrationProviderImpl.java

@@ -115,7 +115,7 @@ public AccessConfig getSubscopedCreds(
                   @Nonnull Set<String> allowedReadLocations,
                   @Nonnull Set<String> allowedWriteLocations,
                   Optional<String> refreshCredentialsEndpoint) {
-                return AccessConfig.builder().build();
+                return AccessConfig.builder().supportsCredentialVending(false).build();
               }
 
               @Override