AI-Agent/ipam
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases 5 Wiki Activity

fix(snmp): 修复异步函数同步调用导致'coroutine object is not an iterator'错误

Browse Source 
问题原因:
- pysnmp.hlapi.asyncio 的 getCmd/bulkCmd/nextCmd 都是 async def 协程函数
- 返回的是 coroutine 对象，需要用 await 获取结果
- 原代码用 next() 或 for...in 同步迭代协程对象，报错:
  'coroutine' object is not an iterator

解决方案:
1. snmp_service.py: 所有 SNMP 操作改为 async def
   - test_connection(): getCmd 改为 await getCmd()
   - get_arp_table(): bulkCmd 改为 await bulkCmd()
   - get_mac_address_table(): bulkCmd 改为 await bulkCmd()
   - get_interfaces(): bulkCmd 改为 await bulkCmd()
   - poll_device(): 所有子调用加 await
2. api/v1/snmp.py: API 路由改为 async def + await

bulkCmd await 后返回 (errorIndication, errorStatus, errorIndex, varBindTable)
其中 varBindTable 是列表的列表（table 形式），需双层循环解析
This commit is contained in:
Your Name
2026-07-23 15:01:18 +08:00
parent 8b1df674ec
commit 094d2cc112
2 changed files with 117 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/v1/snmp.py
+6 -6
View File
@@ -437,7 +437,7 @@ def delete_network_device(
# ========== SNMP 操作 ========== # ========== SNMP 操作 ==========
@router.post("/devices/{device_id}/test", summary="测试 SNMP 连接") @router.post("/devices/{device_id}/test", summary="测试 SNMP 连接")
def test_snmp_connection(device_id: int, db: Session = Depends(get_db)): async def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
"""测试设备的 SNMP 连接（不写审计，避免刷屏）""" """测试设备的 SNMP 连接（不写审计，避免刷屏）"""
from app.models.snmp import NetworkDevice from app.models.snmp import NetworkDevice
@@ -448,7 +448,7 @@ def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
if not device.snmp_credential: if not device.snmp_credential:
raise HTTPException(status_code=400, detail="设备未配置 SNMP 凭据") raise HTTPException(status_code=400, detail="设备未配置 SNMP 凭据")
success, info = SNMPService.test_connection(device, device.snmp_credential) success, info = await SNMPService.test_connection(device, device.snmp_credential)
return { return {
"device_id": device_id, "device_id": device_id,
@@ -459,9 +459,9 @@ def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
@router.post("/devices/{device_id}/poll", summary="立即轮询设备") @router.post("/devices/{device_id}/poll", summary="立即轮询设备")
def poll_device_now(device_id: int, db: Session = Depends(get_db)): async def poll_device_now(device_id: int, db: Session = Depends(get_db)):
"""立即轮询设备，获取 ARP 表、MAC 地址表、接口信息（不写审计，常规操作）""" """立即轮询设备，获取 ARP 表、MAC 地址表、接口信息（不写审计，常规操作）"""
result = SNMPService.poll_device(db, device_id) result = await SNMPService.poll_device(db, device_id)
if 'error' in result: if 'error' in result:
raise HTTPException(status_code=400, detail=result['error']) raise HTTPException(status_code=400, detail=result['error'])
@@ -470,7 +470,7 @@ def poll_device_now(device_id: int, db: Session = Depends(get_db)):
@router.post("/devices/{device_id}/arp", summary="获取设备 ARP 表") @router.post("/devices/{device_id}/arp", summary="获取设备 ARP 表")
def get_device_arp_table(device_id: int, db: Session = Depends(get_db)): async def get_device_arp_table(device_id: int, db: Session = Depends(get_db)):
"""获取设备的 ARP 表""" """获取设备的 ARP 表"""
from app.models.snmp import NetworkDevice, ARPEntry from app.models.snmp import NetworkDevice, ARPEntry
@@ -481,7 +481,7 @@ def get_device_arp_table(device_id: int, db: Session = Depends(get_db)):
if not device.snmp_credential: if not device.snmp_credential:
raise HTTPException(status_code=400, detail="设备未配置 SNMP 凭据") raise HTTPException(status_code=400, detail="设备未配置 SNMP 凭据")
entries = SNMPService.get_arp_table(device, device.snmp_credential, db) entries = await SNMPService.get_arp_table(device, device.snmp_credential, db)
return { return {
"device_id": device_id, "device_id": device_id,
backend/app/services/snmp_service.py
+111 -112
View File
@@ -3,6 +3,7 @@ from sqlalchemy.orm import Session
from datetime import datetime from datetime import datetime
import ipaddress import ipaddress
import logging import logging
import asyncio
from pysnmp.hlapi.asyncio import ( from pysnmp.hlapi.asyncio import (
SnmpEngine, CommunityData, UsmUserData, SnmpEngine, CommunityData, UsmUserData,
@@ -117,7 +118,7 @@ class SNMPService:
return None return None
@staticmethod @staticmethod
def test_connection(device: NetworkDevice, credential: SNMPCredential) -> Tuple[bool, Dict[str, Any]]: async def test_connection(device: NetworkDevice, credential: SNMPCredential) -> Tuple[bool, Dict[str, Any]]:
""" """
测试 SNMP 连接 测试 SNMP 连接
返回: (成功, 设备信息字典) 返回: (成功, 设备信息字典)
@@ -131,11 +132,11 @@ class SNMPService:
retries=credential.retries retries=credential.retries
) )
errorIndication, errorStatus, errorIndex, varBinds = next( errorIndication, errorStatus, errorIndex, varBinds = await getCmd(
getCmd(SnmpEngine(), auth_data, transport, ContextData(), SnmpEngine(), auth_data, transport, ContextData(),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_DESCR)), ObjectType(ObjectIdentity(SNMPService.OID_SYS_DESCR)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_NAME)), ObjectType(ObjectIdentity(SNMPService.OID_SYS_NAME)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_LOCATION))) ObjectType(ObjectIdentity(SNMPService.OID_SYS_LOCATION))
) )
if errorIndication: if errorIndication:
@@ -162,7 +163,7 @@ class SNMPService:
return False, {'error': str(e)} return False, {'error': str(e)}
@staticmethod @staticmethod
def get_arp_table(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]: async def get_arp_table(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
""" """
获取 ARP 表 (IP -> MAC 映射) 获取 ARP 表 (IP -> MAC 映射)
""" """
@@ -178,61 +179,60 @@ class SNMPService:
) )
# 使用 bulkCmd 获取 ARP 表 # 使用 bulkCmd 获取 ARP 表
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd( errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(), SnmpEngine(), auth_data, transport, ContextData(),
0, 50, # nonRepeaters, maxRepetitions 0, 50, # nonRepeaters, maxRepetitions
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS)), 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_NET_ADDRESS)),
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX)), ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX)),
lexicographicMode=False lexicographicMode=False
): )
if errorIndication: if errorIndication:
logger.error(f"SNMP 错误: {errorIndication}") logger.error(f"SNMP 错误: {errorIndication}")
break elif errorStatus:
logger.error(f"SNMP 错误: {errorStatus}")
else:
# varBindTable 是列表的列表，每行是一条 SNMP 响应记录
for varBinds in varBindTable:
# 解析结果
arp_data = {}
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
if errorStatus: # 提取 IP 地址 (OID 后缀: ifIndex.IP)
logger.error(f"SNMP 错误: {errorStatus}") parts = oid.split('.')
break 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:
arp_data = {} mac = SNMPService._normalize_mac(value)
for varBind in varBinds: if mac:
oid = str(varBind[0]) if ip_address not in arp_data:
value = varBind[1] 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
# 提取 IP 地址 (OID 后缀: ifIndex.IP) # 转换为列表
parts = oid.split('.') now = datetime.utcnow()
if len(parts) >= 4: for ip, data in arp_data.items():
# 从 OID 中提取 IP 地址 if 'mac_address' in data:
ip_parts = parts[-4:] entry = {
ip_address = '.'.join(ip_parts) 'ip_address': ip,
'mac_address': data.get('mac_address'),
if SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS in oid: 'interface': str(data.get('if_index', '')),
mac = SNMPService._normalize_mac(value) 'device_id': device.id,
if mac: 'last_seen': now.isoformat(),
if ip_address not in arp_data: 'discovered_at': now.isoformat()
arp_data[ip_address] = {} }
arp_data[ip_address]['mac_address'] = mac arp_entries.append(entry)
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: for entry_data in arp_entries:
@@ -281,7 +281,7 @@ class SNMPService:
return arp_entries return arp_entries
@staticmethod @staticmethod
def get_mac_address_table(device: NetworkDevice, credential: SNMPCredential, db: Session, vlan_id: Optional[int] = None) -> List[Dict[str, Any]]: async def get_mac_address_table(device: NetworkDevice, credential: SNMPCredential, db: Session, vlan_id: Optional[int] = None) -> List[Dict[str, Any]]:
""" """
获取 MAC 地址表 (MAC -> 端口 映射) 获取 MAC 地址表 (MAC -> 端口 映射)
""" """
@@ -297,39 +297,39 @@ class SNMPService:
) )
# 获取 MAC 地址表 # 获取 MAC 地址表
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd( errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(), SnmpEngine(), auth_data, transport, ContextData(),
0, 100, 0, 100,
ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_PORT)), ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_PORT)),
ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_STATUS)), ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_STATUS)),
lexicographicMode=False lexicographicMode=False
): )
if errorIndication: if errorIndication:
break logger.error(f"获取 MAC 地址表失败: {errorIndication}")
elif errorStatus:
logger.error(f"获取 MAC 地址表失败: {errorStatus}")
else:
for varBinds in varBindTable:
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
if errorStatus: # OID 后缀是 MAC 地址: .a.b.c.d.e.f
break parts = oid.split('.')
if len(parts) >= 6:
mac_parts = parts[-6:]
mac_address = ':'.join(f'{int(p):02X}' for p in mac_parts)
for varBind in varBinds: if SNMPService.OID_DOT1D_TP_FDB_PORT in oid:
oid = str(varBind[0]) port = int(value) if isinstance(value, Integer32) else None
value = varBind[1]
# OID 后缀是 MAC 地址: .a.b.c.d.e.f entry = {
parts = oid.split('.') 'mac_address': mac_address,
if len(parts) >= 6: 'port_number': port,
mac_parts = parts[-6:] 'device_id': device.id,
mac_address = ':'.join(f'{int(p):02X}' for p in mac_parts) 'vlan_id': vlan_id
}
if SNMPService.OID_DOT1D_TP_FDB_PORT in oid: mac_entries.append(entry)
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() now = datetime.utcnow()
@@ -361,7 +361,7 @@ class SNMPService:
return mac_entries return mac_entries
@staticmethod @staticmethod
def get_interfaces(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]: async def get_interfaces(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
""" """
获取设备接口信息 获取设备接口信息
""" """
@@ -379,7 +379,7 @@ class SNMPService:
# 获取接口信息 # 获取接口信息
interface_data = {} interface_data = {}
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd( errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(), SnmpEngine(), auth_data, transport, ContextData(),
0, 100, 0, 100,
ObjectType(ObjectIdentity(SNMPService.OID_IF_DESCR)), ObjectType(ObjectIdentity(SNMPService.OID_IF_DESCR)),
@@ -390,35 +390,34 @@ class SNMPService:
ObjectType(ObjectIdentity(SNMPService.OID_IF_ADMIN_STATUS)), ObjectType(ObjectIdentity(SNMPService.OID_IF_ADMIN_STATUS)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_OPER_STATUS)), ObjectType(ObjectIdentity(SNMPService.OID_IF_OPER_STATUS)),
lexicographicMode=False lexicographicMode=False
): )
if errorIndication or errorStatus: if not errorIndication and not errorStatus:
break for varBinds in varBindTable:
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
parts = oid.split('.')
if_index = parts[-1]
for varBind in varBinds: if if_index not in interface_data:
oid = str(varBind[0]) interface_data[if_index] = {'if_index': int(if_index)}
value = varBind[1]
parts = oid.split('.')
if_index = parts[-1]
if if_index not in interface_data: if SNMPService.OID_IF_DESCR in oid:
interface_data[if_index] = {'if_index': int(if_index)} interface_data[if_index]['if_descr'] = value.prettyPrint()
elif SNMPService.OID_IF_TYPE in oid:
if SNMPService.OID_IF_DESCR in oid: interface_data[if_index]['if_type'] = str(value)
interface_data[if_index]['if_descr'] = value.prettyPrint() elif SNMPService.OID_IF_MTU in oid:
elif SNMPService.OID_IF_TYPE in oid: interface_data[if_index]['if_mtu'] = int(value) if isinstance(value, Integer32) else None
interface_data[if_index]['if_type'] = str(value) elif SNMPService.OID_IF_SPEED in oid:
elif SNMPService.OID_IF_MTU in oid: interface_data[if_index]['if_speed'] = int(value) if isinstance(value, (Integer32, Gauge32)) else None
interface_data[if_index]['if_mtu'] = int(value) if isinstance(value, Integer32) else None elif SNMPService.OID_IF_PHYS_ADDRESS in oid:
elif SNMPService.OID_IF_SPEED in oid: interface_data[if_index]['if_phys_address'] = SNMPService._normalize_mac(value)
interface_data[if_index]['if_speed'] = int(value) if isinstance(value, (Integer32, Gauge32)) else None elif SNMPService.OID_IF_ADMIN_STATUS in oid:
elif SNMPService.OID_IF_PHYS_ADDRESS in oid: status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_phys_address'] = SNMPService._normalize_mac(value) interface_data[if_index]['if_admin_status'] = status_map.get(int(value), 'unknown')
elif SNMPService.OID_IF_ADMIN_STATUS in oid: elif SNMPService.OID_IF_OPER_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'} status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_admin_status'] = status_map.get(int(value), 'unknown') interface_data[if_index]['if_oper_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() now = datetime.utcnow()
@@ -485,7 +484,7 @@ class SNMPService:
return device return device
@staticmethod @staticmethod
def poll_device(db: Session, device_id: int) -> Dict[str, Any]: async def poll_device(db: Session, device_id: int) -> Dict[str, Any]:
"""轮询设备，获取所有信息""" """轮询设备，获取所有信息"""
device = db.query(NetworkDevice).filter(NetworkDevice.id == device_id).first() device = db.query(NetworkDevice).filter(NetworkDevice.id == device_id).first()
if not device: if not device:
@@ -505,17 +504,17 @@ class SNMPService:
} }
# 获取 ARP 表 # 获取 ARP 表
arp_entries = SNMPService.get_arp_table(device, credential, db) arp_entries = await SNMPService.get_arp_table(device, credential, db)
result['arp_entries'] = arp_entries result['arp_entries'] = arp_entries
result['arp_count'] = len(arp_entries) result['arp_count'] = len(arp_entries)
# 获取 MAC 地址表 # 获取 MAC 地址表
mac_entries = SNMPService.get_mac_address_table(device, credential, db) mac_entries = await SNMPService.get_mac_address_table(device, credential, db)
result['mac_entries'] = mac_entries result['mac_entries'] = mac_entries
result['mac_count'] = len(mac_entries) result['mac_count'] = len(mac_entries)
# 获取接口信息 # 获取接口信息
interfaces = SNMPService.get_interfaces(device, credential, db) interfaces = await SNMPService.get_interfaces(device, credential, db)
result['interfaces'] = interfaces result['interfaces'] = interfaces
result['interface_count'] = len(interfaces) result['interface_count'] = len(interfaces)