diff --git a/requirements.txt b/requirements.txt index 383199c..6d0be75 100644 --- a/requirements.txt +++ b/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 diff --git a/setup.py b/setup.py index 227874b..3b29520 100644 --- a/setup.py +++ b/setup.py @@ -19,7 +19,7 @@ setup( install_requires=[ "keyring", "PyQt5", - "paramiko", + "asyncssh", "revpimodio2", "zeroconf" ], diff --git a/src/revpicommander/helper.py b/src/revpicommander/helper.py index d8d30a0..a2e674c 100644 --- a/src/revpicommander/helper.py +++ b/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" diff --git a/src/revpicommander/proginit.py b/src/revpicommander/proginit.py index e45f8da..4101308 100644 --- a/src/revpicommander/proginit.py +++ b/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_record = remove_record or record.name.startswith("paramiko") + # Remove asyncssh module + remove_record = remove_record or record.name.startswith("asyncssh") return not remove_record diff --git a/src/revpicommander/ssh_tunneling/server.py b/src/revpicommander/ssh_tunneling/server.py index a2b6542..691a46d 100644 --- a/src/revpicommander/ssh_tunneling/server.py +++ b/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 -from paramiko.rsakey import RSAKey -from paramiko.ssh_exception import PasswordRequiredException -from paramiko.transport import Transport +import asyncssh 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._ssh_client.set_missing_host_key_policy(MissingHostKeyPolicy()) + self._started = threading.Event() + self._stopped = threading.Event() - self._ssh_transport = None # type: Transport - self._forward_server = None # type: ThreadingTCPServer - self._local_tunnel_port = None # type: int + self._startup_error: Exception | None = None + self._runtime_error: Exception | None = None - def __th_target(self): - """Server thread for socket mirror.""" - self._forward_server.serve_forever() + self._conn: asyncssh.SSHClientConnection | None = None + self._server: asyncssh.SSHForwarder | None = None + self._local_tunnel_port: int | None = None - def _configure_forward_server(self) -> int: - """ - Configure forward server for port mirror. + def _thread_main(self, username, password=None, client_keys=None, passphrase=None): + try: + 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 - """ - self._ssh_transport = self._ssh_client.get_transport() + async def _run(self, username, password=None, client_keys=None, passphrase=None): + self._loop = asyncio.get_running_loop() + self._stop_event = asyncio.Event() - class SubHandler(Handler): - chain_host = "127.0.0.1" - chain_port = self._remote_tunnel_port - ssh_transport = self._ssh_transport + try: + async with asyncssh.connect( + host=self._ssh_host, + 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._local_tunnel_port = self._forward_server.socket.getsockname()[1] + self._started.set() + await self._stop_event.wait() - self._th_server = Thread(target=self.__th_target) - self._th_server.start() + self._server.close() + 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( - hostname=self._ssh_host, - port=self._ssh_port, - username=username, - password=password, - ) + self._started.clear() + self._stopped.clear() + self._startup_error = None + self._runtime_error = None - 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): - private_key = RSAKey.from_private_key_file(key_file, key_password) - else: - private_key = RSAKey.from_private_key_file(key_file) + self._started.clear() + self._stopped.clear() + self._startup_error = None + self._runtime_error = None - self._ssh_client.connect( - hostname=self._ssh_host, - port=self._ssh_port, - username=username, - pkey=private_key, - look_for_keys=True, + self._thread = threading.Thread( + target=self._thread_main, + args=(username, None, [key_file], key_password), + daemon=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._forward_server.shutdown() - self._forward_server.server_close() - if self._ssh_transport: - self._ssh_transport.close() - self._ssh_client.close() + self._thread = None + self._loop = None + self._stop_event = None @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) - except PasswordRequiredException: + asyncssh.read_private_key(key_file, passphrase=None) + return False + except asyncssh.KeyImportError: + return True + except Exception: return True - return False 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: - _, stdout, stderr = self._ssh_client.exec_command(cmd, 1024, timeout) - buffer_out = stdout.read() - buffer_err = stderr.read() + # Running async command from sync context + async def _exec(): + result = await self._conn.run(cmd, timeout=timeout) + 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: