Apache Kylin: The remote code execution via jdbc url

@@ -39,6 +39,10 @@
             <groupId>org.apache.kylin</groupId>
             <artifactId>kylin-tool</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kylin</groupId>
+            <artifactId>kylin-datasource-sdk</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-aop</artifactId>

@@ -80,7 +80,6 @@
 import org.apache.kylin.common.scheduler.EventBusFactory;
 import org.apache.kylin.common.scheduler.SourceUsageUpdateNotifier;
 import org.apache.kylin.common.util.EncryptUtil;
-import org.apache.kylin.common.util.JdbcUtils;
 import org.apache.kylin.common.util.JsonUtil;
 import org.apache.kylin.common.util.Pair;
 import org.apache.kylin.common.util.SetThreadName;
@@ -135,6 +134,7 @@
 import org.apache.kylin.rest.security.KerberosLoginManager;
 import org.apache.kylin.rest.service.task.QueryHistoryMetaUpdateScheduler;
 import org.apache.kylin.rest.util.AclEvaluate;
+import org.apache.kylin.sdk.datasource.framework.utils.JdbcUtils;
 import org.apache.kylin.streaming.manager.StreamingJobManager;
 import org.apache.kylin.tool.garbage.MetadataCleaner;
 import org.slf4j.Logger;
@@ -446,6 +446,13 @@ private void updateProjectOverrideKylinProps(String project, Map<String, String>
                 throw new KylinException(INVALID_PARAMETER,
                         MsgPicker.getMsg().getIllegalNegative(KYLIN_JOB_MAX_CONCURRENT_JOBS));
         }
+        if (overrideKylinProps.containsKey(KYLIN_SOURCE_JDBC_CONNECTION_URL_KEY)) {
+            String url = overrideKylinProps.get(KYLIN_SOURCE_JDBC_CONNECTION_URL_KEY);
+            if (KylinConfig.getInstanceFromEnv().isSourceJdbcWhiteListEnabled()
+                    && !JdbcUtils.validateUrlByWhiteList(url)) {
+                throw new KylinException(INVALID_JDBC_SOURCE_CONFIG, MsgPicker.getMsg().getJdbcConnectionInfoWrong());
+            }
+        }
         encryptJdbcPassInOverrideKylinProps(overrideKylinProps);
         projectManager.updateProject(project, copyForWrite -> copyForWrite.getOverrideKylinProps()
                 .putAll(KylinConfig.trimKVFromMap(overrideKylinProps)));

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.rest.source;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kylin.sdk.datasource.security.AbstractJdbcSourceConnectionValidator;
+import org.springframework.web.util.UriComponents;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@NoArgsConstructor
+@Slf4j
+public class CommonJdbcSourceConnectionValidator extends AbstractJdbcSourceConnectionValidator {
+
+    private static final String JDBC_COLON = "jdbc:";
+
+    private boolean parsed = false;
+    private String scheme;
+    private String host;
+    private int port;
+    private String path;
+    private Map<String, List<String>> queryParams;
+
+    @Override
+    public boolean isValid() {
+        if (!parsed) {
+            try {
+                parseUrl();
+            } catch (Exception e) {
+                log.error("Error on parseUrl", e);
+                return false;
+            }
+        }
+        // Only query param keys need to be validated currently
+        return validateQueryParamKeys();
+    }
+
+    private boolean validateQueryParamKeys() {
+        Set<String> validUrlParamKeys = settings.getValidUrlParamKeys();
+        Set<String> userInputKeys = queryParams.keySet();
+        return validUrlParamKeys.containsAll(userInputKeys);
+    }
+
+    private void parseUrl() {
+        if (parsed) {
+            return;
+        }
+        if (StringUtils.isBlank(url)) {
+            throw new IllegalStateException("url cannot be empty");
+        }
+        if (!url.startsWith(JDBC_COLON)) {
+            throw new IllegalStateException("url must start with " + JDBC_COLON);
+        }
+
+        String noPrefixUrl = url.substring(JDBC_COLON.length());
+        UriComponents uri = UriComponentsBuilder.fromUriString(noPrefixUrl).build();
+        scheme = uri.getScheme();
+        host = uri.getHost();
+        port = uri.getPort();
+        path = uri.getPath();
+        queryParams = uri.getQueryParams();
+
+        parsed = true;
+    }
+}

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.rest.source;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.celeborn.shaded.com.google.common.collect.Sets;
+import org.apache.kylin.sdk.datasource.security.JdbcSourceValidationSettings;
+import org.junit.Test;
+
+public class CommonJdbcSourceConnectionValidatorTest {
+
+    @Test
+    public void testValidate() {
+        JdbcSourceValidationSettings settings = JdbcSourceValidationSettings.builder()
+                .validUrlParamKeys(Sets.newHashSet("q1", "q2", "q3"))
+                .build();
+
+        {
+            CommonJdbcSourceConnectionValidator validator = new CommonJdbcSourceConnectionValidator();
+            validator.settings(settings).url("jdbc:mysql://localhost:123/data_db?q1=v1&q2=v2&q3=v3");
+            assertTrue(validator.isValid());
+        }
+        {
+            CommonJdbcSourceConnectionValidator validator = new CommonJdbcSourceConnectionValidator();
+            validator.settings(settings).url("jdbc:mysql://localhost:123/data_db?q1=v1&q2=v2&q4=v4");
+            assertFalse(validator.isValid());
+        }
+    }
+
+    @Test
+    public void testValidateFailed() {
+        JdbcSourceValidationSettings settings = JdbcSourceValidationSettings.builder()
+                .validUrlParamKeys(Sets.newHashSet("q1", "q2", "q3"))
+                .build();
+
+        {
+            CommonJdbcSourceConnectionValidator validator = new CommonJdbcSourceConnectionValidator();
+            validator.settings(settings).url("");
+            assertFalse(validator.isValid());
+        }
+        {
+            CommonJdbcSourceConnectionValidator validator = new CommonJdbcSourceConnectionValidator();
+            validator.settings(settings).url("mysql://localhost:123/data_db");
+            assertFalse(validator.isValid());
+        }
+    }
+}

@@ -4466,4 +4466,38 @@ public long getV3DeltaLogCacheExpireThreshold() {
         return TimeUtil.timeStringAs(getOptional("kylin.query.v3.delta-log-cache-expire-threshold", "43200s"),
                 TimeUnit.SECONDS);
     }
+
+    public boolean isSourceJdbcWhiteListEnabled() {
+        return Boolean.parseBoolean(getOptional("kylin.source.jdbc.white-list.enabled", FALSE));
+    }
+
+    public Set<String> getSourceJdbcWhiteListSchemes() {
+        String config = StringUtils.deleteWhitespace(getOptional("kylin.source.jdbc.white-list.schemes", ""));
+        if (StringUtils.isBlank(config)) {
+            return Collections.emptySet();
+        }
+        return Sets.newHashSet(config.split(","));
+    }
+
+    public String getSourceJdbcWhiteListValidatorClassByScheme(String scheme) {
+        Set<String> whiteListSchemes = getSourceJdbcWhiteListSchemes();
+        if (!whiteListSchemes.contains(scheme)) {
+            return null;
+        }
+        return getOptional(String.format(Locale.ROOT, "kylin.source.jdbc.white-list.%s.validator-class", scheme),
+                "org.apache.kylin.rest.source.CommonJdbcSourceConnectionValidator");
+    }
+
+    public Set<String> getSourceJdbcWhiteListUrlParamKeysByScheme(String scheme) {
+        Set<String> whiteListSchemes = getSourceJdbcWhiteListSchemes();
+        if (!whiteListSchemes.contains(scheme)) {
+            return Collections.emptySet();
+        }
+        String config = StringUtils.deleteWhitespace(getOptional(String.format(Locale.ROOT,
+                "kylin.source.jdbc.white-list.%s.url-param-keys", scheme), ""));
+        if (StringUtils.isBlank(config)) {
+            return Collections.emptySet();
+        }
+        return Sets.newHashSet(config.split(","));
+    }
 }

@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kylin.common.util;
-
-import java.sql.Connection;
-import java.util.Properties;
-
-import org.apache.commons.dbcp2.BasicDataSourceFactory;
-
-import lombok.SneakyThrows;
-import lombok.val;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-
-public class JdbcUtils {
-    @SneakyThrows
-    public static boolean checkConnectionParameter(String driver, String url, String username, String password) {
-        Properties connProp = new Properties();
-        connProp.put("driverClassName", driver);
-        connProp.put("url", url);
-        connProp.put("username", username);
-        connProp.put("password", password);
-        try (val dataSource = BasicDataSourceFactory.createDataSource(connProp);
-                Connection conn = dataSource.getConnection()) {
-            return true;
-        } catch (Exception e) {
-            log.debug("jdbc connect check failed", e);
-            return false;
-        }
-
-    }
-}

@@ -17,14 +17,19 @@
  */
 package org.apache.kylin.sdk.datasource.framework;
 
+import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_JDBC_SOURCE_CONFIG;
+
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.kylin.common.KylinConfig;
 import org.apache.kylin.common.KylinConfigExt;
+import org.apache.kylin.common.exception.KylinException;
+import org.apache.kylin.common.msg.MsgPicker;
 import org.apache.kylin.sdk.datasource.adaptor.AdaptorConfig;
 import org.apache.kylin.sdk.datasource.adaptor.DefaultAdaptor;
 import org.apache.kylin.sdk.datasource.adaptor.MysqlAdaptor;
+import org.apache.kylin.sdk.datasource.framework.utils.JdbcUtils;
 
 public class SourceConnectorFactory {
     private SourceConnectorFactory() {
@@ -37,6 +42,11 @@ public static JdbcConnector getJdbcConnector(KylinConfig config) {
         String jdbcPass = config.getJdbcPass();
         String adaptorClazz = config.getJdbcAdaptorClass();
 
+        if (KylinConfig.getInstanceFromEnv().isSourceJdbcWhiteListEnabled()
+                && !JdbcUtils.validateUrlByWhiteList(jdbcUrl)) {
+            throw new KylinException(INVALID_JDBC_SOURCE_CONFIG, MsgPicker.getMsg().getJdbcConnectionInfoWrong());
+        }
+
         Map<String, String> options = new HashMap<>();
         if (config instanceof KylinConfigExt) {
             options = ((KylinConfigExt) config).getExtendedOverrides();

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.sdk.datasource.framework.utils;
+
+import java.sql.Connection;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.commons.dbcp2.BasicDataSourceFactory;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.sdk.datasource.security.JdbcSourceConnectionValidator;
+import org.apache.kylin.sdk.datasource.security.JdbcSourceValidationSettings;
+
+import lombok.SneakyThrows;
+import lombok.val;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class JdbcUtils {
+
+    public static final Pattern JDBC_SCHEMA_PATTERN = Pattern.compile("(?<=jdbc:)[\\w\\-]+(?=:)");
+
+    @SneakyThrows
+    public static boolean checkConnectionParameter(String driver, String url, String username, String password) {
+        Properties connProp = new Properties();
+        connProp.put("driverClassName", driver);
+        connProp.put("url", url);
+        connProp.put("username", username);
+        connProp.put("password", password);
+
+        if (KylinConfig.getInstanceFromEnv().isSourceJdbcWhiteListEnabled() && !validateUrlByWhiteList(url)) {
+            log.warn("jdbc url white list check failed");
+            return false;
+        }
+
+        try (val dataSource = BasicDataSourceFactory.createDataSource(connProp);
+                Connection conn = dataSource.getConnection()) {
+            return true;
+        } catch (Exception e) {
+            log.warn("jdbc connect check failed", e);
+            return false;
+        }
+    }
+
+    public static boolean validateUrlByWhiteList(String url) {
+        try {
+            KylinConfig config = KylinConfig.getInstanceFromEnv();
+            String scheme = null;
+
+            Matcher m = JDBC_SCHEMA_PATTERN.matcher(url);
+            if (m.find()) {
+                scheme = m.group();
+            }
+            if (StringUtils.isBlank(scheme) || !config.getSourceJdbcWhiteListSchemes().contains(scheme)) {
+                return false;
+            }
+
+            JdbcSourceValidationSettings settings = JdbcSourceValidationSettings.builder()
+                    .validUrlParamKeys(config.getSourceJdbcWhiteListUrlParamKeysByScheme(scheme)).build();
+
+            JdbcSourceConnectionValidator validator = (JdbcSourceConnectionValidator) ClassUtil
+                    .newInstance(config.getSourceJdbcWhiteListValidatorClassByScheme(scheme));
+
+            return validator.settings(settings).url(url).isValid();
+        } catch (Exception e) {
+            log.error("Error on validate url", e);
+            return false;
+        }
+    }
+}

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.sdk.datasource.security;
+
+public abstract class AbstractJdbcSourceConnectionValidator implements JdbcSourceConnectionValidator {
+
+    protected JdbcSourceValidationSettings settings;
+    protected String url;
+
+    @Override
+    public JdbcSourceConnectionValidator settings(JdbcSourceValidationSettings settings) {
+        this.settings = settings;
+        return this;
+    }
+
+    @Override
+    public JdbcSourceConnectionValidator url(String url) {
+        this.url = url;
+        return this;
+    }
+}

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.sdk.datasource.security;
+
+public interface JdbcSourceConnectionValidator {
+
+    JdbcSourceConnectionValidator settings(JdbcSourceValidationSettings settings);
+
+    JdbcSourceConnectionValidator url(String url);
+
+    boolean isValid();
+}

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.sdk.datasource.security;
+
+import java.util.Set;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+
+@Getter
+@Builder
+@AllArgsConstructor
+public final class JdbcSourceValidationSettings {
+
+    private Set<String> validUrlParamKeys;
+}

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.sdk.datasource.framework.utils;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.kylin.common.util.NLocalFileMetadataTestCase;
+import org.apache.kylin.sdk.datasource.security.AbstractJdbcSourceConnectionValidator;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class JdbcUtilsTest extends NLocalFileMetadataTestCase {
+
+    @Before
+    public void setUp() throws Exception {
+        createTestMetadata();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        cleanupTestMetadata();
+    }
+
+    @Test
+    public void testValidateUrlByWhiteList_schemeCheck() {
+        overwriteSystemProp("kylin.source.jdbc.white-list.schemes", "mysql, postgresql");
+        overwriteSystemProp("kylin.source.jdbc.white-list.mysql.validator-class",
+                "org.apache.kylin.sdk.datasource.framework.utils.JdbcUtilsTest$MockJdbcSourceConnectionValidator");
+        overwriteSystemProp("kylin.source.jdbc.white-list.postgresql.validator-class",
+                "org.apache.kylin.sdk.datasource.framework.utils.JdbcUtilsTest$MockJdbcSourceConnectionValidator");
+
+        // valid cases
+        assertTrue(JdbcUtils.validateUrlByWhiteList("jdbc:mysql://localhost:3306/db"));
+        assertTrue(JdbcUtils.validateUrlByWhiteList("jdbc:postgresql://localhost:5433/db"));
+
+        // invalid cases
+        assertFalse(JdbcUtils.validateUrlByWhiteList("xxx://localhost:3306/db"));
+        assertFalse(JdbcUtils.validateUrlByWhiteList("jdbc:mongodb://localhost:1234/db"));
+    }
+
+    public static class MockJdbcSourceConnectionValidator extends AbstractJdbcSourceConnectionValidator {
+        @Override
+        public boolean isValid() {
+            return true;
+        }
+    }
+}