import asyncio from typing import Optional, Dict, List, Tuple import logging logger = logging.getLogger(__name__) # Try importing pysnmp with fallback try: from pysnmp.hlapi.v3arch.asyncio import ( SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity, getCmd, nextCmd, bulkCmd, UsmUserData, usmHMACMD5AuthProtocol, usmHMACSHAAuthProtocol, usmDESPrivProtocol, usmAesCfb128Protocol ) SNMP_AVAILABLE = True except ImportError: logger.warning("pysnmp v3arch not available, trying alternative import") try: from pysnmp.hlapi.asyncio import ( SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity, getCmd, nextCmd, bulkCmd, UsmUserData, usmHMACMD5AuthProtocol, usmHMACSHAAuthProtocol, usmDESPrivProtocol, usmAesCfb128Protocol ) SNMP_AVAILABLE = True except ImportError: logger.error("pysnmp not available - SNMP features will be disabled") SNMP_AVAILABLE = False from datetime import datetime class SNMPManager: """Advanced SNMP Manager supporting v1, v2c, and v3""" # Common OIDs OID_SYSTEM = { 'sysDescr': '1.3.6.1.2.1.1.1.0', 'sysObjectID': '1.3.6.1.2.1.1.2.0', 'sysUpTime': '1.3.6.1.2.1.1.3.0', 'sysContact': '1.3.6.1.2.1.1.4.0', 'sysName': '1.3.6.1.2.1.1.5.0', 'sysLocation': '1.3.6.1.2.1.1.6.0', } OID_HOST_RESOURCES = { 'hrProcessorLoad': '1.3.6.1.2.1.25.3.3.1.2', # CPU Load 'hrStorageUsed': '1.3.6.1.2.1.25.2.3.1.6', # Storage Used 'hrStorageSize': '1.3.6.1.2.1.25.2.3.1.5', # Storage Size 'hrMemorySize': '1.3.6.1.2.1.25.2.2.0', # Total Memory } OID_INTERFACES = { 'ifIndex': '1.3.6.1.2.1.2.2.1.1', 'ifDescr': '1.3.6.1.2.1.2.2.1.2', 'ifType': '1.3.6.1.2.1.2.2.1.3', 'ifMtu': '1.3.6.1.2.1.2.2.1.4', 'ifSpeed': '1.3.6.1.2.1.2.2.1.5', 'ifPhysAddress': '1.3.6.1.2.1.2.2.1.6', 'ifAdminStatus': '1.3.6.1.2.1.2.2.1.7', 'ifOperStatus': '1.3.6.1.2.1.2.2.1.8', 'ifInOctets': '1.3.6.1.2.1.2.2.1.10', 'ifOutOctets': '1.3.6.1.2.1.2.2.1.16', 'ifInErrors': '1.3.6.1.2.1.2.2.1.14', 'ifOutErrors': '1.3.6.1.2.1.2.2.1.20', } OID_SENSORS = { 'entPhySensorValue': '1.3.6.1.4.1.9.9.91.1.1.1.1.4', # Cisco Sensors 'temperature': '1.3.6.1.4.1.2021.13.16.2.1.3', # Net-SNMP Temperature } def __init__(self): if not SNMP_AVAILABLE: logger.warning("SNMP Manager initialized but pysnmp is not available") self.engine = None else: self.engine = SnmpEngine() def _get_auth_data(self, version: str, community: str = None, username: str = None, auth_key: str = None, priv_key: str = None, auth_protocol: str = 'MD5', priv_protocol: str = 'DES'): """Get authentication data based on SNMP version""" if version == 'v3': # SNMP v3 with authentication auth_proto = usmHMACMD5AuthProtocol if auth_protocol == 'MD5' else usmHMACSHAAuthProtocol priv_proto = usmDESPrivProtocol if priv_protocol == 'DES' else usmAesCfb128Protocol return UsmUserData( username, authKey=auth_key, privKey=priv_key, authProtocol=auth_proto, privProtocol=priv_proto ) elif version == 'v2c': return CommunityData(community or 'public', mpModel=1) else: # v1 return CommunityData(community or 'public', mpModel=0) async def get(self, ip: str, oid: str, version: str = 'v2c', community: str = 'public', port: int = 161, timeout: int = 5, **kwargs) -> Tuple[bool, Optional[str]]: """Get single OID value""" try: auth_data = self._get_auth_data(version, community, **kwargs) errorIndication, errorStatus, errorIndex, varBinds = await getCmd( self.engine, auth_data, await UdpTransportTarget.create((ip, port), timeout=timeout, retries=1), ContextData(), ObjectType(ObjectIdentity(oid)) ) if errorIndication: logger.error(f"SNMP GET error for {ip}: {errorIndication}") return False, None if errorStatus: logger.error(f"SNMP GET error: {errorStatus.prettyPrint()}") return False, None for varBind in varBinds: return True, str(varBind[1]) return False, None except Exception as e: logger.error(f"SNMP GET exception for {ip}: {e}") return False, None async def bulk_get(self, ip: str, oids: List[str], version: str = 'v2c', community: str = 'public', port: int = 161, timeout: int = 5, **kwargs) -> Dict[str, str]: """Get multiple OIDs at once""" results = {} try: auth_data = self._get_auth_data(version, community, **kwargs) object_types = [ObjectType(ObjectIdentity(oid)) for oid in oids] errorIndication, errorStatus, errorIndex, varBinds = await getCmd( self.engine, auth_data, await UdpTransportTarget.create((ip, port), timeout=timeout, retries=1), ContextData(), *object_types ) if errorIndication or errorStatus: return results for idx, varBind in enumerate(varBinds): results[oids[idx]] = str(varBind[1]) except Exception as e: logger.error(f"SNMP bulk GET exception for {ip}: {e}") return results async def walk(self, ip: str, oid: str, version: str = 'v2c', community: str = 'public', port: int = 161, timeout: int = 5, **kwargs) -> List[Tuple[str, str]]: """Walk OID tree""" results = [] try: auth_data = self._get_auth_data(version, community, **kwargs) async for (errorIndication, errorStatus, errorIndex, varBinds) in nextCmd( self.engine, auth_data, await UdpTransportTarget.create((ip, port), timeout=timeout, retries=1), ContextData(), ObjectType(ObjectIdentity(oid)), lexicographicMode=False ): if errorIndication or errorStatus: break for varBind in varBinds: results.append((str(varBind[0]), str(varBind[1]))) except Exception as e: logger.error(f"SNMP WALK exception for {ip}: {e}") return results async def get_system_info(self, ip: str, version: str = 'v2c', community: str = 'public', **kwargs) -> Dict: """Get system information""" oids = list(self.OID_SYSTEM.values()) results = await self.bulk_get(ip, oids, version, community, **kwargs) system_info = {} for key, oid in self.OID_SYSTEM.items(): system_info[key] = results.get(oid, 'N/A') return system_info async def get_cpu_usage(self, ip: str, version: str = 'v2c', community: str = 'public', **kwargs) -> Optional[float]: """Get CPU usage percentage""" # Try hrProcessorLoad first success, value = await self.get( ip, self.OID_HOST_RESOURCES['hrProcessorLoad'] + '.1', version, community, **kwargs ) if success and value: try: return float(value) except: pass # Try Cisco specific OID success, value = await self.get( ip, '1.3.6.1.4.1.9.2.1.56.0', # Cisco CPU 5sec version, community, **kwargs ) if success and value: try: return float(value) except: pass return None async def get_memory_usage(self, ip: str, version: str = 'v2c', community: str = 'public', **kwargs) -> Dict: """Get memory usage information""" # Get memory info from hrStorage table storage_walk = await self.walk( ip, '1.3.6.1.2.1.25.2.3.1', version, community, **kwargs ) memory_info = { 'total': 0, 'used': 0, 'free': 0, 'percentage': 0 } # Parse storage table for RAM for oid, value in storage_walk: if 'Physical Memory' in value or 'RAM' in value: # Found memory entry, get size and used try: index = oid.split('.')[-1] size_oid = f'1.3.6.1.2.1.25.2.3.1.5.{index}' used_oid = f'1.3.6.1.2.1.25.2.3.1.6.{index}' units_oid = f'1.3.6.1.2.1.25.2.3.1.4.{index}' size_results = await self.bulk_get( ip, [size_oid, used_oid, units_oid], version, community, **kwargs ) units = int(size_results.get(units_oid, 1)) size = int(size_results.get(size_oid, 0)) * units used = int(size_results.get(used_oid, 0)) * units memory_info['total'] = size memory_info['used'] = used memory_info['free'] = size - used if size > 0: memory_info['percentage'] = (used / size) * 100 break except Exception as e: logger.error(f"Memory parse error: {e}") return memory_info async def get_temperature(self, ip: str, version: str = 'v2c', community: str = 'public', **kwargs) -> Optional[float]: """Get device temperature""" # Try different temperature OIDs temp_oids = [ '1.3.6.1.4.1.9.9.13.1.3.1.3.1', # Cisco '1.3.6.1.4.1.2021.13.16.2.1.3.1', # Net-SNMP '1.3.6.1.4.1.2011.5.25.31.1.1.1.1.11', # Huawei ] for oid in temp_oids: success, value = await self.get(ip, oid, version, community, **kwargs) if success and value: try: temp = float(value) if temp > 0 and temp < 200: # Reasonable range return temp except: pass return None async def get_interfaces(self, ip: str, version: str = 'v2c', community: str = 'public', **kwargs) -> List[Dict]: """Get all interfaces with statistics""" interfaces = [] # Walk interface descriptions descr_walk = await self.walk( ip, self.OID_INTERFACES['ifDescr'], version, community, **kwargs ) for oid, descr in descr_walk: try: index = oid.split('.')[-1] # Get interface details oids = [ f"{self.OID_INTERFACES['ifOperStatus']}.{index}", f"{self.OID_INTERFACES['ifAdminStatus']}.{index}", f"{self.OID_INTERFACES['ifSpeed']}.{index}", f"{self.OID_INTERFACES['ifInOctets']}.{index}", f"{self.OID_INTERFACES['ifOutOctets']}.{index}", f"{self.OID_INTERFACES['ifInErrors']}.{index}", f"{self.OID_INTERFACES['ifOutErrors']}.{index}", ] results = await self.bulk_get(ip, oids, version, community, **kwargs) interface = { 'index': index, 'name': descr, 'status': 'up' if results.get(oids[0]) == '1' else 'down', 'admin_status': 'up' if results.get(oids[1]) == '1' else 'down', 'speed': int(results.get(oids[2], 0)), 'in_octets': int(results.get(oids[3], 0)), 'out_octets': int(results.get(oids[4], 0)), 'in_errors': int(results.get(oids[5], 0)), 'out_errors': int(results.get(oids[6], 0)), } interfaces.append(interface) except Exception as e: logger.error(f"Interface parse error: {e}") return interfaces async def get_full_device_info(self, ip: str, version: str = 'v2c', community: str = 'public', **kwargs) -> Dict: """Get comprehensive device information""" device_info = { 'ip': ip, 'timestamp': datetime.utcnow().isoformat(), 'system': {}, 'cpu': None, 'memory': {}, 'temperature': None, 'interfaces': [] } try: # Gather all information concurrently system_info, cpu, memory, temperature, interfaces = await asyncio.gather( self.get_system_info(ip, version, community, **kwargs), self.get_cpu_usage(ip, version, community, **kwargs), self.get_memory_usage(ip, version, community, **kwargs), self.get_temperature(ip, version, community, **kwargs), self.get_interfaces(ip, version, community, **kwargs), return_exceptions=True ) device_info['system'] = system_info if not isinstance(system_info, Exception) else {} device_info['cpu'] = cpu if not isinstance(cpu, Exception) else None device_info['memory'] = memory if not isinstance(memory, Exception) else {} device_info['temperature'] = temperature if not isinstance(temperature, Exception) else None device_info['interfaces'] = interfaces if not isinstance(interfaces, Exception) else [] except Exception as e: logger.error(f"Error gathering device info for {ip}: {e}") return device_info # Global instance snmp_manager = SNMPManager()