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.

* Update call paths leading to `AccessConfig` to indicate whether the caller requires credentials to be vended or not.

* Add checks to `AwsCredentialsStorageIntegration` (leading to 400) that S3 storage credentials are vended when requested.

* 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/AtomicOperationMetaStoreManager.java

@@ -1580,6 +1580,7 @@ private void revokeGrantRecord(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      boolean credentialsRequired,
       Optional<String> refreshCredentialsEndpoint) {
 
     // get meta store session we should be using
@@ -1616,12 +1617,13 @@ private void revokeGrantRecord(
 
     try {
       AccessConfig accessConfig =
-          storageIntegration.getSubscopedCreds(
+          storageIntegration.getAccessConfig(
               callCtx.getRealmConfig(),
               allowListOperation,
               allowedReadLocations,
               allowedWriteLocations,
-              refreshCredentialsEndpoint);
+              refreshCredentialsEndpoint,
+              credentialsRequired);
       return new ScopedCredentialsResult(accessConfig);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(

polaris-core/src/main/java/org/apache/polaris/core/persistence/TransactionWorkspaceMetaStoreManager.java

@@ -346,6 +346,7 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      boolean credentialsRequired,
       Optional<String> refreshCredentialsEndpoint) {
     return delegate.getSubscopedCredsForEntity(
         callCtx,
@@ -355,6 +356,7 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
         allowListOperation,
         allowedReadLocations,
         allowedWriteLocations,
+        credentialsRequired,
         refreshCredentialsEndpoint);
   }
 

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

@@ -2073,6 +2073,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      boolean credentialsRequired,
       Optional<String> refreshCredentialsEndpoint) {
 
     // get meta store session we should be using
@@ -2104,12 +2105,13 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
 
     try {
       AccessConfig accessConfig =
-          storageIntegration.getSubscopedCreds(
+          storageIntegration.getAccessConfig(
               callCtx.getRealmConfig(),
               allowListOperation,
               allowedReadLocations,
               allowedWriteLocations,
-              refreshCredentialsEndpoint);
+              refreshCredentialsEndpoint,
+              credentialsRequired);
       return new ScopedCredentialsResult(accessConfig);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(

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

@@ -42,6 +42,9 @@ public interface PolarisCredentialVendor {
    *     supported by the storage type it will be returned to the client in the appropriate
    *     properties. The endpoint may be relative to the base URI and the client is responsible for
    *     handling the relative path
+   * @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.
    * @return an enum map containing the scoped credentials
    */
   @Nonnull
@@ -53,5 +56,6 @@ ScopedCredentialsResult getSubscopedCredsForEntity(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      boolean credentialsRequired,
       Optional<String> refreshCredentialsEndpoint);
 }

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

@@ -60,14 +60,18 @@ public String getStorageIdentifierOrId() {
    *     supported by the storage type it will be returned to the client in the appropriate
    *     properties. The endpoint may be relative to the base URI and the client is responsible for
    *     handling the relative path
+   * @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.
    * @return An enum map including the scoped credentials
    */
-  public abstract AccessConfig getSubscopedCreds(
+  public abstract AccessConfig getAccessConfig(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint);
+      Optional<String> refreshCredentialsEndpoint,
+      boolean credentialsRequired);
 
   /**
    * Validate access for the provided operation actions and locations.

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

@@ -20,6 +20,7 @@
 
 import static org.apache.polaris.core.config.FeatureConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS;
 
+import com.google.common.base.Preconditions;
 import jakarta.annotation.Nonnull;
 import java.net.URI;
 import java.util.HashMap;
@@ -70,19 +71,25 @@ public AwsCredentialsStorageIntegration(
 
   /** {@inheritDoc} */
   @Override
-  public AccessConfig getSubscopedCreds(
+  public AccessConfig getAccessConfig(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      boolean credentialsRequired) {
     int storageCredentialDurationSeconds =
         realmConfig.getConfig(STORAGE_CREDENTIAL_DURATION_SECONDS);
     AwsStorageConfigurationInfo storageConfig = config();
     String region = storageConfig.getRegion();
     AccessConfig.Builder accessConfig = AccessConfig.builder();
 
-    if (shouldUseSts(storageConfig)) {
+    boolean shouldUseSts = shouldUseSts(storageConfig);
+    Preconditions.checkArgument(
+        !credentialsRequired || shouldUseSts,
+        "Credential vending was requested, but STS is not available");
+
+    if (shouldUseSts) {
       AssumeRoleRequest.Builder request =
           AssumeRoleRequest.builder()
               .externalId(storageConfig.getExternalId())

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureCredentialsStorageIntegration.java

@@ -73,12 +73,14 @@ public AzureCredentialsStorageIntegration(AzureStorageConfigurationInfo config)
   }
 
   @Override
-  public AccessConfig getSubscopedCreds(
+  public AccessConfig getAccessConfig(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      boolean credentialsRequired) {
+
     String loc =
         !allowedWriteLocations.isEmpty()
             ? allowedWriteLocations.stream().findAny().orElse(null)

polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCache.java

@@ -101,6 +101,9 @@ private long maxCacheDurationMs(RealmConfig realmConfig) {
    * @param allowListOperation whether allow list action on the provided read and write locations
    * @param allowedReadLocations a set of allowed to read locations
    * @param allowedWriteLocations a set of allowed to write 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.
    * @return the a map of string containing the scoped creds information
    */
   public AccessConfig getOrGenerateSubScopeCreds(
@@ -110,7 +113,8 @@ public AccessConfig getOrGenerateSubScopeCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      boolean credentialsRequired) {
     if (!isTypeSupported(polarisEntity.getType())) {
       diagnostics.fail(
           "entity_type_not_suppported_to_scope_creds", "type={}", polarisEntity.getType());
@@ -122,7 +126,8 @@ public AccessConfig getOrGenerateSubScopeCreds(
             allowListOperation,
             allowedReadLocations,
             allowedWriteLocations,
-            refreshCredentialsEndpoint);
+            refreshCredentialsEndpoint,
+            credentialsRequired);
     LOGGER.atDebug().addKeyValue("key", key).log("subscopedCredsCache");
     Function<StorageCredentialCacheKey, StorageCredentialCacheEntry> loader =
         k -> {
@@ -136,6 +141,7 @@ public AccessConfig getOrGenerateSubScopeCreds(
                   k.allowedListAction(),
                   k.allowedReadLocations(),
                   k.allowedWriteLocations(),
+                  k.credentialsRequired(),
                   k.refreshCredentialsEndpoint());
           if (scopedCredentialsResult.isSuccess()) {
             long maxCacheDurationMs = maxCacheDurationMs(callCtx.getRealmConfig());

polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCacheKey.java

@@ -51,13 +51,17 @@ public interface StorageCredentialCacheKey {
   @Value.Parameter(order = 7)
   Optional<String> refreshCredentialsEndpoint();
 
+  @Value.Parameter(order = 8)
+  boolean credentialsRequired();
+
   static StorageCredentialCacheKey of(
       String realmId,
       PolarisEntity entity,
       boolean allowedListAction,
       Set<String> allowedReadLocations,
       Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      boolean credentialsRequired) {
     String storageConfigSerializedStr =
         entity
             .getInternalPropertiesAsMap()
@@ -69,6 +73,7 @@ static StorageCredentialCacheKey of(
         allowedListAction,
         allowedReadLocations,
         allowedWriteLocations,
-        refreshCredentialsEndpoint);
+        refreshCredentialsEndpoint,
+        credentialsRequired);
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java

@@ -72,12 +72,14 @@ public GcpCredentialsStorageIntegration(
   }
 
   @Override
-  public AccessConfig getSubscopedCreds(
+  public AccessConfig getAccessConfig(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      boolean credentialsRequired) {
+
     try {
       sourceCredentials.refresh();
     } catch (IOException e) {