Added GrantType and Client Authentication for the HttpOauth2

Author: Amit-CloudSufi

src/main/java/io/cdap/plugin/http/common/BaseHttpConfig.java

@@ -27,6 +27,7 @@
 import io.cdap.plugin.http.common.http.AuthType;
 import io.cdap.plugin.http.common.http.OAuthUtil;
 
+import io.cdap.plugin.http.source.common.BaseHttpSourceConfig;
 import java.io.File;
 import java.util.Optional;
 import javax.annotation.Nullable;
@@ -48,6 +49,8 @@ public abstract class BaseHttpConfig extends ReferencePluginConfig {
     public static final String PROPERTY_PROXY_URL = "proxyUrl";
     public static final String PROPERTY_PROXY_USERNAME = "proxyUsername";
     public static final String PROPERTY_PROXY_PASSWORD = "proxyPassword";
+    public static final String PROPERTY_OAUTH2_GRANT_TYPE = "oauth2GrantType";
+    public static final String PROPERTY_OAUTH2_CLIENT_AUTHENTICATION = "oauth2ClientAuthentication";
 
     public static final String PROPERTY_AUTH_TYPE_LABEL = "Auth type";
 
@@ -93,6 +96,18 @@ public abstract class BaseHttpConfig extends ReferencePluginConfig {
     @Macro
     protected String authUrl;
 
+    @Nullable
+    @Name(PROPERTY_OAUTH2_GRANT_TYPE)
+    @Description("Which Oauth2 grant type flow is used.")
+    @Macro
+    protected String oauth2GrantType;
+
+    @Nullable
+    @Name(PROPERTY_OAUTH2_CLIENT_AUTHENTICATION)
+    @Description("Which Oauth2 client authentication flow is used.")
+    @Macro
+    protected String oauth2ClientAuthentication;
+
     @Nullable
     @Name(PROPERTY_TOKEN_URL)
     @Description("Endpoint for the resource server, which exchanges the authorization code for an access token.")
@@ -208,6 +223,19 @@ public String getOAuth2Enabled() {
         return oauth2Enabled;
     }
 
+    public OAuthGrantType getOauth2GrantType() {
+        OAuthGrantType grantType = OAuthGrantType.getGrantType(oauth2GrantType);
+        return BaseHttpSourceConfig.getEnumValueByString(OAuthGrantType.class, grantType.getValue(),
+            PROPERTY_OAUTH2_GRANT_TYPE);
+    }
+
+    public OAuthClientAuthentication getOauth2ClientAuthentication() {
+        OAuthClientAuthentication clientAuthentication = OAuthClientAuthentication.getClientAuthentication(
+            oauth2ClientAuthentication);
+        return BaseHttpSourceConfig.getEnumValueByString(OAuthClientAuthentication.class,
+            clientAuthentication.getValue(), PROPERTY_OAUTH2_CLIENT_AUTHENTICATION);
+    }
+
     @Nullable
     public String getAuthUrl() {
         return authUrl;
@@ -365,21 +393,7 @@ public void validate(FailureCollector failureCollector) {
         AuthType authType = getAuthType();
         switch (authType) {
             case OAUTH2:
-                String reasonOauth2 = "OAuth2 is enabled";
-                if (!containsMacro(PROPERTY_TOKEN_URL)) {
-                    assertIsSetWithFailureCollector(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2, failureCollector);
-                }
-                if (!containsMacro(PROPERTY_CLIENT_ID)) {
-                    assertIsSetWithFailureCollector(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2, failureCollector);
-                }
-                if (!containsMacro((PROPERTY_CLIENT_SECRET))) {
-                    assertIsSetWithFailureCollector(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2,
-                      failureCollector);
-                }
-                if (!containsMacro(PROPERTY_REFRESH_TOKEN)) {
-                    assertIsSetWithFailureCollector(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2,
-                      failureCollector);
-                }
+                validateOAuth2Fields(failureCollector);
                 break;
             case SERVICE_ACCOUNT:
                 String reasonSA = "Service Account is enabled";
@@ -423,4 +437,37 @@ public static void assertIsSetWithFailureCollector(Object propertyValue, String
               null).withConfigProperty(propertyName);
         }
     }
+
+    private void validateOAuth2Fields(FailureCollector failureCollector) {
+        String reasonOauth2GrantType = String.format("OAuth2 is enabled and grant type is %s.",
+            getOauth2GrantType().getValue());
+        if (!containsMacro(PROPERTY_TOKEN_URL)) {
+            assertIsSetWithFailureCollector(getTokenUrl(), PROPERTY_TOKEN_URL,
+                reasonOauth2GrantType, failureCollector);
+        }
+        if (!containsMacro(PROPERTY_CLIENT_ID)) {
+            assertIsSetWithFailureCollector(getClientId(), PROPERTY_CLIENT_ID,
+                reasonOauth2GrantType, failureCollector);
+        }
+        if (!containsMacro(PROPERTY_CLIENT_SECRET)) {
+            assertIsSetWithFailureCollector(getClientSecret(), PROPERTY_CLIENT_SECRET,
+                reasonOauth2GrantType, failureCollector);
+        }
+        if (!containsMacro(PROPERTY_OAUTH2_CLIENT_AUTHENTICATION)) {
+            assertIsSetWithFailureCollector(getOauth2ClientAuthentication(),
+                PROPERTY_OAUTH2_CLIENT_AUTHENTICATION, reasonOauth2GrantType, failureCollector);
+        }
+        // in case of refresh token grant type, also check 2 additional fields
+        if (OAuthGrantType.REFRESH_TOKEN.equals(getOauth2GrantType())) {
+            if (!containsMacro(PROPERTY_AUTH_URL)) {
+                assertIsSetWithFailureCollector(getAuthUrl(), PROPERTY_AUTH_URL,
+                    reasonOauth2GrantType, failureCollector);
+            }
+            if (!containsMacro(PROPERTY_REFRESH_TOKEN)) {
+                assertIsSetWithFailureCollector(getRefreshToken(), PROPERTY_REFRESH_TOKEN,
+                    reasonOauth2GrantType, failureCollector);
+            }
+        }
+        failureCollector.getOrThrowException();
+    }
 }

src/main/java/io/cdap/plugin/http/common/OAuthClientAuthentication.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright © 2025 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.common;
+
+import java.util.Objects;
+
+/**
+ * Enum encoding the handled Oauth2 Client Authentication
+ */
+public enum OAuthClientAuthentication implements EnumWithValue {
+  BODY("body", "Body"),
+  REQUEST_PARAMETER("request_parameter", "Request Parameter");
+
+  private final String value;
+  private final String label;
+
+  OAuthClientAuthentication(String value, String label) {
+    this.value = value;
+    this.label = label;
+  }
+
+  public static OAuthClientAuthentication getClientAuthentication(String clientAuthentication) {
+    if (Objects.equals(clientAuthentication, BODY.getLabel())) {
+      return BODY;
+    } else {
+      return REQUEST_PARAMETER;
+    }
+  }
+
+  @Override
+  public String getValue() {
+    return value;
+  }
+
+  public String getLabel() {
+    return label;
+  }
+
+  @Override
+  public String toString() {
+    return this.getValue();
+  }
+}

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright © 2025 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.common;
+
+import java.util.Objects;
+
+/**
+ * Enum encoding the handled Oauth2 Grant Types
+ */
+public enum OAuthGrantType implements EnumWithValue {
+  REFRESH_TOKEN("refresh_token", "Refresh Token"),
+  CLIENT_CREDENTIALS("client_credentials", "Client Credentials");
+
+  private final String value;
+  private final String label;
+
+  OAuthGrantType(String value, String label) {
+    this.value = value;
+    this.label = label;
+  }
+
+  public static OAuthGrantType getGrantType(String oauth2GrantType) {
+    if (oauth2GrantType == null || Objects.equals(oauth2GrantType, REFRESH_TOKEN.getLabel())) {
+      return REFRESH_TOKEN;
+    } else {
+      return CLIENT_CREDENTIALS;
+    }
+  }
+
+  @Override
+  public String getValue() {
+    return value;
+  }
+
+  public String getLabel() {
+    return label;
+  }
+
+  @Override
+  public String toString() {
+    return this.getValue();
+  }
+}

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

@@ -32,6 +32,7 @@
 import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.client.HttpClients;
 import org.apache.http.message.BasicHeader;
 
 import java.io.Closeable;
@@ -132,7 +133,14 @@ public CloseableHttpClient createHttpClient(String pageUriStr) throws IOExceptio
       httpClientBuilder.setProxy(proxyHost);
     }
     httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
-
+    ArrayList<Header> clientHeaders = new ArrayList<>();
+    // oAuth2
+    if (config.getOauth2Enabled()) {
+      AccessToken oauthAccessToken = OAuthUtil.getAccessToken(HttpClients.createDefault(), config);
+      clientHeaders.add(new BasicHeader("Authorization",
+          String.format("Bearer %s", oauthAccessToken.getTokenValue())));
+    }
+    httpClientBuilder.setDefaultHeaders(clientHeaders);
     return httpClientBuilder.build();
   }
 

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

@@ -21,13 +21,17 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.gson.JsonElement;
 import io.cdap.plugin.http.common.BaseHttpConfig;
+import io.cdap.plugin.http.common.OAuthClientAuthentication;
+import io.cdap.plugin.http.common.OAuthGrantType;
 import io.cdap.plugin.http.common.pagination.page.JSONUtil;
 import io.cdap.plugin.http.source.common.BaseHttpSourceConfig;
+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.impl.client.HttpClients;
+import org.apache.http.message.BasicNameValuePair;
 import org.apache.http.util.EntityUtils;
 
 import java.io.ByteArrayInputStream;
@@ -37,8 +41,12 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.List;
+import java.util.Objects;
 import javax.annotation.Nullable;
 
 /**
@@ -70,12 +78,77 @@ public static AccessToken getAccessToken(BaseHttpConfig config) throws IOExcepti
         return OAuthUtil.getAccessTokenByServiceAccount(config);
       case OAUTH2:
         try (CloseableHttpClient client = HttpClients.createDefault()) {
-          return OAuthUtil.getAccessTokenByRefreshToken(client, config);
+          return getAccessToken(client, (BaseHttpSourceConfig) config);
         }
     }
     return null;
   }
 
+  public static AccessToken getAccessToken(CloseableHttpClient httpclient,
+      BaseHttpSourceConfig config) throws IOException {
+    switch (config.getOauth2GrantType()) {
+      case REFRESH_TOKEN:
+        return getAccessTokenByRefreshToken(httpclient, config);
+      case CLIENT_CREDENTIALS:
+        return getAccessTokenByClientCredentials(httpclient, config.getTokenUrl(),
+            config.getClientId(), config.getClientSecret(), config.getScopes(),
+            config.getOauth2ClientAuthentication().getValue());
+      default:
+        throw new IOException("Invalid Grant Type. Cannot retrieve access token.");
+    }
+  }
+
+  private static AccessToken getAccessTokenByClientCredentials(CloseableHttpClient httpclient,
+      String tokenUrl, String clientId, String clientSecret, String scope,
+      String clientAuthentication) throws IOException {
+    URI uri;
+    HttpPost httppost;
+    try {
+      if (Objects.equals(clientAuthentication, OAuthClientAuthentication.BODY.getValue())) {
+        uri = new URIBuilder(tokenUrl).build();
+        List<BasicNameValuePair> nameValuePairs = new ArrayList<>();
+        nameValuePairs.add(new BasicNameValuePair("scope", scope));
+        nameValuePairs.add(
+            new BasicNameValuePair("grant_type", OAuthGrantType.CLIENT_CREDENTIALS.getValue()));
+        nameValuePairs.add(new BasicNameValuePair("client_id", clientId));
+        nameValuePairs.add(new BasicNameValuePair("client_secret", clientSecret));
+        httppost = new HttpPost(uri);
+        httppost.setEntity(new UrlEncodedFormEntity(nameValuePairs));
+      } else {
+        uri = new URIBuilder(tokenUrl).setParameter("client_id", clientId)
+            .setParameter("client_secret", clientSecret)
+            .setParameter("grant_type", OAuthGrantType.CLIENT_CREDENTIALS.getValue()).build();
+        httppost = new HttpPost(uri);
+      }
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException(
+          "Failed to build access token URI for OAuth2 with grant type = "
+              + OAuthGrantType.CLIENT_CREDENTIALS.getValue(), e);
+    }
+
+    CloseableHttpResponse response = httpclient.execute(httppost);
+    String responseString = EntityUtils.toString(response.getEntity(), "UTF-8");
+
+    JsonElement accessTokenElement = JSONUtil.toJsonObject(responseString).get("access_token");
+
+    if (accessTokenElement == null) {
+      throw new IllegalArgumentException("Access token not found");
+    }
+
+    JsonElement expiresInElement = JSONUtil.toJsonObject(responseString).get("expires_in");
+    Date expiresInDate = null;
+    if (expiresInElement != null) {
+      Instant now = Instant.now();
+      Duration expiresIn = Duration.ofSeconds(expiresInElement.getAsInt());
+      Duration buffer = Duration.ofMinutes(1);
+
+      Instant expiresAt = now.plus(expiresIn).minus(buffer);
+      expiresInDate = Date.from(expiresAt);
+    }
+
+    return new AccessToken(accessTokenElement.getAsString(), expiresInDate);
+  }
+
   /**
    * Returns true only if the expiration time set in the accessToken is before the current time.
    * @param accessToken AccessToken instance
@@ -131,9 +204,12 @@ public static AccessToken getAccessTokenByRefreshToken(CloseableHttpClient httpc
     JsonElement expiresInElement = JSONUtil.toJsonObject(responseString).get("expires_in");
     Date expiresInDate = null;
     if (expiresInElement != null) {
-      long expiresAtMilliseconds = System.currentTimeMillis()
-              + (long) (expiresInElement.getAsInt() * 1000) - 60000L;
-      expiresInDate = new Date(expiresAtMilliseconds);
+      Instant now = Instant.now();
+      Duration expiresIn = Duration.ofSeconds(expiresInElement.getAsInt());
+      Duration buffer = Duration.ofMinutes(1);
+
+      Instant expiresAt = now.plus(expiresIn).minus(buffer);
+      expiresInDate = Date.from(expiresAt);
     }
 
     return new AccessToken(accessTokenElement.getAsString(), expiresInDate);