feat(ssh): introduce new proxyCommand property

src/main/java/io/kestra/plugin/fs/ssh/Command.java

@@ -20,6 +20,7 @@
 
 import java.io.*;
 import java.nio.charset.StandardCharsets;
+import java.net.Socket;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -76,6 +77,26 @@
                       - touch kestra_was_here
                 """
         ),
+        @Example(
+            title = "Run SSH command through a proxy command",
+            full = true,
+            code = """
+                id: fs_ssh_proxy_command
+                namespace: company.team
+
+                tasks:
+                  - id: command
+                    type: io.kestra.plugin.fs.ssh.Command
+                    host: host
+                    username: user
+                    authMethod: PASSWORD
+                    password: "{{ secret('SSH_PASSWORD') }}"
+                    proxyCommand: |
+                      cloudflared access ssh --service-token-id {{ secret('SSH_PROXY_SERVICE_TOKEN_ID') }} --service-token-secret {{ secret('SSH_PROXY_SERVICE_TOKEN_SECRET') }} --hostname proxy_host
+                    commands:
+                      - mycmd
+                """
+        ),
         @Example(
             title = "Run SSH command using the local OpenSSH configuration",
             full = true,
@@ -121,6 +142,15 @@ public class Command extends Task implements SshInterface, RunnableTask<Command.
     )
     private Property<String> openSSHConfigPath;
 
+    @Schema(
+        title = "Proxy command",
+        description = """
+            Optional local command used to establish the SSH transport (OpenSSH `ProxyCommand` semantics).
+            Example: `cloudflared access ssh --service-token-id ... --service-token-secret ... --hostname ...`
+            """
+    )
+    private Property<String> proxyCommand;
+
     @Schema(
         title = "SSH authentication configuration",
         description = """
@@ -223,6 +253,10 @@ public Output run(RunContext runContext) throws Exception {
                 runContext.render(username).as(String.class).orElse(null),
                 renderedHost, Integer.parseInt(renderedPort)
             );
+            var rProxyCommand = runContext.render(proxyCommand).as(String.class);
+            if (rProxyCommand.isPresent()) {
+                session.setProxy(new ProcessProxyCommand(rProxyCommand.orElseThrow(), session.getUserName()));
+            }
 
             // enable disabled by default weak RSA/SHA1 algorithm
             if (runContext.render(enableSshRsa1).as(Boolean.class).orElseThrow()) {
@@ -345,6 +379,84 @@ public void showMessage(String message) {
         }
     }
 
+    private static final class ProcessProxyCommand implements Proxy {
+        private final String command;
+        private final String username;
+
+        private Process process;
+        private InputStream inputStream;
+        private OutputStream outputStream;
+
+        private ProcessProxyCommand(String command, String username) {
+            this.command = command;
+            this.username = username;
+        }
+
+        @Override
+        public void connect(SocketFactory socketFactory, String host, int port, int timeout) throws Exception {
+            var resolvedCommand = command
+                .replace("%h", host)
+                .replace("%p", String.valueOf(port))
+                .replace("%r", username == null ? "" : username);
+
+            var processBuilder = new ProcessBuilder(shellCommand(resolvedCommand));
+            processBuilder.redirectError(ProcessBuilder.Redirect.DISCARD);
+            process = processBuilder.start();
+            inputStream = process.getInputStream();
+            outputStream = process.getOutputStream();
+
+            if (!process.isAlive()) {
+                throw new JSchException("Proxy command exited immediately: " + resolvedCommand);
+            }
+        }
+
+        private static String[] shellCommand(String command) {
+            if (System.getProperty("os.name").toLowerCase().contains("win")) {
+                return new String[] {"cmd.exe", "/c", command};
+            }
+
+            return new String[] {"/bin/sh", "-c", command};
+        }
+
+        @Override
+        public InputStream getInputStream() {
+            return inputStream;
+        }
+
+        @Override
+        public OutputStream getOutputStream() {
+            return outputStream;
+        }
+
+        @Override
+        public Socket getSocket() {
+            return null;
+        }
+
+        @Override
+        public void close() {
+            if (inputStream != null) {
+                try {
+                    inputStream.close();
+                } catch (IOException ignored) {
+                    // no-op
+                }
+            }
+
+            if (outputStream != null) {
+                try {
+                    outputStream.close();
+                } catch (IOException ignored) {
+                    // no-op
+                }
+            }
+
+            if (process != null) {
+                process.destroy();
+            }
+        }
+    }
+
     private static class LogRunnable implements Runnable {
         private final InputStream inputStream;
 

src/main/java/io/kestra/plugin/fs/ssh/SshInterface.java

@@ -60,6 +60,12 @@ public interface SshInterface {
     )
     Property<String> getOpenSSHConfigPath();
 
+    @Schema(
+        title = "Proxy command",
+        description = "Optional local command used to establish the SSH transport (OpenSSH `ProxyCommand` semantics)."
+    )
+    Property<String> getProxyCommand();
+
     enum AuthMethod {
         PASSWORD,
         PUBLIC_KEY,

src/test/java/io/kestra/plugin/fs/ssh/CommandTest.java

@@ -7,7 +7,7 @@
 import io.kestra.core.utils.TestsUtils;
 import io.kestra.plugin.fs.ssh.SshInterface.AuthMethod;
 import jakarta.inject.Inject;
-import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.io.ByteArrayOutputStream;
@@ -60,6 +60,38 @@ void run_passwordMethod() throws Exception {
         assertThat(run.getVars().get("err"), is("2"));
     }
 
+    @Test
+    void run_passwordMethod_withProxyCommand() throws Exception {
+        var command = Command.builder()
+            .id(IdUtils.create())
+            .type(Command.class.getName())
+            .host(Property.ofValue("unreachable.invalid"))
+            .username(USERNAME)
+            .authMethod(Property.ofValue(AuthMethod.PASSWORD))
+            .password(PASSWORD)
+            .port(Property.ofValue("2222"))
+            .proxyCommand(Property.ofValue("nc {{ inputs.proxyHost }} {{ inputs.proxyPort }}"))
+            .commands(new String[] {
+                "echo 0",
+                "echo 1",
+                ">&2 echo 2",
+                "echo '::{\"outputs\":{\"out\":\"1\"}}::'",
+                ">&2 echo '::{\"outputs\":{\"err\":\"2\"}}::'",
+            })
+            .build();
+
+        var exception = Assertions.assertThrows(Exception.class, () -> command.run(TestsUtils.mockRunContext(
+            runContextFactory,
+            command,
+            Map.of(
+                "proxyHost", "127.0.0.1",
+                "proxyPort", "1"
+            )
+        )));
+
+        assertThat(String.valueOf(exception.getMessage()).contains("UnknownHostException"), is(false));
+    }
+
     @Test
     void run_pubkeyMethod() throws Exception {
         Path tempDir = Files.createTempDirectory("ssh-key");