Enforce that S3 credentials are vended when requested

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)));

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