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.
core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
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 534e1197e1..5e940fba49 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 @@
 
 @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 List<String> getArgumentNames() {
         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

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 39e00e4e07..2d06c1bf39 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 @@
 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 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 @@ void readFile() throws IllegalVariableEvaluationException, IOException {
     }
 
     @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"));
+    }
 }
