How to maintain a connection on SSH using Paramiko to execute further commands?

Question

I am trying to create a Python application that establish a SSH command and sends a few commands.

At the moment, I am at the step where I just want to open a SSH connection, send a pwd command, and disconnect from the session.

This is the code I did so far:

import paramiko


class Host:
    """This class represents the SSH connection to a host.
    It has attributes related to the remote host being connected and the state of the SSH connection."""

    def __init__(self, remote_host, login, password):
        self.remote_host = remote_host
        self.login = login
        self.password = password
        self.ssh_client = None  # Fix Variable Before Assignment Warning.

    def connect(self):
        try:
            # Try to connect and handle errors - Instantiating an object of the class, not a local object.
            self.ssh_client = paramiko.SSHClient()  # Self instantiates the object, not local.
            self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            # Try for only 1 second.
            self.ssh_client.connect(self.remote_host, username=self.login, password=self.password, timeout=1)

            # Multiple Assignment / Data Unpacking
            # stdin, stdout, stderr = ssh_client.exec_command('pwd')

            # response_stdout = stdout.read().decode().strip()
            return "Connected to SSH!"
        except paramiko.ssh_exception.SSHException as ssh_exception:
            return f"SSHException - Failed to connect to the host: {ssh_exception}"
        except TimeoutError as timeout_error:
            return f"TimeoutError - Host unavailable: {timeout_error}"
        # finally:
            # ssh_client.close()

    def isconnected(self):
        if self.ssh_client:
            return self.ssh_client
        else:
            return None

    def disconnect(self):
        # Use the object attribute that represents the connection state to disconnect.
        if self.ssh_client is not None:
            self.ssh_client.close()
            return "Disconnected from SSH!"
        else:
            return "No active SSH connection."

Now, what I expected was the following flow:

--- Connect to SSH (Retain Connection Open) ----

--- Do something ... Maybe a PWD or something ----

--- Close the SSH Connection ---

But for some reason, when calling:

print("Testing SSH Connection.")
host = Host("192.168.1.10", 'administrator', 'SuperStrongP477')
if host.isconnected():
    print("SSH Connected.")
else:
    print("SSH Not Connected.")

I get:

Testing SSH Connection. SSH Not Connected.

I already looked into other posts but they did not help me, since I did not understand clearly how to adapt it to my code... The answers I looked up and did not help, were:

execute git command in a remote machine using paramiko How to keep ssh session not expired using paramiko? Execute multiple commands in Paramiko so that commands are affected by their predecessors

Those two did clarify some doubts but I still was not able to implement their solutions into my own code.

score 0 · Accepted Answer · answered Jul 24 '23 at 22:21

Solution: Since I did'nt want to use invoke_shell(), here is the method I created - Which receives a list of commands and even accepts elevated ones that requires passwords - Like a initial sudo command.

def enviarcomandos(self, comandos: list, senha=None):
        """ This will send commands to a session, and after completed, will end the current session.
            According to Channel docs at Ref. below, when executing a command, it closes the channel,
            so you must use stdin, stdout and stderr approach to use multiple commands.

            Sintaxe:    1. Criar a lista com os comandos.
                        2. Enviar com remoteclient.enviarcomandos(comandos)

            Ref. https://docs.paramiko.org/en/stable/api/channel.html#paramiko.channel.Channel.get_name"""
        contagem = 0
        try:
            # Crio a sessao pra enviar meus comandos.
            self.createsession()

            # Ref. https://docs.paramiko.org/en/stable/api/channel.html#paramiko.channel.Channel.settimeout
            self.ssh_channel.settimeout(1)  # Timeout for writing/sending bytes.

            # Espero um pouco antes de conectar conectar.
            while True:
                print(".", end="")
                self.esperar(100 / 1000)  # 100 Milisegundos
                contagem += 1

                if contagem == 5:
                    break

            print()

            # Entro no Channel que criamos anteriormente e envio os comandos.
            outputcomandos = []
            for command in comandos:
                if "sudo" in command and senha is not None:
                    # exec_command() method closes all channels (stdin, stdout, stderr) when execution is complete.
                    stdin, stdout, stderr = self.ssh_client.exec_command("sudo -S %s" % command)

                    # If stdout is still open then sudo is asking us for a password
                    if stdout.channel.closed is False:
                        stdin.write('%s\n' % senha)
                        stdin.flush()

                else:
                    print(f"{'#' * 5} Executing command : {command} {'#' * 5}")
                    self.ssh_channel = self.ssh_transport.open_session()
                    print(f"{'#' * 5} Tentando Destrinchar os STDIN, STDOUT e STDERR do Exec Command... {'#' * 5}")

                    self.ssh_channel.exec_command(command)
                    print(f"{'#' * 5} Passei do Channel Exec Command {'#' * 5}")

                    print(f"{'#' * 5} Passei do If - Fora dele {'#' * 5}")

                    # https://docs.paramiko.org/en/stable/api/channel.html#paramiko.channel.Channel.recv_exit_status
                    return_code = self.ssh_channel.recv_exit_status()  # Exit code from SSH Server.

                    stdout = self.ssh_channel.makefile('rb').read()
                    stderr = self.ssh_channel.makefile_stderr('rb').read()

                    print(f"{'#' * 5} Passei do Return Code e dos STD Out e Err {'#' * 5}")

                    # Adicionar as informacoes do comando ao meu dict.
                    comandoatual = {
                        "Comando": command,
                        "Codigo de Retorno": return_code,
                        "Output": stdout.decode(),
                        "Erro": stderr.decode()
                    }

                    outputcomandos.append(comandoatual)

            # Depois de executar meus comandos, fecho tudo
            self.ssh_client.close()

            print(f"Output: {outputcomandos}")
            return outputcomandos

        except Exception as e:
            print(f"Channel Send Commands Error => {e}")
            # If something bad happens.
            return False

The code intercalates between English/Portuguese variable names and comments, but basically, what is it doing is:

For each command on my list, I open a new session on Transport (Because after each command, Paramiko by default closes the Channel) and send the command.

It also checks if stdout is still out after one command, and if it is, it means it is asking for a password.

That is it.

How to maintain a connection on SSH using Paramiko to execute further commands?

1 Answers1