[#7547] Validate JDBC (#7611)

### What changes were proposed in this pull request?

 Validate JDBC connection for unsafe operations.

### Why are the changes needed?

To stop potentially unsafe operations.

Fix: #7547

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Locally, plus added unit tests.

Co-authored-by: Justin Mclean <[email protected]>
Co-authored-by: Mini Yu <[email protected]>

Author: 3 people

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java

@@ -19,6 +19,8 @@
 package org.apache.gravitino.catalog.jdbc.utils;
 
 import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import javax.sql.DataSource;
@@ -37,6 +39,26 @@ public class DataSourceUtils {
   /** SQL statements for database connection pool testing. */
   private static final String POOL_TEST_QUERY = "SELECT 1";
 
+  private static final List<String> unsafeMySQLParameters =
+      Arrays.asList(
+          "maxAllowedPacket",
+          "autoDeserialize",
+          "queryInterceptors",
+          "statementInterceptors",
+          "detectCustomCollations",
+          "allowloadlocalinfile",
+          "allowUrlInLocalInfile",
+          "allowLoadLocalInfileInPath");
+
+  private static final List<String> unsafePostgresParameters =
+      Arrays.asList(
+          "socketFactory",
+          "socketFactoryArg",
+          "sslfactory",
+          "sslhostnameverifier",
+          "sslpasswordcallback",
+          "authenticationPluginClassName");
+
   public static DataSource createDataSource(Map<String, String> properties) {
     return createDataSource(new JdbcConfig(properties));
   }
@@ -51,6 +73,8 @@ public static DataSource createDataSource(JdbcConfig jdbcConfig)
   }
 
   private static DataSource createDBCPDataSource(JdbcConfig jdbcConfig) throws Exception {
+    validateJdbcConfig(
+        jdbcConfig.getJdbcDriver(), jdbcConfig.getJdbcUrl(), jdbcConfig.getAllConfig());
     BasicDataSource basicDataSource =
         BasicDataSourceFactory.createDataSource(getProperties(jdbcConfig));
     String jdbcUrl = jdbcConfig.getJdbcUrl();
@@ -76,6 +100,61 @@ private static Properties getProperties(JdbcConfig jdbcConfig) {
     return properties;
   }
 
+  private static String recursiveDecode(String url) {
+    String prev;
+    String decoded = url;
+    int max = 5;
+
+    do {
+      prev = decoded;
+      try {
+        decoded = java.net.URLDecoder.decode(prev, "UTF-8");
+      } catch (Exception e) {
+        throw new GravitinoRuntimeException("Unable to decode JDBC URL");
+      }
+    } while (!prev.equals(decoded) && --max > 0);
+
+    return decoded;
+  }
+
+  private static void checkUnsafeParameters(
+      String url, Map<String, String> config, List<String> unsafeParams, String dbType) {
+
+    String lowerUrl = url.toLowerCase();
+
+    for (String param : unsafeParams) {
+      String lowerParam = param.toLowerCase();
+      if (lowerUrl.contains(lowerParam) || containsValueIgnoreCase(config, param)) {
+        throw new GravitinoRuntimeException(
+            "Unsafe %s parameter '%s' detected in JDBC URL", dbType, param);
+      }
+    }
+  }
+
+  public static void validateJdbcConfig(String driver, String url, Map<String, String> all) {
+    String lowerUrl = url.toLowerCase();
+    String decodedUrl = recursiveDecode(lowerUrl);
+
+    if (driver != null) {
+      if (decodedUrl.startsWith("jdbc:mysql")) {
+        checkUnsafeParameters(decodedUrl, all, unsafeMySQLParameters, "MySQL");
+      } else if (decodedUrl.startsWith("jdbc:mariadb")) {
+        checkUnsafeParameters(decodedUrl, all, unsafeMySQLParameters, "MariaDB");
+      } else if (decodedUrl.startsWith("jdbc:postgresql")) {
+        checkUnsafeParameters(decodedUrl, all, unsafePostgresParameters, "PostgreSQL");
+      }
+    }
+  }
+
+  private static boolean containsValueIgnoreCase(Map<String, String> map, String value) {
+    for (String keyValue : map.values()) {
+      if (keyValue != null && keyValue.equalsIgnoreCase(value)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   public static void closeDataSource(DataSource dataSource) {
     if (null != dataSource) {
       try {

catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/utils/TestDataSourceUtils.java

@@ -24,6 +24,7 @@
 import javax.sql.DataSource;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
+import org.apache.gravitino.exceptions.GravitinoRuntimeException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -42,4 +43,45 @@ public void testCreateDataSource() throws SQLException {
     Assertions.assertTrue(dataSource instanceof org.apache.commons.dbcp2.BasicDataSource);
     ((BasicDataSource) dataSource).close();
   }
+
+  @Test
+  public void testRejectMysqlAllowLoadLocalInfile() {
+    HashMap<String, String> properties = Maps.newHashMap();
+    properties.put(JdbcConfig.JDBC_DRIVER.getKey(), "com.mysql.cj.jdbc.Driver");
+    properties.put(
+        JdbcConfig.JDBC_URL.getKey(), "jdbc:mysql://localhost:3306/test?allowLoadLocalInfile=true");
+    properties.put(JdbcConfig.USERNAME.getKey(), "test");
+    properties.put(JdbcConfig.PASSWORD.getKey(), "test");
+
+    Assertions.assertThrows(
+        GravitinoRuntimeException.class, () -> DataSourceUtils.createDataSource(properties));
+  }
+
+  @Test
+  public void testRejectPostgresSocketFactory() {
+    HashMap<String, String> properties = Maps.newHashMap();
+    properties.put(JdbcConfig.JDBC_DRIVER.getKey(), "org.postgresql.Driver");
+    properties.put(
+        JdbcConfig.JDBC_URL.getKey(),
+        "jdbc:postgresql:///test?socketFactory=java.io.FileOutputStream&socketFactoryArg=/tmp/x");
+    properties.put(JdbcConfig.USERNAME.getKey(), "test");
+    properties.put(JdbcConfig.PASSWORD.getKey(), "test");
+
+    Assertions.assertThrows(
+        GravitinoRuntimeException.class, () -> DataSourceUtils.createDataSource(properties));
+  }
+
+  @Test
+  public void testRejectEncodedAllowLoadLocalInfile() {
+    HashMap<String, String> properties = Maps.newHashMap();
+    properties.put(JdbcConfig.JDBC_DRIVER.getKey(), "org.postgresql.Driver");
+    properties.put(
+        JdbcConfig.JDBC_URL.getKey(),
+        "jdbc:mysql://address=(host=172.18.0.1)(port=3309)(%61llowLoadLocalInfile=true),172.18.0.1:3306/test");
+    properties.put(JdbcConfig.USERNAME.getKey(), "test");
+    properties.put(JdbcConfig.PASSWORD.getKey(), "test");
+
+    Assertions.assertThrows(
+        GravitinoRuntimeException.class, () -> DataSourceUtils.createDataSource(properties));
+  }
 }