Native SSH Tunnel support for database connections #6573 (#6776)

* - Add native SSH tunnel support for database connections, allowing users to connect
  to databases in private networks through an SSH bastion host
- Supports password, keyboard-interactive, and private key authentication
- New "SSH Tunnel" tab in the database connection editor with enable/disable logic
- SSH keepalive (30s interval) to prevent VPN/firewall from dropping idle connections
- Tunnel lifecycle properly managed even for pipeline connection groups

- **SshTunnelManager** (new): manages JSch sessions with local port forwarding
- **IDatabase/BaseDatabaseMeta**: SSH tunnel fields persisted via `@HopMetadataProperty`
- **Database.java**: opens tunnel before JDBC connect, closes in `closeConnectionOnly()`
  (not `disconnect()`) to prevent tunnel leaks with grouped connections
- **DatabaseMetaEditor**: new SSH Tunnel tab with field enable/disable based on config
- i18n: English and Italian labels

- [x] Unit tests for SshTunnelManager (5 tests)
- [x] Unit tests for Database SSH tunnel integration (5 tests)
- [x] All 10 tests pass
- [x] Manual testing with MySQL over SSH tunnel (verified working)

UPGRADE JUNIT 5

revert ldap changes

* Add IT test

---------

Co-authored-by: Giacomo <gscaglione@paa.it>
Co-authored-by: Hans Van Akelyen <hans.van.akelyen@gmail.com>

Author: 3 people

core/pom.xml

@@ -58,6 +58,10 @@
             <groupId>com.fasterxml.jackson.dataformat</groupId>
             <artifactId>jackson-dataformat-avro</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.github.mwiede</groupId>
+            <artifactId>jsch</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>

core/src/main/java/org/apache/hop/core/database/BaseDatabaseMeta.java

@@ -184,6 +184,21 @@ public abstract class BaseDatabaseMeta implements Cloneable, IDatabase {
   @HopMetadataProperty protected String pluginId;
   @HopMetadataProperty protected String pluginName;
 
+  // SSH Tunnel fields
+  @HopMetadataProperty protected boolean sshTunnelEnabled;
+  @HopMetadataProperty protected String sshTunnelHost;
+  @HopMetadataProperty protected String sshTunnelPort;
+  @HopMetadataProperty protected String sshTunnelUsername;
+
+  @HopMetadataProperty(password = true)
+  protected String sshTunnelPassword;
+
+  @HopMetadataProperty protected boolean sshTunnelUsePrivateKey;
+  @HopMetadataProperty protected String sshTunnelPrivateKeyFile;
+
+  @HopMetadataProperty(password = true)
+  protected String sshTunnelPassphrase;
+
   public BaseDatabaseMeta() {
     attributes = Collections.synchronizedMap(new HashMap<>());
     changed = false;

core/src/main/java/org/apache/hop/core/database/Database.java

@@ -160,6 +160,8 @@ public class Database implements IVariables, ILoggingObject, AutoCloseable {
 
   private int nrExecutedCommits;
 
+  private SshTunnelManager sshTunnelManager;
+
   private static final List<IValueMeta> valueMetaPluginClasses;
 
   static {
@@ -435,7 +437,24 @@ private void connectUsingClass(String classname, String partitionId) throws HopD
     }
 
     try {
-      String url = resolve(databaseMeta.getURL(this));
+      // Open SSH tunnel if configured
+      String url;
+      if (databaseMeta.isSshTunnelEnabled() && !Utils.isEmpty(databaseMeta.getSshTunnelHost())) {
+        sshTunnelManager = new SshTunnelManager();
+        int localPort = sshTunnelManager.openTunnel(this, databaseMeta, log);
+
+        // Build URL using tunnel endpoint (localhost + forwarded port)
+        String tunnelUrl =
+            databaseMeta
+                .getIDatabase()
+                .getURL(
+                    "localhost",
+                    String.valueOf(localPort),
+                    resolve(databaseMeta.getDatabaseName()));
+        url = resolve(tunnelUrl);
+      } else {
+        url = resolve(databaseMeta.getURL(this));
+      }
       log.logDebug("Connecting to database using URL: " + url);
 
       String username = resolve(databaseMeta.getUsername());
@@ -608,6 +627,14 @@ public synchronized void closeConnectionOnly() throws HopDatabaseException {
       }
     } catch (SQLException e) {
       throw new HopDatabaseException("Error disconnecting from database '" + this + "'", e);
+    } finally {
+      // Close SSH tunnel after JDBC connection is closed.
+      // This must be here (not in disconnect()) because grouped connections
+      // use closeConnectionOnly() directly and would otherwise leak tunnels.
+      if (sshTunnelManager != null) {
+        sshTunnelManager.closeTunnel(log);
+        sshTunnelManager = null;
+      }
     }
   }
 

core/src/main/java/org/apache/hop/core/database/DatabaseMeta.java

@@ -2328,4 +2328,70 @@ public List<String> getRemoveItems() {
   public boolean isHideUrlInTestConnection() {
     return iDatabase.isHideUrlInTestConnection();
   }
+
+  // SSH Tunnel delegation methods
+
+  public boolean isSshTunnelEnabled() {
+    return iDatabase.isSshTunnelEnabled();
+  }
+
+  public void setSshTunnelEnabled(boolean enabled) {
+    iDatabase.setSshTunnelEnabled(enabled);
+  }
+
+  public String getSshTunnelHost() {
+    return iDatabase.getSshTunnelHost();
+  }
+
+  public void setSshTunnelHost(String host) {
+    iDatabase.setSshTunnelHost(host);
+  }
+
+  public String getSshTunnelPort() {
+    return iDatabase.getSshTunnelPort();
+  }
+
+  public void setSshTunnelPort(String port) {
+    iDatabase.setSshTunnelPort(port);
+  }
+
+  public String getSshTunnelUsername() {
+    return iDatabase.getSshTunnelUsername();
+  }
+
+  public void setSshTunnelUsername(String username) {
+    iDatabase.setSshTunnelUsername(username);
+  }
+
+  public String getSshTunnelPassword() {
+    return iDatabase.getSshTunnelPassword();
+  }
+
+  public void setSshTunnelPassword(String password) {
+    iDatabase.setSshTunnelPassword(password);
+  }
+
+  public boolean isSshTunnelUsePrivateKey() {
+    return iDatabase.isSshTunnelUsePrivateKey();
+  }
+
+  public void setSshTunnelUsePrivateKey(boolean usePrivateKey) {
+    iDatabase.setSshTunnelUsePrivateKey(usePrivateKey);
+  }
+
+  public String getSshTunnelPrivateKeyFile() {
+    return iDatabase.getSshTunnelPrivateKeyFile();
+  }
+
+  public void setSshTunnelPrivateKeyFile(String privateKeyFile) {
+    iDatabase.setSshTunnelPrivateKeyFile(privateKeyFile);
+  }
+
+  public String getSshTunnelPassphrase() {
+    return iDatabase.getSshTunnelPassphrase();
+  }
+
+  public void setSshTunnelPassphrase(String passphrase) {
+    iDatabase.setSshTunnelPassphrase(passphrase);
+  }
 }

core/src/main/java/org/apache/hop/core/database/IDatabase.java

@@ -1209,4 +1209,38 @@ default String getLegacyColumnName(
    * @return true to hide URL information in test connection results
    */
   boolean isHideUrlInTestConnection();
+
+  // SSH Tunnel configuration methods
+
+  boolean isSshTunnelEnabled();
+
+  void setSshTunnelEnabled(boolean enabled);
+
+  String getSshTunnelHost();
+
+  void setSshTunnelHost(String host);
+
+  String getSshTunnelPort();
+
+  void setSshTunnelPort(String port);
+
+  String getSshTunnelUsername();
+
+  void setSshTunnelUsername(String username);
+
+  String getSshTunnelPassword();
+
+  void setSshTunnelPassword(String password);
+
+  boolean isSshTunnelUsePrivateKey();
+
+  void setSshTunnelUsePrivateKey(boolean usePrivateKey);
+
+  String getSshTunnelPrivateKeyFile();
+
+  void setSshTunnelPrivateKeyFile(String privateKeyFile);
+
+  String getSshTunnelPassphrase();
+
+  void setSshTunnelPassphrase(String passphrase);
 }

core/src/main/java/org/apache/hop/core/database/SshTunnelManager.java

@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.core.database;
+
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.UIKeyboardInteractive;
+import com.jcraft.jsch.UserInfo;
+import java.util.Properties;
+import org.apache.hop.core.Const;
+import org.apache.hop.core.encryption.Encr;
+import org.apache.hop.core.exception.HopDatabaseException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.variables.IVariables;
+
+/** Manages SSH tunnels for database connections using JSch port forwarding. */
+public class SshTunnelManager {
+
+  private Session session;
+  private int localPort;
+
+  /**
+   * Opens an SSH tunnel with local port forwarding.
+   *
+   * @param variables variables for resolving expressions
+   * @param databaseMeta the database metadata containing SSH tunnel configuration
+   * @param log the log channel
+   * @return the local port that forwards to the remote database
+   * @throws HopDatabaseException if the tunnel cannot be established
+   */
+  public int openTunnel(IVariables variables, DatabaseMeta databaseMeta, ILogChannel log)
+      throws HopDatabaseException {
+    try {
+      String sshHost = variables.resolve(databaseMeta.getSshTunnelHost());
+      int sshPort = Const.toInt(variables.resolve(databaseMeta.getSshTunnelPort()), 22);
+      String sshUser = variables.resolve(databaseMeta.getSshTunnelUsername());
+      String sshPassword =
+          Encr.decryptPasswordOptionallyEncrypted(
+              variables.resolve(databaseMeta.getSshTunnelPassword()));
+
+      String remoteHost = variables.resolve(databaseMeta.getHostname());
+      int remotePort =
+          Const.toInt(
+              variables.resolve(databaseMeta.getPort()), databaseMeta.getDefaultDatabasePort());
+
+      JSch jsch = new JSch();
+
+      // Private key authentication
+      if (databaseMeta.isSshTunnelUsePrivateKey()) {
+        String keyFile = variables.resolve(databaseMeta.getSshTunnelPrivateKeyFile());
+        String passphrase =
+            Encr.decryptPasswordOptionallyEncrypted(
+                variables.resolve(databaseMeta.getSshTunnelPassphrase()));
+        if (!Utils.isEmpty(passphrase)) {
+          jsch.addIdentity(keyFile, passphrase);
+        } else {
+          jsch.addIdentity(keyFile);
+        }
+      }
+
+      session = jsch.getSession(sshUser, sshHost, sshPort);
+
+      // Password authentication (supports both password and keyboard-interactive)
+      if (!databaseMeta.isSshTunnelUsePrivateKey() && !Utils.isEmpty(sshPassword)) {
+        session.setPassword(sshPassword);
+        session.setUserInfo((UserInfo) new SshPasswordUserInfo(sshPassword));
+      }
+
+      Properties config = new Properties();
+      config.put("StrictHostKeyChecking", "no");
+      config.put("PreferredAuthentications", "publickey,keyboard-interactive,password");
+      session.setConfig(config);
+
+      // Send keepalive every 30s to prevent VPN/firewall from dropping idle connections
+      session.setServerAliveInterval(30000);
+      session.setServerAliveCountMax(3);
+
+      log.logBasic("Opening SSH tunnel to " + sshHost + ":" + sshPort + " as user " + sshUser);
+      session.connect(30000);
+
+      // Local port 0 = auto-assign a free port
+      localPort = session.setPortForwardingL(0, remoteHost, remotePort);
+      log.logBasic(
+          "SSH tunnel established: localhost:"
+              + localPort
+              + " -> "
+              + remoteHost
+              + ":"
+              + remotePort);
+
+      return localPort;
+    } catch (Exception e) {
+      closeTunnel(log);
+      throw new HopDatabaseException("Failed to open SSH tunnel", e);
+    }
+  }
+
+  /** Closes the SSH tunnel and disconnects the session. */
+  public void closeTunnel(ILogChannel log) {
+    if (session != null && session.isConnected()) {
+      try {
+        session.delPortForwardingL(localPort);
+      } catch (Exception e) {
+        log.logDebug("Error removing port forwarding: " + e.getMessage());
+      }
+      session.disconnect();
+      log.logBasic("SSH tunnel closed");
+    }
+    session = null;
+  }
+
+  public boolean isOpen() {
+    return session != null && session.isConnected();
+  }
+
+  public int getLocalPort() {
+    return localPort;
+  }
+
+  /** Handles keyboard-interactive authentication by responding with the password. */
+  private static class SshPasswordUserInfo implements UserInfo, UIKeyboardInteractive {
+    private final String password;
+
+    SshPasswordUserInfo(String password) {
+      this.password = password;
+    }
+
+    @Override
+    public String[] promptKeyboardInteractive(
+        String destination, String name, String instruction, String[] prompt, boolean[] echo) {
+      return new String[] {password};
+    }
+
+    @Override
+    public String getPassphrase() {
+      return null;
+    }
+
+    @Override
+    public String getPassword() {
+      return password;
+    }
+
+    @Override
+    public boolean promptPassword(String message) {
+      return true;
+    }
+
+    @Override
+    public boolean promptPassphrase(String message) {
+      return false;
+    }
+
+    @Override
+    public boolean promptYesNo(String message) {
+      return true;
+    }
+
+    @Override
+    public void showMessage(String message) {}
+  }
+}