Repository: kestra-io/kestra (java)

Issue:
Storing into Kestra Internal Storage does not take into account illegal characters
### Describe the issue

I'm trying to download a file into Kestra's internal storage using the ```io.kestra.plugin.core.http.Download``` task, but this task fails because the filename of the downloaded file has illegal characters for the URI. Here's the error log for the download task : 

```Illegal character in path at index 122: kestra:///plm/complm-cm-dd-005-worker/executions/1QE1b2wn7AS3yoJTtqKKmm/tasks/centric-downloadfile/1E8tP0Znk8oTzN33AEVRYv/[TEST용]-Adobe-Artboard-for-APP_Sarah-23.png```

I have no control over the endpoint and the endpoint itself does not support renaming the file in the request. Kestra fails to add the file to the internal storage because ```[```,```]``` are illegal characters for URI. 

I suspect this issue will occur for any download of files or any time a file is stored in internal storage with the filename set by external factors, not just exclusive to HTTP Download. 

My proposed solution would be to automatically url encode the filename any time Kestra stores something in internal storage, and automatically decode it during retrieval. 


### Environment

- Kestra Version: 0.19.4
- Operating System (OS/Docker/Kubernetes): Docker


Pull request description:
fix(core): encode filename
close #5589

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/Label.java b/core/src/main/java/io/kestra/core/models/Label.java
index 587970308..bd2fbd835 100644
--- a/core/src/main/java/io/kestra/core/models/Label.java
+++ b/core/src/main/java/io/kestra/core/models/Label.java
@@ -2,4 +2,11 @@ package io.kestra.core.models;
 
 import jakarta.validation.constraints.NotNull;
 
-public record Label(@NotNull String key, @NotNull String value) {}
+public record Label(@NotNull String key, @NotNull String value) {
+    public static final String SYSTEM_PREFIX = "system_";
+
+    // system labels
+    public static final String CORRELATION_ID = SYSTEM_PREFIX + "correlationId";
+    public static final String USERNAME = SYSTEM_PREFIX + "username";
+    public static final String APP = SYSTEM_PREFIX + "app";
+}
diff --git a/core/src/main/java/io/kestra/core/models/executions/Execution.java b/core/src/main/java/io/kestra/core/models/executions/Execution.java
index 890c62976..89407f593 100644
--- a/core/src/main/java/io/kestra/core/models/executions/Execution.java
+++ b/core/src/main/java/io/kestra/core/models/executions/Execution.java
@@ -39,6 +39,8 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.zip.CRC32;
 
+import static io.kestra.core.models.Label.SYSTEM_PREFIX;
+
 @Builder(toBuilder = true)
 @Slf4j
 @Getter
@@ -76,7 +78,6 @@ public class Execution implements DeletedInterface, TenantInterface {
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     Map<String, Object> outputs;
 
-    @With
     @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
     @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
     List<Label> labels;
@@ -180,6 +181,12 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.prebuild();
             return super.build();
         }
+
+        @Override
+        public ExecutionBuilder labels(List<Label> labels) {
+            checkForSystemLabels(labels);
+            return super.labels(labels);
+        }
     }
 
     public Execution withState(State.Type state) {
@@ -205,6 +212,75 @@ public class Execution implements DeletedInterface, TenantInterface {
         );
     }
 
+    /**
+     * This method replaces labels with new ones.
+     * It refuses system labels as they must be passed via the withSystemLabels method.
+     */
+    public Execution withLabels(List<Label> labels) {
+        checkForSystemLabels(labels);
+
+        return new Execution(
+            this.tenantId,
+            this.id,
+            this.namespace,
+            this.flowId,
+            this.flowRevision,
+            this.taskRunList,
+            this.inputs,
+            this.outputs,
+            labels,
+            this.variables,
+            this.state,
+            this.parentId,
+            this.originalId,
+            this.trigger,
+            this.deleted,
+            this.metadata,
+            this.scheduleDate,
+            this.error
+        );
+    }
+
+    private static void checkForSystemLabels(List<Label> labels) {
+        if (labels != null) {
+            Optional<Label> first = labels.stream().filter(label -> label.key() != null && label.key().startsWith(SYSTEM_PREFIX)).findFirst();
+            if (first.isPresent()) {
+                throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().key() + "=" + first.get().value());
+            }
+        }
+    }
+
+    /**
+     * This method in <b>only to be used</b> to add system labels to an execution.
+     * It will not replace exisiting labels but add new one (possibly duplicating).
+     */
+    public Execution withSystemLabels(List<Label> labels) {
+        List<Label> newLabels = this.labels == null ? new ArrayList<>() : this.labels;
+        if (labels != null) {
+            newLabels.addAll(labels);
+        }
+        return new Execution(
+            this.tenantId,
+            this.id,
+            this.namespace,
+            this.flowId,
+            this.flowRevision,
+            this.taskRunList,
+            this.inputs,
+            this.outputs,
+            newLabels,
+            this.variables,
+            this.state,
+            this.parentId,
+            this.originalId,
+            this.trigger,
+            this.deleted,
+            this.metadata,
+            this.scheduleDate,
+            this.error
+        );
+    }
+
     public Execution withTaskRun(TaskRun taskRun) throws InternalException {
         ArrayList<TaskRun> newTaskRunList = new ArrayList<>(this.taskRunList);
 
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
index 61acf59fa..af5aaac76 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
@@ -13,10 +13,12 @@ import io.kestra.core.models.tasks.ExecutableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.storages.Storage;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.stream.Streams;
 
 import java.time.Instant;
 import java.time.ZonedDateTime;
 import java.util.*;
+import java.util.stream.Collectors;
 
 @Slf4j
 public final class ExecutableUtils {
@@ -91,6 +93,14 @@ public final class ExecutableUtils {
             "flowRevision", currentFlow.getRevision()
         );
 
+        // propagate system labels and compute correlation ID if not already existing
+        List<Label> systemLabels = Streams.of(currentExecution.getLabels())
+            .filter(label -> label.key().startsWith(Label.SYSTEM_PREFIX))
+            .collect(Collectors.toList());
+        if (systemLabels.stream().noneMatch(label -> label.key().equals(Label.CORRELATION_ID))) {
+            systemLabels.add(new Label(Label.CORRELATION_ID, currentExecution.getId()));
+        }
+
         FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class);
         Instant scheduleOnDate = scheduleDate != null ? scheduleDate.as(runContext, ZonedDateTime.class).toInstant() : null;
         Execution execution = Execution
@@ -105,7 +115,8 @@ public final class ExecutableUtils {
                 .variables(variables)
                 .build()
             )
-            .withScheduleDate(scheduleOnDate);
+            .withScheduleDate(scheduleOnDate)
+            .withSystemLabels(systemLabels);
         return SubflowExecution.builder()
             .parentTask(currentTask)
             .parentTaskRun(currentTaskRun.withState(State.Type.RUNNING))
diff --git a/core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java b/core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java
index 3dbfb6394..c6fd1964a 100644
--- a/core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java
+++ b/core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java
@@ -24,6 +24,8 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
+import static io.kestra.core.models.Label.SYSTEM_PREFIX;
+
 @Singleton
 @Introspected
 public class FlowValidator implements ConstraintValidator<FlowValidation, Flow> {
@@ -57,32 +59,29 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
         }
 
         value.allTasksWithChilds()
-            .forEach(
-                task -> {
-                    if (task instanceof ExecutableTask<?> executableTask
-                        && value.getId().equals(executableTask.subflowId().flowId())
-                        && value.getNamespace().equals(executableTask.subflowId().namespace())) {
-                        violations.add("Recursive call to flow [" + value.getNamespace() + "." + value.getId() + "]");
-                    }
-                }
-            );
+            .stream().filter(task -> task instanceof ExecutableTask<?> executableTask
+                && value.getId().equals(executableTask.subflowId().flowId())
+                && value.getNamespace().equals(executableTask.subflowId().namespace()))
+            .forEach(task -> violations.add("Recursive call to flow [" + value.getNamespace() + "." + value.getId() + "]"));
 
         // input unique name
-        if (value.getInputs() != null) {
-            List<String> duplicates = getDuplicates(value.getInputs().stream().map(Data::getId).toList());
-            if (!duplicates.isEmpty()) {
-                violations.add("Duplicate input with name [" + String.join(", ", duplicates) + "]");
-            }
-            checkFlowInputsDependencyGraph(value, violations);
+        duplicateIds = getDuplicates(ListUtils.emptyOnNull(value.getInputs()).stream().map(Data::getId).toList());
+        if (!duplicateIds.isEmpty()) {
+            violations.add("Duplicate input with name [" + String.join(", ", duplicateIds) + "]");
         }
+        checkFlowInputsDependencyGraph(value, violations);
+
         // output unique name
-        if (value.getOutputs() != null) {
-            List<String> duplicates = getDuplicates(value.getOutputs().stream().map(Data::getId).toList());
-            if (!duplicates.isEmpty()) {
-                violations.add("Duplicate output with name [" + String.join(", ", duplicates) + "]");
-            }
+        duplicateIds = getDuplicates(ListUtils.emptyOnNull(value.getOutputs()).stream().map(Data::getId).toList());
+        if (!duplicateIds.isEmpty()) {
+            violations.add("Duplicate output with name [" + String.join(", ", duplicateIds) + "]");
         }
 
+        // system labels
+        ListUtils.emptyOnNull(value.getLabels()).stream()
+            .filter(label -> label.key() != null && label.key().startsWith(SYSTEM_PREFIX))
+            .forEach(label -> violations.add("System labels can only be set by Kestra itself, offending label: " + label.key() + "=" + label.value()));
+
         if (!violations.isEmpty()) {
             context.disableDefaultConstraintViolation();
             context.buildConstraintViolationWithTemplate("Invalid Flow: " + String.join(", ", violations))
diff --git a/core/src/main/java/io/kestra/plugin/core/execution/Labels.java b/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
index 9b0ffe6a3..aa0b1d50e 100644
--- a/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
+++ b/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
@@ -131,14 +131,12 @@ public class Labels extends Task implements ExecutionUpdatableTask {
                 );
             }));
 
-            return execution.toBuilder()
-                .labels(newLabels.entrySet().stream()
+            return execution.withLabels(newLabels.entrySet().stream()
                     .map(throwFunction(entry -> new Label(
                         entry.getKey(),
                         entry.getValue()
                     )))
-                    .toList())
-                .build();
+                    .toList());
         } else {
             throw new IllegalVariableEvaluationException("Unknown value type: " + labels.getClass());
         }
diff --git a/core/src/main/java/io/kestra/plugin/core/http/Download.java b/core/src/main/java/io/kestra/plugin/core/http/Download.java
index 308efa999..6c1495101 100644
--- a/core/src/main/java/io/kestra/plugin/core/http/Download.java
+++ b/core/src/main/java/io/kestra/plugin/core/http/Download.java
@@ -21,6 +21,8 @@ import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.net.URI;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 
@@ -128,6 +130,9 @@ public class Download extends AbstractHttp implements RunnableTask<Download.Outp
                 String contentDisposition = builder.headers.get("Content-Disposition").getFirst();
                 filename = filenameFromHeader(runContext, contentDisposition);
             }
+            if (filename != null) {
+                filename = URLEncoder.encode(filename, StandardCharsets.UTF_8);
+            }
 
             builder.uri(runContext.storage().putFile(tempFile, filename));
 
@@ -145,7 +150,7 @@ public class Download extends AbstractHttp implements RunnableTask<Download.Outp
             String filename = null;
             for (String part : parts) {
                 if (part.startsWith("filename")) {
-                    filename = part.substring(part.lastIndexOf('=') + 2, part.length() - 1);
+                    filename = part.substring(part.lastIndexOf('=') + 1);
                 }
                 if (part.startsWith("filename*")) {
                     // following https://datatracker.ietf.org/doc/html/rfc5987 the filename* should be <ENCODING>'(lang)'<filename>
diff --git a/storage-local/src/main/java/io/kestra/storage/local/LocalStorage.java b/storage-local/src/main/java/io/kestra/storage/local/LocalStorage.java
index 9669338c4..90946dd59 100644
--- a/storage-local/src/main/java/io/kestra/storage/local/LocalStorage.java
+++ b/storage-local/src/main/java/io/kestra/storage/local/LocalStorage.java
@@ -161,7 +161,7 @@ public class LocalStorage implements StorageInterface {
             }
         }
 
-        return URI.create("kestra://" + uri.getPath());
+        return URI.create("kestra://" + uri.getRawPath());
     }
 
     @Override
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 369a7ca68..8499affaa 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -89,6 +89,7 @@ import lombok.experimental.SuperBuilder;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.io.FilenameUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
@@ -582,7 +583,8 @@ public class ExecutionController {
         @Parameter(description = "The flow revision or latest if null") @QueryValue Optional<Integer> revision
     ) {
         Flow flow = flowService.getFlowIfExecutableOrThrow(tenantService.resolveTenant(), namespace, id, revision);
-        Execution execution = Execution.newExecution(flow, parseLabels(labels));
+        Pair<List<Label>, List<Label>> parsedLabels = parseLabels(labels);
+        Execution execution = Execution.newExecution(flow, parsedLabels.getLeft()).withSystemLabels(parsedLabels.getRight());
         return flowInputOutput
             .validateExecutionInputs(flow.getInputs(), execution, inputs)
             .map(values -> ApiValidateExecutionInputsResponse.of(id, namespace, values));
@@ -603,7 +605,8 @@ public class ExecutionController {
         @Parameter(description = "Schedule the flow on a specific date") @QueryValue Optional<ZonedDateTime> scheduleDate
     ) throws IOException {
         Flow flow = flowService.getFlowIfExecutableOrThrow(tenantService.resolveTenant(), namespace, id, revision);
-        Execution current = Execution.newExecution(flow, null, parseLabels(labels), scheduleDate);
+        Pair<List<Label>, List<Label>> parsedLabels = parseLabels(labels);
+        Execution current = Execution.newExecution(flow, null, parsedLabels.getLeft(), scheduleDate).withSystemLabels(parsedLabels.getRight());
         Mono<CompletableFuture<ExecutionResponse>> handle = flowInputOutput.readExecutionInputs(flow, current, inputs)
             .handle((executionInputs, sink) -> {
                 Execution executionWithInputs = current.withInputs(executionInputs);
@@ -682,10 +685,14 @@ public class ExecutionController {
         }
     }
 
-    private List<Label> parseLabels(List<String> labels) {
-        return labels == null ? null : RequestUtils.toMap(labels).entrySet().stream()
+    protected Pair<List<Label>, List<Label>> parseLabels(List<String> labels) {
+        // We allow passing the correlation id from the API but only this one, other system labels will go throught and fail at execution creation.
+        List<Label> parsedLabels =  labels == null ? Collections.emptyList() : RequestUtils.toMap(labels).entrySet().stream()
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .toList();
+        List<Label> standardLabels = parsedLabels.stream().filter(label -> !label.key().equals(Label.CORRELATION_ID)).toList();
+        List<Label> systemLabels = parsedLabels.stream().filter(label -> label.key().equals(Label.CORRELATION_ID)).toList();
+        return Pair.of(standardLabels, systemLabels);
     }
 
     protected <T> HttpResponse<T> validateFile(String executionId, URI path, String redirect) {
@@ -1705,9 +1712,7 @@ public class ExecutionController {
         }
 
         Execution newExecution = execution
-            .toBuilder()
-            .labels(newLabels.entrySet().stream().map(entry -> new Label(entry.getKey(), entry.getValue())).filter(label -> !label.key().isEmpty() || !label.value().isEmpty()).toList())
-            .build();
+            .withLabels(newLabels.entrySet().stream().map(entry -> new Label(entry.getKey(), entry.getValue())).filter(label -> !label.key().isEmpty() || !label.value().isEmpty()).toList());
         eventPublisher.publishEvent(new CrudEvent<>(newExecution, execution, CrudEventType.UPDATE));
 
         return executionRepository.save(newExecution);


Test patch:
diff --git a/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java b/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
index 41b6be17b..316898cfd 100644
--- a/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
+++ b/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
@@ -22,7 +22,7 @@ class FlowValidationTest {
     @Inject
     private ModelValidator modelValidator;
     @Inject
-    YamlFlowParser yamlFlowParser = new YamlFlowParser();
+    private YamlFlowParser yamlFlowParser = new YamlFlowParser();
 
     @Test
     void invalidRecursiveFlow() {
@@ -33,6 +33,16 @@ class FlowValidationTest {
         assertThat(validate.get().getMessage(), containsString(": Invalid Flow: Recursive call to flow [io.kestra.tests.recursive-flow]"));
     }
 
+    @Test
+    void systemLabelShouldFailValidation() {
+        Flow flow = this.parse("flows/invalids/system-labels.yaml");
+        Optional<ConstraintViolationException> validate = modelValidator.isValid(flow);
+
+        assertThat(validate.isPresent(), is(true));
+        assertThat(validate.get().getMessage(), containsString("System labels can only be set by Kestra itself, offending label: system_label=system_key"));
+        assertThat(validate.get().getMessage(), containsString("System labels can only be set by Kestra itself, offending label: system_id=id"));
+    }
+
     private Flow parse(String path) {
         URL resource = TestsUtils.class.getClassLoader().getResource(path);
         assert resource != null;
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/CorrelationIdTest.java b/core/src/test/java/io/kestra/plugin/core/flow/CorrelationIdTest.java
new file mode 100644
index 000000000..243296b61
--- /dev/null
+++ b/core/src/test/java/io/kestra/plugin/core/flow/CorrelationIdTest.java
@@ -0,0 +1,68 @@
+package io.kestra.plugin.core.flow;
+
+import io.kestra.core.models.Label;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.queues.QueueException;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import io.kestra.core.utils.TestsUtils;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CorrelationIdTest extends AbstractMemoryRunnerTest {
+    @Inject
+    @Named(QueueFactoryInterface.EXECUTION_NAMED)
+    private QueueInterface<Execution> executionQueue;
+
+    @Test
+    void shouldHaveCorrelationId() throws QueueException, TimeoutException, InterruptedException {
+        CountDownLatch countDownLatch = new CountDownLatch(2);
+        AtomicReference<Execution> child = new AtomicReference<>();
+        AtomicReference<Execution> grandChild = new AtomicReference<>();
+
+        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
+            Execution execution = either.getLeft();
+            if (execution.getFlowId().equals("subflow-child") && execution.getState().getCurrent().isTerminated()) {
+                countDownLatch.countDown();
+                child.set(execution);
+            }
+            if (execution.getFlowId().equals("subflow-grand-child") && execution.getState().getCurrent().isTerminated()) {
+                countDownLatch.countDown();
+                grandChild.set(execution);
+            }
+        });
+
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "subflow-parent");
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+
+        assertTrue(countDownLatch.await(1, TimeUnit.MINUTES));
+        receive.blockLast();
+
+        assertThat(child.get(), notNullValue());
+        assertThat(child.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        Optional<Label> correlationId = child.get().getLabels().stream().filter(label -> label.key().equals(Label.CORRELATION_ID)).findAny();
+        assertThat(correlationId.isPresent(), is(true));
+        assertThat(correlationId.get().value(), is(execution.getId()));
+
+        assertThat(grandChild.get(), notNullValue());
+        assertThat(grandChild.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        correlationId = grandChild.get().getLabels().stream().filter(label -> label.key().equals(Label.CORRELATION_ID)).findAny();
+        assertThat(correlationId.isPresent(), is(true));
+        assertThat(correlationId.get().value(), is(execution.getId()));
+    }
+}
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/FlowCaseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/FlowCaseTest.java
index 6bd6a1b75..97be642d1 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/FlowCaseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/FlowCaseTest.java
@@ -100,14 +100,16 @@ public class FlowCaseTest {
 
         if (testInherited) {
             assertThat(triggered.get().getLabels(), hasItems(
+                new Label(Label.CORRELATION_ID, execution.getId()),
                 new Label("mainFlowExecutionLabel", "execFoo"),
                 new Label("mainFlowLabel", "flowFoo"),
                 new Label("launchTaskLabel", "launchFoo"),
                 new Label("switchFlowLabel", "switchFoo")
             ));
         } else {
-            assertThat(triggered.get().getLabels().size(), is(2));
+            assertThat(triggered.get().getLabels().size(), is(3));
             assertThat(triggered.get().getLabels(), hasItems(
+                new Label(Label.CORRELATION_ID, execution.getId()),
                 new Label("launchTaskLabel", "launchFoo"),
                 new Label("switchFlowLabel", "switchFoo")
             ));
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
index 6aa6a9dc3..8990f6095 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
@@ -1,5 +1,6 @@
 package io.kestra.plugin.core.flow;
 
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.queues.QueueException;
@@ -28,6 +29,7 @@ import java.nio.file.Files;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -101,6 +103,9 @@ public class ForEachItemCaseTest {
         assertThat(triggered.get().getFlowId(), is("for-each-item-subflow"));
         assertThat((String) triggered.get().getInputs().get("items"), matchesRegex("kestra:///io/kestra/tests/for-each-item/executions/.*/tasks/each-split/.*\\.txt"));
         assertThat(triggered.get().getTaskRunList(), hasSize(1));
+        Optional<Label> correlationId = triggered.get().getLabels().stream().filter(label -> label.key().equals(Label.CORRELATION_ID)).findAny();
+        assertThat(correlationId.isPresent(), is(true));
+        assertThat(correlationId.get().value(), is(execution.getId()));
     }
 
     public void forEachItemEmptyItems() throws TimeoutException, URISyntaxException, IOException, QueueException {
diff --git a/core/src/test/java/io/kestra/plugin/core/http/DownloadTest.java b/core/src/test/java/io/kestra/plugin/core/http/DownloadTest.java
index 5442df59b..d69d91f0a 100644
--- a/core/src/test/java/io/kestra/plugin/core/http/DownloadTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/http/DownloadTest.java
@@ -1,6 +1,7 @@
 package io.kestra.plugin.core.http;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.storages.StorageInterface;
@@ -12,19 +13,16 @@ import io.micronaut.http.annotation.Controller;
 import io.micronaut.http.annotation.Get;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.runtime.server.EmbeddedServer;
-import io.kestra.core.junit.annotations.KestraTest;
 import jakarta.inject.Inject;
 import org.apache.commons.io.IOUtils;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.net.URI;
-import java.net.URL;
 import java.nio.charset.StandardCharsets;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.endsWith;
-import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.*;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -135,7 +133,7 @@ class DownloadTest {
 
         Download.Output output = task.run(runContext);
 
-        assertThat(output.getUri().toString(), endsWith("filename.jpg"));
+        assertThat(output.getUri().toString(), containsString("filename.jpg"));
     }
 
     @Controller()
diff --git a/core/src/test/resources/flows/invalids/system-labels.yaml b/core/src/test/resources/flows/invalids/system-labels.yaml
new file mode 100644
index 000000000..8f8ff6292
--- /dev/null
+++ b/core/src/test/resources/flows/invalids/system-labels.yaml
@@ -0,0 +1,11 @@
+id: system-labels
+namespace: io.kestra.tests
+
+labels:
+  system_label: system_key
+  system_id: id
+  another: label
+
+tasks:
+  - type: io.kestra.plugin.core.log.Log
+    message: "bla"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/subflow-child.yaml b/core/src/test/resources/flows/valids/subflow-child.yaml
new file mode 100644
index 000000000..bd3196e40
--- /dev/null
+++ b/core/src/test/resources/flows/valids/subflow-child.yaml
@@ -0,0 +1,8 @@
+id: subflow-child
+namespace: io.kestra.tests
+
+tasks:
+  - id: subflow
+    type: io.kestra.plugin.core.flow.Subflow
+    namespace: io.kestra.tests
+    flowId: subflow-grand-child
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/subflow-grand-child.yaml b/core/src/test/resources/flows/valids/subflow-grand-child.yaml
new file mode 100644
index 000000000..09e1ca5a8
--- /dev/null
+++ b/core/src/test/resources/flows/valids/subflow-grand-child.yaml
@@ -0,0 +1,7 @@
+id: subflow-grand-child
+namespace: io.kestra.tests
+
+tasks:
+  - id: firstLevel
+    type: io.kestra.plugin.core.log.Log
+    message: "My grandparent is {{labels.system_correlationId}}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/subflow-parent.yaml b/core/src/test/resources/flows/valids/subflow-parent.yaml
new file mode 100644
index 000000000..a3489e824
--- /dev/null
+++ b/core/src/test/resources/flows/valids/subflow-parent.yaml
@@ -0,0 +1,8 @@
+id: subflow-parent
+namespace: io.kestra.tests
+
+tasks:
+  - id: subflow
+    type: io.kestra.plugin.core.flow.Subflow
+    namespace: io.kestra.tests
+    flowId: subflow-child
\ No newline at end of file
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
index a27053ca6..df7226217 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
@@ -1592,4 +1592,16 @@ class ExecutionControllerTest extends JdbcH2ControllerTest {
         );
         assertThat(response.getCount(), is(3));
     }
+
+    @Test
+    void shouldRefuseSystemLabels() {
+        var error = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().retrieve(
+            HttpRequest
+                .POST("/api/v1/executions/io.kestra.tests/minimal?labels=system_label:system", null)
+                .contentType(MediaType.MULTIPART_FORM_DATA_TYPE),
+            Execution.class
+        ));
+
+        assertThat(error.getStatus(), is(HttpStatus.UNPROCESSABLE_ENTITY));
+    }
 }