I am a member of the TestOps team at my client. Our team is responsible for building test systems. In some releases, it is mandatory to reset the default password right after the application installation and this is time-consuming. This scenario was ideal for automation. I wrote a Python program using AsyncSSH and asyncio to save our team’s time.
Limitation of Paramiko
In the Python world,Paramiko is the default module to work with SSH. We already use Paramiko in our test automation framework to address an almost similar issue. I initially tried the same by adding async keyword to the Paramiko function and it did not work. Googling helped me to know AsyncSSH Python package was the right choice for establishing multi-client SSH. asyncio was the right library to perform the necessary tasks in parallel.
Introduction to AsyncSSH
AsyncSSH provides a nice interface for working with SSH. In this post, we will look at how to connect to a host and interact with the host’s terminal. We will use the Python library asyncio to wrap around our AsyncSSH code. This will make our script connect to all the servers in parallel and perform required operations.
Solution overview
Our solution is going to comprise of the following steps:
1. Connect to a single host
2. Use the create_process() method to start an xterm
3. Interact with the terminal
4. Wrap the code using asyncio
4a. Create an event loop
4b. Load tasks
4c. Gather the result
4d. Display the result
1. Connect to a single host
Let us begin by writing a method to connect to a single host. We make the method async because it calls a `change_password` method which is async. I am reading my old and new passwords from a config file which in turn reads it from an environment variable. You can modify this code to suit your needs.
async def run_client(host) -> asyncssh.SSHCompletedProcess: """ # async def run_client(host) -> asyncssh.SSHCompletedProcess: # Make an ssh client connection # capturing the complete output of the shell # Server host key provided in the known_hosts argument when the connection is opened """ user_name = pwdconfig.USER_NAME old_pwd = pwdconfig.OLD_PASSWORD new_pwd = pwdconfig.NEW_PASSWORD try: async with asyncssh.connect(host,username=user_name, password=old_pwd,known_hosts=None) as conn: pwd_prompt = await change_password(conn,old_pwd,new_pwd) return pwd_prompt except (TimeoutError,asyncssh.misc.ProtocolError) as exc: print(f'Error connecting to or resetting password on server {host}: {exc}') |
2. Use the create_process() method to start an xterm
Once we have a connection setup, we will create a terminal on that connection like so:
process = await conn.create_process(term_type='xterm') |
3. Interact with the terminal
We can write to our newly created terminal like so:
process.stdin.write("Random String"+'\n') |
Now, what you want to write and when will depend on the workflow you are creating. In my case, I wanted to reset a password. So I went and looked at what to expect to see on the terminal. Since I was expecting a sequence of messages, I stored the text I was expecting and the text I wanted to input at each stage in a list. The entire `change_password` function looks like this:
async def is_current_line(process,prompt,timeout=10): """ read the terminal output until what the user wants ensure user supplied prompt exists in the command line and return result_flag. else return empty string """ result_flag = False try: result = await asyncio.wait_for(process.stdout.readuntil(prompt), timeout=timeout) # to ensure user supplied prompt exists in the command line, get the last line of the output pwd_prompt = result.split('\n')[-1] except Exception as e: print(f"error when waiting for prompt {prompt}") else: result_flag=True # returing result_flag when user supplied prompt exists in the command line. return result_flag async def change_password(conn,old_pwd,new_pwd): """ Initiating interactive shell defined the list of prompts to be captured when password message prompts. creating interactive input for password reset. returning successful message once the password reset is completed. Password change order -> (current) UNIX password: -> New password: -> Retype new password: -> passwd: all authentication tokens updated successfully. """ successful_prompt="passwd: all authentication tokens updated successfully." prompts = [['(current) UNIX password:',old_pwd, "Old password when prompted was entered successfully"], ['New password:',new_pwd, "New password when prompted was entered successfully"], ['Retype new password:',new_pwd, "Re-enter new password when prompted was entered successfully"], [successful_prompt,None,None] ] process = await conn.create_process(term_type='xterm') for prompt in prompts: result_flag=await is_current_line(process,prompt[0]) if not result_flag: return "failed to change password" if prompt[0]!=successful_prompt: process.stdin.write(prompt[1]+'\n') print(prompt[2]) return successful_prompt |
4. Wrap the code using asyncio
At this stage, we have successfully written code using AsyncSSH to connect to a single machine. To make this run in parallel, we will use the asyncio module.
4a. Create an event loop
We create an event loop and supply it a method that we want to run in parallel.
asyncio.get_event_loop().run_until_complete(run_multiple_clients()) |
4b. Load tasks
We will load the tasks and gather their results. You can substitute the `hosts` list with a list of your IP addresses.
hosts = ['host1','host2'] tasks = (run_client(host) for host in hosts) |
4c. Gather the result
We can use asyncio.gather(*tasks) to gather the results of all the jobs once they are done. The `run_multiple_clients` method looks like this:
async def run_multiple_clients() -> None: """ mention your lists of hosts calling run_client and capturing the successful result if result list has values initiate successful_pwd_msg_prompt method """ hosts = ['host1','host2'] tasks = (run_client(host) for host in hosts) results = await asyncio.gather(*tasks, return_exceptions=True) if len(results) > 0: successful_pwd_msg_prompt(hosts, results) |
4d. Display the result
And now we can triumphantly display the result like so:
def successful_pwd_msg_prompt(hosts,results): # printing the final result after successful password reset. final_result = list(zip(hosts, results)) for i in final_result: print(f"The host {i[0]} {i[1]}") |
Putting it all together
This code looks long because I have added some defensive code to our script. These safeguards will help us identify cases wherein a server timed out or a connection suddenly died, etc.
""" Reset password on multiple hosts in parallel using asyncssh """ import asyncio import asyncssh import pwdconfig def successful_pwd_msg_prompt(hosts, results): """ prompting the final result after successful password reset. """ final_result = list(zip(hosts, results)) for i in final_result: print(f"The host {i[0]} {i[1]}") async def is_current_line(process,prompt,timeout=10): """ read terminal output until what the user wants ensure user-supplied prompt exists in the command line and return result_flag. else return empty string """ try: result = await asyncio.wait_for(process.stdout.readuntil(prompt), timeout=timeout) # to ensure user supplied prompt exists in the command line, get the last line of the output pwd_prompt = result.split('\n')[-1] except Exception as e: print(f"error when waiting for prompt {prompt}{e}") else: result_flag=True # returing result_flag when user-supplied prompt exists in the command line. return result_flag async def change_password(conn,old_pwd,new_pwd): """ Initiating interactive shell defined the list of prompts to be captured when password message prompts. creating interactive input for password reset. returning successful message once the password reset is completed. Password change order -> (current) UNIX password: -> New password: -> Retype new password: -> passwd: all authentication tokens updated successfully. """ successful_prompt="passwd: all authentication tokens updated successfully." prompts = [['(current) UNIX password:',old_pwd, "Old password when prompted was entered successfully"], ['New password:',new_pwd, "New password when prompted was entered successfully"], ['Retype new password:',new_pwd, "Re-enter new password when prompted was entered successfully"], [successful_prompt,None,None] ] process = await conn.create_process(term_type='xterm') for prompt in prompts: result_flag=await is_current_line(process,prompt[0]) if not result_flag: return "failed to change password" if prompt[0]!=successful_prompt: process.stdin.write(prompt[1]+'\n') print(prompt[2]) return successful_prompt async def run_client(host) -> asyncssh.SSHCompletedProcess: """ #async def run_client(host) -> asyncssh.SSHCompletedProcess: # Make an ssh client connection # capturing the complete output of the shell # Server host key provided in the known_hosts argument when the connection is opened """ user_name = pwdconfig.USER_NAME old_pwd = pwdconfig.OLD_PASSWORD new_pwd = pwdconfig.NEW_PASSWORD try: async with asyncssh.connect(host,username=user_name, password=old_pwd,known_hosts=None) as conn: pwd_prompt = await change_password(conn,old_pwd,new_pwd) return pwd_prompt except (TimeoutError,asyncssh.misc.ProtocolError) as exc: print(f'Error connecting to or resetting password on server {host}: {exc}') async def run_multiple_clients() -> None: """ mention your lists of hosts calling run_client and capturing the successful result if result list has >0 initiate successful_pwd_msg_prompt method """ hosts = ['host1','host2'] tasks = (run_client(host) for host in hosts) results = await asyncio.gather(*tasks, return_exceptions=True) if len(results) > 0: successful_pwd_msg_prompt(hosts, results) # starting point of the code. asyncio.get_event_loop().run_until_complete(run_multiple_clients()) |
Output of successful run:
Hire Qxf2 for your testing needs!
I hope this post helped you. If you are looking for technical testers who can implement engineering solutions that go beyond simple UI/API test automation, reach out via Qxf2’s contact page. We are good at testing early stage software and like working with small engineering teams.
I have total of 18+ years of industry experience and worked in all the phases of project lifecycle. My career started as a manual tester, Test Lead and Program Test Manager. Recently I shifted to new company Qxf2, which provides testing services to startups. From Qxf2 joined one of the Qxf2 client where I got to know about Infrastructure as code. My hobbies are reading Telugu books, watching cricket and movies.
