🎉 Destination postgres: Add SSL certificates and update normalization

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,34 @@ 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")
+        print("====> SSL")
+        if ssl:
+            ssl_mode = config.get("ssl_mode", "allow")
+            dbt_config["sslmode"] = ssl_mode.get("mode")
+            if ssl_mode["mode"] == "verify-ca":
+                print("====> CA")
+                TransformConfig.create_file("ca.crt", ssl_mode["ca_certificate"])
+                dbt_config["sslrootcert"] = "ca.crt"
+            elif ssl_mode["mode"] == "verify-full":
+                print("====> 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.7
+LABEL io.airbyte.version=0.1.8
 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,12 +4,17 @@
 
 package io.airbyte.integrations.destination.postgres;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.spec_modification.SpecModifyingDestination;
 import io.airbyte.protocol.models.ConnectorSpecification;
+import java.util.ArrayList;
+import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -25,6 +30,11 @@ public PostgresDestinationStrictEncrypt() {
   public ConnectorSpecification modifySpec(final ConnectorSpecification originalSpec) {
     final ConnectorSpecification spec = Jsons.clone(originalSpec);
     ((ObjectNode) spec.getConnectionSpecification().get("properties")).remove("ssl");
+    ArrayNode modifiedSslModes = spec.getConnectionSpecification().get("properties").get("ssl_mode").get("oneOf").deepCopy();
+    // Assume that the first item is the "disable" option; remove it
+    modifiedSslModes.remove(0);
+    ((ObjectNode) spec.getConnectionSpecification().get("properties").get("ssl_mode")).remove("oneOf");
+    ((ObjectNode) spec.getConnectionSpecification().get("properties").get("ssl_mode")).put("oneOf", 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("schema", "public")
         .put("port", db.getFirstMappedPort())
         .put("database", db.getDatabaseName())
+        .put("ssl_mode", 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,145 @@
         "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": "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",

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.List;
 import java.util.Map;
 import java.util.Optional;
@@ -33,10 +39,6 @@ public class PostgresDestination extends AbstractJdbcDestination implements Dest
   public static final String USERNAME_KEY = "username";
   public static final String SCHEMA_KEY = "schema";
 
-  static final Map<String, String> SSL_JDBC_PARAMETERS = ImmutableMap.of(
-      "ssl", "true",
-      "sslmode", "require");
-
   public static Destination sshWrappedDestination() {
     return new SshWrappedDestination(new PostgresDestination(), HOST_KEY, PORT_KEY);
   }
@@ -47,12 +49,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("ssl", "true");
+        additionalParameters.put("sslmode", "require");
+      }
     }
+    return additionalParameters;
   }
 
   @Override
@@ -80,7 +90,6 @@ public JsonNode toJdbcConfig(final JsonNode config) {
     return Jsons.jsonNode(configBuilder.build());
   }
 
-
   public static void main(final String[] args) throws Exception {
     final Destination destination = PostgresDestination.sshWrappedDestination();
     LOGGER.info("starting destination: {}", PostgresDestination.class);