This is an automated email from the ASF dual-hosted git repository. liyang-kylin pushed a commit to branch kylin5 in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 731b11628cd4fe5506cf7b5cbb82548025bdd311 Author: Li Yang <[email protected]> AuthorDate: Fri Oct 2 12:34:10 2026 +0800 KYLIN-6106 improve os command line argument validation --- .../kylin/common/util/CliCommandExecutor.java | 98 +++++++++++++++++ .../kylin/common/util/CliCommandExecutorTest.java | 30 ++++++ .../apache/kylin/rest/service/AsyncQueryJob.java | 7 +- .../kylin/rest/service/AsyncQueryJobTest.java | 28 +++-- .../kylin/engine/spark/job/SparkSubmitCommand.java | 45 ++++++++ .../spark/job/DefaultSparkBuildJobHandler.java | 110 ++++++++++--------- .../engine/spark/job/NSparkExecutableTest.java | 63 ++++++----- .../engine/spark/job/SparkBuildJobHandlerTest.java | 117 +++++++++------------ 8 files changed, 339 insertions(+), 159 deletions(-) diff --git a/src/core-common/src/main/java/org/apache/kylin/common/util/CliCommandExecutor.java b/src/core-common/src/main/java/org/apache/kylin/common/util/CliCommandExecutor.java index 70d680636a..a8503733c9 100644 --- a/src/core-common/src/main/java/org/apache/kylin/common/util/CliCommandExecutor.java +++ b/src/core-common/src/main/java/org/apache/kylin/common/util/CliCommandExecutor.java @@ -24,6 +24,8 @@ import java.io.IOException; import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.List; +import java.util.Map; import org.apache.commons.io.FileUtils; import org.apache.kylin.common.KylinConfig; @@ -155,6 +157,25 @@ public class CliCommandExecutor { return r; } + public CliCmdExecResult execute(List<String> command, Map<String, String> environment, Logger logAppender, + String jobId) throws ShellException { + String displayCommand = displayCommand(command, environment); + CliCmdExecResult r; + if (remoteHost == null) { + r = runNativeCommand(command, environment, logAppender, jobId); + } else { + val remoteResult = runRemoteCommand(toShellCommand(command, environment), logAppender); + r = new CliCmdExecResult(remoteResult.getFirst(), remoteResult.getSecond(), null); + } + + if (r.getCode() != 0) { + throw new ShellException("OS command error exit with return code: " + r.getCode() // + + ", error message: " + r.getCmd() + "The command is: \n" + displayCommand + + (remoteHost == null ? "" : " (remoteHost:" + remoteHost + ")")); + } + return r; + } + private Pair<Integer, String> runRemoteCommand(String command, Logger logAppender) throws ShellException { try { SSHClient ssh = getSshClient(); @@ -233,6 +254,83 @@ public class CliCommandExecutor { } } + private CliCmdExecResult runNativeCommand(List<String> command, Map<String, String> environment, Logger logAppender, + String jobId) throws ShellException { + int pid = 0; + + try { + ProcessBuilder builder = new ProcessBuilder(command); + builder.environment().putAll(System.getenv()); + if (environment != null) { + builder.environment().putAll(environment); + } + builder.redirectErrorStream(true); + Process proc = builder.start(); + pid = ProcessUtils.getPid(proc); + logger.info("sub process {} on behalf of job {}, start to run...", pid, jobId); + EventBusFactory.getInstance().postSync(new ProcessStart(pid, jobId)); + int maxCommandLineOutputLength = KylinConfig.getInstanceFromEnv().getMaxCommandLineOutputLength(); + int headSize = maxCommandLineOutputLength / 2; + StringBuilderHelper result = StringBuilderHelper.headTail(headSize, + maxCommandLineOutputLength - headSize); + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(proc.getInputStream(), StandardCharsets.UTF_8))) { + String line; + + while ((line = reader.readLine()) != null) { + result.append(line).append('\n'); + if (logAppender != null) { + logAppender.log(line); + } + if (Thread.currentThread().isInterrupted()) { + String msg = displayCommand(command, environment) + " is interrupt"; + logger.warn(msg); + throw new InterruptedException(msg); + } + } + } + + try { + int exitCode = proc.waitFor(); + String b = result.toString(); + if (b.length() > (100 << 20)) { + logger.info("[LESS_LIKELY_THINGS_HAPPENED]Sub process log larger than 100M"); + } + return new CliCmdExecResult(exitCode, b, pid + ""); + + } catch (InterruptedException e) { + logger.warn("Thread is interrupted, cmd: {}, pid: {}", displayCommand(command, environment), pid, e); + Thread.currentThread().interrupt(); + throw e; + } + } catch (Exception e) { + throw new ShellException(e); + } finally { + EventBusFactory.getInstance().postSync(new ProcessFinished(pid)); + } + } + + private static String displayCommand(List<String> command, Map<String, String> environment) { + return toShellCommand(command, environment); + } + + private static String toShellCommand(List<String> command, Map<String, String> environment) { + StringBuilder builder = new StringBuilder(); + if (environment != null) { + for (Map.Entry<String, String> entry : environment.entrySet()) { + builder.append(entry.getKey()).append('=').append(quoteShellArgument(entry.getValue())).append(' '); + } + } + for (String argument : command) { + builder.append(quoteShellArgument(argument)).append(' '); + } + return builder.toString().trim(); + } + + private static String quoteShellArgument(String argument) { + return "'" + argument.replace("'", "'\"'\"'") + "'"; + } + @Setter @Getter @AllArgsConstructor diff --git a/src/core-common/src/test/java/org/apache/kylin/common/util/CliCommandExecutorTest.java b/src/core-common/src/test/java/org/apache/kylin/common/util/CliCommandExecutorTest.java index 7d83931961..dcbe90de6c 100644 --- a/src/core-common/src/test/java/org/apache/kylin/common/util/CliCommandExecutorTest.java +++ b/src/core-common/src/test/java/org/apache/kylin/common/util/CliCommandExecutorTest.java @@ -22,6 +22,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.File; import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Collections; import org.apache.commons.io.FileUtils; import org.apache.kylin.common.KylinConfig; @@ -30,6 +32,7 @@ import org.junit.Assert; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import lombok.val; @@ -134,6 +137,33 @@ public class CliCommandExecutorTest { assertTrue(res.getCmd().startsWith("testhead")); assertTrue(res.getCmd().endsWith("testtail\n")); } + + @Test + void testExecuteWithArgumentsDoesNotInterpretShellMetacharacters() throws Exception { + CliCommandExecutor executor = new CliCommandExecutor(); + CliCommandExecutor.CliCmdExecResult result = executor.execute( + Arrays.asList("/bin/echo", "$(touch /tmp/not-executed)", "`id`", "a;b"), + Collections.singletonMap("TEST_ENV", "value;command"), null, null); + + assertEquals("$(touch /tmp/not-executed) `id` a;b\n", result.getCmd()); + } + + @Test + void testExecuteWithArgumentsQuotesRemoteCommand() throws Exception { + CliCommandExecutor executor = Mockito.spy(new CliCommandExecutor("localhost", "root", null)); + SSHClient sshClient = Mockito.mock(SSHClient.class); + Mockito.doReturn(sshClient).when(executor).getSshClient(); + Mockito.when(sshClient.execCommand(Mockito.anyString(), Mockito.anyInt(), Mockito.any())) + .thenReturn(new SSHClientOutput(0, "ok")); + + executor.execute(Arrays.asList("/bin/echo", "a b", "c'd", "$(unused)"), + Collections.singletonMap("TEST_ENV", "value;command"), null, null); + + ArgumentCaptor<String> commandCaptor = ArgumentCaptor.forClass(String.class); + Mockito.verify(sshClient).execCommand(commandCaptor.capture(), Mockito.eq(3600), Mockito.isNull()); + assertEquals("TEST_ENV='value;command' '/bin/echo' 'a b' 'c'\"'\"'d' '$(unused)'", + commandCaptor.getValue()); + } } class MockSSHClient extends SSHClient { diff --git a/src/query-service/src/main/java/org/apache/kylin/rest/service/AsyncQueryJob.java b/src/query-service/src/main/java/org/apache/kylin/rest/service/AsyncQueryJob.java index 4bb4b4df4e..e81808a22f 100644 --- a/src/query-service/src/main/java/org/apache/kylin/rest/service/AsyncQueryJob.java +++ b/src/query-service/src/main/java/org/apache/kylin/rest/service/AsyncQueryJob.java @@ -60,6 +60,7 @@ import org.apache.kylin.common.util.HadoopUtil; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.engine.spark.job.DefaultSparkBuildJobHandler; import org.apache.kylin.engine.spark.job.NSparkExecutable; +import org.apache.kylin.engine.spark.job.SparkSubmitCommand; import org.apache.kylin.guava30.shaded.common.annotations.VisibleForTesting; import org.apache.kylin.guava30.shaded.common.collect.Maps; import org.apache.kylin.job.common.ExecutableUtil; @@ -108,9 +109,11 @@ public class AsyncQueryJob extends NSparkExecutable { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkJobHandler.generateSparkCmd(KylinConfig.getInstanceFromEnv(), desc); + SparkSubmitCommand cmd = (SparkSubmitCommand) sparkJobHandler.generateSparkCmd( + KylinConfig.getInstanceFromEnv(), desc); CliCommandExecutor exec = getCliCommandExecutor(); - CliCommandExecutor.CliCmdExecResult r = exec.execute(cmd, patternedLogger, getId()); + CliCommandExecutor.CliCmdExecResult r = exec.execute(cmd.getArguments(), cmd.getEnvironment(), + patternedLogger, getId()); return ExecuteResult.createSucceed(r.getCmd()); } catch (Exception e) { return ExecuteResult.createError(e); diff --git a/src/query-service/src/test/java/org/apache/kylin/rest/service/AsyncQueryJobTest.java b/src/query-service/src/test/java/org/apache/kylin/rest/service/AsyncQueryJobTest.java index ab7a50ed40..059f843a7e 100644 --- a/src/query-service/src/test/java/org/apache/kylin/rest/service/AsyncQueryJobTest.java +++ b/src/query-service/src/test/java/org/apache/kylin/rest/service/AsyncQueryJobTest.java @@ -50,6 +50,7 @@ import org.apache.kylin.common.util.HadoopUtil; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.common.util.NLocalFileMetadataTestCase; import org.apache.kylin.common.util.ShellException; +import org.apache.kylin.engine.spark.job.SparkSubmitCommand; import org.apache.kylin.guava30.shaded.common.collect.Maps; import org.apache.kylin.guava30.shaded.common.collect.Sets; import org.apache.kylin.guava30.shaded.common.io.ByteSource; @@ -119,7 +120,7 @@ public class AsyncQueryJobTest extends NLocalFileMetadataTestCase { public void testAsyncQueryJob() throws ExecuteException, JsonProcessingException, ShellException { CliCommandExecutor executor = Mockito.spy(new CliCommandExecutor()); Mockito.doReturn(new CliCommandExecutor.CliCmdExecResult(0, "mock", "mock")).when(executor) - .execute(Mockito.any(), Mockito.any(), Mockito.any()); + .execute(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()); AsyncQueryJob asyncQueryJob = Mockito.spy(new AsyncQueryJob()); Assert.assertNotNull(asyncQueryJob.getCliCommandExecutor()); Mockito.doNothing().when(asyncQueryJob).killOrphanApplicationIfExists(Mockito.any()); @@ -148,8 +149,9 @@ public class AsyncQueryJobTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) this.sparkJobHandler.generateSparkCmd(getConfig(), desc); - return ExecuteResult.createSucceed(cmd); + SparkSubmitCommand cmd = (SparkSubmitCommand) this.sparkJobHandler.generateSparkCmd(getConfig(), + desc); + return ExecuteResult.createSucceed(cmd.getArguments().toString()); } }; asyncQueryJob.setProject(queryParams.getProject()); @@ -167,8 +169,9 @@ public class AsyncQueryJobTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) this.sparkJobHandler.generateSparkCmd(getConfig(), desc); - return ExecuteResult.createSucceed(cmd); + SparkSubmitCommand cmd = (SparkSubmitCommand) this.sparkJobHandler.generateSparkCmd(getConfig(), + desc); + return ExecuteResult.createSucceed(cmd.getArguments().toString()); } }; asyncQueryJob.setProject(queryParams.getProject()); @@ -196,7 +199,8 @@ public class AsyncQueryJobTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) this.sparkJobHandler.generateSparkCmd(getConfig(), desc); + SparkSubmitCommand ignored = (SparkSubmitCommand) this.sparkJobHandler.generateSparkCmd(getConfig(), + desc); return ExecuteResult.createSucceed(appArgs .substring(appArgs.lastIndexOf("file:") + "file:".length(), appArgs.lastIndexOf("/")).trim()); } @@ -301,7 +305,8 @@ public class AsyncQueryJobTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) this.sparkJobHandler.generateSparkCmd(getConfig(), desc); + SparkSubmitCommand ignored = (SparkSubmitCommand) this.sparkJobHandler.generateSparkCmd(getConfig(), + desc); return ExecuteResult.createSucceed( appArgs.substring(appArgs.lastIndexOf("file:") + "file:".length(), appArgs.lastIndexOf("/")) .trim()); @@ -332,15 +337,16 @@ public class AsyncQueryJobTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) this.sparkJobHandler.generateSparkCmd(getConfig(), desc); - return ExecuteResult.createSucceed(cmd); + SparkSubmitCommand cmd = (SparkSubmitCommand) this.sparkJobHandler.generateSparkCmd(getConfig(), + desc); + return ExecuteResult.createSucceed(cmd.getArguments().toString()); } }; asyncQueryJob.setProject(queryParams.getProject()); ExecuteResult executeResult = asyncQueryJob.submit(queryParams); Assert.assertTrue(executeResult.succeed()); - Assert.assertTrue(executeResult.output().contains("--conf 'spark.executor.memory=513m'")); - Assert.assertTrue(executeResult.output().contains("--conf 'spark.executor.cores=3'")); + Assert.assertTrue(executeResult.output().contains("spark.executor.memory=513m")); + Assert.assertTrue(executeResult.output().contains("spark.executor.cores=3")); } } diff --git a/src/spark-project/engine-build-sdk/src/main/java/org/apache/kylin/engine/spark/job/SparkSubmitCommand.java b/src/spark-project/engine-build-sdk/src/main/java/org/apache/kylin/engine/spark/job/SparkSubmitCommand.java new file mode 100644 index 0000000000..951b9b2818 --- /dev/null +++ b/src/spark-project/engine-build-sdk/src/main/java/org/apache/kylin/engine/spark/job/SparkSubmitCommand.java @@ -0,0 +1,45 @@ +/* + * 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.engine.spark.job; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public class SparkSubmitCommand { + + private final List<String> arguments; + private final Map<String, String> environment; + + public SparkSubmitCommand(List<String> arguments, Map<String, String> environment) { + this.arguments = Collections.unmodifiableList(new ArrayList<>(arguments)); + this.environment = Collections.unmodifiableMap(new LinkedHashMap<>(environment)); + } + + public List<String> getArguments() { + return arguments; + } + + public Map<String, String> getEnvironment() { + return environment; + } +} + diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/DefaultSparkBuildJobHandler.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/DefaultSparkBuildJobHandler.java index 458cd9145a..53c88a17a4 100644 --- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/DefaultSparkBuildJobHandler.java +++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/DefaultSparkBuildJobHandler.java @@ -55,14 +55,12 @@ import lombok.val; public class DefaultSparkBuildJobHandler implements ISparkJobHandler { private static final Logger logger = LoggerFactory.getLogger(DefaultSparkBuildJobHandler.class); - private static final String SPACE = " "; - private static final String SUBMIT_LINE_FORMAT = " \\\n"; private static final String SPARK_JARS_1 = "spark.jars"; private static final String SPARK_JARS_2 = "spark.yarn.dist.jars"; private static final String SPARK_FILES_1 = "spark.files"; private static final String SPARK_FILES_2 = "spark.yarn.dist.files"; - private static final String EQUALS = "="; + private static final Pattern COMMAND_INJECTION_PATTERN = Pattern.compile("(`[^`]*+`)|(\\$\\([^)]*+)"); @Override public void killOrphanApplicationIfExists(String project, String jobStepId, KylinConfig config, @@ -123,66 +121,75 @@ public class DefaultSparkBuildJobHandler implements ISparkJobHandler { @Override public Object generateSparkCmd(KylinConfig config, SparkAppDescription desc) { - // Hadoop conf dir. - StringBuilder cmdBuilder = new StringBuilder("export HADOOP_CONF_DIR="); - cmdBuilder.append(desc.getHadoopConfDir()); - cmdBuilder.append(SPACE).append("&&"); - - // Spark submit. - cmdBuilder.append(SPACE).append(KylinConfigBase.getSparkHome()).append(File.separator); - cmdBuilder.append("bin/spark-submit"); - cmdBuilder.append(SUBMIT_LINE_FORMAT); + List<String> arguments = Lists.newArrayList(); + arguments.add(KylinConfigBase.getSparkHome() + File.separator + "bin/spark-submit"); // Application main class. - cmdBuilder.append(SPACE).append("--class"); - cmdBuilder.append(SPACE).append("org.apache.kylin.engine.spark.application.SparkEntry"); - cmdBuilder.append(SUBMIT_LINE_FORMAT); + arguments.add("--class"); + arguments.add("org.apache.kylin.engine.spark.application.SparkEntry"); // Application name. - cmdBuilder.append(SPACE).append("--name"); - cmdBuilder.append(SPACE).append(desc.getJobNamePrefix()).append(desc.getJobId()); - cmdBuilder.append(SUBMIT_LINE_FORMAT); + arguments.add("--name"); + arguments.add(desc.getJobNamePrefix() + desc.getJobId()); // Spark jars. - cmdBuilder.append(SPACE).append("--jars"); - cmdBuilder.append(SPACE).append(String.join(desc.getComma(), desc.getSparkJars())); - cmdBuilder.append(SUBMIT_LINE_FORMAT); + if (!desc.getSparkJars().isEmpty()) { + arguments.add("--jars"); + arguments.add(String.join(desc.getComma(), desc.getSparkJars())); + } - cmdBuilder.append(SPACE).append("--files"); - cmdBuilder.append(SPACE).append(String.join(desc.getComma(), desc.getSparkFiles())); - cmdBuilder.append(SUBMIT_LINE_FORMAT); + if (!desc.getSparkFiles().isEmpty()) { + arguments.add("--files"); + arguments.add(String.join(desc.getComma(), desc.getSparkFiles())); + } // Spark conf. // Maybe we would rewrite some confs, like 'extraJavaOptions', 'extraClassPath', // and the confs rewrited should be removed from props thru #modifyDump. - wrapSparkConf(cmdBuilder, desc.getSparkConf()); + wrapSparkConf(arguments, desc.getSparkConf()); // Application jar. KylinJobJar is the application-jar (of spark-submit), // path to a bundled jar including your application and all dependencies, // The URL must be globally visible inside of your cluster, // for instance, an hdfs:// path or a file:// path that is present on all nodes. - cmdBuilder.append(SPACE).append(desc.getKylinJobJar()); - cmdBuilder.append(SUBMIT_LINE_FORMAT); + arguments.add(desc.getKylinJobJar()); // Application parameter file. - cmdBuilder.append(SPACE).append(desc.getAppArgs()); + if (StringUtils.isNotBlank(desc.getAppArgs())) { + for (String arg : StringUtils.split(desc.getAppArgs())) { + arguments.add(arg); + } + } - final String command = cmdBuilder.toString(); - logger.info("spark submit cmd: {}", command); + Map<String, String> environment = Maps.newHashMap(); + if (StringUtils.isNotBlank(desc.getHadoopConfDir())) { + environment.put("HADOOP_CONF_DIR", desc.getHadoopConfDir()); + } + checkCommandInjection(arguments, environment.values()); - // Safe check. - checkCommandInjection(command); - return command; + logger.info("spark submit cmd: {}", arguments); + return new SparkSubmitCommand(arguments, environment); } - private void checkCommandInjection(String command) { - if (Objects.isNull(command)) { - return; - } + private void checkCommandInjection(List<String> arguments, Iterable<String> environmentValues) { List<String> illegals = Lists.newArrayList(); - Matcher matcher = Pattern.compile("(`[^`]*+`)|(\\$\\([^)]*+)").matcher(command); - while (matcher.find()) { - illegals.add(matcher.group()); + for (String argument : arguments) { + if (argument == null) { + continue; + } + Matcher matcher = COMMAND_INJECTION_PATTERN.matcher(argument); + while (matcher.find()) { + illegals.add(matcher.group()); + } + } + for (String environmentValue : environmentValues) { + if (environmentValue == null) { + continue; + } + Matcher matcher = COMMAND_INJECTION_PATTERN.matcher(environmentValue); + while (matcher.find()) { + illegals.add(matcher.group()); + } } if (illegals.isEmpty()) { @@ -197,7 +204,7 @@ public class DefaultSparkBuildJobHandler implements ISparkJobHandler { throw new IllegalArgumentException(msg); } - private void wrapSparkConf(StringBuilder cmdBuilder, Map<String, String> sparkConf) { + private void wrapSparkConf(List<String> arguments, Map<String, String> sparkConf) { for (Map.Entry<String, String> entry : sparkConf.entrySet()) { switch (entry.getKey()) { // Avoid duplicated from '--jars' @@ -209,21 +216,15 @@ public class DefaultSparkBuildJobHandler implements ISparkJobHandler { // Do nothing. break; default: - appendSparkConf(cmdBuilder, entry.getKey(), entry.getValue()); + appendSparkConf(arguments, entry.getKey(), entry.getValue()); break; } } } - protected void appendSparkConf(StringBuilder sb, String confKey, String confValue) { - // Multiple parameters in "--conf" need to be enclosed in single quotes - if (confKey.contains("'") || confValue.contains("'")) { - throw new IllegalArgumentException(String.format(Locale.ROOT, - "Spark conf key or value contains invalid single quote: key=%s, value=%s", - confKey, confValue)); - } - sb.append(" --conf '").append(confKey).append(EQUALS).append(confValue).append("' "); - sb.append(SUBMIT_LINE_FORMAT); + protected void appendSparkConf(List<String> arguments, String confKey, String confValue) { + arguments.add("--conf"); + arguments.add(confKey + "=" + confValue); } @Override @@ -232,7 +233,12 @@ public class DefaultSparkBuildJobHandler implements ISparkJobHandler { try { val patternedLogger = new BufferedLogger(logger); CliCommandExecutor exec = new CliCommandExecutor(); - CliCommandExecutor.CliCmdExecResult r = exec.execute((String) cmd, patternedLogger, parentId); + if (!(cmd instanceof SparkSubmitCommand)) { + throw new IllegalArgumentException("Unsupported spark submit command type"); + } + SparkSubmitCommand sparkSubmitCommand = (SparkSubmitCommand) cmd; + CliCommandExecutor.CliCmdExecResult r = exec.execute(sparkSubmitCommand.getArguments(), + sparkSubmitCommand.getEnvironment(), patternedLogger, parentId); if (StringUtils.isNotEmpty(r.getProcessId())) { updateInfo.put("process_id", r.getProcessId()); } diff --git a/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/NSparkExecutableTest.java b/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/NSparkExecutableTest.java index 897827c277..edc8a81e58 100644 --- a/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/NSparkExecutableTest.java +++ b/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/NSparkExecutableTest.java @@ -18,12 +18,15 @@ package org.apache.kylin.engine.spark.job; +import java.io.File; import java.util.Objects; import java.util.Random; import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.fs.Path; import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.KylinConfigBase; +import org.apache.kylin.common.StorageURL; import org.apache.kylin.common.persistence.transaction.UnitOfWork; import org.apache.kylin.common.util.NLocalFileMetadataTestCase; import org.apache.kylin.common.util.RandomUtil; @@ -35,18 +38,13 @@ import org.apache.kylin.query.plugin.profiler.QueryAsyncProfilerSparkPlugin; import org.junit.After; import org.junit.Assert; import org.junit.Before; -import org.junit.Rule; import org.junit.Test; -import org.junit.rules.TemporaryFolder; import lombok.val; import lombok.var; public class NSparkExecutableTest extends NLocalFileMetadataTestCase { - @Rule - public TemporaryFolder temporaryFolder = new TemporaryFolder(); - private NDataModelManager modelManager; @Before @@ -63,10 +61,9 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { @Test public void attachMetadataAndKylinProps() throws Exception { KylinConfig config = getTestConfig(); - val junitFolder = temporaryFolder.getRoot(); - val path = junitFolder.getAbsolutePath(); + StorageURL metaStoreUrl = config.getJobTmpMetaStoreUrl("default", RandomUtil.randomUUIDStr()); MockSparkTestExecutable executable = new MockSparkTestExecutable(); - executable.setMetaUrl(path); + executable.setMetaUrl(metaStoreUrl.toString()); executable.setProject("default"); Assert.assertEquals(8, executable.getMetadataDumpList(config).size()); NDataModel model = modelManager.getDataModelDesc("82fa7671-a935-45f5-8779-85703601f49a"); @@ -74,7 +71,8 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { new Thread(new AddModelRunner(model)).start(); } executable.attachMetadataAndKylinProps(config); - Assert.assertEquals(2, Objects.requireNonNull(junitFolder.listFiles()).length); + File dumpedMetaDir = new File(new Path(metaStoreUrl.getParameter("path")).toUri().getPath()); + Assert.assertEquals(2, Objects.requireNonNull(dumpedMetaDir.listFiles()).length); } class AddModelRunner implements Runnable { @@ -122,15 +120,17 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); + SparkSubmitCommand cmd = (SparkSubmitCommand) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, + desc); Assert.assertNotNull(cmd); - Assert.assertTrue(cmd.contains("spark-submit")); + Assert.assertTrue(cmd.getArguments().get(0).contains("spark-submit")); Assert.assertTrue( - cmd.contains("log4j.configurationFile=file:" + kylinConfig.getLogSparkDriverPropertiesFile())); - Assert.assertTrue(cmd.contains("spark.executor.extraClassPath=job.jar")); - Assert.assertTrue(cmd.contains("spark.driver.log4j.appender.hdfs.File=")); - Assert.assertTrue(cmd.contains("kylin.hdfs.working.dir=")); + cmd.getArguments().toString() + .contains("log4j.configurationFile=file:" + kylinConfig.getLogSparkDriverPropertiesFile())); + Assert.assertTrue(cmd.getArguments().toString().contains("spark.executor.extraClassPath=job.jar")); + Assert.assertTrue(cmd.getArguments().toString().contains("spark.driver.log4j.appender.hdfs.File=")); + Assert.assertTrue(cmd.getArguments().toString().contains("kylin.hdfs.working.dir=")); } overwriteSystemProp("kylin.engine.extra-jars-path", "/this_new_path.jar"); @@ -139,10 +139,11 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); + SparkSubmitCommand cmd = (SparkSubmitCommand) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, + desc); Assert.assertNotNull(cmd); - Assert.assertTrue(cmd.contains("/this_new_path.jar")); + Assert.assertTrue(cmd.getArguments().toString().contains("/this_new_path.jar")); } // Spark plugin @@ -152,10 +153,12 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); + SparkSubmitCommand cmd = (SparkSubmitCommand) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, + desc); Assert.assertNotNull(cmd); - Assert.assertTrue(cmd.contains("spark.plugins=," + BuildAsyncProfilerSparkPlugin.class.getCanonicalName())); + Assert.assertTrue(cmd.getArguments().toString() + .contains("spark.plugins=," + BuildAsyncProfilerSparkPlugin.class.getCanonicalName())); } overwriteSystemProp("kylin.engine.spark-conf.spark.plugins", @@ -165,11 +168,13 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); + SparkSubmitCommand cmd = (SparkSubmitCommand) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, + desc); Assert.assertNotNull(cmd); - Assert.assertTrue(cmd.contains("spark.plugins=" + QueryAsyncProfilerSparkPlugin.class.getCanonicalName() - + "," + BuildAsyncProfilerSparkPlugin.class.getCanonicalName())); + Assert.assertTrue(cmd.getArguments().toString() + .contains("spark.plugins=" + QueryAsyncProfilerSparkPlugin.class.getCanonicalName() + "," + + BuildAsyncProfilerSparkPlugin.class.getCanonicalName())); } overwriteSystemProp("kylin.engine.async-profiler-enabled", "false"); @@ -178,11 +183,12 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); + SparkSubmitCommand cmd = (SparkSubmitCommand) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, + desc); Assert.assertNotNull(cmd); - Assert.assertFalse( - cmd.contains("spark.plugins=," + BuildAsyncProfilerSparkPlugin.class.getCanonicalName())); + Assert.assertFalse(cmd.getArguments().toString() + .contains("spark.plugins=," + BuildAsyncProfilerSparkPlugin.class.getCanonicalName())); } overwriteSystemProp("kylin.engine.spark-conf.spark.driver.extraJavaOptions", @@ -193,9 +199,10 @@ public class NSparkExecutableTest extends NLocalFileMetadataTestCase { desc.setHadoopConfDir(hadoopConf); desc.setKylinJobJar(kylinJobJar); desc.setAppArgs(appArgs); - String cmd = (String) sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); - } catch (IllegalArgumentException iae) { - Assert.assertTrue(iae.getMessage().contains("Not allowed to specify injected command")); + sparkExecutable.sparkJobHandler.generateSparkCmd(kylinConfig, desc); + Assert.fail("Should reject command substitution in spark config"); + } catch (IllegalArgumentException e) { + Assert.assertTrue(e.getMessage().contains("Not allowed to specify injected command")); } } } diff --git a/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/SparkBuildJobHandlerTest.java b/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/SparkBuildJobHandlerTest.java index 8cd046134e..6219856196 100644 --- a/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/SparkBuildJobHandlerTest.java +++ b/src/spark-project/engine-spark/src/test/java/org/apache/kylin/engine/spark/job/SparkBuildJobHandlerTest.java @@ -20,12 +20,16 @@ package org.apache.kylin.engine.spark.job; import static org.apache.kylin.engine.spark.job.NSparkExecutable.SPARK_MASTER; -import java.lang.reflect.Method; +import java.io.File; +import java.util.Arrays; +import java.util.List; import java.util.Map; import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.KylinConfigBase; import org.apache.kylin.common.util.ClassUtil; import org.apache.kylin.engine.spark.NLocalWithSparkSessionTestBase; +import org.apache.kylin.guava30.shaded.common.collect.Lists; import org.apache.kylin.guava30.shaded.common.collect.Maps; import org.apache.kylin.guava30.shaded.common.collect.Sets; import org.apache.kylin.job.exception.ExecuteException; @@ -78,73 +82,28 @@ public class SparkBuildJobHandlerTest extends NLocalWithSparkSessionTestBase { KylinConfig config = getTestConfig(); ISparkJobHandler handler = (ISparkJobHandler) ClassUtil.newInstance(config.getSparkBuildJobHandlerClassName()); Assert.assertTrue(handler instanceof DefaultSparkBuildJobHandler); - String cmd = ""; + SparkSubmitCommand cmd = new SparkSubmitCommand(Arrays.asList("/bin/echo", "spark-submit"), Maps.newHashMap()); Map<String, String> updateInfo = handler.runSparkSubmit(cmd, ""); - Assert.assertEquals(cmd, updateInfo.get("output")); + Assert.assertEquals("spark-submit\n", updateInfo.get("output")); Assert.assertNotNull(updateInfo.get("process_id")); } @Test - public void testAppendSparkConfRejectSingleQuote() { + public void testAppendSparkConf() { DefaultSparkBuildJobHandler handler = new DefaultSparkBuildJobHandler(); - StringBuilder sb = new StringBuilder(); - handler.appendSparkConf(sb, "spark.yarn.queue", "normalQueue"); - Assert.assertTrue(sb.toString().contains("--conf 'spark.yarn.queue=normalQueue'")); + List<String> arguments = Lists.newArrayList(); + handler.appendSparkConf(arguments, "spark.yarn.queue", "normalQueue"); + Assert.assertEquals(Arrays.asList("--conf", "spark.yarn.queue=normalQueue"), arguments); - // conf value containing single quote must be rejected - try { - handler.appendSparkConf(new StringBuilder(), "spark.yarn.queue", - "default'; touch /tmp/pwned; echo '"); - Assert.fail("Should have thrown IllegalArgumentException for value containing single quote"); - } catch (IllegalArgumentException e) { - Assert.assertTrue(e.getMessage().contains("single quote")); - } - - // pipe without single quote is acceptable - StringBuilder sb2 = new StringBuilder(); - handler.appendSparkConf(sb2, "spark.yarn.queue", "a|b"); - Assert.assertTrue(sb2.toString().contains("--conf 'spark.yarn.queue=a|b'")); - } - - @Test - public void testCheckCommandInjectionBlocked() throws Exception { - DefaultSparkBuildJobHandler handler = new DefaultSparkBuildJobHandler(); - Method method = DefaultSparkBuildJobHandler.class.getDeclaredMethod("checkCommandInjection", String.class); - method.setAccessible(true); - - String[] blockedPayloads = { - "--conf 'spark.yarn.queue=`id`'", - "--conf 'spark.yarn.queue=$(whoami)'", - }; - - for (String payload : blockedPayloads) { - try { - method.invoke(handler, payload); - Assert.fail("Should have thrown IllegalArgumentException for payload: " + payload); - } catch (java.lang.reflect.InvocationTargetException e) { - Assert.assertTrue("Payload not blocked: " + payload, - e.getCause() instanceof IllegalArgumentException); - } - } - } - - @Test - public void testCheckCommandInjectionAllowed() throws Exception { - DefaultSparkBuildJobHandler handler = new DefaultSparkBuildJobHandler(); - Method method = DefaultSparkBuildJobHandler.class.getDeclaredMethod("checkCommandInjection", String.class); - method.setAccessible(true); - - String[] allowedPayloads = { - "--conf 'spark.yarn.queue=default' \\\n--conf 'spark.executor.memory=1024m'", - "--conf 'spark.driver.extraJavaOptions=-Dconfig=value'", - "--conf 'spark.yarn.queue=normal_queue.v1'", - }; + arguments.clear(); + handler.appendSparkConf(arguments, "spark.yarn.queue", "default'; touch /tmp/pwned; echo '"); + Assert.assertEquals(Arrays.asList("--conf", "spark.yarn.queue=default'; touch /tmp/pwned; echo '"), arguments); - for (String payload : allowedPayloads) { - method.invoke(handler, payload); - } + arguments.clear(); + handler.appendSparkConf(arguments, "spark.yarn.queue", "a|b"); + Assert.assertEquals(Arrays.asList("--conf", "spark.yarn.queue=a|b"), arguments); } @Test @@ -162,8 +121,8 @@ public class SparkBuildJobHandlerTest extends NLocalWithSparkSessionTestBase { desc.setJobNamePrefix("test_"); desc.setJobId("test-job-id"); desc.setComma(","); - desc.setSparkJars(Sets.newHashSet("jar1.jar")); - desc.setSparkFiles(Sets.newHashSet("file1.conf")); + desc.setSparkJars(Sets.newHashSet("jar1.jar;touch /tmp/pwned")); + desc.setSparkFiles(Sets.newHashSet("file1.conf|cat /tmp/pwned")); Map<String, String> sparkConf = Maps.newHashMap(); sparkConf.put("spark.yarn.queue", "default'; touch /tmp/pwned; echo '"); @@ -171,11 +130,37 @@ public class SparkBuildJobHandlerTest extends NLocalWithSparkSessionTestBase { desc.setSparkConf(sparkConf); ISparkJobHandler handler = new DefaultSparkBuildJobHandler(); - try { - handler.generateSparkCmd(config, desc); - Assert.fail("Should have thrown IllegalArgumentException for conf value containing single quote"); - } catch (IllegalArgumentException e) { - Assert.assertTrue(e.getMessage().contains("single quote")); - } + SparkSubmitCommand cmd = (SparkSubmitCommand) handler.generateSparkCmd(config, desc); + Assert.assertTrue(cmd.getArguments().contains("spark.yarn.queue=default'; touch /tmp/pwned; echo '")); + Assert.assertTrue(cmd.getArguments().contains("jar1.jar;touch /tmp/pwned")); + Assert.assertTrue(cmd.getArguments().contains("file1.conf|cat /tmp/pwned")); + } + + @Test + public void testGenerateSparkCmdPreservesArgumentOrder() { + SparkAppDescription desc = new SparkAppDescription(); + desc.setHadoopConfDir("/etc/hadoop"); + desc.setKylinJobJar("job.jar"); + desc.setAppArgs("-className org.example.Main file:/tmp/args.json"); + desc.setJobNamePrefix("job_"); + desc.setJobId("id"); + desc.setComma(","); + desc.setSparkJars(Sets.newLinkedHashSet(Arrays.asList("a.jar", "b.jar"))); + desc.setSparkFiles(Sets.newLinkedHashSet(Arrays.asList("a.conf", "b.conf"))); + + Map<String, String> sparkConf = Maps.newLinkedHashMap(); + sparkConf.put("spark.executor.memory", "1024m"); + sparkConf.put("spark.yarn.queue", "default"); + desc.setSparkConf(sparkConf); + + SparkSubmitCommand cmd = (SparkSubmitCommand) new DefaultSparkBuildJobHandler().generateSparkCmd(getTestConfig(), + desc); + + Assert.assertEquals(Arrays.asList(KylinConfigBase.getSparkHome() + File.separator + "bin/spark-submit", + "--class", "org.apache.kylin.engine.spark.application.SparkEntry", "--name", "job_id", "--jars", + "a.jar,b.jar", "--files", "a.conf,b.conf", "--conf", "spark.executor.memory=1024m", "--conf", + "spark.yarn.queue=default", "job.jar", "-className", "org.example.Main", "file:/tmp/args.json"), + cmd.getArguments()); + Assert.assertEquals("/etc/hadoop", cmd.getEnvironment().get("HADOOP_CONF_DIR")); } }
