Recently, as part of an automated test, we needed to SSH into a server, toggle a service and then check the response on a web application. We used the Python module Paramiko. We ended up writing simple wrappers around the most common actions test automation may need to perform over SSH. In this post, we will share the same.
We have written this tutorial for absolute beginners. If you are already familiar with SSH (there are people who are not!) and comfortable with Python, please jump ahead to the section ‘Putting it all together’.
Note 1: We have also integrated this code into our open-sourced Python test automation framework.
Note 2: The Python 3 code for this article is here
How SSH Works
The SSH connection is implemented using a client-server model. To establish a connection the server should be running and clients generally authenticated either using passwords or SSH keys. Password Authentication is simple and straightforward. To authenticate using SSH keys, a user must have an SSH key pair (Public and Private key). On the remote server, the list of public keys is maintained (usually) in the ~/.ssh/authorized_keys directory. When the client connects to the remote server using the public key, the server checks for it and sends an encrypted message which can only be decrypted with the associated private key at the client side. We will be using a Python module called Paramiko. The Paramiko module gives an abstraction of the SSHv2 protocol with both the client side and server side functionality. As a client, you can authenticate yourself using a password or key and as a server, you can decide which users are allowed access and the channels you allow.
Installing Paramiko
Installing Paramiko is straightforward, it has only one direct hard dependency: the Cryptography library. If pip is above 8.x, nothing else is required. pip will install statically compiled binary archives of Cryptography its dependencies. For more details on installation, please refer to this Paramiko’s documentation.
To install Paramiko on Windows
pip install paramiko |
Overview of the Parmiko tutorial
In the coming sections, we will be talking about below items:
We have written different methods for each functionality. All the parameters are read from a configuration file. We are reading the parameters like credentials, commands & file paths from the conf file because when there is any change in the machine or any credential changes, changing the configuration file alone is enough instead of changing the script.
After importing Paramiko, to make everything more modular, we created a class called Ssh_Util and wrote different methods for each functionality.
Connecting to the remote server using Paramiko
Firstly, we created an initialization function inside the class and initialized required variables. Some of the parameters like host, username, password etc., are read from a configuration file.
def __init__(self): self.ssh_output = None self.ssh_error = None self.client = None self.host= conf_file.HOST self.username = conf_file.USERNAME self.password = conf_file.PASSWORD self.timeout = float(conf_file.TIMEOUT) self.commands = conf_file.COMMANDS self.pkey = conf_file.PKEY self.port = conf_file.PORT self.uploadremotefilepath = conf_file.UPLOADREMOTEFILEPATH self.uploadlocalfilepath = conf_file.UPLOADLOCALFILEPATH self.downloadremotefilepath = conf_file.DOWNLOADREMOTEFILEPATH self.downloadlocalfilepath = conf_file.DOWNLOADLOCALFILEPATH |
We created a function connect() and placed inside the Ssh_Util class. This function will be used to connect to the remote server.
def connect(self): "Login to the remote server" try: #Paramiko.SSHClient can be used to make connections to the remote server and transfer files print("Establishing ssh connection") self.client = paramiko.SSHClient() #Parsing an instance of the AutoAddPolicy to set_missing_host_key_policy() changes it to allow any host. self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) #Connect to the server if (self.password == ''): private_key = paramiko.RSAKey.from_private_key_file(self.pkey) self.client.connect(hostname=self.host, port=self.port, username=self.username, pkey=private_key, timeout=self.timeout, allow_agent=False,look_for_keys=False) print("Connected to the server",self.host) else: self.client.connect(hostname=self.host, port=self.port, username=self.username, password=self.password,timeout=self.timeout, allow_agent=False,look_for_keys=False) print("Connected to the server",self.host) except paramiko.AuthenticationException: print("Authentication failed, please verify your credentials") result_flag = False except paramiko.SSHException as sshException: print("Could not establish SSH connection: %s" % sshException) result_flag = False except socket.timeout as e: print("Connection timed out") result_flag = False except Exception,e: print("Exception in connecting to the server") print("PYTHON SAYS:",e) result_flag = False self.client.close() else: result_flag = True return result_flag |
Firstly, we created an instance ‘client’ of paramiko.SSHClient(). Paramiko.SSHClient is the primary client used to make connections to the remote server and execute commands. This class wraps Transport, Channel, and SFTPClient to take care of most aspects of authenticating and opening channels. The creation of an SSHClient object allows establishing server connections via the connect() method.
When we try to connect to a server for the first time, we see an error message prompted saying that the machine is not informed about the remote server that we are trying to access. Paramiko’s client.set_missing_host_key_policy is the method used for this purpose. By default, the paramiko.SSHclient sets the policy to the Reject policy which means the policy rejects connection without validating. In our code, we are overriding this by passing AutoAddPolicy() function where the new server will automatically add the server’s host key without prompting. Note:- In a testing environment, we can use set_missing_host_key_policy and set AutoAddPolicy but for security purpose, this is not a good idea to use in production.
Paramiko supports both password-based authentication and key-pair based authentication for accessing the server. In our code, we are checking for a password and if available, authentication is attempted using plain username/password authentication. If the password is not available, authentication is attempted by reading the private key file.
client.Connect() is to connect to an SSH server. This method also allows to provide our own private key, or connect to the SSH agent on the local machine or read from the user’s local key files. We also used timeout (in seconds) to wait for an authentication response. We are capturing a couple of exceptions like paramiko.AuthenticationException, paramiko.SSHException, Socket.timeout to handle the errors while connecting.
Executing commands on the server
Now, you are connected to the remote server. The next step is to execute commands on the SSH server. To run a command on the server the exec_command() function is called on the SSHClient with the command passed as input. When you execute commands using exec_command a new Channel is opened and the requested command is executed. The response is returned as Python file-like objects representing stdin, stdout, and stderr(as a 3-tuple)
- The stdin is a write-only file which can be used for input commands.
- The stdout file give the output of the command.
- The stderr gives the errors returned on executing the command. Will be empty if there is no error.
We have written below function execute_command() to execute the commands. The input for this function is the set of commands. The call to the function connect() is made first and once the connection is established it is followed by a call to function exec_command(). We are storing the stdout and stderr into two variables namely ssh_output and ssh_error respectively. The ssh_output is the output to standard output as produced when the command executed on the remote computer. The ssh_error is the output of standard error produced at the same occasion. The user can use these variables for further processing if needed.
def execute_command(self,commands): """Execute a command on the remote host.Return a tuple containing an integer status and a two strings, the first containing stdout and the second containing stderr from the command.""" self.ssh_output = None result_flag = True try: if self.connect(): for command in commands: print("Executing command --> {}".format(command)) stdin, stdout, stderr = self.client.exec_command(command,timeout=10) self.ssh_output = stdout.read() self.ssh_error = stderr.read() if self.ssh_error: print("Problem occurred while running command:"+ command + " The error is " + self.ssh_error) result_flag = False else: print("Command execution completed successfully",command) self.client.close() else: print("Could not establish SSH connection") result_flag = False except socket.timeout as e: print("Command timed out.", command) self.client.close() result_flag = False except paramiko.SSHException: print("Failed to execute the command!",command) self.client.close() result_flag = False return result_flag |
File Transfers
File transfers are needed to perform remote file operations. Paramiko allows to programmatically send and receive files using the SFTP protocol, the connection with the remote host is established in the same way explained in the previous sections, the call to connect() is followed by a call to open_sftp() that returns a new SFTPClient session object. This object allows to perform common SFTP operations like get(), put(),listdir().
We have created two functions(upload_file() and download_file()) to upload a file to the remote server and download a file from the remote server.
Firstly, we created a function upload_file() which uploads a file to the remote server. The code below establishes the SFTP Connection using the SSH client and uploads a file. The put() method will copy a local file (local path) to the SFTP server as the remote path. Note:- The filename should be included. Only specifying a directory may result in an error. Once the operation is done you may close the SFTP session and its underlying channel using ftp_client.close().
def upload_file(self,uploadlocalfilepath,uploadremotefilepath): "This method uploads the file to remote server" result_flag = True try: if self.connect(): ftp_client= self.client.open_sftp() ftp_client.put(uploadlocalfilepath,uploadremotefilepath) ftp_client.close() self.client.close() else: print("Could not establish SSH connection") result_flag = False except Exception,e: print('\nUnable to upload the file to the remote server',uploadremotefilepath) print('PYTHON SAYS:',e) result_flag = False ftp_client.close() self.client.close() return result_flag |
Similarly, download_file() function downloads a file from the remote server. The code below establishes the SFTP Connection using the SSH client and downloads a file. The get() method will copy a remote file (remote path) from the SFTP server to the local host as local path. Once the operation is done you may close the SFTP session and its underlying channel using ftp_client.close().
def download_file(self,downloadremotefilepath,downloadlocalfilepath): "This method downloads the file from remote server" result_flag = True try: if self.connect(): ftp_client= self.client.open_sftp() ftp_client.get(downloadremotefilepath,downloadlocalfilepath) ftp_client.close() self.client.close() else: print("Could not establish SSH connection") result_flag = False except Exception,e: print('\nUnable to download the file from the remote server',downloadremotefilepath) print('PYTHON SAYS:',e) result_flag = False ftp_client.close() self.client.close() return result_flag |
Putting it all together
Note: The Python 3 code for this article is here. Please use it instead of the file below.
import paramiko import os,sys,time sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from conf import ssh_conf as conf_file import socket class Ssh_Util: "Class to connect to remote server" def __init__(self): self.ssh_output = None self.ssh_error = None self.client = None self.host= conf_file.HOST self.username = conf_file.USERNAME self.password = conf_file.PASSWORD self.timeout = float(conf_file.TIMEOUT) self.commands = conf_file.COMMANDS self.pkey = conf_file.PKEY self.port = conf_file.PORT self.uploadremotefilepath = conf_file.UPLOADREMOTEFILEPATH self.uploadlocalfilepath = conf_file.UPLOADLOCALFILEPATH self.downloadremotefilepath = conf_file.DOWNLOADREMOTEFILEPATH self.downloadlocalfilepath = conf_file.DOWNLOADLOCALFILEPATH def connect(self): "Login to the remote server" try: #Paramiko.SSHClient can be used to make connections to the remote server and transfer files print("Establishing ssh connection") self.client = paramiko.SSHClient() #Parsing an instance of the AutoAddPolicy to set_missing_host_key_policy() changes it to allow any host. self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) #Connect to the server if (self.password == ''): self.pkey = paramiko.RSAKey.from_private_key_file(self.pkey) self.client.connect(hostname=self.host, port=self.port, username=self.username,pkey=self.pkey ,timeout=self.timeout, allow_agent=False, look_for_keys=False) print("Connected to the server",self.host) else: self.client.connect(hostname=self.host, port=self.port,username=self.username,password=self.password,timeout=self.timeout, allow_agent=False, look_for_keys=False) print("Connected to the server",self.host) except paramiko.AuthenticationException: print("Authentication failed, please verify your credentials") result_flag = False except paramiko.SSHException as sshException: print("Could not establish SSH connection: %s" % sshException) result_flag = False except socket.timeout as e: print("Connection timed out") result_flag = False except Exception,e: print('\nException in connecting to the server') print('PYTHON SAYS:',e) result_flag = False self.client.close() else: result_flag = True return result_flag def execute_command(self,commands): """Execute a command on the remote host.Return a tuple containing an integer status and a two strings, the first containing stdout and the second containing stderr from the command.""" self.ssh_output = None result_flag = True try: if self.connect(): for command in commands: print("Executing command --> {}".format(command)) stdin, stdout, stderr = self.client.exec_command(command,timeout=10) self.ssh_output = stdout.read() self.ssh_error = stderr.read() if self.ssh_error: print("Problem occurred while running command:"+ command + " The error is " + self.ssh_error) result_flag = False else: print("Command execution completed successfully",command) self.client.close() else: print("Could not establish SSH connection") result_flag = False except socket.timeout as e: print("Command timed out.", command) self.client.close() result_flag = False except paramiko.SSHException: print("Failed to execute the command!",command) self.client.close() result_flag = False return result_flag def upload_file(self,uploadlocalfilepath,uploadremotefilepath): "This method uploads the file to remote server" result_flag = True try: if self.connect(): ftp_client= self.client.open_sftp() ftp_client.put(uploadlocalfilepath,uploadremotefilepath) ftp_client.close() self.client.close() else: print("Could not establish SSH connection") result_flag = False except Exception,e: print('\nUnable to upload the file to the remote server',uploadremotefilepath) print('PYTHON SAYS:',e) result_flag = False ftp_client.close() self.client.close() return result_flag def download_file(self,downloadremotefilepath,downloadlocalfilepath): "This method downloads the file from remote server" result_flag = True try: if self.connect(): ftp_client= self.client.open_sftp() ftp_client.get(downloadremotefilepath,downloadlocalfilepath) ftp_client.close() self.client.close() else: print("Could not establish SSH connection") result_flag = False except Exception,e: print('\nUnable to download the file from the remote server',downloadremotefilepath) print('PYTHON SAYS:',e) result_flag = False ftp_client.close() self.client.close() return result_flag #---USAGE EXAMPLES if __name__=='__main__': print("Start of %s"%__file__) #Initialize the ssh object ssh_obj = Ssh_Util() #Sample code to execute commands if ssh_obj.execute_command(ssh_obj.commands) is True: print("Commands executed successfully\n") else: print("Unable to execute the commands") """ #Sample code to upload a file to the server if ssh_obj.upload_file(ssh_obj.uploadlocalfilepath,ssh_obj.uploadremotefilepath) is True: print("File uploaded successfully", ssh_obj.uploadremotefilepath) else: print("Failed to upload the file") #Sample code to download a file from the server if ssh_obj.download_file(ssh_obj.downloadremotefilepath,ssh_obj.downloadlocalfilepath) is True: print("File downloaded successfully", ssh_obj.downloadlocalfilepath) else: print("Failed to download the file") """ |
Hope this code helps you get started with Paramiko easily. Happy coding! If you liked this article and want to learn more about Qxf2 and our testing services for startups, click here.
References
1) SSH and transfer file using paramiko
2) How to SSH in Python using paramiko
3) Paramiko Github examples
4) How does SSH works?
I am an experienced engineer who has worked with top IT firms in India, gaining valuable expertise in software development and testing. My journey in QA began at Dell, where I focused on the manufacturing domain. This experience provided me with a strong foundation in quality assurance practices and processes.
I joined Qxf2 in 2016, where I continued to refine my skills, enhancing my proficiency in Python. I also expanded my skill set to include JavaScript, gaining hands-on experience and even build frameworks from scratch using TestCafe. Throughout my journey at Qxf2, I have had the opportunity to work on diverse technologies and platforms which includes working on powerful data validation framework like Great Expectations, AI tools like Whisper AI, and developed expertise in various web scraping techniques. I recently started exploring Rust. I enjoy working with variety of tools and sharing my experiences through blogging.
My interests are vegetable gardening using organic methods, listening to music and reading books.
Hi Indira ,
Thanks so much for this documentation, it really helps me a lot. I’m totally new to paramiko and ssh client. I’m having trouble with connect function now. I have all fields in conf_file, including private key and private key passphrase. But I’m getting “AthenticationException: Authentication failed.” error. I’m wondering if this is because I haven’t include credential yet. I download gcp service key credential as a json file. But I’m not sure how paramiko and connect function will read it. I check the github link in above comments, but I’m still not sure where to put credential in conf folder. Could you give more detail on that? I would really appreciate!
Thank you so much!
Michelle
Hi Michelle,
Here is an example conf file from our SSH implementation in our automation framework – https://github.com/qxf2/qxf2-page-object-model/blob/master/conf/ssh_conf.py.
Hope this helps.
Thanks
Shivahari.P
bellissimo! Grazie!
Hi,
I wanted to connect to multiple hosts and execute a different set of commands on each host.
Can I make Hosts as a list and loop through them before the try
def connect(self):
“Login to the remote server”
try:
Please let me know.
Thanks,
Asha
Hi,
I dont want to prompt user with password everytime. Is there any way to use public ssh keys in this code to establish a connection passwordless?
Hi Sushil,
You can use key_filename argument of SSHClient.connect. Below document references may help you as well.
Document Reference:
====================
http://docs.paramiko.org/en/stable/api/client.html
https://stackoverflow.com/questions/51299834/passwordless-ssh-connection-with-paramiko-fails-where-as-with-ssh-works
https://stackoverflow.com/questions/8382847/how-to-ssh-connect-through-python-paramiko-with-ppk-public-key
Regards,
Rahul
Hi,
May i know, how to establish a connection from Local to Remote Machine ‘A’ and establish to ‘B’ and execute command and perform sftp operation to local machine.
Can you please help me out on this..?
Thanks.
Hi,
We have not tried this but i think you can use port forwarding (SSH tunnel) to connect to a server via another server.
You can refer to
1. forward_tunnel function in Paramiko forward.py demo
2. Port forwarding with Paramiko.
ref:
https://stackoverflow.com/questions/57210100/connecting-to-a-server-via-another-server-using-paramiko?noredirect=1&lq=1
https://askubuntu.com/questions/311447/how-do-i-ssh-to-machine-a-via-b-in-one-command
The code is quite making sense to me. However, I find myself hardly to adapt this since I never expose to SSH setting up. I followed up few tutorial set setting openssh_client and openssh_server for both linux laptop using same wifi. But never make it works and very confused with some terminologies that should be used correctly.
I dont know if you can help with some setting before using this scripts. Such as the setting for both laptops as a client and server.
Hi,
I understand that you are facing challenges in getting client and server ssh setup up n running. You can refer to the below link which explains the steps very clearly.I suggest you do these steps on clean systems to avoid any further confusion.
https://phoenixnap.com/kb/ssh-to-connect-to-remote-server-linux-or-windows
Regards,
Rohini
Hi, I want to automate secure crt login with mobile otp, will it work, can it handle the OTP pop up ?
Regards:
Romita
I am getting the following error while running the command with special character
TypeError: can only concatenate str (not “bytes”) to str
Can you please help me in this?
Hey Mohit,
What version of python you are using ? And can you try to encode your string which errors with your_string.encode()
Hi,
thank you for the helpful example.
I look for implementing the server side of Paramiko supporting sftp and exec_command for clients.
an example with the runing sftp part is working but I’ve no glue how to add exce_command functionality.
See:
https://stackoverflow.com/questions/62268014/server-implementation-using-paramiko-for-sftp-and-command-execution
Ideas ? Any help is welcome.
Martin
Hi
thanks for the helpful example about paramiko module.
I’m trying to create multiple ssh sessions using multi threading but not successful.
can you please suggest How to create multiple ssh sessions (two ssh sessions) with same host and run different commands through ssh. First ssh conncetion to be closed after second one is closing
Can you help me out on this.
Hi,
It is not very clear from your comment whether you are trying to make changes to the code mentioned in the blog, if not you can try the following thread. https://stackoverflow.com/questions/3485428/creating-multiple-ssh-connections-at-a-time-using-paramiko.
This talks about making multiple ssh connections at a time using Paramiko.