morazow commented on code in PR #3658: URL: https://github.com/apache/fluss/pull/3658#discussion_r3601553769
########## fluss-common/src/main/java/org/apache/fluss/config/provider/ConfigProviders.java: ########## @@ -0,0 +1,211 @@ +/* + * 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.provider; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.IllegalConfigurationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.ServiceLoader; +import java.util.Set; + +/** + * Resolves <code>${identifier:[path:]key}</code> markers in a {@link Configuration} through the + * {@link ConfigProvider}s declared in {@link ConfigOptions#CONFIG_PROVIDERS}. A marker must be the + * whole value; a literal leading <code>${</code> can be escaped as <code>$${</code>. + */ +@Internal +public final class ConfigProviders { + + private static final Logger LOG = LoggerFactory.getLogger(ConfigProviders.class); + + private static final String PROVIDER_PREFIX = ConfigOptions.CONFIG_PROVIDERS.key() + "."; + private static final String PARAM_INFIX = ".param."; + + private ConfigProviders() {} + + /** + * Resolves all markers in {@code config} in place. A no-op when {@link + * ConfigOptions#CONFIG_PROVIDERS} is absent. + * + * @throws IllegalConfigurationException if a marker is malformed, references an undeclared + * provider, or fails to resolve + */ + public static void resolve(Configuration config) { + List<String> declared = config.get(ConfigOptions.CONFIG_PROVIDERS); + if (declared == null || declared.isEmpty()) { + warnAboutUnresolvableMarkers(config); + return; + } + + Map<String, ConfigProvider> providers = loadProviders(config, declared); + try { + for (String key : config.keySet()) { + Optional<Object> raw = config.getRawValue(key); + if (!raw.isPresent() || !(raw.get() instanceof String)) { + continue; + } + String value = (String) raw.get(); + if (value.startsWith("$${")) { + // escaped: drop one '$', leave the rest literal + config.setString(key, value.substring(1)); + } else if (isMarker(value)) { + config.setString(key, resolveMarker(value, providers)); + // hide the resolved secret even under a key name that doesn't look sensitive + config.markSensitive(key); + } + } + } finally { + closeAll(providers); + } + } + + private static String resolveMarker(String value, Map<String, ConfigProvider> providers) { + String inner = value.substring(2, value.length() - 1); + int firstColon = inner.indexOf(':'); + if (firstColon <= 0 || firstColon == inner.length() - 1) { + throw new IllegalConfigurationException( + "Malformed config provider marker '" + + value + + "'. Expected ${identifier:[path:]key}."); + } + String identifier = inner.substring(0, firstColon); + String rest = inner.substring(firstColon + 1); + int lastColon = rest.lastIndexOf(':'); + String path = lastColon < 0 ? "" : rest.substring(0, lastColon); + String key = lastColon < 0 ? rest : rest.substring(lastColon + 1); + + ConfigProvider provider = providers.get(identifier); + if (provider == null) { + throw new IllegalConfigurationException( + "Config provider '" + + identifier + + "' referenced by marker '" + + value + + "' is not declared in '" + + ConfigOptions.CONFIG_PROVIDERS.key() + + "'."); + } + try { + return provider.resolve(path, key); + } catch (Exception e) { + throw new IllegalConfigurationException( + "Failed to resolve config provider marker '" + value + "': " + e.getMessage(), + e); + } + } + + /** Whether the value has the <code>${...}</code> marker shape. */ + public static boolean isMarker(String value) { + return value.length() > 3 && value.startsWith("${") && value.endsWith("}"); + } + + private static void warnAboutUnresolvableMarkers(Configuration config) { + for (String key : config.keySet()) { + Optional<Object> raw = config.getRawValue(key); + if (raw.isPresent() && raw.get() instanceof String && isMarker((String) raw.get())) { + LOG.warn( + "Value of '{}' looks like a config provider marker, but '{}' is not set; " + + "it is kept as a literal.", + key, + ConfigOptions.CONFIG_PROVIDERS.key()); + } + } + } + + private static Map<String, ConfigProvider> loadProviders( + Configuration config, List<String> declared) { + Set<String> requested = new HashSet<>(); + for (String name : declared) { + String trimmed = name.trim(); + if (!trimmed.isEmpty()) { + requested.add(trimmed); + } + } + + Map<String, ConfigProvider> providers = new HashMap<>(); + try { + ServiceLoader<ConfigProvider> loader = + ServiceLoader.load(ConfigProvider.class, ConfigProvider.class.getClassLoader()); + for (Iterator<ConfigProvider> it = loader.iterator(); it.hasNext(); ) { + ConfigProvider provider = it.next(); + if (requested.contains(provider.identifier()) + && !providers.containsKey(provider.identifier())) { + try { + provider.configure(paramsFor(config, provider.identifier())); + } catch (Exception e) { + throw new IllegalConfigurationException( + "Invalid configuration for config provider '" + + provider.identifier() + + "': " + + e.getMessage(), + e); + } + providers.put(provider.identifier(), provider); + } + } + + for (String name : requested) { + if (!providers.containsKey(name)) { Review Comment: I wonder if we should fail for duplicate keys. E.g, for cases if 3rd party jar provider with similar name is included. ########## fluss-common/src/test/java/org/apache/fluss/config/provider/ConfigProvidersTest.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.provider; + +import org.apache.fluss.config.ConfigBuilder; +import org.apache.fluss.config.ConfigOption; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.GlobalConfiguration; +import org.apache.fluss.exception.IllegalConfigurationException; +import org.apache.fluss.utils.InstantiationUtils; + +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.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** Tests for {@link ConfigProviders} and the built-in {@link ConfigProvider}s. */ +class ConfigProvidersTest { + + @Test + void testProviderDefaultMethods() { + ConfigProvider provider = + new ConfigProvider() { + @Override + public String identifier() { + return "noop"; + } + + @Override + public String resolve(String path, String key) { + return path + key; + } + }; + provider.configure(Collections.emptyMap()); + assertThat(provider.resolve("a", "b")).isEqualTo("ab"); + provider.close(); + } + + @Test + void testNoProvidersDeclaredLeavesMarkersUntouched() { + Configuration config = new Configuration(); + config.setString("datalake.paimon.oauth.token", "${env:WHATEVER}"); + ConfigProviders.resolve(config); + assertThat(config.toMap().get("datalake.paimon.oauth.token")).isEqualTo("${env:WHATEVER}"); + } + + @Test + void testDirectoryProviderReadsFileVerbatim(@TempDir Path secretsDir) throws Exception { + // K8s Secret volumes expose exact bytes with no trailing newline. + Files.write(secretsDir.resolve("oauth-token"), "s3cr3t".getBytes(StandardCharsets.UTF_8)); + + Configuration config = new Configuration(); + config.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "directory"); + config.setString("config.providers.directory.param.allowed.paths", secretsDir.toString()); + config.setString( + "datalake.paimon.oauth.token", "${directory:" + secretsDir + ":oauth-token}"); + ConfigProviders.resolve(config); + + assertThat(config.toMap().get("datalake.paimon.oauth.token")).isEqualTo("s3cr3t"); + } + + @Test + void testDirectoryProviderRequiresAllowedPaths() { + Configuration config = new Configuration(); + config.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "directory"); + assertThatThrownBy(() -> ConfigProviders.resolve(config)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("requires the 'allowed.paths' parameter"); + } + + @Test + void testUnknownProviderParamFailsFast(@TempDir Path secretsDir) { + Configuration config = new Configuration(); + config.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "directory"); + config.setString("config.providers.directory.param.allowd.paths", secretsDir.toString()); + assertThatThrownBy(() -> ConfigProviders.resolve(config)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("Unknown parameter 'allowd.paths'"); + } + + @Test + void testDirectoryProviderAllowedPathsRejectsOutsideRoot(@TempDir Path base) throws Exception { + Path allowed = base.resolve("allowed"); + Path other = base.resolve("other"); + Files.createDirectories(allowed); + Files.createDirectories(other); + Files.write(other.resolve("token"), "leak".getBytes(StandardCharsets.UTF_8)); + + Configuration config = new Configuration(); + config.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "directory"); + config.setString("config.providers.directory.param.allowed.paths", allowed.toString()); + config.setString("secret", "${directory:" + other + ":token}"); + + assertThatThrownBy(() -> ConfigProviders.resolve(config)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("allowed.paths"); + } + + @Test + void testFileProviderReadsProperty(@TempDir Path dir) throws Exception { + Path props = dir.resolve("creds.properties"); + Files.write(props, Arrays.asList("db.user=alice", "db.password=hunter2")); + + Configuration config = new Configuration(); + config.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "file"); + config.setString("config.providers.file.param.allowed.paths", dir.toString()); + config.setString("kv.password", "${file:" + props + ":db.password}"); + ConfigProviders.resolve(config); + + assertThat(config.toMap().get("kv.password")).isEqualTo("hunter2"); + } + + @Test + void testEnvProviderResolvesAndRejectsMissing() { + assumeTrue(System.getenv("PATH") != null, "requires PATH to be set"); + + Configuration config = new Configuration(); + config.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "env"); + config.setString("some.path", "${env:PATH}"); + ConfigProviders.resolve(config); + assertThat(config.toMap().get("some.path")).isEqualTo(System.getenv("PATH")); + + Configuration missing = new Configuration(); + missing.setString(ConfigOptions.CONFIG_PROVIDERS.key(), "env"); + missing.setString("x", "${env:FLUSS_DEFINITELY_UNSET_VAR_XYZ}"); + assertThatThrownBy(() -> ConfigProviders.resolve(missing)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("FLUSS_DEFINITELY_UNSET_VAR_XYZ"); + } + + @Test + void testDoubleDollarEscapesLiteralMarker() { Review Comment: I think we should have similar test, without config providers defined. It should fail because `$${...}` will not be escaped. Or in general run the escape processing always even if the config providers are not defined. ########## fluss-common/src/main/java/org/apache/fluss/config/provider/AllowedPaths.java: ########## @@ -0,0 +1,97 @@ +/* + * 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.provider; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; + +/** + * The mandatory {@code allowed.paths} parameter of file-reading {@link ConfigProvider}s: a + * comma-separated list of roots the provider may read from. + */ +final class AllowedPaths { + + static final String PARAM = "allowed.paths"; + + private final String identifier; + private final List<Path> roots; + + private AllowedPaths(String identifier, List<Path> roots) { + this.identifier = identifier; + this.roots = roots; + } + + static AllowedPaths parse(String identifier, @Nullable String value) { + List<Path> roots = new ArrayList<>(); + if (value != null) { + for (String root : value.split(",")) { + String trimmed = root.trim(); + if (!trimmed.isEmpty()) { + roots.add(canonicalize(Paths.get(trimmed))); + } + } + } + if (roots.isEmpty()) { + throw new IllegalArgumentException( + "The '" + + identifier + + "' config provider requires the '" + + PARAM + + "' parameter (comma-separated roots it may read from); set " + + "config.providers." + + identifier + + ".param." + + PARAM + + "."); + } + return new AllowedPaths(identifier, roots); + } + + /** Canonicalizes {@code file} and checks it lies under one of the allowed roots. */ + Path check(Path file) { + Path canonical = canonicalize(file); + for (Path root : roots) { + if (canonical.startsWith(root)) { + return canonical; + } + } + throw new IllegalArgumentException( + "File '" + + canonical + + "' is outside the " + + PARAM + + " " + + roots + + " of the '" + + identifier + + "' config provider."); + } + + private static Path canonicalize(Path path) { + try { + return Paths.get(path.toFile().getCanonicalPath()); + } catch (IOException e) { + return path.toAbsolutePath().normalize(); Review Comment: Should we fall through here? Throw with error. ```suggestion throw new IllegalArgumentException("Cannot canonicalize path '" + path + "' for allowed-path check.", e); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
