Add 'client_credentials' OAuth2 authentication flow

docs/HTTP-batchsource.md

@@ -378,6 +378,8 @@ is stopped.
 
 **OAuth2 Enabled:** If true, plugin will perform OAuth2 authentication.
 
+**Grant Type:** The OAuth2 authentication flow that will be used.
+
 **Auth URL:** Endpoint for the authorization server used to retrieve the authorization code.
 
 **Token URL:** Endpoint for the resource server, which exchanges the authorization code for an access token.

docs/HTTP-streamingsource.md

@@ -385,6 +385,8 @@ is stopped.
 
 **OAuth2 Enabled:** If true, plugin will perform OAuth2 authentication.
 
+**Grant Type:** The OAuth2 authentication flow that will be used.
+
 **Auth URL:** Endpoint for the authorization server used to retrieve the authorization code.
 
 **Token URL:** Endpoint for the resource server, which exchanges the authorization code for an access token.

src/main/java/io/cdap/plugin/http/source/common/BaseHttpSourceConfig.java

@@ -27,6 +27,7 @@
 import io.cdap.plugin.http.source.common.error.HttpErrorHandlerEntity;
 import io.cdap.plugin.http.source.common.error.RetryableErrorHandling;
 import io.cdap.plugin.http.source.common.http.KeyStoreType;
+import io.cdap.plugin.http.source.common.http.OAuthGrantType;
 import io.cdap.plugin.http.source.common.pagination.PaginationIteratorFactory;
 import io.cdap.plugin.http.source.common.pagination.PaginationType;
 import io.cdap.plugin.http.source.common.pagination.page.PageFormat;
@@ -81,6 +82,7 @@ public abstract class BaseHttpSourceConfig extends ReferencePluginConfig {
   public static final String PROPERTY_CUSTOM_PAGINATION_CODE = "customPaginationCode";
   public static final String PROPERTY_WAIT_TIME_BETWEEN_PAGES = "waitTimeBetweenPages";
   public static final String PROPERTY_OAUTH2_ENABLED = "oauth2Enabled";
+  public static final String PROPERTY_OAUTH2_GRANT_TYPE = "oauth2GrantType";
   public static final String PROPERTY_AUTH_URL = "authUrl";
   public static final String PROPERTY_TOKEN_URL = "tokenUrl";
   public static final String PROPERTY_CLIENT_ID = "clientId";
@@ -280,6 +282,12 @@ public abstract class BaseHttpSourceConfig extends ReferencePluginConfig {
   @Description("If true, plugin will perform OAuth2 authentication.")
   protected String oauth2Enabled;
 
+  @Nullable
+  @Name(PROPERTY_OAUTH2_GRANT_TYPE)
+  @Description("Which Oauth2 credential flow is used.")
+  @Macro
+  protected String oauth2GrantType;
+
   @Nullable
   @Name(PROPERTY_AUTH_URL)
   @Description("Endpoint for the authorization server used to retrieve the authorization code.")
@@ -533,6 +541,11 @@ public Boolean getOauth2Enabled() {
     return Boolean.parseBoolean(oauth2Enabled);
   }
 
+  @Nullable
+  public OAuthGrantType getOauth2GrantType() {
+    return getEnumValueByString(OAuthGrantType.class, oauth2GrantType, PROPERTY_OAUTH2_GRANT_TYPE);
+  }
+
   @Nullable
   public String getAuthUrl() {
     return authUrl;
@@ -787,11 +800,23 @@ PAGINATION_INDEX_PLACEHOLDER, getPaginationType()),
     // Validate OAuth2 properties
     if (!containsMacro(PROPERTY_OAUTH2_ENABLED) && this.getOauth2Enabled()) {
       String reasonOauth2 = "OAuth2 is enabled";
-      assertIsSet(getAuthUrl(), PROPERTY_AUTH_URL, reasonOauth2);
-      assertIsSet(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2);
-      assertIsSet(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2);
-      assertIsSet(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2);
-      assertIsSet(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2);
+      String reasonOauth2GrantType = reasonOauth2 + " and grant type is " + getOauth2GrantType().getValue();
+
+      switch (getOauth2GrantType()) {
+        case CLIENT_CREDENTIALS:
+          assertIsSet(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2GrantType);
+          assertIsSet(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2GrantType);
+          assertIsSet(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2GrantType);
+          break;
+        case REFRESH_TOKEN:
+
+          assertIsSet(getAuthUrl(), PROPERTY_AUTH_URL, reasonOauth2GrantType);
+          assertIsSet(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2GrantType);
+          assertIsSet(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2GrantType);
+          assertIsSet(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2GrantType);
+          assertIsSet(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2GrantType);
+          break;
+      }
     }
 
     if (!containsMacro(PROPERTY_VERIFY_HTTPS) && !getVerifyHttps()) {

src/main/java/io/cdap/plugin/http/source/common/http/HttpClient.java

@@ -128,9 +128,7 @@ private CloseableHttpClient createHttpClient() throws IOException {
 
     // oAuth2
     if (config.getOauth2Enabled()) {
-      String accessToken = OAuthUtil.getAccessTokenByRefreshToken(HttpClients.createDefault(), config.getTokenUrl(),
-                                                                  config.getClientId(), config.getClientSecret(),
-                                                                  config.getRefreshToken());
+      String accessToken = OAuthUtil.getAccessToken(HttpClients.createDefault(), config);
       clientHeaders.add(new BasicHeader("Authorization", "Bearer " + accessToken));
     }
 

src/main/java/io/cdap/plugin/http/source/common/http/OAuthGrantType.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright © 2019 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.http.source.common.http;
+
+import io.cdap.plugin.http.source.common.EnumWithValue;
+
+/**
+ * Enum encoding the handled Oauth2 Grant Types
+ */
+public enum OAuthGrantType implements EnumWithValue {
+  REFRESH_TOKEN("refresh_token"),
+  CLIENT_CREDENTIALS("client_credentials");
+
+  private final String value;
+
+  OAuthGrantType(String value) {
+    this.value = value;
+  }
+
+  @Override
+  public String getValue() {
+    return value;
+  }
+
+  @Override
+  public String toString() {
+    return this.getValue();
+  }
+}

src/main/java/io/cdap/plugin/http/source/common/http/OAuthUtil.java

@@ -16,21 +16,47 @@
 package io.cdap.plugin.http.source.common.http;
 
 import com.google.gson.JsonElement;
+import io.cdap.plugin.http.source.common.BaseHttpSourceConfig;
 import io.cdap.plugin.http.source.common.pagination.page.JSONUtil;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.message.BasicHeader;
+import org.apache.http.message.BasicNameValuePair;
 import org.apache.http.util.EntityUtils;
 
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+
+import static io.cdap.plugin.http.source.common.http.OAuthGrantType.CLIENT_CREDENTIALS;
+import static io.cdap.plugin.http.source.common.http.OAuthGrantType.REFRESH_TOKEN;
 
 /**
  * A class which contains utilities to make OAuth2 specific calls.
  */
 public class OAuthUtil {
+
+  public static String getAccessToken(CloseableHttpClient httpclient, BaseHttpSourceConfig config) throws IOException {
+    switch (config.getOauth2GrantType()) {
+      case REFRESH_TOKEN:
+        return getAccessTokenByRefreshToken(httpclient, config.getTokenUrl(),
+                config.getClientId(), config.getClientSecret(),
+                config.getRefreshToken());
+      case CLIENT_CREDENTIALS:
+        return getAccessTokenByClientCredentials(httpclient, config.getTokenUrl(),
+                config.getClientId(), config.getClientSecret(), config.getScopes());
+      default:
+        throw new IOException("Invalid Grant Type. Cannot retrieve access token.");
+    }
+  }
+
+
   public static String getAccessTokenByRefreshToken(CloseableHttpClient httpclient, String tokenUrl, String clientId,
                                                     String clientSecret, String refreshToken)
     throws IOException {
@@ -41,16 +67,48 @@ public static String getAccessTokenByRefreshToken(CloseableHttpClient httpclient
         .setParameter("client_id", clientId)
         .setParameter("client_secret", clientSecret)
         .setParameter("refresh_token", refreshToken)
-        .setParameter("grant_type", "refresh_token")
+        .setParameter("grant_type", OAuthGrantType.REFRESH_TOKEN.getValue())
         .build();
     } catch (URISyntaxException e) {
-      throw new IllegalArgumentException("Failed to build token URI for OAuth2", e);
+      throw new IllegalArgumentException("Failed to build access token URI for OAuth2 with grant type = " +
+              OAuthGrantType.REFRESH_TOKEN.getValue(), e);
     }
 
     HttpPost httppost = new HttpPost(uri);
     CloseableHttpResponse response = httpclient.execute(httppost);
     String responseString = EntityUtils.toString(response.getEntity(), "UTF-8");
 
+    JsonElement jsonElement = JSONUtil.toJsonObject(responseString).get("access_token");
+    return jsonElement.getAsString();
+  }
+
+  private static String getAccessTokenByClientCredentials(CloseableHttpClient httpclient, String tokenUrl,
+                                                          String clientId, String clientSecret, String scope)
+          throws IOException {
+    URI uri;
+    try {
+      uri = new URIBuilder(tokenUrl)
+              .build();
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException("Failed to build access token URI for OAuth2 with grant type = " +
+              OAuthGrantType.CLIENT_CREDENTIALS.getValue(), e);
+    }
+
+    HttpPost httppost = new HttpPost(uri);
+    List<BasicNameValuePair> nameValuePairs = new ArrayList<>();
+    nameValuePairs.add(new BasicNameValuePair("scope", scope));
+    nameValuePairs.add(new BasicNameValuePair("grant_type", OAuthGrantType.CLIENT_CREDENTIALS.getValue()));
+
+    httppost.setEntity(new UrlEncodedFormEntity(nameValuePairs));
+
+    String authorizationKey = "Basic " + Base64.getEncoder().encodeToString((clientId + ":" + clientSecret).getBytes());
+
+    httppost.addHeader(new BasicHeader("Authorization", authorizationKey));
+
+    CloseableHttpResponse response = httpclient.execute(httppost);
+    String responseString = EntityUtils.toString(response.getEntity(), "UTF-8");
+
+
     JsonElement jsonElement = JSONUtil.toJsonObject(responseString).get("access_token");
     return jsonElement.getAsString();
   }

widgets/HTTP-batchsource.json

@@ -121,6 +121,18 @@
             }
           }
         },
+        {
+          "widget-type": "select",
+          "label": "Grant Type",
+          "name": "oauth2GrantType",
+          "widget-attributes": {
+            "values": [
+              "refresh_token",
+              "client_credentials"
+            ],
+            "default": "refresh_token"
+          }
+        },
         {
           "widget-type": "textbox",
           "label": "Auth URL",
@@ -503,7 +515,7 @@
       "name": "Proxy authentication",
       "condition": {
         "property": "proxyUrl",
-        "operator": "exists",
+        "operator": "exists"
       },
       "show": [
         {
@@ -625,7 +637,7 @@
       },
       "show": [
         {
-          "name": "authUrl",
+          "name": "oauth2GrantType",
           "type": "property"
         },
         {
@@ -643,6 +655,18 @@
         {
           "name": "scopes",
           "type": "property"
+        }
+      ]
+    },
+    {
+      "name": "OAuth 2 enabled and Grant Type = refresh_token",
+      "condition": {
+        "expression": "oauth2Enabled == true && oauth2GrantType == 'refresh_token'"
+      },
+      "show": [
+        {
+          "name": "authUrl",
+          "type": "property"
         },
         {
           "name": "refreshToken",

widgets/HTTP-streamingsource.json

@@ -126,6 +126,18 @@
             }
           }
         },
+        {
+          "widget-type": "select",
+          "label": "Grant Type",
+          "name": "oauth2GrantType",
+          "widget-attributes": {
+            "values": [
+              "refresh_token",
+              "client_credentials"
+            ],
+            "default": "refresh_token"
+          }
+        },
         {
           "widget-type": "textbox",
           "label": "Auth URL",
@@ -625,7 +637,7 @@
       },
       "show": [
         {
-          "name": "authUrl",
+          "name": "oauth2GrantType",
           "type": "property"
         },
         {
@@ -643,6 +655,18 @@
         {
           "name": "scopes",
           "type": "property"
+        }
+      ]
+    },
+    {
+      "name": "OAuth 2 enabled and Grant Type = refresh_token",
+      "condition": {
+        "expression": "oauth2Enabled == true && oauth2GrantType == 'refresh_token'"
+      },
+      "show": [
+        {
+          "name": "authUrl",
+          "type": "property"
         },
         {
           "name": "refreshToken",