Refactor PolarisPrincipal to contain generic attributes (#5085)

Author: adutra

CHANGELOG.md

@@ -48,6 +48,19 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
   are now always bootstrapped with the latest available schema version.
 - The `MaintenanceService.performMaintenance()` signature now requires an explicit `OptionalLong overrideRunId` argument to supersede the latest unfinished maintenance run.
 - Admin grant APIs now reject table-like privilege targets with an empty namespace. A table-like target without a namespace is considered invalid input.
+- `PolarisPrincipal` now carries a generic `Map<String, Object> attributes` bag instead of the
+  previous `Map<String, String> properties` and `Optional<String> token` fields. Three well-known
+  attribute keys are defined as constants on the interface:
+
+  - `PolarisPrincipal.PRINCIPAL_ENTITY_ATTRIBUTE_KEY`: holds the `PrincipalEntity`;
+  - `PolarisPrincipal.PRINCIPAL_ROLE_ALL_ATTRIBUTE_KEY`: holds a boolean indicating if the 
+    `PRINCIPAL_ROLE:ALL` pseudo-role is present;
+  - `PolarisPrincipal.JWT_ATTRIBUTE_KEY`: holds the raw JWT string.
+
+  Use `PolarisPrincipal.getAttribute(key, type)` for type-safe access. The `Authenticator` interface
+  now accepts a Quarkus `SecurityIdentity` instead of a `PolarisCredential`, and throws
+  `AuthenticationFailedException` (mapped to HTTP 401) instead of Iceberg's
+  `NotAuthorizedException`.
 
 ### New Features
 - Added GCS principal attribution for vended credentials (the GCP counterpart of AWS STS session tags). Set `GCS_PRINCIPAL_ATTRIBUTION_ENABLED=true` to activate; the feature flags `GCS_PRINCIPAL_ATTRIBUTION_WIF_AUDIENCE`, `GCS_PRINCIPAL_ATTRIBUTION_TOKEN_ISSUER`, and `GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_FILE` are then required (a missing value is a fatal configuration error). Also requires a `gcpServiceAccount` on the catalog StorageConfiguration. When enabled, credential vending chains a catalog-signed JWT through a Workload Identity Federation token exchange and service-account impersonation, so the Polaris principal appears in GCS Data Access audit logs (`serviceAccountDelegationInfo.principalSubject`) for any client. `GCS_PRINCIPAL_ATTRIBUTION_SIGNING_KEY_ID` sets the JWT `kid` for JWKS key rotation. Attribution is keyed per-principal in the credential cache; when disabled (default), GCP vending behaviour is unchanged.

extensions/auth/ranger/src/main/java/org/apache/polaris/extension/auth/ranger/utils/RangerUtils.java

@@ -30,6 +30,7 @@
 import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
 import org.apache.ranger.authz.model.RangerAccessInfo;
@@ -127,10 +128,12 @@ private static Map<String, Object> getResourceAttributes(
   }
 
   private static Map<String, Object> getUserAttributes(PolarisPrincipal principal) {
-    Map<String, String> properties = principal.getProperties();
+    Map<String, String> properties =
+        principal
+            .getAttribute(PolarisPrincipal.PRINCIPAL_ENTITY_ATTRIBUTE_KEY, PrincipalEntity.class)
+            .map(PrincipalEntity::getInternalPropertiesAsMap)
+            .orElse(Collections.emptyMap());
 
-    return (properties == null || properties.isEmpty())
-        ? Collections.emptyMap()
-        : new HashMap<>(properties);
+    return properties.isEmpty() ? Collections.emptyMap() : new HashMap<>(properties);
   }
 }

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisManagementServiceIntegrationTest.java

@@ -1028,7 +1028,7 @@ public void testCreatePrincipalAndRotateCredentials() {
           .isNotNull()
           .extracting(ErrorResponse::message)
           .asString()
-          .contains("PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE");
+          .contains("must rotate credentials first");
     }
 
     // Now try to rotate using the principal's token.

polaris-core/src/main/java/org/apache/polaris/core/auth/AuthorizationPreConditions.java

@@ -22,6 +22,7 @@
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
+import org.apache.polaris.core.entity.PrincipalEntity;
 
 /**
  * Common pre-condition checks shared across authorizer implementations for credential-related
@@ -48,12 +49,21 @@ public static void checkCredentialRotationRequired(
     if (realmConfig.getConfig(
             FeatureConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING)
         && authzOp != PolarisAuthorizableOperation.ROTATE_CREDENTIALS
-        && polarisPrincipal
-            .getProperties()
-            .containsKey(PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)) {
+        && mustRotateCredentials(polarisPrincipal)) {
       throw new ForbiddenException(
-          "Principal '%s' is not authorized for op %s due to PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE",
+          "Principal '%s' is not authorized for op %s because it must rotate credentials first",
           polarisPrincipal.getName(), authzOp);
     }
   }
+
+  private static boolean mustRotateCredentials(PolarisPrincipal principal) {
+    return principal
+        .getAttribute(PolarisPrincipal.PRINCIPAL_ENTITY_ATTRIBUTE_KEY, PrincipalEntity.class)
+        .map(PrincipalEntity::getInternalPropertiesAsMap)
+        .map(
+            map ->
+                map.containsKey(
+                    PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE))
+        .orElse(false);
+  }
 }