#!/usr/bin/env python import itertools import sys import getopt import asyncio import puresnmp import time import socket #from multiprocessing import Process, connection, current_process, Queue import multiprocessing import multiprocessing.managers #from multiprocessing.managers import BaseManager, NamespaceProxy from enum import IntEnum from threading import Event import matelib import warnings import fcntl import os import sys import signal import MySQLdb import subprocess import configparser import logging import logging.handlers from datetime import datetime warnings.filterwarnings('ignore') appname="Dwarfguard SNMP Gateway" appversion="0.3.0" agent_version="1.1.0" # knobs (defaults) # standalone variable below: True: ignore DB, get settings from cfgfile below default_standalone=False cfgfile="dwarfg_snmp_gw.ini" logfile="dwarfg_snmp_gw.log" loglevel=logging.WARNING default_daemon=True device_snmp_timeout = 4 device_snmp_retries = 2 lockfile_fn = "/tmp/snmp_gateway.lock" # do not touch below this line dbc = None stop = Event() use_logfile = False single_action = 0 dev_dbid = None dev_addr = None sql_getcfg = "SELECT intval, strval FROM product_conf WHERE name = %s;" sql_get_list = "SELECT id, address, community_string, protocol_version, last_success, last_failure, errors_since_success, registered_as FROM snmp_devices;" sql_get_new = "SELECT id, address, community_string, protocol_version, errors_since_success, registered_as, devtoken_ag_copy, last_success FROM snmp_devices WHERE registered_as is NULL" sql_get_registered = "SELECT id, address, community_string, protocol_version, errors_since_success, registered_as, devtoken_ag_copy, last_success FROM snmp_devices WHERE registered_as is NOT NULL" sql_predelete = "SELECT id FROM snmp_devices WHERE id = %s;" sql_delete = "DELETE FROM snmp_devices WHERE id = %s;" sql_add = "INSERT INTO snmp_devices(address, community_string, protocol_version) VALUES (%s, %s, %s)" sql_predelete_by_addr = "SELECT id FROM snmp_devices WHERE address = %s;" sql_delete_by_addr = "DELETE FROM snmp_devices WHERE address = %s;" sql_reg_push_ok = "UPDATE snmp_devices SET protocol_version = %s, last_success = %s, errors_since_success = %s, registered_as = %s, registered_as_devid = %s, devtoken_ag_copy = %s WHERE id = %s;" data_reg_push_ok = [] sql_reg_push_fail = "UPDATE snmp_devices SET protocol_version = %s, last_failure = %s, errors_since_success = %s, registered_as = %s, registered_as_devid = %s, devtoken_ag_copy = %s WHERE id = %s;" data_reg_push_fail = [] sql_old_push_ok = "UPDATE snmp_devices SET protocol_version = %s, last_success = %s, errors_since_success = %s WHERE id = %s;" data_old_push_ok = [] sql_old_push_fail = "UPDATE snmp_devices SET protocol_version = %s, last_failure = %s, errors_since_success = %s WHERE id = %s;" data_old_push_fail = [] mylogger = None settings = { # for 'Server' and 'Parameters': the list specifies [ value, DB key, convert to int? ] 'Server': { 'server_address': ['', 'ExternalURL', False], 'servid': ['', None, False] }, 'Parameters': { 'standalone' : [default_standalone, None, False], 'daemon': [default_daemon, None, False], 'run_ssl': [True, None, False], 'threads' : [1, 'SNMPGatewayThreads', True], 'batch' : [16, 'SNMPGatewayBatch', True], 'snmp_timeout' : [4, 'SNMPGatewayTimeout', True], 'snmp_retries' : [2, 'SNMPGatewayRetries', True], 'interval' : [240, 'SNMPGatewayInterval', True], 'logfile' : [logfile, None, False], 'loglevel' : [loglevel, 'SNMPGatewayLoglevel', True], 'ssl_override' : [0, 'SNMPSSLOverride', True] }, 'base_defs': {}, 'enforced': {}, } def stop_handler(sig, fr): global stop stop.set() def dump_settings(sig, fr): talk(str(settings), logging.WARNING) # use class instances to store / access device data class DEVICE(object): def __init__(self, dbid, addr, cstring, protov, errs, devid_num, token, any_success): self.delete = False self.dbid = dbid self.addr = addr self.cstring = cstring self.protov = protov self.errors = errs self.devid_num = devid_num self.token = token self.any_success = any_success self.ok_regis = None self.ok_snmp = None self.ok_push = None self.data = None def __str__(self): tlen = 0 dlen = 0 tdlen = 0 if self.token is not None: tlen = len(self.token) if self.data is not None: dlen = len(self.data) for i in self.data: tdlen += len(i) return f"Device {self.addr} SNMPv{self.protov} with \"{self.cstring}\" oper R/S/P:{self.ok_regis}/{self.ok_snmp}/{self.ok_push} devid:{self.devid_num} errors:{self.errors} len T/D/TD:{tlen}/{dlen}/{tdlen}" def pr(self): return self.__str__() def check(self): allowed_chars=set('0123456789.:abcdefghijklmnopqrstuvwxyz-') return set(self.addr) <= allowed_chars class CustomManager(multiprocessing.managers.BaseManager): pass class CustomProxy(multiprocessing.managers.NamespaceProxy): _exposed_ = ('__getattribute__', '__setattr__', '__delattr__', 'error', 'check', 'pr') def pr(self): callmethod = object.__getattribute__(self, '_callmethod') return callmethod('pr') def check(self): callmethod = object.__getattribute__(self, '_callmethod') return callmethod('check') # TODO in UI: when deleting the device by user, the snmp_devices records must be deleted as well - those that sports the same registered_as, e.g. DELETE FROM snmp_devices WHERE registered_as = # TODO in UI: way to insert SNMP devices by batch # TODO finish support for standalone mode (.ini file) def note(msg, *args, **kwargs): if logging.getLogger().isEnabledFor(logging.NOTE): logging.log(logging.NOTE, msg) def verbose(msg, *args, **kwargs): if logging.getLogger().isEnabledFor(logging.VERBOSE): logging.log(logging.VERBOSE, msg) def talk(string, mloglevel = logging.DEBUG, logger = None): global loglevel if logger is None: if mloglevel >= logging.ERROR: print(string, file=sys.stderr) else: logger.log(mloglevel, string) # ------ HELPERS -------- def checktype(string): if string is not None and string.startswith("1.3.6.1.4.1.30140"): return 2 return 6 def uptime(u): return str(u.total_seconds()) def voltage(v): return str('{:.2f}'.format(v/1000))+" V" def activesim(s): if isinstance(s, int): if 2 == int(s): return "3rd" if 1 == int(s): return "2nd" return "1st" def ignoreNA(s): if s is not None: if s == "N/A" or s == "n/a": return None return s def quote(string): return '"'+str(string)+'"' fdict = { 'c': checktype, 'u': uptime, 'v': voltage, 's': activesim, 'i': ignoreNA, '"': 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", "STAT", "various", "snmphint=", True, "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 : ", False, 's'), # 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 Quality : ", True, None), # Signal quality of a cell ( "1.3.6.1.4.1.30140.4.27", "STAT", "status mobile", "CSQ : ", True, None), # Signal strength (CSQ) ( "1.3.6.1.4.1.30140.4.30", "STAT", "status mobile", "RSSI : ", True, None), # Signal strength (RSSI) ( "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, "i"), # 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: pass for subsection in data[section]: if subsection is None: continue 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, oid, key1, key2, prefix, value, dumplog): myoid = oid if oid is None: oid = "Agent-matic" oper = " added as " if key1 is None: oper = " ignored as " msg = value if prefix is not None: msg = str(prefix) + str(value) dumplog.append("OID " + oid + oper + "'" + msg + "'") if key1 is not None: if key1 in mydict: if key2 in mydict[key1]: mydict[key1][key2].append(msg) else: mydict[key1][key2] = [msg] else: mydict[key1] = { key2 : [msg] } def writeout(device_ip, extension, stuff, logger): err=0 fname = settings['Parameters']['logfile'][0] + "_" + str(device_ip) + "." + str(extension) try: with open(fname, "w", encoding='utf-8') as f: try: for item in stuff: f.write(str(item) + "\n") except (IOError, OSError): print("Error writing to file " + fname, file=sys.stderr) err += 1 except (FileNotFoundError, PermissionError, OSError): print("Error opening file " + fname, file=sys.stderr) err += 1 if 0 == err: talk("Written " + str(extension) + " for " + str(device_ip) + " into " + str(fname), logging.VERBOSE, logger) async def getdevice(target_ip, oidmap, creds, logger, protov2c=True): resdict = {} dumplog = [] 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=device_snmp_retries,timeout=device_snmp_timeout) missed=0 if protov2c: try: result = await client.bulkget([item[0] for item in oidmap], []) except puresnmp.exc.NoSuchOID as e: missed += 1 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 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 outval is not None: addstuff(resdict, item[0], item[1], item[2], item[3], str(outval), dumplog) addstuff(resdict, None, "VARS", None, None, "g_appversion="+str(agent_version), dumplog) addstuff(resdict, None, "VARS", None, None, 'g_mysystem="SNMP device"', dumplog) addstuff(resdict, None, "VARS", None, None, "g_machid=\""+get_snmp_machid(target_ip)+"\"", dumplog) if settings['Parameters']['loglevel'][0] <= 15 and settings['Parameters']['logfile'][0] is not None: writeout(str(target_ip), "snmp", dumplog, logger) return resdict def get_snmp_machid(target): return str(target)+";snmp;snmp;snmp" def print_help(): print(appname, appversion, "is a SNMP Gateway that goes through the defined devices and reformats the data before sending it to the server.") print("\tIt basically substitutes a system agent for devices that are unable to run the agent but can provide some (limited) information over SNMP protocol.") print("\tAt present, it supports getting some basic SNMP data from the Advantech router devices.") print("\nGateway supports (on top of single actions) a few modes:") print("\t(1) Running directly on the server machine as a system service (using systemd)") print("\t\tThis mode is automated but it can only reach devices directly accessible from the server.") print("\t(2) Running on a standalone machine sending data to a remote server") print("\t\tIn this mode you need to provide configuration via a standalone INI file as it cannot access server database.") print("\t\tAlso, the addresses of the devices you want to scan needs to be provided/added via command line once so that they can be stored in the embedded DB.") print("\nExamples:") print("\tRunning on a server machine:") print("\t\tTo start: Do nothing. It is started automatically.") print("\t\tTo add new device or remove existing one: Use GUI.") print("\tRunning on a standalone machine:") print("\t\tOne-shot scan:", sys.argv[0], "--standalone --oneshot") print("\t\tService: create systemd configuration file (check the provided example), using following options: --standalone --daemon") print("\t\tIn this mode, you can run the Gateway as a oneshot script (scan all devices once, exit) or as a service from systemd.") print("\nSupported parameters:") print("\t\t--standalone ... use everytime when running outside of application server") print("\t\t--daemon ... use for continuous run (must use superviser, gateway does NOT daemonize itself)") print("\t\t--oneshot ... perform one sweep over the SNMP devices and exit. Log is put on standard outpu unless --logfile option is used.") print("\t\t--interval ... minimal number of seconds for one sweep (if continuous and cycle took less, sleep until interfal matched)") print("\t\t--snmp_timeout ... wait seconds for SNMP response") print("\t\t--snmp_retries ... attempt connections to a device until device considered failed") print("\t\t--threads ... run using threads (expedites larger fleets of SNMP devices)") print("\t\t--batch ... handle SNMP devices in one thread cycle (expedites larger fleets of SNMP devices)") print("\t\t--logfile ... use for logile.") print("\t\t--loglevel ... set minimal loglevel (messages with lower level will not be shown). 15 = VERBOSE, 20 = INFO, 25 = NOTE (default), 30 = WARNING, 40 = ERROR, 50 = CRITICAL only") print("\t\t--ssl ... enforce SSL connection, regardless of what is defined in base_defs and DB config. Useful for debugging, do not use otherwise (set correct config in DB instead).") print("\t\t--nossl ... enforce non-SSL connection, regardless of what is defined in base_defs and DB config. Useful for debugging, do not use otherwise (set correct config in DB instead).") print("\t\t-f|--force ... remove lockfile (use with caution!)") print("\t\t-h|--help ... show help") print("\nSingle actions (Gateway exits after performing the action):") print("\t\t-l|--list ... list all SNMP device requests with some details") print("\t\t--add_device ... add new SNMP device request. You NEED to quote this if providing string or version as ';' character is command delimiter in shell. Use IP address or DNS name of the device. Example: --add_device \"my_device;nonpublic;1\" OR --add_device 10.20.30.40") print("\t\t--del_device ... delete device with id (when numeric) or address (when not) from SNMP device requests") async def device_snmp_get(device, logger): try: myres = await getdevice(device.addr, oids_advantech, device.cstring, logger, device.protov == 2) if myres is None: device.ok_snmp = False device.errors += 1 return False device.data = genlinestore(myres) if settings['Parameters']['loglevel'][0] <= 15 and settings['Parameters']['logfile'][0] is not None: writeout(str(device.addr), "push", device.data, logger) device.ok_snmp = True device.errors = 0 return True except puresnmp.exc.Timeout as e: talk(" ---> device " + str(device.addr) + " timed out.", logging.INFO, logger) except socket.error as e: talk(" ---> device " + str(device.addr) + " refuses connection.", logging.INFO, logger) if device.protov == 2 and device.any_success == False: talk(" ---> degrading device SNMP protocol version from 2 to 1", logging.WARNING, logger) device.protov = 1 device.ok_snmp = False device.errors += 1 return False def device_register(device, logger): reslist = [] if matelib.register(reslist, matelib.Devtype.SNMPDEVICE, 1, server = None, debug = not settings['Parameters']['daemon'][0] and settings['Parameters']['loglevel'][0] <= 15, use_ssl = settings['Parameters']['run_ssl'][0], servid = settings['Server']['servid'][0], silent = settings['Parameters']['loglevel'][0] >= 30, prefdevid = None, initmachids = [get_snmp_machid(device.addr)], superdebug = not settings['Parameters']['daemon'][0] and settings['Parameters']['loglevel'][0] <= 10, iccid = None, emulate_iccid = False) and len(reslist) > 0: device.ok_regis = True device.errors = 0 device.devid_num = matelib.devid_int(reslist[0][0]) device.token = reslist[0][3] else: device.ok_regis = False device.errors += 1 return device.ok_regis def fwd_devdata_to_dg(device, logger): if device.data is None: device.ok_push = False device.errors += 1 return False push_code, dwarfg_code, _ = matelib.push_data(matelib.devid_str(device.devid_num), matelib.Devtype.SNMPDEVICE, get_snmp_machid(device.addr), device.token, iccid = None, dict_keep = {}, data = device.data, use_ssl = settings['Parameters']['run_ssl'][0], debug = not settings['Parameters']['daemon'][0] and settings['Parameters']['loglevel'][0] <= 15, superdebug = not settings['Parameters']['daemon'][0] and settings['Parameters']['loglevel'][0] <= 10, silent = settings['Parameters']['loglevel'][0] >= 30) # handle push or dwarfg error if push_code: if isinstance(dwarfg_code, int): if 0 == dwarfg_code: device.ok_push = True device.errors = 0 return True elif dwarfg_code in (matelib.dgvars.md['aerr_notex'], matelib.dgvars.md['aerr_serfu'], matelib.dgvars.md['aerr_neser'], matelib.dgvars.md['aerr_comme'], matelib.dgvars.md['aerr_invda']): talk("Device " + str(device.addr) + " not exist on server, to be dropped.", logging.ERROR, logger) device.ok_push = False device.delete = True else: device.ok_push = False talk("Non-ok app code from push: " +str(dwarfg_code) + " for device " +str(device_addr), logging.INFO, logger) else: device.ok_push = False talk("Unexpected app code from push: " + str(dwarfg_code) + " for device " + str(device.addr), logging.NOTE, logger) else: device.ok_push = False talk("Push failed for device " + str(device.addr) + " code is " + str(dwarfg_code), logging.NOTE, logger) return True def read_devices(logger): new_devices, old_devices_good, old_devices_bad = [], [], [] if settings['Parameters']['standalone'][0]: pass # TODO get from sqlite (standalone mode) else: # get new (to-be-registered) SNMP devices dbc.execute(sql_get_new) # device markers - SNMP, register, data pushed new_devices = [manager.DEVICE(a[0], a[1], a[2], a[3], a[4], a[5], a[6], a[7] is not None) for a in dbc.fetchall() ] matelib.mdb.commit() # get registered devices dbc.execute(sql_get_registered) old_devices = dbc.fetchall() matelib.mdb.commit() # split devices into behaving and misbehaving old_devices_good = [] old_devices_bad = [] # split in two lists based on errors_since_success -> well-behaving and misbehaving for device in old_devices: if device[4] == 0: old_devices_good.append(manager.DEVICE(device[0], device[1], device[2], device[3], device[4], device[5], device[6], device[7] is not None)) else: old_devices_bad.append(manager.DEVICE(device[0], device[1], device[2], device[3], device[4], device[5], device[6], device[7] is not None)) talk("Read " + str(len(new_devices)+len(old_devices_good)+len(old_devices_bad)) + " devices from DB: new/nice/noty: " + str(len(new_devices)) + "/" + str(len(old_devices_good)) + "/" + str(len(old_devices_bad)), logging.INFO, logger) return (new_devices, old_devices_good, old_devices_bad) async def process_single_device(device, logger): if not device.check(): talk("Invalid device address \"" + str(device.addr) + "\", skipping device", logging.WARNING, logger) return if stop.is_set(): return if await device_snmp_get(device, logger): if stop.is_set(): return if device.token is None: if device_register(device, logger): talk("Device: " + str(device.addr) + " ... register passed, next is push", logging.VERBOSE, logger) else: talk("Device: " +str(device.addr) + " ... registration failed.", logging.NOTE, logger) return else: talk("Device: " + str(device.addr) + " ... SNMP passed, next is push", logging.VERBOSE, logger) fwd_devdata_to_dg(device, logger) async def run_tasks(devlist, logger): tasks = [] async with asyncio.TaskGroup() as tg: for dev in devlist: tasks.append(tg.create_task(process_single_device(dev, logger))) def thread_handle_devices(devlist, q): qh = logging.handlers.QueueHandler(q) root = logging.getLogger() root.setLevel(settings['Parameters']['loglevel'][0]) root.addHandler(qh) name = multiprocessing.current_process().name talk("Starting worker " + multiprocessing.current_process().name + " for " + str(len(devlist)) + " devices at " + datetime.now().strftime('%H:%M:%S'), logging.VERBOSE, root) asyncio.run(run_tasks(devlist, root)) talk("Worker " + multiprocessing.current_process().name + " finished at " + datetime.now().strftime('%H:%M:%S'), logging.VERBOSE, root) def process_devices(new_devices, old_devices_good, old_devices_bad, queue): factors = [2, 1, 5] counters = [0, 0, 0] groups = [new_devices, old_devices_good, old_devices_bad] pool = {} while len(groups[0]) + len(groups[1]) + len(groups[2]) +sum(counters) > 0: # spin up the processes talk("Process counters are " + str(counters[0]) + "/" + str(counters[1]) + "/" + str(counters[2]), logging.VERBOSE) while len(groups[0]) + len(groups[1]) + len(groups[2]) > 0 and sum(counters) < settings['Parameters']['threads'][0]: # I can run another process, count priorities priorities = [len(groups[i]) * factors[i] for i in range(3)] talk("Priorities for new process are " + str(priorities[0]) + "/" + str(priorities[1]) + "/" + str(priorities[2]), logging.VERBOSE) for i in range(3): if counters[i] > 0: priorities[i] /= counters[i] # select the group which has the highest priority target_group = priorities.index(max(priorities)) # prep the devices list for the new process if len(groups[target_group]) < settings['Parameters']['batch'][0]: target_list = groups[target_group] groups[target_group] = [] else: target_list = groups[target_group][0:settings['Parameters']['batch'][0]] groups[target_group] = groups[target_group][settings['Parameters']['batch'][0]:] # run the new process talk("Scheduled " + str(len(target_list)) + " devices from group " + str(target_group), logging.VERBOSE) proc = multiprocessing.Process(target=thread_handle_devices, args=(target_list, queue)) proc.start() pool[proc.sentinel] = (proc, target_group) # increase the appropriate counter counters[target_group] += 1 # wait for at least one to finish ready = multiprocessing.connection.wait(pool[p][0].sentinel for p in pool) # check if we were signalled to stop if stop.is_set(): return # determine which ones ended and decrease the counters for finished in ready: counters[pool[finished][1]] -= 1 del pool[finished] def update_db(new_devices, old_devices_good, old_devices_bad): # prep the DB data now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') data_reg_push_ok, data_reg_push_fail, data_old_push_ok, data_old_push_fail = [], [], [], [] for device in itertools.chain(new_devices, old_devices_good, old_devices_bad): talk("DB process device " + device.pr(), logging.VERBOSE) if device.delete: try: dbc.execute(sql_delete, (device.dbid)) matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to delete SNMP device " + str(device.addr) + " error was: " + str(e), logging.ERROR) matelib.mdb.rollback() else: if device.ok_snmp is not None or device.ok_regis is not None: if device.ok_regis == True: if device.ok_snmp: data_reg_push_ok.append((device.protov, now, device.errors, device.devid_num, matelib.devid_str(device.devid_num), device.token, device.dbid)) else: data_reg_push_fail.append((device.protov, now, device.errors, device.devid_num, matelib.devid_str(device.devid_num), device.token, device.dbid)) elif device.ok_snmp == True: data_old_push_ok.append((device.protov, now, device.errors, device.dbid)) else: data_old_push_fail.append((device.protov, now, device.errors, device.dbid)) # execute the data for item in data_reg_push_ok: try: dbc.execute(sql_reg_push_ok, item) talk("Executed SQL: " + str(dbc._last_executed), logging.VERBOSE) matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to update DB data for newly registered devices (1), error was: " + str(e), logging.ERROR) matelib.mdb.rollback() for item in data_reg_push_fail: try: dbc.execute(sql_reg_push_fail, item) talk("Executed SQL: " + str(dbc._last_executed), logging.VERBOSE) matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to update DB data for newly registered devices (2), error was: " + str(e), logging.ERROR) matelib.mdb.rollback() for item in data_old_push_ok: try: dbc.execute(sql_old_push_ok, item) talk("Executed SQL: " + str(dbc._last_executed), logging.VERBOSE) matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to update DB data for existing devices (1), error was: " + str(e), logging.ERROR) matelib.mdb.rollback() for item in data_old_push_fail: try: dbc.execute(sql_old_push_fail, item) talk("Executed SQL: " + str(dbc._last_executed), logging.VERBOSE) matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to update DB data for existing devices (2), error was: " + str(e), logging.ERROR) matelib.mdb.rollback() talk("New/Old-ok/Old-failed: " + str(len(data_reg_push_ok) + len(data_reg_push_fail)) + "/" + str(len(data_old_push_ok)) + "/" + str(len(data_old_push_fail)), logging.VERBOSE) def single_act(mode): if mode == 1: arr = dev_addr.split(';') if 0 == len(arr) or 0 == len(arr[0]): talk("You must provide at least an address of the added SNMP device.", logging.ERROR) else: if 1 == len(arr): talk("Adding 'public' as community string.", logging.NOTE) arr.append('public') if 2 == len(arr): talk("Using 2 as SNMP protocol version.", logging.NOTE) arr.append(2) # adding the device - need address, community string, protocol version try: talk("Inserting device with address: '" + str(arr[0]) + "', community string '" + str(arr[1]) + "' using SNMP protocol version " + str(arr[2]), logging.WARNING) dbc.execute(sql_add, (arr[0], arr[1], arr[2])) matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to add new SNMP device, error was: " + str(e), logging.ERROR) matelib.mdb.rollback() matelib.mdb_disconnect() elif mode == 2: if dev_dbid is not None: sql_pre = sql_predelete sql = sql_delete tok = (str(dev_dbid),) else: sql_pre = sql_predelete_by_addr sql = sql_delete_by_addr tok = (dev_addr,) try: dbc.execute(sql_pre, tok) result = dbc.fetchall() if dbc.rowcount != 1: talk("Found " + str(dbc.rowcount) + " devices matching the delete pattern. Cannot delete.", logging.ERROR) matelib.mdb.commit() return matelib.mdb.commit() try: dbc.execute(sql, tok) matelib.mdb.commit() talk("Device " + tok[0] + " has been deleted from SNMP device table. If the device is registered, delete it also from registered devices using UI. In the rear case the SNMP Gateway is running and processing the device in the meantime, the device can reappear. To be 100% sure about the deletion, stop SNMP Gateway prior to device deletion.", logging.WARNING) except MySQLdb.Error as e: talk("Failed to delete SNMP device " + str(dev_dbid) + " error was: " + str(e), logging.ERROR) matelib.mdb.rollback() except MySQLdb.Error as e: talk("Failed to get devices matching deletion pattern. Error was: " + str(e), logging.ERROR) matelib.mdb.rollback() matelib.mdb_disconnect() elif mode == 3: try: dbc.execute(sql_get_list) result = dbc.fetchall() print("id / address / community string / protocol version / last success / last failure / errors since success / registered as") for item in result: print(*item, sep=' / ') matelib.mdb.commit() except MySQLdb.Error as e: talk("Failed to get devices from DB. Error was: " + str(e), logging.ERROR) matelib.mdb.rollback() matelib.mdb_disconnect() else: talk("Unsupported mode for single-action: " +str(mode), logging.ERROR) def get_queued_logger(q): logger = logging.getLogger() myhand = logging.handlers.QueueHandler(q) logger.setLevel(settings['Parameters']['loglevel'][0]) logger.addHandler(myhand) return logger def listener_process(queue, fname = None): root = logging.getLogger() if fname is None: h = logging.StreamHandler() else: h = logging.FileHandler(fname, encoding='utf-8', mode='a') f = logging.Formatter(fmt="[{asctime}/{levelname}] {message}", datefmt="%m-%d %H:%M:%S", style="{") h.setFormatter(f) root.addHandler(h) while True: try: record = queue.get() if record is None: # We send this as a sentinel to tell the listener to quit. break logger = logging.getLogger(record.name) logger.handle(record) # No level or filter logic applied - just do it! except Exception: import sys, traceback print('Whoops! Problem:', file=sys.stderr) traceback.print_exc(file=sys.stderr) # --- program start --- if __name__ == '__main__': signal.signal(signal.SIGINT, stop_handler) signal.signal(signal.SIGTERM, stop_handler) signal.signal(signal.SIGHUP, dump_settings) # handle options - only once at startup try: opts, args = getopt.getopt(sys.argv[1:], 'fhl', ['help', 'force', 'ssl', 'nossl', 'standalone', 'daemon', 'oneshot', 'snmp_timeout=', 'snmp_retries=', 'threads=', 'logfile=', 'interval=', 'loglevel=', 'add_device=', 'del_device=', 'list']) for opt, arg, in opts: if opt == "-f" or opt == "--force": os.remove(lockfile_fn) if opt == "--ssl": settings['enforced']['run_ssl'] = True if opt == "--nossl": settings['enforced']['run_ssl'] = False if opt == '-h' or opt == '--help': print_help() sys.exit(0) if opt == '-l' or '--list' == opt: single_action = 3 if opt == "--loglevel": loglevel=int(arg) settings['enforced']['loglevel'] = int(arg) if opt == "--standalone": settings['enforced']['standalone'] = True if opt == "--daemon": settings['enforced']['daemon'] = True if opt == "--oneshot": settings['enforced']['daemon'] = False if opt == "--snmp_timeout": settings['enforced']['snmp_timeout'] = int(arg) if opt == "--snmp_retries": settings['enforced']['snmp_retries'] = int(arg) if opt == "--interval": settings['enforced']['interval'] = int(arg) if opt == "-t" or opt == "--threads": settings['enforced']['threads'] = int(arg) if opt == "-l" or opt == "--logfile": settings['enforced']['logfile'] = str(arg) if opt == "--add_device": dev_addr = str(arg) single_action = 1 if opt == "--del_device": if arg.isnumeric(): dev_dbid = int(arg) else: dev_addr = str(arg) single_action = 2 except getopt.GetoptError: talk("Error when parsing cmdline arguments from: " + ' '.join(sys.argv), logging.CRITICAL) sys.exit(1) # initialize settings if settings['Parameters']['standalone'][0]: # read configuration from file filecfg = configparser.ConfigParser() filecfg.read(sys.path[0] + '/' + cfgfile) for sect in ['Server', 'Parameters']: for key in settings[sect]: if key in filecfg[sect]: if settings[sect][key][2]: settings[sect][key][0] = int(filecfg[sect][key]) else: settings[sect][key][0] = filecfg[sect][key] # now copy base_defs for key in filecfg['base_defs']: settings['base_defs'][key] = filecfg['base_defs'][key] # initialize matelib matelib.init(alt_base_defs_dict = settings['base_defs'], brand_short="dwarfg", brand_long="Dwarfguard", brand_opt=1, brand_email="dwarfg-info@dwarftech.cz", brand_web="www.dwarfguard.com", brand_manuf="Dwarf Technologies") # TODO init sqlite (standalone mode) else: # initialize matelib matelib.init(brand_short="dwarfg", brand_long="Dwarfguard", brand_opt=1, brand_email="dwarfg-info@dwarftech.cz", brand_web="www.dwarfguard.com", brand_manuf="Dwarf Technologies") # read siteconf (DB PWD) matelib.dgvars.get_siteconf() # connect to DB if not matelib.mdb_connect(): raise Exception("Unable to connect to database!") dbc = matelib.mdb.cursor() # get the programmatically-acessible config for sect in ['Server', 'Parameters']: for key in settings[sect].keys(): if settings[sect][key][1] is not None: try: dbc.execute(sql_getcfg, [settings[sect][key][1]]) matelib.mdb.commit() result = dbc.fetchall() if len(result) > 0 and len(result[0]) > 0: if settings[sect][key][2]: settings[sect][key][0] = int(result[0][0]) else: settings[sect][key][0] = result[0][1] else: talk("Configuration value " + str(settings[sect][key][1]) + " not found in DB.", logging.ERROR) except MySQLdb.Error as e: talk("Failed to get CFG value " + str(settings[sect][key][1]) + " from DB. Error was: " + str(e), logging.ERROR) # get the rest of the config - servid settings['Server']['servid'][0] = matelib.dgvars.md['servid'] # create additional logging levels logging.NOTE = 25 logging.addLevelName(logging.NOTE, "NOTE") logging.Logger.note = note logging.VERBOSE = 15 logging.addLevelName(logging.VERBOSE, "VERBOSE") logging.Logger.verbose = verbose # setup between-processes shared class CustomManager.register('DEVICE', DEVICE, CustomProxy) manager = CustomManager() manager.start() # construct logfile path if not a path already if not '/' in settings['Parameters']['logfile'][0]: if settings['Parameters']['standalone'][0]: settings['Parameters']['logfile'][0] = sys.path[0] + '/' + settings['Parameters']['logfile'][0] else: settings['Parameters']['logfile'][0] = matelib.dgvars.md['logdir'] + '/' + settings['Parameters']['logfile'][0] # now apply the enforced parameters (from cmdline) for key in settings['enforced']: if settings['Parameters'][key][2]: settings['Parameters'][key][0] = int(settings['enforced'][key]) else: settings['Parameters'][key][0] = settings['enforced'][key] # setup logging based on daemon or not if settings['Parameters']['daemon'][0]: # daemon -> log into file use_logfile = True else: # not a daemon -> log into stdout only unless logfile specifically requested if 'logfile' in settings['enforced']: use_logfile = True queue = multiprocessing.Queue(-1) if use_logfile: listener = multiprocessing.Process(target=listener_process, args=(queue, settings['Parameters']['logfile'][0])) else: listener = multiprocessing.Process(target=listener_process, args=(queue, None)) listener.start() # prep own log handler qh = logging.handlers.QueueHandler(queue) root = logging.getLogger('main') root.setLevel(settings['Parameters']['loglevel'][0]) root.addHandler(qh) # hail the world in log talk(appname + " " + appversion + " starting with loglevel: " + str(settings['Parameters']['loglevel'][0]), logging.NOTE, root) # evaluate run_ssl if override is set settings['Parameters']['run_ssl'][0] = (int(matelib.dgvars.md['use_ssl']) > 0) if 0 != settings['Parameters']['ssl_override'][0]: if 1 == int(settings['Parameters']['ssl_override'][0]): talk("Enforcing HTTPS (SSL) (Dwarfguard config override)...", logging.NOTE, root) settings['Parameters']['run_ssl'][0] = True else: talk("Enforcing HTTP (non SSL) (Dwarfguard config override)...", logging.NOTE, root) settings['Parameters']['run_ssl'][0] = False if 'run_ssl' in settings['enforced']: settings['Parameters']['run_ssl'][0] = settings['enforced']['run_ssl'] talk("SSL *overriden* by commandline parameter to " + str(settings['enforced']['run_ssl']), logging.NOTE, root) # run single action and exit if requested if single_action > 0: single_act(single_action) sys.exit(0) cycle_time = None # lock or fail lockfile = os.open(lockfile_fn, os.O_CREAT | os.O_WRONLY, 755) try: fcntl.lockf(lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB) except IOError: talk(appname + "is locked by" + lockfile_fn + "- if you are sure it is not, manually remove the file and start the script again.", logging.CRITICAL, root) sys.exit(1) while not stop.is_set(): if cycle_time is not None and cycle_time < settings['Parameters']['interval'][0]: talk("Sleeping for " + str(settings['Parameters']['interval'][0] - cycle_time) + " seconds.", logging.VERBOSE, root) stop.wait(settings['Parameters']['interval'][0] - cycle_time) cycle_start_time = time.time() if stop.is_set(): break (new_devices, old_devices_good, old_devices_bad) = read_devices(mylogger) if stop.is_set(): break if len(new_devices) + len(old_devices_good) + len(old_devices_bad) > 0: talk("PRE - There are " + str(len(new_devices)) + " / " + str(len(old_devices_good)) + " / " + str(len(old_devices_bad)) + " devices to be DB-processed.", logging.VERBOSE, root) process_devices(new_devices, old_devices_good, old_devices_bad, queue) talk("POST - There are " + str(len(new_devices)) + " / " + str(len(old_devices_good)) + " / " + str(len(old_devices_bad)) + " devices to be DB-processed.", logging.VERBOSE, root) if stop.is_set(): break update_db(new_devices, old_devices_good, old_devices_bad) cycle_time = time.time() - cycle_start_time if not settings['Parameters']['daemon'][0]: stop.set() if matelib.mdb: dbc.close() matelib.mdb_disconnect() os.close(lockfile) os.remove(lockfile_fn) queue.put_nowait(None) listener.join()