#16720: Add Support for Salesforce SSL (#16719)

ingestion/src/metadata/examples/workflows/salesforce.yaml

@@ -8,6 +8,11 @@ source:
       password: password
       securityToken: securityToken
       sobjectName: sobjectName
+      # sslConfig:
+      #   caCertificate: |
+      #     -----BEGIN CERTIFICATE-----
+      #     sample caCertificateData  
+      #     -----END CERTIFICATE-----
       # salesforceApiVersion: 42.0
       # salesforceDomain: login
   sourceConfig:

ingestion/src/metadata/ingestion/source/database/salesforce/connection.py

@@ -14,7 +14,7 @@ Source connection handler
 """
 from typing import Optional
 
-from simple_salesforce import Salesforce
+from simple_salesforce.api import Salesforce
 from sqlalchemy.engine import Engine
 
 from metadata.generated.schema.entity.automations.workflow import (
@@ -32,11 +32,12 @@ def get_connection(connection: SalesforceConnection) -> Engine:
     Create connection
     """
     return Salesforce(
-        connection.username,
+        username=connection.username,
         password=connection.password.get_secret_value(),
         security_token=connection.securityToken.get_secret_value(),
         domain=connection.salesforceDomain,
         version=connection.salesforceApiVersion,
+        **connection.connectionArguments.root if connection.connectionArguments else {},
     )
 
 

ingestion/src/metadata/ingestion/source/database/salesforce/metadata.py

@@ -57,6 +57,7 @@ from metadata.utils import fqn
 from metadata.utils.constants import DEFAULT_DATABASE
 from metadata.utils.filters import filter_by_table
 from metadata.utils.logger import ingestion_logger
+from metadata.utils.ssl_manager import SSLManager, check_ssl_and_init
 
 logger = ingestion_logger()
 
@@ -77,6 +78,11 @@ class SalesforceSource(DatabaseServiceSource):
         )
         self.metadata = metadata
         self.service_connection = self.config.serviceConnection.root.config
+        self.ssl_manager: SSLManager = check_ssl_and_init(self.service_connection)
+        if self.ssl_manager:
+            self.service_connection = self.ssl_manager.setup_ssl(
+                self.service_connection
+            )
         self.client = get_connection(self.service_connection)
         self.table_constraints = None
         self.database_source_state = set()

ingestion/src/metadata/utils/ssl_manager.py

@@ -39,11 +39,15 @@ from metadata.generated.schema.entity.services.connections.database.postgresConn
 from metadata.generated.schema.entity.services.connections.database.redshiftConnection import (
     RedshiftConnection,
 )
+from metadata.generated.schema.entity.services.connections.database.salesforceConnection import (
+    SalesforceConnection,
+)
 from metadata.generated.schema.entity.services.connections.messaging.kafkaConnection import (
     KafkaConnection,
 )
 from metadata.generated.schema.security.ssl import verifySSLConfig
 from metadata.ingestion.connections.builders import init_empty_connection_arguments
+from metadata.ingestion.models.custom_pydantic import CustomSecretStr
 from metadata.ingestion.source.connections import get_connection
 from metadata.utils.logger import utils_logger
 
@@ -126,6 +130,25 @@ class SSLManager:
                 )
         return connection
 
+    @setup_ssl.register(SalesforceConnection)
+    def _(self, connection):
+        import requests  # pylint: disable=import-outside-toplevel
+
+        connection: SalesforceConnection = cast(SalesforceConnection, connection)
+        connection.connectionArguments = (
+            connection.connectionArguments or init_empty_connection_arguments()
+        )
+        session = requests.Session()
+        if self.ca_file_path:
+            session.verify = self.ca_file_path
+        if self.cert_file_path and self.key_file_path:
+            session.cert = (self.cert_file_path, self.key_file_path)
+        connection.connectionArguments.root = (
+            connection.connectionArguments.root or {}
+        )  # to satisfy mypy
+        connection.connectionArguments.root["session"] = session
+        return connection
+
     @setup_ssl.register(QlikSenseConnection)
     def _(self, connection):
         return {
@@ -147,7 +170,22 @@ class SSLManager:
 
 
 @singledispatch
-def check_ssl_and_init(_):
+def check_ssl_and_init(_) -> None:
+    return None
+
+
+@check_ssl_and_init.register(cls=SalesforceConnection)
+def _(connection) -> Union[SSLManager, None]:
+    service_connection = cast(SalesforceConnection, connection)
+    ssl: Optional[verifySSLConfig.SslConfig] = service_connection.sslConfig
+    if ssl and ssl.root.caCertificate:
+        ssl_dict: dict[str, Union[CustomSecretStr, None]] = {
+            "ca": ssl.root.caCertificate
+        }
+        if (ssl.root.sslCertificate) and (ssl.root.sslKey):
+            ssl_dict["cert"] = ssl.root.sslCertificate
+            ssl_dict["key"] = ssl.root.sslKey
+        return SSLManager(**ssl_dict)
     return None
 
 
@@ -182,6 +220,7 @@ def _(connection):
 
 def get_ssl_connection(service_config):
     try:
+        # To be cleaned up as part of https://github.com/open-metadata/OpenMetadata/issues/15913
         ssl_manager: SSLManager = check_ssl_and_init(service_config)
         if ssl_manager:
             service_config = ssl_manager.setup_ssl(service_config)

ingestion/tests/unit/topology/database/test_salesforce.py

@@ -433,7 +433,7 @@ class SalesforceUnitTest(TestCase):
     @patch(
         "metadata.ingestion.source.database.salesforce.metadata.SalesforceSource.test_connection"
     )
-    @patch("simple_salesforce.Salesforce")
+    @patch("simple_salesforce.api.Salesforce")
     def __init__(self, methodName, salesforce, test_connection) -> None:
         super().__init__(methodName)
         test_connection.return_value = False
@@ -461,3 +461,41 @@ class SalesforceUnitTest(TestCase):
                 SALESFORCE_FIELDS[i]["type"].upper()
             )
             assert result == EXPECTED_COLUMN_TYPE[i]
+
+    @patch(
+        "metadata.ingestion.source.database.salesforce.metadata.SalesforceSource.test_connection"
+    )
+    @patch("simple_salesforce.api.Salesforce")
+    def test_check_ssl(self, salesforce, test_connection) -> None:
+        mock_salesforce_config["source"]["serviceConnection"]["config"]["sslConfig"] = {
+            "caCertificate": """
+        -----BEGIN CERTIFICATE-----
+        sample caCertificateData  
+        -----END CERTIFICATE-----
+        """
+        }
+
+        mock_salesforce_config["source"]["serviceConnection"]["config"]["sslConfig"][
+            "sslKey"
+        ] = """
+        -----BEGIN CERTIFICATE-----
+        sample caCertificateData  
+        -----END CERTIFICATE-----
+        """
+        mock_salesforce_config["source"]["serviceConnection"]["config"]["sslConfig"][
+            "sslCertificate"
+        ] = """
+        -----BEGIN CERTIFICATE-----
+        sample sslCertificateData
+        -----END CERTIFICATE-----
+        """
+
+        test_connection.return_value = False
+        self.config = OpenMetadataWorkflowConfig.model_validate(mock_salesforce_config)
+        self.salesforce_source = SalesforceSource.create(
+            mock_salesforce_config["source"],
+            self.config.workflowConfig.openMetadataServerConfig,
+        )
+        self.assertTrue(self.salesforce_source.ssl_manager.ca_file_path)
+        self.assertTrue(self.salesforce_source.ssl_manager.cert_file_path)
+        self.assertTrue(self.salesforce_source.ssl_manager.key_file_path)

openmetadata-docs/content/v1.5.x-SNAPSHOT/connectors/database/salesforce/index.md

@@ -53,6 +53,10 @@ These are the permissions you will require to fetch the metadata from Salesforce
 - **Salesforce Domain**: When connecting to Salesforce, you can specify the domain to use for accessing the platform. The common domains include `login` and `test`, and you can also utilize Salesforce My Domain.
 By default, the domain `login` is used for accessing Salesforce.
 
+**SSL Configuration**
+
+In order to integrate SSL in the Metadata Ingestion Config, the user will have to add the SSL config under sslConfig which is placed in the source.
+
 {% partial file="/v1.5/connectors/database/advanced-configuration.md" /%}
 
 {% /extraContent %}

openmetadata-service/src/main/java/org/openmetadata/service/secrets/converter/ClassConverterFactory.java

@@ -67,6 +67,7 @@ public final class ClassConverterFactory {
             Map.entry(SupersetConnection.class, new SupersetConnectionClassConverter()),
             Map.entry(SSOAuthMechanism.class, new SSOAuthMechanismClassConverter()),
             Map.entry(TableauConnection.class, new TableauConnectionClassConverter()),
+            Map.entry(SalesforceConnection.class, new SalesforceConnectorClassConverter()),
             Map.entry(
                 TestServiceConnectionRequest.class,
                 new TestServiceConnectionRequestClassConverter()),

openmetadata-service/src/main/java/org/openmetadata/service/secrets/converter/SalesforceConnectorClassConverter.java

@@ -0,0 +1,43 @@
+/*
+ *  Copyright 2021 Collate
+ *  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 org.openmetadata.service.secrets.converter;
+
+import java.util.List;
+import org.openmetadata.schema.security.ssl.ValidateSSLClientConfig;
+import org.openmetadata.schema.services.connections.database.SalesforceConnection;
+import org.openmetadata.service.util.JsonUtils;
+
+/**
+ * Converter class to get an `Salesforce` object.
+ */
+public class SalesforceConnectorClassConverter extends ClassConverter {
+
+  private static final List<Class<?>> SSL_SOURCE_CLASS = List.of(ValidateSSLClientConfig.class);
+
+  public SalesforceConnectorClassConverter() {
+    super(SalesforceConnection.class);
+  }
+
+  @Override
+  public Object convert(Object object) {
+    SalesforceConnection salesforceConnection =
+        (SalesforceConnection) JsonUtils.convertValue(object, this.clazz);
+
+    // Convert the `sslConfig` field to the appropriate class
+    tryToConvert(salesforceConnection.getSslConfig(), SSL_SOURCE_CLASS)
+        .ifPresent(obj -> salesforceConnection.setSslConfig((ValidateSSLClientConfig) obj));
+
+    return salesforceConnection;
+  }
+}

openmetadata-service/src/test/java/org/openmetadata/service/secrets/converter/ClassConverterFactoryTest.java

@@ -17,11 +17,12 @@ import org.openmetadata.schema.services.connections.dashboard.SupersetConnection
 import org.openmetadata.schema.services.connections.dashboard.TableauConnection;
 import org.openmetadata.schema.services.connections.database.BigQueryConnection;
 import org.openmetadata.schema.services.connections.database.DatalakeConnection;
+import org.openmetadata.schema.services.connections.database.IcebergConnection;
 import org.openmetadata.schema.services.connections.database.MysqlConnection;
 import org.openmetadata.schema.services.connections.database.PostgresConnection;
+import org.openmetadata.schema.services.connections.database.SalesforceConnection;
 import org.openmetadata.schema.services.connections.database.TrinoConnection;
 import org.openmetadata.schema.services.connections.database.datalake.GCSConfig;
-import org.openmetadata.schema.services.connections.metadata.OpenMetadataConnection;
 import org.openmetadata.schema.services.connections.pipeline.AirflowConnection;
 import org.openmetadata.schema.services.connections.search.ElasticSearchConnection;
 import org.openmetadata.schema.services.connections.storage.GCSConnection;
@@ -42,14 +43,15 @@ public class ClassConverterFactoryTest {
         GCSConnection.class,
         ElasticSearchConnection.class,
         LookerConnection.class,
-        OpenMetadataConnection.class,
         SSOAuthMechanism.class,
         SupersetConnection.class,
         GCPCredentials.class,
         TableauConnection.class,
         TestServiceConnectionRequest.class,
         TrinoConnection.class,
-        Workflow.class
+        Workflow.class,
+        SalesforceConnection.class,
+        IcebergConnection.class,
       })
   void testClassConverterIsSet(Class<?> clazz) {
     assertFalse(
@@ -58,6 +60,6 @@ public class ClassConverterFactoryTest {
 
   @Test
   void testClassConvertedMapIsNotModified() {
-    assertEquals(20, ClassConverterFactory.getConverterMap().size());
+    assertEquals(26, ClassConverterFactory.getConverterMap().size());
   }
 }

openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/salesforceConnection.json

@@ -59,6 +59,11 @@
       "type": "string",
       "default": "login"
     },
+    "sslConfig": {
+      "title": "SSL Configuration",
+      "description": "SSL Configuration details.",
+      "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/sslConfig"
+    },
     "connectionOptions": {
       "title": "Connection Options",
       "$ref": "../connectionBasicType.json#/definitions/connectionOptions"

openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Salesforce.md

@@ -1,3 +1,4 @@
+
 # Salesforce
 
 In this section, we provide guides and references to use the Salesforce connector.
@@ -67,6 +68,13 @@ By default, the domain `login` is used for accessing Salesforce.
 
 $$
 
+
+$$section
+### SSL CA $(id="caCertificate")
+The CA certificate used for SSL validation to connect to Salesforce.
+$$
+
+
 $$section
 ### Connection Options $(id="connectionOptions")