3 Commits

Author SHA1 Message Date
Your Name 9e34f6f0f9 feat(ips): 批量补全厂商功能 + MAC厂商替换为IEEE OUI数据库
前端:
- IPs.vue: 新增「补全厂商」按钮,选中网段后一键补全
- scanApi: 新增 batchFillVendor/batchDiscoverHostname + enable_netbios 参数

后端:
- enhanced_scan.py: 新增 POST /scan/batch/vendor 批量补全厂商端点
- enhanced_scan.py: 新增 POST /scan/batch/hostname 批量发现主机名端点
- scan_tasks.py: 透传 enable_netbios 参数
2026-07-18 10:59:27 +08:00
Your Name ccd7fbe2f0 feat(scan): 增强MAC厂商识别和主机名自动发现
- MAC厂商: 替换硬编码OUI_MAP为mac_vendor_lookup库(IEEE OUI官方数据库)
- 主机名发现新增NetBIOS广播(nmblookup),支持Windows/SMB设备
- 综合扫描支持enable_netbios参数
- 新增NetBIOS主机名发现API端点
- 扫描时自动补全缺失厂商(有MAC无vendor时自动识别)
2026-07-18 10:53:33 +08:00
Your Name bd68abee93 fix(snmp): ARP表缺少发现时间字段,在get_arp_table返回值中添加last_seen和discovered_at 2026-07-18 09:50:20 +08:00
14 changed files with 300 additions and 80 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+93 -1
View File
@@ -24,6 +24,7 @@ def comprehensive_scan(
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2,
db: Session = Depends(get_db)
):
@@ -32,7 +33,8 @@ def comprehensive_scan(
- Ping 扫描检测在线状态
- ARP 扫描获取MAC地址
- 反向DNS解析获取主机名
- MAC厂商识别
- NetBIOS 广播获取主机名(Windows/SMB 设备)
- MAC 厂商识别(IEEE OUI 数据库)
"""
network = NetworkService.get_by_id(db, network_id)
if not network:
@@ -51,6 +53,7 @@ def comprehensive_scan(
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
)
@@ -90,6 +93,7 @@ def comprehensive_scan_ip(
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2,
db: Session = Depends(get_db)
):
@@ -101,6 +105,7 @@ def comprehensive_scan_ip(
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
)
@@ -140,6 +145,93 @@ def reverse_dns(ip_address: str, timeout: int = 2):
"hostname": hostname or "Unknown"
}
@router.post("/netbios/lookup/{ip_address}", summary="NetBIOS 主机名发现")
def netbios_lookup(ip_address: str, timeout: int = 3):
"""
通过 NetBIOS 广播协议获取主机名(Windows / SMB 设备)
"""
hostname = EnhancedScanService.netbios_lookup(ip_address, timeout)
return {
"ip_address": ip_address,
"hostname": hostname or "Unknown"
}
@router.post("/batch/vendor", summary="批量补全厂商信息")
def batch_fill_vendor(
network_id: Optional[int] = None,
db: Session = Depends(get_db)
):
"""
对已有 MAC 地址但缺少厂商信息的 IP 批量补全厂商(使用 IEEE OUI 数据库)
"""
from app.models.network import IPAddress as IPAddressModel
query = db.query(IPAddressModel).filter(
IPAddressModel.mac_address.isnot(None),
IPAddressModel.vendor.is_(None)
)
if network_id:
query = query.filter(IPAddressModel.network_id == network_id)
items = query.all()
updated_count = 0
for ip in items:
vendor = EnhancedScanService.get_mac_vendor(ip.mac_address)
if vendor:
ip.vendor = vendor
updated_count += 1
if updated_count > 0:
db.commit()
return {
"message": f"已补全 {updated_count} 条厂商信息",
"checked": len(items),
"updated": updated_count
}
@router.post("/batch/hostname", summary="批量发现主机名")
def batch_discover_hostname(
network_id: Optional[int] = None,
enable_dns: bool = True,
enable_netbios: bool = True,
db: Session = Depends(get_db)
):
"""
对已有 MAC 地址但缺少主机名的 IP 批量执行主机名发现(DNS PTR + NetBIOS
"""
from app.models.network import IPAddress as IPAddressModel
query = db.query(IPAddressModel).filter(
IPAddressModel.mac_address.isnot(None),
IPAddressModel.hostname.is_(None)
)
if network_id:
query = query.filter(IPAddressModel.network_id == network_id)
items = query.all()
updated_count = 0
for ip in items:
hostname = None
if enable_dns:
hostname = EnhancedScanService.reverse_dns_lookup(ip.ip_address, timeout=2)
if not hostname and enable_netbios:
hostname = EnhancedScanService.netbios_lookup(ip.ip_address, timeout=3)
if hostname:
ip.hostname = hostname
updated_count += 1
if updated_count > 0:
db.commit()
return {
"message": f"已发现 {updated_count} 条主机名",
"checked": len(items),
"updated": updated_count
}
@router.get("/online/ips", summary="获取所有在线IP列表")
def get_online_ips(
+159 -72
View File
@@ -6,96 +6,102 @@ import subprocess
import socket
import ipaddress
import re
import logging
try:
from mac_vendor_lookup import MacLookup, VendorNotFoundError
MAC_LOOKUP_AVAILABLE = True
except ImportError:
MAC_LOOKUP_AVAILABLE = False
from app.models.network import ScanTask, Network, IPAddress
from app.models.network import TaskStatus, TaskType, IPStatus
from app.core.config import settings
logger = __import__('logging').getLogger(__name__)
logger = logging.getLogger(__name__)
class EnhancedScanService:
"""增强扫描服务 - 包含ARP、DNS解析、MAC厂商识别"""
"""增强扫描服务 - 包含ARP、DNS解析、MAC厂商识别、NetBIOS主机名发现"""
# MAC OUI 厂商映射(常用厂商
OUI_MAP = {
'00:1C:14': 'VMware',
'00:0C:29': 'VMware',
'00:50:56': 'VMware',
'00:15:5D': 'Microsoft',
'08:00:27': 'VirtualBox',
'52:54:00': 'QEMU/KVM',
'00:1A:2A': 'Dell',
'00:21:9B': 'Dell',
'00:26:B9': 'Dell',
'00:1B:21': 'HP',
'00:23:7D': 'HP',
'00:1E:0B': 'Cisco',
'00:21:55': 'Cisco',
'00:23:04': 'Cisco',
'00:24:14': 'Cisco',
'00:1F:CA': 'Cisco',
'00:19:BB': 'Huawei',
'00:25:9E': 'Huawei',
'00:E0:FC': 'Huawei',
'AC:85:3D': 'Huawei',
'28:6E:D4': 'Huawei',
'00:16:6F': 'H3C',
'00:23:89': 'H3C',
'00:0F:E2': 'Intel',
'00:1B:77': 'Intel',
'00:21:6A': 'Intel',
'1C:6F:65': 'Intel',
'00:19:D2': 'Realtek',
'00:24:1D': 'Realtek',
'00:E0:4C': 'Realtek',
'F4:6D:04': 'Apple',
'E0:F8:47': 'Apple',
'C8:BC:C8': 'Apple',
'04:15:52': 'Apple',
'28:CF:E9': 'Apple',
'04:7D:7B': 'Xiaomi',
'18:59:36': 'Xiaomi',
'34:CE:00': 'Xiaomi',
'64:09:80': 'TP-Link',
'E8:94:F6': 'TP-Link',
'50:FA:84': 'TP-Link',
'00:24:01': 'TP-Link',
'30:FC:68': 'NETGEAR',
'00:09:5B': 'NETGEAR',
'00:FF:FF': 'Broadcast',
'FF:FF:FF': 'Broadcast',
}
# 初始化 MAC 厂商查询(单例,共享 OUI 数据库
_mac_lookup = None
@classmethod
def _get_mac_lookup(cls):
"""获取或初始化 MacLookup 实例"""
if cls._mac_lookup is None and MAC_LOOKUP_AVAILABLE:
cls._mac_lookup = MacLookup()
return cls._mac_lookup
@staticmethod
def get_mac_vendor(mac_address: str) -> Optional[str]:
"""根据MAC地址OUI识别厂商"""
"""根据 MAC 地址 OUI 识别厂商"""
if not mac_address:
return None
# 标准化MAC格式为 AA:BB:CC:DD:EE:FF
# 标准化 MAC 格式为 AA:BB:CC:DD:EE:FF
mac = mac_address.upper().replace('-', ':')
if len(mac) < 8:
return None
# 提取前3字节 OUI
# 使用 mac_vendor_lookup 库(IEEE OUI 官方数据库)
if MAC_LOOKUP_AVAILABLE:
lookup = EnhancedScanService._get_mac_lookup()
if lookup is not None:
try:
# 提取 OUI(前 3 字节)
oui = mac[:8] # "AA:BB:CC"
vendor = lookup.lookup(oui)
if vendor:
return vendor
except (VendorNotFoundError, AttributeError, TypeError):
pass
except Exception:
pass
# 回退:硬编码常用厂商映射
return EnhancedScanService._legacy_oui_lookup(mac)
# 硬编码回退映射(mac_vendor_lookup 不可用时使用)
_LEGACY_OUI_MAP = {
'00:1C:14': 'VMware', '00:0C:29': 'VMware', '00:50:56': 'VMware',
'00:15:5D': 'Microsoft', '08:00:27': 'VirtualBox', '52:54:00': 'QEMU/KVM',
'00:1A:2A': 'Dell', '00:21:9B': 'Dell', '00:26:B9': 'Dell',
'00:1B:21': 'HP', '00:23:7D': 'HP',
'00:1E:0B': 'Cisco', '00:21:55': 'Cisco', '00:23:04': 'Cisco',
'00:24:14': 'Cisco', '00:1F:CA': 'Cisco',
'00:19:BB': 'Huawei', '00:25:9E': 'Huawei', '00:E0:FC': 'Huawei',
'AC:85:3D': 'Huawei', '28:6E:D4': 'Huawei',
'00:16:6F': 'H3C', '00:23:89': 'H3C',
'00:0F:E2': 'Intel', '00:1B:77': 'Intel', '00:21:6A': 'Intel',
'1C:6F:65': 'Intel',
'00:19:D2': 'Realtek', '00:24:1D': 'Realtek', '00:E0:4C': 'Realtek',
'F4:6D:04': 'Apple', 'E0:F8:47': 'Apple', 'C8:BC:C8': 'Apple',
'04:15:52': 'Apple', '28:CF:E9': 'Apple',
'04:7D:7B': 'Xiaomi', '18:59:36': 'Xiaomi', '34:CE:00': 'Xiaomi',
'64:09:80': 'TP-Link', 'E8:94:F6': 'TP-Link', '50:FA:84': 'TP-Link',
'00:24:01': 'TP-Link',
'30:FC:68': 'NETGEAR', '00:09:5B': 'NETGEAR',
'00:FF:FF': 'Broadcast', 'FF:FF:FF': 'Broadcast',
}
@staticmethod
def _legacy_oui_lookup(mac: str) -> Optional[str]:
"""硬编码 OUI 回退查询"""
oui = mac[:8]
# 精确匹配
if oui in EnhancedScanService.OUI_MAP:
return EnhancedScanService.OUI_MAP[oui]
# 尝试模糊匹配(前2字节)
if oui in EnhancedScanService._LEGACY_OUI_MAP:
return EnhancedScanService._LEGACY_OUI_MAP[oui]
# 模糊匹配(前 2 字节)
oui_short = mac[:5]
for key, vendor in EnhancedScanService.OUI_MAP.items():
for key, vendor in EnhancedScanService._LEGACY_OUI_MAP.items():
if key.startswith(oui_short):
return vendor
return None
@staticmethod
def reverse_dns_lookup(ip_address: str, timeout: int = 2) -> Optional[str]:
"""反向DNS解析获取主机名"""
"""反向 DNS 解析获取主机名"""
try:
socket.setdefaulttimeout(timeout)
hostname, _, _ = socket.gethostbyaddr(ip_address)
@@ -103,6 +109,64 @@ class EnhancedScanService:
except (socket.herror, socket.timeout, socket.gaierror):
return None
@staticmethod
def netbios_lookup(ip_address: str, timeout: int = 3) -> Optional[str]:
"""
通过 NetBIOS 广播协议获取主机名(Windows / SMB 设备)
使用 nmblookup -A <ip> 查询
"""
try:
result = subprocess.run(
['nmblookup', '-A', ip_address],
capture_output=True, text=True, timeout=timeout
)
if result.returncode == 0:
# nmblookup 输出格式示例:
# querying host 192.168.1.100 on 192.168.1.255
#
# WORKGROUP<00> - <ACTIVE> B
# WORKGROUP<00> - <GROUP> B
# WORKGROUP<1D> - <GROUP> B
# SERVER01<20> - <ACTIVE> B
# 主机名通常是 <20> 服务的第一列
for line in result.stdout.strip().split('\n'):
line = line.strip()
if not line or line.startswith('querying'):
continue
# 匹配 "HOSTNAME<20> - <ACTIVE> B" 格式
match = re.match(
r'^(\S+?)<20>\s+-\s+<ACTIVE>',
line,
re.IGNORECASE
)
if match:
return match.group(1).upper()
except Exception:
pass
return None
@staticmethod
def snmp_sysname_lookup(device_id: int, db: Session) -> Optional[str]:
"""
通过 SNMP sysName 获取主机名(网络设备 / 服务器)
仅在扫描时已配置 SNMP 凭据的设备上调用
"""
from app.models.snmp import NetworkDevice
from app.services.snmp_service import SNMPService
try:
device = db.query(NetworkDevice).filter(
NetworkDevice.id == device_id
).first()
if device and device.snmp_credential:
success, info = SNMPService.test_connection(
device, device.snmp_credential
)
if success:
return info.get('sys_name')
except Exception:
pass
return None
@staticmethod
def arp_scan_single(ip_address: str, timeout: int = 2) -> Optional[str]:
"""
@@ -116,7 +180,6 @@ class EnhancedScanService:
capture_output=True, text=True, timeout=timeout + 2
)
if result.returncode == 0:
# arp-scan 输出格式: IP\tMAC\tVendor
for line in result.stdout.strip().split('\n'):
line = line.strip()
if not line:
@@ -155,7 +218,6 @@ class EnhancedScanService:
timeout=2
)
if result.returncode == 0:
# 匹配 MAC 地址格式
mac_pattern = r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})'
match = re.search(mac_pattern, result.stdout)
if match:
@@ -170,9 +232,20 @@ class EnhancedScanService:
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2) -> Dict[str, Any]:
"""
综合扫描单个IP
厂商识别:
- 主要: mac_vendor_lookup 库(IEEE OUI 官方数据库,~17万条)
- 回退: 硬编码 OUI_MAP
主机名发现(多源,按优先级):
1. 反向 DNS 解析(PTR
2. NetBIOS 广播(Windows / SMB 设备)
3. SNMP sysName(网络设备 / 服务器,需配置 SNMP)
返回: {
'ip_address': str,
'status': 'online'|'offline',
@@ -203,14 +276,21 @@ class EnhancedScanService:
mac = EnhancedScanService.arp_scan_single(ip_address, timeout)
if mac:
result['mac_address'] = mac
# 厂商识别:优先 mac_vendor_lookup 库,失败后回退硬编码映射
result['vendor'] = EnhancedScanService.get_mac_vendor(mac)
# 如果获取到MAC,说明设备实际上在线
result['status'] = 'online'
result['success'] = True
# 3. 反向 DNS 解析
if enable_dns and result.get('success'):
hostname = EnhancedScanService.reverse_dns_lookup(ip_address, timeout)
# 3. 主机名发现(多源,有任一在线即执行)
if result.get('success'):
hostname = None
# 3a. 反向 DNS
if enable_dns:
hostname = EnhancedScanService.reverse_dns_lookup(ip_address, timeout)
# 3b. NetBIOSWindows / SMB
if not hostname and enable_netbios:
hostname = EnhancedScanService.netbios_lookup(ip_address, timeout=3)
if hostname:
result['hostname'] = hostname
@@ -249,6 +329,7 @@ class EnhancedScanService:
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2,
max_concurrent: int = 50) -> List[Dict[str, Any]]:
"""
@@ -260,7 +341,6 @@ class EnhancedScanService:
if not ips:
return []
# 并发执行;并发数随网段大小自适应
worker_count = min(max_concurrent, len(ips))
results: List[Dict[str, Any]] = []
with ThreadPoolExecutor(max_workers=worker_count) as executor:
@@ -271,6 +351,7 @@ class EnhancedScanService:
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
): ip
for ip in ips
@@ -303,12 +384,10 @@ class EnhancedScanService:
# 如果IP不存在,尝试自动创建
if not db_ip:
# 从ip_address推断network_id
network = db.query(Network).filter(
Network.cidr.op('@>')((ip_address + '/32').encode('utf-8'))
).first()
if not network:
# 尝试精确匹配 network.cidr
for net in db.query(Network).all():
try:
if ip_address in ipaddress.ip_network(net.cidr, strict=False):
@@ -342,11 +421,15 @@ class EnhancedScanService:
if hostname:
db_ip.hostname = hostname
# 更新厂商信息
# 更新厂商信息(即使没有 MAC,也尝试用现有 MAC 补全)
vendor = scan_result.get('vendor')
if vendor:
db_ip.vendor = vendor
# 如果没有拿到 vendor 但有 MAC,用现有 MAC 补全厂商
if not db_ip.vendor and db_ip.mac_address:
db_ip.vendor = EnhancedScanService.get_mac_vendor(db_ip.mac_address)
# 更新最后发现时间
if scan_result.get('success'):
db_ip.last_seen = datetime.utcnow()
@@ -360,6 +443,7 @@ class EnhancedScanService:
mac_found_count = 0
hostname_found_count = 0
created_count = 0
vendor_found_count = 0
for result in scan_results:
ip = EnhancedScanService.update_ip_from_scan_result(db, result)
@@ -370,6 +454,8 @@ class EnhancedScanService:
mac_found_count += 1
if result.get('hostname'):
hostname_found_count += 1
if result.get('vendor'):
vendor_found_count += 1
db.commit()
@@ -377,5 +463,6 @@ class EnhancedScanService:
'online_count': online_count,
'mac_found_count': mac_found_count,
'hostname_found_count': hostname_found_count,
'vendor_found_count': vendor_found_count,
'total_scanned': len(scan_results)
}
+4 -2
View File
@@ -221,18 +221,20 @@ class SNMPService:
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
'device_id': device.id,
'last_seen': now.isoformat(),
'discovered_at': now.isoformat()
}
arp_entries.append(entry)
# 保存到数据库
now = datetime.utcnow()
for entry_data in arp_entries:
# 查找是否已存在
existing = db.query(ARPEntry).filter(
+10 -3
View File
@@ -21,6 +21,7 @@ def register_tasks(celery_app):
enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True,
enable_netbios: bool = True,
timeout: int = 2):
"""
Celery 异步任务:扫描指定网段
@@ -54,6 +55,7 @@ def register_tasks(celery_app):
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns,
enable_netbios=enable_netbios,
timeout=timeout
)
@@ -120,7 +122,8 @@ def register_tasks(celery_app):
@celery_app.task(bind=True)
def full_network_scan(self, enable_ping: bool = True,
enable_arp: bool = True,
enable_dns: bool = True):
enable_dns: bool = True,
enable_netbios: bool = True):
"""
Celery 定时任务:扫描所有网段
"""
@@ -144,7 +147,8 @@ def register_tasks(celery_app):
network.cidr,
enable_ping=enable_ping,
enable_arp=enable_arp,
enable_dns=enable_dns
enable_dns=enable_dns,
enable_netbios=enable_netbios
)
stats = EnhancedScanService.bulk_update_ips_from_scan(db, results)
@@ -191,7 +195,10 @@ def register_tasks(celery_app):
results = []
for ip in ip_addresses:
result = EnhancedScanService.comprehensive_scan(ip, enable_ping=True, enable_arp=True, enable_dns=False)
result = EnhancedScanService.comprehensive_scan(
ip, enable_ping=True, enable_arp=True, enable_dns=False,
enable_netbios=True
)
EnhancedScanService.update_ip_from_scan_result(db, result)
results.append(result)
+11 -2
View File
@@ -63,20 +63,29 @@ export const scanApi = {
return api.post(`/scan/ping/${id}`)
},
// 综合扫描网段
// options: { enable_ping, enable_arp, enable_dns, timeout }
// options: { enable_ping, enable_arp, enable_dns, enable_netbios, timeout }
// 单次调用覆盖全局 30s 超时,因为 /24 网段扫描要数分钟
comprehensiveScan(id, options = {}) {
const {
enable_ping = true,
enable_arp = true,
enable_dns = true,
enable_netbios = true,
timeout = 2
} = options
return api.post(`/scan/comprehensive/${id}`, {}, {
params: { enable_ping, enable_arp, enable_dns, timeout },
params: { enable_ping, enable_arp, enable_dns, enable_netbios, timeout },
timeout: 600000 // 10 分钟,足以应付 /16 级别网段
})
},
// 批量补全厂商信息
batchFillVendor(params) {
return api.post('/scan/batch/vendor', {}, { params })
},
// 批量发现主机名
batchDiscoverHostname(params) {
return api.post('/scan/batch/hostname', {}, { params })
},
// 单个IP扫描
scanIP(ipAddress) {
return api.post(`/scan/comprehensive/ip/${ipAddress}`)
+23
View File
@@ -8,6 +8,10 @@
<el-icon><Search /></el-icon>
批量扫描
</el-button>
<el-button type="warning" @click="batchFillVendor" :loading="vendorLoading">
<el-icon><Refresh /></el-icon>
补全厂商
</el-button>
</div>
</template>
@@ -176,6 +180,7 @@ const route = useRoute()
const loading = ref(false)
const submitting = ref(false)
const scanning = ref(false)
const vendorLoading = ref(false)
const dialogVisible = ref(false)
const tableData = ref([])
@@ -278,6 +283,24 @@ const startBatchScan = async () => {
}
}
const batchFillVendor = async () => {
if (!filters.networkId) {
ElMessage.warning('请先选择网段')
return
}
vendorLoading.value = true
try {
const res = await scanApi.batchFillVendor({ network_id: filters.networkId })
ElMessage.success(res.message)
loadIPs()
} catch (error) {
ElMessage.error('补全厂商失败')
console.error('补全厂商错误:', error)
} finally {
vendorLoading.value = false
}
}
const editIP = (row) => {
ipForm.id = row.id
ipForm.ip_address = row.ip_address