gharris1727 commented on code in PR #14995: URL: https://github.com/apache/kafka/pull/14995#discussion_r1448064344
########## clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java: ########## @@ -0,0 +1,79 @@ +/* + * 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.kafka.common.config.internals; + +import org.apache.kafka.common.config.ConfigException; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class AllowedPaths { + private List<Path> allowedPaths; + + private AllowedPaths(List<Path> allowedPaths) { + this.allowedPaths = allowedPaths; + } + + /** + * Constructs AllowedPaths with a list of Paths retrieved from {@code configValue}. + * @param configValue {@code allowed.paths} config value which is a string containing comma separated list of paths + * @return AllowedPaths with a list of Paths or null list if the {@code configValue} is null or empty string. + */ + public static AllowedPaths configureAllowedPaths(String configValue) { + if (configValue != null && !configValue.isEmpty()) { + List<Path> allowedPaths = new ArrayList<>(); + + Arrays.stream(configValue.split(",")).forEach(b -> { + Path normalisedPath = Paths.get(b).normalize(); + + if (normalisedPath.isAbsolute() && Files.exists(normalisedPath)) { + allowedPaths.add(normalisedPath); + } else { + throw new ConfigException("Path " + normalisedPath + " is not valid. The path should be absolute and exist"); + } + }); + + return new AllowedPaths(allowedPaths); + } + + return new AllowedPaths(null); + } + + /** + * Checks if the given {@code path} resides in the configured {@code allowed.paths}. + * If {@code allowed.paths} is not configured, the given Path is returned as allowed. + * @param path the Path to check if allowed + * @return Path that can be accessed or null if the given Path does not reside in the configured {@code allowed.paths}. + */ + public Path getIfPathIsAllowed(Path path) { Review Comment: This name sounds like it should return a boolean. Also I noticed that the call-sites duplicate the Path.get(). ```suggestion public Path parseUntrustedPath(String untrustedPath) { ``` ########## clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java: ########## @@ -0,0 +1,79 @@ +/* + * 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.kafka.common.config.internals; + +import org.apache.kafka.common.config.ConfigException; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class AllowedPaths { + private List<Path> allowedPaths; + + private AllowedPaths(List<Path> allowedPaths) { + this.allowedPaths = allowedPaths; + } + + /** + * Constructs AllowedPaths with a list of Paths retrieved from {@code configValue}. + * @param configValue {@code allowed.paths} config value which is a string containing comma separated list of paths + * @return AllowedPaths with a list of Paths or null list if the {@code configValue} is null or empty string. Review Comment: This should probably also have `@throws` explaining under what conditions this throws exceptions. ########## clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java: ########## @@ -0,0 +1,79 @@ +/* + * 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.kafka.common.config.internals; + +import org.apache.kafka.common.config.ConfigException; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class AllowedPaths { + private List<Path> allowedPaths; Review Comment: ```suggestion private final List<Path> allowedPaths; ``` ########## clients/src/main/java/org/apache/kafka/common/config/provider/FileConfigProvider.java: ########## @@ -40,7 +42,13 @@ public class FileConfigProvider implements ConfigProvider { private static final Logger log = LoggerFactory.getLogger(FileConfigProvider.class); + public static final String ALLOWED_PATHS_CONFIG = "allowed.paths"; + public static final String ALLOWED_PATHS_DOC = "A comma separated list of paths that this config provider is " + + "allowed to access. If not set, all paths are allowed."; + private AllowedPaths allowedPaths = null; Review Comment: If someone does not call configure() before get() (as the FileConfigProvider tests previously did) then this will cause an NPE. I think either that should preserve the old behavior (allowing all paths) or should throw an IllegalStateException. There might be an opportunity to attack this class by preventing configure() from being called somehow, so perhaps the IllegalStateException is more secure. ########## clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java: ########## @@ -0,0 +1,79 @@ +/* + * 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.kafka.common.config.internals; + +import org.apache.kafka.common.config.ConfigException; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class AllowedPaths { + private List<Path> allowedPaths; + + private AllowedPaths(List<Path> allowedPaths) { + this.allowedPaths = allowedPaths; + } + + /** + * Constructs AllowedPaths with a list of Paths retrieved from {@code configValue}. + * @param configValue {@code allowed.paths} config value which is a string containing comma separated list of paths + * @return AllowedPaths with a list of Paths or null list if the {@code configValue} is null or empty string. + */ + public static AllowedPaths configureAllowedPaths(String configValue) { Review Comment: Since this is a static factory method, it is called like `AllowedPaths.configureAllowedPaths(configValue)`. This feels repetitive, including the name of the class twice. Also, "configure" is a technical term that I associate with the `Configurable` interface; I would expect this method to mutate an existing AllowedPaths but it constructs a new one. WDYT about instead just having the class constructor take a String, so that the entrypoint to this class is `new AllowedPaths(configValue)`? That would remove the repetition and make it clear that a new AllowedPaths is being constructed. -- 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]
