""" 飞书机器人告警通知服务 ======================== 通过飞书自定义机器人 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)