🎉 Destination postgres: Add SSL certificates and update normalization (#14743)

* added ssl certificates for postgres source

* added command for remove client private key after transformation to encrypted key with .pk8 extension

* added connection with CA and client certificates for postgres destination

* updated code style

* moved common methods to the common class

* moved common methods to the common class

* fixed remarks

* updated postgres source tests

* added minor changes to spec and added fixes to password mechanism

* updated postgres source tests

* updated strict-encrypt postgres source and destination and added tests for SSL certificates for all postgres connectors

* fixed check style

* updated documentation and versions of connectors

* updated ordrs in test spec

* fixed minor remarks in specs and expected_specs

* fixed minor remarks in specs and expected_specs

* fixed Dockerfile

* fixed remarks

* fixed remarks

* fixed remarks

* fixed remarks

* fixed remarks

* fixed code style

* fixed connectors version in definition file

* updated postgres destination normalization

* updated postgres destination tests

* fixed code style for postgres source and destination

* pulled master changes

* removed allow mode for destination-postgres-strect-encrypt

* updated connectors version

* fixed custom DBT transformation and enabled test for it

* updated normalization version

* updated keystore password generation method

* auto-bump connector version [ci skip]

Co-authored-by: Octavia Squidington III <[email protected]>

Author: 2 people

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

@@ -203,7 +203,7 @@
 - name: Postgres
   destinationDefinitionId: 25c5221d-dce2-4163-ade9-739ef790f503
   dockerRepository: airbyte/destination-postgres
-  dockerImageTag: 0.3.21
+  dockerImageTag: 0.3.22
   documentationUrl: https://docs.airbyte.io/integrations/destinations/postgres
   icon: postgresql.svg
   releaseStage: alpha

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

@@ -3302,7 +3302,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-postgres:0.3.21"
+- dockerImage: "airbyte/destination-postgres:0.3.22"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/postgres"
     connectionSpecification:
@@ -3360,17 +3360,161 @@
           order: 5
         ssl:
           title: "SSL Connection"
-          description: "Encrypt data using SSL."
+          description: "Encrypt data using SSL. When activating SSL, please select\
+            \ one of the connection modes."
           type: "boolean"
           default: false
           order: 6
+        ssl_mode:
+          title: "SSL modes"
+          description: "SSL connection modes. \n <b>disable</b> - Chose this mode\
+            \ to disable encryption of communication between Airbyte and destination\
+            \ database\n <b>allow</b> - Chose this mode to enable encryption only\
+            \ when required by the source database\n <b>prefer</b> - Chose this mode\
+            \ to allow unencrypted connection only if the source database does not\
+            \ support encryption\n <b>require</b> - Chose this mode to always require\
+            \ encryption. If the source database server does not support encryption,\
+            \ connection will fail\n  <b>verify-ca</b> - Chose this mode to always\
+            \ require encryption and to verify that the source database server has\
+            \ a valid SSL certificate\n  <b>verify-full</b> - This is the most secure\
+            \ mode. Chose this mode to always require encryption and to verify the\
+            \ identity of the source database server\n See more information - <a href=\"\
+            https://jdbc.postgresql.org/documentation/head/ssl-client.html\"> in the\
+            \ docs</a>."
+          type: "object"
+          order: 7
+          oneOf:
+          - title: "disable"
+            additionalProperties: false
+            description: "Disable SSL."
+            required:
+            - "mode"
+            properties:
+              mode:
+                type: "string"
+                const: "disable"
+                enum:
+                - "disable"
+                default: "disable"
+                order: 0
+          - title: "allow"
+            additionalProperties: false
+            description: "Allow SSL mode."
+            required:
+            - "mode"
+            properties:
+              mode:
+                type: "string"
+                const: "allow"
+                enum:
+                - "allow"
+                default: "allow"
+                order: 0
+          - title: "prefer"
+            additionalProperties: false
+            description: "Prefer SSL mode."
+            required:
+            - "mode"
+            properties:
+              mode:
+                type: "string"
+                const: "prefer"
+                enum:
+                - "prefer"
+                default: "prefer"
+                order: 0
+          - title: "require"
+            additionalProperties: false
+            description: "Require SSL mode."
+            required:
+            - "mode"
+            properties:
+              mode:
+                type: "string"
+                const: "require"
+                enum:
+                - "require"
+                default: "require"
+                order: 0
+          - title: "verify-ca"
+            additionalProperties: false
+            description: "Verify-ca SSL mode."
+            required:
+            - "mode"
+            - "ca_certificate"
+            properties:
+              mode:
+                type: "string"
+                const: "verify-ca"
+                enum:
+                - "verify-ca"
+                default: "verify-ca"
+                order: 0
+              ca_certificate:
+                type: "string"
+                title: "CA certificate"
+                description: "CA certificate"
+                airbyte_secret: true
+                multiline: true
+                order: 1
+              client_key_password:
+                type: "string"
+                title: "Client key password (Optional)"
+                description: "Password for keystorage. This field is optional. If\
+                  \ you do not add it - the password will be generated automatically."
+                airbyte_secret: true
+                order: 4
+          - title: "verify-full"
+            additionalProperties: false
+            description: "Verify-full SSL mode."
+            required:
+            - "mode"
+            - "ca_certificate"
+            - "client_certificate"
+            - "client_key"
+            properties:
+              mode:
+                type: "string"
+                const: "verify-full"
+                enum:
+                - "verify-full"
+                default: "verify-full"
+                order: 0
+              ca_certificate:
+                type: "string"
+                title: "CA certificate"
+                description: "CA certificate"
+                airbyte_secret: true
+                multiline: true
+                order: 1
+              client_certificate:
+                type: "string"
+                title: "Client certificate"
+                description: "Client certificate"
+                airbyte_secret: true
+                multiline: true
+                order: 2
+              client_key:
+                type: "string"
+                title: "Client key"
+                description: "Client key"
+                airbyte_secret: true
+                multiline: true
+                order: 3
+              client_key_password:
+                type: "string"
+                title: "Client key password (Optional)"
+                description: "Password for keystorage. This field is optional. If\
+                  \ you do not add it - the password will be generated automatically."
+                airbyte_secret: true
+                order: 4
         jdbc_url_params:
           description: "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)."
           title: "JDBC URL Params"
           type: "string"
-          order: 7
+          order: 8
         tunnel_method:
           type: "object"
           title: "SSH Tunnel Method"

airbyte-db/db-lib/src/main/java/io/airbyte/db/PostgresUtils.java

@@ -33,7 +33,7 @@ public static Certificate getCertificate(final PostgreSQLContainer<?> container)
     container.execInContainer("su", "-c", "psql -U test -c \"ALTER USER postgres WITH SUPERUSER;\"");
 
     container.execInContainer("su", "-c", "openssl ecparam -name prime256v1 -genkey -noout -out ca.key");
-    container.execInContainer("su", "-c", "openssl req -new -x509 -sha256 -key ca.key -out ca.crt -subj \"/CN=localhost\"");
+    container.execInContainer("su", "-c", "openssl req -new -x509 -sha256 -key ca.key -out ca.crt -subj \"/CN=127.0.0.1\"");
     container.execInContainer("su", "-c", "openssl ecparam -name prime256v1 -genkey -noout -out server.key");
     container.execInContainer("su", "-c", "openssl req -new -sha256 -key server.key -out server.csr -subj \"/CN=localhost\"");
     container.execInContainer("su", "-c",

airbyte-db/db-lib/src/main/java/io/airbyte/db/jdbc/JdbcUtils.java

@@ -29,6 +29,7 @@ public class JdbcUtils {
   // NOTE: this is the plural version of SCHEMA_KEY
   public static final String SCHEMAS_KEY = "schemas";
   public static final String SSL_KEY = "ssl";
+  public static final String SSL_MODE_KEY = "ssl_mode";
   public static final String TLS_KEY = "tls";
   public static final String USERNAME_KEY = "username";
   private static final JdbcSourceOperations defaultSourceOperations = new JdbcSourceOperations();

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/util/PostgresSslConnectionUtils.java

@@ -5,6 +5,10 @@
 package io.airbyte.integrations.util;
 
 import com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.nio.charset.StandardCharsets;
@@ -18,7 +22,6 @@
 public class PostgresSslConnectionUtils {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PostgresSslConnectionUtils.class);
-  private static final String KEY_STORE_PASS = RandomStringUtils.randomAlphanumeric(10);
   private static final String CA_CERTIFICATE = "ca.crt";
   private static final String CLIENT_CERTIFICATE = "client.crt";
   private static final String CLIENT_KEY = "client.key";
@@ -37,17 +40,14 @@ public class PostgresSslConnectionUtils {
   public static final String VERIFY_FULL = "verify-full";
   public static final String DISABLE = "disable";
   public static final String TRUE_STRING_VALUE = "true";
+  public static final String ENCRYPT_FILE_NAME = "encrypt";
   public static final String FACTORY_VALUE = "org.postgresql.ssl.DefaultJavaSSLFactory";
 
   public static Map<String, String> obtainConnectionOptions(final JsonNode encryption) {
     final Map<String, String> additionalParameters = new HashMap<>();
     if (!encryption.isNull()) {
       final var method = encryption.get(PARAM_MODE).asText();
-      String sslPassword = encryption.has(PARAM_CLIENT_KEY_PASSWORD) ? encryption.get(PARAM_CLIENT_KEY_PASSWORD).asText() : "";
-      var keyStorePassword = KEY_STORE_PASS;
-      if (!sslPassword.isEmpty()) {
-        keyStorePassword = sslPassword;
-      }
+      var keyStorePassword = checkOrCreatePassword(encryption);
       switch (method) {
         case VERIFY_CA -> {
           additionalParameters.putAll(obtainConnectionCaOptions(encryption, method, keyStorePassword));
@@ -64,6 +64,37 @@ public static Map<String, String> obtainConnectionOptions(final JsonNode encrypt
     return additionalParameters;
   }
 
+  private static String checkOrCreatePassword(final JsonNode encryption) {
+    String sslPassword = encryption.has(PARAM_CLIENT_KEY_PASSWORD) ? encryption.get(PARAM_CLIENT_KEY_PASSWORD).asText() : "";
+    var keyStorePassword = RandomStringUtils.randomAlphanumeric(10);
+    if (sslPassword.isEmpty()) {
+      var file = new File(ENCRYPT_FILE_NAME);
+      if (file.exists()) {
+        keyStorePassword = readFile(file);
+      } else {
+        try {
+          createCertificateFile(ENCRYPT_FILE_NAME, keyStorePassword);
+        } catch (final IOException e) {
+          throw new RuntimeException("Failed to create encryption file ");
+        }
+      }
+    } else {
+      keyStorePassword = sslPassword;
+    }
+    return keyStorePassword;
+  }
+
+  private static String readFile(final File file) {
+    try {
+      BufferedReader reader = new BufferedReader(new FileReader(file));
+      String currentLine = reader.readLine();
+      reader.close();
+      return currentLine;
+    } catch (final IOException e) {
+      throw new RuntimeException("Failed to read file with encryption");
+    }
+  }
+
   private static Map<String, String> obtainConnectionFullOptions(final JsonNode encryption,
                                                                  final String method,
                                                                  final String clientKeyPassword) {

airbyte-integrations/connectors/destination-postgres-strict-encrypt/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-postgres-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.21
+LABEL io.airbyte.version=0.3.22
 LABEL io.airbyte.name=airbyte/destination-postgres-strict-encrypt

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestinationStrictEncrypt.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.destination.postgres;
 
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.db.jdbc.JdbcUtils;
@@ -17,6 +18,8 @@
 public class PostgresDestinationStrictEncrypt extends SpecModifyingDestination implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PostgresDestinationStrictEncrypt.class);
+  private static final String PROPERTIES = "properties";
+  private static final String ONE_OF_PROPERTY = "oneOf";
 
   public PostgresDestinationStrictEncrypt() {
     super(PostgresDestination.sshWrappedDestination());
@@ -25,7 +28,14 @@ public PostgresDestinationStrictEncrypt() {
   @Override
   public ConnectorSpecification modifySpec(final ConnectorSpecification originalSpec) {
     final ConnectorSpecification spec = Jsons.clone(originalSpec);
-    ((ObjectNode) spec.getConnectionSpecification().get("properties")).remove(JdbcUtils.SSL_KEY);
+    ((ObjectNode) spec.getConnectionSpecification().get(PROPERTIES)).remove(JdbcUtils.SSL_KEY);
+    ArrayNode modifiedSslModes = spec.getConnectionSpecification().get(PROPERTIES).get(JdbcUtils.SSL_MODE_KEY).get(ONE_OF_PROPERTY).deepCopy();
+    // Assume that the first item is the "allow" option; remove it
+    modifiedSslModes.remove(1);
+    // Assume that the first item is the "disable" option; remove it
+    modifiedSslModes.remove(0);
+    ((ObjectNode) spec.getConnectionSpecification().get(PROPERTIES).get(JdbcUtils.SSL_MODE_KEY)).remove(ONE_OF_PROPERTY);
+    ((ObjectNode) spec.getConnectionSpecification().get(PROPERTIES).get(JdbcUtils.SSL_MODE_KEY)).put(ONE_OF_PROPERTY, modifiedSslModes);
     return spec;
   }
 

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/postgres/PostgresDestinationStrictEncryptAcceptanceTest.java

@@ -4,10 +4,13 @@
 
 package io.airbyte.integrations.destination.postgres;
 
+import static io.airbyte.db.PostgresUtils.getCertificate;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.db.Database;
+import io.airbyte.db.PostgresUtils;
 import io.airbyte.db.factory.DSLContextFactory;
 import io.airbyte.db.factory.DatabaseDriver;
 import io.airbyte.db.jdbc.JdbcUtils;
@@ -29,6 +32,9 @@ public class PostgresDestinationStrictEncryptAcceptanceTest extends DestinationA
   private PostgreSQLContainer<?> db;
   private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
 
+  protected static final String PASSWORD = "Passw0rd";
+  protected static PostgresUtils.Certificate certs;
+
   @Override
   protected String getImageName() {
     return "airbyte/destination-postgres-strict-encrypt:dev";
@@ -43,6 +49,13 @@ protected JsonNode getConfig() {
         .put(JdbcUtils.SCHEMA_KEY, "public")
         .put(JdbcUtils.PORT_KEY, db.getFirstMappedPort())
         .put(JdbcUtils.DATABASE_KEY, db.getDatabaseName())
+        .put(JdbcUtils.SSL_MODE_KEY, ImmutableMap.builder()
+                .put("mode", "verify-full")
+                .put("ca_certificate", certs.getCaCertificate())
+                .put("client_certificate", certs.getClientCertificate())
+                .put("client_key", certs.getClientKey())
+                .put("client_key_password", PASSWORD)
+                .build())
         .build());
   }
 
@@ -131,10 +144,11 @@ private List<JsonNode> retrieveRecordsFromTable(final String tableName, final St
   }
 
   @Override
-  protected void setup(final TestDestinationEnv testEnv) {
-    db = new PostgreSQLContainer<>(DockerImageName.parse("marcosmarxm/postgres-ssl:dev").asCompatibleSubstituteFor("postgres"))
-        .withCommand("postgres -c ssl=on -c ssl_cert_file=/var/lib/postgresql/server.crt -c ssl_key_file=/var/lib/postgresql/server.key");
+  protected void setup(final TestDestinationEnv testEnv) throws Exception {
+    db = new PostgreSQLContainer<>(DockerImageName.parse("postgres:bullseye")
+        .asCompatibleSubstituteFor("postgres"));
     db.start();
+    certs = getCertificate(db);
   }
 
   @Override