Updates

clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java

@@ -134,7 +134,7 @@ public class SaslConfigs {
     public static final String SASL_OAUTHBEARER_ACCESS_TOKEN_RETRIEVER_CLASS_DOC = "The fully qualified name of a class that implements the AccessTokenRetriever interface.";
 
     public static final String SASL_OAUTHBEARER_ACCESS_TOKEN_VALIDATOR_CLASS = "sasl.oauthbearer.access.token.validator.class";
-    public static final String DEFAULT_SASL_OAUTHBEARER_ACCESS_TOKEN_VALIDATOR_CLASS = "org.apache.kafka.common.security.oauthbearer.DefaultClientAccessTokenValidator";
+    public static final String DEFAULT_SASL_OAUTHBEARER_ACCESS_TOKEN_VALIDATOR_CLASS = "org.apache.kafka.common.security.oauthbearer.DefaultAccessTokenValidator";
     public static final String SASL_OAUTHBEARER_ACCESS_TOKEN_VALIDATOR_CLASS_DOC = "The fully qualified name of a class that implements the AccessTokenValidator interface.";
 
     public static final String SASL_OAUTHBEARER_SCOPE_CLAIM_NAME = "sasl.oauthbearer.scope.claim.name";

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/AccessTokenValidator.java

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.common.security.oauthbearer;
 
+import org.apache.kafka.common.security.oauthbearer.internals.secured.ClientAccessTokenValidator;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ValidateException;
 
 /**
@@ -40,9 +41,9 @@
  *     <li><a href="https://datatracker.ietf.org/doc/html/draft-ietf-oauth-access-token-jwt">RFC 6750, Section 2.1</a></li>
  * </ul>
  *
- * @see DefaultClientAccessTokenValidator A basic AccessTokenValidator used by client-side login
+ * @see ClientAccessTokenValidator A basic AccessTokenValidator used by client-side login
  *                                authentication
- * @see DefaultBrokerAccessTokenValidator A more robust AccessTokenValidator that is used on the broker
+ * @see DefaultAccessTokenValidator A more robust AccessTokenValidator that is used on the broker
  *                                    to validate the token's contents and verify the signature
  */
 

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/DefaultAccessTokenRetriever.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ClientCredentialsAccessTokenRetriever;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.FileAccessTokenRetriever;
+import org.apache.kafka.common.utils.Utils;
 
 import java.io.IOException;
 import java.net.URL;
@@ -56,8 +57,7 @@ public String retrieve() throws IOException {
 
     @Override
     public void close() {
-        if (delegate != null)
-            delegate.close();
+        Utils.closeQuietly(delegate, "delegate");
     }
 
     public AccessTokenRetriever delegate() {

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/DefaultAccessTokenValidator.java

@@ -0,0 +1,57 @@
+/*
+ * 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.kafka.common.security.oauthbearer;
+
+import org.apache.kafka.common.security.oauthbearer.internals.secured.BrokerAccessTokenValidator;
+import org.apache.kafka.common.utils.Time;
+
+/**
+ * Implementation of {@link AccessTokenValidator} that is used
+ * by the broker to perform more extensive validation of the JWT access token that is received
+ * from the client, but ultimately from posting the client credentials to the OAuth/OIDC provider's
+ * token endpoint.
+ *
+ * The validation steps performed (primary by the jose4j library) are:
+ *
+ * <ol>
+ *     <li>
+ *         Basic structural validation of the <code>b64token</code> value as defined in
+ *         <a href="https://tools.ietf.org/html/rfc6750#section-2.1">RFC 6750 Section 2.1</a>
+ *     </li>
+ *     <li>Basic conversion of the token into an in-memory data structure</li>
+ *     <li>
+ *         Presence of scope, <code>exp</code>, subject, <code>iss</code>, and
+ *         <code>iat</code> claims
+ *     </li>
+ *     <li>
+ *         Signature matching validation against the <code>kid</code> and those provided by
+ *         the OAuth/OIDC provider's JWKS
+ *     </li>
+ * </ol>
+ */
+
+public class DefaultAccessTokenValidator extends BrokerAccessTokenValidator {
+
+    public DefaultAccessTokenValidator() {
+        super(Time.SYSTEM);
+    }
+
+    public DefaultAccessTokenValidator(Time time) {
+        super(time);
+    }
+}

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java

@@ -201,7 +201,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
             accessTokenValidator = newInstance(
                 SaslConfigs.SASL_OAUTHBEARER_ACCESS_TOKEN_VALIDATOR_CLASS,
-                DefaultBrokerAccessTokenValidator.class,
+                DefaultAccessTokenValidator.class,
                 configs,
                 saslMechanism,
                 jaasConfigEntries

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java

@@ -108,7 +108,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
         try {
             accessTokenValidator = newInstance(
                 SaslConfigs.SASL_OAUTHBEARER_ACCESS_TOKEN_VALIDATOR_CLASS,
-                DefaultBrokerAccessTokenValidator.class,
+                DefaultAccessTokenValidator.class,
                 configs,
                 saslMechanism,
                 jaasConfigEntries

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/DefaultBrokerAccessTokenValidator.java → clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/BrokerAccessTokenValidator.java

@@ -15,20 +15,13 @@
  * limitations under the License.
  */
 
-package org.apache.kafka.common.security.oauthbearer;
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.BasicOAuthBearerToken;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.ClaimValidationUtils;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.DelegatingVerificationKeyResolver;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.JaasOptionsUtils;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.RefreshingHttpsJwksVerificationKeyResolver;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.SerializedJwt;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.ValidateException;
+import org.apache.kafka.common.security.oauthbearer.AccessTokenValidator;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
-
 import org.jose4j.jws.JsonWebSignature;
 import org.jose4j.jwt.JwtClaims;
 import org.jose4j.jwt.MalformedClaimException;
@@ -43,6 +36,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.security.auth.login.AppConfigurationEntry;
 import java.security.Key;
 import java.util.Collection;
 import java.util.Collections;
@@ -53,8 +47,6 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import javax.security.auth.login.AppConfigurationEntry;
-
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
@@ -87,9 +79,9 @@
  * </ol>
  */
 
-public class DefaultBrokerAccessTokenValidator implements AccessTokenValidator {
+public class BrokerAccessTokenValidator implements AccessTokenValidator {
 
-    private static final Logger log = LoggerFactory.getLogger(DefaultBrokerAccessTokenValidator.class);
+    private static final Logger log = LoggerFactory.getLogger(BrokerAccessTokenValidator.class);
 
     /**
      * Because a {@link CloseableVerificationKeyResolver} instance can spawn threads and issue
@@ -110,11 +102,11 @@ public class DefaultBrokerAccessTokenValidator implements AccessTokenValidator {
 
     protected String subClaimName;
 
-    public DefaultBrokerAccessTokenValidator() {
+    public BrokerAccessTokenValidator() {
         this(Time.SYSTEM);
     }
 
-    public DefaultBrokerAccessTokenValidator(Time time) {
+    public BrokerAccessTokenValidator(Time time) {
         this.time = time;
     }
 

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/DefaultClientAccessTokenValidator.java → clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientAccessTokenValidator.java

@@ -15,13 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.kafka.common.security.oauthbearer;
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
-import org.apache.kafka.common.security.oauthbearer.internals.secured.BasicOAuthBearerToken;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.ClaimValidationUtils;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.SerializedJwt;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.ValidateException;
+import org.apache.kafka.common.security.oauthbearer.AccessTokenValidator;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
 import org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerIllegalTokenException;
 import org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredJws;
 
@@ -59,9 +56,9 @@
  * </ol>
  */
 
-public class DefaultClientAccessTokenValidator implements AccessTokenValidator {
+public class ClientAccessTokenValidator implements AccessTokenValidator {
 
-    private static final Logger log = LoggerFactory.getLogger(DefaultClientAccessTokenValidator.class);
+    private static final Logger log = LoggerFactory.getLogger(ClientAccessTokenValidator.class);
 
     public static final String EXPIRATION_CLAIM_NAME = "exp";
 

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/CloseableVerificationKeyResolver.java → clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/CloseableVerificationKeyResolver.java

@@ -15,8 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.kafka.common.security.oauthbearer;
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerConfigurable;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler;
 import org.jose4j.keys.resolvers.VerificationKeyResolver;
 
 import java.io.Closeable;

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DelegatingVerificationKeyResolver.java

@@ -17,7 +17,6 @@
 
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
-import org.apache.kafka.common.security.oauthbearer.CloseableVerificationKeyResolver;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwksFileVerificationKeyResolver.java

@@ -18,7 +18,6 @@
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.security.oauthbearer.CloseableVerificationKeyResolver;
 import org.apache.kafka.common.utils.Utils;
 
 import org.jose4j.jwk.JsonWebKeySet;

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/RefreshingHttpsJwks.java

@@ -18,7 +18,7 @@
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.security.oauthbearer.DefaultBrokerAccessTokenValidator;
+import org.apache.kafka.common.security.oauthbearer.DefaultAccessTokenValidator;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerConfigurable;
 import org.apache.kafka.common.utils.Time;
 
@@ -60,12 +60,12 @@
  * This instance is created and provided to the
  * {@link org.jose4j.keys.resolvers.HttpsJwksVerificationKeyResolver} that is used when using
  * an HTTP-/HTTPS-based {@link org.jose4j.keys.resolvers.VerificationKeyResolver}, which is then
- * provided to the {@link DefaultBrokerAccessTokenValidator} to use in validating the signature of
+ * provided to the {@link DefaultAccessTokenValidator} to use in validating the signature of
  * a JWT.
  *
  * @see org.jose4j.keys.resolvers.HttpsJwksVerificationKeyResolver
  * @see org.jose4j.keys.resolvers.VerificationKeyResolver
- * @see DefaultBrokerAccessTokenValidator
+ * @see DefaultAccessTokenValidator
  */
 
 public class RefreshingHttpsJwks implements OAuthBearerConfigurable {

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/RefreshingHttpsJwksVerificationKeyResolver.java

@@ -17,7 +17,6 @@
 
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
-import org.apache.kafka.common.security.oauthbearer.CloseableVerificationKeyResolver;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultBrokerAccessTokenValidatorTest.java → clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultAccessTokenValidatorTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
 
+import org.apache.kafka.common.security.oauthbearer.internals.secured.CloseableVerificationKeyResolver;
 import org.jose4j.jwk.PublicJsonWebKey;
 import org.jose4j.jws.AlgorithmIdentifiers;
 import org.jose4j.lang.InvalidAlgorithmException;
@@ -35,15 +36,15 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-public class DefaultBrokerAccessTokenValidatorTest extends AccessTokenValidatorTest {
+public class DefaultAccessTokenValidatorTest extends AccessTokenValidatorTest {
 
     @Override
     protected AccessTokenValidator createAccessTokenValidator(AccessTokenBuilder builder) throws Exception {
         Key key = builder.jwk() != null ? builder.jwk().getKey() : null;
         CloseableVerificationKeyResolver keyResolver = mock(CloseableVerificationKeyResolver.class);
         when(keyResolver.resolveKey(any(), any())).thenReturn(key);
 
-        return new DefaultBrokerAccessTokenValidator() {
+        return new DefaultAccessTokenValidator() {
             @Override
             public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
                 super.configure(keyResolver, configs, saslMechanism, jaasConfigEntries);

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandlerTest.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
 import org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerClientInitialResponse;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
+import org.apache.kafka.common.security.oauthbearer.internals.secured.ClientAccessTokenValidator;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.FileAccessTokenRetriever;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.HttpAccessTokenRetriever;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest;
@@ -137,7 +138,7 @@ public void testInvalidCallbackGeneratesUnsupportedCallbackException() throws IO
         OAuthBearerTestableLoginCallbackHandler handler = new OAuthBearerTestableLoginCallbackHandler();
         AccessTokenRetriever accessTokenRetriever = mock(AccessTokenRetriever.class);
         when(accessTokenRetriever.retrieve()).thenReturn("foo");
-        AccessTokenValidator accessTokenValidator = new DefaultClientAccessTokenValidator();
+        AccessTokenValidator accessTokenValidator = new ClientAccessTokenValidator();
         handler.init(accessTokenRetriever, accessTokenValidator);
 
         try {

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java

@@ -18,6 +18,7 @@
 package org.apache.kafka.common.security.oauthbearer;
 
 import org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
+import org.apache.kafka.common.security.oauthbearer.internals.secured.CloseableVerificationKeyResolver;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest;
 
 import org.jose4j.jws.AlgorithmIdentifiers;
@@ -96,7 +97,7 @@ private OAuthBearerValidatorCallbackHandler createHandler(Map<String, ?> configs
         CloseableVerificationKeyResolver verificationKeyResolver = mock(CloseableVerificationKeyResolver.class);
         Key key = builder.jwk() != null ? builder.jwk().getPublicKey() : null;
         when(verificationKeyResolver.resolveKey(any(), any())).thenReturn(key);
-        DefaultBrokerAccessTokenValidator accessTokenValidator = new DefaultBrokerAccessTokenValidator(time);
+        DefaultAccessTokenValidator accessTokenValidator = new DefaultAccessTokenValidator(time);
         accessTokenValidator.configure(verificationKeyResolver, configs, null, List.of());
         handler.configure(accessTokenValidator);
         return handler;

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultClientAccessTokenValidatorTest.java → clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientAccessTokenValidatorTest.java

@@ -15,10 +15,13 @@
  * limitations under the License.
  */
 
-package org.apache.kafka.common.security.oauthbearer;
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
+import org.apache.kafka.common.security.oauthbearer.AccessTokenRetriever;
+import org.apache.kafka.common.security.oauthbearer.AccessTokenValidator;
+import org.apache.kafka.common.security.oauthbearer.AccessTokenValidatorTest;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerTestableLoginCallbackHandler;
 
 import org.junit.jupiter.api.Test;
 
@@ -28,11 +31,11 @@
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 
-public class DefaultClientAccessTokenValidatorTest extends AccessTokenValidatorTest {
+public class ClientAccessTokenValidatorTest extends AccessTokenValidatorTest {
 
     @Override
     protected AccessTokenValidator createAccessTokenValidator(AccessTokenBuilder builder) {
-        return new DefaultClientAccessTokenValidator();
+        return new ClientAccessTokenValidator();
     }
 
     @Test

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DelegatingVerificationKeyResolverTest.java

@@ -18,7 +18,6 @@
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
 import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.common.security.oauthbearer.CloseableVerificationKeyResolver;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/OAuthBearerTest.java

@@ -23,7 +23,6 @@
 import org.apache.kafka.common.security.authenticator.TestJaasConfig;
 import org.apache.kafka.common.security.oauthbearer.AccessTokenRetriever;
 import org.apache.kafka.common.security.oauthbearer.AccessTokenValidator;
-import org.apache.kafka.common.security.oauthbearer.DefaultClientAccessTokenValidator;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerTestableLoginCallbackHandler;
@@ -114,7 +113,7 @@ protected OAuthBearerLoginCallbackHandler createHandler(AccessTokenRetriever acc
                                                             Map<String, ?> configs) {
         List<AppConfigurationEntry> jaasConfigEntries = List.of();
         OAuthBearerTestableLoginCallbackHandler handler = new OAuthBearerTestableLoginCallbackHandler();
-        AccessTokenValidator accessTokenValidator = new DefaultClientAccessTokenValidator();
+        AccessTokenValidator accessTokenValidator = new ClientAccessTokenValidator();
         accessTokenRetriever.configure(configs, null, jaasConfigEntries);
         accessTokenValidator.configure(configs, null, jaasConfigEntries);
         handler.init(accessTokenRetriever, accessTokenValidator);