[DO NOT MERGE] OAuth jwt-bearer grant type support

.semaphore/semaphore.yml

@@ -17,7 +17,7 @@ version: v1.0
 name: build-test-release
 agent:
   machine:
-    type: s1-prod-ubuntu20-04-arm64-1
+    type: s1-prod-ubuntu24-04-arm64-2
 fail_fast:
   cancel:
     when: "true"

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

@@ -0,0 +1,84 @@
+/*
+ * 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.config.SaslConfigs;
+
+/**
+ * {@code OAuthBearerJaasOptions} holds the different options that can be configured by the user in the
+ * {@link SaslConfigs#SASL_JAAS_CONFIG} configuration.
+ */
+public class OAuthBearerJaasOptions {
+
+    /**
+     * This provides a "namespace" for the jwt-bearer grant type-related JAAS options. The existing implementation
+     * only supported the client_credentials grant type, so those do not have a prefix.
+     */
+    private static final String JWT_BEARER_PREFIX = "jwt-bearer.";
+
+    /**
+     * The {@code grantType} JAAS option is how the user specifies which OAuth grant type to use.
+     */
+    public static final String GRANT_TYPE = "grantType";
+
+    /**
+     * The scope value can be used by multiple grant types, so it has no prefix.
+     */
+    public static final String SCOPE = "scope";
+
+    public static final String CLIENT_CREDENTIALS_CLIENT_ID = "clientId";
+
+    public static final String CLIENT_CREDENTIALS_CLIENT_SECRET = "clientSecret";
+
+    public static final String JWT_BEARER_PRIVATE_KEY_ID = JWT_BEARER_PREFIX + "privateKeyId";
+
+    public static final String JWT_BEARER_PRIVATE_KEY_FILE_NAME = JWT_BEARER_PREFIX + "privateKeyFileName";
+
+    public static final String JWT_BEARER_PRIVATE_KEY_ALGORITHM = JWT_BEARER_PREFIX + "privateKeyAlgorithm";
+
+    /**
+     * The jwt-bearer grant type requires a JWT be created on the client side, signed, and sent in the token
+     * request. The JWT is built up from claims statically set in the JAAS options:
+     *
+     * <ul>
+     *     <li>jwt-bearer.claim.sub=some-service-account</li>
+     *     <li>jwt-bearer.claim.aud=my_audience</li>
+     *     <li>jwt-bearer.claim.iss=https://example.com</li>
+     * </ul>
+     *
+     * Those claims are used to create a JWT that looks like this:
+     *
+     * <pre>
+     *    {
+     *        "iat": 1741121401,
+     *        "exp": 1741125001,
+     *        "sub": "some-service-account",
+     *        "aud": "my_audience",
+     *        "iss": "https://example.com",
+     *        "...": "...",
+     *    }
+     * </pre>
+     *
+     * <b>Note:</b> the JAAS format for specifying claims will likely change in the future to facilitate support for
+     * arbitrarily complicated JWT JSON that includes integers, longs, booleans, lists, and maps.
+     */
+    public static final String JWT_BEARER_CLAIM_PREFIX = JWT_BEARER_PREFIX + "claim.";
+
+    private OAuthBearerJaasOptions() {
+        // Intentionally empty to prevent instantiation.
+    }
+}

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

@@ -30,6 +30,7 @@
 import org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenValidatorFactory;
 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.Time;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,8 +45,6 @@
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.sasl.SaslException;
 
-import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
-
 /**
  * <p>
  * <code>OAuthBearerLoginCallbackHandler</code> is an {@link AuthenticateCallbackHandler} that
@@ -153,30 +152,10 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
 
     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";
-    public static final String SCOPE_CONFIG = "scope";
-
-    public static final String CLIENT_ID_DOC = "The OAuth/OIDC identity provider-issued " +
-        "client ID to uniquely identify the service account to use for authentication for " +
-        "this client. The value must be paired with a corresponding " + CLIENT_SECRET_CONFIG + " " +
-        "value and is provided to the OAuth provider using the OAuth " +
-        "clientcredentials grant type.";
-
-    public static final String CLIENT_SECRET_DOC = "The OAuth/OIDC identity provider-issued " +
-        "client secret serves a similar function as a password to the " + CLIENT_ID_CONFIG + " " +
-        "account and identifies the service account to use for authentication for " +
-        "this client. The value must be paired with a corresponding " + CLIENT_ID_CONFIG + " " +
-        "value and is provided to the OAuth provider using the OAuth " +
-        "clientcredentials grant type.";
-
-    public static final String SCOPE_DOC = "The (optional) HTTP/HTTPS login request to the " +
-        "token endpoint (" + SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL + ") may need to specify an " +
-        "OAuth \"scope\". If so, the " + SCOPE_CONFIG + " is used to provide the value to " +
-        "include with the login request.";
-
     private static final String EXTENSION_PREFIX = "extension_";
 
+    private final Time time;
+
     private Map<String, Object> moduleOptions;
 
     private AccessTokenRetriever accessTokenRetriever;
@@ -185,10 +164,18 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
 
     private boolean isInitialized = false;
 
+    public OAuthBearerLoginCallbackHandler() {
+        this(Time.SYSTEM);
+    }
+
+    public OAuthBearerLoginCallbackHandler(Time time) {
+        this.time = time;
+    }
+
     @Override
     public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
         moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
-        AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
+        AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(time, configs, saslMechanism, moduleOptions);
         AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
         init(accessTokenRetriever, accessTokenValidator);
     }

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

@@ -17,11 +17,19 @@
 
 package org.apache.kafka.common.security.oauthbearer.internals.secured;
 
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
 
+import java.io.IOException;
 import java.net.URL;
+import java.nio.file.Path;
+import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
+import java.util.function.Supplier;
 
 import javax.net.ssl.SSLSocketFactory;
 
@@ -32,9 +40,14 @@
 import static org.apache.kafka.common.config.SaslConfigs.SASL_LOGIN_RETRY_BACKOFF_MS;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
-import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_ID_CONFIG;
-import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_SECRET_CONFIG;
-import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.SCOPE_CONFIG;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.CLIENT_CREDENTIALS_CLIENT_ID;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.CLIENT_CREDENTIALS_CLIENT_SECRET;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.GRANT_TYPE;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.JWT_BEARER_CLAIM_PREFIX;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.JWT_BEARER_PRIVATE_KEY_ALGORITHM;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.JWT_BEARER_PRIVATE_KEY_FILE_NAME;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.JWT_BEARER_PRIVATE_KEY_ID;
+import static org.apache.kafka.common.security.oauthbearer.OAuthBearerJaasOptions.SCOPE;
 
 public class AccessTokenRetrieverFactory  {
 
@@ -44,17 +57,19 @@ public class AccessTokenRetrieverFactory  {
      * <b>Note</b>: the returned <code>AccessTokenRetriever</code> is <em>not</em> initialized
      * here and must be done by the caller prior to use.
      *
+     * @param time       Time
      * @param configs    SASL configuration
      * @param jaasConfig JAAS configuration
      *
      * @return Non-<code>null</code> {@link AccessTokenRetriever}
      */
 
-    public static AccessTokenRetriever create(Map<String, ?> configs, Map<String, Object> jaasConfig) {
-        return create(configs, null, jaasConfig);
+    public static AccessTokenRetriever create(Time time, Map<String, ?> configs, Map<String, Object> jaasConfig) {
+        return create(time, configs, null, jaasConfig);
     }
 
-    public static AccessTokenRetriever create(Map<String, ?> configs,
+    public static AccessTokenRetriever create(Time time,
+        Map<String, ?> configs,
         String saslMechanism,
         Map<String, Object> jaasConfig) {
         ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
@@ -64,27 +79,67 @@ public static AccessTokenRetriever create(Map<String, ?> configs,
             return new FileTokenRetriever(cu.validateFile(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL));
         } else {
             JaasOptionsUtils jou = new JaasOptionsUtils(jaasConfig);
-            String clientId = jou.validateString(CLIENT_ID_CONFIG);
-            String clientSecret = jou.validateString(CLIENT_SECRET_CONFIG);
-            String scope = jou.validateString(SCOPE_CONFIG, false);
-
             SSLSocketFactory sslSocketFactory = null;
 
             if (jou.shouldCreateSSLSocketFactory(tokenEndpointUrl))
                 sslSocketFactory = jou.createSSLSocketFactory();
 
-            boolean urlencodeHeader = validateUrlencodeHeader(cu);
-
-            return new HttpAccessTokenRetriever(clientId,
-                clientSecret,
-                scope,
-                sslSocketFactory,
-                tokenEndpointUrl.toString(),
-                cu.validateLong(SASL_LOGIN_RETRY_BACKOFF_MS),
-                cu.validateLong(SASL_LOGIN_RETRY_BACKOFF_MAX_MS),
-                cu.validateInteger(SASL_LOGIN_CONNECT_TIMEOUT_MS, false),
-                cu.validateInteger(SASL_LOGIN_READ_TIMEOUT_MS, false),
-                urlencodeHeader);
+            String grantType = Optional
+                .ofNullable(jou.validateString(GRANT_TYPE, false))
+                .orElse(ClientCredentialsAccessTokenRetriever.GRANT_TYPE);
+
+            if (grantType.equalsIgnoreCase(ClientCredentialsAccessTokenRetriever.GRANT_TYPE)) {
+                String clientId = jou.validateString(CLIENT_CREDENTIALS_CLIENT_ID);
+                String clientSecret = jou.validateString(CLIENT_CREDENTIALS_CLIENT_SECRET);
+                String scope = jou.validateString(SCOPE, false);
+                boolean urlencodeHeader = validateUrlencodeHeader(cu);
+
+                return new ClientCredentialsAccessTokenRetriever(clientId,
+                    clientSecret,
+                    scope,
+                    sslSocketFactory,
+                    tokenEndpointUrl.toString(),
+                    cu.validateLong(SASL_LOGIN_RETRY_BACKOFF_MS),
+                    cu.validateLong(SASL_LOGIN_RETRY_BACKOFF_MAX_MS),
+                    cu.validateInteger(SASL_LOGIN_CONNECT_TIMEOUT_MS, false),
+                    cu.validateInteger(SASL_LOGIN_READ_TIMEOUT_MS, false),
+                    urlencodeHeader);
+            } else if (grantType.equalsIgnoreCase(JwtBearerAccessTokenRetriever.GRANT_TYPE)) {
+                String privateKeyId = jou.validateString(JWT_BEARER_PRIVATE_KEY_ID);
+                Path privateKeyFileName = jou.validateFile(JWT_BEARER_PRIVATE_KEY_FILE_NAME);
+                String privateKeySigningAlgorithm = jou.validateString(JWT_BEARER_PRIVATE_KEY_ALGORITHM);
+                Map<String, Object> staticClaims = getStaticClaims(jaasConfig);
+
+                Supplier<String> privateKeySupplier = () -> {
+                    String fileName = privateKeyFileName.toFile().getAbsolutePath();
+
+                    try {
+                        return Utils.readFileAsString(fileName);
+                    } catch (IOException e) {
+                        throw new KafkaException("Could not read the private key from the file " + fileName, e);
+                    }
+                };
+
+                AssertionCreator assertionCreator = new DefaultAssertionCreator(
+                    time,
+                    privateKeySupplier,
+                    privateKeyId,
+                    privateKeySigningAlgorithm
+                );
+
+                return new JwtBearerAccessTokenRetriever(
+                    assertionCreator,
+                    staticClaims,
+                    sslSocketFactory,
+                    tokenEndpointUrl.toString(),
+                    cu.validateLong(SASL_LOGIN_RETRY_BACKOFF_MS),
+                    cu.validateLong(SASL_LOGIN_RETRY_BACKOFF_MAX_MS),
+                    cu.validateInteger(SASL_LOGIN_CONNECT_TIMEOUT_MS, false),
+                    cu.validateInteger(SASL_LOGIN_READ_TIMEOUT_MS, false)
+                );
+            } else {
+                throw new KafkaException("Unsupported grant type provided: " + grantType);
+            }
         }
     }
 
@@ -107,4 +162,20 @@ static boolean validateUrlencodeHeader(ConfigurationUtils configurationUtils) {
             return DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE;
     }
 
+    /**
+     * Support for static claims allows the client to pass arbitrary claims to the identity provider.
+     */
+    static Map<String, Object> getStaticClaims(Map<String, Object> jaasConfig) {
+        Map<String, Object> claims = new HashMap<>();
+
+        jaasConfig.forEach((k, v) -> {
+            if (k.startsWith(JWT_BEARER_CLAIM_PREFIX)) {
+                String claimName = k.substring(JWT_BEARER_CLAIM_PREFIX.length());
+                String claimValue = String.valueOf(v);
+                claims.put(claimName, claimValue);
+            }
+        });
+
+        return claims;
+    }
 }