Merge pull request #190 from cloudsufi/cherry-pick/7ca628d3240f67c2b234d44b8f14537afe20752c/r1.4

[🍒][PLUGIN-1872][PLUGIN-1872] Added GrantType and Client Authentication for the HttpOauth2

Author: vikasrathee-cs

docs/HTTP-batchsink.md

@@ -87,6 +87,36 @@ Skip on error - Ignores erroneous records.
 
 **Wait Time Between Request:** Time in milliseconds to wait between HTTP requests. Defaults to 0. (Macro enabled)
 
+### Authentication
+
+* **OAuth2**
+    * **Grant Type:** Which OAuth2 grant type flow is used. It can be Refresh Token or Client Credentials Flow.
+    * **Client Authentication:** Send OAuth2 Credentials in the Request Body or as Query Parameter or as Basic Auth
+      Header.
+    * **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.
+    * **Client ID:** Client identifier obtained during the Application registration process.
+    * **Client Secret:** Client secret obtained during the Application registration process.
+    * **Scopes:** Scope of the access request, which might have multiple space-separated values.
+    * **Refresh Token:** Token used to receive accessToken, which is end product of OAuth2.
+* **Service Account** - service account key used for authorization
+    * **File Path**: Path on the local file system of the service account key used for
+      authorization. Can be set to 'auto-detect' when running on a Dataproc cluster.
+      When running on other clusters, the file must be present on every node in the cluster.
+    * **JSON**: Contents of the service account JSON file.
+    * **Scope**: The additional Google credential scopes required to access entered url, cloud-platform is included by
+      default, visit https://developers.google.com/identity/protocols/oauth2/scopes for more information.
+        * Scope example:
+
+```
+https://www.googleapis.com/auth/bigquery
+https://www.googleapis.com/auth/cloud-platform
+```
+
+* **Basic Authentication**
+    * **Username:** Username for basic authentication.
+    * **Password:** Password for basic authentication.
+
 ### HTTP Proxy
 
 **Proxy URL:** Proxy URL. Must contain a protocol, address and port.

docs/HTTP-batchsource.md

@@ -214,6 +214,9 @@ The newline delimiter cannot be within quotes.
 
 ### Authentication
 * **OAuth2**
+    * **Grant Type:** Which OAuth2 grant type flow is used. It can be Refresh Token or Client Credentials Flow.
+    * **Client Authentication:** Send OAuth2 Credentials in the Request Body or as Query Parameter or as Basic Auth
+      Header.
     * **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.
     * **Client ID:** Client identifier obtained during the Application registration process.

docs/HTTP-streamingsource.md

@@ -209,6 +209,9 @@ can be omitted as long as the field is present in schema.
 
 ### Authentication
 * **OAuth2**
+    * **Grant Type:** Which OAuth2 grant type flow is used. It can be Refresh Token or Client Credentials Flow.
+    * **Client Authentication:** Send OAuth2 Credentials in the Request Body or as Query Parameter or as Basic Auth
+      Header.
     * **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.
     * **Client ID:** Client identifier obtained during the Application registration process.

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

@@ -29,9 +29,9 @@
 
 import java.io.File;
 import java.util.Optional;
+import java.util.stream.Stream;
 import javax.annotation.Nullable;
 
-
 /**
  *  Base configuration for HTTP Source and HTTP Sink
  */
@@ -48,6 +48,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 +95,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("Send auth credentials in the request body or as query param.")
+    @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 +222,19 @@ public String getOAuth2Enabled() {
         return oauth2Enabled;
     }
 
+    public OAuth2GrantType getOauth2GrantType() {
+        OAuth2GrantType grantType = OAuth2GrantType.getGrantType(oauth2GrantType);
+        return getEnumValueByString(OAuth2GrantType.class, grantType.getValue(),
+            PROPERTY_OAUTH2_GRANT_TYPE);
+    }
+
+    public OAuth2ClientAuthentication getOauth2ClientAuthentication() {
+        OAuth2ClientAuthentication clientAuthentication = OAuth2ClientAuthentication.getClientAuthentication(
+            oauth2ClientAuthentication);
+        return getEnumValueByString(OAuth2ClientAuthentication.class,
+            clientAuthentication.getValue(), PROPERTY_OAUTH2_CLIENT_AUTHENTICATION);
+    }
+
     @Nullable
     public String getAuthUrl() {
         return authUrl;
@@ -359,19 +386,7 @@ public void validate(FailureCollector failureCollector) {
         AuthType authType = getAuthType();
         switch (authType) {
             case OAUTH2:
-                String reasonOauth2 = "OAuth2 is enabled";
-                if (!containsMacro(PROPERTY_TOKEN_URL)) {
-                    assertIsSet(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2);
-                }
-                if (!containsMacro(PROPERTY_CLIENT_ID)) {
-                    assertIsSet(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2);
-                }
-                if (!containsMacro((PROPERTY_CLIENT_SECRET))) {
-                    assertIsSet(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2);
-                }
-                if (!containsMacro(PROPERTY_REFRESH_TOKEN)) {
-                    assertIsSet(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2);
-                }
+                validateOAuth2Fields(failureCollector);
                 break;
             case SERVICE_ACCOUNT:
                 String reasonSA = "Service Account is enabled";
@@ -405,4 +420,65 @@ public static void assertIsSet(Object propertyValue, String propertyName, String
                     String.format("Property '%s' must be set, since %s", propertyName, reason), propertyName);
         }
     }
+
+    public static void assertIsSetWithFailureCollector(Object propertyValue, String propertyName, String reason,
+                                                       FailureCollector failureCollector) {
+        if (propertyValue == null) {
+            failureCollector.addFailure(String.format("Property '%s' must be set, since %s", propertyName, reason),
+              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 additional fields
+        if (OAuth2GrantType.REFRESH_TOKEN.equals(getOauth2GrantType())) {
+            if (!containsMacro(PROPERTY_REFRESH_TOKEN)) {
+                assertIsSetWithFailureCollector(getRefreshToken(), PROPERTY_REFRESH_TOKEN,
+                    reasonOauth2GrantType, failureCollector);
+            }
+        }
+        failureCollector.getOrThrowException();
+    }
+
+    /**
+     * Retrieves the corresponding enum constant of a given string value.
+     *
+     * <p>This method takes an enum class that implements {@code EnumWithValue} and searches for an
+     * enum constant that matches the provided string value. If no matching value is found, it throws
+     * an {@code InvalidConfigPropertyException}.</p>
+     *
+     * @param <T>          the type of enum that implements {@code EnumWithValue}
+     * @param enumClass    the class of the enum to search within
+     * @param stringValue  the string representation of the enum value
+     * @param propertyName the name of the property (used for error messages)
+     * @return the corresponding enum constant if a match is found
+     * @throws InvalidConfigPropertyException if the string value does not match any enum constant
+     */
+    public static <T extends EnumWithValue> T
+    getEnumValueByString(Class<T> enumClass, String stringValue, String propertyName) {
+        return Stream.of(enumClass.getEnumConstants())
+            .filter(keyType -> keyType.getValue().equalsIgnoreCase(stringValue))
+            .findAny()
+            .orElseThrow(() -> new InvalidConfigPropertyException(
+                String.format("Unsupported value for '%s': '%s'", propertyName, stringValue),
+                propertyName));
+    }
 }

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

@@ -0,0 +1,73 @@
+/*
+ * 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 OAuth2ClientAuthentication implements EnumWithValue {
+  BODY("body", "Body"),
+  REQUEST_PARAMETER("request_parameter", "Request Parameter"),
+  BASIC_AUTH_HEADER("basic_auth_header", "Basic Auth Header");
+
+  private final String value;
+  private final String label;
+
+  OAuth2ClientAuthentication(String value, String label) {
+    this.value = value;
+    this.label = label;
+  }
+
+  /**
+   * Determines the OAuth2 client authentication method based on the provided input.
+   *
+   * <p>This method checks if the given client authentication type matches the predefined
+   *  authentication type. If it matches, the method returns the same authentication. Otherwise,
+   * it defaults to BASIC_AUTH_HEADER authentication.</p>
+   *
+   * @param clientAuthentication The client authentication type as a {@link String}. It can be
+   *                             either the value or the label of the authentication method.
+   * @return {@link OAuth2ClientAuthentication} The corresponding authentication type.
+   */
+  public static OAuth2ClientAuthentication getClientAuthentication(String clientAuthentication) {
+    if (Objects.equals(clientAuthentication, BODY.getValue()) || Objects.equals(
+        clientAuthentication, BODY.getLabel())) {
+      return BODY;
+    } else if (Objects.equals(clientAuthentication, REQUEST_PARAMETER.getValue()) || Objects.equals(
+      clientAuthentication, REQUEST_PARAMETER.getLabel())) {
+      return REQUEST_PARAMETER;
+    } else {
+      return BASIC_AUTH_HEADER;
+    }
+  }
+
+  @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/OAuth2GrantType.java

@@ -0,0 +1,68 @@
+/*
+ * 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 OAuth2GrantType implements EnumWithValue {
+  REFRESH_TOKEN("refresh_token", "Refresh Token"),
+  CLIENT_CREDENTIALS("client_credentials", "Client Credentials");
+
+  private final String value;
+  private final String label;
+
+  OAuth2GrantType(String value, String label) {
+    this.value = value;
+    this.label = label;
+  }
+
+  /**
+   * Determines the OAuth2 grant type based on the provided string value.
+   *
+   * <p>This method checks whether the given OAuth2 grant type string matches
+   * the CLIENT_CREDENTIALS grant type based on its value or label. If it matches,
+   * CLIENT_CREDENTIALS is returned; otherwise, REFRESH_TOKEN is returned as the default.</p>
+   *
+   * @param oauth2GrantType The OAuth2 grant type as a string.
+   * @return The corresponding {@link OAuth2GrantType}, either CLIENT_CREDENTIALS or REFRESH_TOKEN.
+   */
+  public static OAuth2GrantType getGrantType(String oauth2GrantType) {
+    if (Objects.equals(oauth2GrantType, CLIENT_CREDENTIALS.getValue()) || Objects.equals(
+        oauth2GrantType, CLIENT_CREDENTIALS.getLabel())) {
+      return CLIENT_CREDENTIALS;
+    } else {
+      return REFRESH_TOKEN;
+    }
+  }
+  
+  @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<>();
+    // If OAuth2 is enabled, fetch an access token and add it as an Authorization header.
+    if (Boolean.TRUE.equals(config.getOauth2Enabled())) {
+      AccessToken oauth2AccessToken = OAuthUtil.getAccessToken(httpClientBuilder.build(), config);
+      clientHeaders.add(new BasicHeader("Authorization",
+          String.format("Bearer %s", oauth2AccessToken.getTokenValue())));
+    }
+    httpClientBuilder.setDefaultHeaders(clientHeaders);
     return httpClientBuilder.build();
   }