snmp_service_py310.y #3
cnbugs
commented
Owner
```
#from typing import List, Dict, Any, Optional, Tuple
#from sqlalchemy.orm import Session
#from datetime import datetime
#import ipaddress
#import logging
#
##from pysnmp.hlapi import (
## SnmpEngine, CommunityData, UsmUserData,
## UdpTransportTarget, ContextData,
## ObjectType, ObjectIdentity,
## bulkCmd, getCmd, nextCmd
##)
#from pysnmp.hlapi import SnmpEngine
#from pysnmp.hlapi import (
# SnmpEngine, CommunityData, UsmUserData,
# UdpTransportTarget, ContextData,
# ObjectType, ObjectIdentity,
# bulkCmd, getCmd, nextCmd
#)
#
#
#
#from app.models.snmp import (
# SNMPCredential, NetworkDevice, ARPEntry, MACAddressTable, DeviceInterface,
# SNMPVersion, SNMPAuthProtocol, SNMPPrivProtocol
#)
#from app.services.enhanced_scan_service import EnhancedScanService
#
#logger = logging.getLogger(__name__)
#
from typing import List, Dict, Any, Optional, Tuple
from sqlalchemy.orm import Session
from datetime import datetime
import ipaddress
import logging
# =====================================================================
# 针对 Python 3.10 + PySNMP 4.x 的兼容性补丁 (Monkey Patch)
# =====================================================================
import collections
import collections.abc
import asyncio
# 修复 pyasn1 / pysmi 底层引发的崩溃
if not hasattr(collections, 'Callable'):
collections.Callable = collections.abc.Callable
if not hasattr(collections, 'MutableMapping'):
collections.MutableMapping = collections.abc.MutableMapping
# 修复 pysnmp.hlapi.asyncio 引发的崩溃
if not hasattr(asyncio, 'coroutine'):
asyncio.coroutine = lambda f: f
# =====================================================================
# 补丁打完后，现在可以安全导入最标准的 hlapi 了，再也不会报 unknown location
from pysnmp.hlapi import (
SnmpEngine, CommunityData, UsmUserData,
UdpTransportTarget, ContextData,
ObjectType, ObjectIdentity,
bulkCmd, getCmd, nextCmd
)
# 务必加上这三个解析类型，否则下面解析 MAC 和 端口 遇到 isinstance() 时会报 NameError
from pysnmp.hlapi import (
SnmpEngine, CommunityData, UsmUserData,
UdpTransportTarget, ContextData,
ObjectType, ObjectIdentity,
bulkCmd, getCmd, nextCmd,
Integer32, OctetString, Gauge32 # <--- 直接加在这里
)
from app.models.snmp import (
SNMPCredential, NetworkDevice, ARPEntry, MACAddressTable, DeviceInterface,
SNMPVersion, SNMPAuthProtocol, SNMPPrivProtocol
)
from app.services.enhanced_scan_service import EnhancedScanService
logger = logging.getLogger(__name__)
class SNMPService:
"""SNMP 服务"""
# OID 定义
OID_SYSTEM = '1.3.6.1.2.1.1'
OID_SYS_DESCR = '1.3.6.1.2.1.1.1.0'
OID_SYS_NAME = '1.3.6.1.2.1.1.5.0'
OID_SYS_LOCATION = '1.3.6.1.2.1.1.6.0'
# 接口相关 OID
OID_IF_TABLE = '1.3.6.1.2.1.2.2'
OID_IF_INDEX = '1.3.6.1.2.1.2.2.1.1'
OID_IF_DESCR = '1.3.6.1.2.1.2.2.1.2'
OID_IF_TYPE = '1.3.6.1.2.1.2.2.1.3'
OID_IF_MTU = '1.3.6.1.2.1.2.2.1.4'
OID_IF_SPEED = '1.3.6.1.2.1.2.2.1.5'
OID_IF_PHYS_ADDRESS = '1.3.6.1.2.1.2.2.1.6'
OID_IF_ADMIN_STATUS = '1.3.6.1.2.1.2.2.1.7'
OID_IF_OPER_STATUS = '1.3.6.1.2.1.2.2.1.8'
OID_IF_IN_OCTETS = '1.3.6.1.2.1.2.2.1.10'
OID_IF_OUT_OCTETS = '1.3.6.1.2.1.2.2.1.16'
OID_IF_IN_ERRORS = '1.3.6.1.2.1.2.2.1.14'
OID_IF_OUT_ERRORS = '1.3.6.1.2.1.2.2.1.20'
# IP 地址表
OID_IP_ADDR_TABLE = '1.3.6.1.2.1.4.20'
OID_IP_ADDR_ENTRIES = '1.3.6.1.2.1.4.20.1.1'
OID_IP_ADDR_IF_INDEX = '1.3.6.1.2.1.4.20.1.2'
OID_IP_ADDR_NETMASK = '1.3.6.1.2.1.4.20.1.3'
# ARP 表 (ipNetToMediaTable)
OID_IP_NET_TO_MEDIA_PHYS_ADDRESS = '1.3.6.1.2.1.4.22.1.2'
OID_IP_NET_TO_MEDIA_IF_INDEX = '1.3.6.1.2.1.4.22.1.1'
OID_IP_NET_TO_MEDIA_NET_ADDRESS = '1.3.6.1.2.1.4.22.1.3'
OID_IP_NET_TO_MEDIA_TYPE = '1.3.6.1.2.1.4.22.1.4'
# DOT1D 桥接 MIB (MAC 地址表)
OID_DOT1D_TP_FDB_PORT = '1.3.6.1.2.1.17.4.3.1.2'
OID_DOT1D_TP_FDB_STATUS = '1.3.6.1.2.1.17.4.3.1.3'
# VLAN 相关 (Cisco 特定)
OID_VTP_VLAN_STATE = '1.3.6.1.4.1.9.9.46.1.3.1.1.2'
@staticmethod
def _create_auth_data(credential: SNMPCredential):
"""创建 SNMP 认证数据"""
if credential.version == SNMPVersion.V2C:
if not credential.community_string:
raise ValueError("SNMPv2c 凭据缺少 community_string")
return CommunityData(credential.community_string)
else: # V3
if not credential.username:
raise ValueError("SNMPv3 凭据缺少 username（V3 强制要求 1-32 字符）")
auth_protocol_map = {
SNMPAuthProtocol.MD5: 'MD5',
SNMPAuthProtocol.SHA: 'SHA',
SNMPAuthProtocol.SHA256: 'SHA256',
}
priv_protocol_map = {
SNMPPrivProtocol.DES: 'DES',
SNMPPrivProtocol.AES: 'AES',
SNMPPrivProtocol.AES256: 'AES256',
}
auth_proto = auth_protocol_map.get(credential.auth_protocol) if credential.auth_protocol else None
priv_proto = priv_protocol_map.get(credential.priv_protocol) if credential.priv_protocol else None
return UsmUserData(
userName=credential.username,
authProtocol=auth_proto,
authKey=credential.auth_password,
privProtocol=priv_proto,
privKey=credential.priv_password
)
@staticmethod
def _get_transport_target(ip_address: str, port: int = 161, timeout: int = 3, retries: int = 2):
"""创建传输目标"""
return UdpTransportTarget((ip_address, port), timeout=timeout, retries=retries)
@staticmethod
def _normalize_mac(mac_bytes) -> Optional[str]:
"""将 MAC 地址字节转换为标准格式 AA:BB:CC:DD:EE:FF"""
if not mac_bytes:
return None
if isinstance(mac_bytes, OctetString):
mac_bytes = mac_bytes.asOctets()
if len(mac_bytes) == 6:
return ':'.join(f'{b:02X}' for b in mac_bytes)
return None
@staticmethod
def test_connection(device: NetworkDevice, credential: SNMPCredential) -> Tuple[bool, Dict[str, Any]]:
"""
测试 SNMP 连接
返回: (成功, 设备信息字典)
"""
try:
auth_data = SNMPService._create_auth_data(credential)
transport = SNMPService._get_transport_target(
device.ip_address,
port=device.port,
timeout=credential.timeout,
retries=credential.retries
)
errorIndication, errorStatus, errorIndex, varBinds = next(
getCmd(SnmpEngine(), auth_data, transport, ContextData(),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_DESCR)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_NAME)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_LOCATION)))
)
if errorIndication:
return False, {'error': str(errorIndication)}
if errorStatus:
return False, {'error': f'{errorStatus.prettyPrint()} at {errorIndex and varBinds[int(errorIndex) - 1][0] or "?"}'}
sys_info = {}
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1].prettyPrint()
if SNMPService.OID_SYS_DESCR in oid:
sys_info['sys_descr'] = value
elif SNMPService.OID_SYS_NAME in oid:
sys_info['sys_name'] = value
elif SNMPService.OID_SYS_LOCATION in oid:
sys_info['sys_location'] = value
return True, sys_info
except Exception as e:
logger.error(f"SNMP 连接测试失败: {e}")
return False, {'error': str(e)}
@staticmethod
def get_arp_table(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
"""
获取 ARP 表 (IP -> MAC 映射)
"""
arp_entries = []
try:
auth_data = SNMPService._create_auth_data(credential)
transport = SNMPService._get_transport_target(
device.ip_address,
port=device.port,
timeout=credential.timeout,
retries=credential.retries
)
# 使用 bulkCmd 获取 ARP 表
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(),
0, 50, # nonRepeaters, maxRepetitions
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS)),
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_NET_ADDRESS)),
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX)),
lexicographicMode=False
):
if errorIndication:
logger.error(f"SNMP 错误: {errorIndication}")
break
if errorStatus:
logger.error(f"SNMP 错误: {errorStatus}")
break
# 解析结果
arp_data = {}
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
# 提取 IP 地址 (OID 后缀: ifIndex.IP)
parts = oid.split('.')
if len(parts) >= 4:
# 从 OID 中提取 IP 地址
ip_parts = parts[-4:]
ip_address = '.'.join(ip_parts)
if SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS in oid:
mac = SNMPService._normalize_mac(value)
if mac:
if ip_address not in arp_data:
arp_data[ip_address] = {}
arp_data[ip_address]['mac_address'] = mac
elif SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX in oid:
if isinstance(value, Integer32):
if_index = int(value)
if ip_address not in arp_data:
arp_data[ip_address] = {}
arp_data[ip_address]['if_index'] = if_index
# 转换为列表
now = datetime.utcnow()
for ip, data in arp_data.items():
if 'mac_address' in data:
entry = {
'ip_address': ip,
'mac_address': data.get('mac_address'),
'interface': str(data.get('if_index', '')),
'device_id': device.id,
'last_seen': now.isoformat(),
'discovered_at': now.isoformat()
}
arp_entries.append(entry)
# 保存到数据库
for entry_data in arp_entries:
# 查找是否已存在
existing = db.query(ARPEntry).filter(
ARPEntry.device_id == device.id,
ARPEntry.ip_address == entry_data['ip_address']
).first()
if existing:
existing.mac_address = entry_data['mac_address']
existing.last_seen = now
else:
arp_entry = ARPEntry(
device_id=device.id,
ip_address=entry_data['ip_address'],
mac_address=entry_data['mac_address'],
interface=entry_data['interface'],
discovered_at=now,
last_seen=now
)
db.add(arp_entry)
db.commit()
# 更新 IP 资产台账中的 MAC 地址
for entry in arp_entries:
from app.models.network import IPAddress as IPAddressModel
ip_addr = db.query(IPAddressModel).filter(
IPAddressModel.ip_address == entry['ip_address']
).first()
if ip_addr and not ip_addr.mac_address:
ip_addr.mac_address = entry['mac_address']
# 识别厂商
ip_addr.vendor = EnhancedScanService.get_mac_vendor(entry['mac_address'])
db.commit()
device.last_successful_poll = now
except Exception as e:
logger.error(f"获取 ARP 表失败: {e}", exc_info=True)
device.last_polled_at = datetime.utcnow()
db.commit()
return arp_entries
@staticmethod
def get_mac_address_table(device: NetworkDevice, credential: SNMPCredential, db: Session, vlan_id: Optional[int] = None) -> List[Dict[str, Any]]:
"""
获取 MAC 地址表 (MAC -> 端口 映射)
"""
mac_entries = []
try:
auth_data = SNMPService._create_auth_data(credential)
transport = SNMPService._get_transport_target(
device.ip_address,
port=device.port,
timeout=credential.timeout,
retries=credential.retries
)
# 获取 MAC 地址表
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(),
0, 100,
ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_PORT)),
ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_STATUS)),
lexicographicMode=False
):
if errorIndication:
break
if errorStatus:
break
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
# OID 后缀是 MAC 地址: .a.b.c.d.e.f
parts = oid.split('.')
if len(parts) >= 6:
mac_parts = parts[-6:]
mac_address = ':'.join(f'{int(p):02X}' for p in mac_parts)
if SNMPService.OID_DOT1D_TP_FDB_PORT in oid:
port = int(value) if isinstance(value, Integer32) else None
entry = {
'mac_address': mac_address,
'port_number': port,
'device_id': device.id,
'vlan_id': vlan_id
}
mac_entries.append(entry)
# 保存到数据库
now = datetime.utcnow()
for entry_data in mac_entries:
existing = db.query(MACAddressTable).filter(
MACAddressTable.device_id == device.id,
MACAddressTable.mac_address == entry_data['mac_address']
).first()
if existing:
existing.port_number = entry_data['port_number']
existing.last_seen = now
else:
mac_entry = MACAddressTable(
device_id=device.id,
mac_address=entry_data['mac_address'],
port_number=entry_data['port_number'],
vlan_id=vlan_id,
discovered_at=now,
last_seen=now
)
db.add(mac_entry)
db.commit()
except Exception as e:
logger.error(f"获取 MAC 地址表失败: {e}", exc_info=True)
return mac_entries
@staticmethod
def get_interfaces(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
"""
获取设备接口信息
"""
interfaces = []
try:
auth_data = SNMPService._create_auth_data(credential)
transport = SNMPService._get_transport_target(
device.ip_address,
port=device.port,
timeout=credential.timeout,
retries=credential.retries
)
# 获取接口信息
interface_data = {}
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(),
0, 100,
ObjectType(ObjectIdentity(SNMPService.OID_IF_DESCR)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_TYPE)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_MTU)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_SPEED)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_PHYS_ADDRESS)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_ADMIN_STATUS)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_OPER_STATUS)),
lexicographicMode=False
):
if errorIndication or errorStatus:
break
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
parts = oid.split('.')
if_index = parts[-1]
if if_index not in interface_data:
interface_data[if_index] = {'if_index': int(if_index)}
if SNMPService.OID_IF_DESCR in oid:
interface_data[if_index]['if_descr'] = value.prettyPrint()
elif SNMPService.OID_IF_TYPE in oid:
interface_data[if_index]['if_type'] = str(value)
elif SNMPService.OID_IF_MTU in oid:
interface_data[if_index]['if_mtu'] = int(value) if isinstance(value, Integer32) else None
elif SNMPService.OID_IF_SPEED in oid:
interface_data[if_index]['if_speed'] = int(value) if isinstance(value, (Integer32, Gauge32)) else None
elif SNMPService.OID_IF_PHYS_ADDRESS in oid:
interface_data[if_index]['if_phys_address'] = SNMPService._normalize_mac(value)
elif SNMPService.OID_IF_ADMIN_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_admin_status'] = status_map.get(int(value), 'unknown')
elif SNMPService.OID_IF_OPER_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_oper_status'] = status_map.get(int(value), 'unknown')
# 转换为列表并保存
now = datetime.utcnow()
for if_index, data in interface_data.items():
interfaces.append(data)
existing = db.query(DeviceInterface).filter(
DeviceInterface.device_id == device.id,
DeviceInterface.if_index == int(if_index)
).first()
if existing:
for key, val in data.items():
if hasattr(existing, key):
setattr(existing, key, val)
existing.last_polled_at = now
else:
iface = DeviceInterface(
device_id=device.id,
if_index=int(if_index),
if_descr=data.get('if_descr'),
if_type=data.get('if_type'),
if_mtu=data.get('if_mtu'),
if_speed=data.get('if_speed'),
if_phys_address=data.get('if_phys_address'),
if_admin_status=data.get('if_admin_status'),
if_oper_status=data.get('if_oper_status'),
last_polled_at=now
)
db.add(iface)
db.commit()
except Exception as e:
logger.error(f"获取接口信息失败: {e}", exc_info=True)
return interfaces
@staticmethod
def create_credential(db: Session, name: str, version: SNMPVersion, **kwargs) -> SNMPCredential:
"""创建 SNMP 凭据"""
credential = SNMPCredential(
name=name,
version=version,
**kwargs
)
db.add(credential)
db.commit()
db.refresh(credential)
return credential
@staticmethod
def create_device(db: Session, name: str, ip_address: str, credential_id: Optional[int] = None, **kwargs) -> NetworkDevice:
"""创建网络设备"""
device = NetworkDevice(
name=name,
ip_address=ip_address,
snmp_credential_id=credential_id,
**kwargs
)
db.add(device)
db.commit()
db.refresh(device)
return device
@staticmethod
def poll_device(db: Session, device_id: int) -> Dict[str, Any]:
"""轮询设备，获取所有信息"""
device = db.query(NetworkDevice).filter(NetworkDevice.id == device_id).first()
if not device:
return {'error': '设备不存在'}
if not device.snmp_credential:
return {'error': '设备未配置 SNMP 凭据'}
credential = device.snmp_credential
result = {
'device_id': device_id,
'device_name': device.name,
'arp_entries': [],
'mac_entries': [],
'interfaces': []
}
# 获取 ARP 表
arp_entries = SNMPService.get_arp_table(device, credential, db)
result['arp_entries'] = arp_entries
result['arp_count'] = len(arp_entries)
# 获取 MAC 地址表
mac_entries = SNMPService.get_mac_address_table(device, credential, db)
result['mac_entries'] = mac_entries
result['mac_count'] = len(mac_entries)
# 获取接口信息
interfaces = SNMPService.get_interfaces(device, credential, db)
result['interfaces'] = interfaces
result['interface_count'] = len(interfaces)
return result
```
Reference in New Issue
Block a user
Delete Branch "%!s()"
Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?