Repository: kestra-io/kestra (java)

Issue:
Unable to authenticate to private image repo if GetAuthToken is inside working dir
### Describe the issue

If the GetAuthToken task is executed within a working directory, then the subsequent task that authenticates to a private image repo will fail. For example

**this fails**
```yaml
id: ecr_test
namespace: demo
description: check user creds

tasks:
  - id: wdir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
    - id: auth
      type: io.kestra.plugin.aws.ecr.GetAuthToken
      accessKeyId: "{{secret('AWS_ACCESS_KEY_ID')}}"
      secretKeyId: "{{secret('AWS_SECRET_ACCESS_KEY')}}"
      region: "eu-west-3"
    - id: image_test
      type: io.kestra.plugin.scripts.shell.Commands
      runner: DOCKER
      docker:
        credentials:
          password: "{{outputs.auth.token.value)}}"
          username: AWS
        image: "account_id.dkr.ecr.eu-west-3.amazonaws.com/shell:latest"
  
        pullPolicy: ALWAYS
      commands:
        - echo abc
```

Fails with error message
```log
2024-06-11 16:12:10.306Status 404: {"message":"pull access denied for account_id.dkr.ecr.eu-west-3.amazonaws.com/shell, repository does not exist or may require 'docker login': denied: Your Authorization Token is invalid."}

```

**This works**
```yaml
id: ecr_test
namespace: demo
description: check user creds

tasks:
  - id: auth
    type: io.kestra.plugin.aws.ecr.GetAuthToken
    accessKeyId: "{{secret('AWS_ACCESS_KEY_ID')}}"
    secretKeyId: "{{secret('AWS_SECRET_ACCESS_KEY')}}"
    region: "eu-west-3"
  - id: wdir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
    - id: image_test
      type: io.kestra.plugin.scripts.shell.Commands
      runner: DOCKER
      docker:
        credentials:
          password: "{{outputs.auth.token.value)}}"
          username: AWS
        image: "account_id.dkr.ecr.eu-west-3.amazonaws.com/shell:latest"
  
        pullPolicy: ALWAYS
      commands:
        - echo abc
```
```

### Environment

- Kestra Version: 0.17.2
- Operating System Docker



Pull request:
fix(core): decrypt outputs for tasks within WorkingDirectory
closes #4000


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/FilesService.java b/core/src/main/java/io/kestra/core/runners/FilesService.java
index 3e6cb0bb9..36bbee9b8 100644
--- a/core/src/main/java/io/kestra/core/runners/FilesService.java
+++ b/core/src/main/java/io/kestra/core/runners/FilesService.java
@@ -2,6 +2,7 @@ package io.kestra.core.runners;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.tasks.runners.PluginUtilsService;
+import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.ListUtils;
 import org.apache.commons.io.IOUtils;
 import org.slf4j.Logger;
@@ -85,10 +86,14 @@ public abstract class FilesService {
                 .filter(path -> pathMatcher.matches(runContext.tempDir().relativize(path)))
                 .map(throwFunction(path -> new AbstractMap.SimpleEntry<>(
                     runContext.tempDir().relativize(path).toString(),
-                    runContext.storage().putFile(path.toFile())
+                    runContext.storage().putFile(path.toFile(), resolveUniqueNameForFile(path))
                 )))
                 .toList()
                 .stream();
         }
     }
+
+    private static String resolveUniqueNameForFile(final Path path) {
+        return IdUtils.from(path.toString()) + "-" + path.toFile().getName();
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index 5c7385398..eeed44845 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -339,7 +339,7 @@ public class RunContext {
             if (execution.getTaskRunList() != null) {
                 Map<String, Object> outputs = new HashMap<>(execution.outputs());
                 if (decryptVariables) {
-                    decryptOutputs(outputs);
+                    outputs = decryptOutputs(outputs);
                 }
                 builder.put("outputs", outputs);
             }
@@ -404,22 +404,30 @@ public class RunContext {
         return builder.build();
     }
 
-    private void decryptOutputs(Map<String, Object> outputs) {
-        for (var entry: outputs.entrySet()) {
+    private Map<String, Object> decryptOutputs(Map<String, Object> mapToDecrypt) {
+        if (mapToDecrypt == null) {
+            return null;
+        }
+
+        Map<String, Object> decryptedMap = new HashMap<>();
+        for (var entry: mapToDecrypt.entrySet()) {
+            decryptedMap.put(entry.getKey(), entry.getValue());
             if (entry.getValue() instanceof Map map) {
-                // if some outputs are of type EncryptedString we decode them and replace the object
+                // if some value are of type EncryptedString we decode them and replace the object
                 if (EncryptedString.TYPE.equalsIgnoreCase((String)map.get("type"))) {
                     try {
                         String decoded = decrypt((String) map.get("value"));
-                        outputs.put(entry.getKey(), decoded);
+                        decryptedMap.put(entry.getKey(), decoded);
                     } catch (GeneralSecurityException e) {
                         throw new RuntimeException(e);
                     }
                 }  else {
-                    decryptOutputs((Map<String, Object>) map);
+                    decryptedMap.put(entry.getKey(), decryptOutputs((Map<String, Object>) map));
                 }
             }
         }
+
+        return decryptedMap;
     }
 
     private Map<String, Object> variables(TaskRun taskRun) {
@@ -502,6 +510,8 @@ public class RunContext {
         runContext.runContextLogger = this.runContextLogger;
         runContext.tempBasedPath = this.tempBasedPath;
         runContext.temporaryDirectory = this.temporaryDirectory;
+        runContext.pluginConfiguration = this.pluginConfiguration;
+        runContext.secretKey = this.secretKey;
 
         return runContext;
     }
@@ -583,6 +593,15 @@ public class RunContext {
         return newContext;
     }
 
+    public RunContext forWorkingDirectoryTask() {
+        Map<String, Object> decryptedVariables = new HashMap<>(this.variables);
+        if (this.variables.get("outputs") != null) {
+            decryptedVariables.put("outputs", decryptOutputs((Map<String, Object>) this.variables.get("outputs")));
+        }
+
+        return this.clone(decryptedVariables);
+    }
+
     public RunContext forTaskRunner(TaskRunner taskRunner) {
         this.initPluginConfiguration(applicationContext, taskRunner.getClass(), taskRunner.getType());
 
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 06779bf58..c7c481326 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -287,7 +287,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
                     WorkerTask currentWorkerTask = workingDirectory.workerTask(
                         workerTask.getTaskRun(),
                         currentTask,
-                        runContext
+                        runContext.forWorkingDirectoryTask()
                     );
 
                     // all tasks will be handled immediately by the worker


Test patch:
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/WorkingDirectoryTest.java b/core/src/test/java/io/kestra/plugin/core/flow/WorkingDirectoryTest.java
index 980f6c694..44baf6b10 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/WorkingDirectoryTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/WorkingDirectoryTest.java
@@ -6,22 +6,23 @@ import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.tasks.common.EncryptedString;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.runners.RunnerUtils;
 import io.kestra.core.storages.InternalStorage;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.IdUtils;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.junit.jupiter.api.Test;
 import org.junitpioneer.jupiter.RetryingTest;
 
 import java.io.ByteArrayInputStream;
-import java.io.File;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.URI;
-import java.nio.file.Files;
+import java.security.GeneralSecurityException;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
@@ -30,7 +31,6 @@ import java.util.concurrent.TimeoutException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.hamcrest.io.FileMatchers.anExistingFile;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -38,6 +38,9 @@ public class WorkingDirectoryTest extends AbstractMemoryRunnerTest {
     @Inject
     Suite suite;
 
+    @Inject
+    RunContextFactory runContextFactory;
+
     @Test
     void success() throws TimeoutException {
        suite.success(runnerUtils);
@@ -83,6 +86,11 @@ public class WorkingDirectoryTest extends AbstractMemoryRunnerTest {
         suite.outputFiles(runnerUtils);
     }
 
+    @Test
+    void encryption() throws Exception {
+        suite.encryption(runnerUtils, runContextFactory);
+    }
+
     @Singleton
     public static class Suite {
         @Inject
@@ -154,8 +162,15 @@ public class WorkingDirectoryTest extends AbstractMemoryRunnerTest {
                 storageContext
                 , storageInterface
             );
-            URI fileURI = URI.create("kestra:" + storageContext.getContextStorageURI() + "/input.txt");
-            assertThat(new String(storage.getFile(fileURI).readAllBytes()), is("Hello World"));
+
+            TaskRun taskRun = execution.getTaskRunList().get(1);
+            Map<String, Object> outputs = taskRun.getOutputs();
+            assertThat(outputs, hasKey("uris"));
+
+            URI uri = URI.create(((Map<String, String>) outputs.get("uris")).get("input.txt"));
+
+            assertTrue(uri.toString().endsWith("input.txt"));
+            assertThat(new String(storage.getFile(uri).readAllBytes()), is("Hello World"));
         }
 
         @SuppressWarnings("unchecked")
@@ -236,6 +251,18 @@ public class WorkingDirectoryTest extends AbstractMemoryRunnerTest {
             assertThat(execution.findTaskRunsByTaskId("t3").get(0).getOutputs().get("value"), is("third"));
         }
 
+        public void encryption(RunnerUtils runnerUtils, RunContextFactory runContextFactory) throws TimeoutException, GeneralSecurityException {
+            Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "working-directory-taskrun-encrypted");
+
+            assertThat(execution.getTaskRunList(), hasSize(3));
+            Map<String, Object> encryptedString = (Map<String, Object>) execution.findTaskRunsByTaskId("encrypted").get(0).getOutputs().get("value");
+            assertThat(encryptedString.get("type"), is(EncryptedString.TYPE));
+            String encryptedValue = (String) encryptedString.get("value");
+            assertThat(encryptedValue, is(not("Hello World")));
+            assertThat(runContextFactory.of().decrypt(encryptedValue), is("Hello World"));
+            assertThat(execution.findTaskRunsByTaskId("decrypted").get(0).getOutputs().get("value"), is("Hello World"));
+        }
+
         private void put(String path, String content) throws IOException {
             storageInterface.put(
                 null,
diff --git a/core/src/test/resources/flows/valids/working-directory-taskrun-encrypted.yml b/core/src/test/resources/flows/valids/working-directory-taskrun-encrypted.yml
new file mode 100644
index 000000000..e79c7cb5c
--- /dev/null
+++ b/core/src/test/resources/flows/valids/working-directory-taskrun-encrypted.yml
@@ -0,0 +1,13 @@
+id: working-directory-taskrun-encrypted
+namespace: io.kestra.tests
+
+tasks:
+  - id: workingDir
+    type: io.kestra.plugin.core.flow.WorkingDirectory
+    tasks:
+      - id: encrypted
+        type: io.kestra.core.tasks.test.Encrypted
+        format: "Hello World"
+      - id: decrypted
+        type: io.kestra.plugin.core.debug.Return
+        format: "{{outputs.encrypted.value}}"
\ No newline at end of file