Moving things around more

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

@@ -17,7 +17,6 @@
 
 package org.apache.kafka.common.security.oauthbearer;
 
-import org.apache.kafka.common.security.auth.AuthenticationConfigurable;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.FileAccessTokenRetriever;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.HttpAccessTokenRetriever;
 
@@ -37,7 +36,7 @@
  * @see FileAccessTokenRetriever
  */
 
-public interface AccessTokenRetriever extends AuthenticationConfigurable {
+public interface AccessTokenRetriever extends OAuthBearerConfigurable {
 
     /**
      * Retrieves a JWT access token in its serialized three-part form. The implementation

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

@@ -17,9 +17,7 @@
 
 package org.apache.kafka.common.security.oauthbearer;
 
-import org.apache.kafka.common.security.auth.AuthenticationConfigurable;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ValidateException;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.DefaultBrokerAccessTokenValidator;
 
 /**
  * An instance of <code>AccessTokenValidator</code> acts as a function object that, given an access
@@ -48,7 +46,7 @@
  *                                    to validate the token's contents and verify the signature
  */
 
-public interface AccessTokenValidator extends AuthenticationConfigurable {
+public interface AccessTokenValidator extends OAuthBearerConfigurable {
 
     /**
      * Accepts an OAuth JWT access token in base-64 encoded format, validates, and returns an

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

@@ -17,7 +17,6 @@
 
 package org.apache.kafka.common.security.oauthbearer;
 
-import org.apache.kafka.common.security.auth.AuthenticationConfigurable;
 import org.jose4j.keys.resolvers.VerificationKeyResolver;
 
 import java.io.Closeable;
@@ -34,6 +33,6 @@
  * @see Closeable
  */
 
-public interface CloseableVerificationKeyResolver extends VerificationKeyResolver, AuthenticationConfigurable {
+public interface CloseableVerificationKeyResolver extends VerificationKeyResolver, OAuthBearerConfigurable {
 
 }

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

@@ -21,14 +21,15 @@
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.FileAccessTokenRetriever;
 
-import javax.security.auth.login.AppConfigurationEntry;
 import java.io.IOException;
 import java.net.URL;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
 
 public class DefaultAccessTokenRetriever implements AccessTokenRetriever {

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

@@ -15,15 +15,20 @@
  * limitations under the License.
  */
 
-package org.apache.kafka.common.security.oauthbearer.internals.secured;
+package org.apache.kafka.common.security.oauthbearer;
 
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
-import org.apache.kafka.common.security.oauthbearer.AccessTokenValidator;
-import org.apache.kafka.common.security.oauthbearer.CloseableVerificationKeyResolver;
-import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
-
+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.utils.Time;
 import org.apache.kafka.common.utils.Utils;
+
 import org.jose4j.jws.JsonWebSignature;
 import org.jose4j.jwt.JwtClaims;
 import org.jose4j.jwt.MalformedClaimException;
@@ -38,8 +43,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.auth.login.AppConfigurationEntry;
-import java.io.IOException;
 import java.security.Key;
 import java.util.Collection;
 import java.util.Collections;
@@ -50,6 +53,8 @@
 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;

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

@@ -28,13 +28,14 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.auth.login.AppConfigurationEntry;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
 import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_SUB_CLAIM_NAME;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;

clients/src/main/java/org/apache/kafka/common/security/auth/AuthenticationConfigurable.java → clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerConfigurable.java

@@ -15,22 +15,23 @@
  * limitations under the License.
  */
 
-package org.apache.kafka.common.security.auth;
+package org.apache.kafka.common.security.oauthbearer;
 
 import org.apache.kafka.common.Configurable;
 
-import javax.security.auth.login.AppConfigurationEntry;
 import java.io.Closeable;
 import java.util.List;
 import java.util.Map;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 /**
  * Analogue to {@link Configurable} for SASL-based authentication.
  *
  * Any resources created in {@link #configure(Map, String, List)} should be cleaned up and released in
  * the call to {@link #close()}.
  */
-public interface AuthenticationConfigurable extends Closeable {
+public interface OAuthBearerConfigurable extends Closeable {
 
     /**
      * Configures this object for the specified SASL mechanism.

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

@@ -17,14 +17,11 @@
 
 package org.apache.kafka.common.security.oauthbearer;
 
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.Retry;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.UnretryableException;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.net.ssl.HttpsURLConnection;
-import javax.net.ssl.SSLSocketFactory;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -35,7 +32,9 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.ExecutionException;
+
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLSocketFactory;
 
 /**
  * <code>OAuthBearerHttpClient</code> is a lightweight client that can be used by callback handlers to

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

@@ -22,27 +22,28 @@
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
-import org.apache.kafka.common.security.auth.AuthenticationConfigurable;
 import org.apache.kafka.common.security.auth.SaslExtensions;
 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.ClientCredentialsAccessTokenRetriever;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.DefaultBrokerAccessTokenValidator;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.JaasOptionsUtils;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ValidateException;
 import org.apache.kafka.common.utils.Utils;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.auth.callback.Callback;
-import javax.security.auth.callback.UnsupportedCallbackException;
-import javax.security.auth.login.AppConfigurationEntry;
-import javax.security.sasl.SaslException;
+import java.io.Closeable;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.sasl.SaslException;
+
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
 
 /**
@@ -148,7 +149,9 @@
  * </p>
  */
 
-public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHandler {
+public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHandler, Closeable {
+
+    private static final Logger log = LoggerFactory.getLogger(OAuthBearerLoginCallbackHandler.class);
 
     public static final String CLIENT_ID_CONFIG = "clientId";
     public static final String CLIENT_SECRET_CONFIG = "clientSecret";
@@ -172,8 +175,6 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
         "OAuth \"scope\". If so, the " + SCOPE_CONFIG + " is used to provide the value to " +
         "include with the login request.";
 
-    private static final Logger log = LoggerFactory.getLogger(OAuthBearerLoginCallbackHandler.class);
-
     private static final String EXTENSION_PREFIX = "extension_";
 
     protected Map<String, Object> moduleOptions;
@@ -285,11 +286,11 @@ protected void checkInitialized() {
     }
 
     @SuppressWarnings("unchecked")
-    private <T extends AuthenticationConfigurable> T newInstance(String configName,
-                                                                 Class<T> defaultClazz,
-                                                                 Map<String, ?> configs,
-                                                                 String saslMechanism,
-                                                                 List<AppConfigurationEntry> jaasConfigEntries) {
+    private <T extends OAuthBearerConfigurable> T newInstance(String configName,
+                                                              Class<T> defaultClazz,
+                                                              Map<String, ?> configs,
+                                                              String saslMechanism,
+                                                              List<AppConfigurationEntry> jaasConfigEntries) {
         Class<T> clazz = (Class<T>) configs.get(configName);
 
         if (clazz == null)

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

@@ -20,14 +20,13 @@
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
-import org.apache.kafka.common.security.auth.AuthenticationConfigurable;
 import org.apache.kafka.common.security.oauthbearer.internals.secured.ValidateException;
-import org.apache.kafka.common.security.oauthbearer.internals.secured.DefaultBrokerAccessTokenValidator;
-
 import org.apache.kafka.common.utils.Utils;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -95,7 +94,7 @@
  * </p>
  */
 
-public class OAuthBearerValidatorCallbackHandler implements AuthenticateCallbackHandler {
+public class OAuthBearerValidatorCallbackHandler implements AuthenticateCallbackHandler, Closeable {
 
     private static final Logger log = LoggerFactory.getLogger(OAuthBearerValidatorCallbackHandler.class);
 
@@ -167,11 +166,11 @@ private void checkInitialized() {
     }
 
     @SuppressWarnings("unchecked")
-    private <T extends AuthenticationConfigurable> T newInstance(String configName,
-                                                                 Class<T> defaultClazz,
-                                                                 Map<String, ?> configs,
-                                                                 String saslMechanism,
-                                                                 List<AppConfigurationEntry> jaasConfigEntries) {
+    private <T extends OAuthBearerConfigurable> T newInstance(String configName,
+                                                              Class<T> defaultClazz,
+                                                              Map<String, ?> configs,
+                                                              String saslMechanism,
+                                                              List<AppConfigurationEntry> jaasConfigEntries) {
         Class<T> clazz = (Class<T>) configs.get(configName);
 
         if (clazz == null)

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

@@ -21,17 +21,16 @@
 import org.apache.kafka.common.security.oauthbearer.AccessTokenRetriever;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
 import org.apache.kafka.common.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-import javax.security.auth.login.AppConfigurationEntry;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE;
 import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_ID_CONFIG;

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

@@ -19,20 +19,20 @@
 
 import org.apache.kafka.common.security.oauthbearer.CloseableVerificationKeyResolver;
 import org.apache.kafka.common.utils.Time;
-
 import org.apache.kafka.common.utils.Utils;
+
 import org.jose4j.jws.JsonWebSignature;
 import org.jose4j.jwx.JsonWebStructure;
 import org.jose4j.lang.UnresolvableKeyException;
 
-import javax.security.auth.login.AppConfigurationEntry;
-import java.io.IOException;
 import java.net.URL;
 import java.security.Key;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL;
 
 public class DelegatingVerificationKeyResolver implements CloseableVerificationKeyResolver {