pyscaniptonetbox/pyscaniptonetbox.py

#!/usr/bin/env python3
import sys
import yaml
import argparse

from pathlib import Path
from pprint import pprint

import networkscan
import requests
import pynetbox
import ipaddress
import itertools

parser = argparse.ArgumentParser(prog="pyscaniptonetbox",description="Network scan an IPV4 subnet to add to netbox IPAM",
                                formatter_class=argparse.ArgumentDefaultsHelpFormatter)

parser.add_argument("tenant", help="tenant/company short name")
parser.add_argument("subnet",help="IPV4 subnet to scan")
args = parser.parse_args()

config = vars(args)
print(config)

tenant_name=config['tenant']

def get_settings():
    full_file_path = Path(__file__).parent.joinpath('local_settings.yaml')
    with open(full_file_path) as settings:
        settings_data = yaml.safe_load(settings)
    return settings_data

settings = get_settings()
nb_url = settings['netbox']['url']
nb_token = settings['netbox']['auth_token']

print("Netbox server: " + nb_url)
#print("Netbox token: " + nb_token)

nb = pynetbox.api(nb_url,token=nb_token)
print("version: -" + nb.version + "-")

tenant_records = nb.tenancy.tenants.filter(name=tenant_name)
if len(tenant_records) == 1:
    t = tenant_records.__next__()
    #pprint(dict(t))
else:
    print("Failed to retrieve exactly one record for tenant:" + tenant_name)
    print("Valid tenant names are:")
    all_tenants = nb.tenancy.tenants.all()
    for t in all_tenants:
        print("  " + t.name)

    sys.exit(1)

tenant_id = t.id
print("tenant id: " + str(tenant_id))
sites = nb.dcim.sites.filter(tenant_id=tenant_id)
if len(sites) == 0:
    print("warning - no site configured for this tenant.")  ;#may not matter
elif len(sites) == 1:
    s = sites.__next__()
else:
    print("Note - more than one site for this tenant. Script may need adjusting to accept '-site' argument")
    #do we even need the site(s)?
    s = sites.__next__()


print("Tenant:" + t.name + " has " + str(t.ipaddress_count) + " ip addresses in netbox")
print("- number of IP prefixes: " + str(t.prefix_count))
prefixes = nb.ipam.prefixes.filter(tenant_id=tenant_id)

def  get_prefix_attribute(pfx):
     return pfx.prefix

prefixes, prefixes_b, prefixes_c  = itertools.tee(prefixes,3)
#strprefix_list =list(tmp_prefixes_1)

for p in prefixes_b:
    print(p)
    #pprint(dict(p))

prefix_list = [get_prefix_attribute(i) for i in prefixes_c]
#print(prefix_list)

scan_net = ipaddress.IPv4Network(config['subnet'])
supernet = None

if config['subnet'] in prefix_list:
    print("Ok - entered prefix directly allocated to " + tenant_name)
else:
    print(" - prefix '" + config['subnet'] + " is not directly allocated to this tenant.. ")


print(" checking if  entered subnet is active and a member/subnet of allocated prefixes...")
is_subnet = False
for p in prefixes:
    pfx = str(p.prefix)
    status = str(p.status)
    #print(str(p.status) + " " +  str(p.display))
    if status == "Active":
        print("active - " + pfx)
        try: 
            pfxnetwork = ipaddress.IPv4Network(pfx)
        except Exception:
            #presumably ipv6
            continue
        
        #print("comparing to:" + pfxnetwork.exploded)
        if scan_net.subnet_of(pfxnetwork):
            is_subnet = True
            supernet = pfxnetwork
            #for now we are assuming last found *active* supernet is the smallest.. todo - check review!

if is_subnet:
    print("OK - found (active) prefix " + supernet.exploded +  " that contains this range")
    strprefixlen = str(supernet.prefixlen)
    prefix_is_writable = True
else:
    print("No suitable (active) prefix found for " + tenant_name + " which contains subnet " + scan_net.exploded)
    strprefixlen = str(scan_net.prefixlen)
    prefix_is_writable = False  ;#we'll do the scan - but won't write to netbox
    #note that netbox will not stop us writing to a non-Active prefix unless the token is locked down with complex rules
    #sys.exit(2)


if prefix_is_writable:
    answer = input("network " +  supernet.exploded + "  Type 'n' to cancel,  'y' to scan only, 'update' to scan and enter IPs into netbox  (n/y/update)")
else:
    answer = input("network " +  scan_net.exploded + "  Type 'n' to cancel,  'y' to scan only  (n/y)")

do_update = False
if any(answer.lower() == f for f in ['yes','y']):
    do_scan = True
elif any(answer.lower() == f for f in ['no', 'n']):
    do_scan = False
elif any(answer.lower() == f for f in ['update']):
    do_scan = True
    if prefix_is_writable:
        do_update = True
else:
    do_scan = False

if do_scan:
    scanner = networkscan.Networkscan(scan_net.exploded)
    scanner.run()
    for i in scanner.list_of_hosts_found:
        print(i)
        if prefix_is_writable & do_update:
            try:
                result = nb.ipam.ip_addresses.create(
                    address = str(i) + "/" + strprefixlen,
                    vrf = 1,
                    tenant = t.id,
                    description = "loaded by pyscaniptonetbox"
                )
                print("  added: " + str(i) + "/" + strprefixlen)
                #todo - add mac-address to custom field if 
                # a) there is no interface to assign it to
                # b) we are on the same subnet and can even get a mac-address
            except pynetbox.RequestError as e:
                print(e.error)

    print("Number of hosts found: " + str(scanner.nbr_host_found))


print("--END--")
sys.exit(0)