Skip to content

Instantly share code, notes, and snippets.

@jondkelley
Created April 6, 2018 20:34
Show Gist options
  • Save jondkelley/e1a9be03413eb34d7dbb80fb8cf58e53 to your computer and use it in GitHub Desktop.
Save jondkelley/e1a9be03413eb34d7dbb80fb8cf58e53 to your computer and use it in GitHub Desktop.
#!/usr/bin/python2.7
# -*- coding: utf-8 -*-
# Written by Jonathan Kelley, March 7, 2018
# March 12, Added config support from yaml..
#pip2.7 install netmiko
#pip2.7 install --upgrade paramiko
#pip2.7 install cryptography
#pip2.7 uninstall gssapi
#pip2.7 install unidecode
###pip2.7 install yamlcfg
#pip2.7 install prettytable
###pip2.7 install configloader
#pip2.7 install loadconfig
#pip3.4 install netmiko
#pip3.4 install unidecode
# 1015 pip3.4 install cryptography
# 1016 vi /bin/carphunter
# 1017 pip3.4 install yamlcfg
# 1018 vi /bin/carphunter
# 1019 pip3.4 install prettytable
# 1020 vi /bin/carphunter
# 1021 pip3.4 install loadconfig
import sys
from netmiko import ConnectHandler
from datetime import date, datetime
from unidecode import unidecode
import re
import json
from pprint import pprint
import prettytable
import argparse
from loadconfig import Config
import time
parser = argparse.ArgumentParser(description="Cisco Arp Hunter v0.1", epilog="This Python based tool uses Netmiko to search your Cisco® branded routers and switches. This tool creates an ARP cache in a local json file (using --poll). --mac and -ip can be used to find MAC, IP, PORT and VLAN associations immediately against this json cache.")
parser.add_argument('-m', '--mac', action="store", dest="mac", help="Search json cache for MAC to port/vlan mappings")
parser.add_argument('-i', '--ip', action="store", dest="ip", help="Search json cache for IP address to MAC mapping")
parser.add_argument('--poll', action="store_true", default=False, dest="pollmode", help="Poll devices now and save to local json-cache-file")
parser.add_argument('--json', action="store_true", default=False, dest="json", help="Print output in JSON format")
parser.add_argument('--all-ports', action="store_true", default=False, dest="allports", help="List all ports in cache search including port channels")
parser.add_argument('--config', action="store", default="/etc/carphunter.yml", dest="configfile", help="Non-default config file location")
parser.add_argument('--json-cache-file', action="store", default="/mnt/data/arp.json", dest="cachedbfile", help="Non-default json-cache-file location")
args = parser.parse_args()
class ConfigHelper(object):
"""
helper class to load configuration properties from local file
"""
def __init__(self, path):
conf = "!include {}".format(path)
config = Config(conf)
self.configuration = Config(conf)
@property
def raw(self):
"""
returns raw configuration as python object, not really for external use.
"""
return self.configuration
@property
def global_logins(self):
"""
return tuple of user/password combintation for global switch login privileges.
"""
return (self.configuration['global']['user'], self.configuration['global']['password'])
@property
def switches(self):
"""
return a list of configuration properties for switching devices.
"""
return self.configuration['devices']['switches']
@property
def routers(self):
"""
return a list of configuration properties for routing devices.
"""
return self.configuration['devices']['routers']
conf = ConfigHelper(path=args.configfile)
# netmiko device constants
a_device = {
'device_type': 'cisco_ios',
'verbose': False,
}
class DateTimeEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, datetime):
return o.isoformat()
return json.JSONEncoder.default(self, o)
class Cache():
"""
updates or loads a network json database object
"""
def __init__(self):
"""
basic minimum datastructure of json class db
"""
self.db = {}
self.db['routers'] = {}
self.db['switches'] = {}
def add_switch(self, switchname):
"""
add a new switch dictionary
"""
self.db['switches'][switchname] = {}
def add_switch_entry(self, switchname, mac, mode, interface, vlan):
"""
adds a new switch entry to class db
"""
self.db['switches'][switchname][mac] = {}
self.db['switches'][switchname][mac]['mode'] = mode
self.db['switches'][switchname][mac]['interface'] = interface
self.db['switches'][switchname][mac]['vlan'] = vlan
def add_router(self, routername):
"""
add a new router dictionary
"""
self.db['routers'][routername] = {}
def add_router_entry(self, routername, ip, mac, proto, age, vlan):
"""
adds a new router entry to class db
"""
self.db['routers'][routername][ip] = {}
self.db['routers'][routername][ip]['ip'] = ip
self.db['routers'][routername][ip]['mac'] = mac
self.db['routers'][routername][ip]['proto'] = proto
self.db['routers'][routername][ip]['age'] = age
self.db['routers'][routername][ip]['vlan'] = vlan
def time_clock(self, timetaken):
"""
records time taken for poll into database
"""
self.runtime = str(timetaken)
@property
def timetaken(self):
"""
return last poller run time
"""
return self.runtime
def write_to_file(self):
"""
writes the database to a file
"""
db = {
'db': self.db,
'lastUpdate': datetime.now(),
'pollTime': self.runtime
}
with open(args.cachedbfile, 'w') as outfile:
json.dump(db, outfile, cls=DateTimeEncoder)
@property
def load_from_file(self):
"""
loads the database from file
"""
return ""
def device_command(device, command):
"""
connects to a network device and issues a command, returns decoded split output
"""
net_connect = ConnectHandler(**device)
output = net_connect.send_command(command)
return unidecode(output).split("\n")
def poll_devices():
"""
initiates a polling process and saves all device data to disk for cached searches in json
"""
cache = Cache()
start_benchmark = datetime.now()
for router, v in conf.routers.iteritems():
print("Show config from router {} ({})".format(router, v.get("name", "No name defined")))
a_device['ip'] = router
a_device['username'] = conf.global_logins[0]
a_device['password'] = conf.global_logins[1]
output = device_command(a_device, "show config")
epoch = str(time.time()).split('.')[0]
filename="/mnt2/network/router.{}.{}.timestamp{}.config.bak".format(v.get("name", "no-name"), router, epoch)
with open(filename, 'w') as file:
for item in output:
if "password 7" in item:
item = "! password 7 <CENSORED>"
elif "enable password" in item:
item = "! enable password <CENSORED>"
elif "enable secret" in item:
item = "! enable secret <CENSORED>"
elif "privilege 15 " in item:
item = "! privilege 15 user <CENSORED>"
elif "privilege 14 " in item:
item = "! privilege 14 user <CENSORED>"
elif "privilege 13 " in item:
item = "! privilege 13 user <CENSORED>"
elif "privilege 12 " in item:
item = "! privilege 12 user <CENSORED>"
elif "privilege 11 " in item:
item = "! privilege 11 user <CENSORED>"
elif "privilege 10 " in item:
item = "! privilege 10 user <CENSORED>"
elif "privilege 9 " in item:
item = "! privilege 9 user <CENSORED>"
elif "privilege 8 " in item:
item = "! privilege 8 user <CENSORED>"
elif "privilege 7 " in item:
item = "! privilege 7 user <CENSORED>"
elif "privilege 6 " in item:
item = "! privilege 6 user <CENSORED>"
elif "privilege 5 " in item:
item = "! privilege 5 user <CENSORED>"
elif "privilege 4 " in item:
item = "! privilege 4 user <CENSORED>"
elif "privilege 3 " in item:
item = "! privilege 3 user <CENSORED>"
elif "privilege 2 " in item:
item = "! privilege 2 user <CENSORED>"
file.write("%s\n" % item)
for switch, v in conf.switches.iteritems():
print("Show config from switch {} ({})".format(switch, v.get("name", "No name defined")))
a_device['ip'] = switch
a_device['username'] = v.get("user", conf.global_logins[0])
a_device['password'] = v.get("password",conf.global_logins[1])
output = device_command(a_device, "show config")
epoch = str(time.time()).split('.')[0]
filename="/mnt2/network/switch.{}.{}.timestamp{}.config.bak".format(v.get("name", "no-name"), switch, epoch)
with open(filename, 'w') as file:
for item in output:
if "password 7" in item:
item = "! password 7 <CENSORED>"
elif "enable password" in item:
item = "! enable password <CENSORED>"
elif "enable secret" in item:
item = "! enable secret <CENSORED>"
elif "privilege 15 " in item:
item = "! privilege 15 user <CENSORED>"
elif "privilege 14 " in item:
item = "! privilege 14 user <CENSORED>"
elif "privilege 13 " in item:
item = "! privilege 13 user <CENSORED>"
elif "privilege 12 " in item:
item = "! privilege 12 user <CENSORED>"
elif "privilege 11 " in item:
item = "! privilege 11 user <CENSORED>"
elif "privilege 10 " in item:
item = "! privilege 10 user <CENSORED>"
elif "privilege 9 " in item:
item = "! privilege 9 user <CENSORED>"
elif "privilege 8 " in item:
item = "! privilege 8 user <CENSORED>"
elif "privilege 7 " in item:
item = "! privilege 7 user <CENSORED>"
elif "privilege 6 " in item:
item = "! privilege 6 user <CENSORED>"
elif "privilege 5 " in item:
item = "! privilege 5 user <CENSORED>"
elif "privilege 4 " in item:
item = "! privilege 4 user <CENSORED>"
elif "privilege 3 " in item:
item = "! privilege 3 user <CENSORED>"
elif "privilege 2 " in item:
item = "! privilege 2 user <CENSORED>"
file.write("%s\n" % item)
end_benchmark = datetime.now()
total_time = end_benchmark - start_benchmark
def print_statistics(data, total_time):
print("\nStatistics:")
print(" * Cache contains {} routers, {} switches".format(len(data['db']['routers']),len(data['db']['switches'])))
print(" * Cache updated {}\n * Last poller took {}".format(data['lastUpdate'], data['pollTime']))
print(" * This lookup took {}".format(total_time))
print("Tool written by: Jonathan Kelley")
def json_serial(obj):
"""JSON serializer for objects not serializable by default json code"""
if isinstance(obj, (datetime, date)):
return obj.isoformat()
raise TypeError ("Type %s not serializable" % type(obj))
def print_json(dictionary):
print(json.dumps(dictionary, indent=2, default=str))
def search_arp_table(search, jsondata=False):
"""
searches the json cache for an arp entry
"""
start_benchmark = datetime.now()
if not jsondata:
print("Checking network for `{}`...".format(search))
data = json.load(open(args.cachedbfile))
jsontable_router = []
for router, arpitem in data['db']['routers'].iteritems():
print_results = 0
#print router, arp
table = prettytable.PrettyTable(["hwaddr", "ip", "vlan", "age", "proto"])
for ip, values in arpitem.iteritems():
if not values['vlan']:
values['vlan'] = "N/A"
if search in values['mac']:
table.add_row([values['mac'], ip, values['vlan'], values['age'], values['proto']])
jsontable_router.append({
ip : {
'mac': values['mac'],
'vlan': values['vlan'],
'age': values['age'],
'proto': values['proto']
}
})
print_results = 1
if print_results and not jsondata:
print("\nRouting Device: {} found match!".format(router))
print("{}".format(table))
jsontable_switch = []
for switch, arpitem in data['db']['switches'].iteritems():
print_results = 0
table = prettytable.PrettyTable(["hwaddr", "interface", "vlan", "mode"])
for mac, values in arpitem.iteritems():
if search in mac:
if not args.allports:
if "/" in values['interface']:
table.add_row([mac, values['interface'], values['vlan'], values['mode']])
jsontable_switch.append({
mac : {
'interface': values['interface'],
'vlan': values['vlan'],
'mode': values['mode']
}
})
print_results = 1
else:
table.add_row([mac, values['interface'], values['vlan'], values['mode']])
jsontable_switch.append({
mac : {
'interface': values['interface'],
'vlan': values['vlan'],
'mode': values['mode']
}
})
print_results = 1
if print_results and not jsondata:
print("\nSwitching Device: {} found match!".format(switch))
print("{}".format(table))
end_benchmark = datetime.now()
total_time = end_benchmark - start_benchmark
if jsondata:
print_json({ 'routers': jsontable_router,
'switches': jsontable_switch,
'seekTime': total_time
})
else:
return print_statistics(data, total_time)
def ip2arp(searchip,jsondata=False):
"""
find arpa in json cache based on ip address
"""
start_benchmark= datetime.now()
if not jsondata:
print("Checking network for `{}` ARPA entry...".format(searchip))
data = json.load(open(args.cachedbfile))
jsondata = {}
for router, arpitem in data['db']['routers'].iteritems():
print_results = 0
exact_match = 0 # if exact match is found, hide ambiguous results
#print router, arp
jsondata[router] = {}
table = prettytable.PrettyTable(["ip", "hwaddr", "vlan", "age", "proto"])
for ip, values in arpitem.iteritems():
if values['vlan'] == '':
values['vlan'] = "?"
if searchip == values['ip']:
table.add_row([ip, values['mac'], values['vlan'], values['age'], values['proto']])
print_results = 1
exact_match = 1
elif searchip in values['ip']:
if not exact_match:
table.add_row([ip, values['mac'], values['vlan'], values['age'], values['proto']])
print_results = 1
if print_results:
print("\nRouting Device: {} found a matching IP!".format(router))
print("{}\n".format(table))
print("")
end_benchmark = datetime.now()
total_time = end_benchmark - start_benchmark
if not jsondata:
print_statistics(data, total_time)
def user_confirm(message=None,default_return=False):
"""
asks the user y or n, then returns a true/false for processing
arguement message (string): optional message to print before asking
arguement default_return (bool): optional default if return pressed
"""
if message:
sys.stdout.write(message)
sys.stdout.write(" Type 'yes' or 'no': ")
#python2 choice = raw_input().lower()
choice = input().lower()
if "y" in choice:
return True
elif "n" in choice:
return False
elif "" in choice:
return default_return
else:
return None
if "__main__" == __name__:
# arg parse
if len(sys.argv) == 1:
choice = user_confirm("WARNING : No arguements will display ALL NETWORK DATA, are you sure?\n")
if choice:
pass
elif choice == False:
print("Exiting...")
exit(1)
if args.pollmode:
# --poll flag
print("Collection polling started. Do not terminate or data will not be saved.")
poll_devices()
else:
print("Invalid selection")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment