[ZEPPELIN-5990] Disable sensitive configuration for JDBC url (#4709)

* [ZEPPELIN-5990] Disable sensitive configuration for JDBC url

* [ZEPPELIN-5990] Disable sensitive configuration for JDBC url

Author: jongyoul

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

@@ -153,6 +153,14 @@ public class JDBCInterpreter extends KerberosInterpreter {
           "KerberosConfigPath", "KerberosKeytabPath", "KerberosCredentialCachePath",
           "extraCredentials", "roles", "sessionProperties"));
 
+  private static final String ALLOW_LOAD_LOCAL_IN_FILE_NAME = "allowLoadLocalInfile";
+
+  private static final String AUTO_DESERIALIZE = "autoDeserialize";
+
+  private static final String ALLOW_LOCAL_IN_FILE_NAME = "allowLocalInfile";
+
+  private static final String ALLOW_URL_IN_LOCAL_IN_FILE_NAME = "allowUrlInLocalInfile";
+
   // database --> Properties
   private final HashMap<String, Properties> basePropertiesMap;
   // username --> User Configuration
@@ -533,6 +541,7 @@ public Connection getConnection(InterpreterContext context)
     String url = properties.getProperty(URL_KEY);
     url = appendProxyUserToURL(url, user);
     String connectionUrl = appendTagsToURL(url, context);
+    validateConnectionUrl(connectionUrl);
 
     String authType = getProperty("zeppelin.jdbc.auth.type", "SIMPLE")
             .trim().toUpperCase();
@@ -576,6 +585,15 @@ public Connection getConnection(InterpreterContext context)
     return connection;
   }
 
+  private void validateConnectionUrl(String url) {
+    if (containsIgnoreCase(url, ALLOW_LOAD_LOCAL_IN_FILE_NAME) ||
+        containsIgnoreCase(url, AUTO_DESERIALIZE) ||
+        containsIgnoreCase(url, ALLOW_LOCAL_IN_FILE_NAME) ||
+        containsIgnoreCase(url, ALLOW_URL_IN_LOCAL_IN_FILE_NAME)) {
+      throw new IllegalArgumentException("Connection URL contains sensitive configuration");
+    }
+  }
+
   private String appendProxyUserToURL(String url, String user) {
     StringBuilder connectionUrl = new StringBuilder(url);
 
@@ -749,6 +767,9 @@ private InterpreterResult executeSql(String sql,
 
     try {
       connection = getConnection(context);
+    } catch (IllegalArgumentException e) {
+      LOGGER.error("Cannot run " + sql, e);
+      return new InterpreterResult(Code.ERROR, "Connection URL contains improper configuration");
     } catch (Exception e) {
       LOGGER.error("Fail to getConnection", e);
       try {
@@ -763,7 +784,7 @@ private InterpreterResult executeSql(String sql,
       }
     }
     if (connection == null) {
-      return new InterpreterResult(Code.ERROR, "User's connectin not found.");
+      return new InterpreterResult(Code.ERROR, "User's connection not found.");
     }
 
     try {

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

@@ -747,6 +747,21 @@ void testSplitSqlQueryWithComments() throws IOException,
     assertEquals(3, resultMessages.size());
   }
 
+  @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());
+  }
+
   private InterpreterContext getInterpreterContext() {
     return InterpreterContext.builder()
             .setAuthenticationInfo(new AuthenticationInfo("testUser"))