Files
SNMP-Influx-collector/cont/snmp/walker.py
YuruC3 10e6ae2347 Complete restructure to modules.
Using modules and imports makes it much more modular
and managable. There are no 600 row files now
2026-03-02 21:46:51 +01:00

65 lines
1.8 KiB
Python

import asyncio, os
from typing import Annotated, Final
# SNMP
from pysnmp.hlapi.v3arch.asyncio import *
# SNMP ENV-------------------------------------------
ROUND_PREC: Final[int] = int(os.getenv("ROUND_PREC", 2))
SNMPUSER: Final[str] = os.getenv("SNMPUSER", None)
SNMPPRIVKEY: Final[str] = os.getenv("SNMPPRIVKEY", None)
SNMPAUTHKEY: Final[str] = os.getenv("SNMPAUTHKEY", None)
# Right now I'll only use SHA
# SNMPAUTHPROTO: Final[str] = os.getenv("SNMPAUTHPROTO", "SHA")
# SNMPPRIVPROTO: Final[str] = os.getenv("SNMPPRIVPROTO", "SHA")
SNMPORT: Final[int] = int(os.getenv("SNMPORT", 161))
# Check if SNMP ENV are empty
if not SNMPUSER or not SNMPPRIVKEY or not SNMPAUTHKEY:
raise Exception("No SNMP user or/and PrivAuth passed")
# SNMP
USMUSRDATA = UsmUserData(
userName=SNMPUSER,
authKey=SNMPAUTHKEY,
privKey=SNMPPRIVKEY,
authProtocol=usmHMACSHAAuthProtocol,
privProtocol=usmAesCfb128Protocol,
)
# Return a following object
# {
# 1: "SomeObject",
# 2: "someOtherObject"
# }
# The value is always a string so it needs to be converted to int if needed
async def walk_column_v3(snmpEngine, remoteIP: str, base_oid: str) -> dict:
rows = {}
async for errInd, errStat, errIdx, varBinds in walk_cmd(
snmpEngine,
USMUSRDATA,
await UdpTransportTarget.create((remoteIP, SNMPORT)),
ContextData(),
ObjectType(ObjectIdentity(base_oid)),
lexicographicMode=False,
):
if errInd:
raise RuntimeError(errInd)
if errStat:
raise RuntimeError(errStat.prettyPrint())
for oid, val in varBinds:
# index is typically the last sub-identifier
idx = int(oid.prettyPrint().split(".")[-1])
rows[idx] = val.prettyPrint()
return rows