#!/usr/bin/env python import sys import getopt import asyncio import puresnmp import time import socket from enum import IntEnum import warnings warnings.filterwarnings('ignore') #from requests.exceptions import ConnectionError agent_version="1.1.0" def checkadv(string): return string def uptime(u): return str(u.total_seconds()) def voltage(v): return str('{:.2f}'.format(v/1000))+" V" def quote(string): return '"'+str(string)+'"' fdict = { 'c': checkadv, 'u': uptime, 'v': voltage, '"': quote } def get_func(string): if string in fdict: return fdict[string] return None oids_advantech = [ # OID, section, subsection, prefix, number, helpers ( "1.3.6.1.2.1.1.1", None, None, None, False, None), # SysDescr ( "1.3.6.1.2.1.1.2", None, None, None, False, "c"), # prodtype (must be 1.3.6.1.4.1.30140.* for advantech) ( "1.3.6.1.2.1.1.3", "STAT", "various", "uptime=", False, 'u"'), # sysUpTime (/100 seconds) ( "1.3.6.1.2.1.1.5", "STAT", "various", "snmpname=", False, '"'), # SysName ( "1.3.6.1.4.1.30140.3.3", "STAT", "status sys", "Temperature : ", True, None), # temperature ( "1.3.6.1.4.1.30140.3.4", "STAT", "status sys", "Supply Voltage : ", True, "v"), # Voltage (mV) ( "1.3.6.1.4.1.30140.3.6", "STAT", "status sys", "CPU Usage : ", True, None), # CPU usage (%) ( "1.3.6.1.4.1.30140.3.9", "STAT", "various", "memavail=", True, None), # RAM free (bytes) ( "1.3.6.1.4.1.30140.3.10", "STAT", "various", "memtotal=", True, None), # RAM total (bytes) ( "1.3.6.1.4.1.30140.4.5", "STAT", "status mobile", "Signal Strength : ", True, None), # Signal strength for cell (dBm) ( "1.3.6.1.4.1.30140.4.19", "STAT", "status mwan", "SIM Card : ", True, None), # SIM card selected ( "1.3.6.1.4.1.30140.4.20", "STAT", "status mwan", "IP Address : ", False, None), # mobile IP ( "1.3.6.1.4.1.30140.4.26", "STAT", "status mobile", "Signal Strength : ", True, None), # Signal quality of a cell ( "1.3.6.1.4.1.30140.4.27", "STAT", "status mobile", "Signal Quality : ", True, None), # Signal strength (CSQ) ( "1.3.6.1.4.1.30140.6.1", "STAT", "status sys", "Product Name : ", False, None), # Product ( "1.3.6.1.4.1.30140.6.2", "STAT", "status sys", "Firmware Version : ", False, None), # Firmware version ( "1.3.6.1.4.1.30140.6.3", "STAT", "status sys", "Serial Number : ", False, None), # Serial NO ( "1.3.6.1.4.1.30140.6.4", "LONGSTAT", "status module", "IMEI : ", False, None), # IMEI ( "1.3.6.1.4.1.30140.6.7", "LONGSTAT", "status module", "ICCID : ", False, None), # ICCID ] def genlinestore(data): linestore=[] for section in data: secdata=[] if None in data[section]: for item in data[section][None]: secdata.append(item) else: print("None is not in section", str(section)) for subsection in data[section]: if subsection is None: print("Skipping None subsection in section ", str(section)) continue print("Processing subsection", str(subsection), "in section", str(section)) subsecdata=[] for item in data[section][subsection]: subsecdata.append(item) # add subsection header subsecdata.insert(0, ":---> " + str(subsection) + ";" + str(len(subsecdata)) + " <---:") # add resulting list into section list secdata.extend(subsecdata) # add section header secdata.insert(0, ":>>>> " + str(section) + ";" + str(len(secdata)) + " <<<<:") linestore.extend(secdata) return linestore def addstuff(mydict, key1, key2, value): if key1 is not None: if key1 in mydict: if key2 in mydict[key1]: mydict[key1][key2].append(value) else: mydict[key1][key2] = [value] else: mydict[key1] = { key2 : [value] } async def getdevice(target_ip, oidmap, creds, protov2c=True): resdict = {} mycreds = None if protov2c: mycreds=puresnmp.V2C(creds) else: mycreds=puresnmp.V1(creds) client = puresnmp.PyWrapper(puresnmp.Client(ip=target_ip, port=161, credentials=mycreds)) client.client.configure(retries=2,timeout=4) #return await asyncio.gather(client.get("1.3.6.1.2.1.1.1.0"),client.bulkget(scalar_oids, repeating_oids, list_max)) missed=0 if protov2c: try: result = await client.bulkget([item[0] for item in oidmap], []) except puresnmp.exc.NoSuchOID as e: missed += 1 print(result) else: result = puresnmp.util.BulkResult({},{}) for i in oidmap: try: result.scalars[puresnmp.ObjectIdentifier(i[0]+".0")] = await client.get(i[0]+".0") except puresnmp.exc.NoSuchOID as e: missed += 1 print("Items:", len(oidmap), "success:", len(result.scalars), "failed:", str(missed)) for item in oidmap: mykey=puresnmp.ObjectIdentifier(item[0]+".0") if mykey in result.scalars: outval=result.scalars[mykey] if isinstance(outval, bytes): outval=outval.decode() if item[5] is not None: for char in item[5]: myf=get_func(char) if myf is not None: outval=myf(outval) if item[1] is not None: addstuff(resdict, item[1], item[2], item[3]+str(outval)) addstuff(resdict, "VARS", None, "g_appversion="+str(agent_version)) addstuff(resdict, "VARS", None, 'g_mysystem="SNMP device"') addstuff(resdict, "VARS", None, "g_machid=\""+str(target_ip)+";snmp;"+str(target_ip)+";snmp\"") return resdict def print_help(): print("... here be the help ...") target="localhost" cstring="public" runv2c=False try: opts, args = getopt.getopt(sys.argv[1:], '2t:c:') for opt, arg, in opts: if opt == "-2": runv2c = True if opt == "-t": target = str(arg) if opt == "-c": cstring = str(arg) except getopt.GetoptError: print("Error when parsing cmdline arguments from: ", ' '.join(sys.argv)) print_help() sys.exit(1) print("Running SNMP against", target, "with community string \""+cstring+"\" using V2C:", runv2c) start_time = time.time() try: myres = asyncio.run(getdevice(target, oids_advantech, cstring, runv2c)) print(myres) x = genlinestore(myres) for item in x: print(item) except puresnmp.exc.Timeout as e: print("Target", str(target), "timed out.") except socket.error as e: print("Target", str(target), "refuses connection.") finally: tot_time = time.time()-start_time print("Runtime was:", tot_time)