feat: 钉钉机器人支持加签安全设置
All checks were successful
continuous-integration/drone/push Build is passing

- 通知渠道增加 sign_secret 字段存储加签密钥
- 发送钉钉消息时自动计算签名
- 前端增加加签密钥输入框(仅钉钉机器人显示)
This commit is contained in:
2026-01-28 17:19:53 +08:00
parent 8430f9dbaa
commit 333bbe57eb
4 changed files with 66 additions and 2 deletions

View File

@@ -282,7 +282,31 @@ class SchedulerService:
async def _send_to_channel(self, channel: TaskNotifyChannel, content: str, title: str):
"""发送消息到通知渠道"""
import time
import hmac
import hashlib
import base64
import urllib.parse
url = channel.webhook_url
if channel.channel_type == 'dingtalk_bot':
# 钉钉加签
if channel.sign_secret:
timestamp = str(round(time.time() * 1000))
string_to_sign = f'{timestamp}\n{channel.sign_secret}'
hmac_code = hmac.new(
channel.sign_secret.encode('utf-8'),
string_to_sign.encode('utf-8'),
digestmod=hashlib.sha256
).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
if '?' in url:
url = f"{url}&timestamp={timestamp}&sign={sign}"
else:
url = f"{url}?timestamp={timestamp}&sign={sign}"
payload = {
"msgtype": "markdown",
"markdown": {
@@ -299,7 +323,7 @@ class SchedulerService:
}
async with httpx.AsyncClient(timeout=10) as client:
response = await client.post(channel.webhook_url, json=payload)
response = await client.post(url, json=payload)
result = response.json()
if result.get('errcode') != 0:
print(f"通知发送失败: {result}")