[ZEPPELIN-6204] do not allow init params in JDBC URLs for H2

### What is this PR for?
ZEPPELIN-6204

Slight tidy of the existing disallow list for strings in JDBC urls so that they are checked against just the query params and not the hostname in the URL. 

### What type of PR is it?
Bug Fix

### Todos
* [ ] - Task

### What is the Jira issue?
* Open an issue on Jira https://issues.apache.org/jira/browse/ZEPPELIN/
* Put link here, and add [ZEPPELIN-*Jira number*] in PR title, eg. [ZEPPELIN-533]

### How should this be tested?
* Strongly recommended: add automated unit tests for any new or changed behavior
* Outline any manual steps to test the PR here.

### Screenshots (if appropriate)

### Questions:
* Does the license files need to update? no
* Is there breaking changes for older versions? no
* Does this needs documentation? no


Closes #4949 from pjfanning/ZEPPELIN-6204-jdbc.

Signed-off-by: ChanHo Lee <chanholee@apache.org>

Author: pjfanning

jdbc/pom.xml

@@ -39,6 +39,7 @@
     <commons.dbcp2.version>2.0.1</commons.dbcp2.version>
     <hive3.version>3.1.3</hive3.version>
     <kyuubi.version>1.9.1</kyuubi.version>
+    <mysql.connector.version>9.3.0</mysql.connector.version>
 
     <!--test library versions-->
     <mockrunner.jdbc.version>1.0.8</mockrunner.jdbc.version>
@@ -59,6 +60,13 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>com.mysql</groupId>
+      <artifactId>mysql-connector-j</artifactId>
+      <version>${mysql.connector.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>

jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCInterpreter.java

@@ -156,14 +156,22 @@ public class JDBCInterpreter extends KerberosInterpreter {
           "KerberosConfigPath", "KerberosKeytabPath", "KerberosCredentialCachePath",
           "extraCredentials", "roles", "sessionProperties"));
 
+  private static final String ALLOW_LOAD_LOCAL = "allowLoadLocal";
+
   private static final String ALLOW_LOAD_LOCAL_IN_FILE_NAME = "allowLoadLocalInfile";
 
-  private static final String AUTO_DESERIALIZE = "autoDeserialize";
+  private static final String ALLOW_LOAD_LOCAL_IN_FILE_IN_PATH = "allowLoadLocalInfileInPath";
 
   private static final String ALLOW_LOCAL_IN_FILE_NAME = "allowLocalInfile";
 
   private static final String ALLOW_URL_IN_LOCAL_IN_FILE_NAME = "allowUrlInLocalInfile";
 
+  private static final String AUTO_DESERIALIZE = "autoDeserialize";
+
+  private static final String SOCKET_FACTORY = "socketFactory";
+
+  private static final String INIT = "INIT";
+
   // database --> Properties
   private final HashMap<String, Properties> basePropertiesMap;
   // username --> User Configuration
@@ -588,18 +596,127 @@ public Connection getConnection(InterpreterContext context)
     return connection;
   }
 
-  private void validateConnectionUrl(String url) {
-    String decodedUrl;
-    decodedUrl = URLDecoder.decode(url, StandardCharsets.UTF_8);
+  // package private for testing purposes
+  static void validateConnectionUrl(String url) {
+    final String decodedUrl = urlDecode(url, url, 0);
+    final Map<String, String> params = parseUrlParameters(decodedUrl);
+
+    if (containsKeyIgnoreCase(params, ALLOW_LOAD_LOCAL) ||
+            containsKeyIgnoreCase(params, ALLOW_LOAD_LOCAL_IN_FILE_NAME) ||
+            containsKeyIgnoreCase(params, ALLOW_LOCAL_IN_FILE_NAME) ||
+            containsKeyIgnoreCase(params, ALLOW_URL_IN_LOCAL_IN_FILE_NAME) ||
+            containsKeyIgnoreCase(params, ALLOW_LOAD_LOCAL_IN_FILE_IN_PATH) ||
+            containsKeyIgnoreCase(params, AUTO_DESERIALIZE) ||
+            containsKeyIgnoreCase(params, SOCKET_FACTORY)) {
+      throw new IllegalArgumentException("Connection URL contains sensitive configuration");
+    }
 
-    if (containsIgnoreCase(decodedUrl, ALLOW_LOAD_LOCAL_IN_FILE_NAME) ||
-            containsIgnoreCase(decodedUrl, AUTO_DESERIALIZE) ||
-            containsIgnoreCase(decodedUrl, ALLOW_LOCAL_IN_FILE_NAME) ||
-            containsIgnoreCase(decodedUrl, ALLOW_URL_IN_LOCAL_IN_FILE_NAME)) {
+    // the INIT parameter name is a bit generic so we check it only for H2
+    if (containsIgnoreCase(decodedUrl, "jdbc:h2") && containsKeyIgnoreCase(params, INIT)) {
       throw new IllegalArgumentException("Connection URL contains sensitive configuration");
     }
   }
 
+  /**
+   * Decode the URL encoded string recursively until no more decoding is needed.
+   * This is to handle cases where the URL might be double-encoded.
+   *
+   * @param url the original URL (for logging purposes)
+   * @param encoded the URL encoded string
+   * @param recurseCount the current recursion depth
+   * @return the decoded string
+   * @throws IllegalArgumentException if the recursion depth exceeds 10
+   */
+  private static String urlDecode(final String url,
+                                  final String encoded,
+                                  final int recurseCount) {
+    if (recurseCount > 10) {
+      throw new IllegalArgumentException("illegal URL encoding detected: " + url);
+    }
+    final String decoded = URLDecoder.decode(encoded, StandardCharsets.UTF_8);
+    if (decoded.equals(encoded)) {
+      return decoded; // No more decoding needed or max recursion reached
+    }
+    return urlDecode(url, decoded, recurseCount + 1);
+  }
+
+  private static Map<String, String> parseUrlParameters(final String url) {
+    final Map<String, String> parameters = new HashMap<>();
+
+    // MySQL supports parentheses in the URL
+    // https://dev.mysql.com/doc/connectors/en/connector-j-reference-jdbc-url-format.html
+    // eg jdbc:mysql://(host=myhost,port=1111,allowLoadLocalInfile=true)/db
+    int parensIndex = extractFromParens(url, 0, parameters);
+    while (parensIndex > 0) {
+      parensIndex = extractFromParens(url, parensIndex, parameters);
+    }
+
+    // Split the URL into the base part and the parameters part
+    String[] parts = url.split("[?&;]");
+    if (parts.length > 1) {
+      // The first part is the base URL, so we start from the second part
+      for (int i = 1; i < parts.length; i++) {
+        splitNameValue(parts[i], parameters, true);
+      }
+    }
+    return parameters;
+  }
+
+  private static boolean containsKeyIgnoreCase(Map<String, String> map, String key) {
+    for (String k : map.keySet()) {
+      if (k.equalsIgnoreCase(key)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Extracts key-value pairs from parentheses in the input string.
+   * The expected format is "(key1=value1, key2=value2, ...)".
+   *
+   * @param input the input string containing parameters in parentheses
+   * @param initIndex the index to start searching for parentheses
+   * @param parameters the map to store extracted key-value pairs
+   * @return the index of the closing parenthesis or -1 if not found
+   */
+  private static int extractFromParens(final String input,
+                                       final int initIndex,
+                                       final Map<String, String> parameters) {
+    final int startIndex = input.indexOf('(', initIndex);
+    if (startIndex == -1) {
+      return -1;
+    }
+    final int endIndex = input.indexOf(')', startIndex);
+    if (startIndex != -1 && endIndex != -1) {
+      String params = input.substring(startIndex + 1, endIndex);
+      String[] keyValuePairs = params.split(",");
+      for (String pair : keyValuePairs) {
+        splitNameValue(pair, parameters, false);
+      }
+    }
+    return endIndex;
+  }
+
+  /**
+   * Splits a name-value pair and adds it to the parameters map.
+   * Handles cases where the value might be missing.
+   *
+   * @param nameValue the name-value pair as a string
+   * @param parameters the map to store the extracted key-value pair
+   * @param allowEmptyValue whether to allow empty values
+   */
+  private static void splitNameValue(String nameValue, Map<String, String> parameters,
+                                     boolean allowEmptyValue) {
+    String[] keyValue = nameValue.split("=");
+    if (keyValue.length >= 2) {
+      parameters.put(keyValue[0].trim(), keyValue[1].trim());
+    } else if (allowEmptyValue) {
+      // Handle cases where there might not be a value
+      parameters.put(keyValue[0].trim(), "");
+    }
+  }
+
   private String appendProxyUserToURL(String url, String user) {
     StringBuilder connectionUrl = new StringBuilder(url);
 

jdbc/src/test/java/org/apache/zeppelin/jdbc/JDBCInterpreterTest.java

@@ -58,11 +58,11 @@
 import static org.apache.zeppelin.jdbc.JDBCInterpreter.DEFAULT_URL;
 import static org.apache.zeppelin.jdbc.JDBCInterpreter.DEFAULT_USER;
 import static org.apache.zeppelin.jdbc.JDBCInterpreter.PRECODE_KEY_TEMPLATE;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+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 static org.junit.jupiter.api.Assertions.fail;
-import static org.junit.jupiter.api.Assertions.assertEquals;
 
 
 /**
@@ -748,33 +748,85 @@ void testSplitSqlQueryWithComments() throws IOException,
   }
 
   @Test
-  void testValidateConnectionUrl() throws IOException, InterpreterException {
-    Properties properties = new Properties();
-    properties.setProperty("default.driver", "org.h2.Driver");
-    properties.setProperty("default.url", getJdbcConnection() + ";allowLoadLocalInfile=true");
-    properties.setProperty("default.user", "");
-    properties.setProperty("default.password", "");
-    JDBCInterpreter jdbcInterpreter = new JDBCInterpreter(properties);
-    jdbcInterpreter.open();
-    InterpreterResult interpreterResult = jdbcInterpreter.interpret("SELECT 1", context);
-    assertEquals(InterpreterResult.Code.ERROR, interpreterResult.code());
-    assertEquals("Connection URL contains improper configuration",
-            interpreterResult.message().get(0).getData());
+  void testValidateConnectionUrlAllowLoadLocalInFile() throws IOException, InterpreterException {
+    testBannedMySQLQueryParam("allowLoadLocalInfile=true");
+    testBannedMySQLQueryParamWithValidParam("allowLoadLocalInfile=true");
+  }
+
+  @Test
+  void testValidateConnectionUrlAllowLoadLocal() throws IOException, InterpreterException {
+    testBannedMySQLQueryParam("allowLoadLocal=true");
+  }
+
+  @Test
+  void testValidateConnectionUrlSocketFactory() throws IOException, InterpreterException {
+    // it easier to unit test with H2 but this is really a Postgres issue
+    testBannedH2QueryParam("socketFactory=com.example.MySocketFactory");
   }
 
   @Test
   void testValidateConnectionUrlEncoded() throws IOException, InterpreterException {
+    testBannedMySQLQueryParam("%61llowLoadLocalInfile=true");
+    // also test encoded param with %2561 (%25 is the encoded form of %)
+    testBannedMySQLQueryParam("%2561llowLoadLocalInfile=true");
+  }
+
+  @Test
+  void testValidateConnectionH2UrlWithInit() throws IOException, InterpreterException {
+    testBannedH2QueryParam("INIT=RUNSCRIPT FROM 'http://localhost/init.sql'");
+  }
+
+  @Test
+  void testValidateConnectionMySQLProps() throws IOException, InterpreterException {
+    testBannedURL("com.mysql.cj.jdbc.Driver",
+        "jdbc:mysql://(host=myhost,port=1111,allowLoadLocalInfile=true)/db");
+  }
+
+  @Test
+  void testValidateConnectionMySQLProps2() throws IOException, InterpreterException {
+    testBannedURL("com.mysql.cj.jdbc.Driver",
+        "jdbc:mysql://[(host=myhost,port=1111,allowLoadLocalInfile=true),(host=myhost2)]/db");
+  }
+
+  @Test
+  void testValidateConnectionMySQLProps3() throws IOException, InterpreterException {
+    testBannedURL("com.mysql.cj.jdbc.Driver",
+        "jdbc:mysql://address=(host=172.18.0.1)(port=3309)" +
+            "(%2561llowLoadLocalInfile=true),localhost:3306/test");
+  }
+
+  @Test
+  void testValidateConnectionMySQLWeirdPassword() throws IOException, InterpreterException {
+    // we strongly discourage putting passwords in the URL
+    assertDoesNotThrow(() -> JDBCInterpreter.validateConnectionUrl
+        ("jdbc:mysql://localhost:3306/test?user=xyz&password=(allowLoadLocalInfile)"));
+  }
+
+  private void testBannedH2QueryParam(String param) throws IOException, InterpreterException {
+    testBannedURL("org.h2.Driver", getJdbcConnection() + ";" + param);
+  }
+
+  private void testBannedMySQLQueryParam(String param) throws IOException, InterpreterException {
+    testBannedURL("org.h2.Driver", "jdbc:mysql://localhost/test?" + param);
+  }
+
+  private void testBannedMySQLQueryParamWithValidParam(String param)
+      throws IOException, InterpreterException {
+    testBannedURL("org.h2.Driver", "jdbc:mysql://localhost/test?paranoid=true&" + param);
+  }
+
+  private void testBannedURL(String driver, String url) throws IOException, InterpreterException {
     Properties properties = new Properties();
-    properties.setProperty("default.driver", "org.h2.Driver");
-    properties.setProperty("default.url", getJdbcConnection() + ";%61llowLoadLocalInfile=true");
+    properties.setProperty("default.driver", driver);
+    properties.setProperty("default.url", url);
     properties.setProperty("default.user", "");
     properties.setProperty("default.password", "");
     JDBCInterpreter jdbcInterpreter = new JDBCInterpreter(properties);
     jdbcInterpreter.open();
     InterpreterResult interpreterResult = jdbcInterpreter.interpret("SELECT 1", context);
     assertEquals(InterpreterResult.Code.ERROR, interpreterResult.code());
     assertEquals("Connection URL contains improper configuration",
-            interpreterResult.message().get(0).getData());
+        interpreterResult.message().get(0).getData());
   }
 
   private InterpreterContext getInterpreterContext() {