# # This file is part of pysnmp software. # # Copyright (c) 2005-2020, Ilya Etingof # License: https://www.pysnmp.com/pysnmp/license.html # import warnings from pysnmp import debug, error from pysnmp.carrier.asyncio.dgram import udp, udp6 from pysnmp.carrier.base import AbstractTransport from pysnmp.entity.engine import SnmpEngine from pysnmp.proto import rfc1902, rfc1905 from pysnmp.proto.secmod.eso.priv import aes192, aes256, des3 from pysnmp.proto.secmod.rfc3414.auth import hmacmd5, hmacsha, noauth from pysnmp.proto.secmod.rfc3414.priv import des, nopriv from pysnmp.proto.secmod.rfc3414.service import SnmpUSMSecurityModel from pysnmp.proto.secmod.rfc3826.priv import aes from pysnmp.proto.secmod.rfc7860.auth import hmacsha2 # Old to new attribute mapping deprecated_attributes = { "snmpUDPDomain": "SNMP_UDP_DOMAIN", "snmpUDP6Domain": "SNMP_UDP6_DOMAIN", "snmpLocalDomain": "SNMP_LOCAL_DOMAIN", "usmHMACMD5AuthProtocol": "USM_AUTH_HMAC96_MD5", "usmHMACSHAAuthProtocol": "USM_AUTH_HMAC96_SHA", "usmHMAC128SHA224AuthProtocol": "USM_AUTH_HMAC128_SHA224", "usmHMAC192SHA256AuthProtocol": "USM_AUTH_HMAC192_SHA256", "usmHMAC256SHA384AuthProtocol": "USM_AUTH_HMAC256_SHA384", "usmHMAC384SHA512AuthProtocol": "USM_AUTH_HMAC384_SHA512", "usmNoAuthProtocol": "USM_AUTH_NONE", "usmDESPrivProtocol": "USM_PRIV_CBC56_DES", "usm3DESEDEPrivProtocol": "USM_PRIV_CBC168_3DES", "usmAesCfb128Protocol": "USM_PRIV_CFB128_AES", "usmAesBlumenthalCfb192Protocol": "USM_PRIV_CFB192_AES_BLUMENTHAL", "usmAesBlumenthalCfb256Protocol": "USM_PRIV_CFB256_AES_BLUMENTHAL", "usmAesCfb192Protocol": "USM_PRIV_CFB192_AES", "usmAesCfb256Protocol": "USM_PRIV_CFB256_AES", "usmNoPrivProtocol": "USM_PRIV_NONE", "usmKeyTypePassphrase": "USM_KEY_TYPE_PASSPHRASE", "usmKeyTypeMaster": "USM_KEY_TYPE_MASTER", "usmKeyTypeLocalized": "USM_KEY_TYPE_LOCALIZED", "authServices": "AUTH_SERVICES", "privServices": "PRIV_SERVICES", "addV1System": "add_v1_system", "delV1System": "delete_v1_system", "addV3User": "add_v3_user", "delV3User": "delete_v3_user", "addTransport": "add_transport", "delTransport": "delete_transport", "addTargetParams": "add_target_parameters", "delTargetParams": "delete_target_parameters", "addTargetAddr": "add_target_address", "delTargetAddr": "delete_target_address", "addVacmUser": "add_vacm_user", "delVacmUser": "delete_vacm_user", } def __getattr__(attr: str): if new_attr := deprecated_attributes.get(attr): warnings.warn( f"{attr} is deprecated. Please use {new_attr} instead.", DeprecationWarning, stacklevel=2, ) return globals()[new_attr] raise AttributeError(f"module '{__name__}' has no attribute '{attr}'") # A shortcut to popular constants # Transports SNMP_UDP_DOMAIN = udp.SNMP_UDP_DOMAIN SNMP_UDP6_DOMAIN = udp6.SNMP_UDP6_DOMAIN # Auth protocol USM_AUTH_HMAC96_MD5 = hmacmd5.HmacMd5.SERVICE_ID USM_AUTH_HMAC96_SHA = hmacsha.HmacSha.SERVICE_ID USM_AUTH_HMAC128_SHA224 = hmacsha2.HmacSha2.SHA224_SERVICE_ID USM_AUTH_HMAC192_SHA256 = hmacsha2.HmacSha2.SHA256_SERVICE_ID USM_AUTH_HMAC256_SHA384 = hmacsha2.HmacSha2.SAH384_SERVICE_ID USM_AUTH_HMAC384_SHA512 = hmacsha2.HmacSha2.SHA512_SERVICE_ID USM_AUTH_NONE = noauth.NoAuth.SERVICE_ID """No authentication service""" # Privacy protocol USM_PRIV_CBC56_DES = des.Des.SERVICE_ID USM_PRIV_CBC168_3DES = des3.Des3.SERVICE_ID USM_PRIV_CFB128_AES = aes.Aes.SERVICE_ID USM_PRIV_CFB192_AES_BLUMENTHAL = ( aes192.AesBlumenthal192.SERVICE_ID ) # semi-standard but not widely used USM_PRIV_CFB256_AES_BLUMENTHAL = ( aes256.AesBlumenthal256.SERVICE_ID ) # semi-standard but not widely used USM_PRIV_CFB192_AES = aes192.Aes192.SERVICE_ID # non-standard but used by many vendors USM_PRIV_CFB256_AES = aes256.Aes256.SERVICE_ID # non-standard but used by many vendors USM_PRIV_NONE = nopriv.NoPriv.SERVICE_ID # USM key types (PYSNMP-USM-MIB::pysnmpUsmKeyType) USM_KEY_TYPE_PASSPHRASE = 0 USM_KEY_TYPE_MASTER = 1 USM_KEY_TYPE_LOCALIZED = 2 # Auth services AUTH_SERVICES = SnmpUSMSecurityModel.AUTH_SERVICES # Privacy services PRIV_SERVICES = SnmpUSMSecurityModel.PRIV_SERVICES def __cook_v1_system_info(snmpEngine: SnmpEngine, communityIndex): mibBuilder = snmpEngine.get_mib_builder() (snmpEngineID,) = mibBuilder.import_symbols("__SNMP-FRAMEWORK-MIB", "snmpEngineID") # type: ignore (snmpCommunityEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-COMMUNITY-MIB", "snmpCommunityEntry" ) tblIdx = snmpCommunityEntry.getInstIdFromIndices(communityIndex) return snmpCommunityEntry, tblIdx, snmpEngineID def add_v1_system( snmpEngine: SnmpEngine, communityIndex: str, communityName: str, contextEngineId=None, contextName=None, transportTag=None, securityName=None, ): """Register SNMPv1 and v2c community.""" (snmpCommunityEntry, tblIdx, snmpEngineID) = __cook_v1_system_info( snmpEngine, communityIndex ) if contextEngineId is None: contextEngineId = snmpEngineID.syntax else: contextEngineId = snmpEngineID.syntax.clone(contextEngineId) if contextName is None: contextName = b"" securityName = securityName is not None and securityName or communityIndex snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpCommunityEntry.name + (8,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpCommunityEntry.name + (1,) + tblIdx, communityIndex), (snmpCommunityEntry.name + (2,) + tblIdx, communityName), ( snmpCommunityEntry.name + (3,) + tblIdx, securityName is not None and securityName or communityIndex, ), (snmpCommunityEntry.name + (4,) + tblIdx, contextEngineId), (snmpCommunityEntry.name + (5,) + tblIdx, contextName), (snmpCommunityEntry.name + (6,) + tblIdx, transportTag), (snmpCommunityEntry.name + (7,) + tblIdx, "nonVolatile"), (snmpCommunityEntry.name + (8,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) debug.logger & debug.FLAG_SM and debug.logger( "addV1System: added new table entry " 'communityIndex "%s" communityName "%s" securityName "%s" ' 'contextEngineId "%s" contextName "%s" transportTag ' '"%s"' % ( communityIndex, communityName, securityName, contextEngineId, contextName, transportTag, ) ) def delete_v1_system(snmpEngine: SnmpEngine, communityIndex): """Unregister SNMPv1 and v2c community.""" (snmpCommunityEntry, tblIdx, snmpEngineID) = __cook_v1_system_info( snmpEngine, communityIndex ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpCommunityEntry.name + (8,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) debug.logger & debug.FLAG_SM and debug.logger( "delV1System: deleted table entry by communityIndex " '"%s"' % (communityIndex,) ) def __cook_v3_user_info(snmpEngine: SnmpEngine, securityName, securityEngineId): mibBuilder = snmpEngine.get_mib_builder() (snmpEngineID,) = mibBuilder.import_symbols("__SNMP-FRAMEWORK-MIB", "snmpEngineID") # type: ignore if securityEngineId is None: securityEngineId = snmpEngineID.syntax else: securityEngineId = snmpEngineID.syntax.clone(securityEngineId) (usmUserEntry,) = mibBuilder.import_symbols("SNMP-USER-BASED-SM-MIB", "usmUserEntry") # type: ignore tblIdx1 = usmUserEntry.getInstIdFromIndices(securityEngineId, securityName) (pysnmpUsmSecretEntry,) = mibBuilder.import_symbols( # type: ignore "PYSNMP-USM-MIB", "pysnmpUsmSecretEntry" ) tblIdx2 = pysnmpUsmSecretEntry.getInstIdFromIndices(securityName) return securityEngineId, usmUserEntry, tblIdx1, pysnmpUsmSecretEntry, tblIdx2 def add_v3_user( snmpEngine: SnmpEngine, userName, authProtocol=USM_AUTH_NONE, authKey=None, privProtocol=USM_PRIV_NONE, privKey=None, securityEngineId=None, securityName=None, authKeyType=USM_KEY_TYPE_PASSPHRASE, privKeyType=USM_KEY_TYPE_PASSPHRASE, ): """Register SNMPv3 user.""" mibBuilder = snmpEngine.get_mib_builder() if securityName is None: securityName = userName ( securityEngineId, usmUserEntry, tblIdx1, pysnmpUsmSecretEntry, tblIdx2, ) = __cook_v3_user_info(snmpEngine, securityName, securityEngineId) # Load augmenting table before creating new row in base one (pysnmpUsmKeyEntry,) = mibBuilder.import_symbols( # type: ignore "PYSNMP-USM-MIB", "pysnmpUsmKeyEntry" ) # Load clone-from (may not be needed) (zeroDotZero,) = mibBuilder.import_symbols("SNMPv2-SMI", "zeroDotZero") # type: ignore snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (usmUserEntry.name + (13,) + tblIdx1, "destroy"), **dict(snmpEngine=snmpEngine) ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (usmUserEntry.name + (2,) + tblIdx1, userName), (usmUserEntry.name + (3,) + tblIdx1, securityName), (usmUserEntry.name + (4,) + tblIdx1, zeroDotZero.name), (usmUserEntry.name + (5,) + tblIdx1, authProtocol), (usmUserEntry.name + (8,) + tblIdx1, privProtocol), (usmUserEntry.name + (13,) + tblIdx1, "createAndGo"), **dict(snmpEngine=snmpEngine), ) if authProtocol not in AUTH_SERVICES: raise error.PySnmpError(f"Unknown auth protocol {authProtocol}") if privProtocol not in PRIV_SERVICES: raise error.PySnmpError(f"Unknown privacy protocol {privProtocol}") (pysnmpUsmKeyType,) = mibBuilder.import_symbols( # type: ignore "__PYSNMP-USM-MIB", "pysnmpUsmKeyType" ) authKeyType = pysnmpUsmKeyType.syntax.clone(authKeyType) # Localize authentication key unless given authKey = authKey and rfc1902.OctetString(authKey) masterAuthKey = localAuthKey = authKey if authKeyType < USM_KEY_TYPE_MASTER: # pass phrase is given masterAuthKey = AUTH_SERVICES[authProtocol].hash_passphrase(authKey or b"") if authKeyType < USM_KEY_TYPE_LOCALIZED: # pass phrase or master key is given localAuthKey = AUTH_SERVICES[authProtocol].localize_key( masterAuthKey, securityEngineId ) # Localize privacy key unless given privKeyType = pysnmpUsmKeyType.syntax.clone(privKeyType) privKey = privKey and rfc1902.OctetString(privKey) masterPrivKey = localPrivKey = privKey if privKeyType < USM_KEY_TYPE_MASTER: # pass phrase is given masterPrivKey = PRIV_SERVICES[privProtocol].hash_passphrase( authProtocol, privKey or b"" ) if privKeyType < USM_KEY_TYPE_LOCALIZED: # pass phrase or master key is given localPrivKey = PRIV_SERVICES[privProtocol].localize_key( authProtocol, masterPrivKey, securityEngineId ) # Commit only the keys we have snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmKeyEntry.name + (1,) + tblIdx1, localAuthKey), (pysnmpUsmKeyEntry.name + (2,) + tblIdx1, localPrivKey), **dict(snmpEngine=snmpEngine), ) if authKeyType < USM_KEY_TYPE_LOCALIZED: snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmKeyEntry.name + (3,) + tblIdx1, masterAuthKey), **dict(snmpEngine=snmpEngine), ) if privKeyType < USM_KEY_TYPE_LOCALIZED: snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmKeyEntry.name + (4,) + tblIdx1, masterPrivKey), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmSecretEntry.name + (4,) + tblIdx2, "destroy"), **dict(snmpEngine=snmpEngine), ) # Commit plain-text pass-phrases if we have them snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmSecretEntry.name + (4,) + tblIdx2, "createAndGo"), **dict(snmpEngine=snmpEngine), ) if authKeyType < USM_KEY_TYPE_MASTER: snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmSecretEntry.name + (1,) + tblIdx2, userName), (pysnmpUsmSecretEntry.name + (2,) + tblIdx2, authKey), **dict(snmpEngine=snmpEngine), ) if privKeyType < USM_KEY_TYPE_MASTER: snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmSecretEntry.name + (1,) + tblIdx2, userName), (pysnmpUsmSecretEntry.name + (3,) + tblIdx2, privKey), **dict(snmpEngine=snmpEngine), ) debug.logger & debug.FLAG_SM and debug.logger( "addV3User: added new table entries " 'userName "%s" securityName "%s" authProtocol %s ' 'privProtocol %s localAuthKey "%s" localPrivKey "%s" ' 'masterAuthKey "%s" masterPrivKey "%s" authKey "%s" ' 'privKey "%s" by index securityName "%s" securityEngineId ' '"%s"' % ( userName, securityName, authProtocol, privProtocol, localAuthKey and localAuthKey.prettyPrint(), localPrivKey and localPrivKey.prettyPrint(), masterAuthKey and masterAuthKey.prettyPrint(), masterPrivKey and masterPrivKey.prettyPrint(), authKey and authKey.prettyPrint(), privKey and privKey.prettyPrint(), securityName, securityEngineId.prettyPrint(), ) ) def delete_v3_user( snmpEngine: SnmpEngine, userName, securityEngineId=None, ): """Unregister SNMPv3 user.""" ( securityEngineId, usmUserEntry, tblIdx1, pysnmpUsmSecretEntry, tblIdx2, ) = __cook_v3_user_info(snmpEngine, userName, securityEngineId) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (usmUserEntry.name + (13,) + tblIdx1, "destroy"), **dict(snmpEngine=snmpEngine) ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (pysnmpUsmSecretEntry.name + (4,) + tblIdx2, "destroy"), **dict(snmpEngine=snmpEngine), ) debug.logger & debug.FLAG_SM and debug.logger( "delV3User: deleted table entries by index " 'userName "%s" securityEngineId ' '"%s"' % (userName, securityEngineId.prettyPrint()) ) # Drop all derived rows varBinds = initialVarBinds = ( (usmUserEntry.name + (1,), None), # usmUserEngineID (usmUserEntry.name + (2,), None), # usmUserName (usmUserEntry.name + (4,), None), # usmUserCloneFrom ) while varBinds: varBinds = ( snmpEngine.message_dispatcher.mib_instrum_controller.read_next_variables( *varBinds, **dict(snmpEngine=snmpEngine) ) ) if varBinds[0][1].isSameTypeWith(rfc1905.endOfMibView): break if varBinds[0][0][: len(initialVarBinds[0][0])] != initialVarBinds[0][0]: break elif varBinds[2][1] == tblIdx1: # cloned from this entry delete_v3_user(snmpEngine, varBinds[1][1], varBinds[0][1]) varBinds = initialVarBinds def __cook_target_parameters_info(snmpEngine: SnmpEngine, name): mibBuilder = snmpEngine.get_mib_builder() (snmpTargetParamsEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-TARGET-MIB", "snmpTargetParamsEntry" ) tblIdx = snmpTargetParamsEntry.getInstIdFromIndices(name) return snmpTargetParamsEntry, tblIdx # mpModel: 0 == SNMPv1, 1 == SNMPv2c, 3 == SNMPv3 def add_target_parameters( snmpEngine: SnmpEngine, name, securityName, securityLevel, mpModel=3 ): """Register target parameters.""" if mpModel == 0: securityModel = 1 elif mpModel in (1, 2): securityModel = 2 elif mpModel == 3: securityModel = 3 else: raise error.PySnmpError("Unknown MP model %s" % mpModel) snmpTargetParamsEntry, tblIdx = __cook_target_parameters_info(snmpEngine, name) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpTargetParamsEntry.name + (7,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpTargetParamsEntry.name + (1,) + tblIdx, name), (snmpTargetParamsEntry.name + (2,) + tblIdx, mpModel), (snmpTargetParamsEntry.name + (3,) + tblIdx, securityModel), (snmpTargetParamsEntry.name + (4,) + tblIdx, securityName), (snmpTargetParamsEntry.name + (5,) + tblIdx, securityLevel), (snmpTargetParamsEntry.name + (7,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_target_parameters(snmpEngine: SnmpEngine, name: str): """Delete target parameters.""" snmpTargetParamsEntry, tblIdx = __cook_target_parameters_info(snmpEngine, name) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpTargetParamsEntry.name + (7,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) def __cook_target_address_info(snmpEngine: SnmpEngine, addrName: str): mibBuilder = snmpEngine.get_mib_builder() (snmpTargetAddrEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-TARGET-MIB", "snmpTargetAddrEntry" ) (snmpSourceAddrEntry,) = mibBuilder.import_symbols( # type: ignore "PYSNMP-SOURCE-MIB", "snmpSourceAddrEntry" ) tblIdx = snmpTargetAddrEntry.getInstIdFromIndices(addrName) return snmpTargetAddrEntry, snmpSourceAddrEntry, tblIdx def add_target_address( snmpEngine: SnmpEngine, addrName: str, transportDomain: "tuple[int, ...]", transportAddress: "tuple[str, int]", params: str, timeout: "float | None" = None, retryCount: "int | None" = None, tagList=b"", sourceAddress=None, ): """Register target address.""" mibBuilder = snmpEngine.get_mib_builder() (snmpTargetAddrEntry, snmpSourceAddrEntry, tblIdx) = __cook_target_address_info( snmpEngine, addrName ) if transportDomain[: len(SNMP_UDP_DOMAIN)] == SNMP_UDP_DOMAIN: (SnmpUDPAddress,) = mibBuilder.import_symbols("SNMPv2-TM", "SnmpUDPAddress") # type: ignore transportAddress = SnmpUDPAddress(transportAddress) if sourceAddress is None: sourceAddress = ("0.0.0.0", 0) sourceAddress = SnmpUDPAddress(sourceAddress) elif transportDomain[: len(SNMP_UDP6_DOMAIN)] == SNMP_UDP6_DOMAIN: (TransportAddressIPv6,) = mibBuilder.import_symbols( # type: ignore "TRANSPORT-ADDRESS-MIB", "TransportAddressIPv6" ) transportAddress = TransportAddressIPv6(transportAddress) if sourceAddress is None: sourceAddress = ("::", 0) sourceAddress = TransportAddressIPv6(sourceAddress) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpTargetAddrEntry.name + (9,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpTargetAddrEntry.name + (1,) + tblIdx, addrName), (snmpTargetAddrEntry.name + (2,) + tblIdx, transportDomain), (snmpTargetAddrEntry.name + (3,) + tblIdx, transportAddress), (snmpTargetAddrEntry.name + (4,) + tblIdx, timeout), (snmpTargetAddrEntry.name + (5,) + tblIdx, retryCount), (snmpTargetAddrEntry.name + (6,) + tblIdx, tagList), (snmpTargetAddrEntry.name + (7,) + tblIdx, params), (snmpSourceAddrEntry.name + (1,) + tblIdx, sourceAddress), (snmpTargetAddrEntry.name + (9,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_target_address(snmpEngine: SnmpEngine, addrName: str): """Delete target address.""" (snmpTargetAddrEntry, snmpSourceAddrEntry, tblIdx) = __cook_target_address_info( snmpEngine, addrName ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpTargetAddrEntry.name + (9,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) def add_transport( snmpEngine: SnmpEngine, transportDomain: "tuple[int, ...]", transport: AbstractTransport, ): """Register transport with dispatcher.""" if snmpEngine.transport_dispatcher: if not transport.is_compatible_with_dispatcher(snmpEngine.transport_dispatcher): raise error.PySnmpError( f"Transport {transport!r} is not compatible with dispatcher {snmpEngine.transport_dispatcher!r}" ) else: protoTransportDispatcher = transport.PROTO_TRANSPORT_DISPATCHER if protoTransportDispatcher is not None: snmpEngine.register_transport_dispatcher(protoTransportDispatcher()) # here we note that we have created transportDispatcher automatically snmpEngine.set_user_context(automaticTransportDispatcher=0) if snmpEngine.transport_dispatcher: snmpEngine.transport_dispatcher.register_transport(transportDomain, transport) automaticTransportDispatcher = snmpEngine.get_user_context( "automaticTransportDispatcher" ) if automaticTransportDispatcher is not None: snmpEngine.set_user_context( automaticTransportDispatcher=automaticTransportDispatcher + 1 ) def get_transport(snmpEngine: SnmpEngine, transportDomain: "tuple[int, ...]"): """Return transport from dispatcher.""" if not snmpEngine.transport_dispatcher: return try: return snmpEngine.transport_dispatcher.get_transport(transportDomain) except error.PySnmpError: return def delete_transport(snmpEngine: SnmpEngine, transportDomain: "tuple[int, ...]"): """Remove transport from dispatcher.""" if not snmpEngine.transport_dispatcher: return transport = get_transport(snmpEngine, transportDomain) snmpEngine.transport_dispatcher.unregister_transport(transportDomain) # automatically shutdown automatically created transportDispatcher automaticTransportDispatcher = snmpEngine.get_user_context( "automaticTransportDispatcher" ) if automaticTransportDispatcher is not None: automaticTransportDispatcher -= 1 snmpEngine.set_user_context( automaticTransportDispatcher=automaticTransportDispatcher ) if not automaticTransportDispatcher: snmpEngine.close_dispatcher() snmpEngine.delete_user_context(automaticTransportDispatcher) return transport addSocketTransport = add_transport # noqa: N816 delSocketTransport = delete_transport # noqa: N816 # VACM shortcuts def __cook_vacm_context_info(snmpEngine: SnmpEngine, contextName): mibBuilder = snmpEngine.get_mib_builder() (vacmContextEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-VIEW-BASED-ACM-MIB", "vacmContextEntry" ) tblIdx = vacmContextEntry.getInstIdFromIndices(contextName) return vacmContextEntry, tblIdx def add_context(snmpEngine: SnmpEngine, contextName): """Setup VACM context.""" vacmContextEntry, tblIdx = __cook_vacm_context_info(snmpEngine, contextName) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmContextEntry.name + (2,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmContextEntry.name + (1,) + tblIdx, contextName), (vacmContextEntry.name + (2,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_context(snmpEngine: SnmpEngine, contextName): """Delete VACM context.""" vacmContextEntry, tblIdx = __cook_vacm_context_info(snmpEngine, contextName) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmContextEntry.name + (2,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) def __cook_vacm_group_info(snmpEngine: SnmpEngine, securityModel, securityName): mibBuilder = snmpEngine.get_mib_builder() (vacmSecurityToGroupEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-VIEW-BASED-ACM-MIB", "vacmSecurityToGroupEntry" ) tblIdx = vacmSecurityToGroupEntry.getInstIdFromIndices(securityModel, securityName) return vacmSecurityToGroupEntry, tblIdx def add_vacm_group(snmpEngine: SnmpEngine, groupName, securityModel, securityName): """Setup VACM group.""" (vacmSecurityToGroupEntry, tblIdx) = __cook_vacm_group_info( snmpEngine, securityModel, securityName ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmSecurityToGroupEntry.name + (5,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmSecurityToGroupEntry.name + (1,) + tblIdx, securityModel), (vacmSecurityToGroupEntry.name + (2,) + tblIdx, securityName), (vacmSecurityToGroupEntry.name + (3,) + tblIdx, groupName), (vacmSecurityToGroupEntry.name + (5,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_vacm_group(snmpEngine: SnmpEngine, securityModel, securityName): """Delete VACM group.""" vacmSecurityToGroupEntry, tblIdx = __cook_vacm_group_info( snmpEngine, securityModel, securityName ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmSecurityToGroupEntry.name + (5,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) def __cook_vacm_access_info( snmpEngine: SnmpEngine, groupName, contextName, securityModel, securityLevel ): mibBuilder = snmpEngine.get_mib_builder() (vacmAccessEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-VIEW-BASED-ACM-MIB", "vacmAccessEntry" ) tblIdx = vacmAccessEntry.getInstIdFromIndices( groupName, contextName, securityModel, securityLevel ) return vacmAccessEntry, tblIdx def add_vacm_access( snmpEngine: SnmpEngine, groupName, contextPrefix, securityModel, securityLevel, contextMatch, readView, writeView, notifyView, ): """Setup VACM access.""" vacmAccessEntry, tblIdx = __cook_vacm_access_info( snmpEngine, groupName, contextPrefix, securityModel, securityLevel ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmAccessEntry.name + (9,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine) ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmAccessEntry.name + (1,) + tblIdx, contextPrefix), (vacmAccessEntry.name + (2,) + tblIdx, securityModel), (vacmAccessEntry.name + (3,) + tblIdx, securityLevel), (vacmAccessEntry.name + (4,) + tblIdx, contextMatch), (vacmAccessEntry.name + (5,) + tblIdx, readView), (vacmAccessEntry.name + (6,) + tblIdx, writeView), (vacmAccessEntry.name + (7,) + tblIdx, notifyView), (vacmAccessEntry.name + (9,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_vacm_access( snmpEngine: SnmpEngine, groupName, contextPrefix, securityModel, securityLevel ): """Delete VACM access.""" vacmAccessEntry, tblIdx = __cook_vacm_access_info( snmpEngine, groupName, contextPrefix, securityModel, securityLevel ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmAccessEntry.name + (9,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine) ) def __cook_vacm_view_info(snmpEngine: SnmpEngine, viewName, subTree): mibBuilder = snmpEngine.get_mib_builder() (vacmViewTreeFamilyEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-VIEW-BASED-ACM-MIB", "vacmViewTreeFamilyEntry" ) tblIdx = vacmViewTreeFamilyEntry.getInstIdFromIndices(viewName, subTree) return vacmViewTreeFamilyEntry, tblIdx def add_vacm_view(snmpEngine: SnmpEngine, viewName, viewType, subTree, subTreeMask): """Setup VACM view.""" vacmViewTreeFamilyEntry, tblIdx = __cook_vacm_view_info( snmpEngine, viewName, subTree ) # Allow bitmask specification in form of an OID if rfc1902.OctetString(".").asOctets() in rfc1902.OctetString(subTreeMask): subTreeMask = rfc1902.ObjectIdentifier(subTreeMask) if isinstance(subTreeMask, rfc1902.ObjectIdentifier): subTreeMask = tuple(subTreeMask) if len(subTreeMask) < len(subTree): subTreeMask += (1,) * (len(subTree) - len(subTreeMask)) subTreeMask = rfc1902.OctetString.fromBinaryString( "".join(str(x) for x in subTreeMask) ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmViewTreeFamilyEntry.name + (6,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmViewTreeFamilyEntry.name + (1,) + tblIdx, viewName), (vacmViewTreeFamilyEntry.name + (2,) + tblIdx, subTree), (vacmViewTreeFamilyEntry.name + (3,) + tblIdx, subTreeMask), (vacmViewTreeFamilyEntry.name + (4,) + tblIdx, viewType), (vacmViewTreeFamilyEntry.name + (6,) + tblIdx, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_vacm_view(snmpEngine: SnmpEngine, viewName, subTree): """Delete VACM view.""" vacmViewTreeFamilyEntry, tblIdx = __cook_vacm_view_info( snmpEngine, viewName, subTree ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (vacmViewTreeFamilyEntry.name + (6,) + tblIdx, "destroy"), **dict(snmpEngine=snmpEngine), ) # VACM simplicity wrappers def __cook_vacm_user_info( snmpEngine: SnmpEngine, securityModel, securityName, securityLevel ): mibBuilder = snmpEngine.get_mib_builder() groupName = "v-%s-%d" % (hash(securityName), securityModel) (SnmpSecurityLevel,) = mibBuilder.import_symbols( # type: ignore "SNMP-FRAMEWORK-MIB", "SnmpSecurityLevel" ) securityLevel = SnmpSecurityLevel(securityLevel) return (groupName, securityLevel, "r" + groupName, "w" + groupName, "n" + groupName) def add_vacm_user( snmpEngine: SnmpEngine, securityModel: int, securityName: str, securityLevel: str, readSubTree=(), writeSubTree=(), notifySubTree=(), contextName=b"", ): """Setup VACM user.""" (groupName, securityLevel, readView, writeView, notifyView) = __cook_vacm_user_info( snmpEngine, securityModel, securityName, securityLevel ) add_context(snmpEngine, contextName) add_vacm_group(snmpEngine, groupName, securityModel, securityName) add_vacm_access( snmpEngine, groupName, contextName, securityModel, securityLevel, "exact", readView, writeView, notifyView, ) if readSubTree: add_vacm_view(snmpEngine, readView, "included", readSubTree, b"") if writeSubTree: add_vacm_view(snmpEngine, writeView, "included", writeSubTree, b"") if notifySubTree: add_vacm_view(snmpEngine, notifyView, "included", notifySubTree, b"") def delete_vacm_user( snmpEngine: SnmpEngine, securityModel, securityName, securityLevel, readSubTree=(), writeSubTree=(), notifySubTree=(), contextName=b"", ): """Delete VACM user.""" (groupName, securityLevel, readView, writeView, notifyView) = __cook_vacm_user_info( snmpEngine, securityModel, securityName, securityLevel ) delete_context(snmpEngine, contextName) delete_vacm_group(snmpEngine, securityModel, securityName) delete_vacm_access(snmpEngine, groupName, contextName, securityModel, securityLevel) if readSubTree: delete_vacm_view(snmpEngine, readView, readSubTree) if writeSubTree: delete_vacm_view(snmpEngine, writeView, writeSubTree) if notifySubTree: delete_vacm_view(snmpEngine, notifyView, notifySubTree) # Notification target setup def __cook_notification_target_info( snmpEngine: SnmpEngine, notificationName, paramsName, filterSubtree=None ): mibBuilder = snmpEngine.get_mib_builder() (snmpNotifyEntry,) = mibBuilder.import_symbols( # type: ignore "SNMP-NOTIFICATION-MIB", "snmpNotifyEntry" ) tblIdx1 = snmpNotifyEntry.getInstIdFromIndices(notificationName) (snmpNotifyFilterProfileEntry,) = mibBuilder.import_symbols( "SNMP-NOTIFICATION-MIB", "snmpNotifyFilterProfileEntry" ) tblIdx2 = snmpNotifyFilterProfileEntry.getInstIdFromIndices(paramsName) profileName = "%s-filter" % hash(notificationName) if filterSubtree: (snmpNotifyFilterEntry,) = mibBuilder.import_symbols( "SNMP-NOTIFICATION-MIB", "snmpNotifyFilterEntry" ) tblIdx3 = snmpNotifyFilterEntry.getInstIdFromIndices(profileName, filterSubtree) else: snmpNotifyFilterEntry = tblIdx3 = None return ( snmpNotifyEntry, tblIdx1, snmpNotifyFilterProfileEntry, tblIdx2, profileName, snmpNotifyFilterEntry, tblIdx3, ) def add_notification_target( snmpEngine: SnmpEngine, notificationName, paramsName, transportTag, notifyType=None, filterSubtree=None, filterMask=None, filterType=None, ): """Setup notification target.""" ( snmpNotifyEntry, tblIdx1, snmpNotifyFilterProfileEntry, tblIdx2, profileName, snmpNotifyFilterEntry, tblIdx3, ) = __cook_notification_target_info( snmpEngine, notificationName, paramsName, filterSubtree ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyEntry.name + (5,) + tblIdx1, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyEntry.name + (2,) + tblIdx1, transportTag), (snmpNotifyEntry.name + (3,) + tblIdx1, notifyType), (snmpNotifyEntry.name + (5,) + tblIdx1, "createAndGo"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyFilterProfileEntry.name + (1,) + tblIdx2, profileName), (snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, "createAndGo"), **dict(snmpEngine=snmpEngine), ) if not snmpNotifyFilterEntry: return snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyFilterEntry.name + (5,) + tblIdx3, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyFilterEntry.name + (1,) + tblIdx3, filterSubtree), (snmpNotifyFilterEntry.name + (2,) + tblIdx3, filterMask), (snmpNotifyFilterEntry.name + (3,) + tblIdx3, filterType), (snmpNotifyFilterEntry.name + (5,) + tblIdx3, "createAndGo"), **dict(snmpEngine=snmpEngine), ) def delete_notification_target( snmpEngine: SnmpEngine, notificationName, paramsName, filterSubtree=None ): """Remove notification target.""" ( snmpNotifyEntry, tblIdx1, snmpNotifyFilterProfileEntry, tblIdx2, profileName, snmpNotifyFilterEntry, tblIdx3, ) = __cook_notification_target_info( snmpEngine, notificationName, paramsName, filterSubtree ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyEntry.name + (5,) + tblIdx1, "destroy"), **dict(snmpEngine=snmpEngine), ) snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyFilterProfileEntry.name + (3,) + tblIdx2, "destroy"), **dict(snmpEngine=snmpEngine), ) if not snmpNotifyFilterEntry: return snmpEngine.message_dispatcher.mib_instrum_controller.write_variables( (snmpNotifyFilterEntry.name + (5,) + tblIdx3, "destroy"), **dict(snmpEngine=snmpEngine), ) # rfc3415: A.1 def set_initial_vacm_parameters(snmpEngine: SnmpEngine): """Set initial VACM parameters as per SNMP-FRAMEWORK-MIB.""" # rfc3415: A.1.1 --> initial-semi-security-configuration # rfc3415: A.1.2 add_context(snmpEngine, "") # rfc3415: A.1.3 add_vacm_group(snmpEngine, "initial", 3, "initial") # rfc3415: A.1.4 add_vacm_access( snmpEngine, "initial", "", 3, "noAuthNoPriv", "exact", "restricted", None, "restricted", ) add_vacm_access( snmpEngine, "initial", "", 3, "authNoPriv", "exact", "internet", "internet", "internet", ) add_vacm_access( snmpEngine, "initial", "", 3, "authPriv", "exact", "internet", "internet", "internet", ) # rfc3415: A.1.5 (semi-secure) add_vacm_view(snmpEngine, "internet", "included", (1, 3, 6, 1), "") add_vacm_view(snmpEngine, "restricted", "included", (1, 3, 6, 1, 2, 1, 1), "") add_vacm_view(snmpEngine, "restricted", "included", (1, 3, 6, 1, 2, 1, 11), "") add_vacm_view( snmpEngine, "restricted", "included", (1, 3, 6, 1, 6, 3, 10, 2, 1), "" ) add_vacm_view( snmpEngine, "restricted", "included", (1, 3, 6, 1, 6, 3, 11, 2, 1), "" ) add_vacm_view( snmpEngine, "restricted", "included", (1, 3, 6, 1, 6, 3, 15, 1, 1), "" )