Cumulus and Netmiko

By Vince

This post will walk through a quick script that connects to a Cumulus switch and runs a command. Cumulus has a virtual machine appliance called VX available for free to run your tests on. Cumulus VX

Much of this can be found on the netmiko github:Netmiko

There is an examples directory that goes into more details for adding things like concurrency. If you are running this on 10+ switches in series, things get slow!

I like to split the SSH’ing into a separate function so you can import and call that from another script, without the command line arguments portion.

Also logging is not necessary, I like to use it for testing and debugging.

import argparse
import sys
import json

import logging
from netmiko import Netmiko
from getpass import getpass

def ssh_run(ip_addr, username, password, commands):
    """from the ip, login and run the commands"""

    # create the dict needed for netmiko
    device = {
        'device_type': 'linux',
        'ip': ip_addr,
        'username': username,
        'password': password,
        'port': 22,  # optional, defaults to 22
        'secret': password,  # optional, defaults to ''
        'verbose': False  # optional, defaults to False
    }
    
    # This will create a file named 'test.log' in your current directory.
    # It will log all reads and writes on the SSH channel.
    logging.basicConfig(filename="test.log", level=logging.DEBUG)
    logger = logging.getLogger("netmiko")
    
    # initiate SSH connection
    try:
        net_connect = Netmiko(**device)
    
        # get the prompt if needed for later
        prompt = net_connect.find_prompt()
        print('Prompt: ', prompt)
    
    except Exception as err:
        print('Exception: %s' % err)
    
        return err
    
    # Now let's try to send the router a command
    output = []
    
    # if we sent a list of commands them run them one by one
    
    if isinstance(commands, list):
        for command in commands:
            output.append(net_connect.send_command(command))
    
    # otherwise just run the one command
    else:
        output.append(net_connect.send_command(commands))
    
    net_connect.disconnect()
    
    return output


if __name__ == '__main__':

    usage = 'usage {prog} -i <target host> -u <user> -c <"command1, command2, etc">'.format(prog=sys.argv[0])
    
    parser = argparse.ArgumentParser(usage=usage)
    # We add the cumulus default login credentials here
    # We are going to run "net show int json" which returns a dict of the interfaces
    parser.add_argument('-i', dest='ip_addr', default="192.168.50.208", help='specify target host')
    parser.add_argument('-c', dest='commands', default="net show int json",
                        help='specify commands separated by a command with quotes "command1,command2"')
    parser.add_argument('-u', dest='user', default="cumulus", help='specify the user')
    parser.add_argument('-p', dest='passw', default="CumulusLinux!", help='specify the password (optional)')
    
    args = parser.parse_args()
    
    ip_addr = args.ip_addr

#split the commands coming in, there might be only one
    argsin = args.commands.split(',')
    username = args.user

    # securely get the input password if not specified
    if args.passw:
        password = args.passw
    else:
        password = getpass()
    
    output = ssh_run(ip_addr, username, password, argsin)
    
    interfaces = json.loads(output[0])
    
    print(f"Interfaces: {interfaces.keys()})
    
    print(f"Interface details:\n{interfaces['vlan10']}")

If we run this, we can see some nice output:

#python cumulus-netmiko.py
dict_keys(['swp5', 'bridge', 'vlan10', 'lo', 'swp2', 'swp3', 'swp1', 'swp6', 'swp4', 'eth0'])
...Output removed

Once you have the interface information saved in the dictionary, you can start to parse it out and look for the correct IPs for things like VLANS:

try:
        ip_address = interfaces['vlan10']['iface_obj']['ip_address']['allentries'][0]
        print(f"IP Details:\n{ip_address}")
        if ip_address == '10.100.100.100/24':
            print(f"Ip address match")

except ValueError as e:
        print(f"Cannot find IP address: {e}")
        exit(1)

This is great for config validation if you are building out a new fabric.

Make sure to add some try/except blocks if you are assigning data from an unknown dict. The keys might not be there! Unlike typed languages (like golang), python is very forgiving when assigning objects to objects.