Make the client_credentials grant-type-string configurable.

README.md

@@ -945,6 +945,10 @@ together with one of authentication options below.
 
 When client starts to establish the connection with the Kafka Broker it will first obtain an access token from the configured Token Endpoint, authenticating with the configured client ID and configured authentication option using client_credentials grant type.
 
+If the OAuth2 server is using an alternative to the "grant_type=client_credentials" string, such as "grant_type=kubernetes", that is achieved by specifying the following:
+- `oauth.client.credentials.grant.type` (e.g.: "kubernetes")
+
+
 ##### Option 1: Using a Client Secret 
 
 Specify the client secret.

oauth-client/src/main/java/io/strimzi/kafka/oauth/client/JaasClientOauthLoginCallbackHandler.java

@@ -73,6 +73,7 @@ public class JaasClientOauthLoginCallbackHandler implements AuthenticateCallback
     private String scope;
     private String audience;
     private URI tokenEndpoint;
+    private String grantType;
 
     private boolean isJwt;
     private int maxTokenExpirySeconds;
@@ -152,6 +153,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
         scope = config.getValue(Config.OAUTH_SCOPE);
         audience = config.getValue(Config.OAUTH_AUDIENCE);
+        grantType = config.getValue(Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
         socketFactory = ConfigUtil.createSSLFactory(config);
         hostnameVerifier = ConfigUtil.createHostnameVerifier(config);
         connectTimeout = getConnectTimeout(config);
@@ -208,6 +210,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
                     + "\n    password: " + mask(password)
                     + "\n    scope: " + scope
                     + "\n    audience: " + audience
+                    + "\n    grantType: " + grantType
                     + "\n    isJwt: " + isJwt
                     + "\n    maxTokenExpirySeconds: " + maxTokenExpirySeconds
                     + "\n    principalExtractor: " + principalExtractor
@@ -397,9 +400,9 @@ private void handleCallback(OAuthBearerTokenCallback callback) throws IOExceptio
             } else if (username != null) {
                 result = loginWithPassword(tokenEndpoint, socketFactory, hostnameVerifier, username, password, clientId, clientSecret, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader);
             } else if (clientSecret != null) {
-                result = loginWithClientSecret(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientSecret, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader);
+                result = loginWithClientSecret(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientSecret, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader, grantType);
             } else if (clientAssertionProvider != null) {
-                result = loginWithClientAssertion(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientAssertionProvider.token(), clientAssertionType, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader);
+                result = loginWithClientAssertion(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientAssertionProvider.token(), clientAssertionType, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader, grantType);
             } else {
                 throw new IllegalStateException("Invalid oauth client configuration - no credentials");
             }

oauth-common/src/main/java/io/strimzi/kafka/oauth/common/Config.java

@@ -23,7 +23,14 @@ public class Config {
     /** The name of 'oauth.client.secret' config option  */
     public static final String OAUTH_CLIENT_SECRET = "oauth.client.secret";
 
-    /** The name of 'oauth.scope' config option  */
+    /** The name of 'oauth.client.credentials.grant.type.string' config option  */
+    public static final String OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE = "oauth.client.credentials.grant.type";
+
+    /** The fallback for 'oauth.client.credentials.grant.type.string' config option  */
+    public static final String OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK = "client_credentials";
+
+
+  /** The name of 'oauth.scope' config option  */
     public static final String OAUTH_SCOPE = "oauth.scope";
 
     /** The name of 'oauth.audience' config option  */

oauth-common/src/main/java/io/strimzi/kafka/oauth/common/OAuthAuthenticator.java

@@ -73,17 +73,19 @@ public static TokenInfo loginWithAccessToken(String token, boolean isJwt, Princi
      * @param principalExtractor A PrincipalExtractor to use to determine the principal (user id)
      * @param scope A scope to request when authenticating
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
      */
     public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFactory socketFactory,
                                                   HostnameVerifier hostnameVerifier,
                                                   String clientId, String clientSecret, boolean isJwt,
-                                                  PrincipalExtractor principalExtractor, String scope, boolean includeAcceptHeader) throws IOException {
+                                                  PrincipalExtractor principalExtractor, String scope, boolean includeAcceptHeader,
+                                                  String grantType) throws IOException {
 
         return loginWithClientSecret(tokenEndpointUrl, socketFactory, hostnameVerifier,
-                clientId, clientSecret, isJwt, principalExtractor, scope, null, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader);
+                clientId, clientSecret, isJwt, principalExtractor, scope, null, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader, grantType);
     }
 
     /**
@@ -100,17 +102,19 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
      * @param scope A scope to request when authenticating
      * @param audience An 'audience' attribute to set on the request when authenticating
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
      */
     public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFactory socketFactory,
                                                   HostnameVerifier hostnameVerifier,
                                                   String clientId, String clientSecret, boolean isJwt,
-                                                  PrincipalExtractor principalExtractor, String scope, String audience, boolean includeAcceptHeader) throws IOException {
+                                                  PrincipalExtractor principalExtractor, String scope, String audience, boolean includeAcceptHeader,
+                                                  String grantType) throws IOException {
 
         return loginWithClientSecret(tokenEndpointUrl, socketFactory, hostnameVerifier,
-                clientId, clientSecret, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader);
+                clientId, clientSecret, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader, grantType);
     }
 
     /**
@@ -132,6 +136,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
      * @param retries A maximum number of retries if the request fails due to network, or unexpected response status
      * @param retryPauseMillis A pause between consecutive requests
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -141,7 +146,8 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
                                                   HostnameVerifier hostnameVerifier,
                                                   String clientId, String clientSecret, boolean isJwt,
                                                   PrincipalExtractor principalExtractor, String scope, String audience,
-                                                  int connectTimeout, int readTimeout, MetricsHandler metrics, int retries, long retryPauseMillis, boolean includeAcceptHeader) throws IOException {
+                                                  int connectTimeout, int readTimeout, MetricsHandler metrics, int retries, long retryPauseMillis, boolean includeAcceptHeader,
+                                                  String grantType) throws IOException {
         if (log.isDebugEnabled()) {
             log.debug("loginWithClientSecret() - tokenEndpointUrl: {}, clientId: {}, clientSecret: {}, scope: {}, audience: {}, connectTimeout: {}, readTimeout: {}, retries: {}, retryPauseMillis: {}",
                     tokenEndpointUrl, clientId, mask(clientSecret), scope, audience, connectTimeout, readTimeout, retries, retryPauseMillis);
@@ -156,7 +162,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
 
         String authorization = "Basic " + base64encode(clientId + ':' + clientSecret);
 
-        StringBuilder body = new StringBuilder("grant_type=client_credentials");
+        StringBuilder body = new StringBuilder("grant_type=" + grantType);
         if (scope != null) {
             body.append("&scope=").append(urlencode(scope));
         }
@@ -182,6 +188,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
      * @param principalExtractor A PrincipalExtractor to use to determine the principal (user id)
      * @param scope A scope to request when authenticating
      * @param audience An 'audience' attribute to set on the request when authenticating
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -196,10 +203,11 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
                                                      boolean isJwt,
                                                      PrincipalExtractor principalExtractor,
                                                      String scope,
-                                                     String audience) throws IOException {
+                                                     String audience,
+                                                     String grantType) throws IOException {
 
         return loginWithClientAssertion(tokenEndpointUrl, socketFactory, hostnameVerifier,
-                clientId, clientAssertion, clientAssertionType, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, true);
+                clientId, clientAssertion, clientAssertionType, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, true, grantType);
     }
 
     /**
@@ -222,6 +230,7 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
      * @param retries A maximum number of retries if the request fails due to network, or unexpected response status
      * @param retryPauseMillis A pause between consecutive requests
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -242,7 +251,8 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
                                                      MetricsHandler metrics,
                                                      int retries,
                                                      long retryPauseMillis,
-                                                     boolean includeAcceptHeader) throws IOException {
+                                                     boolean includeAcceptHeader,
+                                                     String grantType) throws IOException {
         if (log.isDebugEnabled()) {
             log.debug("loginWithClientAssertion() - tokenEndpointUrl: {}, clientId: {}, clientAssertion: {}, clientAssertionType: {}, scope: {}, audience: {}, connectTimeout: {}, readTimeout: {}, retries: {}, retryPauseMillis: {}",
                     tokenEndpointUrl, clientId, mask(clientAssertion), clientAssertionType, scope, audience, connectTimeout, readTimeout, retries, retryPauseMillis);
@@ -252,7 +262,7 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
             throw new IllegalArgumentException("No clientId specified");
         }
 
-        StringBuilder body = new StringBuilder("grant_type=client_credentials")
+        StringBuilder body = new StringBuilder("grant_type=" + grantType)
                 .append("&client_id=").append(urlencode(clientId))
                 .append("&client_assertion=").append(urlencode(clientAssertion))
                 .append("&client_assertion_type=").append(urlencode(clientAssertionType));

oauth-server-plain/src/main/java/io/strimzi/kafka/oauth/server/plain/JaasServerOauthOverPlainValidatorCallbackHandler.java

@@ -120,6 +120,7 @@ public class JaasServerOauthOverPlainValidatorCallbackHandler extends JaasServer
     private URI tokenEndpointUri;
     private String scope;
     private String audience;
+    private String grantType;
 
     private OAuthMetrics metrics;
     private boolean enableMetrics;
@@ -148,6 +149,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
         scope = config.getValue(ServerConfig.OAUTH_SCOPE);
         audience = config.getValue(ServerConfig.OAUTH_AUDIENCE);
+        grantType = config.getValue(Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         super.delegatedConfigure(configs, "PLAIN", jaasConfigEntries);
 
@@ -247,7 +249,7 @@ private void authenticate(String username, String password) throws UnsupportedCa
                 checkUsernameMatch = true;
             } else if (tokenEndpointUri != null) {
                 accessToken = OAuthAuthenticator.loginWithClientSecret(tokenEndpointUri, getSocketFactory(), getVerifier(),
-                        username, password, isJwt(), getPrincipalExtractor(), scope, audience, getConnectTimeout(), getReadTimeout(), authMetrics, getRetries(), getRetryPauseMillis(), includeAcceptHeader())
+                        username, password, isJwt(), getPrincipalExtractor(), scope, audience, getConnectTimeout(), getReadTimeout(), authMetrics, getRetries(), getRetryPauseMillis(), includeAcceptHeader(), grantType)
                         .token();
             } else {
                 throw new ValidationException("Empty password where access token was expected");

testsuite/hydra-test/src/test/java/io/strimzi/testsuite/oauth/HydraAuthenticationTest.java

@@ -42,6 +42,7 @@
 import java.util.Map;
 import java.util.Properties;
 
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.common.TestUtil.logStart;
 
 /**
@@ -135,7 +136,7 @@ public void opaqueAccessTokenWithIntrospectValidationTest(String title) throws E
         // first, request access token using client id and secret
         TokenInfo info = OAuthAuthenticator.loginWithClientSecret(URI.create(tokenEndpointUri),
                 ConfigUtil.createSSLFactory(new ClientConfig()),
-                null, clientId, clientSecret, true, null, null, true);
+                null, clientId, clientSecret, true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         oauthConfig.put(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, tokenEndpointUri);

testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java

@@ -28,6 +28,7 @@
 import java.util.Properties;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.loginWithClientSecret;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.kafka.oauth.common.TokenIntrospection.introspectAccessToken;
 import static io.strimzi.testsuite.oauth.auth.Common.buildConsumerConfigOAuthBearer;
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigOAuthBearer;
@@ -253,7 +254,8 @@ void accessTokenWithIntrospection() throws Exception {
         final String clientSecret = "kafka-producer-client-secret";
 
         // First, request access token using client id and secret
-        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true);
+        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true,
+                                               OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         oauthConfig.put(ClientConfig.OAUTH_ACCESS_TOKEN, info.token());

testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/JwtManipulationTests.java

@@ -40,6 +40,7 @@
 
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigOAuthBearer;
 import static io.strimzi.testsuite.oauth.auth.Common.loginWithUsernamePassword;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 
 public class JwtManipulationTests {
 
@@ -224,7 +225,8 @@ private String getOriginalToken() throws IOException {
 
         // first, request access token using client id and secret
         TokenInfo info = OAuthAuthenticator.loginWithClientSecret(URI.create(tokenEndpointUri), null, null,
-                "kafka-producer-client", "kafka-producer-client-secret", true, null, null, true);
+                "kafka-producer-client", "kafka-producer-client-secret", true, null, null, true,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         return info.token();
     }