From 353dbba3c004a2dfbbbb42bfbc694783d3291716 Mon Sep 17 00:00:00 2001
From: Felix Richter <Felix.Richter@syntax-fehler.de>
Date: Sun, 16 Jan 2011 07:38:58 +0100
Subject: Adds the arping script, as well as snmp-test
added Readme and stuff,
have a nice day.
---
README.md | 14 ++++++++++
arping.py | 50 ++++++++++++++++++++++++++++++++++
main.py | 1 +
snmp_users.py | 87 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
4 files changed, 152 insertions(+)
create mode 100644 README.md
create mode 100644 arping.py
create mode 100644 main.py
create mode 100755 snmp_users.py
diff --git a/README.md b/README.md
new file mode 100644
index 00000000..d28100d3
--- /dev/null
+++ b/README.md
@@ -0,0 +1,14 @@
+SNMP Users
+==========
+
+asks an snmp-router for its arp-list and tries to verify this list via
+ARPING. The snmping is done via snmp-net and command line parsing,
+the arping uses 'scapy'.
+
+This script needs superuser rights and otherwise will just skip the
+verification
+
+SNMPWALK Command
+===============
+
+snmpwalk -c shammunity 10.42.0.1 1.3.6.1.2.1.3.1.1.2
diff --git a/arping.py b/arping.py
new file mode 100644
index 00000000..60f271d2
--- /dev/null
+++ b/arping.py
@@ -0,0 +1,50 @@
+#!/usr/bin/python2
+
+import logging
+log = logging.getLogger('pingy')
+import os
+try:
+ if (os.geteuid() != 0):
+ raise Exception('no root permissions')
+ from scapy.all import *
+
+ def arpingy(iprange="10.42.1.0/24",iface='eth0'):
+ log.debug("pinging"+ iprange)
+ """Arping function takes IP Address or Network, returns nested mac/ip list"""
+ try:
+ conf.verb=0
+ ans,unans=arping(iprange,iface=iface,timeout=1)
+
+ collection = []
+ for snd, rcv in ans:
+ result = rcv.sprintf(r"%ARP.psrc% %Ether.src%").split()
+ log.debug(result)
+ collection.append(result)
+ return collection
+ except Exception as e:
+ print ("something went wrong while arpinging " + str(e))
+ return []
+
+ def pingy(ip="10.42.1.0/24",iface='eth0'):
+ log.debug("pinging"+ ip)
+ """Arping function takes IP Address or Network, returns nested mac/ip list"""
+ try:
+ conf.verb=0
+ ans,unans=srp(Ether()/IP(dst=ip)/ICMP(),timeout=1)
+
+ collection = []
+ for snd, rcv in ans:
+ result = rcv.sprintf(r"%IP.src% %Ether.src%").split()
+ log.debug(result)
+ collection.append(result)
+ return collection
+ except Exception as e:
+ print ("something went wrong while arpinging " + str(e))
+ return []
+
+except Exception as e:
+ log.error("Cannot Arping!" + str(e))
+ def pingy(iprange):
+ return True
+ def arpingy(iprange):
+ return True
diff --git a/main.py b/main.py
new file mode 100644
index 00000000..0e1d5dde
--- /dev/null
+++ b/main.py
@@ -0,0 +1 @@
+#!/usr/bin/python2
diff --git a/snmp_users.py b/snmp_users.py
new file mode 100755
index 00000000..df78cc8d
--- /dev/null
+++ b/snmp_users.py
@@ -0,0 +1,87 @@
+#!/usr/bin/python2
+import subprocess,re
+import logging
+from arping import arpingy,pingy
+log = logging.getLogger('snmp_users')
+
+class snmp_users:
+ mac_list = {}
+
+ def __init__(self,server='10.42.0.1',community='shammunity'):
+ self.server = server
+ self.community = community
+
+ def call_external(self):
+ """ creates a list of lines produced by snmpwal
+ """
+ out = subprocess.Popen(
+ ['snmpwalk',
+ '-v2c',
+ '-c',self.community,
+ self.server,
+ '1.3.6.1.2.1.3.1.1.2'],
+ stdout=subprocess.PIPE).communicate()[0]
+ return out.split('\n')
+
+ def parse_output(self,output):
+ """ parses output lines produced by snmpwalk"""
+ for i in output:
+ if i == '':
+ continue
+
+ ip,mac = re.sub(r'.*\.(\d+\.\d+\.\d+\.\d+) = Hex-STRING: ([ 0-9A-F]*) ',
+ r'\1 : \2',i).split(' : ')
+
+ verified = False
+ if self.verify_data(mac,ip,arpingy(ip)):
+ log.info("verfied by arping %s => %s" % (mac,ip))
+ verified = True
+
+ #elif self.verify_data(mac,ip,pingy(ip) ): builds up onto arp'ing...
+ # log.info("verfied by fallback ping %s => %s" % (mac,ip))
+ # verfied = True
+
+ if verified:
+ if mac in self.mac_list.keys():
+ log.info( "%s already assigned for ip %s (new IP %s) " %
+ ( mac ,self.mac_list[mac], ip))
+ self.mac_list[mac]['counter'] +=1
+ continue
+ else:
+ log.debug("%s => %s" % (mac,ip))
+ self.mac_list[mac] = { 'ip' : ip, 'counter' : 1}
+ else:
+ log.warning("Verification failed %s => %s" % (mac,ip))
+
+ def verify_data(self,mac,ip,ret):
+ """ Verifies ip and mac via ARP Scan """
+ mac = ':'.join(mac.split())
+ log.debug(ret)
+
+ if type(ret) == type(True):
+ return True
+ elif ret != []:
+ ret = ret[0]
+ ret_mac = ret[1].upper()
+ if ret_mac == mac:
+ return True
+ else:
+ log.info('return mac not equal to expected, %s != %s' % (ret_mac,mac))
+ else:
+ return False
+
+ def collect(self):
+ output = self.call_external()
+ self.parse_output(output)
+ self.print_results()
+ def print_results(self):
+ print '\n'.join([ mac + " => " + ip['ip'] + " ( %d active leases )" %
+ ip['counter'] for mac,ip in self.mac_list.items() ])
+ print '%d *unique* nodes in network' % len(self.mac_list)
+
+
+if __name__ == "__main__":
+ logging.basicConfig(level=logging.INFO)
+ a = snmp_users()
+ a.collect()
+
--
cgit v1.2.3