Custom client credentials grant type (#279)

Signed-off-by: George Jahad <[email protected]>
Signed-off-by: Marko Strukelj <[email protected]>
Co-authored-by: George Jahad <[email protected]>

Author: mstruk

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;
@@ -140,6 +141,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
         clientId = config.getValue(Config.OAUTH_CLIENT_ID);
         clientSecret = config.getValue(Config.OAUTH_CLIENT_SECRET);
+        grantType = config.getValue(Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE);
 
         final String clientAssertion = config.getValue(ClientConfig.OAUTH_CLIENT_ASSERTION);
         final String clientAssertionLocation = config.getValue(ClientConfig.OAUTH_CLIENT_ASSERTION_LOCATION);
@@ -201,6 +203,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
                     + "\n    tokenEndpointUri: " + tokenEndpoint
                     + "\n    clientId: " + clientId
                     + "\n    clientSecret: " + mask(clientSecret)
+                    + "\n    grantType: " + grantType
                     + "\n    clientAssertion: " + mask(clientAssertion)
                     + "\n    clientAssertionLocation: " + clientAssertionLocation
                     + "\n    clientAssertionType: " + clientAssertionType
@@ -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,6 +23,12 @@ 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.client.credentials.grant.type.string' config option  */
+    public static final String OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE = "oauth.client.credentials.grant.type";
+
+    /** The default value for 'oauth.client.credentials.grant.type.string' config option  */
+    public static final String OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE = "client_credentials";
+
     /** The name of 'oauth.scope' config option  */
     public static final String OAUTH_SCOPE = "oauth.scope";
 

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

@@ -18,6 +18,7 @@
 import java.util.Base64;
 import java.util.concurrent.ExecutionException;
 
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE;
 import static io.strimzi.kafka.oauth.common.LogUtil.mask;
 import static io.strimzi.kafka.oauth.common.TokenIntrospection.introspectAccessToken;
 
@@ -83,7 +84,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
                                                   PrincipalExtractor principalExtractor, String scope, boolean includeAcceptHeader) 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, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE);
     }
 
     /**
@@ -110,7 +111,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
                                                   PrincipalExtractor principalExtractor, String scope, String audience, boolean includeAcceptHeader) 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, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE);
     }
 
     /**
@@ -132,6 +133,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,10 +143,11 @@ 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);
+            log.debug("loginWithClientSecret() - tokenEndpointUrl: {}, clientId: {}, clientSecret: {}, grantType:{}. scope: {}, audience: {}, connectTimeout: {}, readTimeout: {}, retries: {}, retryPauseMillis: {}",
+                    tokenEndpointUrl, clientId, mask(clientSecret), grantType, scope, audience, connectTimeout, readTimeout, retries, retryPauseMillis);
         }
 
         if (clientId == null) {
@@ -156,7 +159,11 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
 
         String authorization = "Basic " + base64encode(clientId + ':' + clientSecret);
 
-        StringBuilder body = new StringBuilder("grant_type=client_credentials");
+        if (grantType == null) {
+            grantType = OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE;
+        }
+
+        StringBuilder body = new StringBuilder("grant_type=" + urlencode(grantType));
         if (scope != null) {
             body.append("&scope=").append(urlencode(scope));
         }
@@ -199,7 +206,7 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
                                                      String audience) 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, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE);
     }
 
     /**
@@ -222,6 +229,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,17 +250,21 @@ 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);
+            log.debug("loginWithClientAssertion() - tokenEndpointUrl: {}, clientId: {}, grantType: {}. clientAssertion: {}, clientAssertionType: {}, scope: {}, audience: {}, connectTimeout: {}, readTimeout: {}, retries: {}, retryPauseMillis: {}",
+                    tokenEndpointUrl, clientId, grantType, mask(clientAssertion), clientAssertionType, scope, audience, connectTimeout, readTimeout, retries, retryPauseMillis);
         }
 
         if (clientId == null) {
             throw new IllegalArgumentException("No clientId specified");
         }
+        if (grantType == null) {
+            grantType = OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE;
+        }
 
-        StringBuilder body = new StringBuilder("grant_type=client_credentials")
+        StringBuilder body = new StringBuilder("grant_type=" + urlencode(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

@@ -104,6 +104,8 @@
  * The token endpoint is used to authenticate to authorization server with the <em>clientId</em> and the <em>secret</em> received over username and password parameters.
  * If set, both clientId + secret, and userId + access token are available. Otherwise only userId + access token authentication is available.
  * </li>
+ * <li><em>oauth.client.credentials.grant.type</em> A custom value of `grant_type` parameter passed to token endpoint when authenticating with <em>clientId</em> and the <em>secret</em> to obtain the token.
+ * </li>
  * <li><em>oauth.scope</em> A `scope` parameter passed to token endpoint when authenticating with <em>clientId</em> and the <em>secret</em> to obtain the token.
  * </li>
  * <li><em>oauth.audience</em> An `audience` parameter passed to token endpoint when authenticating with <em>clientId</em> and the <em>secret</em> to obtain the token.
@@ -120,6 +122,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 +151,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_DEFAULT_VALUE);
 
         super.delegatedConfigure(configs, "PLAIN", jaasConfigEntries);
 
@@ -159,6 +163,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
         log.debug("Configured OAuth over PLAIN:"
                 + "\n    configId: " + configId
                 + "\n    tokenEndpointUri: " + tokenEndpointUri
+                + "\n    grantType: " + grantType
                 + "\n    scope: " + scope
                 + "\n    audience: " + audience
                 + "\n    enableMetrics: " + enableMetrics);
@@ -247,7 +252,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/common/src/main/java/io/strimzi/testsuite/oauth/common/TestUtil.java

@@ -116,7 +116,7 @@ public static void log(String format, String... objects) {
         if (objects == null || objects.length == 0) {
             System.out.println(format);
         } else {
-            System.out.printf(format, objects);
+            System.out.printf(format, (Object[]) objects);
         }
     }
 

testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/Common.java

@@ -111,7 +111,6 @@ static String loginWithClientSecret(String tokenEndpoint, String clientId, Strin
                 true,
                 new PrincipalExtractor(),
                 "all",
-                null,
                 true);
 
         return tokenInfo.token();

testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JWKSKeyUseTest.java

@@ -53,7 +53,6 @@ public void doTest() throws Exception {
                 true,
                 null,
                 null,
-                null,
                 true);
 
         TokenIntrospection.debugLogJWT(log, tokenInfo.token());

testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JaasClientConfigTest.java

@@ -78,6 +78,8 @@ public void doTest() throws Exception {
         testRefreshTokenLocation();
 
         testClientAssertionLocation();
+
+        testInvalidGrantType();
     }
 
     private void testAllConfigOptions() throws IOException {
@@ -90,6 +92,7 @@ private void testAllConfigOptions() throws IOException {
         attrs.put(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, "https://sso/token");
         attrs.put(ClientConfig.OAUTH_CLIENT_ID, "client-id");
         attrs.put(ClientConfig.OAUTH_CLIENT_SECRET, "client-secret");
+        attrs.put(ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, "non-default-grant-type");
         attrs.put(ClientConfig.OAUTH_CLIENT_ASSERTION, "client-assertion");
         attrs.put(ClientConfig.OAUTH_CLIENT_ASSERTION_TYPE, "urn:ietf:params:oauth:client-assertion-type:saml2-bearer");
         attrs.put(ClientConfig.OAUTH_PASSWORD_GRANT_USERNAME, "username");
@@ -140,6 +143,7 @@ private void testAllConfigOptions() throws IOException {
             "tokenEndpointUri", "https://sso/token",
             "clientId", "client-id",
             "clientSecret", "c\\*\\*",
+            "grantType", "non-default-grant-type",
             "clientAssertion", "c\\*\\*",
             "clientAssertionType", "urn:ietf:params:oauth:client-assertion-type:saml2-bearer",
             "username", "username",
@@ -569,6 +573,46 @@ private void testClientAssertionLocation() throws Exception {
         }
     }
 
+    private void testInvalidGrantType() throws Exception {
+        String testClient = "testclient";
+        String testSecret = "testsecret";
+
+        changeAuthServerMode("jwks", "mode_200");
+        changeAuthServerMode("token", "mode_200");
+        createOAuthClient(testClient, testSecret);
+
+        Map<String, String> oauthConfig = new HashMap<>();
+        oauthConfig.put(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, TOKEN_ENDPOINT_URI);
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_ID, testClient);
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_SECRET, testSecret);
+        oauthConfig.put(ClientConfig.OAUTH_SSL_TRUSTSTORE_LOCATION, "../docker/target/kafka/certs/ca-truststore.p12");
+        oauthConfig.put(ClientConfig.OAUTH_SSL_TRUSTSTORE_PASSWORD, "changeit");
+
+        // Confirm fails with invalid grant type
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, "dummy-grant-type");
+
+        try {
+            initJaasWithRetry(oauthConfig);
+            Assert.fail("Should have failed");
+
+        } catch (KafkaException e) {
+            assertLoginException(e);
+        }
+
+        // Confirm succeeds with valid grant type
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_DEFAULT_VALUE);
+
+        LogLineReader logReader = new LogLineReader(Common.LOG_PATH);
+        logReader.readNext();
+
+        initJaasWithRetry(oauthConfig);
+        List<String> lines = logReader.readNext();
+        boolean found = checkLogForRegex(lines, "Login succeeded");
+        Assert.assertTrue("Login succeeded", found);
+
+    }
+
+
     /**
      * If signing keys have not yet been loaded by kafka broker,
      * keep trying for up to 10 attempts with 2 second pause.

testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/PasswordAuthAndPrincipalExtractionTest.java

@@ -131,7 +131,6 @@ public void doTest() throws Exception {
                 true,
                 null,
                 null,
-                null,
                 true);
 
         token = tokenInfo.token();