Repository: kestra-io/kestra (java)

Issue:
Provide a Pebble function to read an intenal storage file
### Feature description

Currently, there is a new `read(path)` Pebble function that allows reading a namespace file from its name.

We need this function to also be able to read any files from the internal storage, those files will start with the kestra scheme.
For security purpose, we need to only allow reading files from the current execution.

Pull request description:
feat(core): allow the Pebble read() function to read from the interna…
…l storage

Fixes #2426

## QA

This flow shows how to use it for inputs, launch it twice, and update the last task to use a previous URI to check that a file that exists is not accessible if it's not for the current execution.

It also works for tasks and trigger outputs as they share the same prefix as inputs so the check will validate them.

```yaml
id: read-internal-storage
namespace: dev

inputs:
  - name: file
    type: FILE

tasks:
  - id: log-uri
    type: io.kestra.core.tasks.log.Log
    message: "{{ inputs.file }}"

  - id: log-content
    type: io.kestra.core.tasks.log.Log
    message: "{{ read(inputs.file) }}"    

  # use the URI of a previous execution
  - id: log-unauthorized
    type: io.kestra.core.tasks.log.Log
    message: "{{ read('kestra:///dev/read-internal-storage/executions/7Twz4FXx0Rx2WHwdQWXG17/inputs/file/file_12387099138682102581.upl') }}"
```

Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
index 534e1197e..5e940fba4 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
@@ -17,7 +17,8 @@ import java.util.Map;
 
 @Singleton
 public class ReadFileFunction implements Function {
-    private static final String ERROR_MESSAGE = "The 'read' function expects an argument 'path' with structure {filePath}.";
+    private static final String ERROR_MESSAGE = "The 'read' function expects an argument 'path' that is a path to a namespace file or an internal storage URI.";
+    private static final String KESTRA_SCHEME = "kestra:///";
 
     @Inject
     private StorageInterface storageInterface;
@@ -29,24 +30,41 @@ public class ReadFileFunction implements Function {
         return List.of("path");
     }
 
-    @SuppressWarnings("unchecked")
     @Override
     public Object execute(Map<String, Object> args, PebbleTemplate self, EvaluationContext context, int lineNumber) {
-        Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
+        if (!args.containsKey("path")) {
+            throw new PebbleException(null, ERROR_MESSAGE, lineNumber, self.getName());
+        }
 
+        String path = (String) args.get("path");
         try {
-            return new String(storageInterface.get(tenantService.resolveTenant(), getStorageUri(flow.get("namespace"), args, self, lineNumber)).readAllBytes(), StandardCharsets.UTF_8);
-        } catch (IOException e) {
+            return path.startsWith(KESTRA_SCHEME) ? readFromInternalStorageUri(context, path) : readFromNamespaceFile(context, path);
+        }
+        catch (IOException e) {
             throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
         }
     }
 
-    protected URI getStorageUri(String namespace, Map<String, Object> args, PebbleTemplate self, int lineNumber) {
-        if (!args.containsKey("path")) {
-            throw new PebbleException(null, ERROR_MESSAGE, lineNumber, self.getName());
-        }
+    private String readFromNamespaceFile(EvaluationContext context, String path) throws IOException {
+        Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
+        URI namespaceFile = URI.create(storageInterface.namespaceFilePrefix(flow.get("namespace")) + "/" + path);
+        return new String(storageInterface.get(tenantService.resolveTenant(), namespaceFile).readAllBytes(), StandardCharsets.UTF_8);
+    }
 
-        String path = (String) args.get("path");
-        return URI.create(storageInterface.namespaceFilePrefix(namespace) + "/" + path);
+    private String readFromInternalStorageUri(EvaluationContext context, String path) throws IOException {
+        Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
+        Map<String, String> execution = (Map<String, String>) context.getVariable("execution");
+        validateFileUri(flow.get("namespace"), flow.get("id"), execution.get("id"), path);
+        URI internalStorageFile = URI.create(path);
+        return new String(storageInterface.get(tenantService.resolveTenant(), internalStorageFile).readAllBytes(), StandardCharsets.UTF_8);
+    }
+
+    private void validateFileUri(String namespace, String flowId, String executionId, String path) {
+        // Internal storage URI should be: kestra:///$namespace/$flowId/executions/$executionId/tasks/$taskName/$taskRunId/$random.ion or kestra:///$namespace/$flowId/executions/$executionId/trigger/$triggerName/$random.ion
+        // We check that the file is for the given flow execution
+        String authorizedBasePath = KESTRA_SCHEME + namespace + "/" + flowId + "/executions/" + executionId + "/";
+        if (!path.startsWith(authorizedBasePath)) {
+            throw new IllegalArgumentException("Unable to read a file that didn't belong to the current execution");
+        }
     }
 }
\ No newline at end of file


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java b/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
index 39e00e4e0..2d06c1bf3 100644
--- a/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
+++ b/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
@@ -3,6 +3,7 @@ package io.kestra.core.runners.pebble.functions;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.runners.VariableRenderer;
 import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.IdUtils;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Assertions;
@@ -16,6 +17,7 @@ import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @MicronautTest
 class ReadFileFunctionTest {
@@ -26,7 +28,7 @@ class ReadFileFunctionTest {
     StorageInterface storageInterface;
 
     @Test
-    void readFile() throws IllegalVariableEvaluationException, IOException {
+    void readNamespaceFile() throws IllegalVariableEvaluationException, IOException {
         String namespace = "io.kestra.tests";
         String filePath = "file.txt";
         storageInterface.createDirectory(null, URI.create(storageInterface.namespaceFilePrefix(namespace)));
@@ -37,8 +39,45 @@ class ReadFileFunctionTest {
     }
 
     @Test
-    void readUnknownFile() {
-        IllegalVariableEvaluationException illegalVariableEvaluationException = Assertions.assertThrows(IllegalVariableEvaluationException.class, () -> variableRenderer.render("{{ read('unknown.txt') }}", Map.of("flow", Map.of("namespace", "io.kestra.tests"))));
+    void readUnknownNamespaceFile() {
+        IllegalVariableEvaluationException illegalVariableEvaluationException = assertThrows(IllegalVariableEvaluationException.class, () -> variableRenderer.render("{{ read('unknown.txt') }}", Map.of("flow", Map.of("namespace", "io.kestra.tests"))));
         assertThat(illegalVariableEvaluationException.getCause().getCause().getClass(), is(FileNotFoundException.class));
     }
+
+    @Test
+    void readInternalStorageFile() throws IOException, IllegalVariableEvaluationException {
+        // task output URI format: 'kestra:///$namespace/$flowId/executions/$executionId/tasks/$taskName/$taskRunId/$random.ion'
+        String namespace = "namespace";
+        String flowId = "flow";
+        String executionId = IdUtils.create();
+        URI internalStorageURI = URI.create("/" + namespace + "/" + flowId + "/executions/" + executionId + "/tasks/task/" + IdUtils.create() + "/123456.ion");
+        URI internalStorageFile = storageInterface.put(null, internalStorageURI, new ByteArrayInputStream("Hello from a task output".getBytes()));
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", flowId,
+                "namespace", namespace),
+            "execution", Map.of("id", executionId)
+        );
+
+        String render = variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables);
+        assertThat(render, is("Hello from a task output"));
+    }
+
+    @Test
+    void readUnauthorizedInternalStorageFile() throws IOException, IllegalVariableEvaluationException {
+        String namespace = "namespace";
+        String flowId = "flow";
+        String executionId = IdUtils.create();
+        URI internalStorageURI = URI.create("/" + namespace + "/" + flowId + "/executions/" + executionId + "/tasks/task/" + IdUtils.create() + "/123456.ion");
+        URI internalStorageFile = storageInterface.put(null, internalStorageURI, new ByteArrayInputStream("Hello from a task output".getBytes()));
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", "notme",
+                "namespace", "notme"),
+            "execution", Map.of("id", "notme")
+        );
+
+        var exception = assertThrows(IllegalArgumentException.class, () -> variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables));
+        assertThat(exception.getMessage(), is("Unable to read a file that didn't belong to the current execution"));
+    }
 }