{"id":905,"date":"2013-02-24T00:19:13","date_gmt":"2013-02-24T00:19:13","guid":{"rendered":"http:\/\/joelinoff.com\/blog\/?p=905"},"modified":"2026-06-28T17:38:11","modified_gmt":"2026-06-29T00:38:11","slug":"remote-command-execution-in-python-using-paramiko-that-supports-arbitrary-input","status":"publish","type":"post","link":"https:\/\/joelinoff.com\/blog\/?p=905","title":{"rendered":"Remote command execution in python using paramiko that supports arbitrary input"},"content":{"rendered":"I recently decided to use paramiko to develop a remote command execution tool. \n\nIt was very easy to setup initially and ran much faster than my existing pexpect implementation but it had a problem with sudo commands because they required the password to be provided as input. \n\nI solved the problem by using a pseudo-terminal and by creating my own ChannelFile objects for stdin and stdout\/stderr. The solution should be general enough to handle any case that requires simple input but it is not as flexible as pexpect. I hope that you find it useful.\n<!--more-->\n<h1>Current Version<\/h1>\nThis is similar to the older version below but it handles more than 64KiB by eliminating the stdout buffer, the stdin buffer and using recv polling. It even does a silly check for input. That check could be expanded to make it behave more like pexpect but that is for another day.\n\n<pre class=\"wp-block-code language-python\"><code class=\"language-python\">#!\/usr\/bin\/env python\n'''\nThis class allows you to run commands on a remote host and provide\ninput if necessary.\n\nVERSION 1.2\n'''\nimport paramiko\nimport logging\nimport socket\nimport time\nimport datetime\n\n\n# ================================================================\n# class MySSH\n# ================================================================\nclass MySSH:\n    '''\n    Create an SSH connection to a server and execute commands.\n    Here is a typical usage:\n\n        ssh = MySSH()\n        ssh.connect('host', 'user', 'password', port=22)\n        if ssh.connected() is False:\n            sys.exit('Connection failed')\n\n        # Run a command that does not require input.\n        status, output = ssh.run('uname -a')\n        print 'status = %d' % (status)\n        print 'output (%d):' % (len(output))\n        print '%s' % (output)\n\n        # Run a command that does requires input.\n        status, output = ssh.run('sudo uname -a', 'sudo-password')\n        print 'status = %d' % (status)\n        print 'output (%d):' % (len(output))\n        print '%s' % (output)\n    '''\n    def __init__(self, compress=True, verbose=False):\n        '''\n        Setup the initial verbosity level and the logger.\n\n        @param compress  Enable\/disable compression.\n        @param verbose   Enable\/disable verbose messages.\n        '''\n        self.ssh = None\n        self.transport = None\n        self.compress = compress\n        self.bufsize = 65536\n\n        # Setup the logger\n        self.logger = logging.getLogger('MySSH')\n        self.set_verbosity(verbose)\n\n        fmt = '%(asctime)s MySSH:%(funcName)s:%(lineno)d %(message)s'\n        format = logging.Formatter(fmt)\n        handler = logging.StreamHandler()\n        handler.setFormatter(format)\n        self.logger.addHandler(handler)\n        self.info = self.logger.info\n\n    def __del__(self):\n        if self.transport is not None:\n            self.transport.close()\n            self.transport = None\n\n    def connect(self, hostname, username, password, port=22):\n        '''\n        Connect to the host.\n\n        @param hostname  The hostname.\n        @param username  The username.\n        @param password  The password.\n        @param port      The port (default=22).\n\n        @returns True if the connection succeeded or false otherwise.\n        '''\n        self.info('connecting %s@%s:%d' % (username, hostname, port))\n        self.hostname = hostname\n        self.username = username\n        self.port = port\n        self.ssh = paramiko.SSHClient()\n        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())\n        try:\n            self.ssh.connect(hostname=hostname,\n                             port=port,\n                             username=username,\n                             password=password)\n            self.transport = self.ssh.get_transport()\n            self.transport.use_compression(self.compress)\n            self.info('succeeded: %s@%s:%d' % (username,\n                                               hostname,\n                                               port))\n        except socket.error as e:\n            self.transport = None\n            self.info('failed: %s@%s:%d: %s' % (username,\n                                                hostname,\n                                                port,\n                                                str(e)))\n        except paramiko.BadAuthenticationType as e:\n            self.transport = None\n            self.info('failed: %s@%s:%d: %s' % (username,\n                                                hostname,\n                                                port,\n                                                str(e)))\n\n        return self.transport is not None\n\n    def run(self, cmd, input_data=None, timeout=10):\n        '''\n        Run a command with optional input data.\n\n        Here is an example that shows how to run commands with no input:\n\n            ssh = MySSH()\n            ssh.connect('host', 'user', 'password')\n            status, output = ssh.run('uname -a')\n            status, output = ssh.run('uptime')\n\n        Here is an example that shows how to run commands that require input:\n\n            ssh = MySSH()\n            ssh.connect('host', 'user', 'password')\n            status, output = ssh.run('sudo uname -a', '&lt;sudo-password&gt;')\n\n        @param cmd         The command to run.\n        @param input_data  The input data (default is None).\n        @param timeout     The timeout in seconds (default is 10 seconds).\n        @returns The status and the output (stdout and stderr combined).\n        '''\n        self.info('running command: (%d) %s' % (timeout, cmd))\n\n        if self.transport is None:\n            self.info('no connection to %s@%s:%s' % (str(self.username),\n                                                     str(self.hostname),\n                                                     str(self.port)))\n            return -1, 'ERROR: connection not established\\n'\n\n        # Fix the input data.\n        input_data = self._run_fix_input_data(input_data)\n\n        # Initialize the session.\n        self.info('initializing the session')\n        session = self.transport.open_session()\n        session.set_combine_stderr(True)\n        session.get_pty()\n        session.exec_command(cmd)\n        output = self._run_poll(session, timeout, input_data)\n        status = session.recv_exit_status()\n        self.info('output size %d' % (len(output)))\n        self.info('status %d' % (status))\n        return status, output\n\n    def connected(self):\n        '''\n        Am I connected to a host?\n\n        @returns True if connected or false otherwise.\n        '''\n        return self.transport is not None\n\n    def set_verbosity(self, verbose):\n        '''\n        Turn verbose messages on or off.\n\n        @param verbose  Enable\/disable verbose messages.\n        '''\n        if verbose &gt; 0:\n            self.logger.setLevel(logging.INFO)\n        else:\n            self.logger.setLevel(logging.ERROR)\n\n    def _run_fix_input_data(self, input_data):\n        '''\n        Fix the input data supplied by the user for a command.\n\n        @param input_data  The input data (default is None).\n        @returns the fixed input data.\n        '''\n        if input_data is not None:\n            if len(input_data) &gt; 0:\n                if '\\\\n' in input_data:\n                    # Convert \\n in the input into new lines.\n                    lines = input_data.split('\\\\n')\n                    input_data = '\\n'.join(lines)\n            return input_data.split('\\n')\n        return []\n\n    def _run_send_input(self, session, stdin, input_data):\n        '''\n        Send the input data.\n\n        @param session     The session.\n        @param stdin       The stdin stream for the session.\n        @param input_data  The input data (default is None).\n        '''\n        if input_data is not None:\n            self.info('session.exit_status_ready() %s' % str(session.exit_status_ready()))\n            self.info('stdin.channel.closed %s' % str(stdin.channel.closed))\n            if stdin.channel.closed is False:\n                self.info('sending input data')\n                stdin.write(input_data)\n\n    def _run_poll(self, session, timeout, input_data):\n        '''\n        Poll until the command completes.\n\n        @param session     The session.\n        @param timeout     The timeout in seconds.\n        @param input_data  The input data.\n        @returns the output\n        '''\n        interval = 0.1\n        maxseconds = timeout\n        maxcount = maxseconds \/ interval\n\n        # Poll until completion or timeout\n        # Note that we cannot directly use the stdout file descriptor\n        # because it stalls at 64K bytes (65536).\n        input_idx = 0\n        timeout_flag = False\n        self.info('polling (%d, %d)' % (maxseconds, maxcount))\n        start = datetime.datetime.now()\n        start_secs = time.mktime(start.timetuple())\n        output = ''\n        session.setblocking(0)\n        while True:\n            if session.recv_ready():\n                data = session.recv(self.bufsize)\n                output += data\n                self.info('read %d bytes, total %d' % (len(data), len(output)))\n\n                if session.send_ready():\n                    # We received a potential prompt.\n                    # In the future this could be made to work more like\n                    # pexpect with pattern matching.\n                    if input_idx &lt; len(input_data):\n                        data = input_data[input_idx] + '\\n'\n                        input_idx += 1\n                        self.info('sending input data %d' % (len(data)))\n                        session.send(data)\n\n            self.info('session.exit_status_ready() = %s' % (str(session.exit_status_ready())))\n            if session.exit_status_ready():\n                break\n\n            # Timeout check\n            now = datetime.datetime.now()\n            now_secs = time.mktime(now.timetuple()) \n            et_secs = now_secs - start_secs\n            self.info('timeout check %d %d' % (et_secs, maxseconds))\n            if et_secs &gt; maxseconds:\n                self.info('polling finished - timeout')\n                timeout_flag = True\n                break\n            time.sleep(0.200)\n\n        self.info('polling loop ended')\n        if session.recv_ready():\n            data = session.recv(self.bufsize)\n            output += data\n            self.info('read %d bytes, total %d' % (len(data), len(output)))\n\n        self.info('polling finished - %d output bytes' % (len(output)))\n        if timeout_flag:\n            self.info('appending timeout message')\n            output += '\\nERROR: timeout after %d seconds\\n' % (timeout)\n            session.close()\n\n        return output\n\n\n# ================================================================\n# MAIN\n# ================================================================\nif __name__ == '__main__':\n    import sys\n\n    # Access variables.\n    hostname = 'hostname'\n    port = 22\n    username = 'username'\n    password = 'password'\n    sudo_password = password  # assume that it is the same password\n\n    # Create the SSH connection\n    ssh = MySSH()\n    ssh.set_verbosity(False)\n    ssh.connect(hostname=hostname,\n                username=username,\n                password=password,\n                port=port)\n    if ssh.connected() is False:\n        print 'ERROR: connection failed.'\n        sys.exit(1)\n\n    def run_cmd(cmd, indata=None):\n        '''\n        Run a command with optional input.\n\n        @param cmd    The command to execute.\n        @param indata The input data.\n        @returns The command exit status and output.\n                 Stdout and stderr are combined.\n        '''\n        print\n        print '=' * 64\n        print 'command: %s' % (cmd)\n        status, output = ssh.run(cmd, indata)\n        print 'status : %d' % (status)\n        print 'output : %d bytes' % (len(output))\n        print '=' * 64\n        print '%s' % (output)\n\n    run_cmd('uname -a')\n    run_cmd('sudo ls -ltrh \/var\/log | tail', sudo_password)  # sudo command<\/code><\/pre>\n\n\n<h1>Older Version<\/h1>\nThis version will not handle more than 64KiB of output. I am not sure why.\n\n<pre class=\"wp-block-code language-python\"><code class=\"language-python\">#!\/usr\/bin\/env python\n'''\nThis class allows you to run commands on a remote host and provide\ninput if necessary.\n'''\nimport paramiko\nimport logging\nimport socket\nimport time\n\n\n# ================================================================\n# class MySSH\n# ================================================================\nclass MySSH:\n    '''\n    Create an SSH connection to a server and execute commands.\n    Here is a typical usage:\n\n        ssh = MySSH()\n        ssh.connect('host', 'user', 'password', port=22)\n        if ssh.connected() is False:\n            sys.exit('Connection failed')\n\n        # Run a command that does not require input.\n        status, output = ssh.run('uname -a')\n        print 'status = %d' % (status)\n        print 'output (%d):' % (len(output))\n        print '%s' % (output)\n\n        # Run a command that does requires input.\n        status, output = ssh.run('sudo uname -a', 'sudo-password')\n        print 'status = %d' % (status)\n        print 'output (%d):' % (len(output))\n        print '%s' % (output)\n    '''\n    def __init__(self, compress=True, verbose=False):\n        '''\n        Setup the initial verbosity level and the logger.\n\n        @param compress  Enable\/disable compression.\n        @param verbose   Enable\/disable verbose messages.\n        '''\n        self.ssh = None\n        self.transport = None\n        self.compress = compress\n\n        # Setup the logger\n        self.logger = logging.getLogger('MySSH')\n        self.set_verbosity(verbose)\n\n        fmt = '%(asctime)s MySSH:%(funcName)s:%(lineno)d %(message)s'\n        format = logging.Formatter(fmt)\n        handler = logging.StreamHandler()\n        handler.setFormatter(format)\n        self.logger.addHandler(handler)\n        self.info = self.logger.info\n\n    def __del__(self):\n        if self.transport is not None:\n            self.transport.close()\n            self.transport = None\n\n    def connect(self, hostname, username, password, port=22):\n        '''\n        Connect to the host.\n\n        @param hostname  The hostname.\n        @param username  The username.\n        @param password  The password.\n        @param port      The port (default=22).\n\n        @returns True if the connection succeeded or false otherwise.\n        '''\n        self.info('connecting %s@%s:%d' % (username, hostname, port))\n        self.hostname = hostname\n        self.username = username\n        self.port = port\n        self.ssh = paramiko.SSHClient()\n        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())\n        try:\n            self.ssh.connect(hostname=hostname,\n                             port=port,\n                             username=username,\n                             password=password)\n            self.transport = self.ssh.get_transport()\n            self.transport.use_compression(self.compress)\n            self.info('succeeded: %s@%s:%d' % (username,\n                                               hostname,\n                                               port))\n        except socket.error:\n            self.transport = None\n            self.info('failed: %s@%s:%d: %s' % (username,\n                                                hostname,\n                                                port,\n                                                str(e)))\n        except paramiko.BadAuthenticationType as e:\n            self.transport = None\n            self.info('failed: %s@%s:%d: %s' % (username,\n                                                hostname,\n                                                port,\n                                                str(e)))\n\n        return self.transport is not None\n\n    def run(self, cmd, input_data=None, timeout=10):\n        '''\n        Run a command with optional input data.\n\n        Here is an example that shows how to run commands with no input:\n\n            ssh = MySSH()\n            ssh.connect('host', 'user', 'password')\n            status, output = ssh.run('uname -a')\n            status, output = ssh.run('uptime')\n\n        Here is an example that shows how to run commands that require input:\n\n            ssh = MySSH()\n            ssh.connect('host', 'user', 'password')\n            status, output = ssh.run('sudo uname -a', '&lt;sudo-password&gt;')\n\n        @param cmd         The command to run.\n        @param input_data  The input data (default is None).\n        @param timeout     The timeout in seconds (default is 10 seconds).\n        @returns The status and the output (stdout and stderr combined).\n        '''\n        self.info('running command: (%d) %s' % (timeout, cmd))\n\n        if self.transport is None:\n            self.info('no connection to %s@%s:%s' % (str(self.username),\n                                                     str(self.hostname),\n                                                     str(self.port)))\n            return -1, 'ERROR: connection not established\\n'\n\n        # Fix the input data.\n        input_data = self._run_fix_input_data(input_data)\n\n        # Initialize the session.\n        self.info('initializing the session')\n        session = self.transport.open_session()\n        session.set_combine_stderr(True)\n        session.get_pty()\n        session.exec_command(cmd)\n        stdin = session.makefile('wb', -1)\n        stdout = session.makefile('rb', -1)\n\n        self._run_send_input(stdout, stdin, input_data)\n        output = self._run_poll(stdout, timeout, input_data)\n        status = stdout.channel.recv_exit_status()\n        self.info('output size %d' % (len(output)))\n        self.info('status %d' % (status))\n        return status, output\n\n    def connected(self):\n        '''\n        Am I connected to a host?\n\n        @returns True if connected or false otherwise.\n        '''\n        return self.transport is not None\n\n    def set_verbosity(self, verbose):\n        '''\n        Turn verbose messages on or off.\n\n        @param verbose  Enable\/disable verbose messages.\n        '''\n        if verbose is True:\n            self.logger.setLevel(logging.INFO)\n        else:\n            self.logger.setLevel(logging.ERROR)\n\n    def _run_fix_input_data(self, input_data):\n        '''\n        Fix the input data supplied by the user for a command.\n\n        @param input_data  The input data (default is None).\n        @returns the fixed input data.\n        '''\n        if input_data is not None:\n            if len(input_data) &gt; 0:\n                if '\\\\n' in input_data:\n                    # Convert \\n in the input into new lines.\n                    lines = input_data.split('\\\\n')\n                    input_data = '\\n'.join(lines)\n                if input_data[-1] != '\\n':\n                    input_data += '\\n'\n        return input_data\n\n    def _run_send_input(self, stdout, stdin, input_data):\n        '''\n        Send the input data.\n\n        @param stdout      The stdout stream for the session.\n        @param stdin       The stdin stream for the session.\n        @param input_data  The input data (default is None).\n        '''\n        if  stdout.channel.closed is False:\n            if input_data is not None:\n                self.info('sending input data')\n                stdin.write(input_data)\n\n    def _run_poll(self, stdout, timeout, input_data):\n        '''\n        Poll until the command completes.\n\n        @param timeout     The timeout in seconds.\n        @param input_data  The input data.\n        @returns the output\n        '''\n        interval = 0.1\n        maxseconds = timeout\n        maxcount = maxseconds \/ interval\n\n        # Poll until completion or timeout\n        self.info('polling (%d, %d)' % (maxseconds, maxcount))\n        count = 0\n        while stdout.channel.closed is False and count &lt;= maxcount:\n            count += 1\n            time.sleep(interval)\n\n        # Polling finished.\n        if stdout.channel.closed is False:\n            # Some sort of error occurred, assume a timeout.\n            self.info('polling finished - timeout')\n            stdout.channel.close()\n            output = stdout.read()\n            output += '\\nERROR: timeout after %d seconds\\n' % (timeout)\n        else:\n            output = stdout.read()\n            self.info('polling finished - %d output bytes' % (len(output)))\n\n        if input_data is not None:\n            # Strip out the input data.\n            output = output[len(input_data):]\n            self.info('stripped %d input bytes' % (len(input_data)))\n\n        return output\n\n\n# ================================================================\n# MAIN\n# ================================================================\nif __name__ == '__main__':\n    import sys\n\n    # Access variables.\n    hostname = 'hostname'\n    port = 22\n    username = 'username'\n    password = 'password'\n    sudo_password = password  # assume that it is the same password\n\n    # Create the SSH connection\n    ssh = MySSH()\n    ssh.set_verbosity(False)\n    ssh.connect(hostname=hostname,\n                username=username,\n                password=password,\n                port=port)\n    if ssh.connected() is False:\n        print 'ERROR: connection failed.'\n        sys.exit(1)\n\n    def run_cmd(cmd, indata=None):\n        '''\n        Run a command with optional input.\n\n        @param cmd    The command to execute.\n        @param indata The input data.\n        @returns The command exit status and output.\n                 Stdout and stderr are combined.\n        '''\n        print\n        print '=' * 64\n        print 'command: %s' % (cmd)\n        status, output = ssh.run(cmd, indata)\n        print 'status : %d' % (status)\n        print 'output : %d bytes' % (len(output))\n        print '=' * 64\n        print '%s' % (output)\n\n    run_cmd('uname -a')\n    run_cmd('sudo ls -ltrh \/var\/log | tail', sudo_password)  # sudo command<\/code><\/pre>\n\n\n","protected":false},"excerpt":{"rendered":"<p>I recently decided to use paramiko to develop a remote command execution tool. It was very easy to setup initially and ran much faster than my existing pexpect implementation but it had a problem with sudo commands because they required the password to be provided as input. I solved the problem by using a pseudo-terminal &hellip; <a href=\"https:\/\/joelinoff.com\/blog\/?p=905\" class=\"more-link\">Continue reading <span class=\"screen-reader-text\">Remote command execution in python using paramiko that supports arbitrary input<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_monsterinsights_skip_tracking":false,"_monsterinsights_sitenote_active":false,"_monsterinsights_sitenote_note":"","_monsterinsights_sitenote_category":0,"footnotes":""},"categories":[5,7,16],"tags":[],"class_list":["post-905","post","type-post","status-publish","format-standard","hentry","category-programming","category-python","category-sysadmin"],"_links":{"self":[{"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=\/wp\/v2\/posts\/905","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=905"}],"version-history":[{"count":13,"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=\/wp\/v2\/posts\/905\/revisions"}],"predecessor-version":[{"id":1782,"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=\/wp\/v2\/posts\/905\/revisions\/1782"}],"wp:attachment":[{"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=905"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=905"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/joelinoff.com\/blog\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=905"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}