Junos PyEZ на примере задачи по поиску свободных подсетей ipv4

Статья о работе с Junos PyEZ — “Python microframework that enables you to manage and automate devices running Junos OS” автоматизация и управление, все как мы любим. Написание скрипта описанного в этой статье преследовало несколько целей — изучение Python и автоматизация задач по сбору информации или изменения конфигурации на оборудовании под управлением Junos OS. Выбор именной этой связки Python + Junos PyEZ был сделан из-за низкого порога вхождения в язык программирования Python и простоты использования библиотеки Junos PyEZ, которая не требует экспертных знаний Junos OS.

Задача

Аудит свободных подсетей ipv4 принадлежащих компании. Критерием того, что подсеть свободна — является отсутствие записи о ней в маршрутах на коммутаторе выполняющем роль маршрутизатора под управлением Junos OS.

Реализация

Python + Junos PyEZ, хотя был соблазн сделать через paramiko и ssh.exec_command, как следствие понадобится на опрашиваемом оборудовании настроить протокол сетевого управления устройствами netconf. Netconf работает с оборудованием посредством удаленного вызова процедур (remote procedure call RPC) и использует XML, в рассматриваемом примере, для предоставления полученной информации.

Установка текущей версии Junos PyEZ из PyPI, выполняется следующей командой:

$ pip install junos-eznc

Можно установить также из основной ветки проекта на GitHub следующей командой:

$ pip install git+https://github.com/Juniper/py-junos-eznc.git

И еще один вариант через

$ pip install -r requirements.txt 

эта команда установит отсутствующие в системе библиотеки необходимые для работы. В моей версии requirements.txt их всего две, версии указаны последние на момент написания скрипта:

junos-eznc netaddr

Скрипт по умолчанию берет имя текущего пользователя в системе, залогиниться под именем другого пользователя можно используя ключ show_route.py -u <user_name> getpass.getpass принимает пароль из stdin так пароль не останется в системе. Для подключения к оборудованию также понадобится ввести по запросу его hostname или ip-адрес. Все необходимые для авторизации на устройстве данные получены.

Junos PyEZ поддерживает подключение к оборудованию под управлением Junos OS используя консоль, telnet или netconf через ssh. В статье рассмотрен последний вариант.

Для подключения к оборудованию используется класс Device модуля jnpr.junos

with jnpr.junos.Device(host=router,                            user=args.name,                            passwd=password) as dev:

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

data = dev.rpc.get_route_information()

Аналогичная команда на Junos OS

user@router> show route | display xml

Добавив в конец команды rpc, получим тег запроса и можем сопоставить его с именем метода RPC, таким способом можно узнать и другие интересующие имена. Стоит отметить, что синтаксис написания тега запроса отличается от имени метода, а именно следует заменить знаки дефиса на нижние подчеркивание.

user@router> show route | display xml rpc <rpc-reply xmlns:junos="http://xml.juniper.net/junos/15.1R1/junos">     <rpc>         <get-route-information>         </get-route-information>     </rpc> </rpc-reply>

Данные о маршрутах получил в xml формате из них выбрал только интересующие меня по тегу <rt-destination>xxx.xxx.xxx.xxx/yy</rt-destination> и записал в переменную в виде списка в строковом формате, таким образом получив список занятых подсетей.

route_list = data.xpath("//rt-destination/text()")

Остальную часть обернул в цикл while, чтобы не выполнять повторно запрос на роутер, если надо будет проверить в другой подсети из тех, о которых роутер уже знает. Стоит упомянуть, что роутер на котором запрашиваю знает маршруты только через OSPF, поэтому для пограничного роутера лучше изменить немного запрос, чтобы сократить время работы скрипта

data = dev.rpc.get_ospf_route_information()

Теперь обратимся к содержимому цикла while

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

tmp = re.search(r'^%s\S*' % subnet_search, route_list[i])

Через IPNetwork, модуля netaddr, получаю подсети в виде списка ipv4 адресов

range_subnet = netaddr.IPNetwork(tmp.group(0))

Используя IPNetwork из введенной пользователем сети с маской получаю диапазон адресов и формирую список всех адресов из этого диапазона для сравнения со списком занятых адресов.

for i in set(net_list).difference(set(busyip)):         freeip.append(i)

Полученный список свободных адресов вывожу в виде подсетей

print(netaddr.IPSet(freeip))

Ниже приведен скрипт полностью, тестировался на коммутаторах используемых в роли маршрутизатора, модели ex4550, ex4600

 #!/usr/bin/env python3 # -*- coding: utf-8 -*- import argparse import getpass import netaddr import re import sys  import jnpr.junos  parser = argparse.ArgumentParser() parser.add_argument('-u', '--user',                     action='store',                     dest='name',                     help='Enter login from tacacs if it differs from the '                          'username in the system.') args = parser.parse_args()  if not args.name:     args.name = getpass.getuser()    # Return the “login name” of the user. router = input("Full routers name: ") password = getpass.getpass("Password: ")  try:     # Authenticates to a device running Junos, for get information about routs     # into xml format and selects by tag.     route_list = []     with jnpr.junos.Device(host=router,                            user=args.name,                            passwd=password) as dev:         data = dev.rpc.get_route_information()     route_list = data.xpath("//rt-destination/text()") except (jnpr.junos.exception.ConnectRefusedError,         jnpr.junos.exception.ConnectUnknownHostError) as err:     print("Equipment name or password wrong.")     sys.exit(1)  while True:     subnet = input("Net with mask: ")     subnet_search = input("Input no more three octet: ")     # Gets a list of busy IP addresses from the received subnets.     busyip = []     for i in range(len(route_list)):         tmp = re.search(r'^%s\S*' % subnet_search, route_list[i])         if tmp:             range_subnet = netaddr.IPNetwork(tmp.group(0))             for ip in range_subnet:                 busyip.append("%s" % ip)     range_subnet = netaddr.IPNetwork(subnet)     # Gets list ip adresses from subnetworks lists.     net_list = []     for ip in range_subnet:         net_list.append("%s" % ip)     # Сomparing lists.     freeip = []     for i in set(net_list).difference(set(busyip)):         freeip.append(i)     print(netaddr.IPSet(freeip))      request = input("To run request again enter yes or y, "                     "press 'enter', complete request: ")     if request in ("yes", "y"):         continue     else:         print('Bye')         break