Categories
Network Automation

Python: Paramiko (Configuring Multiple devices)

In the previous post, you learned how to use python’s paramiko to access a network device, send commands to it only to retrieve output. The topology had two devices, however only one was accessed using paramiko.

In this post we will utilize the same topology and configure both devices at the same time; killing two birds with one stone, the paramiko stone. The same code that was used in the last post will be used here; we’ll just make a few modifications to it to get it to do what we want it to do.

NOTE: This lab assumes that all devices have the same username and password. Another post will be made showing you how to build a script for devices with different passwords.

So here is the code that we’re working with:

#Build SSH Client
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
enter_key = "\n"

#Build datastructure for device_info
router_defaults = {
            'hostname': f'192.168.0.131',
            'username': f'admin',
            'password': f'redEyeCoding123',
            'port': '22',
            'allow_agent': False,
            'look_for_keys': False
}

#Connect to Client
ssh_client.connect(**router_defaults)
is_session_active = ssh_client.get_transport().is_active()
print(f"Is the session active?: {is_session_active}") # // True

#Build The Shell
command_shell = ssh_client.invoke_shell()

#Send Commands
command_shell.send(f"terminal length 0 {enter_key}")
command_shell.send(f"show ip interface brief {enter_key}")
command_shell.send(f"show clock {enter_key}")
command_shell.send(f"show run {enter_key}")

# Go to sleep for 2 seconds and give the 
  # router time to respond!!
time.sleep(2)

router_output = command_shell.recv(10000).decode("utf-8")
# we need to decode the output from the 
	# router so it is readable to us in terminal.
print(router_output)

And here is our topology:

Today we will be configuring something simple: Loopback interfaces.

So lets begin:

Let’s start by using functional programming. I prefer this method as it is easy on the eyes ( my humble opinion ) while reading code.

We’ll start by building a list of ip addresses for us to loop through. In out case we only have two, routers R1 and R2. I’ll do this by asking the user for input and placing them into a list. We will also ask them for the loopback address they would like to assign to the devices:

host1 = input("Enter ip for host1: ").strip()
host2 = input("Enter ip for host2: ").strip()
host1_loop = input("Enter loopback address for host1 (0.0.0.0 0.0.0.0): ").strip()
host2_loop = input("Enter loopback address for host2 (0.0.0.0 0.0.0.0): ").strip()

devices = [host1, host2]

Let us also import the getpass module so we can receive the password as input from the user. We don’t want the user’s password to be seen in clear text, so we will also import the sys module to prevent this from happening ( the import state will be place at the top of the module):

import getpass
pwd = getpass.getpass(prompt="Enter your password: ".strip(), stream=sys.stderr)

Next we’ll transform the code we used to build the SSH client into a function. This function will take in a single positional argument of ip_address , which will be followed by default parameters of : username, pt, username, allow_Ag, and get_Keys. The said arguments will be used when we attempt to connect to our devices. I”ll also add print functions to so we can know what stage we’re at in the program when we run it. I’ll also use the sleep method to make things output a little smoother.

Within this function we’ll also build a dictionary containing the parameters needed to connect to these devices.

# Build SSH Client    
def build_ssh_client(ip_address, username='admin', pt=22, allow_Ag=False, get_Keys=False):
    print(f"Building ssh_client for host {ip_address}...")
    time.sleep(2)
    
    device_info ={      
            "hostname": f"{ip_address}",
            "username": f"{username}",
            "port": f"{pt}",
            "password": f"{pwd}",
            "allow_agent": f"{allow_Ag}",
            "look_for_keys": f"{get_Keys}"  
        }
    ssh_client = paramiko.SSHClient()
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    connect_to_host(ssh_client, device_info)  

    return

As you can see we’re calling the connect_to_host function, while passing in our ssh_client and dictionary as arguments. We will assemble the connect_to_host function now:

#Connect To Host
def connect_to_host(client, device_info):
    print(f"Connecting to host {device_info['hostname']}...")
    time.sleep(2)
    client.connect(**device_info)
    
    # Build shell
    build_shell(client, device_info['hostname'])
    return            
        

Now lets modify the build_shell function:

#Build Shell
def build_shell(client, ip_address):
    print(f"Building shell for host {ip_address}...")
    time.sleep(2)
    command_shell = client.invoke_shell()
    print(f"Shell successfully build for host {ip_address}!")    
    
    #Send commands to shell
    send_commands(command_shell, ip_address, client)  
    return

As per the above, the build_shell function will return nothing. However within it’s body it will call the send_commands function which will accept three positional arguments:

Now for the send_commands function:

def send_commands(shell, ip_address, client):
    enter_key = "\n"
    print(f"Sending commands to host {ip_address}...")
    shell.send(f"terminal length 0 {enter_key}")
    commands = ['conf t', 'interface loopback1', 'ip address']
    address = host1_loop if (ip_address == host1) else host2_loop

    for command in commands:
        if 'address' in command:
            shell.send(f'{command} {address} {enter_key}')
        else:
            shell.send(f'{command}{enter_key}')

    shell.send(f'end {enter_key}')
    shell.send(f'show ip int bri {enter_key}')

    time.sleep(2)
    router_output = shell.recv(10000).decode("utf-8")
    print(router_output)

    session_active = client.get_transport().is_active()
    if session_active:
        print(f"Closing session for host {ip_address}")
        client.close()
    return router_output

I created a variable call session_active so we can use this close the session once commands were sent. The commands were placed into an array so it can be iterated over. In addition, I used pythons ternary operator (conditional expressions) and assigned that value to the address variable to then be use in the for loop.

Moving along…

Now that we have our functions built. We will need to organize them within our module so we it can run without any errors. Unlike JavaScript, Python will yell and complain if you attempt to invoke a function before it is declared…python has no concept of JavaScript’s “hoisting“.

So lets organize the functions accordingly:

import getpass
import paramiko
import time
import sys

# OUR GLOBAL VARIABLES
pwd = getpass.getpass(prompt="Enter your password: ".strip(), stream=sys.stderr)
host1 = input("Enter ip for host1: ").strip()
host2 = input("Enter ip for host2: ").strip()
host1_loop = input("Enter loopback address for host1 (0.0.0.0 0.0.0.0): ").strip()
host2_loop = input("Enter loopback address for host2 (0.0.0.0 0.0.0.0): ").strip()
devices = [host1, host2] 

def send_commands(shell, ip_address, client):
    enter_key = "\n"
    print(f"Sending commands to host {ip_address}...")
    shell.send(f"terminal length 0 {enter_key}")    

    commands = ['conf t', 'interface loopback1', 'ip address']
    address = host1_loop if (ip_address == host1) else host2_loop

    for command in commands:
        if 'address' in command:
            shell.send(f'{command} {address} {enter_key}')
        else:
            shell.send(f'{command}{enter_key}')


    shell.send(f'end {enter_key}')
    shell.send(f'show ip int bri {enter_key}')

    time.sleep(2)
    router_output = shell.recv(10000).decode("utf-8")
    print(router_output)

    session_active = client.get_transport().is_active()
    if session_active:
        print(f"Closing session for host {ip_address}")
        client.close()
    return router_output
  

#Build Shell
def build_shell(client, ip_address):
    print(f"Building shell for host {ip_address}...")
    time.sleep(2)
    command_shell = client.invoke_shell()
    print(f"Shell successfully build for host {ip_address}!")    
    
    #Send commands to shell
    send_commands(command_shell, ip_address, client)  
    return


#Connect To Host
def connect_to_host(client, device_info):
    print(f"Connecting to host {device_info['hostname']}...")
    time.sleep(2)
    client.connect(**device_info)
    
    # Build shell
    build_shell(client, device_info['hostname'])
    return

          
# Build SSH Client    
def build_ssh_client(ip_address, username='admin', pt=22, allow_Ag=False, get_Keys=False):
    print(f"Building ssh_client for host {ip_address}...")
    time.sleep(2)
    
    device_info ={      
            "hostname": f"{ip_address}",
            "username": f"{username}",
            "port": f"{pt}",
            "password": f"{pwd}",
            "allow_agent": f"{allow_Ag}",
            "look_for_keys": f"{get_Keys}"  
        }
    ssh_client = paramiko.SSHClient()
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    connect_to_host(ssh_client, device_info)
    return
        
 

So far things are looking good. Now we need a way to make this thing run so it sends our commands to both devices; a for loop. Lets grab the list of ip addresses we created at the beginning of this post to do this.

for ip_address in devices:
  build_ssh_client(ip_address)

If you run the program everything should work without any issues. I entered the following for loopback addresses:

1.1.1.1 255.255.255.255 & 2.2.2.2 255.255.255.255

The output below is from my terminal:

I’m not a fan of this code because it is dependent on the the static configurations within the send_commands function. With that being said, there is a better way to accomplish the same task; which is to extract the commands from a file. That will be in the next post.

I also know that this program is a bit long; meaning it could easily be shortened and more readable. I sort of through it together on the fly ( while typing this post out ).

Hope this helped.