From e602c68bdb57a1a5f53bc29dbacc162b6e896c2f Mon Sep 17 00:00:00 2001 From: Yanjun Qiu <153984347+qiuyanjun888@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:26:04 +0800 Subject: [PATCH 1/4] [Fix-17398][Task] Avoid leaking sensitive Dinky task logs --- .../task/api/log/SensitiveDataConverter.java | 33 ++++++ .../api/log/SensitiveDataConverterTest.java | 30 +++++ .../plugin/task/dinky/DinkyLogSanitizer.java | 71 +++++++++++ .../plugin/task/dinky/DinkyTask.java | 27 +++-- .../task/dinky/DinkyLogSanitizerTest.java | 111 ++++++++++++++++++ 5 files changed, 259 insertions(+), 13 deletions(-) create mode 100644 dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java create mode 100644 dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java index ea70aaef7678..8f5314a1d30b 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java +++ b/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 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_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 = maskByConfiguredPatterns(logMsg); + return maskKnownSensitiveConfiguration(maskedLogMsg); + } + + private static String maskByConfiguredPatterns(final String logMsg) { final StringBuffer sb = new StringBuffer(logMsg.length()); final Matcher matcher = multilinePattern.matcher(logMsg); @@ -75,4 +92,20 @@ public static String maskSensitiveData(final String logMsg) { return sb.toString(); } + private static String maskKnownSensitiveConfiguration(final String logMsg) { + String maskedLogMsg = maskKnownSensitiveConfiguration(logMsg, QUOTED_SENSITIVE_CONFIGURATION_PATTERN); + return maskKnownSensitiveConfiguration(maskedLogMsg, UNQUOTED_SENSITIVE_CONFIGURATION_PATTERN); + } + + 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(); + } + } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java index b1d242c1105a..25af2a2989f0 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java +++ b/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,33 @@ void testK8SLogMsgConverter() { assertEquals(maskMsg, maskedLog); } + + @Test + void testMaskObjectStorageCredentialsInCommonFormats() { + HashMap 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 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("******")); + } } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java new file mode 100644 index 000000000000..80367efd9e5c --- /dev/null +++ b/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 variables) { + if (variables == null || variables.isEmpty()) { + return "size=0, keys=[]"; + } + Set 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 summarizeLocalParamKeys(final List localParams) { + Set keys = new TreeSet<>(); + if (localParams == null) { + return keys; + } + for (Property property : localParams) { + if (property != null && property.getProp() != null) { + keys.add(property.getProp()); + } + } + return keys; + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java index ca531261472b..4c7bd9b0c72d 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java @@ -80,7 +80,7 @@ public List 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 generateVariables() { } } } - log.info("sending variables to dinky: {}", variables); + log.info("sending variables to dinky: {}", DinkyLogSanitizer.summarizeVariables(variables)); return variables; } @@ -404,7 +402,8 @@ private JsonNode parse(String res) { try { result = mapper.readTree(res); } catch (JsonProcessingException e) { - log.error("dinky task submit failed with error", e); + log.error("dinky task response parse failed, responseLength: {}, errorType: {}", + StringUtils.length(res), e.getClass().getSimpleName()); } return result; } @@ -426,7 +425,8 @@ private String doGet(String url, Map 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 +455,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); } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java new file mode 100644 index 000000000000..7c22c878b754 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java @@ -0,0 +1,111 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.model.Property; + +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; + +class DinkyLogSanitizerTest { + + @Test + void testSummarizeVariablesDoesNotExposeValues() { + Map variables = new HashMap<>(); + variables.put("accessKeyId", "AKIA_TEST"); + variables.put("accessKeySecret", "SECRET_TEST"); + variables.put("businessDate", "2026-07-06"); + + String summary = DinkyLogSanitizer.summarizeVariables(variables); + + assertTrue(summary.contains("size=3")); + assertTrue(summary.contains("accessKeyId")); + assertTrue(summary.contains("accessKeySecret")); + assertTrue(summary.contains("businessDate")); + assertFalse(summary.contains("AKIA_TEST")); + assertFalse(summary.contains("SECRET_TEST")); + assertFalse(summary.contains("2026-07-06")); + } + + @Test + void testSummarizeParametersDoesNotExposeLocalParamValues() { + DinkyParameters parameters = new DinkyParameters(); + parameters.setAddress("http://dinky:8888"); + parameters.setTaskId("1001"); + parameters.setOnline(true); + parameters.setLocalParams(Arrays.asList( + new Property("accessKeyId", null, null, "AKIA_TEST"), + new Property("businessDate", null, null, "2026-07-06"))); + + String summary = DinkyLogSanitizer.summarizeParameters(parameters); + + assertTrue(summary.contains("address=http://dinky:8888")); + assertTrue(summary.contains("taskId=1001")); + assertTrue(summary.contains("online=true")); + assertTrue(summary.contains("localParamKeys=[accessKeyId, businessDate]")); + assertFalse(summary.contains("AKIA_TEST")); + assertFalse(summary.contains("2026-07-06")); + } + + @Test + void testSanitizeMessageMasksSensitiveValues() { + String message = "{\"accessKeyId\":\"AKIA_TEST\",\"accessKeySecret\":\"SECRET_TEST\"}"; + + String sanitized = DinkyLogSanitizer.sanitizeMessage(message); + + assertTrue(sanitized.contains("accessKeyId")); + assertTrue(sanitized.contains("accessKeySecret")); + assertFalse(sanitized.contains("AKIA_TEST")); + assertFalse(sanitized.contains("SECRET_TEST")); + } + + @Test + void testSanitizeMessageMasksSensitiveValuesInJsonNodeMessage() { + String message = "{\"accessKeyId\":\"AKIA_TEST\",\"accessKeySecret\":\"SECRET_TEST\"}"; + + String sanitized = DinkyLogSanitizer.sanitizeMessage(JsonNodeFactory.instance.textNode(message)); + + assertTrue(sanitized.contains("accessKeyId")); + assertTrue(sanitized.contains("accessKeySecret")); + assertFalse(sanitized.contains("AKIA_TEST")); + assertFalse(sanitized.contains("SECRET_TEST")); + } + + @Test + void testParseMalformedResponseReturnsNullWithoutExposingRawResponse() throws Exception { + DinkyTask task = new DinkyTask(new TaskExecutionContext()); + Method parseMethod = DinkyTask.class.getDeclaredMethod("parse", String.class); + parseMethod.setAccessible(true); + + Object result = parseMethod.invoke(task, + "malformed response accessKeyId=AKIA_TEST accessKeySecret=SECRET_TEST"); + + assertNull(result); + } +} From ffe340bc12d1c6b82385bf9dbedd6872f005edfd Mon Sep 17 00:00:00 2001 From: Yanjun Qiu <153984347+qiuyanjun888@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:35:35 +0800 Subject: [PATCH 2/4] [Feature-17398][Task] Fail explicitly on Dinky response parse errors --- .../plugin/task/dinky/DinkyTask.java | 9 ++++----- .../plugin/task/dinky/DinkyLogSanitizerTest.java | 16 +++++++++++----- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java index 4c7bd9b0c72d..d1cf05e8891b 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java @@ -398,14 +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 response parse failed, responseLength: {}, errorType: {}", - StringUtils.length(res), e.getClass().getSimpleName()); + throw new DinkyTaskException( + "dinky task response parse failed, responseLength: " + StringUtils.length(res) + ", errorType: " + + e.getClass().getSimpleName()); } - return result; } private String doGet(String url, Map params) { diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java index 7c22c878b754..d7a23922a6b4 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java @@ -18,12 +18,14 @@ package org.apache.dolphinscheduler.plugin.task.dinky; import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Arrays; import java.util.HashMap; @@ -98,14 +100,18 @@ void testSanitizeMessageMasksSensitiveValuesInJsonNodeMessage() { } @Test - void testParseMalformedResponseReturnsNullWithoutExposingRawResponse() throws Exception { + void testParseMalformedResponseFailsExplicitlyWithoutExposingRawResponse() throws Exception { DinkyTask task = new DinkyTask(new TaskExecutionContext()); Method parseMethod = DinkyTask.class.getDeclaredMethod("parse", String.class); parseMethod.setAccessible(true); + String response = "malformed response accessKeyId=AKIA_TEST accessKeySecret=SECRET_TEST"; - Object result = parseMethod.invoke(task, - "malformed response accessKeyId=AKIA_TEST accessKeySecret=SECRET_TEST"); + InvocationTargetException exception = + assertThrows(InvocationTargetException.class, () -> parseMethod.invoke(task, response)); - assertNull(result); + DinkyTaskException cause = assertInstanceOf(DinkyTaskException.class, exception.getCause()); + assertTrue(cause.getMessage().contains("dinky task response parse failed")); + assertFalse(cause.getMessage().contains("AKIA_TEST")); + assertFalse(cause.getMessage().contains("SECRET_TEST")); } } From 1e61609d4f2bcddc1539def5e963c1c1e8b3be05 Mon Sep 17 00:00:00 2001 From: Yanjun Qiu <153984347+qiuyanjun888@users.noreply.github.com> Date: Tue, 14 Jul 2026 15:45:18 +0800 Subject: [PATCH 3/4] [Feature-17398][Task] Mask sensitive values containing escaped quotes --- .../task/api/log/SensitiveDataConverter.java | 54 +++++++++++++++++-- .../api/log/SensitiveDataConverterTest.java | 8 +++ 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java index 8f5314a1d30b..29e3c88a9b6d 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverter.java @@ -40,9 +40,9 @@ public class SensitiveDataConverter extends MessageConverter { 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_PATTERN = Pattern.compile( + private static final Pattern QUOTED_SENSITIVE_CONFIGURATION_PREFIX_PATTERN = Pattern.compile( "((?:\\\\?\\\"|')?" + KNOWN_SENSITIVE_CONFIGURATION_KEY_REGEX - + "(?:\\\\?\\\"|')?\\s*(?::|=)\\s*(?:\\\\?\\\"|'))(.*?)(\\\\?\\\"|')", + + "(?:\\\\?\\\"|')?\\s*(?::|=)\\s*(\\\\?\\\"|'))", Pattern.CASE_INSENSITIVE | Pattern.MULTILINE); private static final Pattern UNQUOTED_SENSITIVE_CONFIGURATION_PATTERN = Pattern.compile( @@ -76,8 +76,8 @@ public static String maskSensitiveData(final String logMsg) { return logMsg; } - String maskedLogMsg = maskByConfiguredPatterns(logMsg); - return maskKnownSensitiveConfiguration(maskedLogMsg); + String maskedLogMsg = maskKnownSensitiveConfiguration(logMsg); + return maskByConfiguredPatterns(maskedLogMsg); } private static String maskByConfiguredPatterns(final String logMsg) { @@ -93,10 +93,54 @@ private static String maskByConfiguredPatterns(final String logMsg) { } private static String maskKnownSensitiveConfiguration(final String logMsg) { - String maskedLogMsg = maskKnownSensitiveConfiguration(logMsg, QUOTED_SENSITIVE_CONFIGURATION_PATTERN); + 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); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java index 25af2a2989f0..6976294c72b4 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/log/SensitiveDataConverterTest.java @@ -145,6 +145,14 @@ void testMaskObjectStorageCredentialsInCommonFormats() { } } + @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"; From d001ec79ccf65d0791490c0236d3e11c1a2b389c Mon Sep 17 00:00:00 2001 From: Yanjun Qiu <153984347+qiuyanjun888@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:50:59 +0800 Subject: [PATCH 4/4] [Fix-17398][Task] Sanitize credentials in Dinky addresses --- .../plugin/task/dinky/DinkyLogSanitizer.java | 36 +++++- .../plugin/task/dinky/DinkyTask.java | 19 +-- .../task/dinky/DinkyLogSanitizerTest.java | 117 ++++++++++++++++++ 3 files changed, 164 insertions(+), 8 deletions(-) diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java index 80367efd9e5c..b5a0def514ea 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizer.java @@ -17,9 +17,16 @@ 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; @@ -35,12 +42,33 @@ static String summarizeParameters(final DinkyParameters parameters) { if (parameters == null) { return "null"; } - return "address=" + sanitizeMessage(parameters.getAddress()) + 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 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 variables) { if (variables == null || variables.isEmpty()) { return "size=0, keys=[]"; @@ -56,6 +84,12 @@ static String sanitizeMessage(final Object message) { 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 summarizeLocalParamKeys(final List localParams) { Set keys = new TreeSet<>(); if (localParams == null) { diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java index d1cf05e8891b..8e857f935678 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/main/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyTask.java @@ -179,7 +179,8 @@ public void trackApplicationStatusV0() throws TaskException { String taskId = this.dinkyParameters.getTaskId(); if (status && jobInstanceId == null) { // Use address-taskId as app id - setAppIds(String.format(DinkyTaskConstants.APPIDS_FORMAT, address, taskId)); + setAppIds(String.format( + DinkyTaskConstants.APPIDS_FORMAT, DinkyLogSanitizer.sanitizeAddress(address), taskId)); setExitStatusCode(mapStatusToExitCode(true)); log.info("Dinky common sql task finished."); return; @@ -197,7 +198,8 @@ public void trackApplicationStatusV0() throws TaskException { case DinkyTaskConstants.STATUS_FINISHED: final int exitStatusCode = mapStatusToExitCode(status); // Use address-taskId as app id - setAppIds(String.format(DinkyTaskConstants.APPIDS_FORMAT, address, taskId)); + setAppIds(String.format( + DinkyTaskConstants.APPIDS_FORMAT, DinkyLogSanitizer.sanitizeAddress(address), taskId)); setExitStatusCode(exitStatusCode); log.info("dinky task finished, status: {}", jobInstanceStatus); finishFlag = true; @@ -229,7 +231,8 @@ public void trackApplicationStatusV1() throws TaskException { String taskId = this.dinkyParameters.getTaskId(); if (status && jobInstanceId == null) { // Use address-taskId as app id - setAppIds(String.format(DinkyTaskConstants.APPIDS_FORMAT, address, taskId)); + setAppIds(String.format( + DinkyTaskConstants.APPIDS_FORMAT, DinkyLogSanitizer.sanitizeAddress(address), taskId)); setExitStatusCode(mapStatusToExitCode(true)); log.info("Dinky common sql task finished."); return; @@ -247,7 +250,8 @@ public void trackApplicationStatusV1() throws TaskException { case DinkyTaskConstants.STATUS_FINISHED: final int exitStatusCode = mapStatusToExitCode(status); // Use address-taskId as app id - setAppIds(String.format(DinkyTaskConstants.APPIDS_FORMAT, address, taskId)); + setAppIds(String.format( + DinkyTaskConstants.APPIDS_FORMAT, DinkyLogSanitizer.sanitizeAddress(address), taskId)); setExitStatusCode(exitStatusCode); log.info("dinky task finished, status: {}", jobInstanceStatus); finishFlag = true; @@ -321,15 +325,16 @@ public AbstractParameters getParameters() { @Override public void cancelApplication() throws TaskException { String address = this.dinkyParameters.getAddress(); + String sanitizedAddress = DinkyLogSanitizer.sanitizeAddress(address); String taskId = this.dinkyParameters.getTaskId(); log.info("trying terminate dinky task, taskId: {}, address: {}, taskId: {}", this.taskExecutionContext.getTaskInstanceId(), - address, + sanitizedAddress, taskId); cancelTask(address, taskId); log.warn("dinky task terminated, taskId: {}, address: {}, taskId: {}", this.taskExecutionContext.getTaskInstanceId(), - address, + sanitizedAddress, taskId); } @@ -420,7 +425,7 @@ private String doGet(String url, Map params) { } URI uri = uriBuilder.build(); httpGet = new HttpGet(uri); - log.info("access url: {}", uri); + log.info("access url: {}", DinkyLogSanitizer.sanitizeAddress(uri)); HttpResponse response = httpClient.execute(httpGet); if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { result = EntityUtils.toString(response.getEntity()); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java index d7a23922a6b4..a3e89975e531 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java @@ -25,18 +25,38 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import com.fasterxml.jackson.databind.node.JsonNodeFactory; class DinkyLogSanitizerTest { + private static final String URI_USERNAME = "alice"; + private static final String URI_PASSWORD = "s3cr3t"; + private static final String QUERY_TOKEN = "TOKEN"; + private static final String QUERY_USERNAME = "QUERY_USER"; + private static final String QUERY_PASSWORD = "QUERY_PASSWORD"; + private static final String CREDENTIAL_ADDRESS = + "unknown://" + URI_USERNAME + ":" + URI_PASSWORD + + "@dinky:8888?token=" + QUERY_TOKEN + + "&username=" + QUERY_USERNAME + + "&password=" + QUERY_PASSWORD + + "&mode=test"; + @Test void testSummarizeVariablesDoesNotExposeValues() { Map variables = new HashMap<>(); @@ -75,6 +95,71 @@ void testSummarizeParametersDoesNotExposeLocalParamValues() { assertFalse(summary.contains("2026-07-06")); } + @Test + void testSummarizeParametersDoesNotExposeAddressCredentials() { + DinkyParameters parameters = credentialParameters(); + + String summary = DinkyLogSanitizer.summarizeParameters(parameters); + + assertAddressCredentialsMasked(summary); + assertTrue(summary.contains("dinky:8888")); + assertTrue(summary.contains("mode=test")); + } + + @Test + void testApplicationIdDoesNotExposeAddressCredentials() throws Exception { + DinkyTask task = new DinkyTask(new TaskExecutionContext()); + setField(task, "dinkyParameters", credentialParameters()); + setField(task, "status", true); + + task.trackApplicationStatusV0(); + + String applicationId = task.getAppIds(); + assertAddressCredentialsMasked(applicationId); + assertTrue(applicationId.contains("dinky:8888")); + assertTrue(applicationId.endsWith("-1001")); + } + + @Test + void testGetRequestUrlLogDoesNotExposeAddressCredentials() throws Exception { + DinkyTask task = new DinkyTask(new TaskExecutionContext()); + Logger logger = (Logger) LoggerFactory.getLogger(DinkyTask.class); + ListAppender appender = attachListAppender(logger); + try { + Method doGet = DinkyTask.class.getDeclaredMethod("doGet", String.class, Map.class); + doGet.setAccessible(true); + + doGet.invoke(task, CREDENTIAL_ADDRESS, Collections.emptyMap()); + + String messages = formattedMessages(appender); + assertAddressCredentialsMasked(messages); + assertTrue(messages.contains("access url:")); + assertTrue(messages.contains("mode=test")); + } finally { + logger.detachAppender(appender); + appender.stop(); + } + } + + @Test + void testCancelLogsDoNotExposeAddressCredentials() throws Exception { + DinkyTask task = new DinkyTask(new TaskExecutionContext()); + setField(task, "dinkyParameters", credentialParameters()); + Logger logger = (Logger) LoggerFactory.getLogger(DinkyTask.class); + ListAppender appender = attachListAppender(logger); + try { + task.cancelApplication(); + + String messages = formattedMessages(appender); + assertAddressCredentialsMasked(messages); + assertTrue(messages.contains("trying terminate dinky task")); + assertTrue(messages.contains("dinky task terminated")); + } finally { + logger.detachAppender(appender); + appender.stop(); + } + } + @Test void testSanitizeMessageMasksSensitiveValues() { String message = "{\"accessKeyId\":\"AKIA_TEST\",\"accessKeySecret\":\"SECRET_TEST\"}"; @@ -114,4 +199,36 @@ void testParseMalformedResponseFailsExplicitlyWithoutExposingRawResponse() throw assertFalse(cause.getMessage().contains("AKIA_TEST")); assertFalse(cause.getMessage().contains("SECRET_TEST")); } + + private static DinkyParameters credentialParameters() { + DinkyParameters parameters = new DinkyParameters(); + parameters.setAddress(CREDENTIAL_ADDRESS); + parameters.setTaskId("1001"); + return parameters; + } + + private static void setField(DinkyTask task, String fieldName, Object value) throws Exception { + Field field = DinkyTask.class.getDeclaredField(fieldName); + field.setAccessible(true); + field.set(task, value); + } + + private static ListAppender attachListAppender(Logger logger) { + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + return appender; + } + + private static String formattedMessages(ListAppender appender) { + return appender.list.stream().map(ILoggingEvent::getFormattedMessage).collect(Collectors.joining("\n")); + } + + private static void assertAddressCredentialsMasked(String value) { + assertFalse(value.contains(URI_USERNAME)); + assertFalse(value.contains(URI_PASSWORD)); + assertFalse(value.contains(QUERY_TOKEN)); + assertFalse(value.contains(QUERY_USERNAME)); + assertFalse(value.contains(QUERY_PASSWORD)); + } }