Your Name
365ab73e7d
1. 新增 notification_config 模型,飞书配置存数据库(单行id=1) 2. feishu_service 改为从数据库读配置,支持热更新,Web保存即生效 3. alerts.py 新增接口: - GET /alerts/notify/feishu/config 读配置(secret只返回是否已设) - PUT /alerts/notify/feishu/config 保存配置+热更新 (status/test 同步改为走数据库配置) 4. 前端告警中心增加「飞书通知设置」弹窗: - 开关/Webhook/签名密钥/最低级别/类型白名单 - 保存配置、发送测试按钮 5. alert_service.create_alert 推送时按数据库配置判断 配置无需改文件/重启,前端界面直接保存即生效。
285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""
|
||
飞书机器人告警通知服务
|
||
========================
|
||
通过飞书自定义机器人 Webhook 推送告警通知，支持：
|
||
- 卡片消息（interactive）
|
||
- 可选签名校验（加签模式）
|
||
- 按告警级别 / 类型过滤
|
||
- 同步发送（带超时），失败自动记录日志
|
||
"""
|
||
from typing import Optional, Dict, Any, List
|
||
from datetime import datetime
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import time
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 告警级别 → 颜色（飞书卡片 title 颜色）
|
||
SEVERITY_COLORS = {
|
||
"critical": "red",
|
||
"error": "red",
|
||
"warning": "orange",
|
||
"info": "blue",
|
||
}
|
||
|
||
# 告警级别 → 中文
|
||
SEVERITY_TEXTS = {
|
||
"critical": "严重",
|
||
"error": "错误",
|
||
"warning": "警告",
|
||
"info": "信息",
|
||
}
|
||
|
||
# 告警类型 → 中文
|
||
ALERT_TYPE_TEXTS = {
|
||
"ip_conflict": "IP 地址冲突",
|
||
"unauthorized_access": "未授权设备接入",
|
||
"subnet_full": "网段容量不足",
|
||
"scan_failed": "扫描失败",
|
||
"device_offline": "设备离线",
|
||
"new_device_detected": "新设备发现",
|
||
"mac_changed": "MAC 地址变更",
|
||
}
|
||
|
||
# 告警级别优先级（用于 >= 最低级别过滤）
|
||
_SEVERITY_ORDER = {"info": 0, "warning": 1, "error": 2, "critical": 3}
|
||
|
||
|
||
def _apply_sign(timestamp: str, secret: str) -> str:
|
||
"""飞书机器人签名校验：HMAC-SHA256(secret, timestamp + "\n" + secret)，结果 base64"""
|
||
string_to_sign = f"{timestamp}\n{secret}"
|
||
hmac_code = hmac.new(
|
||
secret.encode("utf-8"),
|
||
string_to_sign.encode("utf-8"),
|
||
digestmod=hashlib.sha256,
|
||
).digest()
|
||
return base64.b64encode(hmac_code).decode("utf-8")
|
||
|
||
|
||
def build_card(alert_data: Dict[str, Any]) -> str:
|
||
"""把告警数据构造成飞书卡片消息体（interactive）。"""
|
||
severity = str(alert_data.get("severity") or "warning").lower()
|
||
alert_type = str(alert_data.get("alert_type") or "unknown").lower()
|
||
|
||
color = SEVERITY_COLORS.get(severity, "blue")
|
||
severity_text = SEVERITY_TEXTS.get(severity, severity)
|
||
alert_type_text = ALERT_TYPE_TEXTS.get(alert_type, alert_type)
|
||
|
||
title = alert_data.get("title") or "IPAM 告警"
|
||
message = alert_data.get("message") or ""
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 组装字段（element），过滤空字段
|
||
fields = []
|
||
def add_field(tag_text, content_text):
|
||
if content_text:
|
||
fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**{tag_text}：** {content_text}"}})
|
||
|
||
add_field("告警级别", f"<font color='{color}'><b>{severity_text}</b></font>")
|
||
add_field("告警类型", alert_type_text)
|
||
add_field("详细信息", message)
|
||
add_field("相关 IP", alert_data.get("ip_address_str"))
|
||
add_field("相关 MAC", alert_data.get("mac_address"))
|
||
add_field("冲突 MAC", alert_data.get("conflicting_mac"))
|
||
if alert_data.get("usage_percent") is not None:
|
||
add_field("网段使用率", f"{alert_data['usage_percent']}%")
|
||
add_field("告警时间", now)
|
||
|
||
card = {
|
||
"msg_type": "interactive",
|
||
"card": {
|
||
"config": {"wide_screen_mode": True},
|
||
"header": {
|
||
"title": {"tag": "plain_text", "content": f"⚠️ IPAM 告警 - {title}"},
|
||
"template": color,
|
||
},
|
||
"elements": [
|
||
{"tag": "div", "fields": fields},
|
||
{
|
||
"tag": "note",
|
||
"elements": [
|
||
{"tag": "plain_text", "content": f"本消息由 IPAM 地址管理系统自动发送 · {now}"}
|
||
],
|
||
},
|
||
],
|
||
},
|
||
}
|
||
return json.dumps(card, ensure_ascii=False)
|
||
|
||
|
||
class FeishuService:
|
||
"""飞书机器人告警推送服务"""
|
||
|
||
def __init__(
|
||
self,
|
||
webhook_url: str = "",
|
||
secret: str = "",
|
||
enabled: bool = True,
|
||
min_severity: str = "warning",
|
||
alert_types: Optional[List[str]] = None,
|
||
):
|
||
self.webhook_url = webhook_url
|
||
self.secret = secret
|
||
self.enabled = enabled
|
||
self.min_severity = min_severity
|
||
self.alert_types = alert_types or []
|
||
self._client = httpx.Client(timeout=8.0)
|
||
|
||
def is_configured(self) -> bool:
|
||
"""是否已配置 webhook 地址且开关打开"""
|
||
return bool(self.enabled and self.webhook_url)
|
||
|
||
def should_push(self, severity: str, alert_type: str = "") -> bool:
|
||
"""按最低级别 + 类型白名单判断是否应推送"""
|
||
if not self.is_configured():
|
||
return False
|
||
# 级别过滤
|
||
sev = (severity or "warning").lower()
|
||
if _SEVERITY_ORDER.get(sev, 1) < _SEVERITY_ORDER.get(self.min_severity, 1):
|
||
return False
|
||
# 类型白名单（空 = 全部推送）
|
||
if self.alert_types and alert_type and alert_type not in self.alert_types:
|
||
return False
|
||
return True
|
||
|
||
def _build_headers(self) -> Dict[str, str]:
|
||
"""构造签名头（若配置了 secret）"""
|
||
headers = {"Content-Type": "application/json; charset=utf-8"}
|
||
if self.secret:
|
||
timestamp = str(round(time.time()))
|
||
sign = _apply_sign(timestamp, self.secret)
|
||
headers["X-Lark-Sign"] = sign
|
||
headers["X-Lark-Timestamp"] = timestamp
|
||
return headers
|
||
|
||
def send_card(self, card_payload: str) -> bool:
|
||
"""发送飞书消息（card_payload 为 build_card 构造的 JSON 字符串）"""
|
||
if not self.is_configured():
|
||
logger.warning("飞书通知未启用或未配置 webhook，已跳过")
|
||
return False
|
||
try:
|
||
headers = self._build_headers()
|
||
resp = self._client.post(self.webhook_url, content=card_payload.encode("utf-8"), headers=headers)
|
||
try:
|
||
data = resp.json()
|
||
except Exception:
|
||
data = {"msg": resp.text}
|
||
# 飞书返回 code=0 表示成功
|
||
if resp.status_code == 200 and data.get("code") == 0:
|
||
logger.info(f"飞书通知发送成功: {data.get('msg', 'ok')}")
|
||
return True
|
||
logger.error(f"飞书通知发送失败: HTTP {resp.status_code} body={data}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"飞书通知发送异常: {str(e)}", exc_info=True)
|
||
return False
|
||
|
||
def send_alert(self, alert_data: Dict[str, Any]) -> bool:
|
||
"""发送一条告警通知（自动判断是否应推送）"""
|
||
severity = str(alert_data.get("severity") or "warning").lower()
|
||
alert_type = str(alert_data.get("alert_type") or "unknown").lower()
|
||
if not self.should_push(severity, alert_type):
|
||
logger.debug(f"告警 {severity}/{alert_type} 低于推送门槛或不在白名单，跳过")
|
||
return False
|
||
card = build_card(alert_data)
|
||
return self.send_card(card)
|
||
|
||
|
||
def close(self):
|
||
if self._client:
|
||
self._client.close()
|
||
|
||
|
||
# 全局单例
|
||
_service: Optional[FeishuService] = None
|
||
|
||
|
||
def _build_from_values(webhook_url="", secret="", enabled=False,
|
||
min_severity="warning", alert_types_str="") -> FeishuService:
|
||
"""用一组值构造 FeishuService（api 层用）"""
|
||
if _service is not None:
|
||
# 热更新：复用同一实例，更新内部配置
|
||
_service.webhook_url = webhook_url
|
||
_service.secret = secret
|
||
_service.enabled = bool(enabled)
|
||
_service.min_severity = min_severity or "warning"
|
||
_service.alert_types = [t.strip() for t in (alert_types_str or "").split(",") if t.strip()]
|
||
return _service
|
||
return FeishuService(
|
||
webhook_url=webhook_url,
|
||
secret=secret,
|
||
enabled=bool(enabled),
|
||
min_severity=min_severity or "warning",
|
||
alert_types=[t.strip() for t in (alert_types_str or "").split(",") if t.strip()],
|
||
)
|
||
|
||
|
||
def get_feishu_service(db=None) -> FeishuService:
|
||
"""获取（缓存的）飞书服务实例。
|
||
优先从数据库 notification_config 读取配置（Web 界面可热更新）；
|
||
若无数据库配置则回退到 .env / settings 默认值。
|
||
"""
|
||
global _service
|
||
|
||
# 优先从数据库加载（Web 配置优先）
|
||
try:
|
||
if db is not None:
|
||
from app.models.notification import NotificationConfig
|
||
row = db.query(NotificationConfig).filter(NotificationConfig.id == 1).first()
|
||
if row is not None:
|
||
return _build_from_values(
|
||
webhook_url=row.feishu_webhook_url or "",
|
||
secret=row.feishu_secret or "",
|
||
enabled=row.feishu_enabled or False,
|
||
min_severity=row.feishu_min_severity or "warning",
|
||
alert_types_str=row.feishu_alert_types or "",
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"从数据库读取飞书配置失败，回退 .env: {e}")
|
||
|
||
# 回退到 settings（.env）
|
||
if _service is None:
|
||
try:
|
||
from app.core.config import settings
|
||
_service = _build_from_values(
|
||
webhook_url=settings.FEISHU_WEBHOOK_URL,
|
||
secret=settings.FEISHU_SECRET,
|
||
enabled=settings.FEISHU_ENABLED,
|
||
min_severity=settings.FEISHU_MIN_SEVERITY or "warning",
|
||
alert_types_str=settings.FEISHU_ALERT_TYPES or "",
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"初始化飞书服务失败: {e}")
|
||
_service = FeishuService()
|
||
return _service
|
||
|
||
|
||
def reload_feishu_service(db) -> FeishuService:
|
||
"""强制从数据库重新加载飞书配置（Web 界面保存配置后调用）"""
|
||
global _service
|
||
from app.models.notification import NotificationConfig
|
||
row = db.query(NotificationConfig).filter(NotificationConfig.id == 1).first()
|
||
if row is None:
|
||
row = NotificationConfig.get_singleton(db)
|
||
svc = _build_from_values(
|
||
webhook_url=row.feishu_webhook_url or "",
|
||
secret=row.feishu_secret or "",
|
||
enabled=row.feishu_enabled or False,
|
||
min_severity=row.feishu_min_severity or "warning",
|
||
alert_types_str=row.feishu_alert_types or "",
|
||
)
|
||
global _service
|
||
_service = svc
|
||
return svc
|
||
|
||
|
||
def send_alert_notification(alert_data: Dict[str, Any], db=None) -> bool:
|
||
"""便捷函数：发送一条告警通知"""
|
||
return get_feishu_service(db).send_alert(alert_data)
|