diff --git a/backend/.env.example b/backend/.env.example index 94b47d3..37d8e55 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,3 +17,16 @@ SECRET_KEY=your-super-secret-key-here-change-in-production PING_TIMEOUT=2 PING_RETRIES=2 SCAN_CONCURRENCY=50 + +# 飞书机器人告警通知配置 +# 1. 在飞书群添加"自定义机器人"，复制 Webhook 地址填入 FEISHU_WEBHOOK_URL +# 2. 若机器人开启了"签名校验"，把密钥填 FEISHU_SECRET；未开启留空 +# 3. FEISHU_ENABLED=true 开启总开关 +# 4. FEISHU_MIN_SEVERITY=warning 仅推送 >= 该级别的告警（info/warning/error/critical） +# 5. FEISHU_ALERT_TYPES 可选，逗号分隔的告警类型白名单，留空=全部推送 +FEISHU_ENABLED=false +FEISHU_WEBHOOK_URL= +FEISHU_SECRET= +FEISHU_MIN_SEVERITY=warning +FEISHU_ALERT_TYPES= + diff --git a/backend/app/api/v1/alerts.py b/backend/app/api/v1/alerts.py index b8309eb..9a48cd6 100644 --- a/backend/app/api/v1/alerts.py +++ b/backend/app/api/v1/alerts.py @@ -181,3 +181,45 @@ def detect_offline_devices(db: Session = Depends(get_db)): "count": len(offline_devices), "devices": offline_devices } + + +# ========== 飞书告警通知 ========== + +@router.get("/notify/feishu/status", summary="获取飞书通知配置状态") +def get_feishu_status(): + """查看飞书通知是否已配置（不返回密钥等敏感信息）""" + from app.services.feishu_service import get_feishu_service + svc = get_feishu_service() + return { + "enabled": svc.enabled, + "configured": bool(svc.webhook_url), + "webhook_set": bool(svc.webhook_url), + "secret_set": bool(svc.secret), + "min_severity": svc.min_severity, + "alert_types": svc.alert_types, + "cani_push": svc.is_configured(), + } + + +@router.post("/notify/feishu/test", summary="发送飞书测试告警") +def test_feishu_notification(): + """发送一条测试告警到飞书，验证 webhook 配置是否正确""" + from app.services.feishu_service import get_feishu_service + svc = get_feishu_service() + if not svc.is_configured(): + raise HTTPException( + status_code=400, + detail="飞书通知未启用或未配置 webhook。请在 backend/.env 设置 FEISHU_ENABLED=true 和 FEISHU_WEBHOOK_URL" + ) + ok = svc.send_alert({ + "alert_type": "new_device_detected", + "severity": "warning", + "title": "飞书通知测试", + "message": "这是一条测试告警，用于验证飞书机器人 webhook 配置是否正常。", + "ip_address_str": "172.16.0.1", + "mac_address": "00:11:22:33:44:55", + }) + if ok: + return {"message": "飞书测试通知发送成功", "success": True} + raise HTTPException(status_code=500, detail="飞书测试通知发送失败，请检查 webhook 地址和日志") + diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 4297df3..7b498b3 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -28,6 +28,13 @@ class Settings(BaseSettings): PING_RETRIES: int = 2 SCAN_CONCURRENCY: int = 50 + # 飞书机器人告警通知配置 + FEISHU_ENABLED: bool = False # 总开关 + FEISHU_WEBHOOK_URL: str = "" # 飞书自定义机器人 webhook 地址 + FEISHU_SECRET: str = "" # 可选：机器人签名校验密钥（开启签名校验时必填） + FEISHU_MIN_SEVERITY: str = "warning" # 最低告警级别：info/warning/error/critical，仅推送 >= 该级别 + FEISHU_ALERT_TYPES: str = "" # 可选：逗号分隔的告警类型白名单，空=全部推送 + class Config: env_file = ".env" diff --git a/backend/app/services/alert_service.py b/backend/app/services/alert_service.py index 7ad67e5..ab4ca5d 100644 --- a/backend/app/services/alert_service.py +++ b/backend/app/services/alert_service.py @@ -12,6 +12,18 @@ from app.models.alert import ( from app.models.network import Network, IPAddress from app.models.snmp import NetworkDevice, ARPEntry +# 飞书告警通知（延迟导入避免循环依赖） +_feishu_service = None + + +def _get_feishu_service(): + global _feishu_service + if _feishu_service is None: + from app.services.feishu_service import get_feishu_service + _feishu_service = get_feishu_service() + return _feishu_service + + logger = logging.getLogger(__name__) @@ -70,6 +82,22 @@ class AlertService: db.refresh(alert) logger.info(f"创建告警: {alert_type} - {title}") + + # 飞书告警通知（仅新创建的告警推送；existing 去重返回的不会走到这里，避免刷屏） + try: + _get_feishu_service().send_alert({ + "alert_type": alert_type.value if hasattr(alert_type, "value") else str(alert_type), + "severity": severity.value if hasattr(severity, "value") else str(severity), + "title": title, + "message": message, + "ip_address_str": ip_address_str, + "mac_address": mac_address, + "conflicting_mac": conflicting_mac, + "usage_percent": usage_percent, + }) + except Exception as e: + logger.error(f"飞书告警通知失败（不影响告警创建）: {e}") + return alert @staticmethod diff --git a/backend/app/services/feishu_service.py b/backend/app/services/feishu_service.py new file mode 100644 index 0000000..78eeee9 --- /dev/null +++ b/backend/app/services/feishu_service.py @@ -0,0 +1,225 @@ +""" +飞书机器人告警通知服务 +======================== +通过飞书自定义机器人 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"{severity_text}") + 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 get_feishu_service() -> FeishuService: + """获取（缓存的）飞书服务实例""" + global _service + if _service is None: + try: + from app.core.config import settings + alert_types = [t.strip() for t in (settings.FEISHU_ALERT_TYPES or "").split(",") if t.strip()] + _service = FeishuService( + webhook_url=settings.FEISHU_WEBHOOK_URL, + secret=settings.FEISHU_SECRET, + enabled=settings.FEISHU_ENABLED, + min_severity=settings.FEISHU_MIN_SEVERITY or "warning", + alert_types=alert_types, + ) + except Exception as e: + logger.error(f"初始化飞书服务失败: {e}") + _service = FeishuService() + return _service + + +def send_alert_notification(alert_data: Dict[str, Any]) -> bool: + """便捷函数：发送一条告警通知""" + return get_feishu_service().send_alert(alert_data)