Running an interactive shell with paramiko

If you want to run several commands on the remote host using paramiko, but you encounter the problem that the ssh session is closed when you execute a command, give an exception related with SSH session not active:

paramiko.ssh_exception.SSHException: SSH session not active

To solve this, you can implement an interactive shell using paramiko; that way, the channel does not close after a command is executed in the remote shell.

After creating the SSH client, using connect, you can use the invoke_shell() method, and it will open a channel that it doesn't close after you send something through it.

You can find the following code in the ssh_interactive_shell.py file:

#!/usr/bin/env python3

import paramiko
import re

class ShellHandler:
def __init__(self, host, user, psw):
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(host, username=user, password=psw, port=22)
# we use this method for getting a shell in the host
channel = self.ssh.invoke_shell()
self.stdin = channel.makefile('wb')
self.stdout = channel.makefile('r')

def __del__(self):
self.ssh.close()

@staticmethod
def _print_exec_out(cmd, out_buf, err_buf, exit_status):
print('command executed: {}'.format(cmd))
print('STDOUT:')
for line in out_buf:
print(line, end="")
print('end of STDOUT')
print('STDERR:')
for line in err_buf:
print(line, end="")
print('end of STDERR')
print('finished with exit status: {}'.format(exit_status))
print('------------------------------------')
pass

In the previous code block, we declared the ShellHandler class with the init method constructor and static method to print the output of the executed command. We continue declaring the method to execute a command that's passed as a parameter, as well as our main program that instantiates an object of this class. For each command available in command list, it calls the execute method:

    def execute(self, cmd):
""":param cmd: the command to be executed on the remote computer
:examples: execute('ls')
execute('finger')
execute('cd folder_name')
"""
cmd = cmd.strip(' ')
self.stdin.write(cmd + ' ')
finish = 'end of stdOUT buffer. finished with exit status'
echo_cmd = 'echo {} $?'.format(finish)
self.stdin.write(echo_cmd + ' ')
shin = self.stdin
self.stdin.flush()

shout = []
sherr = []
exit_status = 0

for line in self.stdout:
if str(line).startswith(cmd) or str(line).startswith(echo_cmd):
# up for now filled with shell junk from stdin
shout = []
elif str(line).startswith(finish):
# our finish command ends with the exit status
exit_status = int(str(line).rsplit(maxsplit=1)[1])
if exit_status:
# stderr is combined with stdout.
# thus, swap sherr with shout in a case of failure.
sherr = shout
shout = []
break
else:
# get rid of 'coloring and formatting' special characters
shout.append(re.compile(r'(x9B|x1B[)[0-?]*[ -/]*[@-~]').sub('', line).replace('', '').replace(' ', '').replace(' ', ''))

# first and last lines of shout/sherr contain a prompt
if shout and echo_cmd in shout[-1]:
shout.pop()
if shout and cmd in shout[0]:
shout.pop(0)
if sherr and echo_cmd in sherr[-1]:
sherr.pop()
if sherr and cmd in sherr[0]:
sherr.pop(0)

self._print_exec_out(cmd=cmd, out_buf=shout, err_buf=sherr,exit_status=exit_status)
return shin, shout, sherr

commands = ["ls","whoami","pwd"]
host="localhost"
name="user"
pwd="password"

shell_connection = ShellHandler(host,name,pwd)
for command in commands:
shell_connection.execute(command)
..................Content has been hidden....................

You can't read the all page of ebook, please click here login for view all page.
Reset