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..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 @@ -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_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(); + } + } 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..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 @@ -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 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("******")); + } } 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..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 @@ -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; } @@ -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 params) { @@ -426,7 +424,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 +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); } 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..d7a23922a6b4 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-dinky/src/test/java/org/apache/dolphinscheduler/plugin/task/dinky/DinkyLogSanitizerTest.java @@ -0,0 +1,117 @@ +/* + * 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.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; +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 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"; + + InvocationTargetException exception = + assertThrows(InvocationTargetException.class, () -> parseMethod.invoke(task, response)); + + 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")); + } +}