Merge branch 'main' into hotfix-influxdb_showcolumns_addstorage

.github/scripts/test/cli/test_py_register.sh

@@ -18,7 +18,7 @@ result=$(bash -c "echo '$COMMAND' | xargs -0 -t -i ${SCRIPT_COMMAND}")
 
 if [[ $result =~ 'success' ]]; then
   echo success
-  COMMAND='DROP PYTHON TASK "'"mock_udf"'";'
+  COMMAND='DROP FUNCTION "'"mock_udf"'";'
   bash -c "echo '$COMMAND' | xargs -0 -t -i ${SCRIPT_COMMAND}"
 else
   echo 'Error: failed to register udf mock_udf.'

.github/scripts/test/cli/test_py_register_macos.sh

@@ -16,7 +16,7 @@ result=$(sh -c "echo '$COMMAND' | xargs -0 -t -I F sh start_cli.sh -e 'F'")
 
 if [[ $result =~ 'success' ]]; then
   echo success
-  COMMAND='DROP PYTHON TASK "'"mock_udf"'";'
+  COMMAND='DROP FUNCTION "'"mock_udf"'";'
   sh -c "echo '$COMMAND' | xargs -0 -t -I F sh start_cli.sh -e 'F'"
 else
   echo 'Error: failed to register udf mock_udf.'

.github/scripts/test/cli/test_py_register_windows.sh

@@ -14,7 +14,7 @@ result=$(bash -c "./start_cli.bat -e '$COMMAND'")
 
 if [[ $result =~ 'success' ]]; then
   echo success
-  COMMAND='DROP PYTHON TASK "'"mock_udf"'";'
+  COMMAND='DROP FUNCTION "'"mock_udf"'";'
   bash -c "./start_cli.bat -e '$COMMAND'"
 else
   echo 'Error: failed to register udf mock_udf.'

antlr/src/main/antlr4/cn/edu/tsinghua/iginx/sql/Sql.g4

@@ -16,9 +16,9 @@ statement
    | SHOW REPLICA NUMBER # showReplicationStatement
    | ADD STORAGEENGINE storageEngineSpec # addStorageEngineStatement
    | SHOW CLUSTER INFO # showClusterInfoStatement
-   | SHOW REGISTER PYTHON TASK # showRegisterTaskStatement
+   | SHOW FUNCTIONS # showRegisterTaskStatement
    | CREATE FUNCTION udfType udfClassRef (COMMA (udfType)? udfClassRef)* IN filePath = stringLiteral # registerTaskStatement
-   | DROP PYTHON TASK name = stringLiteral # dropTaskStatement
+   | DROP FUNCTION name = stringLiteral # dropTaskStatement
    | COMMIT TRANSFORM JOB filePath = stringLiteral # commitTransformJobStatement
    | SHOW TRANSFORM JOB STATUS jobId = INT # showJobStatusStatement
    | CANCEL TRANSFORM JOB jobId = INT # cancelJobStatement
@@ -426,9 +426,9 @@ keyWords
    | DATA
    | REPLICA
    | DROP
-   | REGISTER
-   | PYTHON
-   | TASK
+   | CREATE
+   | FUNCTION
+   | FUNCTIONS
    | COMMIT
    | JOB
    | STATUS
@@ -675,18 +675,6 @@ DROP
    : D R O P
    ;
 
-REGISTER
-   : R E G I S T E R
-   ;
-
-PYTHON
-   : P Y T H O N
-   ;
-
-TASK
-   : T A S K
-   ;
-
 COMMIT
    : C O M M I T
    ;
@@ -759,6 +747,10 @@ FUNCTION
    : F U N C T I O N
    ;
 
+FUNCTIONS
+   : F U N C T I O N S
+   ;
+
 CREATED
    : C R E A T E D
    ;

client/src/main/java/cn/edu/tsinghua/iginx/client/IginxClient.java

@@ -653,7 +653,7 @@ private static Completer buildIginxCompleter() {
             Arrays.asList("explain", "select"),
             Arrays.asList("add", "storageengine"),
             Arrays.asList("create", "function"),
-            Arrays.asList("drop", "python", "task"),
+            Arrays.asList("drop", "function"),
             Arrays.asList("commit", "transform", "job"),
             Arrays.asList("show", "transform", "job", "status"),
             Arrays.asList("cancel", "transform", "job"),
@@ -671,7 +671,7 @@ private static Completer buildIginxCompleter() {
             Arrays.asList("clear", "data"),
             Arrays.asList("show", "columns"),
             Arrays.asList("show", "cluster", "info"),
-            Arrays.asList("show", "register", "python", "task"),
+            Arrays.asList("show", "functions"),
             Arrays.asList("show", "sessionid"),
             Arrays.asList("show", "rules"),
             Arrays.asList("remove", "historydatasource"));

core/src/main/java/cn/edu/tsinghua/iginx/IginxWorker.java

@@ -837,16 +837,19 @@ public Status registerTask(RegisterTaskReq req) {
 
     Predicate<String> ruleNameFilter = FilePermissionRuleNameFilters.transformerRulesWithDefault();
 
-    Predicate<Path> sourceChecker =
+    FilePermissionManager.Checker sourceChecker =
         FilePermissionManager.getInstance()
             .getChecker(null, ruleNameFilter, FileAccessType.EXECUTE);
 
-    File sourceFile = new File(filePath);
-    if (!sourceChecker.test(sourceFile.toPath())) {
+    Optional<Path> sourceCheckedPath = sourceChecker.normalize(filePath);
+
+    if (!sourceCheckedPath.isPresent()) {
       errorMsg = String.format("Register file %s has no execute permission", filePath);
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }
+
+    File sourceFile = sourceCheckedPath.get().toFile();
     if (!sourceFile.exists()) {
       errorMsg = String.format("Register file not exist in declared path, path=%s", filePath);
       LOGGER.error(errorMsg);
@@ -880,7 +883,7 @@ public Status registerTask(RegisterTaskReq req) {
     for (UDFClassPair p : pairs) {
       TransformTaskMeta transformTaskMeta = metaManager.getTransformTask(p.name.trim());
       if (transformTaskMeta != null && transformTaskMeta.getIpSet().contains(config.getIp())) {
-        errorMsg = String.format("Register task %s already exist", transformTaskMeta);
+        errorMsg = String.format("Function %s already exist", transformTaskMeta);
         LOGGER.error(errorMsg);
         return RpcUtils.FAILURE.setMessage(errorMsg);
       }
@@ -890,19 +893,21 @@ public Status registerTask(RegisterTaskReq req) {
     String fileName = sourceFile.getName();
     String destPath =
         String.join(File.separator, config.getDefaultUDFDir(), "python_scripts", fileName);
-    File destFile = new File(destPath);
 
-    if (destFile.exists()) {
-      errorMsg = String.format("Register file(s) already exist, name=%s", fileName);
+    FilePermissionManager.Checker destChecker =
+        FilePermissionManager.getInstance().getChecker(null, ruleNameFilter, FileAccessType.WRITE);
+
+    Optional<Path> destCheckedPath = destChecker.normalize(destPath);
+    if (!destCheckedPath.isPresent()) {
+      errorMsg = String.format("Register file %s has no write permission", destPath);
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }
 
-    Predicate<Path> destChecker =
-        FilePermissionManager.getInstance().getChecker(null, ruleNameFilter, FileAccessType.WRITE);
+    File destFile = destCheckedPath.get().toFile();
 
-    if (!destChecker.test(destFile.toPath())) {
-      errorMsg = String.format("Register file %s has no write permission", destPath);
+    if (destFile.exists()) {
+      errorMsg = String.format("Register file(s) already exist, name=%s", fileName);
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }
@@ -960,20 +965,20 @@ public Status dropTask(DropTaskReq req) {
     TransformTaskMeta transformTaskMeta = metaManager.getTransformTask(name);
     String errorMsg = "";
     if (transformTaskMeta == null) {
-      errorMsg = "Register task not exist";
+      errorMsg = "Function does not exist";
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }
 
     TransformJobManager manager = TransformJobManager.getInstance();
     if (manager.isRegisterTaskRunning(name)) {
-      errorMsg = "Register task is running";
+      errorMsg = "Function is running";
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }
 
     if (!transformTaskMeta.getIpSet().contains(config.getIp())) {
-      errorMsg = String.format("Register task exists in node: %s", config.getIp());
+      errorMsg = String.format("Function exists in node: %s", config.getIp());
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }
@@ -984,25 +989,25 @@ public Status dropTask(DropTaskReq req) {
             + "python_scripts"
             + File.separator
             + transformTaskMeta.getFileName();
-    File file = new File(filePath);
-
-    if (!file.exists()) {
-      metaManager.dropTransformTask(name);
-      errorMsg = String.format("Register file not exist, path=%s", filePath);
-      LOGGER.error(errorMsg);
-      return RpcUtils.FAILURE.setMessage(errorMsg);
-    }
 
-    String pythonDir = config.getDefaultUDFDir() + File.separator + "python_scripts";
     Predicate<String> ruleNameFilter = FilePermissionRuleNameFilters.transformerRulesWithDefault();
-    Predicate<Path> destChecker =
+    FilePermissionManager.Checker destChecker =
         FilePermissionManager.getInstance().getChecker(null, ruleNameFilter, FileAccessType.WRITE);
+    Optional<Path> normalizedFile = destChecker.normalize(filePath);
 
-    Path pythonDirPath = Paths.get(pythonDir);
-    if (!destChecker.test(pythonDirPath)) {
+    if (!normalizedFile.isPresent()) {
       errorMsg =
           String.format(
-              "User has no write permission in target directory, udf %s cannot be dropped.", name);
+              "User has no write permission in target directory, task %s cannot be dropped.", name);
+      LOGGER.error(errorMsg);
+      return RpcUtils.FAILURE.setMessage(errorMsg);
+    }
+
+    File file = normalizedFile.get().toFile();
+
+    if (!file.exists()) {
+      metaManager.dropTransformTask(name);
+      errorMsg = String.format("Register file not exist, path=%s", filePath);
       LOGGER.error(errorMsg);
       return RpcUtils.FAILURE.setMessage(errorMsg);
     }

core/src/main/java/cn/edu/tsinghua/iginx/auth/FilePermissionManager.java

@@ -4,6 +4,7 @@
 import cn.edu.tsinghua.iginx.conf.FilePermissionConfig;
 import cn.edu.tsinghua.iginx.conf.entity.FilePermissionDescriptor;
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
@@ -32,7 +33,57 @@ public static FilePermissionManager getInstance() {
     filePermissionConfig.onReload(this::reload);
   }
 
-  public Predicate<Path> getChecker(
+  public interface Checker {
+    boolean test(Path path);
+
+    default Optional<Path> normalize(String path) {
+      Optional<Path> p = cheatNormalize(Paths.get(path));
+      if (!p.isPresent()) {
+        return Optional.empty();
+      }
+      Path cheated = p.get();
+      if (!test(cheated)) {
+        return Optional.empty();
+      }
+      return Optional.of(cheated);
+    }
+
+    default boolean testRelativePath(String path) {
+      return !path.contains("..");
+    }
+
+    /**
+     * cheat CodeQL to not complain about path traversal vulnerability. This method should not be
+     * used except you know what you are doing.
+     *
+     * @param path the path to normalize
+     * @return the normalized path if it is safe, otherwise empty
+     */
+    @Deprecated
+    default Optional<Path> cheatNormalize(Path path) {
+      Path p = path.toAbsolutePath();
+      // split path nodes
+      String root = p.getRoot().toString();
+      String[] nodes = new String[p.getNameCount()];
+      for (int i = 0; i < p.getNameCount(); i++) {
+        nodes[i] = p.getName(i).toString();
+      }
+      // check if any node contains "."
+      if (!testRelativePath(root)) {
+        return Optional.empty();
+      }
+      for (String node : nodes) {
+        if (!testRelativePath(node)) {
+          return Optional.empty();
+        }
+      }
+      // rebuild path
+      Path rebuiltPath = Paths.get(root, nodes);
+      return Optional.of(rebuiltPath);
+    }
+  }
+
+  public Checker getChecker(
       @Nullable String user, Predicate<String> ruleNameFilter, FileAccessType... type) {
     return path -> {
       UserRules userRules = rules.get();

core/src/main/java/cn/edu/tsinghua/iginx/auth/utils/FilePermissionRuleNameFilters.java

@@ -23,4 +23,12 @@ public static Predicate<String> udfRules() {
   public static Predicate<String> udfRulesWithDefault() {
     return udfRules().or(defaultRules());
   }
+
+  public static Predicate<String> filesystemRules() {
+    return ruleName -> ruleName.startsWith("filesystem");
+  }
+
+  public static Predicate<String> filesystemRulesWithDefault() {
+    return filesystemRules().or(defaultRules());
+  }
 }

core/src/main/java/cn/edu/tsinghua/iginx/transform/data/FileAppendWriter.java

@@ -24,13 +24,26 @@ public class FileAppendWriter extends ExportWriter {
 
   private boolean hasWriteHeader;
 
-  public FileAppendWriter(String fileName) {
-    this.fileName = fileName;
+  public FileAppendWriter(String name) {
+    this.fileName = normalizeFileName(name).toString();
     this.hasWriteHeader = false;
     File file = new File(fileName);
     createFileIfNotExist(file);
   }
 
+  private Path normalizeFileName(String fileName) {
+    Predicate<String> ruleNameFilter = FilePermissionRuleNameFilters.transformerRulesWithDefault();
+
+    FilePermissionManager.Checker checker =
+        FilePermissionManager.getInstance().getChecker(null, ruleNameFilter, FileAccessType.WRITE);
+
+    return checker
+        .normalize(fileName)
+        .orElseThrow(
+            () ->
+                new SecurityException("transformer has no permission to write file: " + fileName));
+  }
+
   @Override
   public void write(BatchData batchData) {
     if (!hasWriteHeader) {
@@ -50,20 +63,13 @@ public void write(BatchData batchData) {
   }
 
   private void createFileIfNotExist(File file) {
-    Predicate<String> ruleNameFilter = FilePermissionRuleNameFilters.transformerRulesWithDefault();
-
-    Predicate<Path> pathChecker =
-        FilePermissionManager.getInstance().getChecker(null, ruleNameFilter, FileAccessType.WRITE);
-    if (!pathChecker.test(file.toPath())) {
-      throw new SecurityException(
-          ("transformer has no permission to write file: " + file.getAbsolutePath()));
-    }
     if (!file.exists()) {
       LOGGER.info("File not exists, create it...");
       // get and create parent dir
-      if (!file.getParentFile().exists()) {
-        System.out.println("Parent dir not exists, create it...");
-        file.getParentFile().mkdirs();
+      File normalizeParentFile = normalizeFileName(file.getParentFile().getPath()).toFile();
+      if (!normalizeParentFile.exists()) {
+        LOGGER.info("Parent dir not exists, create it...");
+        normalizeParentFile.mkdirs();
       }
       try {
         // create file

core/src/main/resources/file-permission.properties

@@ -75,10 +75,11 @@
 #  The rule name is used to distinguish different rules. Different modules
 #  determine the subset of rules used by filter. In details:
 #
-#    * the udf module uses rules starting with "udf"
-#    * the transform module uses rules starting with "transformer".
 #    * the default module uses rules starting with "default".
 #      + the rules of the default module is applied to all modules.
+#    * the udf module uses rules starting with "udf"
+#    * the transform module uses rules starting with "transformer".
+#    * the filesystem driver uses rules starting with "filesystem".
 #
 # Rule Application Order:
 #   1. For the same user, the rules are applied in the order of the configuration