Python3.9 transportDispatcher cannot receive the alarm information sent from snmptrap V3 and does not report an error, but it can receive the alarm information from snmpinform

[nishaoshan]

Python3.9 pysnmp4.4.12
transportDispatcher cannot receive the alarm information sent from snmptrap and does not report an error, but it can receive the alarm information from snmpinform

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import time
import sys
import logging

from pysnmp.debug import Debug, setLogger
from pysnmp.entity import engine, config
from pysnmp.carrier.asyncore.dgram import udp, udp6
from pysnmp.entity.rfc3413 import ntfrcv

from .trap_asset import get_device_id, add_asset
from .trap_config import load_config
from .trap_save import trap_insert
from common.call_ext import is_device_online,is_audit_product
from common.sg_alarm import ALARM_T_CPU, ALARM_T_MEM, ALARM_T_DISK, ALARM_T_CPU_TEMP, ALARM_T_FAN_SPEED, \
     ALARM_T_ABNORMAL_EV,ALARM_T_START_EV, ALARM_T_NAT_EV, ALARM_T_INF_EV, ALARM_T_LOG_ROLLBACK, \
     ALARM_T_CONNECTION_STATE, ALARM_ASSET_ALARM, ALARM_T_FALL_LOST, ALARM_T_CONCURRENCY

ALARM_CLEAR = 32


class TrapConfig(object):
    status = 1
    listen_port = 162
    support_v2 = 1
    community = 'public'
    support_v3 = 1
    user_name = ''
    auth_proto = 'MD5'
    auth_key = ''
    priv_proto = 'DES'
    priv_key = ''

g_trapConf = TrapConfig()
debug_log = logging.getLogger(__name__)


# Callback function for receiving notifications
def cbFun(snmpEngine, stateReference, contextEngineId, contextName,
          varBinds, cbCtx):
    log = cbCtx
    sqlDict = { "log_time":0,
                "log_module":'trap',
                "log_level":0,
                "log_type":0,
                "log_from":None,
                "log_flag":0,
                "cn_msg":None,
                "en_msg":None
                }

    # Get an execution context...
    execContext = snmpEngine.observer.getExecutionContext(
        'rfc3412.receiveMessage:request'
    )
    sqlDict["log_time"] = int(time.time())
    sqlDict["log_from"] = execContext['transportAddress'][0]
    ALARM_R = '解除'

    asset = None
    ###
    for oid, val in varBinds:
        val = str(val).encode('iso-8859-1').decode('utf-8')
        log.debug('oid:%r, val:%s', oid, val)
        if oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.4":
            if not is_audit_product() and not is_device_online(str(val)):
                log.warn('The device [%s][%s] is not online.', sqlDict["log_from"], str(val))
                return
            sqlDict["log_from"] = val
        # cpuPercentUsage
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.10.1.2":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_CPU
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_CPU + ALARM_CLEAR
        # memoryPercentUsage
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.10.3.2":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_MEM
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_MEM + ALARM_CLEAR
        # CF-diskPercentUsage/harddiskPercentUsage
        elif oid.prettyPrint() in ("1.3.6.1.4.1.32328.6.1.10.2.2", "1.3.6.1.4.1.32328.6.1.10.2.4"):
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_DISK
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_DISK + ALARM_CLEAR
        # cpuTemp
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.10.1.3.1.3":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_CPU_TEMP
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_CPU_TEMP + ALARM_CLEAR
        # FanSpeed
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.10.5.1.1.3":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_FAN_SPEED
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_FAN_SPEED + ALARM_CLEAR
        # attack Event
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.2":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_ABNORMAL_EV
        # startEvent
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.5":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_START_EV
        # natThresholdEvent
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.6":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_NAT_EV
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_NAT_EV + ALARM_CLEAR
        # interfaces
        elif oid.prettyPrint() in ("1.3.6.1.4.1.32328.6.2.3.1.1.7", "1.3.6.1.4.1.32328.6.2.3.1.1.8"):
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_INF_EV
            if ALARM_R in val:
                sqlDict['log_type'] = ALARM_T_INF_EV + ALARM_CLEAR

        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.7":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_CONNECTION_STATE

        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.8":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_LOG_ROLLBACK

        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.9":
            data = json.loads(str(val))
            if data and data["property_terminal"] and len(data["property_terminal"]) > 0:
                sqlDict["log_type"] = ALARM_ASSET_ALARM
                asset = data["property_terminal"][0]
                log.info("asset %s", asset)
                if asset["action"] == 'pending':
                    sqlDict["en_msg"] = "There is a new pending terminal to access network, please audit it at once."
                    sqlDict["cn_msg"] = "您有1个新待审批设备接入网络，请审核是否准入网络。"
                elif asset["action"] == 'abnormal':
                    sqlDict["en_msg"] = "There is a new abnormal terminal to access network, please handle it at once."
                    sqlDict["cn_msg"] = "您有1个新异常仿冒设备接入网络，请审核是否准入网络。"
                else:
                    log.error("unknown type asset: %s", asset)
        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.10":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_CONCURRENCY

        elif oid.prettyPrint() == "1.3.6.1.4.1.32328.6.1.11.11":
            sqlDict["en_msg"] = val
            sqlDict["cn_msg"] = val
            sqlDict["log_type"] = ALARM_T_FALL_LOST

    if sqlDict["log_type"] == ALARM_ASSET_ALARM:
        if sqlDict["log_from"] and asset:
            ne_id = get_device_id(str(sqlDict["log_from"]))
            add_asset(ne_id, asset)
        else:
            log.error("trap does not contain device sn: %s", varBinds)
            return

    if sqlDict["cn_msg"] is None or sqlDict["en_msg"] is None:
        return
    log.debug('ALARM:%s', sqlDict)
    trap_insert(sqlDict, logger=log)


# AuthProtocolConvert
def AuthProtocolConvert(proto):
    auth_proto = config.usmNoAuthProtocol

    if proto == "MD5":
        auth_proto = config.usmHMACMD5AuthProtocol
    elif proto == "SHA1":
        auth_proto = config.usmHMACSHAAuthProtocol

    return auth_proto


# PrivProtocolConvert
def PrivProtocolConvert(proto):
    priv_proto = config.usmNoPrivProtocol

    if proto == "DES":
        priv_proto = config.usmDESPrivProtocol
    elif proto == "AES-128":
        priv_proto = config.usmAesCfb128Protocol

    return priv_proto


# ################ main function ################
def main_loop(logger=None):
    if logger:
        global debug_log
        debug_log = logger
    # load trap config
    load_config(g_trapConf, debug_log)

    # local variables
    auth_proto = AuthProtocolConvert(g_trapConf.auth_proto)
    priv_proto = PrivProtocolConvert(g_trapConf.priv_proto)

    # global enable
    if not g_trapConf.status:
        debug_log.info('trap status disable, exit')
        return

    # debug
    # setLogger(Debug('msgproc', 'secmod', 'app', 'dsp', 'io'))

    # Create SNMP engine with auto-generated engineID and pre-bound
    # to socket transport dispatcher
    snmpEngine = engine.SnmpEngine()

    # Transport setup
    # UDP over IPv4
    domain_name = udp6.domainName
    transport = udp6.Udp6Transport().openServerMode(('::', g_trapConf.listen_port))
    config.addTransport(snmpEngine, domain_name, transport)

    # SNMPv2 setup
    if g_trapConf.support_v2:
        config.addV1System(snmpEngine, 'my-area', g_trapConf.community)

    # SNMPv3/USM setup
    if g_trapConf.support_v3:
        config.addV3User(
            snmpEngine, g_trapConf.user_name,
            auth_proto, g_trapConf.auth_key,
            priv_proto, g_trapConf.priv_key
        )

    # Register SNMP Application at the SNMP engine
    ntfrcv.NotificationReceiver(snmpEngine, cbFun, cbCtx=debug_log)

    snmpEngine.transportDispatcher.jobStarted(1)  # this job would never finish

    # Run I/O dispatcher which would receive queries and send confirmations
    try:
        snmpEngine.transportDispatcher.runDispatcher()
    except Exception as e:
        snmpEngine.transportDispatcher.closeDispatcher()
        debug_log.error(e)

[nishaoshan]

How can I receive the alarm information sent by snmptrap v3, not only snmpinform, because the sender uses snmptrap instead of snmpinform。But there is no such problem in python 2。In addition, there is no problem with v1 and v2 in Python 3.9。
If python3 wants to use snmptrap to receive messages, the trap receiver may need to add a securityEngineId. If you do not add a securityEngineId, you cannot obtain snmptrap. Only snmpinform can be obtained

[lextm]

SNMP v3 TRAPs work differently from INFORMs.

About that, Ilya explained enough in #324

[nishaoshan]

您好！您的邮件我已查收！谢谢！