Junos PyEZ pe exemplul unei sarcini de căutare a subrețelelor libere ipv4

Articol despre lucrul cu Junos PyEZ — „microframework Python care vă permite să gestionați și să automatizați dispozitive care rulează Junos OS” automatizare și gestionare, tot ce ne place. Scrierea scriptului descris în acest articol a avut mai multe scopuri — studiul Python și automatizarea sarcinilor de colectare a informațiilor sau modificarea configurației pe echipamentele care rulează Junos OS. Alegerea acestei combinații Python + Junos PyEZ a fost făcută datorită plafonului de intrare scăzut în limbajul de programare Python și a ușurinței de utilizare a bibliotecii Junos PyEZ, care nu necesită cunoștințe de expert în Junos OS.

Sarcină

Auditul subrețelelor ipv4 libere aparținând companiei. Criteriul prin care o subrețea este considerată liberă este absența unei înregistrări despre aceasta în rutele de pe switch-ul care îndeplinește rolul de rutare sub Junos OS.

Implementarea

Python + Junos PyEZ, deși a fost tentant să facem prin paramiko și ssh.exec_command, în consecință, va fi necesar să configurăm pe echipamentele interogate protocolul de gestionare a rețelei netconf. Netconf interacționează cu echipamentele prin apeluri de procedură la distanță (remote procedure call RPC) și utilizează XML, în exemplul analizat, pentru a furniza informațiile obținute.

Instalarea versiunii curente Junos PyEZ din PyPI se realizează cu următoarea comandă:

$ pip install junos-eznc

De asemenea, poate fi instalat din ramura principală a proiectului de pe GitHub cu următoarea comandă:

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

Și încă o variantă prin

$ pip install -r requirements.txt 

această comandă va instala librăriile lipsă din sistem necesare funcționării. În versiunea mea requirements.txt sunt în total două, versiunile fiind cele mai recente în momentul scrierii scriptului:

junos-eznc
netaddr

Scriptul, implicit, preia numele utilizatorului curent din sistem, autentificarea sub un alt nume de utilizator se poate face folosind cheia show_route.py -u getpass.getpass acceptă parola din stdin, astfel parola nu va rămâne în sistem. De asemenea, va fi necesar să introduceți, la solicitare, hostname-ul sau adresa ip a echipamentului. Toate datele necesare pentru autorizarea pe dispozitiv au fost obținute.

Junos PyEZ suportă conexiuni la echipamentele care rulează Junos OS folosind consolă, telnet sau netconf prin ssh. În articol este analizată ultima variantă.

Pentru conectarea la echipamente, se utilizează clasa Device din modulul jnpr.junos

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

Se efectuează o cerere pentru toate rutele cunoscute de router prin apeluri la proceduri externe sau așa cum le numesc unii.

data = dev.rpc.get_route_information()

Comanda echivalentă pe Junos OS

user@router> show route | display xml

Adăugând la sfârșitul comenzii rpc, vom obține eticheta cererii și putem să o asociem cu numele metodei RPC, astfel putem afla și alte nume interesante. Merită menționat că sintaxa scrierii etichetei cererii diferă de numele metodei, și anume semnele de minus ar trebui înlocuite cu underscore.

user@router> show route | display xml rpc
route_list = data.xpath("//rt-destination/text()")

Restul am împachetat într-un ciclu while, pentru a nu efectua din nou o cerere către router, dacă va trebui să verific într-o altă subrețea din cele pe care routerul deja le cunoaște. Merită să menționăm că routerul de pe care fac cererea cunoaște rutele doar prin OSPF, așa că pentru routerul de frontieră este mai bine să modificăm puțin cererea, pentru a reduce timpul de execuție al scriptului.

data = dev.rpc.get_ospf_route_information()

Acum să ne îndreptăm atenția către conținutul ciclului while

La început, utilizatorului i se va propune să introducă subrețeaua cu masca și nu mai mult de trei octete din rețeaua acelei subrețele, aceasta este necesară pentru a stabili intervalul de căutare. Nu îmi place foarte mult această implementare a setării criteriului și intervalului de căutare, dar momentan nu am găsit o soluție mai bună. Apoi, din lista obținută de subrețele, folosind variabila care conține nu mai mult de trei octete, aleg subrețelele de interes.

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

Prin IPNetwork, modulul netaddr, obțin subrețelele sub formă de listă de adrese ipv4

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

Folosind IPNetwork din rețeaua introdusă de utilizator cu masca, obțin intervalul de adrese și formez o listă cu toate adresele din acest interval pentru a le compara cu lista de adrese ocupate.

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

Lista obținută de adrese libere o afișez sub formă de subrețele

print(netaddr.IPSet(freeip))

Mai jos este prezentat scriptul complet, testat pe switch-uri folosite în rol de router, modelele 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'^%sS*' % 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

Sursa: habr.com