Refactor Authenticator and PolarisPrincipal (#2307)

The main goal of this change is to facilitate future integration of federated principals:

- `AuthenticatedPolarisPrincipal` becomes an interface `PolarisPrincipal`, as the original class leaks implementation details (references to `PrincipalEntity` and thus to the storage layer). The new interface does not reference the storage layer. This is one step further towards easy pluggability of authentication in Polaris.

- The `Authenticator.authenticate()` method does not return an `Optional` anymore, as this was ambiguous (returning `Optional.empty()` vs throwing `NotAuthorizedException`).

- Also the `Authenticator` interface is not generic anymore. This was an artifact of times when there were two kinds of `Authenticators` in Polaris (one for internal auth, the other for external) and is not necessary anymore.

Author: adutra

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

@@ -29,14 +29,14 @@
 public interface PolarisAuthorizer {
 
   void authorizeOrThrow(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable PolarisResolvedPathWrapper target,
       @Nullable PolarisResolvedPathWrapper secondary);
 
   void authorizeOrThrow(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable List<PolarisResolvedPathWrapper> targets,

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

@@ -558,13 +558,13 @@ public boolean matchesOrIsSubsumedBy(
 
   @Override
   public void authorizeOrThrow(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable PolarisResolvedPathWrapper target,
       @Nullable PolarisResolvedPathWrapper secondary) {
     authorizeOrThrow(
-        authenticatedPrincipal,
+        polarisPrincipal,
         activatedEntities,
         authzOp,
         target == null ? null : List.of(target),
@@ -573,7 +573,7 @@ public void authorizeOrThrow(
 
   @Override
   public void authorizeOrThrow(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable List<PolarisResolvedPathWrapper> targets,
@@ -582,20 +582,18 @@ public void authorizeOrThrow(
         realmConfig.getConfig(
             FeatureConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING);
     if (enforceCredentialRotationRequiredState
-        && authenticatedPrincipal
-            .getPrincipalEntity()
-            .getInternalPropertiesAsMap()
+        && polarisPrincipal
+            .getProperties()
             .containsKey(PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)
         && authzOp != PolarisAuthorizableOperation.ROTATE_CREDENTIALS) {
       throw new ForbiddenException(
           "Principal '%s' is not authorized for op %s due to PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE",
-          authenticatedPrincipal.getName(), authzOp);
-    } else if (!isAuthorized(
-        authenticatedPrincipal, activatedEntities, authzOp, targets, secondaries)) {
+          polarisPrincipal.getName(), authzOp);
+    } else if (!isAuthorized(polarisPrincipal, activatedEntities, authzOp, targets, secondaries)) {
       throw new ForbiddenException(
           "Principal '%s' with activated PrincipalRoles '%s' and activated grants via '%s' is not authorized for op %s",
-          authenticatedPrincipal.getName(),
-          authenticatedPrincipal.getActivatedPrincipalRoleNames(),
+          polarisPrincipal.getName(),
+          polarisPrincipal.getRoles(),
           activatedEntities.stream().map(PolarisEntityCore::getName).collect(Collectors.toSet()),
           authzOp);
     }
@@ -607,21 +605,21 @@ public void authorizeOrThrow(
    * the operation.
    */
   public boolean isAuthorized(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPolarisPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable PolarisResolvedPathWrapper target,
       @Nullable PolarisResolvedPathWrapper secondary) {
     return isAuthorized(
-        authenticatedPolarisPrincipal,
+        polarisPrincipal,
         activatedEntities,
         authzOp,
         target == null ? null : List.of(target),
         secondary == null ? null : List.of(secondary));
   }
 
   public boolean isAuthorized(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPolarisPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable List<PolarisResolvedPathWrapper> targets,
@@ -636,8 +634,7 @@ public boolean isAuthorized(
           authzOp,
           privilegeOnTarget);
       for (PolarisResolvedPathWrapper target : targets) {
-        if (!hasTransitivePrivilege(
-            authenticatedPolarisPrincipal, entityIdSet, privilegeOnTarget, target)) {
+        if (!hasTransitivePrivilege(polarisPrincipal, entityIdSet, privilegeOnTarget, target)) {
           // TODO: Collect missing privileges to report all at the end and/or return to code
           // that throws NotAuthorizedException for more useful messages.
           return false;
@@ -652,7 +649,7 @@ public boolean isAuthorized(
           privilegeOnSecondary);
       for (PolarisResolvedPathWrapper secondary : secondaries) {
         if (!hasTransitivePrivilege(
-            authenticatedPolarisPrincipal, entityIdSet, privilegeOnSecondary, secondary)) {
+            polarisPrincipal, entityIdSet, privilegeOnSecondary, secondary)) {
           return false;
         }
       }
@@ -670,7 +667,7 @@ public boolean isAuthorized(
    * errors/exceptions.
    */
   public boolean hasTransitivePrivilege(
-      @Nonnull AuthenticatedPolarisPrincipal authenticatedPolarisPrincipal,
+      @Nonnull PolarisPrincipal polarisPrincipal,
       Set<Long> activatedGranteeIds,
       PolarisPrivilege desiredPrivilege,
       PolarisResolvedPathWrapper resolvedPath) {
@@ -693,7 +690,7 @@ public boolean hasTransitivePrivilege(
                 desiredPrivilege,
                 grantRecord,
                 resolvedSecurableEntity,
-                authenticatedPolarisPrincipal.getName(),
+                polarisPrincipal.getName(),
                 activatedGranteeIds);
             return true;
           }
@@ -704,7 +701,7 @@ public boolean hasTransitivePrivilege(
     LOGGER.debug(
         "Failed to satisfy privilege {} for principalName {} on resolvedPath {}",
         desiredPrivilege,
-        authenticatedPolarisPrincipal.getName(),
+        polarisPrincipal.getName(),
         resolvedPath);
     return false;
   }

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

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.auth;
+
+import java.security.Principal;
+import java.util.Map;
+import java.util.Set;
+import org.apache.polaris.core.entity.PrincipalEntity;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/** Represents a {@link Principal} in the Polaris system. */
+@PolarisImmutable
+public interface PolarisPrincipal extends Principal {
+
+  /**
+   * Creates a new instance of {@link PolarisPrincipal} from the given {@link PrincipalEntity} and
+   * roles.
+   *
+   * <p>The created principal will have the same ID and name as the {@link PrincipalEntity}, and its
+   * properties will be derived from the internal properties of the entity.
+   *
+   * @param principalEntity the principal entity representing the user or service
+   * @param roles the set of roles associated with the principal
+   */
+  static PolarisPrincipal of(PrincipalEntity principalEntity, Set<String> roles) {
+    return of(
+        principalEntity.getId(),
+        principalEntity.getName(),
+        principalEntity.getInternalPropertiesAsMap(),
+        roles);
+  }
+
+  /**
+   * Creates a new instance of {@link PolarisPrincipal} with the specified ID, name, roles, and
+   * properties.
+   *
+   * @param id the unique identifier of the principal
+   * @param name the name of the principal
+   * @param properties additional properties associated with the principal
+   * @param roles the set of roles associated with the principal
+   */
+  static PolarisPrincipal of(
+      long id, String name, Map<String, String> properties, Set<String> roles) {
+    return ImmutablePolarisPrincipal.builder()
+        .id(id)
+        .name(name)
+        .properties(properties)
+        .roles(roles)
+        .build();
+  }
+
+  /**
+   * Returns the unique identifier of the principal.
+   *
+   * <p>This identifier is used to uniquely identify the principal within a Polaris realm.
+   */
+  long getId();
+
+  /**
+   * Returns the set of activated principal role names. Activated role names are the roles that were
+   * explicitly requested by the client when authenticating, through JWT claims or other means. It
+   * may be a subset of the roles that the principal has in the system.
+   */
+  Set<String> getRoles();
+
+  /**
+   * Returns the properties of this principal.
+   *
+   * <p>Properties are key-value pairs that provide additional information about the principal, such
+   * as permissions, preferences, or other metadata.
+   */
+  Map<String, String> getProperties();
+}

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

@@ -29,7 +29,7 @@
 import java.util.Map;
 import java.util.Set;
 import org.apache.polaris.core.PolarisDiagnostics;
-import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -83,7 +83,7 @@ public PolarisResolutionManifest(
     this.diagnostics.checkNotNull(securityContext, "null_security_context_for_resolution_manifest");
     this.securityContext = securityContext;
     diagnostics.check(
-        securityContext.getUserPrincipal() instanceof AuthenticatedPolarisPrincipal,
+        securityContext.getUserPrincipal() instanceof PolarisPrincipal,
         "invalid_principal_type_for_resolution_manifest",
         "principal={}",
         securityContext.getUserPrincipal());

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/PolarisResolutionManifest.java

@@ -32,7 +32,7 @@
 import java.util.stream.Collectors;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
-import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
@@ -68,7 +68,7 @@ public class Resolver {
   @Nullable private final EntityCache cache;
 
   // the id of the principal making the call or 0 if unknown
-  private final @Nonnull AuthenticatedPolarisPrincipal polarisPrincipal;
+  private final @Nonnull PolarisPrincipal polarisPrincipal;
   private final @Nonnull SecurityContext securityContext;
 
   // reference catalog name for name resolution
@@ -116,7 +116,7 @@ public class Resolver {
    *
    * @param polarisCallContext the polaris call context
    * @param polarisMetaStoreManager meta store manager
-   * @param securityContext The {@link AuthenticatedPolarisPrincipal} for the current request
+   * @param securityContext The {@link SecurityContext} for the current request
    * @param cache shared entity cache
    * @param referenceCatalogName if not null, specifies the name of the reference catalog. The
    *     reference catalog is the catalog used to resolve catalog roles and catalog path. Also, if a
@@ -147,12 +147,12 @@ public Resolver(
     this.diagnostics.checkNotNull(
         securityContext.getUserPrincipal(), "principal_must_be_specified");
     this.diagnostics.check(
-        securityContext.getUserPrincipal() instanceof AuthenticatedPolarisPrincipal,
+        securityContext.getUserPrincipal() instanceof PolarisPrincipal,
         "unexpected_principal_type",
         "class={}",
         securityContext.getUserPrincipal().getClass().getName());
 
-    this.polarisPrincipal = (AuthenticatedPolarisPrincipal) securityContext.getUserPrincipal();
+    this.polarisPrincipal = (PolarisPrincipal) securityContext.getUserPrincipal();
     // paths to resolve
     this.pathsToResolve = new ArrayList<>();
     this.resolvedPaths = new ArrayList<>();
@@ -758,7 +758,7 @@ private ResolverStatus resolveCallerPrincipalAndPrincipalRoles(
             toValidate,
             PolarisEntityType.PRINCIPAL,
             PolarisEntityConstants.getNullId(),
-            polarisPrincipal.getPrincipalEntity().getId());
+            polarisPrincipal.getId());
 
     // if the principal was not found, we can end right there
     if (this.resolvedCallerPrincipal == null
@@ -768,10 +768,9 @@ private ResolverStatus resolveCallerPrincipalAndPrincipalRoles(
 
     // activate all principal roles specified in the authenticated principal
     resolvedCallerPrincipalRoles =
-        this.polarisPrincipal.getActivatedPrincipalRoleNames().isEmpty()
+        this.polarisPrincipal.getRoles().isEmpty()
             ? resolveAllPrincipalRoles(toValidate, resolvedCallerPrincipal)
-            : resolvePrincipalRolesByName(
-                toValidate, this.polarisPrincipal.getActivatedPrincipalRoleNames());
+            : resolvePrincipalRolesByName(toValidate, this.polarisPrincipal.getRoles());
 
     // total success
     return new ResolverStatus(ResolverStatus.StatusEnum.SUCCESS);

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/Resolver.java

@@ -31,7 +31,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.polaris.core.PolarisCallContext;
-import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
+import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntityCore;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
@@ -481,8 +481,8 @@ private Resolver allocateResolver(
                         .filter(Optional::isPresent)
                         .map(Optional::get)
                         .collect(Collectors.toList()));
-    AuthenticatedPolarisPrincipal authenticatedPrincipal =
-        new AuthenticatedPolarisPrincipal(
+    PolarisPrincipal authenticatedPrincipal =
+        PolarisPrincipal.of(
             PrincipalEntity.of(P1), Optional.ofNullable(principalRolesScope).orElse(Set.of()));
     return new Resolver(
         callCtx(),