(CDAP-17611) Added OAuth function support

pom.xml

@@ -439,8 +439,12 @@
           <version>1.1.0</version>
           <configuration>
             <cdapArtifacts>
-              <parent>system:cdap-data-pipeline[6.1.0-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
-              <parent>system:cdap-data-streams[6.1.0-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
+              <!--
+                Plugins use the new macro substitution that supports object, which is introduced in 6.4.0.
+                The cdap API version that this project depends on is still 6.2.0 since there is no new API needed.
+               -->
+              <parent>system:cdap-data-pipeline[6.4.0-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
+              <parent>system:cdap-data-streams[6.4.0-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
             </cdapArtifacts>
           </configuration>
           <executions>

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

@@ -20,6 +20,7 @@
 import com.sforce.ws.ConnectorConfig;
 import io.cdap.plugin.salesforce.authenticator.Authenticator;
 import io.cdap.plugin.salesforce.authenticator.AuthenticatorCredentials;
+import io.cdap.plugin.salesforce.plugin.OAuthInfo;
 import org.apache.hadoop.conf.Configuration;
 
 /**
@@ -41,33 +42,23 @@ public static PartnerConnection getPartnerConnection(AuthenticatorCredentials cr
     return new PartnerConnection(connectorConfig);
   }
 
-  /**
-   * Creates {@link AuthenticatorCredentials} instance based on given parameters.
-   *
-   * @param username Salesforce username
-   * @param password Salesforce password
-   * @param consumerKey Salesforce consumer key
-   * @param consumerSecret Salesforce consumer secret
-   * @param loginUrl Salesforce authentication url
-   * @return authenticator credentials
-   */
-  public static AuthenticatorCredentials getAuthenticatorCredentials(String username, String password,
-                                                                     String consumerKey, String consumerSecret,
-                                                                     String loginUrl) {
-    return new AuthenticatorCredentials(username, password, consumerKey, consumerSecret, loginUrl);
-  }
-
   /**
    * Creates {@link AuthenticatorCredentials} instance based on given {@link Configuration}.
    *
    * @param conf hadoop job configuration
    * @return authenticator credentials
    */
   public static AuthenticatorCredentials getAuthenticatorCredentials(Configuration conf) {
-    return getAuthenticatorCredentials(conf.get(SalesforceConstants.CONFIG_USERNAME),
-                                       conf.get(SalesforceConstants.CONFIG_PASSWORD),
-                                       conf.get(SalesforceConstants.CONFIG_CONSUMER_KEY),
-                                       conf.get(SalesforceConstants.CONFIG_CONSUMER_SECRET),
-                                       conf.get(SalesforceConstants.CONFIG_LOGIN_URL));
+    String oAuthToken = conf.get(SalesforceConstants.CONFIG_OAUTH_TOKEN);
+    String instanceURL = conf.get(SalesforceConstants.CONFIG_OAUTH_INSTANCE_URL);
+    if (oAuthToken != null && instanceURL != null) {
+      return new AuthenticatorCredentials(new OAuthInfo(oAuthToken, instanceURL));
+    }
+
+    return new AuthenticatorCredentials(conf.get(SalesforceConstants.CONFIG_USERNAME),
+                                        conf.get(SalesforceConstants.CONFIG_PASSWORD),
+                                        conf.get(SalesforceConstants.CONFIG_CONSUMER_KEY),
+                                        conf.get(SalesforceConstants.CONFIG_CONSUMER_SECRET),
+                                        conf.get(SalesforceConstants.CONFIG_LOGIN_URL));
   }
 }

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

@@ -29,7 +29,10 @@ public class SalesforceConstants {
   public static final String PROPERTY_PASSWORD = "password";
   public static final String PROPERTY_SECURITY_TOKEN = "securityToken";
   public static final String PROPERTY_LOGIN_URL = "loginUrl";
+  public static final String PROPERTY_OAUTH_INFO = "oAuthInfo";
 
+  public static final String CONFIG_OAUTH_TOKEN = "mapred.salesforce.oauth.token";
+  public static final String CONFIG_OAUTH_INSTANCE_URL = "mapred.salesforce.oauth.instance.url";
   public static final String CONFIG_CONSUMER_KEY = "mapred.salesforce.consumer.key";
   public static final String CONFIG_PASSWORD = "mapred.salesforce.password";
   public static final String CONFIG_USERNAME = "mapred.salesforce.user";

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

@@ -20,6 +20,7 @@
 import com.google.gson.Gson;
 import com.sforce.ws.ConnectorConfig;
 import io.cdap.plugin.salesforce.SalesforceConstants;
+import io.cdap.plugin.salesforce.plugin.OAuthInfo;
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 
@@ -39,12 +40,12 @@ public class Authenticator {
    */
   public static ConnectorConfig createConnectorConfig(AuthenticatorCredentials credentials) {
     try {
-      AuthResponse authResponse = oauthLogin(credentials);
+      OAuthInfo oAuthInfo = getOAuthInfo(credentials);
       ConnectorConfig connectorConfig = new ConnectorConfig();
-      connectorConfig.setSessionId(authResponse.getAccessToken());
+      connectorConfig.setSessionId(oAuthInfo.getAccessToken());
       String apiVersion = SalesforceConstants.API_VERSION;
-      String restEndpoint = String.format("%s/services/async/%s", authResponse.getInstanceUrl(), apiVersion);
-      String serviceEndPoint = String.format("%s/services/Soap/u/%s", authResponse.getInstanceUrl(), apiVersion);
+      String restEndpoint = String.format("%s/services/async/%s", oAuthInfo.getInstanceURL(), apiVersion);
+      String serviceEndPoint = String.format("%s/services/Soap/u/%s", oAuthInfo.getInstanceURL(), apiVersion);
       connectorConfig.setRestEndpoint(restEndpoint);
       connectorConfig.setServiceEndpoint(serviceEndPoint);
       // This should only be false when doing debugging.
@@ -65,7 +66,12 @@ public static ConnectorConfig createConnectorConfig(AuthenticatorCredentials cre
    *
    * @return AuthResponse response to http request
    */
-  public static AuthResponse oauthLogin(AuthenticatorCredentials credentials) throws Exception {
+  public static OAuthInfo getOAuthInfo(AuthenticatorCredentials credentials) throws Exception {
+    OAuthInfo oAuthInfo = credentials.getOAuthInfo();
+    if (oAuthInfo != null) {
+      return oAuthInfo;
+    }
+
     SslContextFactory sslContextFactory = new SslContextFactory();
     HttpClient httpClient = new HttpClient(sslContextFactory);
     try {
@@ -83,7 +89,7 @@ public static AuthResponse oauthLogin(AuthenticatorCredentials credentials) thro
           String.format("Cannot authenticate to Salesforce with given credentials. ServerResponse='%s'", response));
       }
 
-      return authResponse;
+      return new OAuthInfo(authResponse.getAccessToken(), authResponse.getInstanceUrl());
     } finally {
       httpClient.stop();
     }

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

@@ -16,44 +16,74 @@
 
 package io.cdap.plugin.salesforce.authenticator;
 
+import io.cdap.plugin.salesforce.plugin.OAuthInfo;
+
 import java.io.Serializable;
 import java.util.Objects;
+import javax.annotation.Nullable;
 
 /**
  * Stores information to connect to salesforce via oauth2
  */
 public class AuthenticatorCredentials implements Serializable {
+
+  private final OAuthInfo oAuthInfo;
   private final String username;
   private final String password;
   private final String consumerKey;
   private final String consumerSecret;
   private final String loginUrl;
 
+  public AuthenticatorCredentials(OAuthInfo oAuthInfo) {
+    this(Objects.requireNonNull(oAuthInfo), null, null, null, null, null);
+  }
+
   public AuthenticatorCredentials(String username, String password,
                                   String consumerKey, String consumerSecret, String loginUrl) {
+    this(null, Objects.requireNonNull(username), Objects.requireNonNull(password), Objects.requireNonNull(consumerKey),
+         Objects.requireNonNull(consumerSecret), Objects.requireNonNull(loginUrl));
+  }
+
+  private AuthenticatorCredentials(@Nullable OAuthInfo oAuthInfo,
+                                   @Nullable String username,
+                                   @Nullable String password,
+                                   @Nullable String consumerKey,
+                                   @Nullable String consumerSecret,
+                                   @Nullable String loginUrl) {
+    this.oAuthInfo = oAuthInfo;
     this.username = username;
     this.password = password;
     this.consumerKey = consumerKey;
     this.consumerSecret = consumerSecret;
     this.loginUrl = loginUrl;
   }
 
+  @Nullable
+  public OAuthInfo getOAuthInfo() {
+    return oAuthInfo;
+  }
+
+  @Nullable
   public String getUsername() {
     return username;
   }
 
+  @Nullable
   public String getPassword() {
     return password;
   }
 
+  @Nullable
   public String getConsumerKey() {
     return consumerKey;
   }
 
+  @Nullable
   public String getConsumerSecret() {
     return consumerSecret;
   }
 
+  @Nullable
   public String getLoginUrl() {
     return loginUrl;
   }

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

@@ -32,65 +32,95 @@
  */
 public class BaseSalesforceConfig extends ReferencePluginConfig {
 
+  @Name(SalesforceConstants.PROPERTY_OAUTH_INFO)
+  @Description("OAuth information for connecting to Salesforce. " +
+    "It is expected to be an json string containing two properties, \"accessToken\" and \"instanceURL\", " +
+    "which carry the OAuth access token and the URL to connect to respectively. " +
+    "Use the ${oauth(provider, credentialId)} macro function for acquiring OAuth information dynamically. ")
+  @Macro
+  @Nullable
+  private OAuthInfo oAuthInfo;
+
   @Name(SalesforceConstants.PROPERTY_CONSUMER_KEY)
   @Description("Salesforce connected app's consumer key")
   @Macro
+  @Nullable
   private String consumerKey;
 
   @Name(SalesforceConstants.PROPERTY_CONSUMER_SECRET)
   @Description("Salesforce connected app's client secret key")
   @Macro
+  @Nullable
   private String consumerSecret;
 
   @Name(SalesforceConstants.PROPERTY_USERNAME)
   @Description("Salesforce username")
   @Macro
+  @Nullable
   private String username;
 
   @Name(SalesforceConstants.PROPERTY_PASSWORD)
   @Description("Salesforce password")
   @Macro
+  @Nullable
   private String password;
 
   @Name(SalesforceConstants.PROPERTY_SECURITY_TOKEN)
   @Description("Salesforce security token")
-  @Nullable
   @Macro
+  @Nullable
   private String securityToken;
 
   @Name(SalesforceConstants.PROPERTY_LOGIN_URL)
   @Description("Endpoint to authenticate to")
   @Macro
+  @Nullable
   private String loginUrl;
 
-  public BaseSalesforceConfig(String referenceName, String consumerKey, String consumerSecret,
-                              String username, String password, String loginUrl,
-                              @Nullable String securityToken) {
+  public BaseSalesforceConfig(String referenceName,
+                              @Nullable String consumerKey,
+                              @Nullable String consumerSecret,
+                              @Nullable String username,
+                              @Nullable String password,
+                              @Nullable String loginUrl,
+                              @Nullable String securityToken,
+                              @Nullable OAuthInfo oAuthInfo) {
     super(referenceName);
     this.consumerKey = consumerKey;
     this.consumerSecret = consumerSecret;
     this.username = username;
     this.password = password;
     this.loginUrl = loginUrl;
     this.securityToken = securityToken;
+    this.oAuthInfo = oAuthInfo;
+  }
+
+  @Nullable
+  public OAuthInfo getOAuthInfo() {
+    return oAuthInfo;
   }
 
+  @Nullable
   public String getConsumerKey() {
     return consumerKey;
   }
 
+  @Nullable
   public String getConsumerSecret() {
     return consumerSecret;
   }
 
+  @Nullable
   public String getUsername() {
     return username;
   }
 
+  @Nullable
   public String getPassword() {
     return constructPasswordWithToken(password, securityToken);
   }
 
+  @Nullable
   public String getLoginUrl() {
     return loginUrl;
   }
@@ -99,15 +129,21 @@ public void validate(FailureCollector collector) {
     try {
       validateConnection();
     } catch (Exception e) {
-      collector.addFailure("Error encountered while establishing connection: " + e.getMessage(), null)
+      collector.addFailure("Error encountered while establishing connection: " + e.getMessage(),
+                           "Please verify authentication properties are provided correctly")
         .withStacktrace(e.getStackTrace());
     }
     collector.getOrThrowException();
   }
 
   public AuthenticatorCredentials getAuthenticatorCredentials() {
-    return SalesforceConnectionUtil.getAuthenticatorCredentials(username, getPassword(),
-                                                                consumerKey, consumerSecret, loginUrl);
+    OAuthInfo oAuthInfo = getOAuthInfo();
+    if (oAuthInfo != null) {
+      return new AuthenticatorCredentials(oAuthInfo);
+    }
+
+    return new AuthenticatorCredentials(getUsername(), getPassword(), getConsumerKey(),
+                                        getConsumerSecret(), getLoginUrl());
   }
 
   /**
@@ -117,6 +153,16 @@ public AuthenticatorCredentials getAuthenticatorCredentials() {
    * @return true if none of the connection properties contains macro, false otherwise
    */
   public boolean canAttemptToEstablishConnection() {
+    // If OAuth token is configured, use it to establish connection
+    if (getOAuthInfo() != null) {
+      return true;
+    }
+
+    // At configurePipeline time, macro is not resolved, hence the OAuth field will be null.
+    if (containsMacro(SalesforceConstants.PROPERTY_OAUTH_INFO)) {
+      return false;
+    }
+
     return !(containsMacro(SalesforceConstants.PROPERTY_CONSUMER_KEY)
       || containsMacro(SalesforceConstants.PROPERTY_CONSUMER_SECRET)
       || containsMacro(SalesforceConstants.PROPERTY_USERNAME)

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright © 2021 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.salesforce.plugin;
+
+/**
+ * Class to carry OAuth information returned by the {@code ${oauth}} macro function.
+ */
+public final class OAuthInfo {
+
+  private final String accessToken;
+  private final String instanceURL;
+
+  public OAuthInfo(String accessToken, String instanceURL) {
+    this.accessToken = accessToken;
+    this.instanceURL = instanceURL;
+  }
+
+  public String getAccessToken() {
+    return accessToken;
+  }
+
+  public String getInstanceURL() {
+    return instanceURL;
+  }
+}