Device Failover

Prerequisites

• Experience with Python and install Python packages
• RESTful API experience
• Conductor user account with API access enabled

Required Python Packages

• multiping
• requests

Sample Script

#!/usr/bin/python3

from ipaddress import ip_address
import json
from multiping import MultiPing
import os
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import sys
import time

# Suppress cert warning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

url = "https://<Conductor IP>/api/v1/"

# pre-2.1.3 API headers
# headers = {
#     'x-person-token': '1234',
#     'x-person-email': 'api@temperednetworks.com',
#     'content-type': 'application/json'
# }

headers = {
    'x-api-client-id': 'KWFCITL4VX_B-ZSdywD0Eg',
    'x-api-token': 'grui5TWLvvic8mZ7fgglbA',
    'content-type': 'application/json'
}


def generate_menu_select(cont, msg):
    """
    Generate a menu structure from a list of dictionaries

    :param cont: content to be processed
    :param msg: message to be asked

    :returns: selected object from list
    """

    data = sorted(cont, key=lambda k: k['name'])
    for d in data:

        print('{0}) {1}'.format(data.index(d) + 1, d['name']))

    i = input(msg)

    try:
        return(data[int(i) - 1])
    except (IndexError, ValueError):
        print('Selected input {0} not found or invalid'.format(i))
        sys.exit()


def get_overlay():
    """
    Get all overlays in Conductor

    :returns: selected overlay network
    """

    r = requests.get(url + "overlay_networks", headers=headers, verify=False)

    if r.status_code == requests.codes.ok:
        print('Collected overlays:')
        return generate_menu_select(r.json(),
                                    "Select overlay network to failover: ")
    else:
        print('Error getting overlay networks')
        print(r.json())
        sys.exit()


def get_object_in_overlay(devs, dgs, ovl_gps):
    """
    Get all devices/device groups in the overlay. Move to own list.

    :param devs: all devices
    :param dgs: all device groups
    :param ovl_gps: all devices/device groups in overlay

    :returns: selected device/device group
    """

    # devices and device groups
    grps = [d for d in devs for o in ovl_gps if o == d['uuid']]
    grps.extend([d for d in dgs for o in ovl_gps if o == d['uuid']])

    print('Collected devices and device groups in overlay network:')
    return generate_menu_select(grps, 'Select device/device group to replace: ')


def get_device_groups():
    """
    Get all device groups in Conductor

    :returns: all Conductor device groups
    """

    r = requests.get(url + "device_groups", headers=headers, verify=False)

    if r.status_code == requests.codes.ok:
        return r.json()
    else:
        print('Error getting device groups')
        print(r.json())
        sys.exit()


def get_devices():
    """
    Get all devices in Conductor

    :returns: all Conductor devices
    """

    r = requests.get(url + "devices", headers=headers, verify=False)

    if r.status_code == requests.codes.ok:
        return r.json()
    else:
        print('Error getting devices')
        print(r.json())
        sys.exit()


def add_device_to_overlay(ovl_uuid, d_uuid):
    """
    Add a device to an overlay network

    :param ovl_uuid: overlay network UUID
    :param d_uuid: device UUID
    """

    payload = {'network_id': ovl_uuid,
               'device_group_ids': [d_uuid]}

    r = requests.post(url + "overlay_network_devices", headers=headers,
                      data=json.dumps(payload), verify=False)

    if not r.status_code == requests.codes.ok:
        print('Error adding to overlay')
        print(r.json())
        sys.exit()


def remove_device_from_overlay(ovl_uuid, d_uuid):
    """
    Remove a device from an overlay network

    :param ovl_uuid: overlay network UUID
    :param d_uuid: device UUID
    """

    payload = {'network_id': ovl_uuid,
               'device_group_ids': [d_uuid]}

    r = requests.delete(url + "overlay_network_devices", headers=headers,
                        data=json.dumps(payload), verify=False)

    if not r.status_code == requests.codes.ok:
        print('Error removing from overlay')
        print(r.json())
        sys.exit()


def build_overlay_policy(ovl_uuid, d_uuid, ds_uuid):
    """
    Build the overlay policy

    :param ovl_uuid: overlay network UUID
    :param d_uuid: device UUID
    :param ds_uuid: UUIDs of devices in policy with previous device (target)
    """

    for uuid in ds_uuid:
        payload = {'network_id': ovl_uuid,
                   'device_group_1': d_uuid,
                   'device_group_2': uuid}

        r = requests.post(url + "overlay_network_devices/trust", headers=headers,
                          data=json.dumps(payload), verify=False)

        if not r.status_code == requests.codes.ok:
            print('Error adding policy in overlay')
            print(r.json())


def get_replacement_object(devs, dgs):
    """
    Select which object to use as a replacement

    :param devs: all devices
    :param dgs: all device groups

    :returns: device/device group JSON data
    """

    selection = [{'name': 'Device'}, {'name': 'Device Group'}]

    sel = generate_menu_select(selection, 'Type to replace with: ')

    if sel['name'] == 'Device':
        return generate_menu_select(devs, 'Select replacement device: ')
    elif sel['name'] == 'Device Group':
        return generate_menu_select(dgs, 'Select replacement device group: ')


def replace_overlay_object(ovl, target, replacement):
    """
    Replace a given target with a replacement device/device group object

    :param ovl: overlay JSON data
    :param target: target device JSON data
    :param replacement: replacement device JSON data
    """

    policies = [p for p in [t['from'] for t in ovl['policy'] if t['to'] == target['uuid']]]

    remove_device_from_overlay(ovl['uuid'], target['uuid'])
    print('Removing device from overlay')
    add_device_to_overlay(ovl['uuid'], replacement['uuid'])
    print('Adding replacement device to overlay')
    build_overlay_policy(ovl['uuid'], replacement['uuid'], policies)
    print('Build overlay policy with replacement device')


def select_mon_target():
    """
    Input an IP to monitor

    :returns: IP address to monitor
    """

    selection = True
    mon_target = None

    while selection:
        mon_target = input('Enter IP target to monitor: ')
        try:
            ip_address(mon_target)
            selection = False
        except ValueError:
            print('Invalid IP address.')
            continue

    return mon_target


def monitor_target(mon_target):
    """
    Monitor the given target import ip

    :param mon_target: IP address to monitor
    """

    active = True
    while active:
        if mon_target:
            mp = MultiPing([mon_target])
            mp.send()
            resp, no_resp = mp.receive(.1)
            stamp = time.strftime('%Y-%m-%d %H:%M:%S')

            if no_resp:
                print('{0}: Monitor failed'.format(stamp))
                break
            else:
                print('{0}: Ping monitor successful'.format(stamp))
                time.sleep(1)


def main():

    if not os.geteuid() == 0:
        print('Must run as root or Administrator. Exiting...')

    else:

        # get content
        ovl = get_overlay()
        devs = get_devices()
        dgs = get_device_groups()

        # ask questions
        target = get_object_in_overlay(devs, dgs, ovl['device_groups'])
        replacement = get_replacement_object(devs, dgs)

        # monitor ip
        mon_target = select_mon_target()
        monitor_target(mon_target)

        # do work
        replace_overlay_object(ovl, target, replacement)
        print('Device failover completed')


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit()