Source postgres: Add SSL certificates to source postgres

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

@@ -0,0 +1,161 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.util;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.RandomStringUtils;
+
+public class PostgresSslConnectionUtils {
+
+  private static final String KEY_STORE_PASS = RandomStringUtils.randomAlphanumeric(8);
+  private static final String CA_CERTIFICATE = "ca.crt";
+  private static final String CLIENT_CERTIFICATE = "client.crt";
+  private static final String CLIENT_KEY = "client.key";
+  private static final String CLIENT_ENCRYPTED_KEY = "client.pk8";
+
+  public static final String PARAM_MODE = "mode";
+  public static final String PARAM_SSL = "ssl";
+  public static final String PARAM_SSL_MODE = "ssl_mode";
+  public static final String PARAM_SSLMODE = "sslmode";
+  public static final String PARAM_CLIENT_KEY_PASSWORD = "client_key_password";
+  public static final String PARAM_CA_CERTIFICATE = "ca_certificate";
+  public static final String PARAM_CLIENT_CERTIFICATE = "client_certificate";
+  public static final String PARAM_CLIENT_KEY = "client_key";
+
+  public static final String VERIFY_CA = "verify-ca";
+  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 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();
+      switch (method) {
+        case VERIFY_CA -> {
+          final var clientKeyPassword = getKeyStorePassword(encryption.get(PARAM_CLIENT_KEY_PASSWORD));
+          additionalParameters.putAll(obtainConnectionCaOptions(encryption, method, clientKeyPassword));
+        }
+        case VERIFY_FULL -> {
+          final var clientKeyPassword = getKeyStorePassword(encryption.get(PARAM_CLIENT_KEY_PASSWORD));
+          additionalParameters.putAll(obtainConnectionFullOptions(encryption, method, clientKeyPassword));
+        }
+        default -> {
+          additionalParameters.put(PARAM_SSL, TRUE_STRING_VALUE);
+          additionalParameters.put(PARAM_SSLMODE, method);
+        }
+      }
+    }
+    return additionalParameters;
+  }
+
+  private static Map<String, String> obtainConnectionFullOptions(final JsonNode encryption,
+                                                                 final String method,
+                                                                 final String clientKeyPassword) {
+    final Map<String, String> additionalParameters = new HashMap<>();
+    try {
+      convertAndImportFullCertificate(encryption.get(PARAM_CA_CERTIFICATE).asText(),
+          encryption.get(PARAM_CLIENT_CERTIFICATE).asText(), encryption.get(PARAM_CLIENT_KEY).asText(), clientKeyPassword);
+    } catch (final IOException | InterruptedException e) {
+      throw new RuntimeException("Failed to import certificate into Java Keystore");
+    }
+    additionalParameters.put("ssl", TRUE_STRING_VALUE);
+    additionalParameters.put("sslmode", method);
+    additionalParameters.put("sslrootcert", CA_CERTIFICATE);
+    additionalParameters.put("sslcert", CLIENT_CERTIFICATE);
+    additionalParameters.put("sslkey", CLIENT_ENCRYPTED_KEY);
+    additionalParameters.put("sslfactory", FACTORY_VALUE);
+    return additionalParameters;
+  }
+
+  private static Map<String, String> obtainConnectionCaOptions(final JsonNode encryption,
+                                                               final String method,
+                                                               final String clientKeyPassword) {
+    final Map<String, String> additionalParameters = new HashMap<>();
+    try {
+      convertAndImportCaCertificate(encryption.get(PARAM_CA_CERTIFICATE).asText(), clientKeyPassword);
+    } catch (final IOException | InterruptedException e) {
+      throw new RuntimeException("Failed to import certificate into Java Keystore");
+    }
+    additionalParameters.put("ssl", TRUE_STRING_VALUE);
+    additionalParameters.put("sslmode", method);
+    additionalParameters.put("sslrootcert", CA_CERTIFICATE);
+    additionalParameters.put("sslfactory", FACTORY_VALUE);
+    return additionalParameters;
+  }
+
+  private static void convertAndImportFullCertificate(final String caCertificate,
+                                                      final String clientCertificate,
+                                                      final String clientKey,
+                                                      final String clientKeyPassword)
+      throws IOException, InterruptedException {
+    final Runtime run = Runtime.getRuntime();
+    createCaCertificate(caCertificate, clientKeyPassword, run);
+    createCertificateFile(CLIENT_CERTIFICATE, clientCertificate);
+    createCertificateFile(CLIENT_KEY, clientKey);
+    // add client certificate to the custom keystore
+    runProcess("keytool -alias client-certificate -keystore customkeystore"
+        + " -import -file " + CLIENT_CERTIFICATE + " -storepass " + clientKeyPassword + " -noprompt", run);
+    // convert client.key to client.pk8 based on the documentation
+    runProcess("openssl pkcs8 -topk8 -inform PEM -in " + CLIENT_KEY + " -outform DER -out "
+        + CLIENT_ENCRYPTED_KEY + " -nocrypt", run);
+    runProcess("rm " + CLIENT_KEY, run);
+
+    updateTrustStoreSystemProperty(clientKeyPassword);
+  }
+
+  private static void convertAndImportCaCertificate(final String caCertificate,
+                                                    final String clientKeyPassword)
+      throws IOException, InterruptedException {
+    final Runtime run = Runtime.getRuntime();
+    createCaCertificate(caCertificate, clientKeyPassword, run);
+    updateTrustStoreSystemProperty(clientKeyPassword);
+  }
+
+  private static void createCaCertificate(final String caCertificate,
+                                          final String clientKeyPassword,
+                                          final Runtime run) throws IOException, InterruptedException {
+    createCertificateFile(CA_CERTIFICATE, caCertificate);
+    // add CA certificate to the custom keystore
+    runProcess("keytool -import -alias rds-root -keystore customkeystore"
+            + " -file " + CA_CERTIFICATE + " -storepass " + clientKeyPassword + " -noprompt", run);
+  }
+
+  private static void updateTrustStoreSystemProperty(final String clientKeyPassword) {
+    String result = System.getProperty("user.dir") + "/customkeystore";
+    System.setProperty("javax.net.ssl.trustStore", result);
+    System.setProperty("javax.net.ssl.trustStorePassword", clientKeyPassword);
+  }
+
+  private static void createCertificateFile(String fileName, String fileValue) throws IOException {
+    try (final PrintWriter out = new PrintWriter(fileName, StandardCharsets.UTF_8)) {
+      out.print(fileValue);
+    }
+  }
+
+  private static String getKeyStorePassword(final JsonNode sslMode) {
+    var keyStorePassword = KEY_STORE_PASS;
+    if (!sslMode.isNull() || !sslMode.isEmpty()) {
+      keyStorePassword = sslMode.asText();
+    }
+    return keyStorePassword;
+  }
+
+  private static void runProcess(final String cmd, final Runtime run) throws IOException, InterruptedException {
+    final Process pr = run.exec(cmd);
+    if (!pr.waitFor(30, TimeUnit.SECONDS)) {
+      pr.destroy();
+      throw new RuntimeException("Timeout while executing: " + cmd);
+    }
+  }
+
+}

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;
@@ -12,7 +18,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;
@@ -30,14 +36,9 @@ public class PostgresDestination extends AbstractJdbcDestination implements Dest
   public static final String JDBC_URL_KEY = "jdbc_url";
   public static final String JDBC_URL_PARAMS_KEY = "jdbc_url_params";
   public static final String PASSWORD_KEY = "password";
-  public static final String SSL_KEY = "ssl";
   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);
   }
@@ -48,12 +49,15 @@ public PostgresDestination() {
 
   @Override
   protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
-    if (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 (DISABLE.equals(config.get(PARAM_SSL_MODE).get(PARAM_MODE).asText())) {
+        additionalParameters.put("sslmode", DISABLE);
+      } else {
+        additionalParameters.putAll(obtainConnectionOptions(config.get(PARAM_SSL_MODE)));
+      }
     }
+    return additionalParameters;
   }
 
   @Override
@@ -81,10 +85,6 @@ public JsonNode toJdbcConfig(final JsonNode config) {
     return Jsons.jsonNode(configBuilder.build());
   }
 
-  private boolean useSsl(final JsonNode config) {
-    return !config.has(SSL_KEY) || config.get(SSL_KEY).asBoolean();
-  }
-
   public static void main(final String[] args) throws Exception {
     final Destination destination = PostgresDestination.sshWrappedDestination();
     LOGGER.info("starting destination: {}", PostgresDestination.class);

airbyte-integrations/connectors/destination-postgres/src/main/resources/spec.json

@@ -61,11 +61,155 @@
         "default": false,
         "order": 6
       },
+      "ssl_mode": {
+        "title": "SSL modes",
+        "description": "SSL modes.",
+        "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"],
+            "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"],
+            "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
       }
     }
   }