[improve] Validate user paths in Functions utils

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java

@@ -24,6 +24,7 @@
 import javax.servlet.ServletResponse;
 import javax.servlet.http.HttpServletResponse;
 import javax.ws.rs.core.Response;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.common.intercept.InterceptException;
 import org.apache.pulsar.common.policies.data.ErrorData;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -36,6 +37,7 @@
 /**
  *  Exception handler for handle exception.
  */
+@Slf4j
 public class ExceptionHandler {
 
     public void handle(ServletResponse response, Exception ex) throws IOException {

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java

@@ -853,14 +853,24 @@ public static void doCommonChecks(FunctionConfig functionConfig) {
         if (!isEmpty(functionConfig.getPy()) && !org.apache.pulsar.common.functions.Utils
                 .isFunctionPackageUrlSupported(functionConfig.getPy())
                 && functionConfig.getPy().startsWith(BUILTIN)) {
-            if (!new File(functionConfig.getPy()).exists()) {
+            String filename = functionConfig.getPy();
+            if (filename.contains("..")) {
+                throw new IllegalArgumentException("Invalid filename: " + filename);
+            }
+
+            if (!new File(filename).exists()) {
                 throw new IllegalArgumentException("The supplied python file does not exist");
             }
         }
         if (!isEmpty(functionConfig.getGo()) && !org.apache.pulsar.common.functions.Utils
                 .isFunctionPackageUrlSupported(functionConfig.getGo())
                 && functionConfig.getGo().startsWith(BUILTIN)) {
-            if (!new File(functionConfig.getGo()).exists()) {
+            String filename = functionConfig.getGo();
+            if (filename.contains("..")) {
+                throw new IllegalArgumentException("Invalid filename: " + filename);
+            }
+
+            if (!new File(filename).exists()) {
                 throw new IllegalArgumentException("The supplied go file does not exist");
             }
         }

pulsar-package-management/filesystem-storage/src/main/java/org/apache/pulsar/packages/management/storage/filesystem/FileSystemPackagesStorage.java

@@ -58,7 +58,11 @@ public class FileSystemPackagesStorage implements PackagesStorage {
         }
     }
 
-    private File getPath(String path) {
+    private File getPath(String path) throws IOException {
+        if (path.contains("..")) {
+            throw new IOException("Invalid path: " + path);
+        }
+
         File f = Paths.get(storagePath.toString(), path).toFile();
         if (!f.getParentFile().exists()) {
             if (!f.getParentFile().mkdirs()) {
@@ -119,28 +123,40 @@ public CompletableFuture<Void> readAsync(String path, OutputStream outputStream)
 
     @Override
     public CompletableFuture<Void> deleteAsync(String path) {
-        if (getPath(path).delete()) {
-            return CompletableFuture.completedFuture(null);
-        } else {
-            CompletableFuture<Void> f = new CompletableFuture<>();
-            f.completeExceptionally(new IOException("Failed to delete file at " + path));
-            return f;
+        try {
+            if (getPath(path).delete()) {
+                return CompletableFuture.completedFuture(null);
+            } else {
+                CompletableFuture<Void> f = new CompletableFuture<>();
+                f.completeExceptionally(new IOException("Failed to delete file at " + path));
+                return f;
+            }
+        } catch (IOException e) {
+            return CompletableFuture.failedFuture(e);
         }
     }
 
     @Override
     public CompletableFuture<List<String>> listAsync(String path) {
-        String[] files = getPath(path).list();
-        if (files == null) {
-            return CompletableFuture.completedFuture(Collections.emptyList());
-        } else {
-            return CompletableFuture.completedFuture(Arrays.asList(files));
+        try {
+            String[] files = getPath(path).list();
+            if (files == null) {
+                return CompletableFuture.completedFuture(Collections.emptyList());
+            } else {
+                return CompletableFuture.completedFuture(Arrays.asList(files));
+            }
+        } catch (IOException e) {
+            return CompletableFuture.failedFuture(e);
         }
     }
 
     @Override
     public CompletableFuture<Boolean> existAsync(String path) {
-        return CompletableFuture.completedFuture(getPath(path).exists());
+        try {
+            return CompletableFuture.completedFuture(getPath(path).exists());
+        } catch (IOException e) {
+            return CompletableFuture.failedFuture(e);
+        }
     }
 
     @Override