add changelog, address comments

Author: tokoko

CHANGELOG.md

@@ -57,6 +57,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Added `--no-sts` flag to CLI to support S3-compatible storage systems that do not have Security Token Service available.
 - 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 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/storage/aws/AwsCredentialsStorageIntegration.java

@@ -98,7 +98,7 @@ public StorageAccessConfig getSubscopedCreds(
     String roleSessionName =
         includePrincipalNameInSubscopedCredential
             ? "polaris-" + polarisPrincipal.getName()
-            : "polaris";
+            : "PolarisAwsCredentialsStorageIntegration";
     String cappedRoleSessionName =
         roleSessionName.substring(0, Math.min(roleSessionName.length(), 64));
 

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

@@ -129,10 +129,10 @@ public StorageAccessConfig getOrGenerateSubScopeCreds(
             allowListOperation,
             allowedReadLocations,
             allowedWriteLocations,
+            refreshCredentialsEndpoint,
             includePrincipalNameInSubscopedCredential
                 ? Optional.of(polarisPrincipal)
-                : Optional.empty(),
-            refreshCredentialsEndpoint);
+                : Optional.empty());
     LOGGER.atDebug().addKeyValue("key", key).log("subscopedCredsCache");
     Function<StorageCredentialCacheKey, StorageCredentialCacheEntry> loader =
         k -> {

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

@@ -50,19 +50,19 @@ public interface StorageCredentialCacheKey {
   Set<String> allowedWriteLocations();
 
   @Value.Parameter(order = 7)
-  Optional<String> principalName();
+  Optional<String> refreshCredentialsEndpoint();
 
   @Value.Parameter(order = 8)
-  Optional<String> refreshCredentialsEndpoint();
+  Optional<String> principalName();
 
   static StorageCredentialCacheKey of(
       String realmId,
       PolarisEntity entity,
       boolean allowedListAction,
       Set<String> allowedReadLocations,
       Set<String> allowedWriteLocations,
-      Optional<PolarisPrincipal> polarisPrincipal,
-      Optional<String> refreshCredentialsEndpoint) {
+      Optional<String> refreshCredentialsEndpoint,
+      Optional<PolarisPrincipal> polarisPrincipal) {
     String storageConfigSerializedStr =
         entity
             .getInternalPropertiesAsMap()
@@ -74,7 +74,7 @@ static StorageCredentialCacheKey of(
         allowedListAction,
         allowedReadLocations,
         allowedWriteLocations,
-        polarisPrincipal.map(PolarisPrincipal::getName),
-        refreshCredentialsEndpoint);
+        refreshCredentialsEndpoint,
+        polarisPrincipal.map(PolarisPrincipal::getName));
   }
 }

polaris-core/src/test/java/org/apache/polaris/core/storage/cache/StorageCredentialCacheTest.java

@@ -300,8 +300,8 @@ public void testCacheEvict() throws Exception {
             true,
             Set.of("s3://bucket1/path", "s3://bucket2/path"),
             Set.of("s3://bucket/path"),
-            Optional.of(polarisPrincipal),
-            Optional.empty());
+            Optional.empty(),
+            Optional.of(polarisPrincipal));
 
     // the entry will be evicted immediately because the token is expired
     storageCredentialCache.getOrGenerateSubScopeCreds(

polaris-core/src/test/java/org/apache/polaris/service/storage/aws/AwsCredentialsStorageIntegrationTest.java

@@ -86,7 +86,7 @@ public String getConfiguration(@Nonnull RealmContext ctx, String configName) {
           .build();
   public static final String AWS_PARTITION = "aws";
   public static final PolarisPrincipal POLARIS_PRINCIPAL =
-      PolarisPrincipal.of("principal", Map.of(), Set.of());
+      PolarisPrincipal.of("test-principal", Map.of(), Set.of());
 
   @ParameterizedTest
   @ValueSource(strings = {"s3a", "s3"})
@@ -103,7 +103,8 @@ public void testGetSubscopedCreds(String scheme) {
                   .asInstanceOf(InstanceOfAssertFactories.type(AssumeRoleRequest.class))
                   .returns(externalId, AssumeRoleRequest::externalId)
                   .returns(roleARN, AssumeRoleRequest::roleArn)
-                  .returns("polaris", AssumeRoleRequest::roleSessionName)
+                  .returns(
+                      "PolarisAwsCredentialsStorageIntegration", AssumeRoleRequest::roleSessionName)
                   // ensure that the policy content does not refer to S3A
                   .extracting(AssumeRoleRequest::policy)
                   .doesNotMatch(s -> s.contains("s3a"));
@@ -153,7 +154,7 @@ public void testGetSubscopedCredsWithNameInclude() {
                   .asInstanceOf(InstanceOfAssertFactories.type(AssumeRoleRequest.class))
                   .returns(externalId, AssumeRoleRequest::externalId)
                   .returns(roleARN, AssumeRoleRequest::roleArn)
-                  .returns("polaris-principal", AssumeRoleRequest::roleSessionName);
+                  .returns("polaris-test-principal", AssumeRoleRequest::roleSessionName);
               return ASSUME_ROLE_RESPONSE;
             });
     String warehouseDir = "s3://bucket/path/to/warehouse";

runtime/service/src/main/java/org/apache/polaris/service/task/TaskExecutorImpl.java

@@ -40,6 +40,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.function.TriConsumer;
+import org.apache.polaris.core.auth.ImmutablePolarisPrincipal;
 import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
@@ -72,8 +73,8 @@ public class TaskExecutorImpl implements TaskExecutor {
   private final MetaStoreManagerFactory metaStoreManagerFactory;
   private final TaskFileIOSupplier fileIOSupplier;
   private final RealmContextHolder realmContextHolder;
-  @Inject PolarisPrincipalHolder polarisPrincipalHolder;
-  @Inject PolarisPrincipal polarisPrincipal;
+  private final PolarisPrincipalHolder polarisPrincipalHolder;
+  private final PolarisPrincipal polarisPrincipal;
   private final List<TaskHandler> taskHandlers = new CopyOnWriteArrayList<>();
   private final Optional<TriConsumer<Long, Boolean, Throwable>> errorHandler;
   private final PolarisEventListener polarisEventListener;
@@ -82,7 +83,7 @@ public class TaskExecutorImpl implements TaskExecutor {
 
   @SuppressWarnings("unused") // Required by CDI
   protected TaskExecutorImpl() {
-    this(null, null, null, null, null, null, null, null, null);
+    this(null, null, null, null, null, null, null, null, null, null, null);
   }
 
   @Inject
@@ -96,7 +97,9 @@ public TaskExecutorImpl(
       RealmContextHolder realmContextHolder,
       PolarisEventListener polarisEventListener,
       PolarisEventMetadataFactory eventMetadataFactory,
-      @Nullable Tracer tracer) {
+      @Nullable Tracer tracer,
+      PolarisPrincipalHolder polarisPrincipalHolder,
+      PolarisPrincipal polarisPrincipal) {
     this.executor = executor;
     this.clock = clock;
     this.metaStoreManagerFactory = metaStoreManagerFactory;
@@ -105,6 +108,8 @@ public TaskExecutorImpl(
     this.polarisEventListener = polarisEventListener;
     this.eventMetadataFactory = eventMetadataFactory;
     this.tracer = tracer;
+    this.polarisPrincipalHolder = polarisPrincipalHolder;
+    this.polarisPrincipal = polarisPrincipal;
 
     if (errorHandler != null && errorHandler.isResolvable()) {
       this.errorHandler = Optional.of(errorHandler.get());
@@ -167,10 +172,7 @@ public void addTaskHandlerContext(long taskEntityId, CallContext callContext) {
     String realmId = callContext.getRealmContext().getRealmIdentifier();
 
     PolarisPrincipal principalClone =
-        PolarisPrincipal.of(
-            polarisPrincipal.getName(),
-            polarisPrincipal.getProperties(),
-            polarisPrincipal.getRoles());
+        ImmutablePolarisPrincipal.builder().from(polarisPrincipal).build();
 
     return CompletableFuture.runAsync(
             () -> {
@@ -253,6 +255,9 @@ protected void handleTaskWithTracing(
     // Note: each call to this method runs in a new CDI request context
 
     realmContextHolder.set(() -> realmId);
+    // since this is now a different context we store clone of the principal in a holder object
+    // which essentially reauthenticates the principal. PolarisPrincipal bean always looks for a
+    // principal set in PolarisPrincipalHolder first and assumes that identity if set.
     polarisPrincipalHolder.set(principal);
 
     if (tracer == null) {

runtime/service/src/test/java/org/apache/polaris/service/catalog/Profiles.java

@@ -19,7 +19,6 @@
 
 package org.apache.polaris.service.catalog;
 
-import com.google.common.collect.ImmutableMap;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import java.util.Map;
 
@@ -42,14 +41,4 @@ public Map<String, String> getConfigOverrides() {
           "true");
     }
   }
-
-  public static class AuthProfile extends DefaultProfile {
-    @Override
-    public Map<String, String> getConfigOverrides() {
-      return ImmutableMap.<String, String>builder()
-          .putAll(super.getConfigOverrides())
-          .put("polaris.test.rootAugmentor.enabled", "true")
-          .build();
-    }
-  }
 }

runtime/service/src/test/java/org/apache/polaris/service/catalog/generic/PolarisGenericTableCatalogRelationalTest.java

@@ -23,6 +23,6 @@
 import org.apache.polaris.service.catalog.Profiles;
 
 @QuarkusTest
-@TestProfile(Profiles.AuthProfile.class)
+@TestProfile(Profiles.DefaultProfile.class)
 public class PolarisGenericTableCatalogRelationalTest
     extends AbstractPolarisGenericTableCatalogTest {}

runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/PolicyCatalogRelationalTest.java

@@ -23,5 +23,5 @@
 import org.apache.polaris.service.catalog.Profiles;
 
 @QuarkusTest
-@TestProfile(Profiles.AuthProfile.class)
+@TestProfile(Profiles.DefaultProfile.class)
 public class PolicyCatalogRelationalTest extends AbstractPolicyCatalogTest {}