Auth: reorganize internal authentication components (#2634)

This PR contains no functional and no user-facing change. It is merely a refactor to better organize auth code.

Summary of changes:

- Moved all internal authentication components to the `org.apache.polaris.service.auth.internal` package and subpackages
- Reduced visibility of utility classes
- Renamed `TokenBroker` class hierarchy to stick to the naming standard: `<Algorithm>JWTBroker`
- Introduced `@PolarisImmutable` whenever appropriate
- Removed unused `NoneTokenBrokerFactory` (we already have `DisabledOAuth2ApiService`)
- Removed unused `TokenBrokerFactoryConfig`

Author: adutra

runtime/service/src/main/java/org/apache/polaris/service/auth/AuthenticationRealmConfiguration.java

@@ -22,6 +22,8 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Optional;
+import org.apache.polaris.service.auth.internal.broker.TokenBrokerFactory;
+import org.apache.polaris.service.catalog.api.IcebergRestOAuth2ApiService;
 
 public interface AuthenticationRealmConfiguration {
 
@@ -38,10 +40,7 @@ public interface AuthenticationRealmConfiguration {
 
   interface AuthenticatorConfiguration {
 
-    /**
-     * The type of the identity provider. Must be a registered {@link
-     * org.apache.polaris.service.auth.Authenticator} identifier.
-     */
+    /** The type of the identity provider. Must be a registered {@link Authenticator} identifier. */
     @WithDefault("default")
     String type();
   }
@@ -54,8 +53,8 @@ interface AuthenticatorConfiguration {
 
   interface TokenServiceConfiguration {
     /**
-     * The type of the OAuth2 service. Must be a registered {@link
-     * org.apache.polaris.service.catalog.api.IcebergRestOAuth2ApiService} identifier.
+     * The type of the OAuth2 service. Must be a registered {@link IcebergRestOAuth2ApiService}
+     * identifier.
      */
     @WithDefault("default")
     String type();
@@ -75,8 +74,8 @@ interface TokenBrokerConfiguration {
     Duration maxTokenGeneration();
 
     /**
-     * The type of the token broker factory. Must be a registered {@link
-     * org.apache.polaris.service.auth.TokenBrokerFactory} identifier.
+     * The type of the token broker factory. Must be a registered {@link TokenBrokerFactory}
+     * identifier.
      */
     @WithDefault("rsa-key-pair")
     String type();

runtime/service/src/main/java/org/apache/polaris/service/auth/NoneTokenBrokerFactory.java

@@ -39,7 +39,7 @@
 import org.apache.polaris.service.auth.AuthenticationRealmConfiguration;
 import org.apache.polaris.service.auth.AuthenticationType;
 import org.apache.polaris.service.auth.PolarisCredential;
-import org.apache.polaris.service.auth.TokenBroker;
+import org.apache.polaris.service.auth.internal.broker.TokenBroker;
 
 /**
  * A custom {@link HttpAuthenticationMechanism} that handles internal token authentication, that is,

runtime/service/src/main/java/org/apache/polaris/service/auth/OAuthTokenErrorResponse.java

@@ -16,13 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.polaris.service.auth;
+package org.apache.polaris.service.auth.internal.broker;
 
 import com.google.common.base.Splitter;
 import jakarta.annotation.Nonnull;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.polaris.immutables.PolarisImmutable;
+import org.apache.polaris.service.auth.PolarisCredential;
 import org.immutables.value.Value;
 
 /**

runtime/service/src/main/java/org/apache/polaris/service/auth/TokenResponse.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.polaris.service.auth;
+package org.apache.polaris.service.auth.internal.broker;
 
 import com.auth0.jwt.JWT;
 import com.auth0.jwt.algorithms.Algorithm;
@@ -32,7 +32,10 @@
 import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
-import org.apache.polaris.service.auth.OAuthTokenErrorResponse.Error;
+import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
+import org.apache.polaris.service.auth.DefaultAuthenticator;
+import org.apache.polaris.service.auth.PolarisCredential;
+import org.apache.polaris.service.auth.internal.service.OAuthError;
 import org.apache.polaris.service.types.TokenType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -88,35 +91,35 @@ public TokenResponse generateFromToken(
       PolarisCallContext polarisCallContext,
       TokenType requestedTokenType) {
     if (requestedTokenType != null && !TokenType.ACCESS_TOKEN.equals(requestedTokenType)) {
-      return new TokenResponse(OAuthTokenErrorResponse.Error.invalid_request);
+      return TokenResponse.of(OAuthError.invalid_request);
     }
     if (!TokenType.ACCESS_TOKEN.equals(subjectTokenType)) {
-      return new TokenResponse(OAuthTokenErrorResponse.Error.invalid_request);
+      return TokenResponse.of(OAuthError.invalid_request);
     }
     if (subjectToken == null || subjectToken.isBlank()) {
-      return new TokenResponse(OAuthTokenErrorResponse.Error.invalid_request);
+      return TokenResponse.of(OAuthError.invalid_request);
     }
     InternalPolarisToken decodedToken;
     try {
       decodedToken = verifyInternal(subjectToken);
     } catch (NotAuthorizedException e) {
       LOGGER.error("Failed to verify the token", e.getCause());
-      return new TokenResponse(Error.invalid_client);
+      return TokenResponse.of(OAuthError.invalid_client);
     }
     EntityResult principalLookup =
         metaStoreManager.loadEntity(
             polarisCallContext, 0L, decodedToken.getPrincipalId(), PolarisEntityType.PRINCIPAL);
     if (!principalLookup.isSuccess()
         || principalLookup.getEntity().getType() != PolarisEntityType.PRINCIPAL) {
-      return new TokenResponse(OAuthTokenErrorResponse.Error.unauthorized_client);
+      return TokenResponse.of(OAuthError.unauthorized_client);
     }
     String tokenString =
         generateTokenString(
             decodedToken.getPrincipalName(),
             decodedToken.getPrincipalId(),
             decodedToken.getClientId(),
             decodedToken.getScope());
-    return new TokenResponse(
+    return TokenResponse.of(
         tokenString, TokenType.ACCESS_TOKEN.getValue(), maxTokenGenerationInSeconds);
   }
 
@@ -130,21 +133,20 @@ public TokenResponse generateFromClientSecrets(
       TokenType requestedTokenType) {
     // Initial sanity checks
     TokenRequestValidator validator = new TokenRequestValidator();
-    Optional<OAuthTokenErrorResponse.Error> initialValidationResponse =
+    Optional<OAuthError> initialValidationResponse =
         validator.validateForClientCredentialsFlow(clientId, clientSecret, grantType, scope);
     if (initialValidationResponse.isPresent()) {
-      return new TokenResponse(initialValidationResponse.get());
+      return TokenResponse.of(initialValidationResponse.get());
     }
 
     Optional<PrincipalEntity> principal =
-        TokenBroker.findPrincipalEntity(
-            metaStoreManager, clientId, clientSecret, polarisCallContext);
+        findPrincipalEntity(clientId, clientSecret, polarisCallContext);
     if (principal.isEmpty()) {
-      return new TokenResponse(OAuthTokenErrorResponse.Error.unauthorized_client);
+      return TokenResponse.of(OAuthError.unauthorized_client);
     }
     String tokenString =
         generateTokenString(principal.get().getName(), principal.get().getId(), clientId, scope);
-    return new TokenResponse(
+    return TokenResponse.of(
         tokenString, TokenType.ACCESS_TOKEN.getValue(), maxTokenGenerationInSeconds);
   }
 
@@ -177,4 +179,27 @@ public boolean supportsRequestedTokenType(TokenType tokenType) {
   private String scopes(String scope) {
     return scope == null || scope.isBlank() ? DefaultAuthenticator.PRINCIPAL_ROLE_ALL : scope;
   }
+
+  private Optional<PrincipalEntity> findPrincipalEntity(
+      String clientId, String clientSecret, PolarisCallContext polarisCallContext) {
+    // Validate the principal is present and secrets match
+    PrincipalSecretsResult principalSecrets =
+        metaStoreManager.loadPrincipalSecrets(polarisCallContext, clientId);
+    if (!principalSecrets.isSuccess()) {
+      return Optional.empty();
+    }
+    if (!principalSecrets.getPrincipalSecrets().matchesSecret(clientSecret)) {
+      return Optional.empty();
+    }
+    EntityResult result =
+        metaStoreManager.loadEntity(
+            polarisCallContext,
+            0L,
+            principalSecrets.getPrincipalSecrets().getPrincipalId(),
+            PolarisEntityType.PRINCIPAL);
+    if (!result.isSuccess() || result.getEntity().getType() != PolarisEntityType.PRINCIPAL) {
+      return Optional.empty();
+    }
+    return Optional.of(PrincipalEntity.of(result.getEntity()));
+  }
 }