Your Name
5369d8d6e5
detect_ip_conflicts 误用 db.func.count/db.func.distinct，而 func 是 SQLAlchemy 模块属性而非 Session 方法，触发 AttributeError 使 /alerts/detect/run 返回500（前端弹'请求失败'）。 改为 from sqlalchemy import func 后使用 func.count/func.distinct
498 lines
17 KiB
Python
498 lines
17 KiB
Python
from typing import List, Dict, Any, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
import json
|
|
|
|
from app.models.alert import (
|
|
Alert, AlertRule, WhitelistedMAC,
|
|
AlertType, AlertSeverity, AlertStatus
|
|
)
|
|
from app.models.network import Network, IPAddress
|
|
from app.models.snmp import NetworkDevice, ARPEntry
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AlertService:
|
|
"""告警检测与管理服务"""
|
|
|
|
@staticmethod
|
|
def create_alert(db: Session,
|
|
alert_type: AlertType,
|
|
title: str,
|
|
message: str = "",
|
|
severity: AlertSeverity = AlertSeverity.WARNING,
|
|
network_id: Optional[int] = None,
|
|
ip_address_id: Optional[int] = None,
|
|
device_id: Optional[int] = None,
|
|
ip_address_str: Optional[str] = None,
|
|
mac_address: Optional[str] = None,
|
|
conflicting_mac: Optional[str] = None,
|
|
usage_percent: Optional[float] = None) -> Alert:
|
|
"""创建告警"""
|
|
# 检查是否已有相同的活跃告警，避免重复告警
|
|
existing = db.query(Alert).filter(
|
|
Alert.alert_type == alert_type,
|
|
Alert.status == AlertStatus.ACTIVE
|
|
)
|
|
|
|
if ip_address_str:
|
|
existing = existing.filter(Alert.ip_address_str == ip_address_str)
|
|
if network_id:
|
|
existing = existing.filter(Alert.network_id == network_id)
|
|
if device_id:
|
|
existing = existing.filter(Alert.device_id == device_id)
|
|
|
|
existing = existing.first()
|
|
|
|
if existing:
|
|
logger.debug(f"已有相同的活跃告警: {alert_type} - {ip_address_str or network_id or device_id}")
|
|
return existing
|
|
|
|
alert = Alert(
|
|
alert_type=alert_type,
|
|
title=title,
|
|
message=message,
|
|
severity=severity,
|
|
network_id=network_id,
|
|
ip_address_id=ip_address_id,
|
|
device_id=device_id,
|
|
ip_address_str=ip_address_str,
|
|
mac_address=mac_address,
|
|
conflicting_mac=conflicting_mac,
|
|
usage_percent=usage_percent
|
|
)
|
|
|
|
db.add(alert)
|
|
db.commit()
|
|
db.refresh(alert)
|
|
|
|
logger.info(f"创建告警: {alert_type} - {title}")
|
|
return alert
|
|
|
|
@staticmethod
|
|
def detect_ip_conflicts(db: Session) -> List[Dict[str, Any]]:
|
|
"""检测 IP 冲突"""
|
|
conflicts = []
|
|
|
|
# 检查 ARP 表中同一个 IP 对应多个不同 MAC 的情况
|
|
subquery = db.query(
|
|
ARPEntry.ip_address
|
|
).group_by(ARPEntry.ip_address).having(
|
|
func.count(func.distinct(ARPEntry.mac_address)) > 1
|
|
).subquery()
|
|
|
|
conflict_entries = db.query(ARPEntry).filter(
|
|
ARPEntry.ip_address.in_(subquery)
|
|
).order_by(ARPEntry.ip_address).all()
|
|
|
|
# 按 IP 分组
|
|
ip_macs = {}
|
|
for entry in conflict_entries:
|
|
if entry.ip_address not in ip_macs:
|
|
ip_macs[entry.ip_address] = set()
|
|
ip_macs[entry.ip_address].add(entry.mac_address)
|
|
|
|
# 为每个冲突创建告警
|
|
for ip, macs in ip_macs.items():
|
|
if len(macs) > 1:
|
|
mac_list = list(macs)
|
|
conflict_info = {
|
|
'ip_address': ip,
|
|
'mac_addresses': mac_list
|
|
}
|
|
conflicts.append(conflict_info)
|
|
|
|
# 创建告警
|
|
AlertService.create_alert(
|
|
db=db,
|
|
alert_type=AlertType.IP_CONFLICT,
|
|
title=f"IP 冲突检测: {ip}",
|
|
message=f"IP 地址 {ip} 检测到多个 MAC 地址: {', '.join(mac_list)}",
|
|
severity=AlertSeverity.ERROR,
|
|
ip_address_str=ip,
|
|
mac_address=mac_list[0],
|
|
conflicting_mac=mac_list[1] if len(mac_list) > 1 else None
|
|
)
|
|
|
|
return conflicts
|
|
|
|
@staticmethod
|
|
def detect_unauthorized_access(db: Session) -> List[Dict[str, Any]]:
|
|
"""检测未授权接入（新 MAC 地址不在白名单中）"""
|
|
unauthorized = []
|
|
|
|
# 获取白名单 MAC
|
|
whitelist = set()
|
|
whitelist_entries = db.query(WhitelistedMAC).filter(
|
|
WhitelistedMAC.is_active == True
|
|
).all()
|
|
for entry in whitelist_entries:
|
|
whitelist.add(entry.mac_address.upper())
|
|
|
|
# 获取所有 ARP 表中的 MAC 地址
|
|
arp_entries = db.query(ARPEntry).all()
|
|
|
|
for entry in arp_entries:
|
|
if not entry.mac_address:
|
|
continue
|
|
|
|
mac = entry.mac_address.upper()
|
|
if mac and mac not in whitelist:
|
|
# 检查是否是最近发现的（1小时内）
|
|
is_recent = entry.discovered_at >= datetime.utcnow() - timedelta(hours=1)
|
|
|
|
if is_recent:
|
|
unauth_info = {
|
|
'ip_address': entry.ip_address,
|
|
'mac_address': mac,
|
|
'discovered_at': entry.discovered_at
|
|
}
|
|
unauthorized.append(unauth_info)
|
|
|
|
# 创建告警
|
|
AlertService.create_alert(
|
|
db=db,
|
|
alert_type=AlertType.UNAUTHORIZED_ACCESS,
|
|
title=f"未授权接入: {mac}",
|
|
message=f"检测到未授权设备接入: IP {entry.ip_address}, MAC {mac}",
|
|
severity=AlertSeverity.WARNING,
|
|
ip_address_str=entry.ip_address,
|
|
mac_address=mac
|
|
)
|
|
|
|
return unauthorized
|
|
|
|
@staticmethod
|
|
def detect_subnet_exhaustion(db: Session) -> List[Dict[str, Any]]:
|
|
"""检测网段耗尽"""
|
|
exhausted = []
|
|
|
|
# 获取所有网段的统计信息
|
|
networks = db.query(Network).all()
|
|
|
|
for network in networks:
|
|
if network.total_ips == 0:
|
|
continue
|
|
|
|
# 计算使用中的 IP 数量（在线或已分配）
|
|
used_count = db.query(IPAddress).filter(
|
|
IPAddress.network_id == network.id
|
|
).filter(
|
|
(IPAddress.mac_address.isnot(None)) |
|
|
(IPAddress.hostname.isnot(None))
|
|
).count()
|
|
|
|
usage_percent = (used_count / network.total_ips) * 100
|
|
|
|
# 默认阈值 90%
|
|
threshold = 90.0
|
|
|
|
if usage_percent >= threshold:
|
|
exhaustion_info = {
|
|
'network_id': network.id,
|
|
'cidr': network.cidr,
|
|
'name': network.name,
|
|
'total_ips': network.total_ips,
|
|
'used_ips': used_count,
|
|
'usage_percent': round(usage_percent, 2)
|
|
}
|
|
exhausted.append(exhaustion_info)
|
|
|
|
# 创建告警
|
|
AlertService.create_alert(
|
|
db=db,
|
|
alert_type=AlertType.SUBNET_FULL,
|
|
title=f"网段使用率告警: {network.cidr}",
|
|
message=f"网段 {network.cidr} ({network.name}) 使用率达到 {round(usage_percent, 2)}%，总IP {network.total_ips}，已使用 {used_count}",
|
|
severity=AlertSeverity.WARNING if usage_percent < 95 else AlertSeverity.CRITICAL,
|
|
network_id=network.id,
|
|
usage_percent=usage_percent
|
|
)
|
|
|
|
return exhausted
|
|
|
|
@staticmethod
|
|
def detect_new_devices(db: Session) -> List[Dict[str, Any]]:
|
|
"""检测新发现的设备"""
|
|
new_devices = []
|
|
|
|
# 查找最近 1 小时内发现的新 MAC 地址
|
|
one_hour_ago = datetime.utcnow() - timedelta(hours=1)
|
|
|
|
recent_entries = db.query(ARPEntry).filter(
|
|
ARPEntry.discovered_at >= one_hour_ago
|
|
).all()
|
|
|
|
for entry in recent_entries:
|
|
# 检查此 MAC 是否是新出现的
|
|
older_entries = db.query(ARPEntry).filter(
|
|
ARPEntry.mac_address == entry.mac_address,
|
|
ARPEntry.discovered_at < one_hour_ago
|
|
).count()
|
|
|
|
if older_entries == 0 and entry.mac_address:
|
|
device_info = {
|
|
'ip_address': entry.ip_address,
|
|
'mac_address': entry.mac_address,
|
|
'discovered_at': entry.discovered_at
|
|
}
|
|
new_devices.append(device_info)
|
|
|
|
# 创建告警
|
|
AlertService.create_alert(
|
|
db=db,
|
|
alert_type=AlertType.NEW_DEVICE_DETECTED,
|
|
title=f"新设备发现: {entry.ip_address}",
|
|
message=f"检测到新设备: IP {entry.ip_address}, MAC {entry.mac_address}",
|
|
severity=AlertSeverity.INFO,
|
|
ip_address_str=entry.ip_address,
|
|
mac_address=entry.mac_address
|
|
)
|
|
|
|
return new_devices
|
|
|
|
@staticmethod
|
|
def detect_device_offline(db: Session) -> List[Dict[str, Any]]:
|
|
"""检测设备离线"""
|
|
offline_devices = []
|
|
|
|
# 检查网络设备是否长时间没有成功轮询
|
|
threshold_time = datetime.utcnow() - timedelta(hours=2)
|
|
|
|
offline_network_devices = db.query(NetworkDevice).filter(
|
|
NetworkDevice.is_active == True,
|
|
(NetworkDevice.last_successful_poll < threshold_time) |
|
|
(NetworkDevice.last_successful_poll == None)
|
|
).all()
|
|
|
|
for device in offline_network_devices:
|
|
offline_info = {
|
|
'device_id': device.id,
|
|
'device_name': device.name,
|
|
'ip_address': device.ip_address,
|
|
'last_polled_at': device.last_polled_at,
|
|
'last_successful_poll': device.last_successful_poll
|
|
}
|
|
offline_devices.append(offline_info)
|
|
|
|
# 创建告警
|
|
AlertService.create_alert(
|
|
db=db,
|
|
alert_type=AlertType.DEVICE_OFFLINE,
|
|
title=f"设备离线: {device.name}",
|
|
message=f"网络设备 {device.name} ({device.ip_address}) 已超过 2 小时未成功轮询",
|
|
severity=AlertSeverity.ERROR,
|
|
device_id=device.id
|
|
)
|
|
|
|
return offline_devices
|
|
|
|
@staticmethod
|
|
def run_all_detections(db: Session) -> Dict[str, Any]:
|
|
"""运行所有检测"""
|
|
results = {}
|
|
|
|
logger.info("开始运行告警检测...")
|
|
|
|
# 检测 IP 冲突
|
|
conflicts = AlertService.detect_ip_conflicts(db)
|
|
results['ip_conflicts'] = {
|
|
'count': len(conflicts),
|
|
'items': conflicts
|
|
}
|
|
|
|
# 检测未授权接入
|
|
unauthorized = AlertService.detect_unauthorized_access(db)
|
|
results['unauthorized_access'] = {
|
|
'count': len(unauthorized),
|
|
'items': unauthorized
|
|
}
|
|
|
|
# 检测网段耗尽
|
|
subnet_exhaustion = AlertService.detect_subnet_exhaustion(db)
|
|
results['subnet_exhaustion'] = {
|
|
'count': len(subnet_exhaustion),
|
|
'items': subnet_exhaustion
|
|
}
|
|
|
|
# 检测新设备
|
|
new_devices = AlertService.detect_new_devices(db)
|
|
results['new_devices'] = {
|
|
'count': len(new_devices),
|
|
'items': new_devices
|
|
}
|
|
|
|
# 检测设备离线
|
|
offline_devices = AlertService.detect_device_offline(db)
|
|
results['offline_devices'] = {
|
|
'count': len(offline_devices),
|
|
'items': offline_devices
|
|
}
|
|
|
|
total_alerts = sum(v['count'] for v in results.values())
|
|
logger.info(f"告警检测完成，共发现 {total_alerts} 个问题")
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def get_alerts(db: Session,
|
|
status: Optional[AlertStatus] = None,
|
|
severity: Optional[AlertSeverity] = None,
|
|
alert_type: Optional[AlertType] = None,
|
|
skip: int = 0,
|
|
limit: int = 100) -> tuple[int, List[Alert]]:
|
|
"""获取告警列表"""
|
|
query = db.query(Alert)
|
|
|
|
if status:
|
|
query = query.filter(Alert.status == status)
|
|
if severity:
|
|
query = query.filter(Alert.severity == severity)
|
|
if alert_type:
|
|
query = query.filter(Alert.alert_type == alert_type)
|
|
|
|
total = query.count()
|
|
items = query.order_by(Alert.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
return total, items
|
|
|
|
@staticmethod
|
|
def acknowledge_alert(db: Session, alert_id: int, acknowledged_by: str = "system") -> Optional[Alert]:
|
|
"""确认告警"""
|
|
alert = db.query(Alert).filter(Alert.id == alert_id).first()
|
|
if not alert:
|
|
return None
|
|
|
|
alert.status = AlertStatus.ACKNOWLEDGED
|
|
alert.acknowledged_by = acknowledged_by
|
|
alert.acknowledged_at = datetime.utcnow()
|
|
db.commit()
|
|
db.refresh(alert)
|
|
|
|
return alert
|
|
|
|
@staticmethod
|
|
def resolve_alert(db: Session, alert_id: int, resolved_by: str = "system", notes: str = "") -> Optional[Alert]:
|
|
"""解决告警"""
|
|
alert = db.query(Alert).filter(Alert.id == alert_id).first()
|
|
if not alert:
|
|
return None
|
|
|
|
alert.status = AlertStatus.RESOLVED
|
|
alert.resolved_by = resolved_by
|
|
alert.resolved_at = datetime.utcnow()
|
|
alert.resolution_notes = notes
|
|
db.commit()
|
|
db.refresh(alert)
|
|
|
|
return alert
|
|
|
|
@staticmethod
|
|
def ignore_alert(db: Session, alert_id: int) -> Optional[Alert]:
|
|
"""忽略告警"""
|
|
alert = db.query(Alert).filter(Alert.id == alert_id).first()
|
|
if not alert:
|
|
return None
|
|
|
|
alert.status = AlertStatus.IGNORED
|
|
db.commit()
|
|
db.refresh(alert)
|
|
|
|
return alert
|
|
|
|
@staticmethod
|
|
def get_statistics(db: Session) -> Dict[str, Any]:
|
|
"""获取告警统计"""
|
|
active_count = db.query(Alert).filter(Alert.status == AlertStatus.ACTIVE).count()
|
|
acknowledged_count = db.query(Alert).filter(Alert.status == AlertStatus.ACKNOWLEDGED).count()
|
|
resolved_count = db.query(Alert).filter(Alert.status == AlertStatus.RESOLVED).count()
|
|
|
|
# 按严重级别统计
|
|
critical_count = db.query(Alert).filter(
|
|
Alert.status == AlertStatus.ACTIVE,
|
|
Alert.severity == AlertSeverity.CRITICAL
|
|
).count()
|
|
error_count = db.query(Alert).filter(
|
|
Alert.status == AlertStatus.ACTIVE,
|
|
Alert.severity == AlertSeverity.ERROR
|
|
).count()
|
|
warning_count = db.query(Alert).filter(
|
|
Alert.status == AlertStatus.ACTIVE,
|
|
Alert.severity == AlertSeverity.WARNING
|
|
).count()
|
|
info_count = db.query(Alert).filter(
|
|
Alert.status == AlertStatus.ACTIVE,
|
|
Alert.severity == AlertSeverity.INFO
|
|
).count()
|
|
|
|
# 按类型统计
|
|
type_stats = {}
|
|
for alert_type in AlertType:
|
|
count = db.query(Alert).filter(
|
|
Alert.status == AlertStatus.ACTIVE,
|
|
Alert.alert_type == alert_type
|
|
).count()
|
|
type_stats[alert_type.value] = count
|
|
|
|
return {
|
|
'by_status': {
|
|
'active': active_count,
|
|
'acknowledged': acknowledged_count,
|
|
'resolved': resolved_count
|
|
},
|
|
'by_severity': {
|
|
'critical': critical_count,
|
|
'error': error_count,
|
|
'warning': warning_count,
|
|
'info': info_count
|
|
},
|
|
'by_type': type_stats
|
|
}
|
|
|
|
@staticmethod
|
|
def add_mac_to_whitelist(db: Session, mac_address: str, description: str = "", owner: str = "") -> WhitelistedMAC:
|
|
"""添加 MAC 地址到白名单"""
|
|
existing = db.query(WhitelistedMAC).filter(WhitelistedMAC.mac_address == mac_address.upper()).first()
|
|
if existing:
|
|
existing.is_active = True
|
|
if description:
|
|
existing.description = description
|
|
if owner:
|
|
existing.owner = owner
|
|
db.commit()
|
|
db.refresh(existing)
|
|
return existing
|
|
|
|
whitelisted_mac = WhitelistedMAC(
|
|
mac_address=mac_address.upper(),
|
|
description=description,
|
|
owner=owner
|
|
)
|
|
db.add(whitelisted_mac)
|
|
db.commit()
|
|
db.refresh(whitelisted_mac)
|
|
|
|
return whitelisted_mac
|
|
|
|
@staticmethod
|
|
def get_mac_whitelist(db: Session, skip: int = 0, limit: int = 100) -> tuple[int, List[WhitelistedMAC]]:
|
|
"""获取 MAC 白名单"""
|
|
query = db.query(WhitelistedMAC).filter(WhitelistedMAC.is_active == True)
|
|
total = query.count()
|
|
items = query.order_by(WhitelistedMAC.id.desc()).offset(skip).limit(limit).all()
|
|
return total, items
|
|
|
|
@staticmethod
|
|
def remove_mac_from_whitelist(db: Session, mac_id: int) -> bool:
|
|
"""从白名单移除 MAC 地址"""
|
|
mac = db.query(WhitelistedMAC).filter(WhitelistedMAC.id == mac_id).first()
|
|
if not mac:
|
|
return False
|
|
|
|
mac.is_active = False
|
|
db.commit()
|
|
return True
|