fix(ssh-tunnel): support ed25519 and ECDSA keys, not just RSA

`SSHManager.create_tunnel` unconditionally called
`RSAKey.from_private_key` regardless of key type, so users who pasted a
modern key (ed25519, ECDSA) got an opaque
`paramiko.ssh_exception.SSHException: unpack requires a buffer of 4
bytes` and the tunnel never opened.

paramiko does not expose a polymorphic `PKey.from_private_key`; each
key class only accepts its own format. Add `_load_private_key` that
tries Ed25519Key → ECDSAKey → RSAKey in order and returns the first
that parses cleanly. Modern types are tried first because RSA's parser
is the most permissive and would happily accept a non-RSA buffer with
the same vague error otherwise.

If none of the loaders succeed, surface a single `SSHException` whose
message lists every type that was attempted — much more actionable
than the original 4-bytes-buffer error.

Companion regression tests (added in this PR's earlier commit) cover
both the new ed25519 path and the backward-compatible RSA path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

superset/extensions/ssh.py

@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING
 
 import sshtunnel
 from flask import Flask
-from paramiko import RSAKey
+from paramiko import ECDSAKey, Ed25519Key, PKey, RSAKey, SSHException
 
 from superset.commands.database.ssh_tunnel.exceptions import SSHTunnelDatabasePortError
 from superset.databases.utils import make_url_safe
@@ -30,6 +30,32 @@ from superset.utils.class_utils import load_class_from_name
 if TYPE_CHECKING:
     from superset.databases.ssh_tunnel.models import SSHTunnel
 
+# Order matters: paramiko's per-class loaders raise SSHException with vague
+# "unpack requires 4 bytes" messages on type mismatches, so we try the more
+# modern key types first (ed25519, ECDSA) and fall back to RSA, which is the
+# most permissive parser and the historical default in this codebase.
+_SSH_KEY_TYPES: tuple[type[PKey], ...] = (Ed25519Key, ECDSAKey, RSAKey)
+
+
+def _load_private_key(pem: str, password: str | None) -> PKey:
+    """Load a private key PEM regardless of algorithm (ed25519, ECDSA, RSA).
+
+    paramiko does not expose a polymorphic ``PKey.from_private_key``; each
+    key class only accepts its own format. Iterate over the supported types
+    and return the first that parses cleanly.
+    """
+    last_exc: Exception | None = None
+    for key_class in _SSH_KEY_TYPES:
+        try:
+            return key_class.from_private_key(StringIO(pem), password=password)
+        except SSHException as exc:
+            last_exc = exc
+            continue
+    raise SSHException(
+        "Unable to parse SSH private key as any of "
+        f"{', '.join(k.__name__ for k in _SSH_KEY_TYPES)}: {last_exc}"
+    )
+
 
 class SSHManager:
     def __init__(self, app: Flask) -> None:
@@ -71,11 +97,9 @@ class SSHManager:
         if ssh_tunnel.password:
             params["ssh_password"] = ssh_tunnel.password
         elif ssh_tunnel.private_key:
-            private_key_file = StringIO(ssh_tunnel.private_key)
-            private_key = RSAKey.from_private_key(
-                private_key_file, ssh_tunnel.private_key_password
+            params["ssh_pkey"] = _load_private_key(
+                ssh_tunnel.private_key, ssh_tunnel.private_key_password
             )
-            params["ssh_pkey"] = private_key
 
         return sshtunnel.open_tunnel(**params)
 

tests/unit_tests/extensions/ssh_test.py

@@ -14,11 +14,17 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import sshtunnel
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import (
+    Encoding,
+    NoEncryption,
+    PrivateFormat,
+)
 
-from superset.extensions.ssh import SSHManagerFactory
+from superset.extensions.ssh import SSHManager, SSHManagerFactory
 
 
 def test_ssh_tunnel_timeout_setting() -> None:
@@ -34,3 +40,100 @@ def test_ssh_tunnel_timeout_setting() -> None:
     factory.init_app(app)
     assert sshtunnel.TUNNEL_TIMEOUT == 123.0
     assert sshtunnel.SSH_TIMEOUT == 321.0
+
+
+def _make_ed25519_pem() -> str:
+    """Generate a fresh OpenSSH-format ed25519 private key PEM."""
+    key = Ed25519PrivateKey.generate()
+    return key.private_bytes(
+        encoding=Encoding.PEM,
+        format=PrivateFormat.OpenSSH,
+        encryption_algorithm=NoEncryption(),
+    ).decode()
+
+
+def test_create_tunnel_accepts_ed25519_private_key() -> None:
+    """
+    Regression for #24180: ed25519 SSH keys must be loadable for tunnel
+    setup. The bug surfaces as ``unpack requires 4 bytes`` because the
+    code unconditionally calls ``RSAKey.from_private_key`` regardless of
+    key type, which mis-parses the ed25519 byte stream.
+
+    The Superset UI accepts any key the user pastes, so the fix is to
+    detect the key type (or use ``paramiko.PKey.from_private_key`` once
+    paramiko exposes a polymorphic loader) rather than hard-coding RSA.
+
+    This test exercises ``create_tunnel`` end-to-end with a freshly
+    generated ed25519 key. It mocks ``sshtunnel.open_tunnel`` so the
+    test does not actually open a network connection — only the key
+    parsing path is exercised.
+    """
+    app = Mock()
+    app.config = {
+        "SSH_TUNNEL_LOCAL_BIND_ADDRESS": "127.0.0.1",
+        "SSH_TUNNEL_TIMEOUT_SEC": 10.0,
+        "SSH_TUNNEL_PACKET_TIMEOUT_SEC": 10.0,
+    }
+    manager = SSHManager(app)
+
+    ssh_tunnel = Mock()
+    ssh_tunnel.server_address = "ssh.example.com"
+    ssh_tunnel.server_port = 22
+    ssh_tunnel.username = "tunneluser"
+    ssh_tunnel.password = None
+    ssh_tunnel.private_key = _make_ed25519_pem()
+    ssh_tunnel.private_key_password = None
+
+    with patch("superset.extensions.ssh.sshtunnel.open_tunnel") as mock_open:
+        manager.create_tunnel(
+            ssh_tunnel, "postgresql://user:pass@db.example.com:5432/x"
+        )
+
+    # Key-type-agnostic loader must produce a paramiko PKey usable as ssh_pkey.
+    assert mock_open.called, "open_tunnel was never invoked — key parsing aborted"
+    forwarded_pkey = mock_open.call_args.kwargs["ssh_pkey"]
+    assert forwarded_pkey is not None
+
+
+def test_create_tunnel_accepts_rsa_private_key_unchanged() -> None:
+    """
+    Companion to test_create_tunnel_accepts_ed25519_private_key: pin the
+    backward-compatible RSA path so a fix for ed25519 doesn't regress the
+    historically-supported RSA case. Uses a freshly generated RSA key in
+    OpenSSH format.
+    """
+    from cryptography.hazmat.primitives.asymmetric.rsa import generate_private_key
+
+    rsa_pem = (
+        generate_private_key(public_exponent=65537, key_size=2048)
+        .private_bytes(
+            encoding=Encoding.PEM,
+            format=PrivateFormat.OpenSSH,
+            encryption_algorithm=NoEncryption(),
+        )
+        .decode()
+    )
+
+    app = Mock()
+    app.config = {
+        "SSH_TUNNEL_LOCAL_BIND_ADDRESS": "127.0.0.1",
+        "SSH_TUNNEL_TIMEOUT_SEC": 10.0,
+        "SSH_TUNNEL_PACKET_TIMEOUT_SEC": 10.0,
+    }
+    manager = SSHManager(app)
+
+    ssh_tunnel = Mock()
+    ssh_tunnel.server_address = "ssh.example.com"
+    ssh_tunnel.server_port = 22
+    ssh_tunnel.username = "tunneluser"
+    ssh_tunnel.password = None
+    ssh_tunnel.private_key = rsa_pem
+    ssh_tunnel.private_key_password = None
+
+    with patch("superset.extensions.ssh.sshtunnel.open_tunnel") as mock_open:
+        manager.create_tunnel(
+            ssh_tunnel, "postgresql://user:pass@db.example.com:5432/x"
+        )
+
+    assert mock_open.called
+    assert mock_open.call_args.kwargs["ssh_pkey"] is not None