Your Name
ccd7fbe2f0
- MAC厂商: 替换硬编码OUI_MAP为mac_vendor_lookup库(IEEE OUI官方数据库) - 主机名发现新增NetBIOS广播(nmblookup),支持Windows/SMB设备 - 综合扫描支持enable_netbios参数 - 新增NetBIOS主机名发现API端点 - 扫描时自动补全缺失厂商(有MAC无vendor时自动识别)
469 lines
17 KiB
Python
469 lines
17 KiB
Python
from typing import List, Dict, Any, Optional
|
||
from sqlalchemy.orm import Session
|
||
from datetime import datetime
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
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 = logging.getLogger(__name__)
|
||
|
||
|
||
class EnhancedScanService:
|
||
"""增强扫描服务 - 包含ARP、DNS解析、MAC厂商识别、NetBIOS主机名发现"""
|
||
|
||
# 初始化 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 识别厂商"""
|
||
if not mac_address:
|
||
return None
|
||
|
||
# 标准化 MAC 格式为 AA:BB:CC:DD:EE:FF
|
||
mac = mac_address.upper().replace('-', ':')
|
||
if len(mac) < 8:
|
||
return None
|
||
|
||
# 使用 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._LEGACY_OUI_MAP:
|
||
return EnhancedScanService._LEGACY_OUI_MAP[oui]
|
||
# 模糊匹配（前 2 字节）
|
||
oui_short = mac[:5]
|
||
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 解析获取主机名"""
|
||
try:
|
||
socket.setdefaulttimeout(timeout)
|
||
hostname, _, _ = socket.gethostbyaddr(ip_address)
|
||
return hostname
|
||
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]:
|
||
"""
|
||
对单个IP执行ARP扫描获取MAC地址
|
||
返回: MAC地址 或 None
|
||
"""
|
||
# 方法1: 使用 arp-scan 命令（比 scapy 更可靠，不依赖 raw socket）
|
||
try:
|
||
result = subprocess.run(
|
||
['arp-scan', '-I', 'auto', ip_address, '-q'],
|
||
capture_output=True, text=True, timeout=timeout + 2
|
||
)
|
||
if result.returncode == 0:
|
||
for line in result.stdout.strip().split('\n'):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
parts = line.split('\t')
|
||
if len(parts) >= 2:
|
||
mac = parts[1].strip()
|
||
if mac and mac != '<incomplete>':
|
||
return mac.upper()
|
||
except Exception:
|
||
pass
|
||
|
||
# 方法2: 尝试读取系统 ARP 缓存
|
||
return EnhancedScanService._get_arp_from_cache(ip_address)
|
||
|
||
@staticmethod
|
||
def _get_arp_from_cache(ip_address: str) -> Optional[str]:
|
||
"""从系统ARP缓存读取MAC地址"""
|
||
try:
|
||
# 读取 /proc/net/arp (Linux)
|
||
with open('/proc/net/arp', 'r') as f:
|
||
for line in f.readlines()[1:]: # 跳过表头
|
||
parts = line.split()
|
||
if len(parts) >= 4 and parts[0] == ip_address:
|
||
mac = parts[3].upper()
|
||
if mac != '00:00:00:00:00:00':
|
||
return mac
|
||
except Exception:
|
||
pass
|
||
|
||
# 尝试 arp -a 命令
|
||
try:
|
||
result = subprocess.run(
|
||
['arp', '-a', ip_address],
|
||
capture_output=True, text=True,
|
||
timeout=2
|
||
)
|
||
if result.returncode == 0:
|
||
mac_pattern = r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})'
|
||
match = re.search(mac_pattern, result.stdout)
|
||
if match:
|
||
return match.group().upper()
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
@staticmethod
|
||
def comprehensive_scan(ip_address: str,
|
||
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',
|
||
'mac_address': str|None,
|
||
'hostname': str|None,
|
||
'vendor': str|None,
|
||
'response_time': float|None
|
||
}
|
||
"""
|
||
result = {
|
||
'ip_address': ip_address,
|
||
'status': 'offline',
|
||
'mac_address': None,
|
||
'hostname': None,
|
||
'vendor': None,
|
||
'success': False
|
||
}
|
||
|
||
# 1. Ping 扫描
|
||
if enable_ping:
|
||
ping_result = EnhancedScanService.ping_single_ip(ip_address, timeout)
|
||
result['status'] = ping_result.get('status', 'offline')
|
||
result['success'] = ping_result.get('success', False)
|
||
result['response_time'] = ping_result.get('response_time')
|
||
|
||
# 2. ARP 扫描 (即使ping不通也尝试，有些设备禁ping)
|
||
if enable_arp:
|
||
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. 主机名发现（多源，有任一在线即执行）
|
||
if result.get('success'):
|
||
hostname = None
|
||
# 3a. 反向 DNS
|
||
if enable_dns:
|
||
hostname = EnhancedScanService.reverse_dns_lookup(ip_address, timeout)
|
||
# 3b. NetBIOS（Windows / SMB）
|
||
if not hostname and enable_netbios:
|
||
hostname = EnhancedScanService.netbios_lookup(ip_address, timeout=3)
|
||
if hostname:
|
||
result['hostname'] = hostname
|
||
|
||
return result
|
||
|
||
@staticmethod
|
||
def ping_single_ip(ip_address: str, timeout: int = 2) -> Dict[str, Any]:
|
||
"""Ping 单个IP"""
|
||
start_time = datetime.now()
|
||
try:
|
||
result = subprocess.run(
|
||
['ping', '-c', '1', '-W', str(timeout), ip_address],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
|
||
is_alive = result.returncode == 0
|
||
response_time = (datetime.now() - start_time).total_seconds() * 1000
|
||
|
||
return {
|
||
'ip_address': ip_address,
|
||
'status': 'online' if is_alive else 'offline',
|
||
'response_time': round(response_time, 2),
|
||
'success': is_alive
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
'ip_address': ip_address,
|
||
'status': 'error',
|
||
'error': str(e),
|
||
'success': False
|
||
}
|
||
|
||
@staticmethod
|
||
def scan_network(cidr: str,
|
||
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]]:
|
||
"""
|
||
扫描整个网段（并发）
|
||
"""
|
||
network = ipaddress.ip_network(cidr, strict=False)
|
||
ips = [str(ip) for ip in network.hosts()]
|
||
|
||
if not ips:
|
||
return []
|
||
|
||
worker_count = min(max_concurrent, len(ips))
|
||
results: List[Dict[str, Any]] = []
|
||
with ThreadPoolExecutor(max_workers=worker_count) as executor:
|
||
future_to_ip = {
|
||
executor.submit(
|
||
EnhancedScanService.comprehensive_scan,
|
||
ip,
|
||
enable_ping=enable_ping,
|
||
enable_arp=enable_arp,
|
||
enable_dns=enable_dns,
|
||
enable_netbios=enable_netbios,
|
||
timeout=timeout
|
||
): ip
|
||
for ip in ips
|
||
}
|
||
for future in as_completed(future_to_ip):
|
||
try:
|
||
results.append(future.result())
|
||
except Exception as e:
|
||
ip = future_to_ip[future]
|
||
results.append({
|
||
'ip_address': ip,
|
||
'status': 'offline',
|
||
'mac_address': None,
|
||
'hostname': None,
|
||
'vendor': None,
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
return results
|
||
|
||
@staticmethod
|
||
def update_ip_from_scan_result(db: Session, scan_result: Dict[str, Any]):
|
||
"""根据扫描结果更新IP信息"""
|
||
ip_address = scan_result.get('ip_address')
|
||
if not ip_address:
|
||
return
|
||
|
||
db_ip = db.query(IPAddress).filter(IPAddress.ip_address == ip_address).first()
|
||
|
||
# 如果IP不存在，尝试自动创建
|
||
if not db_ip:
|
||
network = db.query(Network).filter(
|
||
Network.cidr.op('@>')((ip_address + '/32').encode('utf-8'))
|
||
).first()
|
||
if not network:
|
||
for net in db.query(Network).all():
|
||
try:
|
||
if ip_address in ipaddress.ip_network(net.cidr, strict=False):
|
||
network = net
|
||
break
|
||
except Exception:
|
||
pass
|
||
if not network:
|
||
logger.warning(f"无法为扫描到的IP {ip_address} 找到所属网段，跳过创建")
|
||
return
|
||
|
||
db_ip = IPAddress(ip_address=ip_address, network_id=network.id)
|
||
db.add(db_ip)
|
||
db.commit()
|
||
db.refresh(db_ip)
|
||
|
||
# 更新状态
|
||
status = scan_result.get('status', 'offline')
|
||
if status == 'online':
|
||
db_ip.status = IPStatus.ONLINE
|
||
elif status == 'offline' and db_ip.status == IPStatus.ONLINE:
|
||
db_ip.status = IPStatus.OFFLINE
|
||
|
||
# 更新 MAC 地址
|
||
mac = scan_result.get('mac_address')
|
||
if mac:
|
||
db_ip.mac_address = mac
|
||
|
||
# 更新主机名
|
||
hostname = scan_result.get('hostname')
|
||
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()
|
||
|
||
return db_ip
|
||
|
||
@staticmethod
|
||
def bulk_update_ips_from_scan(db: Session, scan_results: List[Dict[str, Any]]):
|
||
"""批量更新IP信息"""
|
||
online_count = 0
|
||
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)
|
||
if ip:
|
||
if result.get('success'):
|
||
online_count += 1
|
||
if result.get('mac_address'):
|
||
mac_found_count += 1
|
||
if result.get('hostname'):
|
||
hostname_found_count += 1
|
||
if result.get('vendor'):
|
||
vendor_found_count += 1
|
||
|
||
db.commit()
|
||
|
||
return {
|
||
'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)
|
||
}
|