handle null identity, refactor tests

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 `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/test/java/org/apache/polaris/core/storage/cache/StorageCredentialCacheTest.java

@@ -161,9 +161,7 @@ public void testCacheHit() {
     Assertions.assertThat(storageCredentialCache.getEstimatedSize()).isEqualTo(1);
   }
 
-  @Test
-  public void testCacheMissForAnotherPrincipal() {
-
+  private void testCacheForAnotherPrincipal(boolean hitExpected) {
     List<ScopedCredentialsResult> mockedScopedCreds =
         getFakeScopedCreds(3, /* expireSoon= */ false);
     Mockito.when(
@@ -196,7 +194,6 @@ public void testCacheMissForAnotherPrincipal() {
         Optional.empty());
     Assertions.assertThat(storageCredentialCache.getEstimatedSize()).isEqualTo(1);
 
-    // subscope for the same entity and same allowed locations, will hit the cache
     storageCredentialCache.getOrGenerateSubScopeCreds(
         storageCredentialsVendor,
         polarisEntity,
@@ -205,11 +202,16 @@ public void testCacheMissForAnotherPrincipal() {
         Set.of("s3://bucket3/path", "s3://bucket4/path"),
         anotherPolarisPrincipal,
         Optional.empty());
-    Assertions.assertThat(storageCredentialCache.getEstimatedSize()).isEqualTo(1);
+    Assertions.assertThat(storageCredentialCache.getEstimatedSize()).isEqualTo(hitExpected ? 1 : 2);
   }
 
   @Test
   public void testCacheHitForAnotherPrincipal() {
+    testCacheForAnotherPrincipal(true);
+  }
+
+  @Test
+  public void testCacheMissForAnotherPrincipal() {
     Mockito.when(storageCredentialsVendor.getRealmConfig())
         .thenReturn(
             new RealmConfigImpl(
@@ -227,48 +229,7 @@ public String getConfiguration(@Nonnull RealmContext ctx, String configName) {
                 },
                 () -> "realm"));
 
-    List<ScopedCredentialsResult> mockedScopedCreds =
-        getFakeScopedCreds(3, /* expireSoon= */ false);
-    Mockito.when(
-            storageCredentialsVendor.getSubscopedCredsForEntity(
-                Mockito.any(),
-                Mockito.anyBoolean(),
-                Mockito.anySet(),
-                Mockito.anySet(),
-                Mockito.any(),
-                Mockito.any()))
-        .thenReturn(mockedScopedCreds.get(0))
-        .thenReturn(mockedScopedCreds.get(1))
-        .thenReturn(mockedScopedCreds.get(1));
-    PolarisBaseEntity baseEntity =
-        new PolarisBaseEntity(
-            1, 2, PolarisEntityType.CATALOG, PolarisEntitySubType.ICEBERG_TABLE, 0, "name");
-    PolarisEntity polarisEntity = new PolarisEntity(baseEntity);
-    PolarisPrincipal polarisPrincipal = PolarisPrincipal.of("principal", Map.of(), Set.of());
-    PolarisPrincipal anotherPolarisPrincipal =
-        PolarisPrincipal.of("anotherPrincipal", Map.of(), Set.of());
-
-    // add an item to the cache
-    storageCredentialCache.getOrGenerateSubScopeCreds(
-        storageCredentialsVendor,
-        polarisEntity,
-        true,
-        Set.of("s3://bucket1/path", "s3://bucket2/path"),
-        Set.of("s3://bucket3/path", "s3://bucket4/path"),
-        polarisPrincipal,
-        Optional.empty());
-    Assertions.assertThat(storageCredentialCache.getEstimatedSize()).isEqualTo(1);
-
-    // subscope for the same entity and same allowed locations, will hit the cache
-    storageCredentialCache.getOrGenerateSubScopeCreds(
-        storageCredentialsVendor,
-        polarisEntity,
-        true,
-        Set.of("s3://bucket1/path", "s3://bucket2/path"),
-        Set.of("s3://bucket3/path", "s3://bucket4/path"),
-        anotherPolarisPrincipal,
-        Optional.empty());
-    Assertions.assertThat(storageCredentialCache.getEstimatedSize()).isEqualTo(2);
+    testCacheForAnotherPrincipal(false);
   }
 
   @RepeatedTest(10)

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

@@ -140,8 +140,10 @@ public void testGetSubscopedCreds(String scheme) {
             "/namespace/table/credentials");
   }
 
+  // uses different realm config with INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL set to true
+  // tests that the resulting role session name includes principal name
   @Test
-  public void testGetSubscopedCredsWithNameInclude() {
+  public void testGetSubscopedCredsRoleSessionNameWithPrincipalIncluded() {
     StsClient stsClient = Mockito.mock(StsClient.class);
     String roleARN = "arn:aws:iam::012345678901:role/jdoe";
     String externalId = "externalId";
@@ -173,18 +175,6 @@ public void testGetSubscopedCredsWithNameInclude() {
                 Set.of(warehouseDir + "/namespace/table"),
                 POLARIS_PRINCIPAL,
                 Optional.of("/namespace/table/credentials"));
-    assertThat(storageAccessConfig.credentials())
-        .isNotEmpty()
-        .containsEntry(StorageAccessProperty.AWS_TOKEN.getPropertyName(), "sess")
-        .containsEntry(StorageAccessProperty.AWS_KEY_ID.getPropertyName(), "accessKey")
-        .containsEntry(StorageAccessProperty.AWS_SECRET_KEY.getPropertyName(), "secretKey")
-        .containsEntry(
-            StorageAccessProperty.AWS_SESSION_TOKEN_EXPIRES_AT_MS.getPropertyName(),
-            String.valueOf(EXPIRE_TIME.toEpochMilli()));
-    assertThat(storageAccessConfig.extraProperties())
-        .containsEntry(
-            StorageAccessProperty.AWS_REFRESH_CREDENTIALS_ENDPOINT.getPropertyName(),
-            "/namespace/table/credentials");
   }
 
   @ParameterizedTest

runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisPrincipalHolder.java

@@ -23,6 +23,7 @@
 import io.quarkus.security.identity.SecurityIdentity;
 import jakarta.enterprise.context.RequestScoped;
 import jakarta.enterprise.inject.Produces;
+import jakarta.enterprise.inject.UnsatisfiedResolutionException;
 import java.security.Principal;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.polaris.core.PolarisDiagnostics;
@@ -46,6 +47,10 @@ public PolarisPrincipal get(
     SecurityIdentity identity =
         currentIdentityAssociation.getDeferredIdentity().subscribeAsCompletionStage().getNow(null);
 
+    if (identity == null) {
+      throw new UnsatisfiedResolutionException("Not authenticated");
+    }
+
     Principal userPrincipal = identity.getPrincipal();
 
     diagnostics.check(