paramiko ssh sudo and scp

SSH.py

import logging
from paramiko import SSHClient, AutoAddPolicy
from scp import SCPClient


class SshClient:
    def __init__(self, ssh_machine, ssh_username, ssh_password):
        self.ssh_machine = ssh_machine
        self.ssh_username = ssh_username
        self.ssh_password = ssh_password
        self.conn = self.get_ssh_connection()
        self.logger = logging.getLogger()
        self.logger.setLevel(logging.INFO)
        self.scp = SCPClient(self.conn.get_transport())

    def close(self):
        self.conn.close()
        self.scp.close()

    def get_ssh_connection(self):
        """Establishes a ssh connection to execute command.

          returns connection Object
          """
        client = SSHClient()
        client.set_missing_host_key_policy(AutoAddPolicy())
        client.connect(hostname=self.ssh_machine, username=self.ssh_username, password=self.ssh_password, timeout=10)

        return client

    def run_sudo_command(self, command="ls", jobid="None", ignore_exit_status=False):
        """Executes a command over a established SSH connectio.
        :param command: Command or commands.
        :param jobid: Adds jobid to logging.
        :param ignore_exit_status: Ignore exit status for example on a reboot of a machine.
        returns status of the command executed and Output of the command.
        """
        # conn = self.get_ssh_connection(ssh_machine=ssh_machine, ssh_username=ssh_username, ssh_password=ssh_password)
        logging.info("Job[%s]: Executing: %s" % (jobid, command))
        # command = "sudo -S -p '' %s" % command
        command = join_sudo_commands(command)

        stdin, stdout, stderr = self.conn.exec_command(command=command)
        stdin.write(self.ssh_password + "\n")
        stdin.flush()

        stdoutput = [line for line in stdout]
        stderroutput = [line for line in stderr]
        for output in stdoutput:
            self.logger.info("Job[%s]: %s" % (jobid, output.strip()))
        # Check exit code.
        self.logger.debug("Job[%s]:stdout: %s" % (jobid, stdoutput))
        self.logger.debug("Job[%s]:stderror: %s" % (jobid, stderroutput))

        if not ignore_exit_status:
            self.logger.info("Job[%s]: Command status: %s" % (jobid, stdout.channel.recv_exit_status()))

        if not stdout.channel.recv_exit_status() or ignore_exit_status:
            self.logger.info("Job[%s]: Command executed." % jobid)

            if not stdoutput:
                stdoutput = True
            return True, stdoutput
        else:
            self.logger.error("Job[%s]: Command failed." % jobid)
            for output in stderroutput:
                self.logger.error("Job[%s]: %s" % (jobid, output))

            return False, stderroutput


def join_sudo_commands(commands_input: str):
    """ Joins sudo and non sudo commands in one string that exec_command can take as input
    :param commands_input: a command(s) separated by ;
    :return: a joined string
    """
    commands = commands_input.split(';')
    sudo_prompt_added = False
    output_list = []

    for cmd in commands:
        if "sudo" in cmd and not sudo_prompt_added:
            command = "sudo -S -p '' %s" % cmd
            output_list.append(command)
            sudo_prompt_added = True
        else:
            output_list.append(cmd)

    return ';'.join(output_list)

This was useful for setting up some DevOps stuff we needed, thanks for sharing.