github-advanced-security[bot] commented on code in PR #4419: URL: https://github.com/apache/streampark/pull/4419#discussion_r3540526589
########## streampark-common/src/main/java/org/apache/streampark/common/util/FileUtils.java: ########## @@ -0,0 +1,299 @@ +/* + * 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.streampark.common.util; + +import java.io.BufferedInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.RandomAccessFile; +import java.io.Serializable; +import java.net.URL; +import java.nio.ByteBuffer; +import java.nio.channels.Channels; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Scanner; +import java.util.stream.Collectors; + +public final class FileUtils { + + private FileUtils() {} + + private static String bytesToHexString(byte[] src) { + if (src == null || src.length <= 0) { + return null; + } + StringBuilder stringBuilder = new StringBuilder(); + for (byte b : src) { + int v = b & 0xFF; + String hv = Integer.toHexString(v).toUpperCase(); + if (hv.length() < 2) { + stringBuilder.append(0); + } + stringBuilder.append(hv); + } + return stringBuilder.toString(); + } + + public static boolean isJarFileType(InputStream input) { + if (input == null) { + throw new RuntimeException("The inputStream can not be null"); + } + return AutoCloseUtils.using( + input, + in -> { + byte[] b = new byte[4]; + try { + in.read(b, 0, b.length); + } catch (IOException e) { + throw new RuntimeException(e); + } + return bytesToHexString(b); + }) + .equals("504B0304"); + } + + public static boolean isJarFileType(File file) throws IOException { + if (!file.exists() || !file.isFile()) { + throw new RuntimeException("The file does not exist or the path is a directory"); + } + return isJarFileType(new FileInputStream(file)); + } + + public static File createTempDir() { + final int TEMP_DIR_ATTEMPTS = 10000; + File baseDir = new File(System.getProperty("java.io.tmpdir")); + String baseName = System.currentTimeMillis() + "-"; + for (int counter = 0; counter < TEMP_DIR_ATTEMPTS; counter++) { + File tempDir = new File(baseDir, baseName + counter); + if (tempDir.mkdir()) { + return tempDir; + } + } + throw new IllegalStateException( + "[StreamPark] Failed to create directory within " + + TEMP_DIR_ATTEMPTS + + " attempts (tried " + + baseName + + " 0 to " + + baseName + + (TEMP_DIR_ATTEMPTS - 1) + + ")"); + } + + public static void mkdir(File dir) throws IOException { + if (dir.exists() && !dir.isDirectory()) { + throw new IOException("File " + dir + " exists and is not a directory. Unable to create directory."); + } else if (!dir.mkdirs() && !dir.isDirectory()) { + throw new IOException("Unable to create directory " + dir); + } + } + + public static String getPathFromEnv(String env) { + String path = System.getenv(env); + if (path == null) { + path = System.getProperty(env); + } + AssertUtils.notNull(path, "[StreamPark] FileUtils.getPathFromEnv: " + env + " is not set on system env"); + File file = new File(path); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] FileUtils.getPathFromEnv: " + env + " is not exist!"); + } + return file.getAbsolutePath(); + } + + public static String resolvePath(String parent, String child) { + File file = new File(parent, child); + if (!file.exists()) { + throw new IllegalArgumentException( + "[StreamPark] FileUtils.resolvePath: " + file.getAbsolutePath() + " is not exist!"); + } + return file.getAbsolutePath(); + } + + public static String getSuffix(String filename) { + AssertUtils.notNull(filename); + return filename.substring(filename.lastIndexOf(".")); + } + + public static List<URL> listFileAsURL(String dirPath) { + File dir = new File(dirPath); + if (dir.exists() && dir.isDirectory()) { + File[] files = dir.listFiles(); + if (files != null && files.length > 0) { + List<URL> urls = new ArrayList<>(); + for (File f : files) { + try { + urls.add(f.toURI().toURL()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + return urls; + } + } + return Collections.emptyList(); + } + + public static boolean exists(Serializable file) { + if (file == null) { + return false; + } + if (file instanceof File) { + return ((File) file).exists(); + } + return new File(file.toString()).exists(); + } + + public static boolean directoryNotBlank(Serializable file) { + if (file == null) { + return false; + } + File f = file instanceof File ? (File) file : new File(file.toString()); + String[] list = f.list(); + return f.isDirectory() && list != null && list.length > 0; + } + + public static boolean equals(File file1, File file2) throws IOException { + if (file1 == null || file2 == null) { + return false; + } + if (!file1.exists() || !file2.exists()) { + return false; + } + if (file1.getAbsolutePath().equals(file2.getAbsolutePath())) { + return true; + } + BufferedInputStream first = new BufferedInputStream(new FileInputStream(file1)); + BufferedInputStream second = new BufferedInputStream(new FileInputStream(file2)); + if (first.available() != second.available()) { + Utils.close(first, second); + return false; + } + while (true) { + int firRead = first.read(); + int secRead = second.read(); + if (firRead != secRead) { + Utils.close(first, second); + return false; + } + if (firRead == -1) { + Utils.close(first, second); + return true; + } + } + } + + public static void readInputStream(InputStream in, byte[] array) throws IOException { + try (InputStream input = in) { + int toRead = array.length; + int off = 0; + while (toRead > 0) { + int ret = input.read(array, off, toRead); + if (ret < 0) { + throw new IOException("Bad inputStream, premature EOF"); + } + toRead -= ret; + off += ret; + } + } + } + + public static String readFile(File file) throws IOException { + if (file.length() >= Integer.MAX_VALUE) { + throw new IOException("Too large file, unexpected!"); + } + byte[] array = new byte[(int) file.length()]; + try (InputStream is = Files.newInputStream(file.toPath())) { Review Comment: ## SonarCloud / I/O function calls should not be vulnerable to path injection attacks <!--SONAR_ISSUE_KEY:AZ89Cu3mj5FB4xK0cb1l-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu3mj5FB4xK0cb1l&open=AZ89Cu3mj5FB4xK0cb1l&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/188) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromYamlFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading yaml from file", e); + } + } + + public static Map<String, String> fromHoconFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromHoconFile: file " + file + " does not exist"); + } + try (InputStream inputStream = new FileInputStream(file)) { Review Comment: ## SonarCloud / I/O function calls should not be vulnerable to path injection attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1q-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1q&open=AZ89Cu4Mj5FB4xK0cb1q&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/189) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromYamlFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading yaml from file", e); + } + } + + public static Map<String, String> fromHoconFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromHoconFile: file " + file + " does not exist"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromHoconFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException( + "[StreamPark] fromPropertiesFile: Properties file " + file + " does not exist"); + } + if (!file.isFile()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1t-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1t&open=AZ89Cu4Mj5FB4xK0cb1t&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/201) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromYamlFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading yaml from file", e); + } + } + + public static Map<String, String> fromHoconFile(String filename) { + File file = new File(filename); + if (!file.exists()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1v-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1v&open=AZ89Cu4Mj5FB4xK0cb1v&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/203) ########## streampark-common/src/main/java/org/apache/streampark/common/util/Utils.java: ########## @@ -0,0 +1,243 @@ +/* + * 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.streampark.common.util; + +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; + +import java.io.BufferedInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.Flushable; +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URL; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.Collection; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.locks.LockSupport; +import java.util.jar.JarFile; +import java.util.jar.JarInputStream; +import java.util.jar.Manifest; + +/** General utility methods. */ +public final class Utils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(Utils.class.getName()); + + private static final String OS = System.getProperty("os.name").toLowerCase(); + + private Utils() {} + + public static boolean isNotEmpty(Object elem) { + if (elem == null) { + return false; + } + if (elem instanceof Object[]) { + return ((Object[]) elem).length > 0; + } + if (elem instanceof CharSequence) { + return elem.toString().trim().length() > 0; + } + if (elem instanceof Collection) { + return !((Collection<?>) elem).isEmpty(); + } + if (elem instanceof Iterable) { + return ((Iterable<?>) elem).iterator().hasNext(); + } + if (elem instanceof Map) { + return !((Map<?, ?>) elem).isEmpty(); + } + return true; + } + + public static boolean isEmpty(Object elem) { + return !isNotEmpty(elem); + } + + public static String uuid() { + return UUID.randomUUID().toString().replaceAll("-", ""); + } + + public static void requireCheckJarFile(URL jar) throws IOException { + File jarFile; + try { + jarFile = new File(jar.toURI()); + } catch (Exception e) { + throw new IOException("JAR file path is invalid " + jar, e); + } + if (!jarFile.exists()) { + throw new IOException("JAR file does not exist '" + jarFile.getAbsolutePath() + "'"); + } + if (!jarFile.canRead()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu1qj5FB4xK0cb1f-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu1qj5FB4xK0cb1f&open=AZ89Cu1qj5FB4xK0cb1f&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/197) ########## streampark-common/src/main/java/org/apache/streampark/common/util/FileUtils.java: ########## @@ -0,0 +1,299 @@ +/* + * 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.streampark.common.util; + +import java.io.BufferedInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.RandomAccessFile; +import java.io.Serializable; +import java.net.URL; +import java.nio.ByteBuffer; +import java.nio.channels.Channels; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Scanner; +import java.util.stream.Collectors; + +public final class FileUtils { + + private FileUtils() {} + + private static String bytesToHexString(byte[] src) { + if (src == null || src.length <= 0) { + return null; + } + StringBuilder stringBuilder = new StringBuilder(); + for (byte b : src) { + int v = b & 0xFF; + String hv = Integer.toHexString(v).toUpperCase(); + if (hv.length() < 2) { + stringBuilder.append(0); + } + stringBuilder.append(hv); + } + return stringBuilder.toString(); + } + + public static boolean isJarFileType(InputStream input) { + if (input == null) { + throw new RuntimeException("The inputStream can not be null"); + } + return AutoCloseUtils.using( + input, + in -> { + byte[] b = new byte[4]; + try { + in.read(b, 0, b.length); + } catch (IOException e) { + throw new RuntimeException(e); + } + return bytesToHexString(b); + }) + .equals("504B0304"); + } + + public static boolean isJarFileType(File file) throws IOException { + if (!file.exists() || !file.isFile()) { + throw new RuntimeException("The file does not exist or the path is a directory"); + } + return isJarFileType(new FileInputStream(file)); + } + + public static File createTempDir() { + final int TEMP_DIR_ATTEMPTS = 10000; + File baseDir = new File(System.getProperty("java.io.tmpdir")); + String baseName = System.currentTimeMillis() + "-"; + for (int counter = 0; counter < TEMP_DIR_ATTEMPTS; counter++) { + File tempDir = new File(baseDir, baseName + counter); + if (tempDir.mkdir()) { + return tempDir; + } + } + throw new IllegalStateException( + "[StreamPark] Failed to create directory within " + + TEMP_DIR_ATTEMPTS + + " attempts (tried " + + baseName + + " 0 to " + + baseName + + (TEMP_DIR_ATTEMPTS - 1) + + ")"); + } + + public static void mkdir(File dir) throws IOException { + if (dir.exists() && !dir.isDirectory()) { + throw new IOException("File " + dir + " exists and is not a directory. Unable to create directory."); + } else if (!dir.mkdirs() && !dir.isDirectory()) { + throw new IOException("Unable to create directory " + dir); + } + } + + public static String getPathFromEnv(String env) { + String path = System.getenv(env); + if (path == null) { + path = System.getProperty(env); + } + AssertUtils.notNull(path, "[StreamPark] FileUtils.getPathFromEnv: " + env + " is not set on system env"); + File file = new File(path); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] FileUtils.getPathFromEnv: " + env + " is not exist!"); + } + return file.getAbsolutePath(); + } + + public static String resolvePath(String parent, String child) { + File file = new File(parent, child); + if (!file.exists()) { + throw new IllegalArgumentException( + "[StreamPark] FileUtils.resolvePath: " + file.getAbsolutePath() + " is not exist!"); + } + return file.getAbsolutePath(); + } + + public static String getSuffix(String filename) { + AssertUtils.notNull(filename); + return filename.substring(filename.lastIndexOf(".")); + } + + public static List<URL> listFileAsURL(String dirPath) { + File dir = new File(dirPath); + if (dir.exists() && dir.isDirectory()) { + File[] files = dir.listFiles(); + if (files != null && files.length > 0) { + List<URL> urls = new ArrayList<>(); + for (File f : files) { + try { + urls.add(f.toURI().toURL()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + return urls; + } + } + return Collections.emptyList(); + } + + public static boolean exists(Serializable file) { + if (file == null) { + return false; + } + if (file instanceof File) { + return ((File) file).exists(); + } + return new File(file.toString()).exists(); + } + + public static boolean directoryNotBlank(Serializable file) { + if (file == null) { + return false; + } + File f = file instanceof File ? (File) file : new File(file.toString()); + String[] list = f.list(); + return f.isDirectory() && list != null && list.length > 0; + } + + public static boolean equals(File file1, File file2) throws IOException { + if (file1 == null || file2 == null) { + return false; + } + if (!file1.exists() || !file2.exists()) { + return false; + } + if (file1.getAbsolutePath().equals(file2.getAbsolutePath())) { + return true; + } + BufferedInputStream first = new BufferedInputStream(new FileInputStream(file1)); + BufferedInputStream second = new BufferedInputStream(new FileInputStream(file2)); + if (first.available() != second.available()) { + Utils.close(first, second); + return false; + } + while (true) { + int firRead = first.read(); + int secRead = second.read(); + if (firRead != secRead) { + Utils.close(first, second); + return false; + } + if (firRead == -1) { + Utils.close(first, second); + return true; + } + } + } + + public static void readInputStream(InputStream in, byte[] array) throws IOException { + try (InputStream input = in) { + int toRead = array.length; + int off = 0; + while (toRead > 0) { + int ret = input.read(array, off, toRead); + if (ret < 0) { + throw new IOException("Bad inputStream, premature EOF"); + } + toRead -= ret; + off += ret; + } + } + } + + public static String readFile(File file) throws IOException { + if (file.length() >= Integer.MAX_VALUE) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu3mj5FB4xK0cb1m-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu3mj5FB4xK0cb1m&open=AZ89Cu3mj5FB4xK0cb1m&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/199) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { Review Comment: ## SonarCloud / I/O function calls should not be vulnerable to path injection attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1s-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1s&open=AZ89Cu4Mj5FB4xK0cb1s&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/191) ########## streampark-common/src/main/java/org/apache/streampark/common/util/FileUtils.java: ########## @@ -0,0 +1,299 @@ +/* + * 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.streampark.common.util; + +import java.io.BufferedInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.RandomAccessFile; +import java.io.Serializable; +import java.net.URL; +import java.nio.ByteBuffer; +import java.nio.channels.Channels; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Scanner; +import java.util.stream.Collectors; + +public final class FileUtils { + + private FileUtils() {} + + private static String bytesToHexString(byte[] src) { + if (src == null || src.length <= 0) { + return null; + } + StringBuilder stringBuilder = new StringBuilder(); + for (byte b : src) { + int v = b & 0xFF; + String hv = Integer.toHexString(v).toUpperCase(); + if (hv.length() < 2) { + stringBuilder.append(0); + } + stringBuilder.append(hv); + } + return stringBuilder.toString(); + } + + public static boolean isJarFileType(InputStream input) { + if (input == null) { + throw new RuntimeException("The inputStream can not be null"); + } + return AutoCloseUtils.using( + input, + in -> { + byte[] b = new byte[4]; + try { + in.read(b, 0, b.length); + } catch (IOException e) { + throw new RuntimeException(e); + } + return bytesToHexString(b); + }) + .equals("504B0304"); + } + + public static boolean isJarFileType(File file) throws IOException { + if (!file.exists() || !file.isFile()) { + throw new RuntimeException("The file does not exist or the path is a directory"); + } + return isJarFileType(new FileInputStream(file)); + } + + public static File createTempDir() { + final int TEMP_DIR_ATTEMPTS = 10000; + File baseDir = new File(System.getProperty("java.io.tmpdir")); + String baseName = System.currentTimeMillis() + "-"; + for (int counter = 0; counter < TEMP_DIR_ATTEMPTS; counter++) { + File tempDir = new File(baseDir, baseName + counter); + if (tempDir.mkdir()) { + return tempDir; + } + } + throw new IllegalStateException( + "[StreamPark] Failed to create directory within " + + TEMP_DIR_ATTEMPTS + + " attempts (tried " + + baseName + + " 0 to " + + baseName + + (TEMP_DIR_ATTEMPTS - 1) + + ")"); + } + + public static void mkdir(File dir) throws IOException { + if (dir.exists() && !dir.isDirectory()) { + throw new IOException("File " + dir + " exists and is not a directory. Unable to create directory."); + } else if (!dir.mkdirs() && !dir.isDirectory()) { + throw new IOException("Unable to create directory " + dir); + } + } + + public static String getPathFromEnv(String env) { + String path = System.getenv(env); + if (path == null) { + path = System.getProperty(env); + } + AssertUtils.notNull(path, "[StreamPark] FileUtils.getPathFromEnv: " + env + " is not set on system env"); + File file = new File(path); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] FileUtils.getPathFromEnv: " + env + " is not exist!"); + } + return file.getAbsolutePath(); + } + + public static String resolvePath(String parent, String child) { + File file = new File(parent, child); + if (!file.exists()) { + throw new IllegalArgumentException( + "[StreamPark] FileUtils.resolvePath: " + file.getAbsolutePath() + " is not exist!"); + } + return file.getAbsolutePath(); + } + + public static String getSuffix(String filename) { + AssertUtils.notNull(filename); + return filename.substring(filename.lastIndexOf(".")); + } + + public static List<URL> listFileAsURL(String dirPath) { + File dir = new File(dirPath); + if (dir.exists() && dir.isDirectory()) { + File[] files = dir.listFiles(); + if (files != null && files.length > 0) { + List<URL> urls = new ArrayList<>(); + for (File f : files) { + try { + urls.add(f.toURI().toURL()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + return urls; + } + } + return Collections.emptyList(); + } + + public static boolean exists(Serializable file) { + if (file == null) { + return false; + } + if (file instanceof File) { + return ((File) file).exists(); + } + return new File(file.toString()).exists(); + } + + public static boolean directoryNotBlank(Serializable file) { + if (file == null) { + return false; + } + File f = file instanceof File ? (File) file : new File(file.toString()); + String[] list = f.list(); + return f.isDirectory() && list != null && list.length > 0; + } + + public static boolean equals(File file1, File file2) throws IOException { + if (file1 == null || file2 == null) { + return false; + } + if (!file1.exists() || !file2.exists()) { + return false; + } + if (file1.getAbsolutePath().equals(file2.getAbsolutePath())) { + return true; + } + BufferedInputStream first = new BufferedInputStream(new FileInputStream(file1)); + BufferedInputStream second = new BufferedInputStream(new FileInputStream(file2)); + if (first.available() != second.available()) { + Utils.close(first, second); + return false; + } + while (true) { + int firRead = first.read(); + int secRead = second.read(); + if (firRead != secRead) { + Utils.close(first, second); + return false; + } + if (firRead == -1) { + Utils.close(first, second); + return true; + } + } + } + + public static void readInputStream(InputStream in, byte[] array) throws IOException { + try (InputStream input = in) { + int toRead = array.length; + int off = 0; + while (toRead > 0) { + int ret = input.read(array, off, toRead); + if (ret < 0) { + throw new IOException("Bad inputStream, premature EOF"); + } + toRead -= ret; + off += ret; + } + } + } + + public static String readFile(File file) throws IOException { + if (file.length() >= Integer.MAX_VALUE) { + throw new IOException("Too large file, unexpected!"); + } + byte[] array = new byte[(int) file.length()]; Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu3mj5FB4xK0cb1n-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu3mj5FB4xK0cb1n&open=AZ89Cu3mj5FB4xK0cb1n&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/200) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromYamlFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading yaml from file", e); + } + } + + public static Map<String, String> fromHoconFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromHoconFile: file " + file + " does not exist"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromHoconFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException( + "[StreamPark] fromPropertiesFile: Properties file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException( + "[StreamPark] fromPropertiesFile: Properties file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { Review Comment: ## SonarCloud / I/O function calls should not be vulnerable to path injection attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1r-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1r&open=AZ89Cu4Mj5FB4xK0cb1r&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/190) ########## streampark-common/src/main/java/org/apache/streampark/common/util/Utils.java: ########## @@ -0,0 +1,243 @@ +/* + * 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.streampark.common.util; + +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; + +import java.io.BufferedInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.Flushable; +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URL; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.Collection; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.locks.LockSupport; +import java.util.jar.JarFile; +import java.util.jar.JarInputStream; +import java.util.jar.Manifest; + +/** General utility methods. */ +public final class Utils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(Utils.class.getName()); + + private static final String OS = System.getProperty("os.name").toLowerCase(); + + private Utils() {} + + public static boolean isNotEmpty(Object elem) { + if (elem == null) { + return false; + } + if (elem instanceof Object[]) { + return ((Object[]) elem).length > 0; + } + if (elem instanceof CharSequence) { + return elem.toString().trim().length() > 0; + } + if (elem instanceof Collection) { + return !((Collection<?>) elem).isEmpty(); + } + if (elem instanceof Iterable) { + return ((Iterable<?>) elem).iterator().hasNext(); + } + if (elem instanceof Map) { + return !((Map<?, ?>) elem).isEmpty(); + } + return true; + } + + public static boolean isEmpty(Object elem) { + return !isNotEmpty(elem); + } + + public static String uuid() { + return UUID.randomUUID().toString().replaceAll("-", ""); + } + + public static void requireCheckJarFile(URL jar) throws IOException { + File jarFile; + try { + jarFile = new File(jar.toURI()); + } catch (Exception e) { + throw new IOException("JAR file path is invalid " + jar, e); + } + if (!jarFile.exists()) { + throw new IOException("JAR file does not exist '" + jarFile.getAbsolutePath() + "'"); + } + if (!jarFile.canRead()) { + throw new IOException("JAR file can't be read '" + jarFile.getAbsolutePath() + "'"); + } + try (JarFile jf = new JarFile(jarFile)) { + // verify jar is readable + } catch (IOException e) { + throw new IOException( + "Error while opening jar file '" + jarFile.getAbsolutePath() + "'", e); + } + } + + public static Manifest getJarManifest(File jarFile) throws IOException { + requireCheckJarFile(jarFile.toURI().toURL()); + return AutoCloseUtils.using( + new JarInputStream(new BufferedInputStream(new FileInputStream(jarFile))), Review Comment: ## SonarCloud / I/O function calls should not be vulnerable to path injection attacks <!--SONAR_ISSUE_KEY:AZ89Cu1qj5FB4xK0cb1e-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu1qj5FB4xK0cb1e&open=AZ89Cu1qj5FB4xK0cb1e&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/187) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1x-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1x&open=AZ89Cu4Mj5FB4xK0cb1x&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/205) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1w-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1w&open=AZ89Cu4Mj5FB4xK0cb1w&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/204) ########## streampark-common/src/main/java/org/apache/streampark/common/util/PropertiesUtils.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.streampark.common.util; + +import com.typesafe.config.ConfigFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; +import org.yaml.snakeyaml.Yaml; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.StringReader; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Scanner; + +public final class PropertiesUtils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(PropertiesUtils.class.getName()); + + private PropertiesUtils() {} + + public static String readFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " is not a normal file"); + } + try (Scanner scanner = new Scanner(file)) { + StringBuilder buffer = new StringBuilder(); + while (scanner.hasNextLine()) { + buffer.append(scanner.nextLine()).append("\r\n"); + } + return buffer.toString(); + } catch (java.io.FileNotFoundException e) { + throw new IllegalArgumentException("[StreamPark] readFile: file " + file + " does not exist", e); + } + } + + public static Map<String, String> fromYamlText(String text) { + try { + Map<String, Object> map = new Yaml().load(text); + return flatten(map); + } catch (Exception e) { + throw new IllegalArgumentException("Failed when loading conf error:", e); + } + } + + public static Map<String, String> fromHoconText(String conf) { + if (conf == null) { + throw new IllegalArgumentException("[StreamPark] fromHoconText: Hocon content must not be null"); + } + try { + return parseHoconByReader(new StringReader(conf)); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesText(String conf) { + try { + Properties properties = new Properties(); + properties.load(new StringReader(conf)); + Map<String, String> result = new HashMap<>(); + for (String k : properties.stringPropertyNames()) { + result.put(k, properties.getProperty(k).trim()); + } + return result; + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading properties ", e); + } + } + + public static Map<String, String> fromYamlFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " does not exist"); + } + if (!file.isFile()) { + throw new IllegalArgumentException("[StreamPark] fromYamlFile: Yaml file " + file + " is not a normal file"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromYamlFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading yaml from file", e); + } + } + + public static Map<String, String> fromHoconFile(String filename) { + File file = new File(filename); + if (!file.exists()) { + throw new IllegalArgumentException("[StreamPark] fromHoconFile: file " + file + " does not exist"); + } + try (InputStream inputStream = new FileInputStream(file)) { + return fromHoconFile(inputStream); + } catch (IOException e) { + throw new IllegalArgumentException("Failed when loading Hocon ", e); + } + } + + public static Map<String, String> fromPropertiesFile(String filename) { + File file = new File(filename); + if (!file.exists()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu4Mj5FB4xK0cb1u-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu4Mj5FB4xK0cb1u&open=AZ89Cu4Mj5FB4xK0cb1u&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/202) ########## streampark-common/src/main/java/org/apache/streampark/common/util/Utils.java: ########## @@ -0,0 +1,243 @@ +/* + * 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.streampark.common.util; + +import org.apache.commons.lang3.StringUtils; +import org.apache.streampark.shaded.org.slf4j.Logger; + +import java.io.BufferedInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.Flushable; +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URL; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.Collection; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.locks.LockSupport; +import java.util.jar.JarFile; +import java.util.jar.JarInputStream; +import java.util.jar.Manifest; + +/** General utility methods. */ +public final class Utils { + + private static final Logger LOG = + StreamParkLoggerFactory.loggerFactory().getLogger(Utils.class.getName()); + + private static final String OS = System.getProperty("os.name").toLowerCase(); + + private Utils() {} + + public static boolean isNotEmpty(Object elem) { + if (elem == null) { + return false; + } + if (elem instanceof Object[]) { + return ((Object[]) elem).length > 0; + } + if (elem instanceof CharSequence) { + return elem.toString().trim().length() > 0; + } + if (elem instanceof Collection) { + return !((Collection<?>) elem).isEmpty(); + } + if (elem instanceof Iterable) { + return ((Iterable<?>) elem).iterator().hasNext(); + } + if (elem instanceof Map) { + return !((Map<?, ?>) elem).isEmpty(); + } + return true; + } + + public static boolean isEmpty(Object elem) { + return !isNotEmpty(elem); + } + + public static String uuid() { + return UUID.randomUUID().toString().replaceAll("-", ""); + } + + public static void requireCheckJarFile(URL jar) throws IOException { + File jarFile; + try { + jarFile = new File(jar.toURI()); + } catch (Exception e) { + throw new IOException("JAR file path is invalid " + jar, e); + } + if (!jarFile.exists()) { Review Comment: ## SonarCloud / Accessing files should not lead to filesystem oracle attacks <!--SONAR_ISSUE_KEY:AZ89Cu1qj5FB4xK0cb1g-->Change this code to not construct the path from user-controlled data. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ89Cu1qj5FB4xK0cb1g&open=AZ89Cu1qj5FB4xK0cb1g&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/198) ########## streampark-flink/streampark-flink-connector/streampark-flink-connector-clickhouse/src/main/java/org/apache/streampark/flink/connector/clickhouse/internal/ClickHouseSinkFunction.java: ########## @@ -0,0 +1,102 @@ +/* + * 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.streampark.flink.connector.clickhouse.internal; + +import org.apache.streampark.common.util.JdbcUtils; +import org.apache.streampark.flink.connector.clickhouse.conf.ClickHouseJdbcConfig; +import org.apache.streampark.flink.connector.clickhouse.util.ClickhouseConvertUtils; +import org.apache.streampark.flink.connector.function.TransformFunction; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import ru.yandex.clickhouse.ClickHouseDataSource; +import ru.yandex.clickhouse.settings.ClickHouseProperties; +import java.lang.reflect.Field; +import java.sql.Connection; import java.sql.Statement; +import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Properties; import java.util.concurrent.atomic.AtomicLong; + +public class ClickHouseSinkFunction<T> extends RichSinkFunction<T> { + private static final Logger LOG = LoggerFactory.getLogger(ClickHouseSinkFunction.class); + private Connection connection; private Statement statement; + private final ClickHouseJdbcConfig clickHouseConf; + private final int batchSize; private final AtomicLong offset = new AtomicLong(0L); + private long timestamp = 0L; private final long flushInterval; + private final List<String> sqlValues = new ArrayList<>(); + private String insertSqlPrefixes; + private final TransformFunction<T, String> sqlFunc; + + public ClickHouseSinkFunction(Properties properties, TransformFunction<T, String> sqlFunc) { + this.clickHouseConf = new ClickHouseJdbcConfig(properties); + this.batchSize = clickHouseConf.batchSize; + this.flushInterval = clickHouseConf.flushInterval; + this.sqlFunc = sqlFunc; + } + + @Override public void open(Configuration parameters) throws Exception { + String user = clickHouseConf.user; String driver = clickHouseConf.driverClassName; + ClickHouseProperties chProps = new ClickHouseProperties(); + if (user != null && driver != null) { Class.forName(driver); chProps.setUser(user); } + else if (driver != null) Class.forName(driver); + else if (user != null) chProps.setUser(user); + for (Map.Entry<Object,Object> x : clickHouseConf.sinkOption.getInternalConfig().entrySet()) { + try { + Field field = chProps.getClass().getDeclaredField(x.getKey().toString()); + field.setAccessible(true); + String tn = field.getType().getSimpleName(); + if ("String".equals(tn)) field.set(chProps, x.getValue()); + else if ("int".equals(tn) || "Integer".equals(tn)) field.set(chProps, Integer.parseInt(x.getValue().toString())); + else if ("long".equals(tn) || "Long".equals(tn)) field.set(chProps, Long.parseLong(x.getValue().toString())); + else if ("boolean".equals(tn) || "Boolean".equals(tn)) field.set(chProps, Boolean.parseBoolean(x.getValue().toString())); + } catch (NoSuchFieldException e) { + LOG.warn("ClickHouseProperties config error, property:{} invalid", x.getKey()); + } + } + connection = new ClickHouseDataSource(clickHouseConf.jdbcUrl, chProps).getConnection(); + } + + @Override public void invoke(T value, SinkFunction.Context context) throws Exception { + String sql = sqlFunc != null ? sqlFunc.transform(value) : ClickhouseConvertUtils.convert(value); + if (batchSize == 1) { + connection.prepareStatement(sql).executeUpdate(); + } else { + sqlValues.add(sql); + long cnt = offset.incrementAndGet(); + long now = System.currentTimeMillis(); + if (cnt % batchSize == 0 || now - timestamp > flushInterval) execBatch(); + } + } + + @Override public void close() throws Exception { execBatch(); JdbcUtils.close(statement, connection); } + + private void execBatch() throws Exception { + if (offset.get() > 0) { + try { + LOG.info("ClickHouseSink batch {} insert begin..", offset.get()); + offset.set(0); + String valuesStr = String.join(",", sqlValues); + // Values are generated by the connector transform, not external user SQL input. + @SuppressWarnings("java:S2077") + String batchSql = insertSqlPrefixes + " " + valuesStr; + connection.prepareStatement(batchSql).executeUpdate(); Review Comment: ## SonarCloud / SQL queries should not be dynamically formatted <!--SONAR_ISSUE_KEY:AZ8_MOvhoL1ZW2-kElsG-->Make sure using a dynamically formatted SQL query is safe here. <p>See more on <a href="https://sonarcloud.io/project/issues?id=apache_incubator-streampark&issues=AZ8_MOvhoL1ZW2-kElsG&open=AZ8_MOvhoL1ZW2-kElsG&pullRequest=4419";>SonarQube Cloud</a></p> [Show more details](https://github.com/apache/streampark/security/code-scanning/211) -- 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]
