[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,105 @@
+/*
+ * 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.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.log.SensitiveDataConverter;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+
+import org.apache.http.NameValuePair;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.message.BasicNameValuePair;
+
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+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=" + sanitizeAddress(parameters.getAddress())
+                + ", taskId=" + sanitizeMessage(parameters.getTaskId())
+                + ", online=" + parameters.isOnline()
+                + ", localParamKeys=" + summarizeLocalParamKeys(parameters.getLocalParams());
+    }
+
+    static String sanitizeAddress(final Object address) {
+        if (address == null) {
+            return null;
+        }
+        try {
+            URIBuilder uriBuilder = new URIBuilder(String.valueOf(address));
+            uriBuilder.setUserInfo(null);
+            List<NameValuePair> sanitizedParameters = new ArrayList<>();
+            for (NameValuePair parameter : uriBuilder.getQueryParams()) {
+                String value = isSensitiveAddressParameter(parameter.getName())
+                        ? TaskConstants.SENSITIVE_DATA_MASK
+                        : parameter.getValue();
+                sanitizedParameters.add(new BasicNameValuePair(parameter.getName(), value));
+            }
+            uriBuilder.setParameters(sanitizedParameters);
+            return sanitizeMessage(uriBuilder.build());
+        } catch (URISyntaxException | IllegalArgumentException ex) {
+            return TaskConstants.SENSITIVE_DATA_MASK;
+        }
+    }
+
+    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 boolean isSensitiveAddressParameter(final String name) {
+        return "username".equalsIgnoreCase(name)
+                || "password".equalsIgnoreCase(name)
+                || "token".equalsIgnoreCase(name);
+    }
+
+    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;
+    }
+}