🚨 Validate source JDBC url parameters 🚨 (#14586)

* Moved toDatabaseConfigStatic subfunction into the main toDatabaseConfig function

* Moved Postgres JDBC_URL_PARAMS_KEY into JdbcUtils to reduce redundancy

* Migrated useSsl method into JdbcUtils to remove duplication

* Created constants for config parameters to match PostgresDestination

* Add validation check for overwritten connection properties

* Added test coverage for useSsl and clarifying javadoc comments

* Updated useSsl tests to match linter and javadoc comments for useSsl

* Fixed isCDC method with proper PostgresUtil function and corresponding imports

* Bumping Dockerfile and adding changelog description

* auto-bump connector version

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

Author: ryankfu

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

@@ -754,7 +754,7 @@
 - name: Postgres
   sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   dockerRepository: airbyte/source-postgres
-  dockerImageTag: 0.4.32
+  dockerImageTag: 0.4.33
   documentationUrl: https://docs.airbyte.io/integrations/sources/postgres
   icon: postgresql.svg
   sourceType: database

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

@@ -6960,7 +6960,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-postgres:0.4.32"
+- dockerImage: "airbyte/source-postgres:0.4.33"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/postgres"
     connectionSpecification:

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

@@ -12,6 +12,9 @@
 
 public class JdbcUtils {
 
+  public static final String JDBC_URL_PARAMS_KEY = "jdbc_url_params";
+  public static final String SSL_KEY = "ssl";
+
   private static final JdbcSourceOperations defaultSourceOperations = new JdbcSourceOperations();
 
   private static final JSONFormat defaultJSONFormat = new JSONFormat().recordFormat(JSONFormat.RecordFormat.OBJECT);
@@ -62,4 +65,15 @@ public static Map<String, String> parseJdbcParameters(final String jdbcPropertie
     return parameters;
   }
 
+  /**
+   * Checks that SSL_KEY has not been set or that an SSL_KEY is set and value can be mapped to true
+   * (e.g. non-zero integers, string true, etc)
+   *
+   * @param config A configuration used to check Jdbc connection
+   * @return true: if ssl has not been set or it has been set with true, false: in all other cases
+   */
+  public static boolean useSsl(final JsonNode config) {
+    return !config.has(SSL_KEY) || config.get(SSL_KEY).asBoolean();
+  }
+
 }

airbyte-db/db-lib/src/test/java/io/airbyte/db/jdbc/TestJdbcUtils.java

@@ -5,6 +5,8 @@
 package io.airbyte.db.jdbc;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -41,6 +43,8 @@
 
 public class TestJdbcUtils {
 
+  private String dbName;
+
   private static final List<JsonNode> RECORDS_AS_JSON = Lists.newArrayList(
       Jsons.jsonNode(ImmutableMap.of("id", 1, "name", "picard")),
       Jsons.jsonNode(ImmutableMap.of("id", 2, "name", "crusher")),
@@ -60,7 +64,7 @@ static void init() {
 
   @BeforeEach
   void setup() throws Exception {
-    final String dbName = Strings.addRandomSuffix("db", "_", 10);
+    dbName = Strings.addRandomSuffix("db", "_", 10);
 
     final JsonNode config = getConfig(PSQL_DB, dbName);
 
@@ -95,6 +99,18 @@ private JsonNode getConfig(final PostgreSQLContainer<?> psqlDb, final String dbN
         .build());
   }
 
+  // Takes in a generic sslValue because useSsl maps sslValue to a boolean
+  private <T> JsonNode getConfigWithSsl(final PostgreSQLContainer<?> psqlDb, final String dbName, final T sslValue) {
+    return Jsons.jsonNode(ImmutableMap.builder()
+        .put("host", psqlDb.getHost())
+        .put("port", psqlDb.getFirstMappedPort())
+        .put("database", dbName)
+        .put("username", psqlDb.getUsername())
+        .put("password", psqlDb.getPassword())
+        .put("ssl", sslValue)
+        .build());
+  }
+
   @Test
   void testRowToJson() throws SQLException {
     try (final Connection connection = dataSource.getConnection()) {
@@ -157,6 +173,41 @@ void testSetStatementField() throws SQLException {
     }
   }
 
+  @Test
+  void testUseSslWithSslNotSet() {
+    final JsonNode config = getConfig(PSQL_DB, dbName);
+    final boolean sslSet = JdbcUtils.useSsl(config);
+    assertTrue(sslSet);
+  }
+
+  @Test
+  void testUseSslWithSslSetAndValueStringFalse() {
+    final JsonNode config = getConfigWithSsl(PSQL_DB, dbName, "false");
+    final boolean sslSet = JdbcUtils.useSsl(config);
+    assertFalse(sslSet);
+  }
+
+  @Test
+  void testUseSslWithSslSetAndValueIntegerFalse() {
+    final JsonNode config = getConfigWithSsl(PSQL_DB, dbName, 0);
+    final boolean sslSet = JdbcUtils.useSsl(config);
+    assertFalse(sslSet);
+  }
+
+  @Test
+  void testUseSslWithSslSetAndValueStringTrue() {
+    final JsonNode config = getConfigWithSsl(PSQL_DB, dbName, "true");
+    final boolean sslSet = JdbcUtils.useSsl(config);
+    assertTrue(sslSet);
+  }
+
+  @Test
+  void testUssSslWithSslSetAndValueIntegerTrue() {
+    final JsonNode config = getConfigWithSsl(PSQL_DB, dbName, 3);
+    final boolean sslSet = JdbcUtils.useSsl(config);
+    assertTrue(sslSet);
+  }
+
   private static void createTableWithAllTypes(final Connection connection) throws SQLException {
     // jdbctype not included because they are not directly supported in postgres: TINYINT, LONGVARCHAR,
     // VARBINAR, LONGVARBINARY

airbyte-integrations/connectors/destination-clickhouse/src/main/java/io/airbyte/integrations/destination/clickhouse/ClickhouseDestination.java

@@ -10,6 +10,7 @@
 import io.airbyte.db.factory.DataSourceFactory;
 import io.airbyte.db.factory.DatabaseDriver;
 import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.db.jdbc.JdbcUtils;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.ssh.SshWrappedDestination;
@@ -65,9 +66,6 @@ public JsonNode toJdbcConfig(final JsonNode config) {
     return Jsons.jsonNode(configBuilder.build());
   }
 
-  private boolean useSsl(final JsonNode config) {
-    return !config.has("ssl") || config.get("ssl").asBoolean();
-  }
 
   @Override
   public AirbyteConnectionStatus check(final JsonNode config) {
@@ -94,7 +92,7 @@ public AirbyteConnectionStatus check(final JsonNode config) {
 
   @Override
   protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
-    if (useSsl(config)) {
+    if (JdbcUtils.useSsl(config)) {
       return SSL_JDBC_PARAMETERS;
     } else {
       // No need for any parameters if the connection doesn't use SSL except socket_timeout

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/AbstractJdbcDestination.java

@@ -31,8 +31,6 @@ public abstract class AbstractJdbcDestination extends BaseConnector implements D
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractJdbcDestination.class);
 
-  public static final String JDBC_URL_PARAMS_KEY = "jdbc_url_params";
-
   private final String driverClass;
   private final NamingConventionTransformer namingResolver;
   private final SqlOperations sqlOperations;
@@ -105,7 +103,7 @@ protected JdbcDatabase getDatabase(final DataSource dataSource) {
   }
 
   protected Map<String, String> getConnectionProperties(final JsonNode config) {
-    final Map<String, String> customProperties = JdbcUtils.parseJdbcParameters(config, JDBC_URL_PARAMS_KEY);
+    final Map<String, String> customProperties = JdbcUtils.parseJdbcParameters(config, JdbcUtils.JDBC_URL_PARAMS_KEY);
     final Map<String, String> defaultProperties = getDefaultConnectionProperties(config);
     assertCustomParametersDontOverwriteDefaultParameters(customProperties, defaultProperties);
     return MoreMaps.merge(customProperties, defaultProperties);

airbyte-integrations/connectors/destination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java

@@ -11,6 +11,7 @@
 import io.airbyte.db.factory.DataSourceFactory;
 import io.airbyte.db.factory.DatabaseDriver;
 import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.db.jdbc.JdbcUtils;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.ssh.SshWrappedDestination;
@@ -99,16 +100,13 @@ public MySQLDestination() {
 
   @Override
   protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
-    if (useSSL(config)) {
+    if (JdbcUtils.useSsl(config)) {
       return DEFAULT_SSL_JDBC_PARAMETERS;
     } else {
       return DEFAULT_JDBC_PARAMETERS;
     }
   }
 
-  private boolean useSSL(final JsonNode config) {
-    return !config.has(SSL_KEY) || config.get(SSL_KEY).asBoolean();
-  }
 
   @Override
   public JsonNode toJdbcConfig(final JsonNode config) {

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

@@ -8,6 +8,7 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.db.factory.DatabaseDriver;
+import io.airbyte.db.jdbc.JdbcUtils;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.ssh.SshWrappedDestination;
@@ -28,9 +29,7 @@ public class PostgresDestination extends AbstractJdbcDestination implements Dest
   public static final List<String> PORT_KEY = List.of("port");
   public static final String DATABASE_KEY = "database";
   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";
 
@@ -48,7 +47,7 @@ public PostgresDestination() {
 
   @Override
   protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
-    if (useSsl(config)) {
+    if (JdbcUtils.useSsl(config)) {
       return SSL_JDBC_PARAMETERS;
     } else {
       // No need for any parameters if the connection doesn't use SSL
@@ -74,16 +73,13 @@ public JsonNode toJdbcConfig(final JsonNode config) {
       configBuilder.put(PASSWORD_KEY, config.get(PASSWORD_KEY).asText());
     }
 
-    if (config.has(JDBC_URL_PARAMS_KEY)) {
-      configBuilder.put(JDBC_URL_PARAMS_KEY, config.get(JDBC_URL_PARAMS_KEY).asText());
+    if (config.has(JdbcUtils.JDBC_URL_PARAMS_KEY)) {
+      configBuilder.put(JdbcUtils.JDBC_URL_PARAMS_KEY, config.get(JdbcUtils.JDBC_URL_PARAMS_KEY).asText());
     }
 
     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();

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -23,6 +23,7 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.functional.CheckedConsumer;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.map.MoreMaps;
 import io.airbyte.commons.util.AutoCloseableIterator;
 import io.airbyte.commons.util.AutoCloseableIterators;
 import io.airbyte.db.JdbcCompatibleSourceOperations;
@@ -49,6 +50,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -297,7 +299,7 @@ protected DataSource createDataSource(final JsonNode config) {
         jdbcConfig.has("password") ? jdbcConfig.get("password").asText() : null,
         driverClass,
         jdbcConfig.get("jdbc_url").asText(),
-        JdbcUtils.parseJdbcParameters(jdbcConfig, "connection_properties", getJdbcParameterDelimiter()));
+        getConnectionProperties(config));
     // Record the data source so that it can be closed.
     dataSources.add(dataSource);
     return dataSource;
@@ -316,6 +318,48 @@ public JdbcDatabase createDatabase(final JsonNode config) throws SQLException {
     return database;
   }
 
+  /**
+   * Retrieves connection_properties from config and also validates if custom
+   * jdbc_url parameters overlap with the default properties
+   *
+   * @param config A configuration used to check Jdbc connection
+   * @return A mapping of connection properties
+   */
+  protected Map<String, String> getConnectionProperties(final JsonNode config) {
+    final Map<String, String> customProperties = JdbcUtils.parseJdbcParameters(config, JdbcUtils.JDBC_URL_PARAMS_KEY);
+    final Map<String, String> defaultProperties = getDefaultConnectionProperties(config);
+    assertCustomParametersDontOverwriteDefaultParameters(customProperties, defaultProperties);
+    return MoreMaps.merge(customProperties, defaultProperties);
+  }
+
+  /**
+   * Validates for duplication parameters
+   *
+   * @param customParameters custom connection properties map as specified by each Jdbc source
+   * @param defaultParameters connection properties map as specified by each Jdbc source
+   * @throws IllegalArgumentException
+   */
+  private void assertCustomParametersDontOverwriteDefaultParameters(final Map<String, String> customParameters,
+      final Map<String, String> defaultParameters) {
+    for (final String key : defaultParameters.keySet()) {
+      if (customParameters.containsKey(key) && !Objects.equals(customParameters.get(key), defaultParameters.get(key))) {
+        throw new IllegalArgumentException("Cannot overwrite default JDBC parameter " + key);
+      }
+    }
+  }
+
+  /**
+   * Retrieves default connection_properties from config
+   *
+   * TODO: make this method abstract and add parity features to
+   * destination connectors
+   * @param config A configuration used to check Jdbc connection
+   * @return A mapping of the default connection properties
+   */
+  protected Map<String, String> getDefaultConnectionProperties(final JsonNode config) {
+    return JdbcUtils.parseJdbcParameters(config, "connection_properties", getJdbcParameterDelimiter());
+  };
+
   protected String getJdbcParameterDelimiter() {
     return "&";
   }

airbyte-integrations/connectors/source-postgres/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-postgres
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.4.32
+LABEL io.airbyte.version=0.4.33
 LABEL io.airbyte.name=airbyte/source-postgres