refactor: Replace paramiko with asyncssh for SSH connections

Switches the SSH implementation from `paramiko` to `asyncssh` for
improved capabilities and performance. Updates logic in SSH tunnel
handling, authentication, and dependencies.

Signed-off-by: Sven Sager <akira@narux.de>

requirements.txt

@@ -7,6 +7,6 @@ wheel
 # Runtime dependencies, must match install_requires in setup.py
 keyring>=23.13.1
 PyQt5>=5.14.1
-paramiko>=2.12.0
+asyncssh>=2.14.0
 revpimodio2>=2.5.6
 zeroconf>=0.24.4

setup.py

@@ -19,7 +19,7 @@ setup(
     install_requires=[
         "keyring",
         "PyQt5",
-        "paramiko",
+        "asyncssh",
         "revpimodio2",
         "zeroconf"
     ],

src/revpicommander/helper.py

@@ -18,7 +18,7 @@ from uuid import uuid4
 from xmlrpc.client import Binary, ServerProxy, Transport
 
 from PyQt5 import QtCore
-from paramiko.ssh_exception import AuthenticationException
+import asyncssh
 
 from . import proginit as pi
 from .ssh_tunneling.server import SSHLocalTunnel
@@ -380,7 +380,7 @@ class ConnectionManager(QtCore.QThread):
                 if getattr(revpi_settings, "ssh_enable_revpipyload", False):
                     ssh_tunnel_server.send_cmd("sudo systemctl enable --now revpipyload")
 
-            except AuthenticationException:
+            except asyncssh.PermissionDenied:
                 self.connect_error.emit(
                     self.tr("Error"), self.tr(
                         "The combination of username and password was rejected from the SSH server.\n\n"

src/revpicommander/proginit.py

@@ -100,8 +100,8 @@ def reconfigure_logger():
         def filter(self, record: logging.LogRecord) -> bool:
             remove_record = False
 
-            # Remove paramiko ssh module
+            # Remove asyncssh module
-            remove_record = remove_record or record.name.startswith("paramiko")
+            remove_record = remove_record or record.name.startswith("asyncssh")
 
             return not remove_record
 

src/revpicommander/ssh_tunneling/server.py

@@ -1,65 +1,21 @@
 # -*- coding: utf-8 -*-
 """
 Connect to a remote host and tunnel a port.
-
-This was crated on base of the paramiko library demo file forward.py, see on
-GitHub https://github.com/paramiko/paramiko/blob/main/demos/forward.py
 """
 __author__ = "Sven Sager"
-__copyright__ = "Copyright (C) 2023 Sven Sager"
+__copyright__ = "Copyright (C) 2023-2026 Sven Sager"
 __license__ = "GPLv2"
 
-import select
+import asyncio
+import threading
 from logging import getLogger
-from socketserver import BaseRequestHandler, ThreadingTCPServer
-from threading import Thread
 from typing import Tuple, Union
 
-from paramiko.client import MissingHostKeyPolicy, SSHClient
+import asyncssh
-from paramiko.rsakey import RSAKey
-from paramiko.ssh_exception import PasswordRequiredException
-from paramiko.transport import Transport
 
 log = getLogger("ssh_tunneling")
 
 
-class ForwardServer(ThreadingTCPServer):
-    daemon_threads = True
-    allow_reuse_address = True
-
-
-class Handler(BaseRequestHandler):
-    def handle(self):
-        try:
-            chan = self.ssh_transport.open_channel(
-                "direct-tcpip",
-                (self.chain_host, self.chain_port),
-                self.request.getpeername(),
-            )
-        except Exception as e:
-            log.error(e)
-            return
-        if chan is None:
-            log.error("Could not create a ssh channel")
-            return
-
-        while True:
-            r, w, x = select.select([self.request, chan], [], [], 5.0)
-            if self.request in r:
-                data = self.request.recv(1024)
-                if len(data) == 0:
-                    break
-                chan.send(data)
-            if chan in r:
-                data = chan.recv(1024)
-                if len(data) == 0:
-                    break
-                self.request.send(data)
-
-        chan.close()
-        self.request.close()
-
-
 class SSHLocalTunnel:
 
     def __init__(self, remote_tunnel_port: int, ssh_host: str, ssh_port: int = 22):
@@ -74,39 +30,66 @@ class SSHLocalTunnel:
         self._ssh_host = ssh_host
         self._ssh_port = ssh_port
 
-        self._th_server = Thread()
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._stop_event: asyncio.Event | None = None
+        self._thread: threading.Thread | None = None
 
-        self._ssh_client = SSHClient()
+        self._started = threading.Event()
-        self._ssh_client.set_missing_host_key_policy(MissingHostKeyPolicy())
+        self._stopped = threading.Event()
 
-        self._ssh_transport = None  # type: Transport
+        self._startup_error: Exception | None = None
-        self._forward_server = None  # type: ThreadingTCPServer
+        self._runtime_error: Exception | None = None
-        self._local_tunnel_port = None  # type: int
 
-    def __th_target(self):
+        self._conn: asyncssh.SSHClientConnection | None = None
-        """Server thread for socket mirror."""
+        self._server: asyncssh.SSHForwarder | None = None
-        self._forward_server.serve_forever()
+        self._local_tunnel_port: int | None = None
 
-    def _configure_forward_server(self) -> int:
+    def _thread_main(self, username, password=None, client_keys=None, passphrase=None):
-        """
+        try:
-        Configure forward server for port mirror.
+            asyncio.run(self._run(username, password, client_keys, passphrase))
+        except Exception as exc:
+            if not self._started.is_set():
+                self._startup_error = exc
+                self._started.set()
+            else:
+                self._runtime_error = exc
+        finally:
+            self._stopped.set()
 
-        :return: Local port on wich the remote port is connected
+    async def _run(self, username, password=None, client_keys=None, passphrase=None):
-        """
+        self._loop = asyncio.get_running_loop()
-        self._ssh_transport = self._ssh_client.get_transport()
+        self._stop_event = asyncio.Event()
 
-        class SubHandler(Handler):
+        try:
-            chain_host = "127.0.0.1"
+            async with asyncssh.connect(
-            chain_port = self._remote_tunnel_port
+                host=self._ssh_host,
-            ssh_transport = self._ssh_transport
+                port=self._ssh_port,
+                username=username,
+                password=password,
+                client_keys=client_keys,
+                passphrase=passphrase,
+                known_hosts=None,  # Analog zu MissingHostKeyPolicy()
+                config=None,  # Do not parse local config
+            ) as conn:
+                self._conn = conn
+                # Forward local port 0 (dynamic) to remote 127.0.0.1:remote_tunnel_port
+                self._server = await conn.forward_local_port(
+                    '127.0.0.1', 0, '127.0.0.1', self._remote_tunnel_port
+                )
+                self._local_tunnel_port = self._server.get_port()
 
-        self._forward_server = ForwardServer(("127.0.0.1", 0), SubHandler)
+                self._started.set()
-        self._local_tunnel_port = self._forward_server.socket.getsockname()[1]
+                await self._stop_event.wait()
 
-        self._th_server = Thread(target=self.__th_target)
+                self._server.close()
-        self._th_server.start()
+                await self._server.wait_closed()
 
-        return self._local_tunnel_port
+        except Exception as exc:
+            if not self._started.is_set():
+                self._startup_error = exc
+                self._started.set()
+                return
+            raise
 
     def connect_by_credentials(self, username: str, password: str) -> int:
         """
@@ -114,17 +97,31 @@ class SSHLocalTunnel:
 
         :return: Local port on wich the remote port is connected
         """
-        if self._th_server.is_alive():
+        if self._thread and self._thread.is_alive():
             raise RuntimeError("Already connected")
 
-        self._ssh_client.connect(
+        self._started.clear()
-            hostname=self._ssh_host,
+        self._stopped.clear()
-            port=self._ssh_port,
+        self._startup_error = None
-            username=username,
+        self._runtime_error = None
-            password=password,
-        )
 
-        return self._configure_forward_server()
+        self._thread = threading.Thread(
+            target=self._thread_main,
+            args=(username, password),
+            daemon=True
+        )
+        self._thread.start()
+
+        if not self._started.wait(20.0):
+            self.disconnect()
+            raise TimeoutError("SSH connection timed out")
+
+        if self._startup_error:
+            error = self._startup_error
+            self.disconnect()
+            raise error
+
+        return self._local_tunnel_port
 
     def connect_by_keyfile(self, username: str, key_file: str, key_password: str = None) -> int:
         """
@@ -132,62 +129,79 @@ class SSHLocalTunnel:
 
         :return: Local port on wich the remote port is connected
         """
-        if self._th_server.is_alive():
+        if self._thread and self._thread.is_alive():
             raise RuntimeError("Already connected")
 
-        if self.key_file_password_protected(key_file):
+        self._started.clear()
-            private_key = RSAKey.from_private_key_file(key_file, key_password)
+        self._stopped.clear()
-        else:
+        self._startup_error = None
-            private_key = RSAKey.from_private_key_file(key_file)
+        self._runtime_error = None
 
-        self._ssh_client.connect(
+        self._thread = threading.Thread(
-            hostname=self._ssh_host,
+            target=self._thread_main,
-            port=self._ssh_port,
+            args=(username, None, [key_file], key_password),
-            username=username,
+            daemon=True
-            pkey=private_key,
-            look_for_keys=True,
         )
+        self._thread.start()
 
-        return self._configure_forward_server()
+        if not self._started.wait(20.0):
+            self.disconnect()
+            raise TimeoutError("SSH connection timed out")
+
+        if self._startup_error:
+            error = self._startup_error
+            self.disconnect()
+            raise error
+
+        return self._local_tunnel_port
 
     def disconnect(self):
         """Close SSH tunnel connection."""
+        if self._loop and self._stop_event:
+            self._loop.call_soon_threadsafe(self._stop_event.set)
+
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=5.0)
+
+        self._conn = None
+        self._server = None
         self._local_tunnel_port = None
-        if self._forward_server:
+        self._thread = None
-            self._forward_server.shutdown()
+        self._loop = None
-            self._forward_server.server_close()
+        self._stop_event = None
-        if self._ssh_transport:
-            self._ssh_transport.close()
-        self._ssh_client.close()
 
     @staticmethod
     def key_file_password_protected(key_file: str) -> bool:
+        # asyncssh doesn't have a direct equivalent without trying to load it.
+        # But we can try to load it with an empty passphrase.
         try:
-            RSAKey.from_private_key_file(key_file)
+            asyncssh.read_private_key(key_file, passphrase=None)
-        except PasswordRequiredException:
-            return True
             return False
+        except asyncssh.KeyImportError:
+            return True
+        except Exception:
+            return True
 
     def send_cmd(self, cmd: str, timeout: float = None) -> Union[Tuple[str, str], Tuple[None, None]]:
         """
         Send simple command to ssh host.
 
-        The output of stdout and stderr is returned as a tuple of two elements.
-        This elements could be None, in case of an internal error.
-
         :param cmd: Shell command to execute on remote host
         :param timeout: Timeout for execution
         :return: Tuple with stdout and stderr
         """
-        if not self._th_server.is_alive():
+        if not self.connected:
             raise RuntimeError("Not connected")
 
-        try:
+        # Running async command from sync context
-            _, stdout, stderr = self._ssh_client.exec_command(cmd, 1024, timeout)
+        async def _exec():
-            buffer_out = stdout.read()
+            result = await self._conn.run(cmd, timeout=timeout)
-            buffer_err = stderr.read()
+            return result.stdout, result.stderr
 
-            return buffer_out.decode(), buffer_err.decode()
+        try:
+            future = asyncio.run_coroutine_threadsafe(_exec(), self._loop)
+            stdout, stderr = future.result(timeout=timeout)
+            return stdout, stderr
         except Exception as e:
             log.error(e)
             return None, None
@@ -195,7 +209,7 @@ class SSHLocalTunnel:
     @property
     def connected(self):
         """Check connection state of ssh tunnel."""
-        return self._ssh_transport and self._ssh_transport.is_active()
+        return self._conn is not None and self._started.is_set() and not self._stopped.is_set()
 
     @property
     def local_tunnel_port(self) -> int: