Python для сетевых инженеров: начало пути

Наверное, многие сетевые инженеры уже поняли, что администрирование сетевого оборудования только через CLI слишком трудоёмко и непродуктивно. Особенно когда под управлением находятся десятки или сотни устройств, часто настроенных по единому шаблону. Удалить локального пользователя со всех устройств, проверить конфигурации всех маршрутизаторов на соответствие каким-то правилам, посчитать количество включенных портов на всех коммутаторах — вот примеры типовых задач, решать которые без автоматизации нецелесообразно.



Эта статья в основном для сетевых инженеров, которые пока не знакомы или очень слабо знакомы с Python. Мы рассмотрим пример скрипта для решения некоторых практических задач, который вы сразу сможете применять в своей работе.


Для начала расскажу, почему я выбрал Python.

Во-первых, это легкий в освоении язык программирования, который позволяет решать очень широкий круг задач.

Во-вторых, крупные производители сетевого оборудования, такие как Cisco, Juniper, Huawei, внедряют поддержку Python на своем оборудовании. У языка есть будущее в сетевой сфере, и его изучение не будет пустой тратой времени.

В-третьих, язык очень распространен. Для него написано много полезных библиотек, есть большое сообщество программистов, и найти ответы на большинство вопросов в интернете можно в первых строках поисковой выдачи.

Я занимаюсь проектированием и немного внедрением сетевых проектов. В одном из них потребовалось решить сразу две задачи.

1. Пройтись по нескольким сотням филиальных маршрутизаторов и убедиться, что они настроены единообразно. Например, что для связи с ЦОД используется интерфейс Tunnel1, а не Tunnel0 или Tunnel99. И что эти интерфейсы настроены одинаково, за исключением их IP-адресов, естественно.
2. Перенастроить все маршрутизаторы, в том числе добавить статический маршрут через IP-адрес местного провайдера. То есть эта команда будет уникальной для каждого маршрутизатора.

На помощь пришел скрипт на Python. Его разработка и тестирование заняли один день.

Первое, что нужно сделать, это установить Python и крайне желательно PyCharm CE. Скачиваем и устанавливаем Python 3 (сейчас последняя версия 3.6.2). При установке выбираем «Customize installation» и на этапе «Advanced Options» устанавливаем галку напротив «Add Python to environment variables».

PyCharm CE — это бесплатная среда разработки с очень удобным отладчиком. Скачиваем и устанавливаем.

Второй шаг — устанавливаем необходимую библиотеку netmiko. Она нужна для взаимодействия с устройствами по SSH или telnet. Библиотеку устанавливаем из командной строки:

pip install netmiko

Третьим шагом будет подготовка исходных данных и скрипта под наши задачи.

В качестве входных данных будем использовать текстовый файл “ip.txt”. В каждой строчке файла должен быть IP-адрес устройства, к которому мы подключаемся. Через запятую можно указать логин и пароль для конкретного устройства. Если этого не сделать, то будут использоваться те, которые вы введёте при запуске скрипта. Пробелы будут проигнорированы. Если первый символ в строке «#», то она считается комментарием и игнорируется. Вот пример корректного файла:

Сам скрипт логически состоит из двух частей: основной программы и функции doRouter(). Внутри неё выполняется подключение к маршрутизатору, отправка команд в CLI, получение и анализ ответов. Входными данными для функции являются: IP-адрес маршрутизатора, логин и пароль. При возникновении проблем функция вернёт IP-адрес маршрутизатора, мы его запишем в отдельный файл fail.txt. Если всё прошло хорошо, то будет просто выведено сообщение на экран.

Почему нужно выносить взаимодействие с маршрутизаторами в отдельную функцию, а не выполнить всё в цикле в основной программе? Главная причина — продолжительность работы скрипта. Подключение поочередно ко всем маршрутизаторам заняло у меня 4 часа. В основном из-за того, что какие-то из них не отвечали и скрипт долго ждал истечения таймаута. Поэтому запускать мы будем параллельно по 10 экземпляров функций. В моём случае это сократило время выполнения скрипта до 10 минут.

Рассмотрим теперь подробнее основную программу.

Ради безопасности не будем хранить логин и пароль в скрипте. Поэтому выведим на экран приглашение для их ввода. Причем при вводе пароля он не будет отображаться. Эти глобальные переменные используем в процедуре doRouter. У меня были проблемы с работой getpass в PyCharm под Windows. Скрипт работал корректно, только если выполнять его в режиме Debug, а не Run. В командной строке всё работало без нареканий. Также скрипт тестировался в OS X, там проблем в PyCharm замечено не было.

user_name = input("Enter Username: ")
pass_word = getpass()

Потом читаем файл с IP-адресами. Конструкция try…except позволит корректно обработать ошибку чтения файла. На выходе получим массив данных для подключения connection_data, содержащий IP-адрес, логин и пароль.

try:
    f = open('ip.txt')
    connection_data=[]
    filelines = f.read().splitlines()
    for line in filelines:
if line == "": continue
        if line[0] == "#": continue
        conn_data = line.split(',')
        ipaddr=conn_data[0].strip()
        username=global_username
        password=global_password
        if len(conn_data) > 1 and conn_data[1].strip() != "": username = conn_data[1].strip()
        if len(conn_data) > 2 and conn_data[2].strip() != "": password = conn_data[2].strip()
        connection_data.append((ipaddr, username, password))
    f.close()
except:
    sys.exit("Couldn't open or read file ip.txt")

Далее создаём список процессов и запускаем их. Метод создания процессов я задал как “spawn”, чтобы в Windows и OS X скрипт работал одинаково. Количество созданных процессов будет равно количеству IP-адресов. Но выполняться одновременно будут не более 10. В список routers_with_issues записываем то, что вернут функции doRouter. В нашем случае это IP-адреса маршрутизаторов, с которыми были проблемы.

multiprocessing.set_start_method("spawn")
with multiprocessing.Pool(maxtasksperchild=10) as process_pool:
    routers_with_issues = process_pool.map(doRouter, connection_data, 1)
    process_pool.close()
    process_pool.join()

Команда process_pool.join() нужна для того, чтобы скрипт дождался завершения выполнения всех экземпляров функций doRouter() и только потом продолжил выполнять основную программу.

В конце создаем/переписываем текстовый файл, в котором у нас будут IP-адреса ненастроенных маршрутизаторов. Также выводим этот список на экран.

failed_file = open('fail.txt', 'w')
for item in routers_with_issues:
    if item != None:
      failed_file.write("%s\n" % item)
      print(item)

Теперь разберем процедуру doRouter(). Первое, что нужно сделать, — обработать входные данные. С помощью ReGex проверяем, что функции был передан корректный IP-адрес.

ip_check = re.findall("^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$", ip_address)
if ip_check == []:
    print(bcolors.FAIL + "Invalid IP - " + str(ip_address) + bcolors.ENDC)
    return ip_address

Далее создаём словарь с необходимыми для подключения данными и подключаемся к маршрутизатору.

device = {
    'device_type': 'cisco_ios',
    'ip': ip_address.strip(),
    'username': username,
    'password': password,
    'port': 22, }
try:
    config_ok = True

    net_connect = ConnectHandler(**device)

Отправляем команды и анализируем полученный ответ от маршрутизатора. Он будет помещён в переменную cli_response. В этом примере мы проверяем текущие настройки. Результат выводим на экран. Данную часть нужно менять под разные задачи. В этом скрипте проверяем текущую конфигурацию маршрутизатора. Если она корректная, то вносим изменения. Если при проверке обнаружены проблемы, то присваиваем переменной config_ok значение False и не применяем изменения.

cli_response = net_connect.send_command("sh dmvpn | i Interface")
cli_response = cli_response.replace("Interface: ", "")
cli_response = cli_response.replace(", IPv4 NHRP Details", "").strip()
if cli_response != "Tunnel1":
    print(str(ip_address)+" - " + bcolors.WARNING + "WARNING - DMVPN not on Tunnel1.  " + cli_response+ " " + bcolors.ENDC)
    config_ok=False

Тут будут полезны следующие операции работы со строками.

Операция	Описание	Пример
+	Объединение строк	s3 = s1 + s2 >>> print('Happy New ' + str(2017) + ' Year') Happy New 2017 Year
len(s)	Определение длины строки
[]	Выделение подстроки (индекс начинается с нуля)	s[5] — шестой символ s[5:7] — символы с шестого по восьмой s[-1] — последний символ, то же, что s[len(s)-1]
s.split() s.join()	Разделить строки Объединить строки	>>> 'Петя, Лёша, Коля'.split(',') ['Петя', 'Лёша', 'Коля'] >>> ','.join({'Петя', 'Лёша', 'Коля'}) 'Лёша, Петя, Коля'
str(L) list(s)	Преобразовать список в строку Преобразовать строку в список	>>> str(['1', '2', '3']) "['1', '2', '3']" >>> list('Test') ['T', 'e', 's', 't']
%	Форматирование по шаблону	>>> s1, s2 = 'Митя', 'Василиса' >>> '%s + %s = любовь' % (s1, s2) 'Митя + Василиса = любовь'
f	Подстановка переменных	>>> a='Максим' >>> f'Имя {a}' 'Имя Максим'
str.find(substr)	Поиск подстроки substr в строке str Возвращает позицию первой найденной подстроки	>>> 'This is a text'.find('a') 8
str.replace(old, new)	Замена подстроки old на подстроку new в строке str	>>> newstr = 'This is a text'.replace(' is ', ' is not ') >>> print(newstr) This is not a text
str.strip() str.rstrip()	Удалить пробелы и табуляции в начале и конце (или только в конце)	>>> ' This is a text \t\t\t'.strip() 'This is a text'

Чтобы решить задачу по добавлению статического маршрута, для начала нужно определить IP-адрес next-hop. В моем случае самый простой способ — посмотреть адрес next-hop у существующих статических маршрутов.

cli_response2=net_connect.send_command("sh run | i ip route 8.8.8.8 255.255.255.255")
if cli_response2.strip() == "":
    print(str(ip_address)+" — " + bcolors.FAIL + "WARNING — couldn't find static route to 8.8.8.8" + bcolors.ENDC)
    config_ok=False

ip_next_hop = ""
if cli_response2 != "":
    ip_next_hop = cli_response2.split(" ")[4]

if ip_next_hop == "":
    print(str(ip_address)+" — " + bcolors.FAIL + "WARNING — couldn't find next-hop IP address " + bcolors.ENDC)
    config_ok=False

Можно отправлять одну или несколько конфигурационных команд сразу. У меня плохо работала отправка больше 5 команд одновременно, при необходимости можно просто повторить конструкцию несколько раз.

config_commands = ['ip route 1.1.1.1 255.255.255.255 '+ip_next_hop,
                   'ip route 2.2.2.2 255.255.255.255 '+ip_next_hop]
net_connect.send_config_set(config_commands)

Полный скрипт.

import sys
from netmiko import ConnectHandler
from getpass import getpass
import time
import multiprocessing
import re

start_time = time.time()

class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

def doRouter(connection_data):

    ip_address = connection_data[0]
    username = connection_data[1]
    password = connection_data[2]

    ip_check = re.findall("^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$", ip_address)
    if ip_check == []:
        print(bcolors.FAIL + "Invalid IP - " + str(ip_address) + bcolors.ENDC)
        return ip_address

    device = {
        'device_type': 'cisco_ios',
        'ip': ip_address.strip(),
        'username': username,
        'password': password,
        'port': 22, }
    try:
        config_ok = True

        net_connect = ConnectHandler(**device)

        cli_response = net_connect.send_command("sh dmvpn | i Interface")
        cli_response = cli_response.replace("Interface: ", "")
        cli_response = cli_response.replace(", IPv4 NHRP Details", "").strip()
        if cli_response != "Tunnel1":
            print(str(ip_address)+" - " + bcolors.WARNING + "WARNING - DMVPN not on Tunnel1.  " + cli_response+ " " + bcolors.ENDC)
            config_ok=False

        cli_response2=net_connect.send_command("sh run | i ip route 1.1.1.1 255.255.255.255")
        if cli_response2.strip() == "":
            print(str(ip_address)+" - " + bcolors.WARNING + "WARNING - couldn't find static route to 8.8.8.8" + bcolors.ENDC)
            config_ok=False

        ip_next_hop = ""
        if cli_response2 != "":
            ip_next_hop = cli_response2.split(" ")[4]

        if ip_next_hop == "":
            print(str(ip_address)+" - " + bcolors.WARNING + "WARNING - couldn't find next-hop IP address " + bcolors.ENDC)
            config_ok=False


        if config_ok:
            config_commands = ['ip route 1.1.1.1 255.255.255.255 '+ip_next_hop,
                               'ip route 2.2.2.2 255.255.255.255 '+ip_next_hop]
         	    net_connect.send_config_set(config_commands)
            print(str(ip_address) + " - " + "Static routes added")
        else:
            print(str(ip_address) + " - " + bcolors.FAIL + "Routes weren't added because config is incorrect" + bcolors.ENDC)
            return ip_address

        if config_ok:
   	                  net_connect.send_command_expect('write memory')
            print(str(ip_address) + " - " + "Config saved")

        net_connect.disconnect()
    except:
        print(str(ip_address)+" - "+bcolors.FAIL+"Cannot connect to this device."+bcolors.ENDC)
        return ip_address
    print(str(ip_address) + " - " + bcolors.OKGREEN + "Router configured sucessfully" + bcolors.ENDC)


if __name__ == '__main__':

    # Enter valid username and password. Note password is blanked out using the getpass library
    global_username = input("Enter Username: ")
    global_password = getpass()

    try:
        f = open('ip.txt')
        connection_data=[]
        filelines = f.read().splitlines()
        for line in filelines:
            if line == "": continue
            if line[0] == "#": continue
            conn_data = line.split(',')
            ipaddr=conn_data[0].strip()
            username=global_username
            password=global_password
            if len(conn_data) > 1 and conn_data[1].strip() != "": username = conn_data[1].strip()
            if len(conn_data) > 2 and conn_data[2].strip() != "": password = conn_data[2].strip()
            connection_data.append((ipaddr, username, password))
        f.close()
    except:
        sys.exit("Couldn't open or read file ip.txt")

    multiprocessing.set_start_method("spawn")
    with multiprocessing.Pool(maxtasksperchild=10) as process_pool:
        routers_with_issues = process_pool.map(doRouter, connection_data, 1)  # doRouter - function, iplist - argument
        process_pool.close()
        process_pool.join()

    print("\n")
    print("#These routers weren't configured#")

    failed_file = open('fail.txt', 'w')
    for item in routers_with_issues:
        if item != None:
          failed_file.write("%s\n" % item)
          print(item)

    #Completing the script and print running time
    print("\n")
    print("#This script has now completed#")
    print("\n")
    print("--- %s seconds ---" % (time.time() - start_time))

После подготовки скрипта выполнить его можно из командной строки или из PyCharm CE. Из командной строки запускаем командой:

python script.py

Я рекомендую пользоваться PyCharm CE. Там создаём новый проект, файл Python (File → New…) и вставляем в него наш скрипт. В папку со скриптом кладем файл ip.txt и запускаем скрипт (Run → Run)

Получаем следующий результат:

bash ~/PycharmProjects/p4ne $ python3 script.py 
Enter Username: cisco
Password: 
Invalid IP - 10.1.1.256
127.0.0.1 - Cannot connect to this device.
1.1.1.1 - Cannot connect to this device.
10.10.100.227 - Static routes added
10.10.100.227 - Config saved
10.10.100.227 - Router configured sucessfully
10.10.31.170 - WARNING - couldn't find static route to 8.8.8.8
10.10.31.170 - WARNING - couldn't find next-hop IP address 
10.10.31.170 - Routes weren't added because config is incorrect
2.2.2.2 - Cannot connect to this device.


#These routers weren't configured#
10.1.1.256
127.0.0.1
217.112.31.170
1.1.1.1
2.2.2.2


#This script has now completed#

Пару слов о том, как отладить скрипт. Легче всего это делать в PyCharm. Отмечаем строчку, на которой хотим остановить выполнение скрипта, и запускаем выполнение в режиме отладки. После того, как скрипт остановится, можно будет посмотреть текущие значения всех переменных. Проверить, что передаются и принимаются корректные данные. Кнопками «Step Into» или «Step Into My Code» можно пошагово продолжить выполнение скрипта.



Ограничения описанной версии скрипта:

• тестировался только в Python 3
• не умеет обрабатывать ситуацию, когда вы в первый раз подключаетесь к маршрутизатору и получаете вопрос вида:
  
  The authenticity of host '11.22.33.44 (11.22.33.44)' can't be established.
RSA key fingerprint is SHA256:C+BHaMBjuMIoEewAbjbQbRGdVkjs&840Ve3z4aJo.
Are you sure you want to continue connecting (yes/no)?

Этот скрипт был написан для решения конкретных задач. Однако он универсален и, надеюсь, поможет ещё кому-нибудь в работе. А самое главное — послужит первым шагом в освоении Python.

При написании скрипта использовались следующие ресурсы:

• https://github.com/ktbyers/netmiko — страница библиотеки netmiko на GitHub. Там есть документация и примеры.
• https://pynet.twb-tech.com/blog/automation/netmiko.html — ещё один пример с netmiko
• https://docs.Python .org/3/library/multiprocessing.html — описание и примеры библиотеки multiprocessing

Александр Гаршин, ведущий инженер-проектировщик систем передачи данных компании «Инфосистемы Джет»