pysnmp
pysnmp copied to clipboard
lextudio
walkCmd not returning from loaded custom MIB
Expected behavior
Code Block to Walk SNMP MIB
Consider the following block of code:
import asyncio
from pysnmp.hlapi.asyncio import *
from pysnmp.smi import builder, view, compiler, rfc1902
# Initialize the MIB builder
mib_builder = builder.MibBuilder()
# Add MIB sources, including the directory containing the converted MIB file
custom_mib_path = '/home/conrad/.pysnmp/mibs'
mib_sources = mib_builder.getMibSources() + (builder.DirMibSource(custom_mib_path),)
mib_builder.setMibSources(*mib_sources)
# Load custom MIB files
compiler.addMibCompiler(mib_builder, sources=['file://' + custom_mib_path])
mib_builder.loadModules('ELTEK-BC2000-DC-POWER-MIB-V2-12')
# Create MIB viewer
mib_view_controller = view.MibViewController(mib_builder)
async def snmp_walk_custom_mib(oid, ip, community='public'):
result = []
try:
identity = ObjectIdentity('ELTEK-BC2000-DC-POWER-MIB-V2-12', oid).loadMibs('ELTEK-BC2000-DC-POWER-MIB-V2-12')
identity = identity.resolveWithMib(mib_view_controller)
except Exception as e:
print(f"Error resolving OID {oid}: {e}")
return []
iterator = walkCmd(
SnmpEngine(),
CommunityData(community),
UdpTransportTarget((ip, 161)),
ContextData(),
ObjectType(identity),
lexicographicMode=False
)
async for error_indication, error_status, error_index, var_binds in iterator:
if error_indication:
print(f"Error indication: {error_indication}")
break
elif error_status:
print(f"Error status: {error_status.prettyPrint()} at {error_index}")
break
else:
for var_bind in var_binds:
oid, value = var_bind
try:
oid_label = oid.resolveWithMib(mib_view_controller)
oid_number = oid.getOid()._value
oid_label = mib_view_controller.getNodeName((oid_number))
except Exception as e:
oid_label = 'Unknown'
print(f"Error getting label for OID {oid}: {e}")
result.append(f'{oid.prettyPrint()} = {value.prettyPrint()} ({oid_label})')
return result
async def main():
value1 = await snmp_walk_custom_mib('vpwrPanelModuleSerNum', '10.115.0.10')
print("\n".join(value1))
# Run the main function
asyncio.run(main())
Expected Output
The code should return the correct MIB from ELTEK-BC2000-DC-POWER-MIB-V2-12:
ELTEK-BC2000-DC-POWER-MIB-V2-12::vpwrPanelModuleSerNum.0 = 183886010176
Actual Output
However, it returns:
SNMPv2-SMI::enterprises.13858.2.4.2.1.3.5.1 = 183886010176
Actual behavior
Issue
The returned object from the walkCmd doesn't use the right MIB but defaults to the SNMPv2-SMI MIB file:
SNMPv2-SMI::enterprises.13858.2.4.2.1.3.5.1 = 183886010176
I believe this is a similar issue than this bug from "etingof" repo: https://github.com/etingof/pysnmp/issues/273
Detailed steps
Issue
From the debug log, we can see that my ELTEK-BC2000-DC-POWER-MIB-V2-12 mib file is loaded properly. Also the mib_view_controller object contains my mib file as well.
Note
This issue only happens for the walkCmd, it works as intended with the getCmd.
The workaround that I've found is to reinstantiate the object based on the returned OID number with the correct MIB:
for device_result in completed_tasks:
task_results = []
for oid_result in device_result:
if oid_result is not None:
for varBind in oid_result:
if varBind[0]._ObjectIdentity__modName == "ELTEK-BC2000-DC-POWER-MIB-V2-12":
full_oid = varBind[0].prettyPrint()
key = full_oid.split('::')[-1]
value = varBind[1].prettyPrint()
task_results.append({key: value})
all_keys.add(key)
else:
full_oid = varBind[0].getOid()._value
resolved_varBind = mibViewController.getNodeName((full_oid))
last_value = resolved_varBind[1][-1]
resolved_varBind_2_str = '.' + '.'.join(map(str, resolved_varBind[2]))
key = last_value + resolved_varBind_2_str
value = varBind[1]._value
if isinstance(value, bytes):
decoded_value = value.decode('utf-8')
else:
decoded_value = value
task_results.append({key: decoded_value})
all_keys.add(key)
Partial log
2024-07-25 16:44:27,548 DEBUG exportSymbols: symbol ELTEK-BC2000-DC-POWER-MIB-V2-12::vpwrDcPowerUnkModule 2024-07-25 16:44:27,548 DEBUG exportSymbols: symbol ELTEK-BC2000-DC-POWER-MIB-V2-12::thirdPartyProduct 2024-07-25 16:44:27,548 DEBUG loadModule: loaded /home/conrad/.pyenv/versions/3.12.0/lib/python3.12/site-packages/pysnmp/smi/mibs/ELTEK-BC2000-DC-POWER-MIB-V2-12/home/conrad/.pyenv/versions/3.12.0/lib/python3.12/site-packages/pysnmp/smi/mibs/ELTEK-BC2000-DC-POWER-MIB-V2-12.py 2024-07-25 16:44:27,548 DEBUG MIB loaded successfully ...
Python package information
pysnmp-lextudio 6.2.1
Operating system information
Ubuntu 22.04.4 LTS
Python information
3.12.0
(Optional) Contents of your test script
No response
Relevant log output
No response
The right way to ensure your custom MIB documents are loaded and used at the right time is to pass mib_view_controller to your SnmpEngine instance,
snmpEngine = SnmpEngine()
snmpEngine.cache["mibViewController"] = mib_view_controller
Don't use resolveWithMib which has a much smaller effective scope.
@lextm This didn't fix the issue. I've tried with a simple code using a different custom mib file and the same error happens again.
import logging
from pysnmp.hlapi.asyncio import (
SnmpEngine,
CommunityData,
UdpTransportTarget,
ContextData,
ObjectType,
ObjectIdentity,
walkCmd
)
from pysnmp.smi import builder, view
logging.basicConfig(level=logging.DEBUG)
device_ip = '10.111.111.111'
community = 'random'
port = 161
mib_module = 'FELE-BATTERY-MONITORING-MIB'
dependency_mib = 'FELE-MIB'
entry_oid_name = 'batteryEntry'
def list_loaded_mibs(mib_builder):
"""
Prints the loaded MIB modules and their symbols for verification.
"""
print("\n--- Loaded MIB Modules ---")
for module_name in mib_builder.mibSymbols.keys():
print(module_name)
print("\n--- Symbols in Each MIB Module ---")
for module_name, symbols in mib_builder.mibSymbols.items():
print(f"\nModule: {module_name}")
for symbol_name in symbols.keys():
print(f" {symbol_name}")
async def snmp_walk(device_ip, community, port, mib_module, dependency_mib, entry_oid_name):
"""
Performs an SNMP walk to retrieve the subtree under the specified table entry OID using symbolic OID.
"""
mib_builder = builder.MibBuilder()
try:
mib_builder.loadModules(dependency_mib)
logging.debug(f"Successfully loaded MIB module: {dependency_mib}")
mib_builder.loadModules(mib_module)
logging.debug(f"Successfully loaded MIB module: {mib_module}")
list_loaded_mibs(mib_builder)
except Exception as e:
print(f"Failed to load MIB modules: {e}")
return
mib_view = view.MibViewController(mib_builder)
snmp_engine = SnmpEngine()
snmp_engine.cache["mibViewController"] = mib_view
try:
battery_entry_oid = mib_builder.importSymbols(mib_module, entry_oid_name)[0].getName()
logging.debug(f"'batteryEntry' OID: {'.'.join(str(x) for x in battery_entry_oid)}")
except Exception as e:
print(f"Failed to import 'batteryEntry' from MIB '{mib_module}': {e}")
return
try:
oid = ObjectIdentity(mib_module, entry_oid_name).resolveWithMib(mib_view)
logging.debug(f"Resolved OID: {oid.prettyPrint()}")
except Exception as e:
print(f"Failed to resolve OID '{entry_oid_name}' in MIB '{mib_module}': {e}")
return
walk_iterator = walkCmd(
snmp_engine,
CommunityData(community, mpModel=1),
UdpTransportTarget((device_ip, port)),
ContextData(),
ObjectType(oid),
lexicographicMode=False
)
try:
async for (errorIndication, errorStatus, errorIndex, varBinds) in walk_iterator:
if errorIndication:
print(f"Error: {errorIndication}")
break
elif errorStatus:
print(f"Error Status: {errorStatus.prettyPrint()} at {errorIndex}")
break
else:
for varBind in varBinds:
try:
symbolic_oid = varBind[0].prettyPrint()
value = varBind[1].prettyPrint()
except Exception as e:
logging.debug(f"Failed to resolve OID '{varBind[0]}' with MIB view: {e}")
symbolic_oid = varBind[0].prettyPrint()
value = varBind[1].prettyPrint()
print(f'{symbolic_oid} = {value}')
except Exception as e:
print(f"An exception occurred during SNMP walk: {e}")
finally:
snmp_engine.transportDispatcher.closeDispatcher()
def main():
asyncio.run(snmp_walk(device_ip, community, port, mib_module, dependency_mib, entry_oid_name))
if __name__ == '__main__':
main()
SNMPv2-SMI is still being used. This is the partial output:
SNMPv2-SMI::enterprises.56952.1.1.2.3.3.2.1.2.1.1 = SNDGCAJWBS1
SNMPv2-SMI::enterprises.56952.1.1.2.3.3.2.1.2.1.2 = SNDGCAJWBS1
SNMPv2-SMI::enterprises.56952.1.1.2.3.3.2.1.2.1.3 = SNDGCAJWBS1
SNMPv2-SMI::enterprises.56952.1.1.2.3.3.2.1.2.1.4 = SNDGCAJWBS1
SNMPv2-SMI::enterprises.56952.1.1.2.3.3.2.1.2.2.1 = SNDGCAJWBS2
SNMPv2-SMI::enterprises.56952.1.1.2.3.3.2.1.2.2.2 = SNDGCAJWBS2
I've tried using different releases as well, this code is running on the latest release:
pysmi 1.5.0
pysnmp 7.1.3
The compiled '.py' mib files reside in:
.venv/lib/python3.8/site-packages/pysnmp/smi/mibs/
Given the resource constraints, our focus is to cover the essentials, which are working as expected like test_v1_walk_mib in this commit.
We aren’t able to assist troubleshooting every possible case unless it is backed by commercial support contracts.