Snowflake destination: support key pair authentication

airbyte-integrations/bases/base-normalization/Dockerfile

@@ -28,5 +28,5 @@ WORKDIR /airbyte
 ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh"
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
 
-LABEL io.airbyte.version=0.2.6
+LABEL io.airbyte.version=0.2.7
 LABEL io.airbyte.name=airbyte/normalization

airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py

@@ -219,6 +219,13 @@ def transform_snowflake(config: Dict[str, Any]):
             dbt_config["oauth_client_id"] = credentials["client_id"]
             dbt_config["oauth_client_secret"] = credentials["client_secret"]
             dbt_config["token"] = credentials["refresh_token"]
+        elif credentials.get("private_key"):
+            f = open("private_key_path.txt", "w")
+            f.write(credentials["private_key"])
+            f.close()
+            dbt_config["private_key_path"] = "private_key_path.txt"
+            if credentials.get("passphrase"):
+                dbt_config["private_key_passphrase"] = credentials["passphrase"]
         elif credentials.get("password"):
             dbt_config["password"] = credentials["password"]
         else:

airbyte-integrations/connectors/destination-snowflake/src/main/java/io/airbyte/integrations/destination/snowflake/SnowflakeDatabase.java

@@ -12,6 +12,7 @@
 import io.airbyte.db.jdbc.DefaultJdbcDatabase;
 import io.airbyte.db.jdbc.JdbcDatabase;
 import java.io.IOException;
+import java.io.PrintWriter;
 import java.net.URI;
 import java.net.URLEncoder;
 import java.net.http.HttpClient;
@@ -48,6 +49,9 @@ public class SnowflakeDatabase {
       .version(HttpClient.Version.HTTP_2)
       .connectTimeout(Duration.ofSeconds(10))
       .build();
+  public static final String PRIVATE_KEY_FILE_NAME = "rsa_key.p8";
+  public static final String PRIVATE_KEY_FIELD_NAME = "private_key";
+  public static final String PASSPHRASE = "passphrase";
 
   public static HikariDataSource createDataSource(final JsonNode config) {
     final HikariDataSource dataSource = new HikariDataSource();
@@ -92,6 +96,15 @@ public static HikariDataSource createDataSource(final JsonNode config) {
       dataSource.setUsername(username);
       dataSource.setPassword(credentials.get("password").asText());
 
+    } else if (credentials != null && credentials.has(PRIVATE_KEY_FIELD_NAME)) {
+      LOGGER.debug("Login mode with key pair is used");
+      dataSource.setUsername(username);
+      final String privateKeyValue = credentials.get(PRIVATE_KEY_FIELD_NAME).asText();
+      createPrivateKeyFile(PRIVATE_KEY_FILE_NAME, privateKeyValue);
+      properties.put("private_key_file", PRIVATE_KEY_FILE_NAME);
+      if (credentials.has(PASSPHRASE)) {
+        properties.put("private_key_file_pwd", credentials.get(PASSPHRASE).asText());
+      }
     } else {
       LOGGER.warn(
           "Obsolete User/password login mode is used. Please re-create a connection to use the latest connector's version");
@@ -128,6 +141,14 @@ public static HikariDataSource createDataSource(final JsonNode config) {
     return dataSource;
   }
 
+  private static void createPrivateKeyFile(final String fileName, final String fileValue) {
+    try (final PrintWriter out = new PrintWriter(fileName, StandardCharsets.UTF_8)) {
+      out.print(fileValue);
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to create file for private key");
+    }
+  }
+
   private static String getAccessTokenUsingRefreshToken(final String hostName,
                                                         final String clientId,
                                                         final String clientSecret,

airbyte-integrations/connectors/destination-snowflake/src/main/resources/spec.json

@@ -62,8 +62,8 @@
         "type": "object",
         "oneOf": [
           {
-            "type": "object",
             "title": "OAuth2.0",
+            "type": "object",
             "order": 0,
             "required": ["access_token", "refresh_token"],
             "properties": {
@@ -100,11 +100,39 @@
               }
             }
           },
+          {
+            "title": "Key Pair Authentication",
+            "type": "object",
+            "order": 1,
+            "required": ["private_key"],
+            "properties": {
+              "auth_type": {
+                "type": "string",
+                "const": "Key Pair Authentication",
+                "enum": ["Key Pair Authentication"],
+                "default": "Key Pair Authentication",
+                "order": 0
+              },
+              "private_key": {
+                "type": "string",
+                "title": "Private Key",
+                "description": "RSA Private key to use for Snowflake connection. See the <a href=\"https://docs.airbyte.io/integrations/destinations/snowflake\">docs</a> for more information on how to obtain this key.",
+                "multiline": true,
+                "airbyte_secret": true
+              },
+              "passphrase": {
+                "type": "string",
+                "title": "Passphrase (Optional)",
+                "description": "Passphrase for private key",
+                "airbyte_secret": true
+              }
+            }
+          },
           {
             "title": "Username and Password",
             "type": "object",
             "required": ["password"],
-            "order": 1,
+            "order": 2,
             "properties": {
               "password": {
                 "description": "Enter the password associated with the username.",

airbyte-integrations/connectors/destination-snowflake/src/test-integration/java/io/airbyte/integrations/destination/snowflake/SnowflakeInsertDestinationAcceptanceTest.java

@@ -23,6 +23,7 @@
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
 import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import io.airbyte.protocol.models.AirbyteCatalog;
+import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.CatalogHelpers;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
@@ -188,6 +189,13 @@ public void testBackwardCompatibilityAfterAddingOauth() {
     assertEquals(Status.SUCCEEDED, runCheckWithCatchedException(deprecatedStyleConfig));
   }
 
+  @Test
+  void testCheckWithKeyPairAuth() throws Exception {
+    final JsonNode credentialsJsonString = Jsons.deserialize(IOs.readFile(Path.of("secrets/config_key_pair.json")));
+    final AirbyteConnectionStatus check = new SnowflakeDestination().check(credentialsJsonString);
+    assertEquals(AirbyteConnectionStatus.Status.SUCCEEDED, check.getStatus());
+  }
+
   /**
    * This test is disabled because it is very slow, and should only be run manually for now.
    */

airbyte-integrations/connectors/destination-snowflake/src/test-integration/java/io/airbyte/integrations/destination/snowflake/SnowflakeInternalStagingDestinationAcceptanceTest.java

@@ -8,7 +8,18 @@
 import com.google.common.base.Preconditions;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.resources.MoreResources;
+import io.airbyte.integrations.standardtest.destination.DataArgumentsProvider;
+import io.airbyte.protocol.models.AirbyteCatalog;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
+import io.airbyte.protocol.models.CatalogHelpers;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ArgumentsSource;
 
 public class SnowflakeInternalStagingDestinationAcceptanceTest extends SnowflakeInsertDestinationAcceptanceTest {
 
@@ -19,4 +30,34 @@ public JsonNode getStaticConfig() {
     return internalStagingConfig;
   }
 
+  @ParameterizedTest
+  @ArgumentsSource(DataArgumentsProvider.class)
+  public void testSyncWithNormalization(final String messagesFilename, final String catalogFilename) throws Exception {
+    testSyncWithNormalizationWithKeyPairAuth(messagesFilename, catalogFilename, "secrets/config_key_pair.json");
+  }
+
+  @ParameterizedTest
+  @ArgumentsSource(DataArgumentsProvider.class)
+  public void testSyncWithNormalizationWithKeyPairEncrypt(final String messagesFilename, final String catalogFilename) throws Exception {
+    testSyncWithNormalizationWithKeyPairAuth(messagesFilename, catalogFilename, "secrets/config_key_pair_encrypted.json");
+  }
+
+  private void testSyncWithNormalizationWithKeyPairAuth(String messagesFilename, String catalogFilename, String configName) throws Exception {
+    if (!normalizationFromSpec()) {
+      return;
+    }
+
+    final AirbyteCatalog catalog = Jsons.deserialize(MoreResources.readResource(catalogFilename), AirbyteCatalog.class);
+    final ConfiguredAirbyteCatalog configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog);
+    final List<AirbyteMessage> messages = MoreResources.readResource(messagesFilename).lines()
+        .map(record -> Jsons.deserialize(record, AirbyteMessage.class)).collect(Collectors.toList());
+
+    final JsonNode config = Jsons.deserialize(IOs.readFile(Path.of(configName)));
+    runSyncAndVerifyStateOutput(config, messages, configuredCatalog, true);
+
+    final String defaultSchema = getDefaultSchema(config);
+    final List<AirbyteRecordMessage> actualMessages = retrieveNormalizedRecords(catalog, defaultSchema);
+    assertSameMessages(messages, actualMessages, true);
+  }
+
 }

airbyte-workers/src/main/java/io/airbyte/workers/normalization/NormalizationRunnerFactory.java

@@ -14,7 +14,7 @@
 public class NormalizationRunnerFactory {
 
   public static final String BASE_NORMALIZATION_IMAGE_NAME = "airbyte/normalization";
-  public static final String NORMALIZATION_VERSION = "0.2.6";
+  public static final String NORMALIZATION_VERSION = "0.2.7";
 
   static final Map<String, ImmutablePair<String, DefaultNormalizationRunner.DestinationType>> NORMALIZATION_MAPPING =
       ImmutableMap.<String, ImmutablePair<String, DefaultNormalizationRunner.DestinationType>>builder()

docs/integrations/destinations/snowflake.md

@@ -183,6 +183,31 @@ Field | Description |
 | [JDBC URL Params](https://docs.snowflake.com/en/user-guide/jdbc-parameters.html) (Optional) | Additional properties to pass to the JDBC URL string when connecting to the database formatted as `key=value` pairs separated by the symbol `&`. Example: `key1=value1&key2=value2&key3=value3` |
 
 
+### Key pair authentication
+    In order to configure key pair authentication you will need a private/public key pair.
+    If you do not have the key pair yet, you can generate one using openssl command line tool
+    Use this command in order to generate an unencrypted private key file:
+
+       `openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt`
+
+    Alternatively, use this command to generate an encrypted private key file:
+
+      `openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -v1 PBE-SHA1-RC4-128 -out rsa_key.p8`
+
+    Once you have your private key, you need to generate a matching public key.
+    You can do so with the following command:
+
+      `openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub`
+
+    Finally, you need to add the public key to your Snowflake user account.
+    You can do so with the following SQL command in Snowflake:
+
+      `alter user <user_name> set rsa_public_key=<public_key_value>;`
+
+    and replace <user_name> with your user name and <public_key_value> with your public key.
+
+
+
 To use AWS S3 as the cloud storage, enter the information for the S3 bucket you created in Step 2:
 
 | Field | Description |
@@ -249,11 +274,12 @@ Now that you have set up the Snowflake destination connector, check out the foll
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                             |
 |:--------|:-----------|:-----------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.4.32  | 2022-07-14 | [\#14388](https://github.com/airbytehq/airbyte/pull/14388) | Add support for key pair authentication  |
 | 0.4.31  | 2022-07-07 | [\#13729](https://github.com/airbytehq/airbyte/pull/13729) | Improve configuration field description  |
 | 0.4.30  | 2022-06-24 | [\#14114](https://github.com/airbytehq/airbyte/pull/14114) | Remove "additionalProperties": false from specs for connectors with staging  |
 | 0.4.29  | 2022-06-17 | [\#13753](https://github.com/airbytehq/airbyte/pull/13753) | Deprecate and remove PART_SIZE_MB fields from connectors based on StreamTransferManager  |
 | 0.4.28  | 2022-05-18 | [\#12952](https://github.com/airbytehq/airbyte/pull/12952) | Apply buffering strategy on GCS staging |
-| 0.4.27  | 2022-05-17 | [12820](https://github.com/airbytehq/airbyte/pull/12820) | Improved 'check' operation performance |
+| 0.4.27  | 2022-05-17 | [\#12820](https://github.com/airbytehq/airbyte/pull/12820) | Improved 'check' operation performance |
 | 0.4.26  | 2022-05-12 | [\#12805](https://github.com/airbytehq/airbyte/pull/12805) | Updated to latest base-java to emit AirbyteTraceMessages on error.                                                                                  |
 | 0.4.25  | 2022-05-03 | [\#12452](https://github.com/airbytehq/airbyte/pull/12452) | Add support for encrypted staging on S3; fix the purge_staging_files option                                                                         |
 | 0.4.24  | 2022-03-24 | [\#11093](https://github.com/airbytehq/airbyte/pull/11093) | Added OAuth support (Compatible with Airbyte Version 0.35.60+)                                                                                      |