Merge pull request data-integrations#250 from data-integrations/clien…

…t-credentials-oauth

Add support for other access token grant types in Authenticator

src/e2e-test/java/io/cdap/plugin/utils/SalesforceClient.java

@@ -178,7 +178,7 @@ public static List<JsonObject> queryObject(String id, String objectName) {
   public static void deletePushTopic(String pushTopicName) {
     try {
       PartnerConnection partnerConnection = new PartnerConnection(
-        Authenticator.createConnectorConfig(new AuthenticatorCredentials(USERNAME, PASSWORD + SECURITYTOKEN,
+        Authenticator.createConnectorConfig(AuthenticatorCredentials.fromParameters(USERNAME, PASSWORD + SECURITYTOKEN,
                                                                          CLIENTID, CLIENTSECRET, PluginPropertyUtils.
                                                                            pluginProp("login.url"),
                                                                          30000, 3600, "")));

src/main/java/io/cdap/plugin/salesforce/SalesforceConnectionUtil.java

@@ -64,11 +64,11 @@ public static AuthenticatorCredentials getAuthenticatorCredentials(Configuration
     }
     String proxyUrl = conf.get(SalesforceConstants.CONFIG_PROXY_URL);
     if (oAuthToken != null && instanceURL != null) {
-      return new AuthenticatorCredentials(new OAuthInfo(oAuthToken, instanceURL), connectTimeout, readTimeout,
-                                          proxyUrl);
+      return AuthenticatorCredentials.fromParameters(
+              new OAuthInfo(oAuthToken, instanceURL), connectTimeout, readTimeout, proxyUrl);
     }
 
-    return new AuthenticatorCredentials(conf.get(SalesforceConstants.CONFIG_USERNAME),
+    return AuthenticatorCredentials.fromParameters(conf.get(SalesforceConstants.CONFIG_USERNAME),
                                         conf.get(SalesforceConstants.CONFIG_PASSWORD),
                                         conf.get(SalesforceConstants.CONFIG_CONSUMER_KEY),
                                         conf.get(SalesforceConstants.CONFIG_CONSUMER_SECRET),

src/main/java/io/cdap/plugin/salesforce/authenticator/Authenticator.java

@@ -25,6 +25,7 @@
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.HttpProxy;
 import org.eclipse.jetty.client.ProxyConfiguration;
+import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 
 import java.net.URI;
@@ -84,6 +85,10 @@ public static OAuthInfo getOAuthInfo(AuthenticatorCredentials credentials) throw
       return oAuthInfo;
     }
 
+    if (credentials.getGrantType() == null) {
+      throw new IllegalArgumentException("Grant type cannot be null for OAuth flow to fetch access token.");
+    }
+
     SslContextFactory sslContextFactory = new SslContextFactory();
     HttpClient httpClient = new HttpClient(sslContextFactory);
     httpClient.setConnectTimeout(credentials.getConnectTimeout());
@@ -92,12 +97,24 @@ public static OAuthInfo getOAuthInfo(AuthenticatorCredentials credentials) throw
     }
     try {
       httpClient.start();
-      String response = httpClient.POST(credentials.getLoginUrl()).param("grant_type", "password")
-        .param("client_id", credentials.getConsumerKey())
-        .param("client_secret", credentials.getConsumerSecret())
-        .param("username", credentials.getUsername())
-        .param("password", credentials.getPassword()).send().getContentAsString();
 
+      Request authRequest = httpClient.POST(credentials.getLoginUrl())
+              .param("grant_type", credentials.getGrantType().getType());
+
+      switch(credentials.getGrantType()) {
+        case CLIENT_CREDENTIALS:
+          authRequest = authRequest.param("client_id", credentials.getConsumerKey())
+                  .param("client_secret", credentials.getConsumerSecret());
+          break;
+        case PASSWORD:
+          authRequest = authRequest.param("client_id", credentials.getConsumerKey())
+                  .param("client_secret", credentials.getConsumerSecret())
+                  .param("username", credentials.getUsername())
+                  .param("password", credentials.getPassword());
+          break;
+      }
+
+      String response = authRequest.send().getContentAsString();
       AuthResponse authResponse = GSON.fromJson(response, AuthResponse.class);
 
       if (!Strings.isNullOrEmpty(authResponse.getError())) {

src/main/java/io/cdap/plugin/salesforce/authenticator/AuthenticatorCredentials.java

@@ -20,6 +20,7 @@
 
 import java.io.Serializable;
 import java.util.Objects;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 /**
@@ -33,28 +34,18 @@ public class AuthenticatorCredentials implements Serializable {
   private final String consumerKey;
   private final String consumerSecret;
   private final String loginUrl;
+  private final GrantType grantType;
   private final Integer connectTimeout;
   private final Integer readTimeout;
   private final String proxyUrl;
 
-  public AuthenticatorCredentials(OAuthInfo oAuthInfo, Integer connectTimeout, Integer readTimeout, String proxyUrl) {
-    this(Objects.requireNonNull(oAuthInfo), null, null, null, null, null, connectTimeout, readTimeout, proxyUrl);
-  }
-
-  public AuthenticatorCredentials(String username, String password,
-                                  String consumerKey, String consumerSecret, String loginUrl,
-                                  Integer connectTimeout, Integer readTimeout, String proxyUrl) {
-    this(null, Objects.requireNonNull(username), Objects.requireNonNull(password), Objects.requireNonNull(consumerKey),
-         Objects.requireNonNull(consumerSecret), Objects.requireNonNull(loginUrl),
-         Objects.requireNonNull(connectTimeout), Objects.requireNonNull(readTimeout), proxyUrl);
-  }
-
-  private AuthenticatorCredentials(@Nullable OAuthInfo oAuthInfo,
+  public AuthenticatorCredentials(@Nullable OAuthInfo oAuthInfo,
                                    @Nullable String username,
                                    @Nullable String password,
                                    @Nullable String consumerKey,
                                    @Nullable String consumerSecret,
                                    @Nullable String loginUrl,
+                                   @Nullable GrantType grantType,
                                    @Nullable Integer connectTimeout,
                                    @Nullable Integer readTimeout,
                                    @Nullable String proxyUrl) {
@@ -64,6 +55,7 @@ private AuthenticatorCredentials(@Nullable OAuthInfo oAuthInfo,
     this.consumerKey = consumerKey;
     this.consumerSecret = consumerSecret;
     this.loginUrl = loginUrl;
+    this.grantType = grantType;
     this.connectTimeout = connectTimeout;
     this.readTimeout = readTimeout;
     this.proxyUrl = proxyUrl;
@@ -99,6 +91,11 @@ public String getLoginUrl() {
     return loginUrl;
   }
 
+  @Nullable
+  public GrantType getGrantType() {
+    return grantType;
+  }
+
   @Nullable
   public Integer getConnectTimeout() {
     return connectTimeout;
@@ -113,6 +110,62 @@ public String getProxyUrl() {
     return proxyUrl;
   }
 
+  /**
+   * Builder for {@link AuthenticatorCredentials} with credentials for username-password OAuth flow, where
+   * Salesforce OAuth {@link GrantType} is set as "password".
+   */
+  public static Builder getBuilder(String username, String password,
+                                   String consumerKey, String consumerSecret, String loginUrl) {
+    return new Builder(
+            Objects.requireNonNull(username), Objects.requireNonNull(password),
+            Objects.requireNonNull(consumerKey), Objects.requireNonNull(consumerSecret),
+            Objects.requireNonNull(loginUrl)
+    );
+  }
+
+  /**
+   * Builder for {@link AuthenticatorCredentials} with credentials for client-credentials OAuth flow,
+   * where Salesforce OAuth {@link GrantType} is set as "client-credentials".
+   */
+  public static Builder getBuilder(String consumerKey, String consumerSecret, String loginUrl) {
+    return new Builder(
+            Objects.requireNonNull(consumerKey), Objects.requireNonNull(consumerSecret),
+            Objects.requireNonNull(loginUrl)
+    );
+  }
+
+  /**
+   * Builder for {@link AuthenticatorCredentials} with already fetched {@link OAuthInfo}.
+   */
+  public static Builder getBuilder(OAuthInfo oAuthInfo) {
+    return new Builder(Objects.requireNonNull(oAuthInfo));
+  }
+
+  /**
+   * Create an instance of {@link AuthenticatorCredentials} from given set of parameters. Uses "password" grant type
+   * if username and password are non-null. Uses client-credentials grant type otherwise.
+   */
+  public static AuthenticatorCredentials fromParameters(String username, String password,
+                                                        String consumerKey, String consumerSecret, String loginUrl,
+                                                        Integer connectTimeout, Integer readTimeout, String proxyUrl) {
+    AuthenticatorCredentials.Builder builder;
+    if (username != null && password != null) {
+      builder = AuthenticatorCredentials.getBuilder(username, password, consumerKey, consumerSecret, loginUrl);
+    } else {
+      builder = AuthenticatorCredentials.getBuilder(consumerKey, consumerSecret, loginUrl);
+    }
+    return builder.setConnectTimeout(connectTimeout).setReadTimeout(readTimeout).setProxyUrl(proxyUrl).build();
+  }
+
+  /**
+   * Create an instance of {@link AuthenticatorCredentials} from given {@link OAuthInfo} and parameters.
+   */
+  public static AuthenticatorCredentials fromParameters(OAuthInfo oAuthInfo,
+                                                        Integer connectTimeout, Integer readTimeout, String proxyUrl) {
+    return AuthenticatorCredentials.getBuilder(oAuthInfo)
+            .setConnectTimeout(connectTimeout).setReadTimeout(readTimeout).setProxyUrl(proxyUrl).build();
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -131,12 +184,99 @@ public boolean equals(Object o) {
       Objects.equals(loginUrl, that.loginUrl) &&
       Objects.equals(connectTimeout, that.connectTimeout) &&
       Objects.equals(readTimeout, that.readTimeout) &&
-      Objects.equals(proxyUrl, that.proxyUrl);
+      Objects.equals(proxyUrl, that.proxyUrl) &&
+      Objects.equals(grantType, that.grantType);
   }
 
   @Override
   public int hashCode() {
     return Objects.hash(username, password, consumerKey, consumerSecret, loginUrl, connectTimeout, readTimeout,
-                        proxyUrl);
+                        proxyUrl, grantType);
+  }
+
+  /**
+   * Builder class for creating an instance of {@link AuthenticatorCredentials}.
+   */
+  public static final class Builder {
+    private OAuthInfo oAuthInfo;
+    private String username;
+    private String password;
+    private String consumerKey;
+    private String consumerSecret;
+    private String loginUrl;
+    private GrantType grantType;
+    private Integer connectTimeout;
+    private Integer readTimeout;
+    private String proxyUrl;
+
+    /**
+     * Create an instance of {@link AuthenticatorCredentials} with credentials for username-password OAuth flow, where
+     * Salesforce OAuth {@link GrantType} is set as "password".
+     */
+    public Builder(@Nonnull String username, @Nonnull String password,
+                   @Nonnull String consumerKey, @Nonnull String consumerSecret, @Nonnull String loginUrl) {
+      this.username = username;
+      this.password = password;
+      this.consumerKey = consumerKey;
+      this.consumerSecret = consumerSecret;
+      this.loginUrl = loginUrl;
+      this.grantType = GrantType.PASSWORD;
+    }
+
+    /**
+     * Create an instance of {@link AuthenticatorCredentials} with credentials for client-credentials OAuth flow,
+     * where Salesforce OAuth {@link GrantType} is set as "client-credentials".
+     */
+    public Builder(@Nonnull String consumerKey, @Nonnull String consumerSecret, @Nonnull String loginUrl) {
+      this.consumerKey = consumerKey;
+      this.consumerSecret = consumerSecret;
+      this.loginUrl = loginUrl;
+      this.grantType = GrantType.CLIENT_CREDENTIALS;
+    }
+
+    /**
+     * Create an instance of {@link AuthenticatorCredentials} with already fetched {@link OAuthInfo}.
+     */
+    public Builder(@Nonnull OAuthInfo oAuthInfo) {
+      this.oAuthInfo = oAuthInfo;
+    }
+
+    public Builder setConnectTimeout(Integer connectTimeout) {
+      this.connectTimeout = connectTimeout;
+      return this;
+    }
+
+    public Builder setReadTimeout(Integer readTimeout) {
+      this.readTimeout = readTimeout;
+      return this;
+    }
+
+    public Builder setProxyUrl(String proxyUrl) {
+      this.proxyUrl = proxyUrl;
+      return this;
+    }
+
+    public AuthenticatorCredentials build() {
+      return new AuthenticatorCredentials(oAuthInfo, username, password, consumerKey, consumerSecret,
+              loginUrl, grantType, connectTimeout, readTimeout, proxyUrl);
+    }
+  }
+
+  /**
+   * Represents the grant type parameter for OAuth
+   */
+  public enum GrantType {
+    PASSWORD("password"),
+    CLIENT_CREDENTIALS("client_credentials");
+
+    private final String type;
+
+    GrantType(String type) {
+      this.type = type;
+    }
+
+    public String getType() {
+      return type;
+    }
   }
 }

src/main/java/io/cdap/plugin/salesforce/plugin/SalesforceConnectorBaseConfig.java

@@ -162,11 +162,10 @@ private void validateConnection(@Nullable OAuthInfo oAuthInfo) {
     if (oAuthInfo == null) {
       return;
     }
-
+    AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(
+            oAuthInfo, this.getConnectTimeout(), this.getReadTimeoutInMillis(), getProxyUrl());
     try {
-      SalesforceConnectionUtil.getPartnerConnection(new AuthenticatorCredentials(oAuthInfo, this.getConnectTimeout(),
-                                                                                 this.getReadTimeoutInMillis(),
-                                                                                 getProxyUrl()));
+      SalesforceConnectionUtil.getPartnerConnection(credentials);
     } catch (ConnectionException e) {
       String message = SalesforceConnectionUtil.getSalesforceErrorMessageFromException(e);
       throw new RuntimeException(

src/main/java/io/cdap/plugin/salesforce/plugin/SalesforceConnectorInfo.java

@@ -101,10 +101,10 @@ public void validate(FailureCollector collector, @Nullable OAuthInfo oAuthInfo)
   public AuthenticatorCredentials getAuthenticatorCredentials() {
     OAuthInfo oAuthInfo = getOAuthInfo();
     if (oAuthInfo != null) {
-      return new AuthenticatorCredentials(oAuthInfo, config.getConnectTimeout(), config.getReadTimeoutInMillis(),
-                                          config.getProxyUrl());
+      return AuthenticatorCredentials.fromParameters(
+              oAuthInfo, config.getConnectTimeout(), config.getReadTimeoutInMillis(), config.getProxyUrl());
     }
-    return new AuthenticatorCredentials(config.getUsername(), config.getPassword(), config.getConsumerKey(),
+    return AuthenticatorCredentials.fromParameters(config.getUsername(), config.getPassword(), config.getConsumerKey(),
                                         config.getConsumerSecret(), config.getLoginUrl(), config.getConnectTimeout(),
                                         config.getReadTimeoutInMillis(), config.getProxyUrl());
   }
@@ -140,11 +140,11 @@ private void validateConnection(@Nullable OAuthInfo oAuthInfo) {
     if (oAuthInfo == null) {
       return;
     }
+    AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(
+            oAuthInfo, config.getConnectTimeout(), config.getReadTimeoutInMillis(), config.getProxyUrl());
 
     try {
-      SalesforceConnectionUtil.getPartnerConnection(new AuthenticatorCredentials(oAuthInfo, config.getConnectTimeout(),
-                                                                                 config.getReadTimeoutInMillis(),
-                                                                                 config.getProxyUrl()));
+      SalesforceConnectionUtil.getPartnerConnection(credentials);
     } catch (ConnectionException e) {
       String message = SalesforceConnectionUtil.getSalesforceErrorMessageFromException(e);
       throw new RuntimeException(

src/main/java/io/cdap/plugin/salesforce/plugin/connector/SalesforceConnector.java

@@ -102,7 +102,8 @@ public BrowseDetail browse(ConnectorContext connectorContext, BrowseRequest brow
    * @throws IOException In case of Salesforce connection failure while browsing
    */
   public BrowseDetail browse(boolean onlyReturnQueryableObjects) throws IOException {
-    AuthenticatorCredentials credentials = new AuthenticatorCredentials(config.getUsername(), config.getPassword(),
+    AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(
+                                                                        config.getUsername(), config.getPassword(),
                                                                         config.getConsumerKey(),
                                                                         config.getConsumerSecret(),
                                                                         config.getLoginUrl(),

src/main/java/io/cdap/plugin/salesforce/plugin/sink/batch/SalesforceSinkConfig.java

@@ -268,7 +268,7 @@ public String getFQN(String orgId, String sObject) {
   }
 
   public String getOrgId(OAuthInfo oAuthInfo) throws ConnectionException {
-    AuthenticatorCredentials credentials = new AuthenticatorCredentials(oAuthInfo,
+    AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(oAuthInfo,
                                                                         this.getConnection().getConnectTimeout(),
                                                                         this.getConnection().getReadTimeout(),
                                                                         this.connection.getProxyUrl());
@@ -430,7 +430,7 @@ private Set<String> getCreatableSObjectFields(SObjectsDescribeResult describeRes
   }
 
   private SObjectsDescribeResult getSObjectDescribeResult(FailureCollector collector, OAuthInfo oAuthInfo) {
-    AuthenticatorCredentials credentials = new AuthenticatorCredentials(oAuthInfo,
+    AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(oAuthInfo,
                                                                         this.getConnection().getConnectTimeout(),
                                                                         this.getConnection().getReadTimeout(),
                                                                         this.getConnection().getProxyUrl());

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBaseSourceConfig.java

@@ -214,7 +214,7 @@ public String getFQN(String orgId, String sObject) {
   }
 
   public String getOrgId(OAuthInfo oAuthInfo) throws ConnectionException {
-    AuthenticatorCredentials credentials = new AuthenticatorCredentials(oAuthInfo,
+    AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(oAuthInfo,
                                                                         this.getConnection().getConnectTimeout(),
                                                                         this.getConnection().getReadTimeout(),
                                                                         this.connection.getProxyUrl());
@@ -277,7 +277,7 @@ public void validateFilters(FailureCollector collector) {
    */
   protected String getSObjectQuery(String sObjectName, Schema schema, long logicalStartTime, OAuthInfo oAuthInfo) {
     try {
-      AuthenticatorCredentials credentials = new AuthenticatorCredentials(oAuthInfo,
+      AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(oAuthInfo,
                                                                           this.getConnection().getConnectTimeout(),
                                                                           this.getConnection().getReadTimeout(),
                                                                           this.connection.getProxyUrl());

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBatchSource.java

@@ -211,7 +211,7 @@ public static Schema getSchema(SalesforceSourceConfig config, OAuthInfo oAuthInf
     String query = config.getQuery(System.currentTimeMillis(), oAuthInfo);
     SObjectDescriptor sObjectDescriptor = SObjectDescriptor.fromQuery(query);
     try {
-      AuthenticatorCredentials credentials = new AuthenticatorCredentials(oAuthInfo,
+      AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(oAuthInfo,
                                                                           config.getConnection().getConnectTimeout(),
                                                                           config.getConnection().getReadTimeout(),
                                                                           config.getConnection().getProxyUrl());