KAFKA-18573: Add support for OAuth jwt-bearer grant type

checkstyle/import-control.xml

@@ -145,6 +145,7 @@
       <subpackage name="oauthbearer">
         <allow pkg="com.fasterxml.jackson.databind" />
         <allow pkg="org.jose4j" />
+        <allow pkg="javax.crypto"/>
       </subpackage>
     </subpackage>
 

clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java

@@ -136,6 +136,10 @@ public class BrokerSecurityConfigs {
     // The allowlist of the SASL OAUTHBEARER endpoints
     public static final String ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG = "org.apache.kafka.sasl.oauthbearer.allowed.urls";
     public static final String ALLOWED_SASL_OAUTHBEARER_URLS_DEFAULT = "";
+
+    public static final String ALLOWED_SASL_OAUTHBEARER_FILES_CONFIG = "org.apache.kafka.sasl.oauthbearer.allowed.files";
+    public static final String ALLOWED_SASL_OAUTHBEARER_FILES_DEFAULT = "";
+
     public static final ConfigDef CONFIG_DEF =  new ConfigDef()
             // General Security Configuration
             .define(BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_CONFIG, LONG, BrokerSecurityConfigs.DEFAULT_CONNECTIONS_MAX_REAUTH_MS, MEDIUM, BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_DOC)
@@ -190,6 +194,22 @@ public class BrokerSecurityConfigs {
             .define(SaslConfigs.SASL_LOGIN_READ_TIMEOUT_MS, INT, null, LOW, SaslConfigs.SASL_LOGIN_READ_TIMEOUT_MS_DOC)
             .define(SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MAX_MS, LONG, SaslConfigs.DEFAULT_SASL_LOGIN_RETRY_BACKOFF_MAX_MS, LOW, SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MAX_MS_DOC)
             .define(SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MS, LONG, SaslConfigs.DEFAULT_SASL_LOGIN_RETRY_BACKOFF_MS, LOW, SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MS_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_JWT_RETRIEVER_CLASS, CLASS, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_JWT_RETRIEVER_CLASS, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_JWT_RETRIEVER_CLASS_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_JWT_VALIDATOR_CLASS, CLASS, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_JWT_VALIDATOR_CLASS, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_JWT_VALIDATOR_CLASS_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_SCOPE, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_SCOPE_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_ID, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_ID_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_SECRET, PASSWORD, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_SECRET_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM, STRING, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_ASSERTION_ALGORITHM, ConfigDef.CaseInsensitiveValidString.in("ES256", "RS256"), MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS, INT, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS, ConfigDef.Range.between(0, 86400), LOW, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE, BOOLEAN, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS, INT, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS, ConfigDef.Range.between(0, 3600), LOW, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE, PASSWORD, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE_DOC)
+            .define(SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE_DOC)
             .define(SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME, STRING, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_SCOPE_CLAIM_NAME, LOW, SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME_DOC)
             .define(SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME, STRING, SaslConfigs.DEFAULT_SASL_OAUTHBEARER_SUB_CLAIM_NAME, LOW, SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME_DOC)
             .define(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, STRING, null, MEDIUM, SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL_DOC)

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

@@ -15,9 +15,14 @@
  * 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.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.CloseableVerificationKeyResolver;
+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.VerificationKeyResolverFactory;
 
 import org.jose4j.jwt.JwtClaims;
 import org.jose4j.jwt.MalformedClaimException;
@@ -27,14 +32,23 @@
 import org.jose4j.jwt.consumer.JwtConsumer;
 import org.jose4j.jwt.consumer.JwtConsumerBuilder;
 import org.jose4j.jwt.consumer.JwtContext;
-import org.jose4j.keys.resolvers.VerificationKeyResolver;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
+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;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
 import static org.jose4j.jwa.AlgorithmConstraints.DISALLOW_NONE;
 
 /**
@@ -43,16 +57,18 @@
  * 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:
+ * The validation steps performed (primarily 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
+ *         Basic conversion of the token into an in-memory data structure
+ *     </li>
+ *     <li>
+ *         Presence of <code>scope</code>, <code>exp</code>, <code>subject</code>, <code>iss</code>, and
  *         <code>iat</code> claims
  *     </li>
  *     <li>
@@ -61,63 +77,46 @@
  *     </li>
  * </ol>
  */
-
 public class BrokerJwtValidator implements JwtValidator {
 
     private static final Logger log = LoggerFactory.getLogger(BrokerJwtValidator.class);
 
-    private final JwtConsumer jwtConsumer;
+    private final Optional<CloseableVerificationKeyResolver> verificationKeyResolverOpt;
 
-    private final String scopeClaimName;
+    private JwtConsumer jwtConsumer;
 
-    private final String subClaimName;
+    private String scopeClaimName;
+
+    private String subClaimName;
 
     /**
-     * Creates a new {@code BrokerJwtValidator} that will be used by the broker for more
-     * thorough validation of the JWT.
-     *
-     * @param clockSkew               The optional value (in seconds) to allow for differences
-     *                                between the time of the OAuth/OIDC identity provider and
-     *                                the broker. If <code>null</code> is provided, the broker
-     *                                and the OAUth/OIDC identity provider are assumed to have
-     *                                very close clock settings.
-     * @param expectedAudiences       The (optional) set the broker will use to verify that
-     *                                the JWT was issued for one of the expected audiences.
-     *                                The JWT will be inspected for the standard OAuth
-     *                                <code>aud</code> claim and if this value is set, the
-     *                                broker will match the value from JWT's <code>aud</code>
-     *                                claim to see if there is an <b>exact</b> match. If there is no
-     *                                match, the broker will reject the JWT and authentication
-     *                                will fail. May be <code>null</code> to not perform any
-     *                                check to verify the JWT's <code>aud</code> claim matches any
-     *                                fixed set of known/expected audiences.
-     * @param expectedIssuer          The (optional) value for the broker to use to verify that
-     *                                the JWT was created by the expected issuer. The JWT will
-     *                                be inspected for the standard OAuth <code>iss</code> claim
-     *                                and if this value is set, the broker will match it
-     *                                <b>exactly</b> against what is in the JWT's <code>iss</code>
-     *                                claim. If there is no match, the broker will reject the JWT
-     *                                and authentication will fail. May be <code>null</code> to not
-     *                                perform any check to verify the JWT's <code>iss</code> claim
-     *                                matches a specific issuer.
-     * @param verificationKeyResolver jose4j-based {@link VerificationKeyResolver} that is used
-     *                                to validate the signature matches the contents of the header
-     *                                and payload
-     * @param scopeClaimName          Name of the scope claim to use; must be non-<code>null</code>
-     * @param subClaimName            Name of the subject claim to use; must be
-     *                                non-<code>null</code>
-     *
-     * @see JwtConsumerBuilder
-     * @see JwtConsumer
-     * @see VerificationKeyResolver
+     * A public, no-args constructor is necessary for instantiation via configuration.
      */
+    public BrokerJwtValidator() {
+        this.verificationKeyResolverOpt = Optional.empty();
+    }
+
+    /*
+     * Package-visible for testing.
+     */
+    BrokerJwtValidator(CloseableVerificationKeyResolver verificationKeyResolver) {
+        this.verificationKeyResolverOpt = Optional.of(verificationKeyResolver);
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
+        ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+        List<String> expectedAudiencesList = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+        Set<String> expectedAudiences = expectedAudiencesList != null ? Set.copyOf(expectedAudiencesList) : null;
+        Integer clockSkew = cu.validateInteger(SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, false);
+        String expectedIssuer = cu.validateString(SASL_OAUTHBEARER_EXPECTED_ISSUER, false);
+        String scopeClaimName = cu.validateString(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME);
+        String subClaimName = cu.validateString(SASL_OAUTHBEARER_SUB_CLAIM_NAME);
+
+        CloseableVerificationKeyResolver verificationKeyResolver = verificationKeyResolverOpt.orElseGet(
+            () -> VerificationKeyResolverFactory.get(configs, saslMechanism, jaasConfigEntries)
+        );
 
-    public BrokerJwtValidator(Integer clockSkew,
-                              Set<String> expectedAudiences,
-                              String expectedIssuer,
-                              VerificationKeyResolver verificationKeyResolver,
-                              String scopeClaimName,
-                              String subClaimName) {
         final JwtConsumerBuilder jwtConsumerBuilder = new JwtConsumerBuilder();
 
         if (clockSkew != null)
@@ -145,19 +144,19 @@ public BrokerJwtValidator(Integer clockSkew,
      *
      * @param accessToken Non-<code>null</code> JWT access token
      * @return {@link OAuthBearerToken}
-     * @throws ValidateException Thrown on errors performing validation of given token
+     * @throws JwtValidatorException Thrown on errors performing validation of given token
      */
 
     @SuppressWarnings("unchecked")
-    public OAuthBearerToken validate(String accessToken) throws ValidateException {
+    public OAuthBearerToken validate(String accessToken) throws JwtValidatorException {
         SerializedJwt serializedJwt = new SerializedJwt(accessToken);
 
         JwtContext jwt;
 
         try {
             jwt = jwtConsumer.process(serializedJwt.getToken());
         } catch (InvalidJwtException e) {
-            throw new ValidateException(String.format("Could not validate the access token: %s", e.getMessage()), e);
+            throw new JwtValidatorException(String.format("Could not validate the access token: %s", e.getMessage()), e);
         }
 
         JwtClaims claims = jwt.getJwtClaims();
@@ -190,13 +189,13 @@ else if (scopeRaw instanceof Collection)
             issuedAt);
     }
 
-    private <T> T getClaim(ClaimSupplier<T> supplier, String claimName) throws ValidateException {
+    private <T> T getClaim(ClaimSupplier<T> supplier, String claimName) throws JwtValidatorException {
         try {
             T value = supplier.get();
             log.debug("getClaim - {}: {}", claimName, value);
             return value;
         } catch (MalformedClaimException e) {
-            throw new ValidateException(String.format("Could not extract the '%s' claim from the access token", claimName), e);
+            throw new JwtValidatorException(String.format("Could not extract the '%s' claim from the access token", claimName), e);
         }
     }
 
@@ -205,5 +204,4 @@ public interface ClaimSupplier<T> {
         T get() throws MalformedClaimException;
 
     }
-
 }