[Feature-17398][Task] Avoid leaking sensitive Dinky task logs

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java

@@ -37,6 +37,18 @@ public class SensitiveDataConverter extends MessageConverter {
     private static Pattern multilinePattern;
     private static final Set<String> maskPatterns = new HashSet<>();
 
+    private static final String KNOWN_SENSITIVE_CONFIGURATION_KEY_REGEX =
+            "(?:password|access[._-]?key(?:[._-]?(?:id|secret))?|secret[._-]?access[._-]?key|secret[._-]?key)";
+
+    private static final Pattern QUOTED_SENSITIVE_CONFIGURATION_PREFIX_PATTERN = Pattern.compile(
+            "((?:\\\\?\\\"|')?" + KNOWN_SENSITIVE_CONFIGURATION_KEY_REGEX
+                    + "(?:\\\\?\\\"|')?\\s*(?::|=)\\s*(\\\\?\\\"|'))",
+            Pattern.CASE_INSENSITIVE | Pattern.MULTILINE);
+
+    private static final Pattern UNQUOTED_SENSITIVE_CONFIGURATION_PATTERN = Pattern.compile(
+            "(" + KNOWN_SENSITIVE_CONFIGURATION_KEY_REGEX + "\\s*(?::|=)\\s*)([^\\s,;&}'\\\"]+)()",
+            Pattern.CASE_INSENSITIVE | Pattern.MULTILINE);
+
     static {
         addMaskPattern(TaskConstants.DATASOURCE_PASSWORD_REGEX);
     }
@@ -64,6 +76,11 @@ public static String maskSensitiveData(final String logMsg) {
             return logMsg;
         }
 
+        String maskedLogMsg = maskKnownSensitiveConfiguration(logMsg);
+        return maskByConfiguredPatterns(maskedLogMsg);
+    }
+
+    private static String maskByConfiguredPatterns(final String logMsg) {
         final StringBuffer sb = new StringBuffer(logMsg.length());
         final Matcher matcher = multilinePattern.matcher(logMsg);
 
@@ -75,4 +92,64 @@ public static String maskSensitiveData(final String logMsg) {
         return sb.toString();
     }
 
+    private static String maskKnownSensitiveConfiguration(final String logMsg) {
+        String maskedLogMsg = maskQuotedSensitiveConfiguration(logMsg);
+        return maskKnownSensitiveConfiguration(maskedLogMsg, UNQUOTED_SENSITIVE_CONFIGURATION_PATTERN);
+    }
+
+    private static String maskQuotedSensitiveConfiguration(final String logMsg) {
+        final StringBuilder sb = new StringBuilder(logMsg.length());
+        final Matcher matcher = QUOTED_SENSITIVE_CONFIGURATION_PREFIX_PATTERN.matcher(logMsg);
+        int appendFrom = 0;
+        int searchFrom = 0;
+        while (matcher.find(searchFrom)) {
+            final String quote = matcher.group(2);
+            final int closingQuoteStart = findClosingQuote(logMsg, matcher.end(), quote);
+            if (closingQuoteStart < 0) {
+                searchFrom = matcher.end();
+                continue;
+            }
+            final int closingQuoteEnd = closingQuoteStart + quote.length();
+            sb.append(logMsg, appendFrom, matcher.end());
+            sb.append(TaskConstants.SENSITIVE_DATA_MASK);
+            sb.append(logMsg, closingQuoteStart, closingQuoteEnd);
+            appendFrom = closingQuoteEnd;
+            searchFrom = closingQuoteEnd;
+        }
+        sb.append(logMsg, appendFrom, logMsg.length());
+        return sb.toString();
+    }
+
+    private static int findClosingQuote(final String logMsg, final int valueStart, final String quote) {
+        final char quoteCharacter = quote.charAt(quote.length() - 1);
+        final boolean escapedQuote = quote.length() == 2 && quote.charAt(0) == '\\';
+        for (int i = valueStart; i < logMsg.length(); i++) {
+            if (logMsg.charAt(i) != quoteCharacter) {
+                continue;
+            }
+            int precedingBackslashes = 0;
+            for (int j = i - 1; j >= valueStart && logMsg.charAt(j) == '\\'; j--) {
+                precedingBackslashes++;
+            }
+            if (escapedQuote && precedingBackslashes % 4 == 1) {
+                return i - 1;
+            }
+            if (!escapedQuote && precedingBackslashes % 2 == 0) {
+                return i;
+            }
+        }
+        return -1;
+    }
+
+    private static String maskKnownSensitiveConfiguration(final String logMsg, final Pattern pattern) {
+        final StringBuffer sb = new StringBuffer(logMsg.length());
+        final Matcher matcher = pattern.matcher(logMsg);
+        while (matcher.find()) {
+            matcher.appendReplacement(sb,
+                    Matcher.quoteReplacement(matcher.group(1) + TaskConstants.SENSITIVE_DATA_MASK + matcher.group(3)));
+        }
+        matcher.appendTail(sb);
+        return sb.toString();
+    }
+
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java

@@ -19,6 +19,7 @@
 
 import static org.apache.dolphinscheduler.common.constants.Constants.K8S_CONFIG_REGEX;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 
 import java.util.HashMap;
 
@@ -124,4 +125,41 @@ void testK8SLogMsgConverter() {
 
         assertEquals(maskMsg, maskedLog);
     }
+
+    @Test
+    void testMaskObjectStorageCredentialsInCommonFormats() {
+        HashMap<String, String> tcs = new HashMap<>();
+        tcs.put("{\"accessKeyId\":\"AKIA_TEST\",\"accessKeySecret\":\"SECRET_TEST\",\"bucket\":\"ds\"}",
+                "{\"accessKeyId\":\"******\",\"accessKeySecret\":\"******\",\"bucket\":\"ds\"}");
+        tcs.put("connectionParams=\"{\\\"accessKeyId\\\":\\\"AKIA_TEST\\\",\\\"accessKeySecret\\\":\\\"SECRET_TEST\\\"}\"",
+                "connectionParams=\"{\\\"accessKeyId\\\":\\\"******\\\",\\\"accessKeySecret\\\":\\\"******\\\"}\"");
+        tcs.put("OssConnection{accessKeyId='AKIA_TEST', accessKeySecret='SECRET_TEST', endPoint='oss-cn'}",
+                "OssConnection{accessKeyId='******', accessKeySecret='******', endPoint='oss-cn'}");
+        tcs.put("remote.logging.oss.access.key.id=AKIA_TEST&remote.logging.oss.access.key.secret=SECRET_TEST&bucket=ds",
+                "remote.logging.oss.access.key.id=******&remote.logging.oss.access.key.secret=******&bucket=ds");
+        tcs.put("resource.aws.access.key.id=AKIA_TEST\nresource.aws.secret.access.key=SECRET_TEST\nresource.aws.region=cn-north-1",
+                "resource.aws.access.key.id=******\nresource.aws.secret.access.key=******\nresource.aws.region=cn-north-1");
+
+        for (String logMsg : tcs.keySet()) {
+            assertEquals(tcs.get(logMsg), SensitiveDataConverter.maskSensitiveData(logMsg));
+        }
+    }
+
+    @Test
+    void testMaskQuotedSensitiveValueContainingEscapedQuote() {
+        String logMsg = "{\"password\":\"abc\\\"VISIBLE_SUFFIX\",\"bucket\":\"ds\"}";
+        String expected = "{\"password\":\"******\",\"bucket\":\"ds\"}";
+
+        assertEquals(expected, SensitiveDataConverter.maskSensitiveData(logMsg));
+    }
+
+    @Test
+    void testDoesNotMaskNonSensitiveWordsContainingKeyOrSecret() {
+        String logMsg = "secretary=alice, monkey=banana, keystore=/tmp/ks, access=public, bucket=ds";
+
+        final String maskedLog = SensitiveDataConverter.maskSensitiveData(logMsg);
+
+        assertEquals(logMsg, maskedLog);
+        assertFalse(maskedLog.contains("******"));
+    }
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java

@@ -0,0 +1,71 @@
+/*
+ * 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.dolphinscheduler.plugin.task.dinky;
+
+import org.apache.dolphinscheduler.plugin.task.api.log.SensitiveDataConverter;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+final class DinkyLogSanitizer {
+
+    private DinkyLogSanitizer() {
+        throw new UnsupportedOperationException("Utility class");
+    }
+
+    static String summarizeParameters(final DinkyParameters parameters) {
+        if (parameters == null) {
+            return "null";
+        }
+        return "address=" + sanitizeMessage(parameters.getAddress())
+                + ", taskId=" + sanitizeMessage(parameters.getTaskId())
+                + ", online=" + parameters.isOnline()
+                + ", localParamKeys=" + summarizeLocalParamKeys(parameters.getLocalParams());
+    }
+
+    static String summarizeVariables(final Map<String, String> variables) {
+        if (variables == null || variables.isEmpty()) {
+            return "size=0, keys=[]";
+        }
+        Set<String> keys = new TreeSet<>(variables.keySet());
+        return "size=" + variables.size() + ", keys=" + keys;
+    }
+
+    static String sanitizeMessage(final Object message) {
+        if (message == null) {
+            return null;
+        }
+        return SensitiveDataConverter.maskSensitiveData(String.valueOf(message));
+    }
+
+    private static Set<String> summarizeLocalParamKeys(final List<Property> localParams) {
+        Set<String> keys = new TreeSet<>();
+        if (localParams == null) {
+            return keys;
+        }
+        for (Property property : localParams) {
+            if (property != null && property.getProp() != null) {
+                keys.add(property.getProp());
+            }
+        }
+        return keys;
+    }
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java

@@ -80,7 +80,7 @@ public List<String> getApplicationIds() throws TaskException {
     public void init() {
         final String taskParams = taskExecutionContext.getTaskParams();
         this.dinkyParameters = JSONUtils.parseObject(taskParams, DinkyParameters.class);
-        log.info("Initialize dinky task params: {}", JSONUtils.toPrettyJsonString(dinkyParameters));
+        log.info("Initialize dinky task: {}", DinkyLogSanitizer.summarizeParameters(dinkyParameters));
         if (this.dinkyParameters == null || !this.dinkyParameters.checkParameters()) {
             throw new DinkyTaskException("dinky task params is not valid");
         }
@@ -160,10 +160,10 @@ private void submitApplicationV1() {
                             result.get(apiResultDataKey).get(DinkyTaskConstants.API_RESULT_JOB_INSTANCE_ID).asText();
                 }
             } else {
-                log.error(DinkyTaskConstants.SUBMIT_FAILED_MSG + "{}", result.get(DinkyTaskConstants.API_RESULT_MSG));
+                String errorMessage = DinkyLogSanitizer.sanitizeMessage(result.get(DinkyTaskConstants.API_RESULT_MSG));
+                log.error(DinkyTaskConstants.SUBMIT_FAILED_MSG + "{}", errorMessage);
                 setExitStatusCode(EXIT_CODE_FAILURE);
-                throw new TaskException(
-                        DinkyTaskConstants.SUBMIT_FAILED_MSG + result.get(DinkyTaskConstants.API_RESULT_MSG));
+                throw new TaskException(DinkyTaskConstants.SUBMIT_FAILED_MSG + errorMessage);
             }
         } catch (Exception ex) {
             Thread.currentThread().interrupt();
@@ -199,8 +199,7 @@ public void trackApplicationStatusV0() throws TaskException {
                         // Use address-taskId as app id
                         setAppIds(String.format(DinkyTaskConstants.APPIDS_FORMAT, address, taskId));
                         setExitStatusCode(exitStatusCode);
-                        log.info("dinky task finished with results: {}",
-                                jobInstanceInfoResult.get(apiResultDatasKey));
+                        log.info("dinky task finished, status: {}", jobInstanceStatus);
                         finishFlag = true;
                         break;
                     case DinkyTaskConstants.STATUS_FAILED:
@@ -250,8 +249,7 @@ public void trackApplicationStatusV1() throws TaskException {
                         // Use address-taskId as app id
                         setAppIds(String.format(DinkyTaskConstants.APPIDS_FORMAT, address, taskId));
                         setExitStatusCode(exitStatusCode);
-                        log.info("dinky task finished with results: {}",
-                                jobInstanceInfoResult.get(apiResultDataKey));
+                        log.info("dinky task finished, status: {}", jobInstanceStatus);
                         finishFlag = true;
                         break;
                     case DinkyTaskConstants.STATUS_FAILED:
@@ -312,7 +310,7 @@ private boolean checkResultV1(JsonNode result) {
 
     private void errorHandle(Object msg) {
         setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);
-        log.error("dinky task submit failed with error: {}", msg);
+        log.error("dinky task submit failed with error: {}", DinkyLogSanitizer.sanitizeMessage(msg));
     }
 
     @Override
@@ -352,7 +350,7 @@ private Map<String, String> generateVariables() {
                 }
             }
         }
-        log.info("sending variables to dinky: {}", variables);
+        log.info("sending variables to dinky: {}", DinkyLogSanitizer.summarizeVariables(variables));
         return variables;
     }
 
@@ -400,13 +398,13 @@ private JsonNode getJobInstanceInfo(String address, String taskId) {
 
     private JsonNode parse(String res) {
         ObjectMapper mapper = new ObjectMapper();
-        JsonNode result = null;
         try {
-            result = mapper.readTree(res);
+            return mapper.readTree(res);
         } catch (JsonProcessingException e) {
-            log.error("dinky task submit failed with error", e);
+            throw new DinkyTaskException(
+                    "dinky task response parse failed, responseLength: " + StringUtils.length(res) + ", errorType: "
+                            + e.getClass().getSimpleName());
         }
-        return result;
     }
 
     private String doGet(String url, Map<String, String> params) {
@@ -426,7 +424,8 @@ private String doGet(String url, Map<String, String> params) {
             HttpResponse response = httpClient.execute(httpGet);
             if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                 result = EntityUtils.toString(response.getEntity());
-                log.info("dinky task succeed with results: {}", result);
+                log.info("dinky task request succeeded, statusCode: {}, responseLength: {}",
+                        response.getStatusLine().getStatusCode(), result.length());
             } else {
                 log.error("dinky task terminated,response: {}", response);
             }
@@ -455,7 +454,8 @@ private String sendJsonStr(String url, String params) {
             HttpResponse response = httpClient.execute(httpPost);
             if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                 result = EntityUtils.toString(response.getEntity());
-                log.info("dinky task succeed with results: {}", result);
+                log.info("dinky task request succeeded, statusCode: {}, responseLength: {}",
+                        response.getStatusLine().getStatusCode(), result.length());
             } else {
                 log.error("dinky task terminated,response: {}", response);
             }