docker/clonedb/core/ssh.py

"""
An SSH Tunneling tool

@author: olivier.massot, 05-2020
"""
from sshtunnel import SSHTunnelForwarder


class SshTunnel:
    LOCAL_ADRESS = ('127.0.0.1', 6000)

    def __init__(self, host, remote_port, port=22, user="root", key_file="~/.ssh/id_rsa"):
        self.host = host
        self.remote_port = remote_port
        self.port = int(port)
        self.user = user
        self.key_file = key_file

        self._tunnel = SSHTunnelForwarder(
            (self.host, self.port),
            ssh_username=self.user,
            ssh_pkey=self.key_file,
            local_bind_address=self.LOCAL_ADRESS,
            remote_bind_address=('127.0.0.1', self.remote_port)
        )

    def start(self):
        """ Start the ssh tunnel
        """
        self._tunnel.start()
        if not self._tunnel.tunnel_is_up[self.LOCAL_ADRESS]:
            raise RuntimeError('Unable to open the SSH Tunnel')

    def stop(self):
        """ Stop the ssh tunnel
        """
        self._tunnel.stop()