Статья о работе с 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