Introduction

I am testing some functions for sending configuration over to cisco asa with netmiko. Netmiko support sending commands and sending commands as a set. In order to deliver the command as a set Jinja2 template engine is used to fill up the variables of the template.

The purpose of this post is to record how I did the script with decorator, on every function that requires to send command set the function will be decorated which adds on deploying command set to the original function.

Another decorator is used for logging netmiko functions.

Credential handling

Username and password are pre-obtained and encrypted in a file. The encryption is using the cryptography module, using the Fernet recipe to easily generate symmetric keys for encryption and decryption.

The method of encryption and decryption has been mentioned many times in this blog, the code for handling credentials as below.

import logging
from cryptography.fernet import Fernet
import os
from device_logging.cisco_logger import today_date, date_fmt_log_config
import sys
from getpass import getpass
import json

logging.basicConfig(filename=f"d:\\temp\\log_{today_date}.log",
                    format="%(asctime)s %(levelname)s:%(message)s",
                    datefmt=date_fmt_log_config,
                    level=logging.INFO,
                    filemode="w")


def get_key(key_name):
    logging.info("Trying to find the specified key.")
    if not os.path.exists(key_name):
        logging.warning("Specified key is not found, a new key will be generated.")
        key = Fernet.generate_key()
        logging.info("New key is generated.")
        with open(key_name, "wb") as write_key:
            logging.info("Writing key bytes into a file.")
            write_key.write(key)
        logging.info("Key is stored in a file - {}".format(key_name))
    with open(key_name, "rb") as read_key:
        logging.info("Opening key file.")
        enc_key = read_key.read()
    logging.info("key file retrieved.")
    return enc_key


def data_crypt(key, filename):
    cipher = Fernet(key)
    if not os.path.exists(filename):
        print("Specified file - {} cannot be found".format(filename))
        user_device = get_user_device()
        encrypted_data = cipher.encrypt(json.dumps(user_device).encode('utf-8'))
        try:
            logging.info("Writing encrypted data to file - {}".format(filename))
            with open(filename, "wb") as enc_file:
                enc_file.write(encrypted_data)
            logging.info("Encrypted data saved.")
        except OSError as e:
            logging.error(e)
            sys.exit(1)
    try:
        logging.info("Attempt to open specified file - {}".format(filename))
        with open(filename, "rb") as file:
            logging.info("Opening encrypted file {}.".format(filename))
            data_byte = file.read()
            logging.info("Encrypted data retrieved from file - {}.".format(filename))
        logging.info("Decrypting data byte to plain text.")
        plaintext = cipher.decrypt(data_byte)
        logging.info("Data decrypted.")
        return json.loads(plaintext)
    except OSError as e:
        logging.error(e)
        sys.exit(1)


def get_user_device():
    host = input("IP address of device: ")
    username = input("Username: ")
    password = getpass()
    return {
        "device_type": "cisco_asa",
        "host": host,
        "username": username,
        "password": password
    }

A brief explanation of the functions.
get_key, reads the key from the enc.key file which stores the symmetric key generated by Fernet, if the key is not found it regenerates the key with Fernet.generate_key() method, however doing this will cause previously encrypted file to be irretrievable hence if the key is lost the entire credential acquisition has to be restarted.

data_crypt function encrypts and decrypts the credential file, if the credential file is not found, this function prompts for user’s input by calling get_user_device() function.

In order to encrypt the file, the dictionary obtained from get_user_device() has to be converted to string by using json.dumps() then encode the string into bytes, encrypts the bytes and save into a file by using the "wb" option.

To decrypt the file, data_crypt is called, the contents of the encrypted file will be read as bytes, then decrypts the bytes, then use json.loads to convert the string into dictionary. The return dictionary will be the device configuration required by netmiko ConnectHandler to establish a ssh connection to the device specified in the device configuration.

Command set handling

A decorator is defined so that every functions that requires to send command set can be decorated. The decorator function code is below:

# Decorator to help to push the command set.
def deploy_config_set(function):
    @device_logging_decorator
    def device_connection_wrapper(*args, **kwargs):
        # The wrapper uses the return values of the function,
        # to establish a connection to the device.
        # After connection to device is established,
        # the command set is sent over. If there is no exception,
        # the wrapper write the configuration set to memory.
        cmd, device = function(*args, **kwargs)
        try:
            with ConnectHandler(**device) as conn:
                try:
                    conn.send_config_set(cmd, delay_factor=2)
                    logging.info("Command executed successfully...")
                except netmiko_exceptions as e:
                    logging.error(e)
                conn.send_command("write memory")
                logging.info("Configuration saved to memory.")
        except netmiko_exceptions as e:
            logging.error(e)
    return device_connection_wrapper

The decorator expects a function as the argument, the function that is passed into the decorator is executed to get the command set and the device configuration, the decorator then runs the ConnectHandler to get a ssh connection to the cisco asa, and use the command set returned by the function to send the command set to the cisco asa.

Log handling

A logging decorator is defined to log general logs of sending commands over to the device.
The code is below.

import logging
from datetime import datetime
from netmiko.ssh_exception import NetMikoAuthenticationException, NetMikoTimeoutException
from paramiko.ssh_exception import SSHException


date_fmt_log_file = "%d-%b-%Y_%H.%M.%S"
date_fmt_log_config = "%d/%b/%Y %H:%M:%S"
today_date = datetime.today().strftime(date_fmt_log_file)
netmiko_exceptions = NetMikoAuthenticationException, NetMikoTimeoutException, SSHException
logging.basicConfig(filename=f"d:\\temp\\log_{today_date}.log",
                            format="%(asctime)s %(levelname)s:%(message)s",
                            datefmt=date_fmt_log_config,
                            level=logging.INFO,
                            filemode="w")


def device_logging_decorator(function):
    def log_wrapper(*args, **kwargs):
        try:
            resp = function(*args, **kwargs)
            logging.info("Login to {}".format(kwargs.get("host")))
            if resp is None:
                logging.warning("No output received from {}".format(kwargs.get("host")))
            logging.info("Task has been executed successfully.")
            return resp
        except netmiko_exceptions as e:
            logging.error(e)
    return log_wrapper

This decorator logs the normal configuration deployment by netmiko and also logs the error if deployment has failed. netmiko generally does not check your cisco configuration, the error is returned when netmiko has issued establish ssh over to the device such as authentication failure, timeout and other SSH exceptions.
It is possible to concatenate all exceptions into a single object and just do a single exception catching with the concatenated object.
netmiko_exceptions = NetMikoAuthenticationException, NetMikoTimeoutException, SSHException

Then:
except netmiko_exceptions as e:
logging.error(e)

Function that configures sub interfaces

This function gets all the configuration information required for configuring sub interfaces, and also get the device configuration for netmiko to establish ssh to cisco asa. This functions then returns both the sub interfaces information and device configuration and pass over to the decorator.

@deploy_config_set
def configure_subif(*args, **kwargs):
    device_config = kwargs
    config_template = template_env.get_template("subif_base.txt")
    user_input_config = args[0]
    cmd_set = config_template.render(
        intf_id=user_input_config.get("interface", None),
        vlan_id=user_input_config.get("vlan", None),
        nameif=user_input_config.get("nameif", None),
        description=user_input_config.get("description", None),
        security_level=user_input_config.get("sec_level", None),
        ip_addr=user_input_config.get("ip", None),
        netmask=user_input_config.get("netmask", None)
    )
    return cmd_set, device_config

Jinja2 template

I have many templates for the cisco asa, for this post I am only posting the configuration template for sub interface.

interface {{intf_id}}
 no nameif
 no security-level
 no ip address
 no shut
 exit
interface {{intf_id}}.{{vlan_id}}
{% if description is not none %}
 description {{description}}
{% endif %}
{% if vlan_id is not none %}
 vlan {{vlan_id}}
{% else %}
 vlan 1
{% endif %}
 nameif {{nameif}}
 security-level {{security_level}}
 ip address {{ip_addr}} {{netmask}}
 end

Testing for the functions

The code test the function for configuring sub interfaces.

from asa import *
from confidentiality.security import *


if __name__ == "__main__":
    key = get_key("enc.key")
    network_device = data_crypt(key, "device.cfg")
    all_interfaces = get_firewall_interfaces(**network_device)
    interface_index = [i + 1 for i in range(len(all_interfaces))]
    interface_dict = dict(zip(interface_index, all_interfaces))
    exit_menu = False
    subif_list = list()
    while not exit_menu:
        print("[1] Configure interfaces.\n"
              "[0] Exit\n")
        choice = int(input("Your choice: "))
        if choice == 0:
            exit_menu = True
        elif choice == 1:
            while True:
                print("Configure sub interfaces...\n")
                for key, value in interface_dict.items():
                    print(key, value)
                k = int(input("Choice (0 to quit): "))
                if k == 0:
                    break
                vlan = int(input("Vlan: "))
                description = input("Interface description: ")
                nameif = input("nameif: ")
                sec_level = int(input("Security level: "))
                ip_addr = input("IP address: ")
                netmask = input("netmask: ")
                subif_config = {
                    "interface": interface_dict[k],
                    "vlan": vlan,
                    "description": description,
                    "nameif": nameif,
                    "sec_level": sec_level,
                    "ip": ip_addr,
                    "netmask": netmask
                }
                subif_list.append(subif_config)
            for subif in subif_list:
                configure_subif(subif, **network_device)

Testing in action

The lab topology looks like this:

So when the code is executed, the test collects all configuration from users until users terminate the collection.

I have not presented all codes for my asa.py library, there are many more functions which I have tested before, the intention for this post is to remind myself on how to use decorator with netmiko configuration. I can still achieve the same result by calling the function within a function, a decorator simply adds a single line like this @deploy_config_set.