fix(core)!: don't nest inputs by in hierarchical maps (#7964)

Fixes #7964

Nesting inputs in a hierarchical map lead to not decrypting secret.
A secret ID is an ID, not a structure, so we should not create a nested map if a secret ID has a dot in it.

core/src/main/java/io/kestra/core/runners/FlowInputOutput.java

@@ -203,7 +203,7 @@ public class FlowInputOutput {
         final Execution execution,
         final Map<String, ?> data
     ) {
-        Map<String, Object> resolved = this.resolveInputs(inputs, flow, execution, data)
+        return this.resolveInputs(inputs, flow, execution, data)
             .stream()
             .filter(InputAndValue::enabled)
             .map(it -> {
@@ -214,7 +214,6 @@ public class FlowInputOutput {
                 return new AbstractMap.SimpleEntry<>(it.input().getId(), it.value());
             })
             .collect(HashMap::new, (m,v)-> m.put(v.getKey(), v.getValue()), HashMap::putAll);
-        return MapUtils.flattenToNestedMap(resolved);
     }
 
     /**

core/src/test/java/io/kestra/core/runners/InputsTest.java

@@ -87,6 +87,9 @@ public class InputsTest {
     @Inject
     private FlowInputOutput flowIO;
 
+    @Inject
+    private FlowInputOutput flowInputOutput;
+
     private Map<String, Object> typedInputs(Map<String, Object> map) {
         return typedInputs(map, flowRepository.findById(null, "io.kestra.tests", "inputs").get());
     }
@@ -370,7 +373,7 @@ public class InputsTest {
         assertThat((String) execution.findTaskRunsByTaskId("jsonOutput").getFirst().getOutputs().get("value")).isEqualTo("{}");
     }
 
-    @RetryingTest(5) // it can happen that a log from another execution arrives first, so we enable retry
+    @Test
     @LoadFlows({"flows/valids/input-log-secret.yaml"})
     void shouldNotLogSecretInput() throws TimeoutException, QueueException {
         Flux<LogEntry> receive = TestsUtils.receive(logQueue, l -> {});
@@ -378,7 +381,9 @@ public class InputsTest {
         Execution execution = runnerUtils.runOne(
             null,
             "io.kestra.tests",
-            "input-log-secret"
+            "input-log-secret",
+            null,
+            (flow, exec) -> flowInputOutput.readExecutionInputs(flow, exec, Map.of("nested.key", "pass"))
         );
 
         assertThat(execution.getTaskRunList()).hasSize(1);
@@ -386,6 +391,6 @@ public class InputsTest {
 
         var logEntry = receive.blockLast();
         assertThat(logEntry).isNotNull();
-        assertThat(logEntry.getMessage()).isEqualTo("This is my secret: ********");
+        assertThat(logEntry.getMessage()).isEqualTo("These are my secrets: **** - ********");
     }
 }

core/src/test/resources/flows/valids/input-log-secret.yaml

@@ -6,7 +6,11 @@ inputs:
     type: SECRET
     defaults: password
 
+  - id: nested.key
+    type: SECRET
+    defaults: password
+
 tasks:
   - id: log-secret
     type: io.kestra.plugin.core.log.Log
-    message: "This is my secret: {{inputs.secret}}"
+    message: "These are my secrets: {{inputs['nested.key']}} - {{inputs.secret}}"