Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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_PATTERN = Pattern.compile(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

.*? does not distinguish an escaped quote from the actual string terminator.

eg, given valid JSON:

{"password":"abc\"VISIBLE_SUFFIX","bucket":"ds"}

result output:

{"password":"******\"VISIBLE_SUFFIX","bucket":"ds"}

@qiuyanjun888 qiuyanjun888 Jul 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 1e61609d. I replaced the lazy quoted-value match with quote-aware delimiter scanning, so escaped quotes are not treated as the end of the value. Known-sensitive masking now also runs before the legacy configured patterns, preventing the old password regex from truncating the value first.

The exact JSON example now becomes {"password":"******","bucket":"ds"}.

Validation: SensitiveDataConverterTest passed 7/7, DinkyLogSanitizerTest passed 5/5, and Spotless passed for both task-api and task-dinky.

"((?:\\\\?\\\"|')?" + 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);
}
Expand Down Expand Up @@ -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);

Expand All @@ -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();
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -124,4 +125,33 @@ 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 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("******"));
}
}
Original file line number Diff line number Diff line change
@@ -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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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;
}

Expand Down Expand Up @@ -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: {}",

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The parse error should fail explicitly.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed by throwing DinkyTaskException when the Dinky response cannot be parsed. The exception only includes the response length and parser exception type, and the regression test verifies explicit failure without exposing sensitive response values.

Verification: ./mvnw -pl dolphinscheduler-task-plugin/dolphinscheduler-task-dinky -Dtest='DinkyLogSanitizerTest#testParseMalformedResponseFailsExplicitlyWithoutExposingRawResponse' clean test passed.

StringUtils.length(res), e.getClass().getSimpleName());
}
return result;
}
Expand All @@ -426,7 +425,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);
}
Expand Down Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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<String, String> 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 {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This only asserts that the return value is null; it does not capture or inspect any log event. The test would also pass with the old implementation even if the raw response were written through the exception log.

@qiuyanjun888 qiuyanjun888 Jul 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in ffe340bc and revalidated at 1e61609d. The old null-return test was replaced with an explicit-failure regression, so it would fail against the old implementation.

parse no longer emits a log event with the Jackson throwable or raw response. It throws a cause-free DinkyTaskException containing only the response length and parser exception type. I also traced the production paths: submit-side and worker terminal loggers can only receive that safe exception graph, not the raw response. Therefore an appender assertion at parse would no longer exercise an actual log event.

Validation: DinkyLogSanitizerTest passed 5/5 in the final targeted test run.

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);
}
}
Loading