Repository: kestra-io/kestra (java)

Issue:
The error "Unable to read a file that didn't belong to the current execution" is raised for the current execution
### Explain the bug

Reproducer

Subflow:

```yaml
id: subflow_raw_string_input
namespace: qa

inputs:
  - name: string_input
    type: STRING
    defaults: hey there

tasks:
  - id: for_each_item
    type: io.kestra.core.tasks.debugs.Return
    format: "{{ inputs.string_input }}"
```

Parent reading the file for the current batch using the `read()` function and passing it to the subflow as raw string input:
```yaml
id: each_parent
namespace: qa

tasks:
  - id: extract
    type: io.kestra.plugin.jdbc.duckdb.Query
    sql: |
      INSTALL httpfs;
      LOAD httpfs;
      SELECT *
      FROM read_csv_auto('https://raw.githubusercontent.com/kestra-io/datasets/main/csv/orders.csv', header=True);
    store: true

  - id: each_raw
    type: io.kestra.core.tasks.flows.ForEachItem
    items: "{{ outputs.extract.uri }}"
    namespace: qa
    flowId: subflow_raw_string_input
    wait: true
    transmitFailed: true
    batch: 
      rows: 10
    inputs:
      string_input: "{{ read(taskrun.items) }}"
```

![image](https://github.com/kestra-io/kestra/assets/86264395/7c5994ae-d9f4-4589-af43-d6b101a20037)



### Environment Information

- Kestra Version: 0.13.0

Pull request description:
Fix/read function parent execution
Fixes #2442

Also allow a subflow to read a file from the parent flow.

This allow to pass the URI of the internal storage and read the file in a subflow.

For example:

```yaml
id: each-item
namespace: dev


tasks:
  - id: extract
    type: io.kestra.plugin.gcp.bigquery.Query
    sql: |
      SELECT DATETIME(datehour) as date, title, views FROM `bigquery-public-data.wikipedia.pageviews_2023` 
      WHERE DATE(datehour) = current_date() and wiki = 'fr' and title not in ('Cookie_(informatique)', 'Wikipédia:Accueil_principal', 'Spécial:Recherche')
      ORDER BY datehour desc, views desc
      LIMIT 10
    store: true
  
  - id: each
    type: io.kestra.core.tasks.flows.ForEachItem
    items: "{{ outputs.extract.uri }}"
    batch: 
      rows: 4
    namespace: dev
    flowId: per-item
    wait: true
    transmitFailed: true
    inputs:
      items: "{{taskrun.items}}"
```


With a child flow:

```yaml
id: per-item
namespace: dev

inputs:
  - name: items
    type: STRING

tasks:
  - id: per-item
    type: io.kestra.core.tasks.log.Log
    message: "{{ inputs.items }}"
  - id: read
    type: io.kestra.core.tasks.log.Log
    message: "{{ read(inputs.items) }}"    
```

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 5e940fba4..28dcf9c2e 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
@@ -2,6 +2,7 @@ package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.tenant.TenantService;
+import io.kestra.core.utils.Slugify;
 import io.pebbletemplates.pebble.error.PebbleException;
 import io.pebbletemplates.pebble.extension.Function;
 import io.pebbletemplates.pebble.template.EvaluationContext;
@@ -54,17 +55,33 @@ public class ReadFileFunction implements Function {
     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);
+
+        // check if the file is from the current execution
+        if (!validateFileUri(flow.get("namespace"), flow.get("id"), execution.get("id"), path)) {
+            // if not, it can be from the parent execution, so we check if there is a trigger of type execution
+            if (context.getVariable("trigger") != null) {
+                // if there is a trigger of type execution, we also allow accessing a file from the parent execution
+                Map<String, String> trigger = (Map<String, String>) context.getVariable("trigger");
+                if (!validateFileUri(trigger.get("namespace"), trigger.get("flowId"), trigger.get("executionId"), path)) {
+                    throw new IllegalArgumentException("Unable to read the file '" + path + "' as it didn't belong to the current execution");
+                }
+            }
+            else {
+                throw new IllegalArgumentException("Unable to read the file '" + path + "' as it didn't belong to the current execution");
+            }
+        }
         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) {
+    private boolean 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");
+        if (namespace == null || flowId == null || executionId == null) {
+            return false;
         }
+        
+        String authorizedBasePath = KESTRA_SCHEME + Slugify.of(namespace) + "/" + Slugify.of(flowId) + "/executions/" + executionId + "/";
+        return path.startsWith(authorizedBasePath);
     }
 }
\ 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 2d06c1bf3..4e52da151 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
@@ -52,6 +52,8 @@ class ReadFileFunctionTest {
         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()));
+
+        // test for an authorized execution
         Map<String, Object> variables = Map.of(
             "flow", Map.of(
                 "id", flowId,
@@ -61,15 +63,33 @@ class ReadFileFunctionTest {
 
         String render = variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables);
         assertThat(render, is("Hello from a task output"));
+
+        // test for an authorized parent execution (execution trigger)
+        variables = Map.of(
+            "flow", Map.of(
+                "id", "subflow",
+                "namespace", namespace),
+            "execution", Map.of("id", IdUtils.create()),
+            "trigger", Map.of(
+                "flowId", flowId,
+                "namespace", namespace,
+                "executionId", executionId
+            )
+        );
+
+        render = variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables);
+        assertThat(render, is("Hello from a task output"));
     }
 
     @Test
-    void readUnauthorizedInternalStorageFile() throws IOException, IllegalVariableEvaluationException {
+    void readUnauthorizedInternalStorageFile() throws IOException {
         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()));
+
+        // test for an un-authorized execution with no trigger
         Map<String, Object> variables = Map.of(
             "flow", Map.of(
                 "id", "notme",
@@ -78,6 +98,37 @@ class ReadFileFunctionTest {
         );
 
         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"));
+        assertThat(exception.getMessage(), is("Unable to read the file '" + internalStorageFile + "' as it didn't belong to the current execution"));
+
+        // test for an un-authorized execution with a trigger of type execution
+        Map<String, Object> executionTriggerVariables = Map.of(
+            "flow", Map.of(
+                "id", "notme",
+                "namespace", "notme"),
+            "execution", Map.of("id", "notme"),
+            "trigger", Map.of(
+                "flowId", "notme",
+                "namespace", "notme",
+                "executionId", "notme"
+            )
+        );
+
+        exception = assertThrows(IllegalArgumentException.class, () -> variableRenderer.render("{{ read('" + internalStorageFile + "') }}", executionTriggerVariables));
+        assertThat(exception.getMessage(), is("Unable to read the file '" + internalStorageFile + "' as it didn't belong to the current execution"));
+
+        // test for an un-authorized execution with a trigger of another type
+        Map<String, Object> triggerVariables = Map.of(
+            "flow", Map.of(
+                "id", "notme",
+                "namespace", "notme"),
+            "execution", Map.of("id", "notme"),
+            "trigger", Map.of(
+                "date", "somedate",
+                "row", "somerow"
+            )
+        );
+
+        exception = assertThrows(IllegalArgumentException.class, () -> variableRenderer.render("{{ read('" + internalStorageFile + "') }}", triggerVariables));
+        assertThat(exception.getMessage(), is("Unable to read the file '" + internalStorageFile + "' as it didn't belong to the current execution"));
     }
 }