feat: pass principal name as part of aws subscoped credentials session

CHANGELOG.md

@@ -58,6 +58,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Support credential vending for federated catalogs. `ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING` (default: true) was added to toggle this feature.
 - Enhanced catalog federation with SigV4 authentication support, additional authentication types for credential vending, and location-based access restrictions to block credential vending for remote tables outside allowed location lists.
 - Added `topologySpreadConstraints` support in Helm chart.
+- Added support for including principal name in subscoped credentials. `INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL` (default: false) can be used to toggle this feature. If enabled, cached credentials issued to one principal will no longer be available for others.
 
 ### Changes
 

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -79,6 +79,18 @@ public static void enforceFeatureEnabledOrThrow(
           .defaultValue(false)
           .buildFeatureConfiguration();
 
+  public static final FeatureConfiguration<Boolean> INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL =
+      PolarisConfiguration.<Boolean>builder()
+          .key("INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL")
+          .description(
+              "If set to true, principal name will be included in temporary subscoped credentials.\n"
+                  + "Currently only AWS credentials are supported for which session name of the generated credentials \n"
+                  + "will look like 'polaris-<principal>' rather than simple 'polaris'.\n"
+                  + "Note that enabling this feature leads to degradation in temporary credential caching as \n"
+                  + "catalog will no longer be able to reuse credentials for multiple principals.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
+
   public static final FeatureConfiguration<Boolean> ALLOW_SETTING_S3_ENDPOINTS =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_SETTING_S3_ENDPOINTS")

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

@@ -35,6 +35,7 @@
 import java.util.stream.IntStream;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.entity.AsyncTaskType;
 import org.apache.polaris.core.entity.CatalogEntity;
@@ -1600,6 +1601,7 @@ public void deletePrincipalSecrets(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
 
     // get meta store session we should be using
@@ -1641,6 +1643,7 @@ public void deletePrincipalSecrets(
               allowListOperation,
               allowedReadLocations,
               allowedWriteLocations,
+              polarisPrincipal,
               refreshCredentialsEndpoint);
       return new ScopedCredentialsResult(storageAccessConfig);
     } catch (Exception ex) {

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

@@ -28,6 +28,7 @@
 import java.util.Set;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.entity.LocationBasedEntity;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
@@ -324,6 +325,7 @@ public void deletePrincipalSecrets(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
     return delegate.getSubscopedCredsForEntity(
         callCtx,
@@ -333,6 +335,7 @@ public void deletePrincipalSecrets(
         allowListOperation,
         allowedReadLocations,
         allowedWriteLocations,
+        polarisPrincipal,
         refreshCredentialsEndpoint);
   }
 

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

@@ -34,6 +34,7 @@
 import java.util.stream.IntStream;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.entity.AsyncTaskType;
 import org.apache.polaris.core.entity.CatalogEntity;
@@ -2094,6 +2095,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
 
     // get meta store session we should be using
@@ -2130,6 +2132,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
               allowListOperation,
               allowedReadLocations,
               allowedWriteLocations,
+              polarisPrincipal,
               refreshCredentialsEndpoint);
       return new ScopedCredentialsResult(storageAccessConfig);
     } catch (Exception ex) {

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

@@ -22,6 +22,7 @@
 import java.util.Optional;
 import java.util.Set;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.persistence.dao.entity.ScopedCredentialsResult;
 
@@ -53,5 +54,6 @@ ScopedCredentialsResult getSubscopedCredsForEntity(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint);
 }

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

@@ -22,6 +22,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.RealmConfig;
 
 /**
@@ -67,6 +68,7 @@ public abstract StorageAccessConfig getSubscopedCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint);
 
   /**

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

@@ -22,6 +22,7 @@
 import jakarta.annotation.Nonnull;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
@@ -67,6 +68,7 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
     return polarisCredentialVendor.getSubscopedCredsForEntity(
         callContext.getPolarisCallContext(),
@@ -76,6 +78,7 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
         allowListOperation,
         allowedReadLocations,
         allowedWriteLocations,
+        polarisPrincipal,
         refreshCredentialsEndpoint);
   }
 }

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

@@ -28,6 +28,8 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessConfig;
@@ -81,6 +83,7 @@ public StorageAccessConfig getSubscopedCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
     int storageCredentialDurationSeconds =
         realmConfig.getConfig(STORAGE_CREDENTIAL_DURATION_SECONDS);
@@ -89,12 +92,22 @@ public StorageAccessConfig getSubscopedCreds(
     String accountId = storageConfig.getAwsAccountId();
     StorageAccessConfig.Builder accessConfig = StorageAccessConfig.builder();
 
+    boolean includePrincipalNameInSubscopedCredential =
+        realmConfig.getConfig(FeatureConfiguration.INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL);
+
+    String roleSessionName =
+        includePrincipalNameInSubscopedCredential
+            ? "polaris-" + polarisPrincipal.getName()
+            : "PolarisAwsCredentialsStorageIntegration";
+    String cappedRoleSessionName =
+        roleSessionName.substring(0, Math.min(roleSessionName.length(), 64));
+
     if (shouldUseSts(storageConfig)) {
       AssumeRoleRequest.Builder request =
           AssumeRoleRequest.builder()
               .externalId(storageConfig.getExternalId())
               .roleArn(storageConfig.getRoleARN())
-              .roleSessionName("PolarisAwsCredentialsStorageIntegration")
+              .roleSessionName(cappedRoleSessionName)
               .policy(
                   policyString(
                           storageConfig,

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

@@ -53,6 +53,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessConfig;
@@ -84,6 +85,7 @@ public StorageAccessConfig getSubscopedCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
     String loc =
         !allowedWriteLocations.isEmpty()

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

@@ -31,6 +31,7 @@
 import java.util.function.Function;
 import org.apache.iceberg.exceptions.UnprocessableEntityException;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
@@ -109,21 +110,29 @@ public StorageAccessConfig getOrGenerateSubScopeCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
     RealmContext realmContext = storageCredentialsVendor.getRealmContext();
     RealmConfig realmConfig = storageCredentialsVendor.getRealmConfig();
     if (!isTypeSupported(polarisEntity.getType())) {
       diagnostics.fail(
           "entity_type_not_suppported_to_scope_creds", "type={}", polarisEntity.getType());
     }
+
+    boolean includePrincipalNameInSubscopedCredential =
+        realmConfig.getConfig(FeatureConfiguration.INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL);
+
     StorageCredentialCacheKey key =
         StorageCredentialCacheKey.of(
             realmContext.getRealmIdentifier(),
             polarisEntity,
             allowListOperation,
             allowedReadLocations,
             allowedWriteLocations,
-            refreshCredentialsEndpoint);
+            refreshCredentialsEndpoint,
+            includePrincipalNameInSubscopedCredential
+                ? Optional.of(polarisPrincipal)
+                : Optional.empty());
     LOGGER.atDebug().addKeyValue("key", key).log("subscopedCredsCache");
     Function<StorageCredentialCacheKey, StorageCredentialCacheEntry> loader =
         k -> {
@@ -134,6 +143,7 @@ public StorageAccessConfig getOrGenerateSubScopeCreds(
                   allowListOperation,
                   allowedReadLocations,
                   allowedWriteLocations,
+                  polarisPrincipal,
                   refreshCredentialsEndpoint);
           if (scopedCredentialsResult.isSuccess()) {
             long maxCacheDurationMs = maxCacheDurationMs(realmConfig);

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

@@ -21,6 +21,7 @@
 import jakarta.annotation.Nullable;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.immutables.PolarisImmutable;
@@ -51,13 +52,17 @@ public interface StorageCredentialCacheKey {
   @Value.Parameter(order = 7)
   Optional<String> refreshCredentialsEndpoint();
 
+  @Value.Parameter(order = 8)
+  Optional<String> principalName();
+
   static StorageCredentialCacheKey of(
       String realmId,
       PolarisEntity entity,
       boolean allowedListAction,
       Set<String> allowedReadLocations,
       Set<String> allowedWriteLocations,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      Optional<PolarisPrincipal> polarisPrincipal) {
     String storageConfigSerializedStr =
         entity
             .getInternalPropertiesAsMap()
@@ -69,6 +74,7 @@ static StorageCredentialCacheKey of(
         allowedListAction,
         allowedReadLocations,
         allowedWriteLocations,
-        refreshCredentialsEndpoint);
+        refreshCredentialsEndpoint,
+        polarisPrincipal.map(PolarisPrincipal::getName));
   }
 }

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

@@ -38,6 +38,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
@@ -77,6 +78,7 @@ public StorageAccessConfig getSubscopedCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Optional<String> refreshCredentialsEndpoint) {
     try {
       sourceCredentials.refresh();

polaris-core/src/test/java/org/apache/polaris/core/storage/InMemoryStorageIntegrationTest.java

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.config.RealmConfigImpl;
@@ -199,6 +200,7 @@ public StorageAccessConfig getSubscopedCreds(
         boolean allowListOperation,
         @Nonnull Set<String> allowedReadLocations,
         @Nonnull Set<String> allowedWriteLocations,
+        @Nonnull PolarisPrincipal polarisPrincipal,
         Optional<String> refreshCredentialsEndpoint) {
       return null;
     }