🎉 Oracle Source: Add encryption options

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b39a7370-74c3-45a6-ac3a-380d48520a83.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "b39a7370-74c3-45a6-ac3a-380d48520a83",
   "name": "Oracle DB",
   "dockerRepository": "airbyte/source-oracle",
-  "dockerImageTag": "0.3.5",
+  "dockerImageTag": "0.3.7",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/oracle"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -326,7 +326,7 @@
 - sourceDefinitionId: b39a7370-74c3-45a6-ac3a-380d48520a83
   name: Oracle DB
   dockerRepository: airbyte/source-oracle
-  dockerImageTag: 0.3.5
+  dockerImageTag: 0.3.7
   documentationUrl: https://docs.airbyte.io/integrations/sources/oracle
   sourceType: database
 - sourceDefinitionId: c8630570-086d-4a40-99ae-ea5b18673071

airbyte-integrations/connectors/source-oracle/Dockerfile

@@ -9,5 +9,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.3.5
+LABEL io.airbyte.version=0.3.7
 LABEL io.airbyte.name=airbyte/source-oracle

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

@@ -15,11 +15,15 @@
 import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
 import io.airbyte.integrations.source.relationaldb.TableInfo;
 import io.airbyte.protocol.models.CommonField;
+import java.io.IOException;
+import java.io.PrintWriter;
 import java.sql.JDBCType;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.RandomStringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -31,18 +35,35 @@ 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) {
+    List<String> additionalParameters = new ArrayList<>();
+
+    Protocol protocol = config.has("encryption")
+        ? obtainConnectionProtocol(config.get("encryption"), additionalParameters)
+        : Protocol.TCP;
+    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());
+
     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",
-            config.get("host").asText(),
-            config.get("port").asText(),
-            config.get("sid").asText()));
+        .put("jdbc_url", connectionString);
 
     if (config.has("password")) {
       configBuilder.put("password", config.get("password").asText());
@@ -56,10 +77,58 @@ 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(30, 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

@@ -14,12 +14,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,6 +44,71 @@
         },
         "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
+              }
+            }
+          }
+        ]
       }
     }
   }

airbyte-integrations/connectors/source-oracle/src/test-integration/java/io/airbyte/integrations/source/oracle/AbstractSshOracleSourceAcceptanceTest.java

@@ -103,7 +103,10 @@ public ImmutableMap.Builder<Object, Object> getBasicOracleDbConfigBuider(OracleC
         .put("password", db.getPassword())
         .put("port", db.getExposedPorts().get(0))
         .put("sid", db.getSid())
-        .put("schemas", List.of("JDBC_SPACE"));
+        .put("schemas", List.of("JDBC_SPACE"))
+        .put("encryption", Jsons.jsonNode(ImmutableMap.builder()
+            .put("encryption_method", "unencrypted")
+            .build()));
   }
 
   @Override

airbyte-integrations/connectors/source-oracle/src/test-integration/java/io/airbyte/integrations/source/oracle/OracleSourceAcceptanceTest.java

@@ -30,8 +30,8 @@ public class OracleSourceAcceptanceTest extends SourceAcceptanceTest {
   private static final String STREAM_NAME = "JDBC_SPACE.ID_AND_NAME";
   private static final String STREAM_NAME2 = "JDBC_SPACE.STARSHIPS";
 
-  private OracleContainer container;
-  private JsonNode config;
+  protected OracleContainer container;
+  protected JsonNode config;
 
   @Override
   protected void setupEnvironment(TestDestinationEnv environment) throws Exception {
@@ -45,6 +45,9 @@ protected void setupEnvironment(TestDestinationEnv environment) throws Exception
         .put("username", container.getUsername())
         .put("password", container.getPassword())
         .put("schemas", List.of("JDBC_SPACE"))
+        .put("encryption", Jsons.jsonNode(ImmutableMap.builder()
+            .put("encryption_method", "unencrypted")
+            .build()))
         .build());
 
     JdbcDatabase database = Databases.createJdbcDatabase(config.get("username").asText(),

airbyte-integrations/connectors/source-oracle/src/test-integration/java/io/airbyte/integrations/source/oracle/OracleSourceNneAcceptanceTest.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.source.oracle;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.db.Databases;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+public class OracleSourceNneAcceptanceTest extends OracleSourceAcceptanceTest {
+
+  @Test
+  public void testEncrytion() throws SQLException {
+    final JsonNode clone = Jsons.clone(getConfig());
+    ((ObjectNode) clone).put("encryption", Jsons.jsonNode(ImmutableMap.builder()
+        .put("encryption_method", "client_nne")
+        .put("encryption_algorithm", "3DES168")
+        .build()));
+
+    String algorithm = clone.get("encryption")
+        .get("encryption_algorithm").asText();
+
+    JdbcDatabase database = Databases.createJdbcDatabase(clone.get("username").asText(),
+        clone.get("password").asText(),
+        String.format("jdbc:oracle:thin:@//%s:%s/%s",
+            clone.get("host").asText(),
+            clone.get("port").asText(),
+            clone.get("sid").asText()),
+        "oracle.jdbc.driver.OracleDriver",
+        "oracle.net.encryption_client=REQUIRED;" +
+            "oracle.net.encryption_types_client=( "
+            + algorithm + " )");
+
+    String network_service_banner = "select network_service_banner from v$session_connect_info where sid in (select distinct sid from v$mystat)";
+    List<JsonNode> collect = database.query(network_service_banner).collect(Collectors.toList());
+
+    assertTrue(collect.get(2).get("NETWORK_SERVICE_BANNER").asText()
+        .contains("Oracle Advanced Security: " + algorithm + " encryption"));
+  }
+
+  @Test
+  public void testNoneEncrytion() throws SQLException {
+
+    JdbcDatabase database = Databases.createJdbcDatabase(config.get("username").asText(),
+        config.get("password").asText(),
+        String.format("jdbc:oracle:thin:@//%s:%s/%s",
+            config.get("host").asText(),
+            config.get("port").asText(),
+            config.get("sid").asText()),
+        "oracle.jdbc.driver.OracleDriver");
+
+    String network_service_banner = "select network_service_banner from v$session_connect_info where sid in (select distinct sid from v$mystat)";
+    List<JsonNode> collect = database.query(network_service_banner).collect(Collectors.toList());
+
+    assertTrue(collect.get(1).get("NETWORK_SERVICE_BANNER").asText()
+        .contains("Oracle Advanced Security: encryption"));
+  }
+
+  @Test
+  public void testCheckProtocol() throws SQLException {
+    final JsonNode clone = Jsons.clone(getConfig());
+    ((ObjectNode) clone).put("encryption", Jsons.jsonNode(ImmutableMap.builder()
+        .put("encryption_method", "client_nne")
+        .put("encryption_algorithm", "AES256")
+        .build()));
+
+    String algorithm = clone.get("encryption")
+        .get("encryption_algorithm").asText();
+
+    JdbcDatabase database = Databases.createJdbcDatabase(clone.get("username").asText(),
+        clone.get("password").asText(),
+        String.format("jdbc:oracle:thin:@//%s:%s/%s",
+            clone.get("host").asText(),
+            clone.get("port").asText(),
+            clone.get("sid").asText()),
+        "oracle.jdbc.driver.OracleDriver",
+        "oracle.net.encryption_client=REQUIRED;" +
+            "oracle.net.encryption_types_client=( "
+            + algorithm + " )");
+
+    String network_service_banner = "SELECT sys_context('USERENV', 'NETWORK_PROTOCOL') as network_protocol FROM dual";
+    List<JsonNode> collect = database.query(network_service_banner).collect(Collectors.toList());
+
+    assertEquals("tcp", collect.get(0).get("NETWORK_PROTOCOL").asText());
+  }
+
+}

docs/integrations/sources/oracle.md

@@ -123,10 +123,22 @@ If you can't find the data type you are looking for or have any problems feel fr
 
 If you do not see a type in this list, assume that it is coerced into a string. We are happy to take feedback on preferred mappings.
 
+## Encryption Options
+
+Airbite has the ability to connect to the Oracle source with 3 network connectivity options:
+
+1.`Unencrypted` the connection will be made using the TCP protocol. In this case, all data over the network will be transmitted in unencrypted form.
+2.`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.
+In this case, the *SQLNET.ENCRYPTION_CLIENT* option will always be set as *REQUIRED* by default: The client or server will only accept encrypted traffic,
+but the user has the opportunity to choose an `Encryption algorithm` according to the security policies he needs.
+3.`TLS Encrypted` (verify certificate) - if this option is selected, data transfer will be transfered using the TLS protocol, taking into account the handshake procedure and certificate verification.
+To use this option, insert the content of the certificate issued by the server into the `SSL PEM file` field
+
 ## Changelog
 
 | Version | Date       | Pull Request | Subject |
 | :------ | :--------  | :-----       | :------ |
+| 0.3.7   | 2021-10-01 | [6616](https://github.com/airbytehq/airbyte/pull/6616) | Added network encryption options |
 | 0.3.6   | 2021-09-30 | [6585](https://github.com/airbytehq/airbyte/pull/6585) | Improved SSH Tunnel key generation steps |
 | 0.3.5   | 2021-09-22 | [6356](https://github.com/airbytehq/airbyte/pull/6356) | Added option to connect to DB via SSH. |
 | 0.3.4   | 2021-09-01 | [6038](https://github.com/airbytehq/airbyte/pull/6038) | Remove automatic filtering of system schemas. |