mask secrets in MM task command log (#10128)

* mask secrets in MM task command log

* unit test for masked iterator

* checkstyle fix

indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java

@@ -28,6 +28,7 @@ import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
 import com.google.common.io.ByteSink;
 import com.google.common.io.ByteSource;
 import com.google.common.io.ByteStreams;
@@ -58,6 +59,7 @@ import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.query.DruidMetrics;
 import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.log.StartupLoggingConfig;
 import org.apache.druid.server.metrics.MonitorsConfig;
 import org.apache.druid.tasklogs.TaskLogPusher;
 import org.apache.druid.tasklogs.TaskLogStreamer;
@@ -95,6 +97,7 @@ public class ForkingTaskRunner
   private final DruidNode node;
   private final ListeningExecutorService exec;
   private final PortFinder portFinder;
+  private final StartupLoggingConfig startupLoggingConfig;
 
   private volatile boolean stopping = false;
 
@@ -106,7 +109,8 @@ public class ForkingTaskRunner
       Properties props,
       TaskLogPusher taskLogPusher,
       ObjectMapper jsonMapper,
-      @Self DruidNode node
+      @Self DruidNode node,
+      StartupLoggingConfig startupLoggingConfig
   )
   {
     super(jsonMapper, taskConfig);
@@ -115,6 +119,7 @@ public class ForkingTaskRunner
     this.taskLogPusher = taskLogPusher;
     this.node = node;
     this.portFinder = new PortFinder(config.getStartPort(), config.getEndPort(), config.getPorts());
+    this.startupLoggingConfig = startupLoggingConfig;
     this.exec = MoreExecutors.listeningDecorator(
         Execs.multiThreaded(workerConfig.getCapacity(), "forking-task-runner-%d")
     );
@@ -338,7 +343,7 @@ public class ForkingTaskRunner
                           jsonMapper.writeValue(taskFile, task);
                         }
 
-                        LOGGER.info("Running command: %s", Joiner.on(" ").join(command));
+                        LOGGER.info("Running command: %s", getMaskedCommand(startupLoggingConfig.getMaskProperties(), command));
                         taskWorkItem.processHolder = new ProcessHolder(
                           new ProcessBuilder(ImmutableList.copyOf(command)).redirectErrorStream(true).start(),
                           logFile,
@@ -627,6 +632,23 @@ public class ForkingTaskRunner
     }
   }
 
+  String getMaskedCommand(List<String> maskedProperties, List<String> command)
+  {
+    final Set<String> maskedPropertiesSet = Sets.newHashSet(maskedProperties);
+    final Iterator<String> maskedIterator = command.stream().map(element -> {
+      String[] splits = element.split("=", 2);
+      if (splits.length == 2) {
+        for (String masked : maskedPropertiesSet) {
+          if (splits[0].contains(masked)) {
+            return StringUtils.format("%s=%s", splits[0], "<masked>");
+          }
+        }
+      }
+      return element;
+    }).iterator();
+    return Joiner.on(" ").join(maskedIterator);
+  }
+
   protected static class ForkingTaskRunnerWorkItem extends TaskRunnerWorkItem
   {
     private final Task task;

indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerFactory.java

@@ -26,6 +26,7 @@ import org.apache.druid.indexing.common.config.TaskConfig;
 import org.apache.druid.indexing.overlord.config.ForkingTaskRunnerConfig;
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.server.DruidNode;
+import org.apache.druid.server.log.StartupLoggingConfig;
 import org.apache.druid.tasklogs.TaskLogPusher;
 
 import java.util.Properties;
@@ -41,6 +42,7 @@ public class ForkingTaskRunnerFactory implements TaskRunnerFactory<ForkingTaskRu
   private final ObjectMapper jsonMapper;
   private final TaskLogPusher persistentTaskLogs;
   private final DruidNode node;
+  private final StartupLoggingConfig startupLoggingConfig;
 
   @Inject
   public ForkingTaskRunnerFactory(
@@ -50,7 +52,8 @@ public class ForkingTaskRunnerFactory implements TaskRunnerFactory<ForkingTaskRu
       final Properties props,
       final ObjectMapper jsonMapper,
       final TaskLogPusher persistentTaskLogs,
-      @Self DruidNode node
+      @Self DruidNode node,
+      final StartupLoggingConfig startupLoggingConfig
   )
   {
     this.config = config;
@@ -60,11 +63,12 @@ public class ForkingTaskRunnerFactory implements TaskRunnerFactory<ForkingTaskRu
     this.jsonMapper = jsonMapper;
     this.persistentTaskLogs = persistentTaskLogs;
     this.node = node;
+    this.startupLoggingConfig = startupLoggingConfig;
   }
 
   @Override
   public ForkingTaskRunner build()
   {
-    return new ForkingTaskRunner(config, taskConfig, workerConfig, props, persistentTaskLogs, jsonMapper, node);
+    return new ForkingTaskRunner(config, taskConfig, workerConfig, props, persistentTaskLogs, jsonMapper, node, startupLoggingConfig);
   }
 }

indexing-service/src/test/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerTest.java

@@ -22,9 +22,16 @@ package org.apache.druid.indexing.overlord;
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterators;
+import org.apache.druid.indexing.overlord.config.ForkingTaskRunnerConfig;
+import org.apache.druid.indexing.worker.config.WorkerConfig;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.server.log.StartupLoggingConfig;
+import org.assertj.core.util.Lists;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.List;
+
 public class ForkingTaskRunnerTest
 {
   // This tests the test to make sure the test fails when it should.
@@ -121,4 +128,29 @@ public class ForkingTaskRunnerTest
         ImmutableList.copyOf(new QuotableWhiteSpaceSplitter(Joiner.on(" ").join(strings)))
     );
   }
+
+  @Test
+  public void testMaskedIterator()
+  {
+    Pair<List<String>, String> originalAndExpectedCommand = new Pair<>(
+        Lists.list(
+            "java -cp",
+            "/path/to/somewhere:some-jars.jar",
+            "/some===file",
+            "/asecretFileNa=me", // this should not be masked but there is not way to know this not a property and probably this is an unrealistic scenario anyways
+            "-Dsome.property=random",
+            "-Dsome.otherproperty = random=random",
+            "-Dsome.somesecret = secretvalue",
+            "-Dsome.somesecret=secretvalue",
+            "-Dsome.somepassword = secret=value",
+            "-Dsome.some=notasecret",
+            "-Dsome.otherSecret= =asfdhkj352872598====fasdlkjfa="
+            ),
+        "java -cp /path/to/somewhere:some-jars.jar /some===file /asecretFileNa=<masked> -Dsome.property=random -Dsome.otherproperty = random=random " +
+            "-Dsome.somesecret =<masked> -Dsome.somesecret=<masked> -Dsome.somepassword =<masked> -Dsome.some=notasecret -Dsome.otherSecret=<masked>"
+    );
+    StartupLoggingConfig startupLoggingConfig = new StartupLoggingConfig();
+    ForkingTaskRunner forkingTaskRunner = new ForkingTaskRunner(new ForkingTaskRunnerConfig(), null, new WorkerConfig(), null, null, null, null, startupLoggingConfig);
+    Assert.assertEquals(originalAndExpectedCommand.rhs, forkingTaskRunner.getMaskedCommand(startupLoggingConfig.getMaskProperties(), originalAndExpectedCommand.lhs));
+  }
 }