🎉 Oracle Source: Add encryption options

airbyte-integrations/connectors/source-oracle/Dockerfile

@@ -6,7 +6,6 @@ ENV APPLICATION source-oracle
 ENV TZ UTC
 
 COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
-
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
 LABEL io.airbyte.version=0.3.5

airbyte-integrations/connectors/source-oracle/src/main/java/io/airbyte/integrations/source/oracle/OracleSource.java

@@ -15,11 +15,13 @@
 import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
 import io.airbyte.integrations.source.relationaldb.TableInfo;
 import io.airbyte.protocol.models.CommonField;
+
+import java.io.*;
 import java.sql.JDBCType;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Locale;
-import java.util.Set;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.RandomStringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -31,18 +33,33 @@ public class OracleSource extends AbstractJdbcSource implements Source {
 
   private List<String> schemas;
 
+  private static final String KEY_STORE_FILE_PATH = "clientkeystore.jks";
+  private static final String KEY_STORE_PASS = RandomStringUtils.randomAlphanumeric(8);
+
+  enum Protocol{
+    TCP,TCPS
+  }
+
   public OracleSource() {
     super(DRIVER_CLASS, new OracleJdbcStreamingQueryConfiguration());
   }
 
   @Override
   public JsonNode toDatabaseConfig(JsonNode config) {
-    final ImmutableMap.Builder<Object, Object> configBuilder = ImmutableMap.builder()
-        .put("username", config.get("username").asText())
-        .put("jdbc_url", String.format("jdbc:oracle:thin:@//%s:%s/%s",
+    List<String> additionalParameters = new ArrayList<>();
+
+    JsonNode encryption = config.get("encryption");
+    Protocol protocol = obtainConnectionProtocol(encryption, additionalParameters);
+    final String connectionString = String.format(
+            "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=%s)(HOST=%s)(PORT=%s))(CONNECT_DATA=(SID=%s)))",
+            protocol,
             config.get("host").asText(),
             config.get("port").asText(),
-            config.get("sid").asText()));
+            config.get("sid").asText());
+
+    final ImmutableMap.Builder<Object, Object> configBuilder = ImmutableMap.builder()
+        .put("username", config.get("username").asText())
+        .put("jdbc_url", connectionString);
 
     if (config.has("password")) {
       configBuilder.put("password", config.get("password").asText());
@@ -56,10 +73,59 @@ public JsonNode toDatabaseConfig(JsonNode config) {
         schemas.add(schema.asText());
       }
     }
+    if (!additionalParameters.isEmpty()) {
+      String connectionParams = String.join(";", additionalParameters);
+      configBuilder.put("connection_properties",connectionParams);
+    }
 
     return Jsons.jsonNode(configBuilder.build());
   }
 
+  private Protocol obtainConnectionProtocol(JsonNode encryption, List<String> additionalParameters) {
+    String encryptionMethod = encryption.get("encryption_method").asText();
+    switch (encryptionMethod) {
+      case "unencrypted" -> {
+        return Protocol.TCP;
+      }
+      case "client_nne" -> {
+        String algorithm = encryption.get("encryption_algorithm").asText();
+        additionalParameters.add("oracle.net.encryption_client=REQUIRED");
+        additionalParameters.add("oracle.net.encryption_types_client=( "+algorithm+" )");
+        return Protocol.TCP;
+      }
+      case "encrypted_verify_certificate" -> {
+        try {
+          convertAndImportCertificate(encryption.get("ssl_certificate").asText());
+        } catch (IOException | InterruptedException e) {
+          throw new RuntimeException("Failed to import certificate into Java Keystore");
+        }
+        additionalParameters.add("javax.net.ssl.trustStore="+KEY_STORE_FILE_PATH);
+        additionalParameters.add("javax.net.ssl.trustStoreType=JKS");
+        additionalParameters.add("javax.net.ssl.trustStorePassword="+KEY_STORE_PASS);
+        return Protocol.TCPS;
+      }
+    }
+    throw new RuntimeException("Failed to obtain connection protocol from config"+encryption.asText());
+  }
+
+  private static void convertAndImportCertificate(String certificate) throws IOException, InterruptedException {
+    Runtime run = Runtime.getRuntime();
+    try (PrintWriter out = new PrintWriter("certificate.pem")) {
+      out.print(certificate);
+    }
+    runProcess("openssl x509 -outform der -in certificate.pem -out certificate.der", run);
+    runProcess("keytool -import -alias rds-root -keystore "+KEY_STORE_FILE_PATH+" -file certificate.der -storepass "+KEY_STORE_PASS+" -noprompt", run);
+  }
+
+  private static void runProcess(String cmd, Runtime run) throws IOException, InterruptedException {
+    Process pr = run.exec(cmd);
+    if (!pr.waitFor(10, TimeUnit.SECONDS)){
+      pr.destroy();
+      throw new RuntimeException("Timeout while executing: "+ cmd);
+    };
+  }
+
+
   @Override
   public List<TableInfo<CommonField<JDBCType>>> discoverInternal(JdbcDatabase database) throws Exception {
     List<TableInfo<CommonField<JDBCType>>> internals = new ArrayList<>();

airbyte-integrations/connectors/source-oracle/src/main/resources/spec.json

@@ -4,7 +4,13 @@
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "Oracle Source Spec",
     "type": "object",
-    "required": ["host", "port", "sid", "username"],
+    "required": [
+      "host",
+      "port",
+      "sid",
+      "username",
+      "encryption"
+    ],
     "additionalProperties": false,
     "properties": {
       "host": {
@@ -14,12 +20,11 @@
       },
       "port": {
         "title": "Port",
-        "description": "Port of the database.",
+        "description": "Port of the database.\nOracle Corporations recommends the following port numbers:\n1521 - Default listening port for client connections to the listener. \n2484 - Recommended and officially registered listening port for client connections to the listener using TCP/IP with SSL",
         "type": "integer",
         "minimum": 0,
         "maximum": 65536,
-        "default": 1521,
-        "examples": ["1521"]
+        "default": 1521
       },
       "sid": {
         "title": "SID (Oracle System Identifier)",
@@ -45,7 +50,89 @@
         },
         "minItems": 1,
         "uniqueItems": true
+      },
+      "encryption": {
+        "title": "Encryption",
+        "type": "object",
+        "description": "Encryption method to use when communicating with the database",
+        "order": 6,
+        "oneOf": [
+          {
+            "title": "Unencrypted",
+            "additionalProperties": false,
+            "description": "Data transfer will not be encrypted.",
+            "required": [
+              "encryption_method"
+            ],
+            "properties": {
+              "encryption_method": {
+                "type": "string",
+                "const": "unencrypted",
+                "enum": [
+                  "unencrypted"
+                ],
+                "default": "unencrypted"
+              }
+            }
+          },
+          {
+            "title": "Native Network Ecryption (NNE)",
+            "additionalProperties": false,
+            "description": "Native network encryption gives you the ability to encrypt database connections, without the configuration overhead of TCP/IP and SSL/TLS and without the need to open and listen on different ports.",
+            "required": [
+              "encryption_method"
+            ],
+            "properties": {
+              "encryption_method": {
+                "type": "string",
+                "const": "client_nne",
+                "enum": [
+                  "client_nne"
+                ],
+                "default": "client_nne"
+              },
+                "encryption_algorithm": {
+                  "type": "string",
+                  "description": "This parameter defines the encryption algorithm to be used",
+                  "title": "Encryption Algorithm",
+                  "default": "AES256",
+                  "enum": [
+                    "AES256",
+                    "RC4_56",
+                    "3DES168"
+                  ]
+                }
+            }
+          },
+          {
+            "title": "TLS Encrypted (verify certificate)",
+            "additionalProperties": false,
+            "description": "Verify and use the cert provided by the server.",
+            "required": [
+              "encryption_method",
+              "ssl_certificate"
+            ],
+            "properties": {
+              "encryption_method": {
+                "type": "string",
+                "const": "encrypted_verify_certificate",
+                "enum": [
+                  "encrypted_verify_certificate"
+                ],
+                "default": "encrypted_verify_certificate"
+              },
+              "ssl_certificate": {
+                "title": "SSL PEM file",
+                "description": "Privacy Enhanced Mail (PEM) files are concatenated certificate containers frequently used in certificate installations",
+                "type": "string",
+                "airbyte_secret": true,
+                "multiline": true,
+                "order": 4
+              }
+            }
+          }
+        ]
       }
     }
   }
-}
+}