Implement BYOC Oauth

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 3.0.0
+  - Replace Auth code based OAuth with Client Credentials OAuth. [#220](https://github.com/singer-io/tap-salesforce/pull/220)
+
 ## 2.9.0
   - Optimize discovery by using composite batch API [#218](https://github.com/singer-io/tap-salesforce/pull/218)
 

README.md

@@ -27,15 +27,17 @@ $ tap-salesforce --config config.json --properties properties.json --state state
 {
   "client_id": "secret_client_id",
   "client_secret": "secret_client_secret",
-  "refresh_token": "abc123",
+  "instance_url": "https://test.my.salesforce.com",
   "start_date": "2017-11-02T00:00:00Z",
   "api_type": "BULK",
-  "select_fields_by_default": true
+  "select_fields_by_default": true,
   "lookback_window": 10
 }
 ```
 
-The `client_id` and `client_secret` keys are your OAuth Salesforce App secrets. The `refresh_token` is a secret created during the OAuth flow. For more info on the Salesforce OAuth flow, visit the [Salesforce documentation](https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/intro_understanding_web_server_oauth_flow.htm). Additionnaly, if the Salesforce Sandbox is to be used to run the tap, the parameter `"is_sandbox": true` must be passed to the config.
+The `client_id` and `client_secret` keys are your ECA(External Client App)'s secrets. Find details [here](https://help.salesforce.com/s/articleView?id=xcloud.create_a_local_external_client_app.htm&type=5) on how to create External Client App.
+The `instance_url` is My Domain URL which you can find at `My Domain → My Domain Settings → My Domain Details → Current My Domain URL`
+
 
 The `start_date` is used by the tap as a bound on SOQL queries when searching for records.  This should be an [RFC3339](https://www.ietf.org/rfc/rfc3339.txt) formatted date-time, like "2018-01-08T00:00:00Z". For more details, see the [Singer best practices for dates](https://github.com/singer-io/getting-started/blob/master/BEST_PRACTICES.md#dates).
 

setup.py

@@ -3,7 +3,7 @@
 from setuptools import setup
 
 setup(name='tap-salesforce',
-      version='2.9.0',
+      version='3.0.0',
       description='Singer.io tap for extracting data from the Salesforce API',
       author='Stitch',
       url='https://singer.io',

tap_salesforce/__init__.py

@@ -19,14 +19,13 @@
 
 LOGGER = singer.get_logger()
 
-REQUIRED_CONFIG_KEYS = ['refresh_token',
-                        'client_id',
+REQUIRED_CONFIG_KEYS = ['client_id',
                         'client_secret',
                         'start_date',
-                        'api_type']
+                        'api_type',
+                        'instance_url']
 
 CONFIG = {
-    'refresh_token': None,
     'client_id': None,
     'client_secret': None,
     'start_date': None
@@ -403,17 +402,16 @@ def main_impl():
         lookback_window = int(lookback_window) if lookback_window else None
 
         sf = Salesforce(
-            refresh_token=CONFIG['refresh_token'],
             sf_client_id=CONFIG['client_id'],
             sf_client_secret=CONFIG['client_secret'],
             quota_percent_total=CONFIG.get('quota_percent_total'),
             quota_percent_per_run=CONFIG.get('quota_percent_per_run'),
-            is_sandbox=CONFIG.get('is_sandbox'),
             select_fields_by_default=CONFIG.get('select_fields_by_default'),
             default_start_date=CONFIG.get('start_date'),
             api_type=CONFIG.get('api_type'),
             lookback_window=lookback_window,
-            config_path=args.config_path)
+            config_path=args.config_path,
+            instance_url=CONFIG['instance_url'])
         sf.login()
 
         if args.discover:

tap_salesforce/salesforce/__init__.py

@@ -22,6 +22,7 @@
 # The minimum expiration setting for SF Refresh Tokens is 15 minutes
 REFRESH_TOKEN_EXPIRATION_PERIOD = 900
 
+
 BULK_API_TYPE = "BULK"
 REST_API_TYPE = "REST"
 BATCH_DESCRIBE_SIZE = 25
@@ -212,27 +213,25 @@ def field_to_property_schema(field, mdata): # pylint:disable=too-many-branches
 class Salesforce():
     # pylint: disable=too-many-instance-attributes,too-many-arguments,too-many-positional-arguments
     def __init__(self,
-                 refresh_token=None,
                  token=None,
                  sf_client_id=None,
                  sf_client_secret=None,
                  quota_percent_per_run=None,
                  quota_percent_total=None,
-                 is_sandbox=None,
                  select_fields_by_default=None,
                  default_start_date=None,
                  api_type=None,
                  lookback_window=None,
-                 config_path=None):
+                 config_path=None,
+                 instance_url=None):
         self.api_type = api_type.upper() if api_type else None
-        self.refresh_token = refresh_token
         self.token = token
         self.config_path = config_path
         self.sf_client_id = sf_client_id
         self.sf_client_secret = sf_client_secret
         self.session = requests.Session()
         self.access_token = None
-        self.instance_url = None
+        self.instance_url = instance_url
         if isinstance(quota_percent_per_run, str) and quota_percent_per_run.strip() == '':
             quota_percent_per_run = None
         if isinstance(quota_percent_total, str) and quota_percent_total.strip() == '':
@@ -241,7 +240,6 @@ def __init__(self,
             quota_percent_per_run) if quota_percent_per_run is not None else 25
         self.quota_percent_total = float(
             quota_percent_total) if quota_percent_total is not None else 80
-        self.is_sandbox = is_sandbox is True or (isinstance(is_sandbox, str) and is_sandbox.lower() == 'true')
         self.select_fields_by_default = select_fields_by_default is True or (isinstance(select_fields_by_default, str) and select_fields_by_default.lower() == 'true')
         self.default_start_date = default_start_date
         self.rest_requests_attempted = 0
@@ -257,15 +255,6 @@ def __init__(self,
     def _get_standard_headers(self):
         return {"Authorization": "Bearer {}".format(self.access_token)}
 
-    def _write_config(self):
-        """Save updated config (with new token) back to config file."""
-        with open(self.config_path, 'r', encoding='utf-8') as f:
-            config = json.load(f)
-        config['refresh_token'] = self.refresh_token
-        with open(self.config_path, 'w', encoding='utf-8') as f:
-            json.dump(config, f, indent=2)
-        LOGGER.info("Updated config file with new refresh token.")
-
     # pylint: disable=anomalous-backslash-in-string,line-too-long
     def check_rest_quota_usage(self, headers):
         match = re.search('^api-usage=(\d+)/(\d+)$', headers.get('Sforce-Limit-Info'))
@@ -314,7 +303,7 @@ def _make_request(self, http_method, url, headers=None, body=None, stream=False,
                                         params=params,
                                         timeout=request_timeout,)
             elif http_method == "POST":
-                LOGGER.info("Making %s request to %s with body %s", http_method, url, body)
+                LOGGER.info("Making %s request to %s", http_method, url)
                 resp = self.session.post(url,
                                          headers=headers,
                                          data=body,
@@ -343,35 +332,21 @@ def _make_request(self, http_method, url, headers=None, body=None, stream=False,
         return resp
 
     def login(self):
-        if self.is_sandbox:
-            login_url = 'https://test.salesforce.com/services/oauth2/token'
-        else:
-            login_url = 'https://login.salesforce.com/services/oauth2/token'
+        login_url = '{}/services/oauth2/token'.format(self.instance_url.rstrip('/'))
 
-        login_body = {'grant_type': 'refresh_token', 'client_id': self.sf_client_id,
-                      'client_secret': self.sf_client_secret, 'refresh_token': self.refresh_token}
-
-        LOGGER.info("Attempting login via OAuth2")
+        login_body = {'grant_type': 'client_credentials',
+                      'client_id': self.sf_client_id,
+                      'client_secret': self.sf_client_secret}
+        LOGGER.info("Attempting login via OAuth2 client_credentials flow")
 
         resp = None
         try:
             resp = self._make_request("POST", login_url, body=login_body, headers={"Content-Type": "application/x-www-form-urlencoded"})
 
-            LOGGER.info("OAuth2 login successful")
+            LOGGER.info("OAuth2 login successful using Client Credentials flow")
 
             auth = resp.json()
-
             self.access_token = auth['access_token']
-            self.instance_url = auth['instance_url']
-            # Salesforce may or may not return a new refresh token. If it does,
-            # we should update to use the new one.
-            new_refresh_token = auth.get('refresh_token')
-            if new_refresh_token and new_refresh_token != self.refresh_token:
-                LOGGER.info("Refresh token rotation detected. Updating refresh token.")
-                self.refresh_token = new_refresh_token
-                self._write_config()
-            else:
-                LOGGER.info("No refresh token rotation detected.")
         except Exception as e:
             error_message = str(e)
             if resp is None and hasattr(e, 'response') and e.response is not None: #pylint:disable=no-member

tests/base.py

@@ -46,16 +46,15 @@ def tap_name():
     @staticmethod
     def get_type():
         """the expected url route ending"""
-        return "platform.salesforce"
+        return "platform.salesforce-byoc"
 
     def get_properties(self, original: bool = True):
         """Configuration properties required for the tap."""
         return_value = {
             'start_date': '2020-11-23T00:00:00Z',
-            'instance_url': 'https://singer2-dev-ed.my.salesforce.com',
+            'instance_url': 'https://qlik76-dev-ed.develop.my.salesforce.com',
             'quota_percent_total': '95',
             'api_type': self.salesforce_api,
-            'is_sandbox': 'false'
         }
 
         if original:
@@ -70,9 +69,8 @@ def get_properties(self, original: bool = True):
     @staticmethod
     def get_credentials():
         """Authentication information for the test account"""
-        return {'refresh_token': os.getenv('TAP_SALESFORCE_REFRESH_TOKEN'),
-                'client_id': os.getenv('TAP_SALESFORCE_CLIENT_ID'),
-                'client_secret': os.getenv('TAP_SALESFORCE_CLIENT_SECRET')}
+        return {'client_id': os.getenv('TAP_SALESFORCE_BYOC_CLIENT_ID'),
+                'client_secret': os.getenv('TAP_SALESFORCE_BYOC_CLIENT_SECRET')}
 
     def expected_metadata(self):
         """The expected streams and metadata about the streams"""
@@ -1182,9 +1180,8 @@ def expected_replication_method(self):
 
     def setUp(self):
         """Verify that you have set the prerequisites to run the tap (creds, etc.)"""
-        missing_envs = [x for x in ['TAP_SALESFORCE_CLIENT_ID',
-                                    'TAP_SALESFORCE_CLIENT_SECRET',
-                                    'TAP_SALESFORCE_REFRESH_TOKEN']
+        missing_envs = [x for x in ['TAP_SALESFORCE_BYOC_CLIENT_ID',
+                                    'TAP_SALESFORCE_BYOC_CLIENT_SECRET']
                         if os.getenv(x) is None]
 
         if missing_envs:

tests/sfbase.py

@@ -33,27 +33,25 @@ def tap_name():
     @staticmethod
     def get_type():
         """the expected url route ending"""
-        return "platform.salesforce"
+        return "platform.salesforce-byoc"
 
     def get_properties(self):
         """Configuration properties required for the tap."""
 
         return {
             'start_date': self.start_date,
-            'instance_url': 'https://singer2-dev-ed.my.salesforce.com',
+            'instance_url': 'https://qlik76-dev-ed.develop.my.salesforce.com',
             'select_fields_by_default': 'true',
             'quota_percent_total': self.total_quota,
             'quota_percent_per_run' : self.per_run_quota,
             'api_type': self.salesforce_api,
-            'is_sandbox': 'false'
         }
 
     @staticmethod
     def get_credentials():
         """Authentication information for the test account"""
-        return {'refresh_token': os.getenv('TAP_SALESFORCE_REFRESH_TOKEN'),
-                'client_id': os.getenv('TAP_SALESFORCE_CLIENT_ID'),
-                'client_secret': os.getenv('TAP_SALESFORCE_CLIENT_SECRET')}
+        return {'client_id': os.getenv('TAP_SALESFORCE_BYOC_CLIENT_ID'),
+                'client_secret': os.getenv('TAP_SALESFORCE_BYOC_CLIENT_SECRET')}
 
     def run_and_verify_check_mode(self, conn_id):
         """
@@ -77,7 +75,9 @@ def run_and_verify_check_mode(self, conn_id):
         # TODO do we want this?
         # Verify the expected streams are present in the catalog
         found_stream_names = {catalog['stream_name'] for catalog in found_catalogs}
-        self.assertTrue(self.expected_stream_names().issubset(found_stream_names),
+        # Streams no longer discovered from the Salesforce instance
+        missing_streams = {'TapTester__Share', 'OrgMetricScanResult', 'TapTester__c'}
+        self.assertTrue(self.expected_stream_names().difference(missing_streams).issubset(found_stream_names),
                         logging="Expected streams are present in catalog.")
 
         return found_catalogs
@@ -1174,9 +1174,8 @@ def set_replication_methods(self, conn_id, catalogs, replication_methods):
     @classmethod
     def setUpClass(cls):
         """Verify that you have set the prerequisites to run the tap (creds, etc.)"""
-        missing_envs = [x for x in ['TAP_SALESFORCE_CLIENT_ID',
-                                    'TAP_SALESFORCE_CLIENT_SECRET',
-                                    'TAP_SALESFORCE_REFRESH_TOKEN']
+        missing_envs = [x for x in ['TAP_SALESFORCE_BYOC_CLIENT_ID',
+                                    'TAP_SALESFORCE_BYOC_CLIENT_SECRET']
                         if os.getenv(x) is None]
 
         if missing_envs:

tests/test_salesforce_discovery_rest.py

@@ -28,7 +28,8 @@ def discovery_test(self):
         # BUG | https://jira.talendforge.org/browse/TDL-15748
         #      The following streams stopped being discovered 10/10/2021
         #      When bug is addressed fix the marked lines
-        missing_streams = {'DataAssetUsageTrackingInfo', 'DataAssetSemanticGraphEdge'}
+        missing_streams = {'DataAssetUsageTrackingInfo', 'DataAssetSemanticGraphEdge',
+                           'TapTester__Share', 'OrgMetricScanResult', 'TapTester__c'}
 
         streams_to_test = self.expected_streams() - missing_streams # BUG_TDL-15748
         # streams_to_test_prime = self.expected_streams().difference(self.get_unsupported_by_bulk_api())
@@ -139,7 +140,7 @@ def discovery_test(self):
                     'ApiEvent', 'WorkOrder', 'ContactCleanInfo', 'ResourceAbsence', 'ReturnOrder',
                     'LegalEntity', 'PaymentMethod', 'EventLogFile', 'ServiceAppointment',
                     'DandBCompany', 'AccountCleanInfo', 'Organization', 'Document', 'Account',
-                    'Address', 'FulfillmentOrder', 'Asset'
+                    'Address', 'FulfillmentOrder', 'Asset', 'Location'
                 }
 
                 # verify that all other fields have inclusion of available

tests/test_salesforce_lookback_window.py

@@ -18,10 +18,9 @@ def name():
     def get_properties(self):  # pylint: disable=arguments-differ
         return {
             'start_date' : '2021-11-10T00:00:00Z',
-            'instance_url': 'https://singer2-dev-ed.my.salesforce.com',
+            'instance_url': 'https://qlik76-dev-ed.develop.my.salesforce.com',
             'select_fields_by_default': 'true',
             'api_type': self.salesforce_api,
-            'is_sandbox': 'false',
             'lookback_window': 86400
         }
 

tests/test_salesforce_sync_canary.py

@@ -20,10 +20,9 @@ def name():
     def get_properties():  # pylint: disable=arguments-differ
         return {
             'start_date' : '2024-03-12T00:00:00Z',
-            'instance_url': 'https://singer2-dev-ed.my.salesforce.com',
+            'instance_url': 'https://qlik76-dev-ed.develop.my.salesforce.com',
             'select_fields_by_default': 'true',
             'api_type': 'BULK',
-            'is_sandbox': 'false'
         }
 
     def expected_sync_streams(self):