diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java index ad800aa8f6..4b4ae09fc5 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java @@ -17,6 +17,7 @@ package org.apache.fluss.config; +import org.apache.fluss.utils.CollectionUtils; import org.apache.fluss.utils.TimeUtils; import javax.annotation.Nonnull; @@ -38,6 +39,37 @@ /** Utility class for {@link Configuration} related helper functions. */ public class ConfigurationUtils { + + private static final String[] SENSITIVE_KEY_PARTS = { + "password", + "secret", + "fs.azure.account.key", + "apikey", + "api-key", + "auth-params", + "service-key", + "token", + "basic-auth", + "jaas.config", + "http-headers", + "private.key", + "private-key", + "access.key", + "access-key", + "access_key", + "accesskey" + }; + + // Exact allowlist for non-secret options that match broad sensitive key parts. + private static final Set NON_SENSITIVE_KEYS = + Arrays.stream( + new String[] { + ConfigOptions.FILESYSTEM_SECURITY_TOKEN_RENEWAL_RETRY_BACKOFF.key(), + ConfigOptions.FILESYSTEM_SECURITY_TOKEN_RENEWAL_TIME_RATIO.key() + }) + .map(key -> key.toLowerCase(Locale.ROOT)) + .collect(Collectors.toSet()); + // -------------------------------------------------------------------------------------------- // Type conversion // -------------------------------------------------------------------------------------------- @@ -81,6 +113,47 @@ public static T convertValue(Object rawValue, Class clazz) { throw new IllegalArgumentException("Unsupported type: " + clazz); } + /** + * Returns a log-safe representation for a configuration value. + * + * @param key configuration key + * @param value configuration value + * @return {@link Password#HIDDEN_CONTENT} for sensitive values, otherwise the original value + */ + public static Object hideSensitiveValue(String key, Object value) { + if (value == null) { + return null; + } + return value instanceof Password || isSensitiveKey(key) ? Password.HIDDEN_CONTENT : value; + } + + /** + * Returns a copy of the given configuration map with sensitive values hidden. + * + * @param values configuration key/value pairs + * @return a new map containing log-safe values + */ + public static Map hideSensitiveValues(Map values) { + Map hiddenValues = + CollectionUtils.newHashMapWithExpectedSize(values.size()); + values.forEach( + (key, value) -> hiddenValues.put(key, (String) hideSensitiveValue(key, value))); + return hiddenValues; + } + + static boolean isSensitiveKey(String key) { + String lowerKey = key.toLowerCase(Locale.ROOT); + if (NON_SENSITIVE_KEYS.contains(lowerKey)) { + return false; + } + for (String sensitiveKeyPart : SENSITIVE_KEY_PARTS) { + if (lowerKey.contains(sensitiveKeyPart)) { + return true; + } + } + return false; + } + @SuppressWarnings("unchecked") static T convertToList(Object rawValue, Class atomicClass) { if (rawValue instanceof List) { diff --git a/fluss-common/src/main/java/org/apache/fluss/config/GlobalConfiguration.java b/fluss-common/src/main/java/org/apache/fluss/config/GlobalConfiguration.java index fa7b6da916..b3d0412bfe 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/GlobalConfiguration.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/GlobalConfiguration.java @@ -126,7 +126,7 @@ private static void logConfiguration(String prefix, Configuration config) { "{} configuration property: {}={}", prefix, key, - value instanceof Password ? Password.HIDDEN_CONTENT : value)); + ConfigurationUtils.hideSensitiveValue(key, value))); } /** diff --git a/fluss-common/src/test/java/org/apache/fluss/config/ConfigurationTest.java b/fluss-common/src/test/java/org/apache/fluss/config/ConfigurationTest.java index 77dfec1680..d080999485 100644 --- a/fluss-common/src/test/java/org/apache/fluss/config/ConfigurationTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/config/ConfigurationTest.java @@ -23,6 +23,7 @@ import java.util.Arrays; import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Properties; import java.util.stream.Collectors; @@ -58,6 +59,11 @@ public class ConfigurationTest { private static final ConfigOption SECRET_OPTION = ConfigBuilder.key("secret").passwordType().noDefaultValue(); + private static final String GS_SERVICE_ACCOUNT_PRIVATE_KEY = + "fs.gs.auth.service.account.private.key"; + private static final String GS_SERVICE_ACCOUNT_PRIVATE_KEY_HYPHEN = + "fs.gs.auth.service.account.private-key"; + private static final Map PROPERTIES_MAP = new HashMap<>(); static { @@ -456,6 +462,77 @@ void testPasswordType() { assertThat(cfg.get(SECRET_OPTION).value()).isEqualTo(secretValue); } + @Test + void testSensitiveKeyParts() { + List sensitiveKeys = + Arrays.asList( + "client.security.sasl.password", + "fs.s3a.secret.key", + "fs.azure.account.key.account.blob.core.windows.net", + "connector.apikey", + "connector.api-key", + "connector.auth-params", + "connector.service-key", + "session.token", + "security.basic-auth", + "security.sasl.plain.jaas.config", + "client.http-headers", + GS_SERVICE_ACCOUNT_PRIVATE_KEY, + GS_SERVICE_ACCOUNT_PRIVATE_KEY_HYPHEN, + "fs.s3a.access.key", + "s3.access-key", + "fs.oss.accessKeyId", + "datalake.lance.access_key_id"); + + for (String key : sensitiveKeys) { + assertThat(ConfigurationUtils.isSensitiveKey(key)).as(key).isTrue(); + assertThat(ConfigurationUtils.isSensitiveKey(key.toUpperCase(Locale.ROOT))) + .as(key) + .isTrue(); + } + + assertThat(ConfigurationUtils.isSensitiveKey("client.request.timeout")).isFalse(); + assertThat(ConfigurationUtils.isSensitiveKey("remote.data.dir")).isFalse(); + + List nonSensitiveKeys = + Arrays.asList( + ConfigOptions.FILESYSTEM_SECURITY_TOKEN_RENEWAL_RETRY_BACKOFF.key(), + ConfigOptions.FILESYSTEM_SECURITY_TOKEN_RENEWAL_TIME_RATIO.key()); + for (String key : nonSensitiveKeys) { + assertThat(ConfigurationUtils.isSensitiveKey(key)).as(key).isFalse(); + assertThat(ConfigurationUtils.isSensitiveKey(key.toUpperCase(Locale.ROOT))) + .as(key) + .isFalse(); + } + } + + @Test + void testHideSensitiveValue() { + assertThat(ConfigurationUtils.hideSensitiveValue("fs.s3a.access.key", "ak")) + .isEqualTo(Password.HIDDEN_CONTENT); + assertThat(ConfigurationUtils.hideSensitiveValue("client.security.sasl.password", "pwd")) + .isEqualTo(Password.HIDDEN_CONTENT); + assertThat(ConfigurationUtils.hideSensitiveValue("plain.key", new Password("pwd"))) + .isEqualTo(Password.HIDDEN_CONTENT); + assertThat(ConfigurationUtils.hideSensitiveValue("plain.key", "value")).isEqualTo("value"); + } + + @Test + void testHideSensitiveValues() { + Map values = new HashMap<>(); + values.put("fs.s3a.access.key", "access-key"); + values.put(GS_SERVICE_ACCOUNT_PRIVATE_KEY, "private-key"); + values.put(ConfigOptions.FILESYSTEM_SECURITY_TOKEN_RENEWAL_RETRY_BACKOFF.key(), "10 s"); + values.put("plain.key", "value"); + + assertThat(ConfigurationUtils.hideSensitiveValues(values)) + .containsEntry("fs.s3a.access.key", Password.HIDDEN_CONTENT) + .containsEntry(GS_SERVICE_ACCOUNT_PRIVATE_KEY, Password.HIDDEN_CONTENT) + .containsEntry( + ConfigOptions.FILESYSTEM_SECURITY_TOKEN_RENEWAL_RETRY_BACKOFF.key(), "10 s") + .containsEntry("plain.key", "value"); + } + @Test void testPasswordParserErrorDoesNotLeakSensitiveData() { assertThat(SECRET_OPTION.isSensitive()).isTrue(); diff --git a/fluss-common/src/test/java/org/apache/fluss/config/GlobalConfigurationTest.java b/fluss-common/src/test/java/org/apache/fluss/config/GlobalConfigurationTest.java new file mode 100644 index 0000000000..3baedb8811 --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/config/GlobalConfigurationTest.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.config; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link GlobalConfiguration}. */ +public class GlobalConfigurationTest { + + @Test + void testSensitiveConfigurationValuesAreHiddenInLogs(@TempDir Path tempDir) throws Exception { + Files.write( + tempDir.resolve(GlobalConfiguration.FLUSS_CONF_FILENAME), + Arrays.asList( + "fs.s3a.access.key: s3-access-key", + "fs.gs.auth.service.account.private.key: gs-private-key", + "fs.oss.accessKeyId: oss-access-key", + "client.security.sasl.password: sasl-password", + "client.filesystem.security.token.renewal.backoff: 10 s", + "client.request.timeout: 30 s"), + StandardCharsets.UTF_8); + + Logger logger = (Logger) LogManager.getLogger(GlobalConfiguration.class); + Level previousLevel = logger.getLevel(); + boolean previousAdditive = logger.isAdditive(); + TestAppender appender = new TestAppender(); + appender.start(); + logger.addAppender(appender); + logger.setLevel(Level.INFO); + logger.setAdditive(false); + + try { + GlobalConfiguration.loadConfiguration(tempDir.toString(), null); + } finally { + logger.removeAppender(appender); + logger.setLevel(previousLevel); + logger.setAdditive(previousAdditive); + appender.stop(); + } + + String logs = String.join("\n", appender.getMessages()); + assertThat(logs) + .contains("Loading configuration property: fs.s3a.access.key=******") + .contains( + "Loading configuration property: " + + "fs.gs.auth.service.account.private.key=******") + .contains("Loading configuration property: fs.oss.accessKeyId=******") + .contains("Loading configuration property: client.security.sasl.password=******") + .contains( + "Loading configuration property: " + + "client.filesystem.security.token.renewal.backoff=10 s") + .contains("Loading configuration property: client.request.timeout=30 s") + .doesNotContain( + "s3-access-key", "gs-private-key", "oss-access-key", "sasl-password"); + } + + private static class TestAppender extends AbstractAppender { + + private final List messages = new ArrayList<>(); + + TestAppender() { + super("test-appender", null, null, true, Property.EMPTY_ARRAY); + } + + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + + private List getMessages() { + return messages; + } + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java index dee0b1fade..df82f606db 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java @@ -21,6 +21,7 @@ import org.apache.fluss.config.ConfigOption; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.ConfigurationUtils; import org.apache.fluss.config.cluster.ConfigValidator; import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.exception.ConfigException; @@ -168,7 +169,9 @@ private void updateCurrentConfig(Map newDynamicConfigs, boolean // Early return if no effective changes if (effectiveChanges.isEmpty()) { - LOG.info("No effective config changes detected for: {}", newDynamicConfigs); + LOG.info( + "No effective config changes detected for: {}", + ConfigurationUtils.hideSensitiveValues(newDynamicConfigs)); return; } @@ -181,7 +184,9 @@ private void updateCurrentConfig(Map newDynamicConfigs, boolean // Update internal state updateInternalState(newConfig, newConfigMap, newDynamicConfigs); - LOG.info("Dynamic configs changed: {}", effectiveChanges); + LOG.info( + "Dynamic configs changed: {}", + ConfigurationUtils.hideSensitiveValues(effectiveChanges)); } /** @@ -287,8 +292,8 @@ private boolean validateConfigChange( LOG.error( "Config validation failed for '{}': {} -> {}. {}", configKey, - oldValue, - newValue, + ConfigurationUtils.hideSensitiveValue(configKey, oldValue), + ConfigurationUtils.hideSensitiveValue(configKey, newValue), e.getMessage()); if (skipErrorConfig) { skippedConfigs.add(configKey); @@ -366,7 +371,7 @@ private void validateSingleConfig(String configKey, String oldValueStr, String n throw new ConfigException( String.format( "Cannot parse '%s' as %s for config '%s': %s", - newValueStr, + ConfigurationUtils.hideSensitiveValue(configKey, newValueStr), configOption.isList() ? "List<" + configOption.getClazz().getSimpleName() + ">" : configOption.getClazz().getSimpleName(), diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java index cb34a2f2ed..48b0e8a423 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java @@ -21,6 +21,7 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.ConfigurationUtils; import org.apache.fluss.config.FlussConfigUtils; import org.apache.fluss.exception.CoordinatorEpochFencedException; import org.apache.fluss.metadata.DatabaseSummary; @@ -1518,7 +1519,7 @@ public void upsertEntityConfigs(Map configs) throws Exception { .forPath(path, ConfigEntityZNode.encode(configs)); } - LOG.info("upsert entity configs {}", configs); + LOG.info("upsert entity configs {}", ConfigurationUtils.hideSensitiveValues(configs)); insertConfigChangeNotification(); }