diff --git a/docs/HTTP-batchsink.md b/docs/HTTP-batchsink.md index 7f3081bf..5d5caf61 100644 --- a/docs/HTTP-batchsink.md +++ b/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. diff --git a/docs/HTTP-batchsource.md b/docs/HTTP-batchsource.md index 45b85127..f3eb3a45 100644 --- a/docs/HTTP-batchsource.md +++ b/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. diff --git a/docs/HTTP-streamingsource.md b/docs/HTTP-streamingsource.md index 9c9ce140..c8153d83 100644 --- a/docs/HTTP-streamingsource.md +++ b/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. diff --git a/src/main/java/io/cdap/plugin/http/common/BaseHttpConfig.java b/src/main/java/io/cdap/plugin/http/common/BaseHttpConfig.java index c78b8249..c37f3629 100644 --- a/src/main/java/io/cdap/plugin/http/common/BaseHttpConfig.java +++ b/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. + * + *
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}.
+ * + * @paramThis 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.
+ * + * @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(); + } +} diff --git a/src/main/java/io/cdap/plugin/http/common/OAuth2GrantType.java b/src/main/java/io/cdap/plugin/http/common/OAuth2GrantType.java new file mode 100644 index 00000000..4a36b31d --- /dev/null +++ b/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. + * + *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.
+ * + * @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(); + } +} diff --git a/src/main/java/io/cdap/plugin/http/common/http/HttpClient.java b/src/main/java/io/cdap/plugin/http/common/http/HttpClient.java index 51a6880e..2d624589 100644 --- a/src/main/java/io/cdap/plugin/http/common/http/HttpClient.java +++ b/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); - + ArrayListThis method supports obtaining an access token using either the {@code REFRESH_TOKEN} + * or {@code CLIENT_CREDENTIALS} grant type. If an invalid grant type is provided, an + * {@link IOException} is thrown.
+ * + * @param httpclient the {@link CloseableHttpClient} instance used to execute HTTP requests. + * @param config the {@link BaseHttpConfig} instance containing OAuth 2.0 configuration + * details. + * @return an {@link AccessToken} object containing the retrieved access token. + * @throws IOException if an error occurs during the HTTP request or if the grant type is + * invalid. + */ + public static AccessToken getAccessToken(CloseableHttpClient httpclient, BaseHttpConfig config) + throws IOException { + switch (config.getOauth2GrantType()) { + case REFRESH_TOKEN: + return getAccessTokenByRefreshToken(httpclient, config); + case CLIENT_CREDENTIALS: + return getAccessTokenByClientCredentials(httpclient, config); + default: + throw new IllegalArgumentException( + String.format("Invalid Grant Type: %s. Cannot retrieve access token.", + config.getOauth2GrantType())); + } + } + + /** + * Retrieves an OAuth2 access token using the Client Credentials grant type. + * + *This method constructs an HTTP POST request to fetch an access token from the authorization + * server. The client authentication method (either "BODY" or "REQUEST" or "BASIC_AUTH_HEADER") determines whether + * client credentials are sent in the request body or as query parameters or as basic auth header.
+ * + *Steps:
+ * 1. If client authentication is set to "BODY": - Constructs a URI using the token URL. - Adds
+ * necessary parameters (scope, grant_type, client_id, client_secret) in the request body. -
+ * Creates an HTTP POST request and sets the entity with encoded parameters.
+ * 404. That’s an error.\n";
+
+ Mockito.when(EntityUtils.toString(entity, "UTF-8")).thenReturn(response);
+ PowerMockito.mockStatic(PaginationIteratorFactory.class);
+ BaseHttpPaginationIterator baseHttpPaginationIterator = Mockito.mock(
+ BaseHttpPaginationIterator.class);
+ PowerMockito.when(PaginationIteratorFactory.createInstance(Mockito.any(), Mockito.any()))
+ .thenReturn(baseHttpPaginationIterator);
+ PowerMockito.when(baseHttpPaginationIterator.supportsSkippingPages()).thenReturn(true);
+ PowerMockito.mockStatic(HttpClients.class);
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ Mockito.when(httpClientBuilder.build()).thenReturn(httpClientMock);
+ try {
+ config.validate(collector);
+ } catch (IllegalStateException e) {
+ Assert.assertEquals(1, collector.getValidationFailures().size());
+ }
+ }
+
+ // Client credentials unit test cases for "Request Parameter" authentication
+ @Test
+ public void testValidateOAuth2WithClientCredentialsAndRequestParamAuthentication()
+ throws Exception {
+ FailureCollector collector = new MockFailureCollector();
+ HttpBatchSourceConfig config = HttpBatchSourceConfig.builder().setReferenceName("test")
+ .setUrl("http://localhost").setHttpMethod("GET").setHeaders("Auth:auth").setFormat("JSON")
+ .setErrorHandling(StringUtils.EMPTY).setRetryPolicy(StringUtils.EMPTY)
+ .setMaxRetryDuration(600L).setConnectTimeout(120).setReadTimeout(120)
+ .setPaginationType("NONE").setVerifyHttps("true").setAuthType("oAuth2").setClientId("id")
+ .setClientSecret("secret").setScopes("scope").setTokenUrl("https//:token")
+ .setRetryPolicy("exponential").setOauth2GrantType("client_credentials")
+ .setOauth2ClientAuthentication("request_parameter").build();
+ PowerMockito.mockStatic(PaginationIteratorFactory.class);
+ BaseHttpPaginationIterator baseHttpPaginationIterator = Mockito.mock(
+ BaseHttpPaginationIterator.class);
+ PowerMockito.when(PaginationIteratorFactory.createInstance(Mockito.any(), Mockito.any()))
+ .thenReturn(baseHttpPaginationIterator);
+ PowerMockito.when(baseHttpPaginationIterator.supportsSkippingPages()).thenReturn(true);
+ PowerMockito.mockStatic(HttpClients.class);
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ AccessToken accessToken = Mockito.mock(AccessToken.class);
+ Mockito.when(accessToken.getTokenValue()).thenReturn("1234");
+ PowerMockito.mockStatic(OAuthUtil.class);
+ Mockito.when(OAuthUtil.getAccessTokenByClientCredentials(Mockito.any(), Mockito.any()))
+ .thenReturn(accessToken);
+ config.validate(collector);
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+
+ @Test
+ public void testValidateOAuth2CredentialsWithProxyWithClientCredentialsAndRequestParamAuthentication()
+ throws IOException {
+ FailureCollector collector = new MockFailureCollector();
+ FailureCollector collectorMock = new MockFailureCollector();
+ HttpBatchSourceConfig config = HttpBatchSourceConfig.builder().setReferenceName("test")
+ .setUrl("http://localhost").setHttpMethod("GET").setHeaders("Auth:auth").setFormat("JSON")
+ .setErrorHandling(StringUtils.EMPTY).setRetryPolicy(StringUtils.EMPTY)
+ .setMaxRetryDuration(600L).setConnectTimeout(120).setReadTimeout(120)
+ .setPaginationType("NONE").setVerifyHttps("true").setAuthType("oAuth2").setClientId("id")
+ .setClientSecret("secret").setRefreshToken("token").setScopes("scope")
+ .setTokenUrl("https//:token").setRetryPolicy("exponential").setProxyUrl("https://proxy")
+ .setProxyUsername("proxyuser").setProxyPassword("proxypassword")
+ .setOauth2GrantType("client_credentials").setOauth2ClientAuthentication("request_parameter")
+ .build();
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ CredentialsProvider credentialsProvider = Mockito.mock(CredentialsProvider.class);
+ HttpHost proxy = PowerMockito.mock(HttpHost.class);
+ httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+ httpClientBuilder.setProxy(proxy);
+ PowerMockito.mockStatic(HttpClients.class);
+ CloseableHttpClient closeableHttpClient = Mockito.mock(CloseableHttpClient.class);
+ Mockito.when(HttpClients.createDefault()).thenReturn(closeableHttpClient);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ Mockito.when(
+ HttpClients.custom().setDefaultCredentialsProvider(credentialsProvider).setProxy(proxy)
+ .build()).thenReturn(closeableHttpClient);
+ AccessToken accessToken = Mockito.mock(AccessToken.class);
+ Mockito.when(accessToken.getTokenValue()).thenReturn("1234");
+ PowerMockito.mockStatic(OAuthUtil.class);
+ Mockito.when(OAuthUtil.getAccessTokenByRefreshToken(Mockito.any(), Mockito.any()))
+ .thenReturn(accessToken);
+ config.validate(collectorMock);
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+ @Test
+ public void testValidateCredentialsOAuth2WithInvalidAccessTokenRequestForClientCredAndRequestParamAuthentication()
+ throws Exception {
+ FailureCollector collector = new MockFailureCollector();
+ HttpBatchSourceConfig config = HttpBatchSourceConfig.builder().setReferenceName("test")
+ .setUrl("http://localhost").setHttpMethod("GET").setHeaders("Auth:auth").setFormat("JSON")
+ .setErrorHandling(StringUtils.EMPTY).setRetryPolicy(StringUtils.EMPTY)
+ .setMaxRetryDuration(600L).setConnectTimeout(120).setReadTimeout(120)
+ .setPaginationType("NONE").setVerifyHttps("true").setAuthType("oAuth2").setClientId("id")
+ .setClientSecret("secret").setRefreshToken("token").setScopes("scope")
+ .setTokenUrl("https//:token").setRetryPolicy("exponential")
+ .setOauth2GrantType("client_credentials").setOauth2ClientAuthentication("request_parameter")
+ .build();
+ CloseableHttpClient httpClientMock = Mockito.mock(CloseableHttpClient.class);
+ CloseableHttpResponse httpResponse = Mockito.mock(CloseableHttpResponse.class);
+ Mockito.when(httpClientMock.execute(Mockito.any())).thenReturn(httpResponse);
+ HttpEntity entity = Mockito.mock(HttpEntity.class);
+ Mockito.when(httpResponse.getEntity()).thenReturn(entity);
+ PowerMockito.mockStatic(EntityUtils.class);
+ String response = " 404. That’s an error.\n";
+
+ Mockito.when(EntityUtils.toString(entity, "UTF-8")).thenReturn(response);
+ PowerMockito.mockStatic(PaginationIteratorFactory.class);
+ BaseHttpPaginationIterator baseHttpPaginationIterator = Mockito.mock(
+ BaseHttpPaginationIterator.class);
+ PowerMockito.when(PaginationIteratorFactory.createInstance(Mockito.any(), Mockito.any()))
+ .thenReturn(baseHttpPaginationIterator);
+ PowerMockito.when(baseHttpPaginationIterator.supportsSkippingPages()).thenReturn(true);
+ PowerMockito.mockStatic(HttpClients.class);
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ Mockito.when(httpClientBuilder.build()).thenReturn(httpClientMock);
+ try {
+ config.validate(collector);
+ } catch (IllegalStateException e) {
+ Assert.assertEquals(1, collector.getValidationFailures().size());
+ }
+ }
+
+ // Client credentials unit test cases for "Basic Auth Header" authentication
+ @Test
+ public void testValidateOAuth2WithClientCredentialsAndBasicAuthHeaderAuthentication()
+ throws Exception {
+ FailureCollector collector = new MockFailureCollector();
+ HttpBatchSourceConfig config = HttpBatchSourceConfig.builder().setReferenceName("test")
+ .setUrl("http://localhost").setHttpMethod("GET").setHeaders("Auth:auth").setFormat("JSON")
+ .setErrorHandling(StringUtils.EMPTY).setRetryPolicy(StringUtils.EMPTY)
+ .setMaxRetryDuration(600L).setConnectTimeout(120).setReadTimeout(120)
+ .setPaginationType("NONE").setVerifyHttps("true").setAuthType("oAuth2").setClientId("id")
+ .setClientSecret("secret").setScopes("scope").setTokenUrl("https//:token")
+ .setRetryPolicy("exponential").setOauth2GrantType("client_credentials")
+ .setOauth2ClientAuthentication("basic_auth_header").build();
+ PowerMockito.mockStatic(PaginationIteratorFactory.class);
+ BaseHttpPaginationIterator baseHttpPaginationIterator = Mockito.mock(
+ BaseHttpPaginationIterator.class);
+ PowerMockito.when(PaginationIteratorFactory.createInstance(Mockito.any(), Mockito.any()))
+ .thenReturn(baseHttpPaginationIterator);
+ PowerMockito.when(baseHttpPaginationIterator.supportsSkippingPages()).thenReturn(true);
+ PowerMockito.mockStatic(HttpClients.class);
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ AccessToken accessToken = Mockito.mock(AccessToken.class);
+ Mockito.when(accessToken.getTokenValue()).thenReturn("1234");
+ PowerMockito.mockStatic(OAuthUtil.class);
+ Mockito.when(OAuthUtil.getAccessTokenByClientCredentials(Mockito.any(), Mockito.any()))
+ .thenReturn(accessToken);
+ config.validate(collector);
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+
+ @Test
+ public void testValidateOAuth2CredentialsWithProxyWithClientCredentialsAndBasicAuthHeaderAuthentication()
+ throws IOException {
+ FailureCollector collector = new MockFailureCollector();
+ FailureCollector collectorMock = new MockFailureCollector();
+ HttpBatchSourceConfig config = HttpBatchSourceConfig.builder().setReferenceName("test")
+ .setUrl("http://localhost").setHttpMethod("GET").setHeaders("Auth:auth").setFormat("JSON")
+ .setErrorHandling(StringUtils.EMPTY).setRetryPolicy(StringUtils.EMPTY)
+ .setMaxRetryDuration(600L).setConnectTimeout(120).setReadTimeout(120)
+ .setPaginationType("NONE").setVerifyHttps("true").setAuthType("oAuth2").setClientId("id")
+ .setClientSecret("secret").setRefreshToken("token").setScopes("scope")
+ .setTokenUrl("https//:token").setRetryPolicy("exponential").setProxyUrl("https://proxy")
+ .setProxyUsername("proxyuser").setProxyPassword("proxypassword")
+ .setOauth2GrantType("client_credentials").setOauth2ClientAuthentication("basic_auth_header")
+ .build();
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ CredentialsProvider credentialsProvider = Mockito.mock(CredentialsProvider.class);
+ HttpHost proxy = PowerMockito.mock(HttpHost.class);
+ httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+ httpClientBuilder.setProxy(proxy);
+ PowerMockito.mockStatic(HttpClients.class);
+ CloseableHttpClient closeableHttpClient = Mockito.mock(CloseableHttpClient.class);
+ Mockito.when(HttpClients.createDefault()).thenReturn(closeableHttpClient);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ Mockito.when(
+ HttpClients.custom().setDefaultCredentialsProvider(credentialsProvider).setProxy(proxy)
+ .build()).thenReturn(closeableHttpClient);
+ AccessToken accessToken = Mockito.mock(AccessToken.class);
+ Mockito.when(accessToken.getTokenValue()).thenReturn("1234");
+ PowerMockito.mockStatic(OAuthUtil.class);
+ Mockito.when(OAuthUtil.getAccessTokenByRefreshToken(Mockito.any(), Mockito.any()))
+ .thenReturn(accessToken);
+ config.validate(collectorMock);
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+ @Test
+ public void testValidateCredentialsOAuth2WithInvalidAccessTokenRequestForClientCredAndBasicAuthHeaderAuthentication()
+ throws Exception {
+ FailureCollector collector = new MockFailureCollector();
+ HttpBatchSourceConfig config = HttpBatchSourceConfig.builder().setReferenceName("test")
+ .setUrl("http://localhost").setHttpMethod("GET").setHeaders("Auth:auth").setFormat("JSON")
+ .setErrorHandling(StringUtils.EMPTY).setRetryPolicy(StringUtils.EMPTY)
+ .setMaxRetryDuration(600L).setConnectTimeout(120).setReadTimeout(120)
+ .setPaginationType("NONE").setVerifyHttps("true").setAuthType("oAuth2").setClientId("id")
+ .setClientSecret("secret").setRefreshToken("token").setScopes("scope")
+ .setTokenUrl("https//:token").setRetryPolicy("exponential")
+ .setOauth2GrantType("client_credentials").setOauth2ClientAuthentication("basic_auth_header")
+ .build();
+ CloseableHttpClient httpClientMock = Mockito.mock(CloseableHttpClient.class);
+ CloseableHttpResponse httpResponse = Mockito.mock(CloseableHttpResponse.class);
+ Mockito.when(httpClientMock.execute(Mockito.any())).thenReturn(httpResponse);
+ HttpEntity entity = Mockito.mock(HttpEntity.class);
+ Mockito.when(httpResponse.getEntity()).thenReturn(entity);
+ PowerMockito.mockStatic(EntityUtils.class);
+ String response = " 404. That’s an error.\n";
+
+ Mockito.when(EntityUtils.toString(entity, "UTF-8")).thenReturn(response);
+ PowerMockito.mockStatic(PaginationIteratorFactory.class);
+ BaseHttpPaginationIterator baseHttpPaginationIterator = Mockito.mock(
+ BaseHttpPaginationIterator.class);
+ PowerMockito.when(PaginationIteratorFactory.createInstance(Mockito.any(), Mockito.any()))
+ .thenReturn(baseHttpPaginationIterator);
+ PowerMockito.when(baseHttpPaginationIterator.supportsSkippingPages()).thenReturn(true);
+ PowerMockito.mockStatic(HttpClients.class);
+ HttpClientBuilder httpClientBuilder = Mockito.mock(HttpClientBuilder.class);
+ Mockito.when(HttpClients.custom()).thenReturn(httpClientBuilder);
+ Mockito.when(httpClientBuilder.build()).thenReturn(httpClientMock);
+ config.validate(collector);
+ Assert.assertEquals(1, collector.getValidationFailures().size());
+ }
}
diff --git a/widgets/HTTP-batchsink.json b/widgets/HTTP-batchsink.json
index 4741833a..6fd3fa4e 100644
--- a/widgets/HTTP-batchsink.json
+++ b/widgets/HTTP-batchsink.json
@@ -275,6 +275,31 @@
]
}
},
+ {
+ "widget-type": "select",
+ "label": "Grant Type",
+ "name": "oauth2GrantType",
+ "widget-attributes": {
+ "values": [
+ "Refresh Token",
+ "Client Credentials"
+ ],
+ "default": "Refresh Token"
+ }
+ },
+ {
+ "widget-type": "select",
+ "label": "Client Authentication",
+ "name": "oauth2ClientAuthentication",
+ "widget-attributes": {
+ "values": [
+ "Body",
+ "Request Parameter",
+ "Basic Auth Header"
+ ],
+ "default": "Body"
+ }
+ },
{
"widget-type": "textbox",
"label": "Auth URL",
@@ -466,6 +491,30 @@
}
]
},
+ {
+ "name": "Show Client Authentication when with Grant type is Client Credentials",
+ "condition": {
+ "expression": "authType == 'oAuth2' && oauth2GrantType == 'Client Credentials'"
+ },
+ "show": [
+ {
+ "name": "oauth2ClientAuthentication",
+ "type": "property"
+ }
+ ]
+ },
+ {
+ "name": "Show Refresh_Token when Grant type is 'Refresh Token' or NULL for older version pipeline",
+ "condition": {
+ "expression": "authType == 'oAuth2' && oauth2GrantType != 'Client Credentials'"
+ },
+ "show": [
+ {
+ "name": "refreshToken",
+ "type": "property"
+ }
+ ]
+ },
{
"name": "Authenticate with OAuth2",
"condition": {
@@ -474,6 +523,10 @@
"value": "oAuth2"
},
"show": [
+ {
+ "name": "oauth2GrantType",
+ "type": "property"
+ },
{
"name": "authUrl",
"type": "property"
@@ -493,10 +546,6 @@
{
"name": "scopes",
"type": "property"
- },
- {
- "name": "refreshToken",
- "type": "property"
}
]
},
diff --git a/widgets/HTTP-batchsource.json b/widgets/HTTP-batchsource.json
index 9eed8feb..3b4153cc 100644
--- a/widgets/HTTP-batchsource.json
+++ b/widgets/HTTP-batchsource.json
@@ -165,6 +165,31 @@
]
}
},
+ {
+ "widget-type": "select",
+ "label": "Grant Type",
+ "name": "oauth2GrantType",
+ "widget-attributes": {
+ "values": [
+ "Refresh Token",
+ "Client Credentials"
+ ],
+ "default": "Refresh Token"
+ }
+ },
+ {
+ "widget-type": "select",
+ "label": "Client Authentication",
+ "name": "oauth2ClientAuthentication",
+ "widget-attributes": {
+ "values": [
+ "Body",
+ "Request Parameter",
+ "Basic Auth Header"
+ ],
+ "default": "Body"
+ }
+ },
{
"widget-type": "textbox",
"label": "Auth URL",
@@ -720,6 +745,30 @@
}
]
},
+ {
+ "name": "Show Client Authentication when with Grant type is Client Credentials",
+ "condition": {
+ "expression": "authType == 'oAuth2' && oauth2GrantType == 'Client Credentials'"
+ },
+ "show": [
+ {
+ "name": "oauth2ClientAuthentication",
+ "type": "property"
+ }
+ ]
+ },
+ {
+ "name": "Show Refresh Token when Grant type is 'Refresh Token' or NULL for older version pipeline",
+ "condition": {
+ "expression": "authType == 'oAuth2' && oauth2GrantType != 'Client Credentials'"
+ },
+ "show": [
+ {
+ "name": "refreshToken",
+ "type": "property"
+ }
+ ]
+ },
{
"name": "Authenticate with OAuth2",
"condition": {
@@ -728,6 +777,10 @@
"value": "oAuth2"
},
"show": [
+ {
+ "name": "oauth2GrantType",
+ "type": "property"
+ },
{
"name": "authUrl",
"type": "property"
@@ -747,10 +800,6 @@
{
"name": "scopes",
"type": "property"
- },
- {
- "name": "refreshToken",
- "type": "property"
}
]
},
diff --git a/widgets/HTTP-streamingsource.json b/widgets/HTTP-streamingsource.json
index e47c0b0a..88e6833e 100644
--- a/widgets/HTTP-streamingsource.json
+++ b/widgets/HTTP-streamingsource.json
@@ -133,6 +133,31 @@
]
}
},
+ {
+ "widget-type": "select",
+ "label": "Grant Type",
+ "name": "oauth2GrantType",
+ "widget-attributes": {
+ "values": [
+ "Refresh Token",
+ "Client Credentials"
+ ],
+ "default": "Refresh Token"
+ }
+ },
+ {
+ "widget-type": "select",
+ "label": "Client Authentication",
+ "name": "oauth2ClientAuthentication",
+ "widget-attributes": {
+ "values": [
+ "Body",
+ "Request Parameter",
+ "Basic Auth Header"
+ ],
+ "default": "Body"
+ }
+ },
{
"widget-type": "textbox",
"label": "Auth URL",
@@ -676,6 +701,30 @@
}
]
},
+ {
+ "name": "Show Client Authentication when with Grant type is Client Credentials",
+ "condition": {
+ "expression": "authType == 'oAuth2' && oauth2GrantType == 'Client Credentials'"
+ },
+ "show": [
+ {
+ "name": "oauth2ClientAuthentication",
+ "type": "property"
+ }
+ ]
+ },
+ {
+ "name": "Show Refresh_Token when Grant type is 'Refresh Token' or NULL for older version pipeline",
+ "condition": {
+ "expression": "authType == 'oAuth2' && oauth2GrantType != 'Client Credentials'"
+ },
+ "show": [
+ {
+ "name": "refreshToken",
+ "type": "property"
+ }
+ ]
+ },
{
"name": "Authenticate with OAuth2",
"condition": {
@@ -684,6 +733,10 @@
"value": "oAuth2"
},
"show": [
+ {
+ "name": "oauth2GrantType",
+ "type": "property"
+ },
{
"name": "authUrl",
"type": "property"
@@ -703,10 +756,6 @@
{
"name": "scopes",
"type": "property"
- },
- {
- "name": "refreshToken",
- "type": "property"
}
]
},
+ * 2. If client authentication is set to "REQUEST": - Constructs a URI with client credentials as
+ * query parameters. - Creates an HTTP POST request with the URI.
+ *
+ * 3. If client authentication is set to "BASIC_AUTH_HEADER": - Constructs a URI with client credentials first
+ * concatenated and encoded to Base64 and passed a Basic Authorization Header and
+ * grant type and scope as part of body.
+ *
+ * 4. Calls `fetchAccessToken(httpclient,httppost)` to execute the request and retrieve the
+ * token.
+ *
+ * @param httpclient The HTTP client to execute the request.
+ * @return An AccessToken object containing the token and expiration details.
+ * @throws IOException If an error occurs while executing the request.
+ * @throws IllegalArgumentException If the token URL cannot be built properly.
+ */
+ public static AccessToken getAccessTokenByClientCredentials(CloseableHttpClient httpclient,
+ BaseHttpConfig config) throws IOException {
+ URI uri;
+ HttpPost httppost;
+
+ try {
+ List