fix encrypted rsa key issue

Author: binglihub

pom.xml

@@ -12,7 +12,7 @@
 
     <groupId>com.snowflake</groupId>
     <artifactId>snowflake-kafka-connector</artifactId>
-    <version>0.4.0</version>
+    <version>0.5.0</version>
     <packaging>jar</packaging>
     <name>Snowflake Kafka Connector</name>
     <description>Snowflake Kafka Connect Sink Connector</description>
@@ -173,6 +173,7 @@
 
             <plugin>
                 <artifactId>maven-assembly-plugin</artifactId>
+                <version>3.1.1</version>
                 <executions>
                     <execution>
                         <phase>package</phase>
@@ -264,10 +265,15 @@
         <dependency>
             <groupId>org.bouncycastle</groupId>
             <artifactId>bcprov-ext-jdk15on</artifactId>
-            <version>1.60</version>
-            <scope>test</scope>
+            <version>1.61</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcpkix-jdk15on</artifactId>
+            <version>1.61</version>
+            <scope>provided</scope>
         </dependency>
-
 
         <!-- https://mvnrepository.com/artifact/org.apache.kafka/connect-api -->
         <dependency>

src/main/java/com/snowflake/kafka/connector/Utils.java

@@ -37,7 +37,7 @@ public class Utils
 {
 
   //Connector version, change every release
-  static final String VERSION = "0.4.0";
+  static final String VERSION = "0.5.0";
 
   //connector parameter list
   public static final String NAME = "name";

src/main/java/com/snowflake/kafka/connector/internal/InternalUtils.java

@@ -3,18 +3,20 @@
 import com.snowflake.kafka.connector.Utils;
 import net.snowflake.client.jdbc.internal.apache.commons.codec.binary.Base64;
 import net.snowflake.ingest.connection.IngestStatus;
+import org.bouncycastle.asn1.pkcs.PrivateKeyInfo;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.bouncycastle.openssl.PEMParser;
+import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
+import org.bouncycastle.openssl.jcajce.JceOpenSSLPKCS8DecryptorProviderBuilder;
+import org.bouncycastle.operator.InputDecryptorProvider;
+import org.bouncycastle.pkcs.PKCS8EncryptedPrivateKeyInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.crypto.Cipher;
-import javax.crypto.EncryptedPrivateKeyInfo;
-import javax.crypto.SecretKeyFactory;
-import javax.crypto.spec.PBEKeySpec;
-import java.security.AlgorithmParameters;
-import java.security.Key;
+import java.io.StringReader;
 import java.security.KeyFactory;
 import java.security.PrivateKey;
-import java.security.spec.KeySpec;
+import java.security.Security;
 import java.security.spec.PKCS8EncodedKeySpec;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -86,22 +88,28 @@ static PrivateKey parseEncryptedPrivateKey(String key, String passphrase)
     key = key.replaceAll("-+[A-Za-z ]+-+", "");
     key = key.replaceAll("\\s", "");
 
+    StringBuilder builder = new StringBuilder();
+    builder.append("-----BEGIN ENCRYPTED PRIVATE KEY-----");
+    for (int i = 0; i < key.length(); i++)
+    {
+      if(i%64==0)
+      {
+        builder.append("\n");
+      }
+      builder.append(key.charAt(i));
+    }
+    builder.append("\n-----END ENCRYPTED PRIVATE KEY-----");
+    key = builder.toString();
+    Security.addProvider(new BouncyCastleProvider());
     try
     {
-      byte[] code = Base64.decodeBase64(key);
-      EncryptedPrivateKeyInfo encryptPKInfo =
-        new EncryptedPrivateKeyInfo(code);
-
-      Cipher cipher = Cipher.getInstance(encryptPKInfo.getAlgName());
-      PBEKeySpec pbeKeySpec = new PBEKeySpec(passphrase.toCharArray());
-      SecretKeyFactory secretKeyFactory =
-        SecretKeyFactory.getInstance(encryptPKInfo.getAlgName());
-      Key pbeKey = secretKeyFactory.generateSecret(pbeKeySpec);
-      AlgorithmParameters algorithmParameters = encryptPKInfo.getAlgParameters();
-      cipher.init(Cipher.DECRYPT_MODE, pbeKey, algorithmParameters);
-      KeySpec keySpec = encryptPKInfo.getKeySpec(cipher);
-      KeyFactory kf = KeyFactory.getInstance("RSA");
-      return kf.generatePrivate(keySpec);
+      PEMParser pemParser = new PEMParser(new StringReader(key));
+      PKCS8EncryptedPrivateKeyInfo encryptedPrivateKeyInfo = (PKCS8EncryptedPrivateKeyInfo) pemParser.readObject();
+      pemParser.close();
+      InputDecryptorProvider pkcs8Prov = new JceOpenSSLPKCS8DecryptorProviderBuilder().build(passphrase.toCharArray());
+      JcaPEMKeyConverter converter = new JcaPEMKeyConverter().setProvider(BouncyCastleProvider.PROVIDER_NAME);
+      PrivateKeyInfo decryptedPrivateKeyInfo = encryptedPrivateKeyInfo.decryptPrivateKeyInfo(pkcs8Prov);
+      return converter.getPrivateKey(decryptedPrivateKeyInfo);
     }
     catch (Exception e)
     {
@@ -115,7 +123,7 @@ static PrivateKey parsePrivateKey(String key)
     key = key.replaceAll("-+[A-Za-z ]+-+", "");
     key = key.replaceAll("\\s", "");
 
-    java.security.Security.addProvider(new net.snowflake.client.jdbc.internal.org.bouncycastle.jce.provider.BouncyCastleProvider());
+    java.security.Security.addProvider(new BouncyCastleProvider());
     byte[] encoded = Base64.decodeBase64(key);
     try
     {

src/main/java/com/snowflake/kafka/connector/records/RecordService.java

@@ -108,7 +108,8 @@ public String processRecord(SinkRecord record)
     }
 
     //include String key
-    if (record.keySchema().toString().equals(Schema.STRING_SCHEMA.toString()))
+    if (record.keySchema().toString().equals(Schema.STRING_SCHEMA.toString())
+      && record.key() != null)
     {
       meta.put(KEY, record.key().toString());
     }

src/test/java/com/snowflake/kafka/connector/UtilsTest.java

@@ -5,7 +5,6 @@
 import org.junit.Test;
 
 import java.util.Map;
-import java.util.regex.Matcher;
 
 public class UtilsTest
 {