🎉 Destination postgres: Add SSL certificates and update normalization

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

@@ -28,6 +28,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-normalization/normalization/transform_config/transform.py

@@ -8,6 +8,7 @@
 import os
 import pkgutil
 import socket
+import subprocess
 from typing import Any, Dict
 
 import yaml
@@ -167,12 +168,31 @@ def transform_postgres(config: Dict[str, Any]):
             "threads": 8,
         }
 
-        # if unset, we assume true.
-        if config.get("ssl", True):
-            config["sslmode"] = "require"
+        ssl = config.get("ssl")
+        if ssl:
+            ssl_mode = config.get("ssl_mode", "allow")
+            dbt_config["sslmode"] = ssl_mode.get("mode")
+            if ssl_mode["mode"] == "verify-ca":
+                TransformConfig.create_file("ca.crt", ssl_mode["ca_certificate"])
+                dbt_config["sslrootcert"] = "ca.crt"
+            elif ssl_mode["mode"] == "verify-full":
+                TransformConfig.create_file("ca.crt", ssl_mode["ca_certificate"])
+                TransformConfig.create_file("client.crt", ssl_mode["client_certificate"])
+                TransformConfig.create_file("client.key", ssl_mode["client_key"])
+                dbt_config["sslrootcert"] = "ca.crt"
+                dbt_config["sslcert"] = "client.crt"
+                subprocess.call("openssl pkcs8 -topk8 -inform PEM -in client.key -outform DER -out client.pk8 -nocrypt", shell=True)
+                subprocess.call("rm client.key", shell=True)
+                dbt_config["sslkey"] = "client.pk8"
 
         return dbt_config
 
+    @staticmethod
+    def create_file(name, content):
+        f = open(name, "x")
+        f.write(content)
+        f.close()
+
     @staticmethod
     def transform_redshift(config: Dict[str, Any]):
         print("transform_redshift")

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.1.8
+LABEL io.airbyte.version=0.1.9
 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

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test/resources/expected_spec.json

@@ -54,11 +54,130 @@
         "airbyte_secret": true,
         "order": 5
       },
+      "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": "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",

airbyte-integrations/connectors/destination-postgres/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-postgres
 
 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

airbyte-integrations/connectors/destination-postgres/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestination.java

@@ -4,6 +4,12 @@
 
 package io.airbyte.integrations.destination.postgres;
 
+import static io.airbyte.integrations.util.PostgresSslConnectionUtils.DISABLE;
+import static io.airbyte.integrations.util.PostgresSslConnectionUtils.PARAM_MODE;
+import static io.airbyte.integrations.util.PostgresSslConnectionUtils.PARAM_SSL;
+import static io.airbyte.integrations.util.PostgresSslConnectionUtils.PARAM_SSL_MODE;
+import static io.airbyte.integrations.util.PostgresSslConnectionUtils.obtainConnectionOptions;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
@@ -13,7 +19,7 @@
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.ssh.SshWrappedDestination;
 import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import org.slf4j.Logger;
@@ -25,10 +31,6 @@ public class PostgresDestination extends AbstractJdbcDestination implements Dest
 
   public static final String DRIVER_CLASS = DatabaseDriver.POSTGRESQL.getDriverClassName();
 
-  static final Map<String, String> SSL_JDBC_PARAMETERS = ImmutableMap.of(
-      JdbcUtils.SSL_KEY, "true",
-      "sslmode", "require");
-
   public static Destination sshWrappedDestination() {
     return new SshWrappedDestination(new PostgresDestination(), JdbcUtils.HOST_LIST_KEY, JdbcUtils.PORT_LIST_KEY);
   }
@@ -39,12 +41,20 @@ public PostgresDestination() {
 
   @Override
   protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
-    if (JdbcUtils.useSsl(config)) {
-      return SSL_JDBC_PARAMETERS;
-    } else {
-      // No need for any parameters if the connection doesn't use SSL
-      return Collections.emptyMap();
+    final Map<String, String> additionalParameters = new HashMap<>();
+    if (!config.has(PARAM_SSL) || config.get(PARAM_SSL).asBoolean()) {
+      if (config.has(PARAM_SSL_MODE)) {
+        if (DISABLE.equals(config.get(PARAM_SSL_MODE).get(PARAM_MODE).asText())) {
+          additionalParameters.put("sslmode", DISABLE);
+        } else {
+          additionalParameters.putAll(obtainConnectionOptions(config.get(PARAM_SSL_MODE)));
+        }
+      } else {
+        additionalParameters.put(JdbcUtils.SSL_KEY, "true");
+        additionalParameters.put("sslmode", "require");
+      }
     }
+    return additionalParameters;
   }
 
   @Override