This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c4ddfbaa66a [fix](fe) Mask Kafka routine load sensitive properties
(#64786)
c4ddfbaa66a is described below
commit c4ddfbaa66a6039a0383fd3fa1fb8c549bfa4381
Author: Refrain <[email protected]>
AuthorDate: Thu Jul 2 20:48:11 2026 +0800
[fix](fe) Mask Kafka routine load sensitive properties (#64786)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Kafka routine load custom properties were rendered
directly in SHOW ROUTINE LOAD, SHOW CREATE ROUTINE LOAD, and
information_schema.routine_load_jobs, so SASL, SSL, and AWS credential
values could be exposed through job metadata. This change masks
sensitive Kafka custom property values when formatting external display
output while preserving the internal properties used by Kafka clients.
Added unit and regression coverage for masked display output and
unchanged internal values.
---
.../routineload/kafka/KafkaRoutineLoadJob.java | 33 +++++++--
.../routineload/kinesis/KinesisRoutineLoadJob.java | 30 ++++----
.../load/routineload/KafkaRoutineLoadJobTest.java | 81 ++++++++++++++++++++++
.../routineload/KinesisRoutineLoadJobTest.java | 28 ++++++++
.../test_kinesis_show_routine_load.groovy | 2 +-
.../routine_load/test_show_routine_load.groovy | 41 +++++++++++
6 files changed, 194 insertions(+), 21 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
index dcb683e9e22..8b824747979 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
@@ -33,6 +33,7 @@ import org.apache.doris.common.InternalErrorCode;
import org.apache.doris.common.LoadException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DatasourcePrintableMap;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.common.util.LogBuilder;
@@ -82,6 +83,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.TimeZone;
import java.util.UUID;
@@ -102,6 +104,7 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
private static final String READ_COMMITTED_ZERO_ROWS_WITH_LAG_MESSAGE =
"Kafka routine load consumed 0 rows "
+ "while lag is still positive under
isolation.level=read_committed. If the upstream producer uses "
+ "Kafka transactions, some records may be in uncommitted
transactions and are not visible yet.";
+ private static final String SENSITIVE_PROPERTY_MASK = "******";
@SerializedName("bl")
private String brokerList;
@@ -723,7 +726,31 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
@Override
public String customPropertiesJsonToString() {
Gson gson = new GsonBuilder().disableHtmlEscaping().create();
- return gson.toJson(customProperties);
+ return gson.toJson(getMaskedCustomProperties(""));
+ }
+
+ private Map<String, String> getMaskedCustomProperties(String keyPrefix) {
+ Map<String, String> maskedProperties = new HashMap<>();
+ customProperties.forEach((key, value) -> {
+ String lowerKey = key.toLowerCase(Locale.ROOT);
+ boolean sensitive =
KafkaConfiguration.SASL_JAAS_CONFIG.equalsIgnoreCase(key)
+ || KafkaConfiguration.AWS_ACCESS_KEY.equalsIgnoreCase(key)
+ || DatasourcePrintableMap.SENSITIVE_KEY.contains(key)
+ || lowerKey.endsWith(".password")
+ || lowerKey.endsWith(".secret")
+ || lowerKey.endsWith(".secret_key")
+ || lowerKey.endsWith(".secret.key")
+ || lowerKey.endsWith(".session_key")
+ || lowerKey.endsWith(".session.token")
+ || lowerKey.contains(".private.key.")
+ || lowerKey.endsWith(".private.key")
+ || lowerKey.endsWith(".private_key")
+ || lowerKey.endsWith(".passphrase")
+ || "ssl.keystore.key".equals(lowerKey)
+ || "ssl.key.pem".equals(lowerKey);
+ maskedProperties.put(keyPrefix + key, sensitive ?
SENSITIVE_PROPERTY_MASK : value);
+ });
+ return maskedProperties;
}
@Override
@@ -736,9 +763,7 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
@Override
public Map<String, String> getCustomProperties() {
- Map<String, String> ret = new HashMap<>();
- customProperties.forEach((k, v) -> ret.put("property." + k, v));
- return ret;
+ return getMaskedCustomProperties("property.");
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
index 0c058899299..7cebc3f5165 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
@@ -92,6 +92,7 @@ import java.util.UUID;
*/
public class KinesisRoutineLoadJob extends RoutineLoadJob {
private static final Logger LOG =
LogManager.getLogger(KinesisRoutineLoadJob.class);
+ private static final String SENSITIVE_PROPERTY_MASK = "******";
public static final String KINESIS_FILE_CATALOG = "kinesis";
@@ -640,15 +641,7 @@ public class KinesisRoutineLoadJob extends RoutineLoadJob {
@Override
public String customPropertiesJsonToString() {
Gson gson = new GsonBuilder().disableHtmlEscaping().create();
- // Mask sensitive information
- Map<String, String> maskedProperties = new HashMap<>(customProperties);
- if (maskedProperties.containsKey("aws.secret_key")) {
- maskedProperties.put("aws.secret_key", "******");
- }
- if (maskedProperties.containsKey("aws.session_key")) {
- maskedProperties.put("aws.session_key", "******");
- }
- return gson.toJson(maskedProperties);
+ return gson.toJson(getMaskedCustomProperties(""));
}
@Override
@@ -664,16 +657,21 @@ public class KinesisRoutineLoadJob extends RoutineLoadJob
{
@Override
public Map<String, String> getCustomProperties() {
- Map<String, String> ret = new HashMap<>();
- customProperties.forEach((k, v) -> {
- // Mask sensitive values
- if (k.equals("aws.secret_key") || k.equals("aws.session_key")) {
- ret.put("property." + k, "******");
+ return getMaskedCustomProperties("property.");
+ }
+
+ private Map<String, String> getMaskedCustomProperties(String keyPrefix) {
+ Map<String, String> maskedProperties = new HashMap<>();
+ customProperties.forEach((key, value) -> {
+ if
(KinesisConfiguration.KINESIS_ACCESS_KEY.getName().equalsIgnoreCase(key)
+ ||
KinesisConfiguration.KINESIS_SECRET_KEY.getName().equalsIgnoreCase(key)
+ ||
KinesisConfiguration.KINESIS_SESSION_TOKEN.getName().equalsIgnoreCase(key)) {
+ maskedProperties.put(keyPrefix + key, SENSITIVE_PROPERTY_MASK);
} else {
- ret.put("property." + k, v);
+ maskedProperties.put(keyPrefix + key, value);
}
});
- return ret;
+ return maskedProperties;
}
@Override
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
index 8b442012ff2..ebdad39a274 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
@@ -261,6 +261,87 @@ public class KafkaRoutineLoadJobTest {
Assert.assertTrue(otherMsg.contains("some records may be in
uncommitted transactions"));
}
+ @Test
+ public void testDisplayCustomPropertiesMasksKafkaSecrets() {
+ KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
+ 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+ Map<String, String> customProperties = Maps.newHashMap();
+ customProperties.put("security.protocol", "SASL_PLAINTEXT");
+ customProperties.put("sasl.username", "doris");
+ customProperties.put("sasl.password", "plain_secret");
+ customProperties.put("sasl.jaas.config", "username=\"doris\"
password=\"jaas_secret\"");
+ customProperties.put("sasl.oauthbearer.client.secret",
"oauth_client_secret");
+
customProperties.put("sasl.oauthbearer.client.credentials.client.secret",
"oauth_alias_secret");
+ customProperties.put("sasl.oauthbearer.assertion.private.key.pem",
"oauth_private_key_pem");
+
customProperties.put("sasl.oauthbearer.assertion.private.key.passphrase",
"oauth_private_key_passphrase");
+ customProperties.put("ssl.keystore.password", "keystore_secret");
+ customProperties.put("ssl.keystore.key", "keystore_key_secret");
+ customProperties.put("ssl.key.pem", "key_pem_secret");
+ customProperties.put(KafkaConfiguration.AWS_ACCESS_KEY,
"aws_access_key");
+ customProperties.put("aws.secret_key", "aws_secret");
+ customProperties.put("aws.session_key", "aws_session_secret");
+ customProperties.put("password", "bare_password_secret");
+ customProperties.put("secret_key", "bare_secret_key");
+ customProperties.put("session_token", "bare_session_token");
+ Deencapsulation.setField(routineLoadJob, "customProperties",
customProperties);
+
+ String customPropertiesJson =
routineLoadJob.customPropertiesJsonToString();
+ Map<String, String> showCreateCustomProperties =
routineLoadJob.getCustomProperties();
+
+ Assert.assertFalse(customPropertiesJson.contains("plain_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("jaas_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_client_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_alias_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_private_key_pem"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_private_key_passphrase"));
+ Assert.assertFalse(customPropertiesJson.contains("keystore_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("keystore_key_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("key_pem_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("aws_access_key"));
+ Assert.assertFalse(customPropertiesJson.contains("aws_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("aws_session_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("bare_password_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("bare_secret_key"));
+
Assert.assertFalse(customPropertiesJson.contains("bare_session_token"));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.password\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.jaas.config\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.oauthbearer.client.secret\":\"******\""));
+ Assert.assertTrue(customPropertiesJson.contains(
+
"\"sasl.oauthbearer.client.credentials.client.secret\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.oauthbearer.assertion.private.key.pem\":\"******\""));
+ Assert.assertTrue(customPropertiesJson.contains(
+
"\"sasl.oauthbearer.assertion.private.key.passphrase\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"ssl.keystore.password\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"ssl.keystore.key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"ssl.key.pem\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.access_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.secret_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.session_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"password\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"secret_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"session_token\":\"******\""));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.sasl.password"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.sasl.jaas.config"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.sasl.oauthbearer.client.secret"));
+ Assert.assertEquals("******",
+
showCreateCustomProperties.get("property.sasl.oauthbearer.client.credentials.client.secret"));
+ Assert.assertEquals("******",
+
showCreateCustomProperties.get("property.sasl.oauthbearer.assertion.private.key.pem"));
+ Assert.assertEquals("******",
+
showCreateCustomProperties.get("property.sasl.oauthbearer.assertion.private.key.passphrase"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.ssl.keystore.password"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.ssl.keystore.key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.ssl.key.pem"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.access_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.secret_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.session_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.password"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.secret_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.session_token"));
+ Assert.assertEquals("doris",
showCreateCustomProperties.get("property.sasl.username"));
+ Assert.assertEquals("plain_secret",
customProperties.get("sasl.password"));
+ }
+
@Test
public void testReadCommittedZeroRowsWithLagDelaysNextTask() throws
UserException {
RoutineLoadManager routineLoadManager =
Mockito.mock(RoutineLoadManager.class);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KinesisRoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KinesisRoutineLoadJobTest.java
index aa1dc052605..65aebd0084e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KinesisRoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KinesisRoutineLoadJobTest.java
@@ -311,6 +311,34 @@ public class KinesisRoutineLoadJobTest {
Assert.assertFalse((Boolean) Deencapsulation.invoke(routineLoadJob,
"isKinesisShardsChanged"));
}
+ @Test
+ public void testDisplayCustomPropertiesMasksKinesisSecrets() {
+ KinesisRoutineLoadJob routineLoadJob =
+ new KinesisRoutineLoadJob(1L, "kinesis_routine_load_job", 1L,
+ 1L, "ap-southeast-1", "stream-1", UserIdentity.ADMIN);
+ Map<String, String> customProperties = Maps.newHashMap();
+
customProperties.put(KinesisConfiguration.KINESIS_ACCESS_KEY.getName(),
"aws_access_key");
+
customProperties.put(KinesisConfiguration.KINESIS_SECRET_KEY.getName(),
"aws_secret");
+
customProperties.put(KinesisConfiguration.KINESIS_SESSION_TOKEN.getName(),
"aws_session_secret");
+ customProperties.put("aws.role_arn", "role_arn_value");
+ Deencapsulation.setField(routineLoadJob, "customProperties",
customProperties);
+
+ String customPropertiesJson =
routineLoadJob.customPropertiesJsonToString();
+ Map<String, String> showCreateCustomProperties =
routineLoadJob.getCustomProperties();
+
+ Assert.assertFalse(customPropertiesJson.contains("aws_access_key"));
+ Assert.assertFalse(customPropertiesJson.contains("aws_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("aws_session_secret"));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.access_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.secret_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.session_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.role_arn\":\"role_arn_value\""));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.access_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.secret_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.session_key"));
+ Assert.assertEquals("role_arn_value",
showCreateCustomProperties.get("property.aws.role_arn"));
+ }
+
private Set<String> collectAssignedShards(KinesisRoutineLoadJob
routineLoadJob) {
List<RoutineLoadTaskInfo> routineLoadTaskInfoList =
Deencapsulation.getField(routineLoadJob,
"routineLoadTaskInfoList");
diff --git
a/regression-test/suites/load_p0/kinesis_routine_load/test_kinesis_show_routine_load.groovy
b/regression-test/suites/load_p0/kinesis_routine_load/test_kinesis_show_routine_load.groovy
index 3c6cd17da36..929e9dbf50b 100644
---
a/regression-test/suites/load_p0/kinesis_routine_load/test_kinesis_show_routine_load.groovy
+++
b/regression-test/suites/load_p0/kinesis_routine_load/test_kinesis_show_routine_load.groovy
@@ -224,7 +224,7 @@ suite("test_kinesis_show_routine_load") {
def customProperties = parseJson(showRow[13].toString())
assertEquals("******",
customProperties["aws.secret_key"].toString())
- assertTrue(customProperties["aws.access_key"].toString().length()
> 0)
+ assertEquals("******",
customProperties["aws.access_key"].toString())
// Verify additional SHOW ROUTINE LOAD paths.
String dbName = context.config.getDbNameByFile(context.file)
diff --git
a/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
b/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
index f89e89a9dda..fc330bb7b1d 100644
--- a/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
+++ b/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
@@ -204,6 +204,47 @@ suite("test_show_routine_load","p0") {
sql "stop routine load for testShow"
}
+ // test show routine load and information_schema mask sensitive custom
properties
+ try {
+ def kafkaPassword = "doris_routine_load_password_should_be_hidden"
+ def kafkaJaasSecret =
"doris_routine_load_jaas_secret_should_be_hidden"
+ sql """
+ CREATE ROUTINE LOAD testShowSensitiveProperties ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${externalEnvIp}:${kafka_port}",
+ "kafka_topic" = "${kafkaCsvTpoics[0]}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING",
+ "property.security.protocol" = "SASL_PLAINTEXT",
+ "property.sasl.mechanism" = "PLAIN",
+ "property.sasl.username" = "doris",
+ "property.sasl.password" = "${kafkaPassword}",
+ "property.sasl.jaas.config" = "password=${kafkaJaasSecret}"
+ );
+ """
+
+ def showResult = sql "show routine load for
testShowSensitiveProperties"
+ def showCustomProperties = parseJson(showResult[0][13])
+ assertEquals("******",
showCustomProperties["sasl.password"].toString())
+ assertEquals("******",
showCustomProperties["sasl.jaas.config"].toString())
+ assertFalse(showResult[0][13].toString().contains(kafkaPassword))
+ assertFalse(showResult[0][13].toString().contains(kafkaJaasSecret))
+
+ def systemTableResult = sql """
+ SELECT CUSTOM_PROPERTIES
+ FROM information_schema.routine_load_jobs
+ WHERE JOB_NAME = 'testShowSensitiveProperties'
+ """
+ def systemTableCustomProperties =
parseJson(systemTableResult[0][0])
+ assertEquals("******",
systemTableCustomProperties["sasl.password"].toString())
+ assertEquals("******",
systemTableCustomProperties["sasl.jaas.config"].toString())
+
assertFalse(systemTableResult[0][0].toString().contains(kafkaPassword))
+
assertFalse(systemTableResult[0][0].toString().contains(kafkaJaasSecret))
+ } finally {
+ sql "stop routine load for testShowSensitiveProperties"
+ }
+
// test show routine load computegroup
try {
sql """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
