Feature configuration to modify storage credential lifetime

polaris-core/src/main/java/org/apache/polaris/core/PolarisConfiguration.java

@@ -21,9 +21,14 @@
 import java.util.List;
 import java.util.Optional;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.context.CallContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class PolarisConfiguration<T> {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(PolarisConfiguration.class);
+
   public final String key;
   public final String description;
   public final T defaultValue;
@@ -89,6 +94,25 @@ public PolarisConfiguration<T> build() {
     }
   }
 
+  /**
+   * Returns the value of a `PolarisConfiguration`, or the default if it cannot be loaded. This
+   * method does not need to be used when a `CallContext` is already available
+   */
+  public static <T> T loadConfig(PolarisConfiguration<T> configuration) {
+    var callContext = CallContext.getCurrentContext();
+    if (callContext == null) {
+      LOGGER.warn(
+          String.format(
+              "Unable to load current call context; using %s = %s",
+              configuration.key, configuration.defaultValue));
+      return configuration.defaultValue;
+    }
+    return callContext
+        .getPolarisCallContext()
+        .getConfigurationStore()
+        .getConfiguration(callContext.getPolarisCallContext(), configuration);
+  }
+
   public static <T> Builder<T> builder() {
     return new Builder<>();
   }
@@ -199,4 +223,22 @@ public static <T> Builder<T> builder() {
               "If set to true, allows tables to be dropped with the purge parameter set to true.")
           .defaultValue(true)
           .build();
+
+  public static final PolarisConfiguration<Integer> STORAGE_CREDENTIAL_DURATION_SECONDS =
+      PolarisConfiguration.<Integer>builder()
+          .key("STORAGE_CREDENTIAL_DURATION_SECONDS")
+          .description(
+              "The duration of time that vended storage credentials are valid for. Support for"
+                  + " longer (or shorter) durations is dependent on the storage provider.")
+          .defaultValue(60 * 60) // 1 hour
+          .build();
+
+  public static final PolarisConfiguration<Integer> STORAGE_CREDENTIAL_CACHE_DURATION_SECONDS =
+      PolarisConfiguration.<Integer>builder()
+          .key("STORAGE_CREDENTIAL_CACHE_DURATION_SECONDS")
+          .description(
+              "How long to store storage credentials in the local cache. This should be less than "
+                  + STORAGE_CREDENTIAL_DURATION_SECONDS.key)
+          .defaultValue(30 * 60) // 30 minutes
+          .build();
 }

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

@@ -22,8 +22,10 @@
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
+import org.apache.polaris.core.PolarisConfiguration;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.PolarisCredentialProperty;
@@ -69,13 +71,21 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
                             allowedReadLocations,
                             allowedWriteLocations)
                         .toJson())
+                .durationSeconds(
+                    PolarisConfiguration.loadConfig(
+                        PolarisConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS))
                 .build());
     EnumMap<PolarisCredentialProperty, String> credentialMap =
         new EnumMap<>(PolarisCredentialProperty.class);
     credentialMap.put(PolarisCredentialProperty.AWS_KEY_ID, response.credentials().accessKeyId());
     credentialMap.put(
         PolarisCredentialProperty.AWS_SECRET_KEY, response.credentials().secretAccessKey());
     credentialMap.put(PolarisCredentialProperty.AWS_TOKEN, response.credentials().sessionToken());
+    Optional.ofNullable(response.credentials().expiration())
+        .ifPresent(
+            i ->
+                credentialMap.put(
+                    PolarisCredentialProperty.EXPIRATION_TIME, String.valueOf(i.toEpochMilli())));
     return credentialMap;
   }
 

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

@@ -44,6 +44,7 @@
 import java.util.HashSet;
 import java.util.Objects;
 import java.util.Set;
+import org.apache.polaris.core.PolarisConfiguration;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.PolarisCredentialProperty;
@@ -124,8 +125,15 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
     // Azure strictly requires the end time to be <= 7 days from the current time, -1 min to avoid
     // clock skew between the client and server,
     OffsetDateTime startTime = start.truncatedTo(ChronoUnit.SECONDS).atOffset(ZoneOffset.UTC);
-    OffsetDateTime sanitizedEndTime =
+    int intendedDurationSeconds =
+        PolarisConfiguration.loadConfig(PolarisConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS);
+    OffsetDateTime intendedEndTime =
+        start.plusSeconds(intendedDurationSeconds).atOffset(ZoneOffset.UTC);
+    OffsetDateTime maxAllowedEndTime =
         start.plus(Period.ofDays(7)).minusSeconds(60).atOffset(ZoneOffset.UTC);
+    OffsetDateTime sanitizedEndTime =
+        intendedEndTime.isBefore(maxAllowedEndTime) ? intendedEndTime : maxAllowedEndTime;
+
     LOGGER
         .atDebug()
         .addKeyValue("allowedListAction", allowListOperation)
@@ -164,6 +172,9 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
     }
     credentialMap.put(PolarisCredentialProperty.AZURE_SAS_TOKEN, sasToken);
     credentialMap.put(PolarisCredentialProperty.AZURE_ACCOUNT_HOST, storageDnsName);
+    credentialMap.put(
+        PolarisCredentialProperty.EXPIRATION_TIME,
+        String.valueOf(sanitizedEndTime.toInstant().toEpochMilli()));
     return credentialMap;
   }
 

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

@@ -28,6 +28,7 @@
 import java.util.function.Function;
 import org.apache.iceberg.exceptions.UnprocessableEntityException;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.PolarisConfiguration;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
@@ -41,11 +42,10 @@ public class StorageCredentialCache {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(StorageCredentialCache.class);
 
-  private static final long CACHE_MAX_DURATION_MS = 30 * 60 * 1000L; // 30 minutes
   private static final long CACHE_MAX_NUMBER_OF_ENTRIES = 10_000L;
   private final LoadingCache<StorageCredentialCacheKey, StorageCredentialCacheEntry> cache;
 
-  /** Initialize the creds cache, max cache duration is half an hr. */
+  /** Initialize the creds cache */
   public StorageCredentialCache() {
     cache =
         Caffeine.newBuilder()
@@ -62,7 +62,7 @@ public long expireAfterCreate(
                             0,
                             Math.min(
                                 (entry.getExpirationTime() - System.currentTimeMillis()) / 2,
-                                CACHE_MAX_DURATION_MS));
+                                maxCacheDurationMs()));
                     return TimeUnit.MILLISECONDS.toNanos(expireAfterMillis);
                   }
 
@@ -91,6 +91,24 @@ public long expireAfterRead(
                 });
   }
 
+  /** How long credentials should remain in the cache. */
+  private static long maxCacheDurationMs() {
+    var cacheDurationSeconds =
+        PolarisConfiguration.loadConfig(
+            PolarisConfiguration.STORAGE_CREDENTIAL_CACHE_DURATION_SECONDS);
+    var credentialDurationSeconds =
+        PolarisConfiguration.loadConfig(PolarisConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS);
+    if (cacheDurationSeconds >= credentialDurationSeconds) {
+      throw new IllegalArgumentException(
+          String.format(
+              "%s should be less than %s",
+              PolarisConfiguration.STORAGE_CREDENTIAL_CACHE_DURATION_SECONDS.key,
+              PolarisConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS.key));
+    } else {
+      return cacheDurationSeconds * 1000L;
+    }
+  }
+
   /**
    * Either get from the cache or generate a new entry for a scoped creds
    *