文章总结： 本文档提供了用于Zabbix6.4LTS集成的钉钉告警Python脚本，支持Markdown等多种消息格式及加签安全验证。文章包含完整的安装部署脚本、Zabbix媒体类型配置指南及消息模板设置，详细介绍了参数传递和错误处理机制，旨在实现高效稳定的监控告警通知。 综合评分： 88 文章分类： 安全工具,安全运营
Zabbix 6.4 LTS钉钉告警脚本
原创
刘军军 刘军军
运维星火燎原
2026年2月1日 00:01 北京
完整的钉钉告警脚本
1.1 主脚本文件：dingtalk_alert.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Zabbix 6.4 LTS 钉钉告警脚本
支持多种消息格式：Markdown、Text、ActionCard、FeedCard
支持加签安全验证
支持@特定用户
"""
import requests
import json
import sys
import os
import hmac
import hashlib
import base64
import time
import urllib.parse
import logging
from typing import Dict, List, Optional, Union
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('/var/log/zabbix/dingtalk_alert.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger('dingtalk_alert')
classDingTalkAlert:
def__init__(self, webhook_url: str, secret: str = None, timeout: int = 10):
"""
初始化钉钉告警类
Args:
webhook_url: 钉钉Webhook地址
secret: 加签密钥
timeout: 请求超时时间(秒)
"""
self.webhook_url = webhook_url
self.secret = secret
self.timeout = timeout
def_generate_signature(self) -> str:
"""生成加签参数"""
ifnot self.secret:
return self.webhook_url
timestamp = str(round(time.time() * 1000))
string_to_sign = f"{timestamp}\n{self.secret}"
# 计算签名
hmac_code = hmac.new(
self.secret.encode('utf-8'),
string_to_sign.encode('utf-8'),
digestmod=hashlib.sha256
).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
returnf"{self.webhook_url}×tamp={timestamp}&sign={sign}"
def_send_request(self, data: Dict) -> bool:
"""发送请求到钉钉"""
url = self._generate_signature()
headers = {
'Content-Type': 'application/json',
'User-Agent': 'Zabbix-DingTalk-Alert/1.0'
}
try:
response = requests.post(
url,
data=json.dumps(data),
headers=headers,
timeout=self.timeout
)
result = response.json()
if result.get('errcode') == 0:
logger.info("消息发送成功")
returnTrue
else:
logger.error(f"发送失败: {result.get('errmsg')}")
returnFalse
except requests.exceptions.Timeout:
logger.error("请求超时")
returnFalse
except requests.exceptions.ConnectionError:
logger.error("网络连接错误")
returnFalse
except Exception as e:
logger.error(f"请求异常: {str(e)}")
returnFalse
defsend_markdown(self, title: str, text: str, at_mobiles: List[str] = None,
at_user_ids: List[str] = None, is_at_all: bool = False) -> bool:
"""
发送Markdown格式消息
Args:
title: 消息标题
text: Markdown格式文本
at_mobiles: 要@的手机号列表
at_user_ids: 要@的用户ID列表
is_at_all: 是否@所有人
"""
at_data = {
"atMobiles": at_mobiles or [],
"atUserIds": at_user_ids or [],
"isAtAll": is_at_all
}
data = {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": text
},
"at": at_data
}
return self._send_request(data)
defsend_text(self, content: str, at_mobiles: List[str] = None,
at_user_ids: List[str] = None, is_at_all: bool = False) -> bool:
"""发送文本格式消息"""
at_data = {
"atMobiles": at_mobiles or [],
"atUserIds": at_user_ids or [],
"isAtAll": is_at_all
}
data = {
"msgtype": "text",
"text": {
"content": content
},
"at": at_data
}
return self._send_request(data)
defsend_action_card(self, title: str, text: str, single_title: str,
single_url: str, btn_orientation: str = "0") -> bool:
"""发送ActionCard格式消息"""
data = {
"msgtype": "actionCard",
"actionCard": {
"title": title,
"text": text,
"singleTitle": single_title,
"singleURL": single_url,
"btnOrientation": btn_orientation
}
}
return self._send_request(data)
defsend_feed_card(self, links: List[Dict]) -> bool:
"""发送FeedCard格式消息"""
data = {
"msgtype": "feedCard",
"feedCard": {
"links": links
}
}
return self._send_request(data)
classZabbixMessageFormatter:
"""Zabbix消息格式化类"""
@staticmethod
defformat_problem_message(trigger_data: Dict) -> str:
"""格式化问题告警消息"""
severity_emoji = {
'Not classified': '⚪',
'Information': '🔵',
'Warning': '🟡',
'Average': '🟠',
'High': '🔴',
'Disaster': '💀'
}
severity = trigger_data.get('TRIGGER.SEVERITY', 'Not classified')
emoji = severity_emoji.get(severity, '⚪')
message = f"""## {emoji} [{severity}] 告警通知
**🔖 事件ID**: {trigger_data.get('EVENT.ID', 'N/A')}
**🏷️ 主机名**: {trigger_data.get('HOST.NAME', 'N/A')}
**📍 IP地址**: {trigger_data.get('HOST.IP', 'N/A')}
**🚨 问题描述**: {trigger_data.get('TRIGGER.NAME', 'N/A')}
**📊 严重程度**: {severity}
**⏰ 发生时间**: {trigger_data.get('EVENT.DATE', 'N/A')}{trigger_data.get('EVENT.TIME', 'N/A')}
**⏱️ 持续时间**: {trigger_data.get('EVENT.AGE', 'N/A')}
**📈 监控项值**:
{trigger_data.get('ITEM.NAME', 'N/A')}: `{trigger_data.get('ITEM.VALUE', 'N/A')}`
**🔍 触发条件**:
`{trigger_data.get('TRIGGER.EXPRESSION', 'N/A')}`
**📋 详细信息**:
{trigger_data.get('TRIGGER.URL', 'N/A')}
---
💡 *来自 Zabbix 监控系统*"""
return message
@staticmethod
defformat_recovery_message(trigger_data: Dict) -> str:
"""格式化恢复消息"""
message = f"""## ✅ 问题已恢复
**🏷️ 主机名**: {trigger_data.get('HOST.NAME', 'N/A')}
**🚨 问题描述**: {trigger_data.get('TRIGGER.NAME', 'N/A')}
**🕒 恢复时间**: {trigger_data.get('EVENT.DATE', 'N/A')}{trigger_data.get('EVENT.TIME', 'N/A')}
**⏱️ 持续时间**: {trigger_data.get('EVENT.AGE', 'N/A')}
**📈 当前值**:
{trigger_data.get('ITEM.NAME', 'N/A')}: `{trigger_data.get('ITEM.VALUE', 'N/A')}`
**🎉 状态**: 已恢复正常
---
💡 *来自 Zabbix 监控系统*"""
return message
@staticmethod
defformat_simple_text(trigger_data: Dict, is_recovery: bool = False) -> str:
"""格式化简单文本消息"""
if is_recovery:
returnf"✅ 恢复告警: {trigger_data.get('HOST.NAME')} - {trigger_data.get('TRIGGER.NAME')}"
else:
returnf"🚨 问题告警: {trigger_data.get('HOST.NAME')} - {trigger_data.get('TRIGGER.NAME')}"
defparse_zabbix_arguments():
"""解析Zabbix传递的参数"""
if len(sys.argv) < 4:
logger.error("参数不足，至少需要3个参数")
sys.exit(1)
# Zabbix传递的参数顺序
webhook_url = sys.argv[1] # {ALERT.SENDTO}
subject = sys.argv[2] # {ALERT.SUBJECT}
message = sys.argv[3] # {ALERT.MESSAGE}
secret = sys.argv[4] if len(sys.argv) > 4elseNone# {ALERT.SECRET}
at_mobiles = sys.argv[5] if len(sys.argv) > 5elseNone# {ALERT.AT_MOBILES}
# 解析消息内容（Zabbix格式）
trigger_data = {}
lines = message.split('\n')
for line in lines:
if':'in line:
key, value = line.split(':', 1)
trigger_data[key.strip()] = value.strip()
return {
'webhook_url': webhook_url,
'subject': subject,
'message': message,
'secret': secret,
'at_mobiles': at_mobiles.split(',') if at_mobiles else [],
'trigger_data': trigger_data
}
defmain():
"""主函数"""
try:
# 解析参数
args = parse_zabbix_arguments()
# 创建钉钉发送器
dingtalk = DingTalkAlert(
webhook_url=args['webhook_url'],
secret=args['secret']
)
# 判断消息类型（问题或恢复）
is_recovery = '恢复'in args['subject'] or'RESOLVED'in args['subject']
# 格式化消息
formatter = ZabbixMessageFormatter()
if is_recovery:
title = f"✅ {args['subject']}"
message_text = formatter.format_recovery_message(args['trigger_data'])
else:
title = f"🚨 {args['subject']}"
message_text = formatter.format_problem_message(args['trigger_data'])
# 发送消息
success = dingtalk.send_markdown(
title=title,
text=message_text,
at_mobiles=args['at_mobiles']
)
ifnot success:
logger.error("消息发送失败")
sys.exit(1)
logger.info("消息发送成功")
except Exception as e:
logger.error(f"脚本执行异常: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
main()
1.2 安装和配置脚本
#!/bin/bash
# install_dingtalk_alert.sh - 钉钉告警脚本安装脚本
# 创建脚本目录
sudo mkdir -p /usr/lib/zabbix/alertscripts
sudo chown zabbix:zabbix /usr/lib/zabbix/alertscripts
# 复制脚本文件
sudo cp dingtalk_alert.py /usr/lib/zabbix/alertscripts/
sudo chmod 755 /usr/lib/zabbix/alertscripts/dingtalk_alert.py
sudo chown zabbix:zabbix /usr/lib/zabbix/alertscripts/dingtalk_alert.py
# 创建日志目录
sudo mkdir -p /var/log/zabbix
sudo chown zabbix:zabbix /var/log/zabbix
# 安装Python依赖
sudo apt update
sudo apt install python3 python3-pip -y
sudo pip3 install requests
# 测试脚本
echo"测试脚本安装..."
sudo -u zabbix python3 /usr/lib/zabbix/alertscripts/dingtalk_alert.py --help
echo"钉钉告警脚本安装完成！"
1.3 Zabbix媒体类型配置
在Zabbix Web界面配置：
- Administration → Media types → Create media type
- 基本配置：
- Name: DingTalk Alert
- Type: Script
- Script name: dingtalk_alert.py
- 脚本参数：
{ALERT.SENDTO}
{ALERT.SUBJECT}
{ALERT.MESSAGE}
{ALERT.SECRET}
{ALERT.AT_MOBILES}
- 消息模板：
问题主题模板：
[{TRIGGER.SEVERITY}] Problem: {TRIGGER.NAME}
问题消息模板：
TRIGGER.NAME: {TRIGGER.NAME}
TRIGGER.SEVERITY: {TRIGGER.SEVERITY}
HOST.NAME: {HOST.NAME}
HOST.IP: {HOST.IP}
EVENT.DATE: {EVENT.DATE}
EVENT.TIME: {EVENT.TIME}
EVENT.AGE: {EVENT.AGE}
EVENT.ID: {EVENT.ID}
ITEM.NAME: {ITEM.NAME}
ITEM.VALUE: {ITEM.VALUE}
TRIGGER.EXPRESSION: {TRIGGER.EXPRESSION}
TRIGGER.URL: {TRIGGER.URL}
恢复主题模板：
✅ Resolved: {TRIGGER.NAME}
1.4 测试脚本
# 测试脚本功能
cd /usr/lib/zabbix/alertscripts
# 测试帮助信息
sudo -u zabbix python3 dingtalk_alert.py --help
# 测试发送消息
sudo -u zabbix python3 dingtalk_alert.py \
"https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN" \
"测试告警" \
"TRIGGER.NAME: CPU负载过高\nTRIGGER.SEVERITY: High\nHOST.NAME: test-server\nHOST.IP: 192.168.1.100" \
"YOUR_SECRET" \
"13800138000"
1.5 高级功能：支持命令行参数
#!/bin/bash
# dingtalk_wrapper.sh - 包装脚本，支持更多参数
#!/bin/bash
# 钉钉告警包装脚本
WEBHOOK_URL="$1"
SUBJECT="$2"
MESSAGE="$3"
SECRET="$4"
AT_MOBILES="$5"
MESSAGE_TYPE="$6"# problem/recovery
# 调用Python脚本
python3 /usr/lib/zabbix/alertscripts/dingtalk_alert.py \
"$WEBHOOK_URL" \
"$SUBJECT" \
"$MESSAGE" \
"$SECRET" \
"$AT_MOBILES"
# 记录执行结果
echo"$(date): 执行钉钉告警脚本" >> /var/log/zabbix/dingtalk.log
1.6 监控脚本运行状态
#!/bin/bash
# monitor_dingtalk.sh - 监控钉钉告警脚本运行状态
LOG_FILE="/var/log/zabbix/dingtalk_alert.log"
ERROR_PATTERN="ERROR|失败|异常"
# 检查最近错误
recent_errors=$(tail -100 "$LOG_FILE" | grep -E "$ERROR_PATTERN" | wc -l)
if [ "$recent_errors" -gt 5 ]; then
echo"钉钉告警脚本最近出现 $recent_errors 个错误，请检查！"
exit 1
else
echo"钉钉告警脚本运行正常"
exit 0
fi
使用说明
基本用法
# 直接调用
python3 dingtalk_alert.py<webhook_url><subject><message> [secret] [at_mobiles]
# 示例
python3 dingtalk_alert.py \
"https://oapi.dingtalk.com/robot/send?access_token=abc123" \
"CPU负载过高" \
"TRIGGER.NAME: CPU负载过高\nHOST.NAME: web-server01" \
"SECRET123" \
"13800138000,13900139000"
Zabbix集成
- 将脚本放置在 /usr/lib/zabbix/alertscripts/
- 在Zabbix中创建媒体类型
- 配置用户报警媒介
- 设置告警动作
支持的变量
- {ALERT.SENDTO} – Webhook URL
- {ALERT.SUBJECT} – 消息主题
- {ALERT.MESSAGE} – 消息内容
- {ALERT.SECRET} – 加签密钥
- {ALERT.AT_MOBILES} – @的用户手机号
这个脚本提供了完整的钉钉告警功能，包括：
- ✅ 多种消息格式支持
- ✅ 加签安全验证
- ✅ @特定用户功能
- ✅ 详细的日志记录
- ✅ 错误处理和重试机制
- ✅ Zabbix 6.4 LTS完全兼容
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：运维星火燎原 刘军军 刘军军《Zabbix 6.4 LTS钉钉告警脚本》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论