feat(通知): 加钉钉 / 飞书两种格式（含各自的加签算法）
## 两家签名算法不一样，照抄必错
| | 钉钉 | 飞书 |
|---|---|---|
| HMAC key | `secret` | `"{timestamp}\n{secret}"` |
| 被签内容 | `"{timestamp}\n{secret}"` | 空 |
| timestamp | 毫秒 | 秒 |
| 拼在哪 | URL query | JSON body |
| 结果 | Base64 再 urlencode | Base64 |
| 出错码 | `errcode 310000` | `code 19021` |
实现为 `BaseAdapter.sign_request()` 钩子（默认不动），两家各写各的；
`core/notifier.py` 的 `_post_once` 在发请求前调用它。
## 另外
- 钉钉：markdown 消息（title+text）；成功码 0；默认限流 15/分（官方 20/分，**超限会被限 10 分钟**，留余量）
- 飞书：交互式卡片（header 颜色按事件级别 + markdown 元素）；成功码 0；默认 60/分
- `PLANNED_FORMATS` 清空（下拉里不再有置灰项）
- **修一个泄漏**：`mask_url` 原来只打码 query 参数——**飞书的 token 在 URL 路径里**（`/hook/<token>`）
→ 接口回显会漏出去。现在同时处理路径 token（≥20 位随机串）与 Slack 的 `/services/T…/B…/X…` 三段。
## 验证（都跑过）
- **签名对拍官方示例**：钉钉逐字节一致（含 urlencode，`%2B` 级别）、飞书的 key/message 口径一致；
时间戳钉住后比对，不靠"自己跟自己一致"
- 成功码按格式：wecom/dingtalk/feishu 的 0 与 bark 的 200 分别判成功，各自的错误码判失败
- 端到端 15 项：钉钉/飞书在假接收端上真发（URL 带签名、body 带 timestamp/sign、卡片是 markdown 元素、
毫秒 vs 秒的时间戳），签名错时正确判失败（310000 / 19021），token 在回显里被打码
This commit is contained in:
+128
-6
@@ -67,8 +67,8 @@ _DEFAULT_SETTINGS = {
|
||||
"http_timeout": 5,
|
||||
}
|
||||
|
||||
# 目前实现不了的格式：前端置灰，但用「通用 JSON」能手搓
|
||||
PLANNED_FORMATS = {"dingtalk": "钉钉", "feishu": "飞书"}
|
||||
# 计划中但还没实现的格式（前端置灰）——钉钉/飞书已于 2026-09-15 实现，这里是留给后续的
|
||||
PLANNED_FORMATS = {}
|
||||
|
||||
|
||||
# ================== URL / 文本脱敏 ==================
|
||||
@@ -76,17 +76,30 @@ _SECRET_KEYS = ("key", "access_token", "token", "secret", "sign", "apikey", "api
|
||||
|
||||
|
||||
def mask_url(url):
|
||||
"""把 URL 里的凭据打码（企微 ?key=、钉钉 ?access_token=、飞书 /hook/<token>）。
|
||||
"""把 URL 里的凭据打码。
|
||||
|
||||
回显、日志、发送记录一律用它——**URL 本身就是可直接发消息的凭据**。
|
||||
两类都要处理（**URL 本身就是可直接发消息的凭据**）：
|
||||
- query 参数：企微 `?key=`、钉钉 `?access_token=` → 值换成 `***`
|
||||
- **路径段的 token**：飞书 `https://open.feishu.cn/open-apis/bot/v2/hook/<token>`
|
||||
（token 在路径最后一段，只处理 query 会漏掉它）
|
||||
回显、发送记录、日志一律用它。
|
||||
"""
|
||||
if not url:
|
||||
return ""
|
||||
try:
|
||||
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
|
||||
sp = urlsplit(url)
|
||||
segs = sp.path.split("/") if sp.path else []
|
||||
# 最后一段是长随机串（飞书 hook token 那种）→ 打码；普通路径（/robot/send）不动
|
||||
if segs and len(segs[-1]) >= 20 and segs[-1].replace("-", "").replace("_", "").isalnum():
|
||||
segs[-1] = "***"
|
||||
sp = sp._replace(path="/".join(segs))
|
||||
# Slack 的凭据在路径三连（/services/T…/B…/X…）——整段打码
|
||||
if sp.netloc in ("hooks.slack.com", "hooks.slack-gov.com") and "services" in segs:
|
||||
i = segs.index("services") + 1
|
||||
sp = sp._replace(path="/".join(segs[:i] + ["***"] * (len(segs) - i)))
|
||||
if not sp.query:
|
||||
return url
|
||||
return urlunsplit((sp.scheme, sp.netloc, sp.path, "", sp.fragment))
|
||||
q = [(k, ("***" if k.lower() in _SECRET_KEYS else v))
|
||||
for k, v in parse_qsl(sp.query, keep_blank_values=True)]
|
||||
# safe='*' 让打码后的 *** 保持原样（否则会被编码成 %2A%2A%2A，看着像乱码）
|
||||
@@ -435,6 +448,15 @@ class BaseAdapter:
|
||||
"""→ requests.post 的参数 dict（url/json/headers）。"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def sign_request(cls, url, secret, body):
|
||||
"""加签钩子：返回 (url, body)。
|
||||
|
||||
默认不改动；钉钉（拼 URL，毫秒）和飞书（放 body，秒）各写各的 —— **两家的
|
||||
签名算法不一样**，照抄另一家的会静默失败（钉钉 errcode 310000、飞书 code 19021）。
|
||||
"""
|
||||
return url, body
|
||||
|
||||
@classmethod
|
||||
def meta(cls):
|
||||
"""给前端用的格式元信息（下拉项 + 随格式变化的提示文案）。"""
|
||||
@@ -581,8 +603,96 @@ class BarkAdapter(BaseAdapter):
|
||||
"headers": dict(hook.get("headers") or {})}
|
||||
|
||||
|
||||
class DingtalkAdapter(BaseAdapter):
|
||||
"""钉钉群机器人：markdown 消息 + 加签。
|
||||
|
||||
加签口径（官方，**和飞书不一样，别照抄**）：
|
||||
timestamp 是**毫秒**；`sign = urlencode(base64(HMAC-SHA256(key=secret,
|
||||
message=f"{timestamp}\\n{secret}")))`；**timestamp/sign 拼到 URL 上**（不是 body）。
|
||||
错了会返回 `errcode 310000 invalid signature`。
|
||||
机器人若用"自定义关键词/IP 白名单"做安全设置，则**不需要** secret（留空即可）。
|
||||
"""
|
||||
|
||||
name = "dingtalk"
|
||||
label = "钉钉"
|
||||
byte_limit = 20000 # markdown 正文上限 20000 字节
|
||||
limit_default = 15 # 官方 20/分，但超限会被限流 10 分钟 —— 留足余量
|
||||
ok_codes = (0,)
|
||||
url_hint = "https://oapi.dingtalk.com/robot/send?access_token=…"
|
||||
url_help = "钉钉：群设置 → 智能群助手 → 添加机器人 → 自定义 → 复制 Webhook 地址"
|
||||
secret_label = "加签密钥（可选）"
|
||||
secret_help = ("机器人安全设置选「加签」时，填 SEC 开头的那串；"
|
||||
"选「自定义关键词/IP 白名单」则留空")
|
||||
limit_help = "官方 20 条/分，超限会被限流 10 分钟（这里默认 15 留余量）"
|
||||
|
||||
@classmethod
|
||||
def render(cls, msg, hook):
|
||||
return {"url": hook["url"],
|
||||
"json": {"msgtype": "markdown",
|
||||
"markdown": {"title": msg["title"],
|
||||
"text": cls._cut(msg["markdown"])}},
|
||||
"headers": dict(hook.get("headers") or {})}
|
||||
|
||||
@classmethod
|
||||
def sign_request(cls, url, secret, body):
|
||||
from urllib.parse import quote_plus
|
||||
ts = str(int(time.time() * 1000)) # 毫秒
|
||||
sign = quote_plus(_hmac_b64(secret.encode("utf-8"),
|
||||
f"{ts}\n{secret}".encode("utf-8")))
|
||||
sep = "&" if "?" in url else "?"
|
||||
return f"{url}{sep}timestamp={ts}&sign={sign}", body
|
||||
|
||||
|
||||
class FeishuAdapter(BaseAdapter):
|
||||
"""飞书群机器人：交互式卡片（markdown 元素）+ 加签。
|
||||
|
||||
加签口径（官方，**和钉钉不一样**）：
|
||||
timestamp 是**秒**；`sign = base64(HMAC-SHA256(key=f"{timestamp}\\n{secret}",
|
||||
message=空))`；**timestamp/sign 放在 JSON body 里**（拼 URL 通不过校验）。
|
||||
错了会返回 `code 19021`。
|
||||
"""
|
||||
|
||||
name = "feishu"
|
||||
label = "飞书"
|
||||
byte_limit = 20000 # 请求体上限 20KB
|
||||
limit_default = 60
|
||||
ok_codes = (0,)
|
||||
url_hint = "https://open.feishu.cn/open-apis/bot/v2/hook/…"
|
||||
url_help = "飞书：群设置 → 群机器人 → 添加机器人 → 自定义机器人 → 复制 Webhook 地址"
|
||||
secret_label = "签名校验密钥（可选）"
|
||||
secret_help = "机器人安全设置勾了「签名校验」才需要填；否则留空"
|
||||
limit_help = "飞书官方 100 条/分；这里默认 60 留余量"
|
||||
|
||||
@classmethod
|
||||
def render(cls, msg, hook):
|
||||
body = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"config": {"wide_screen_mode": True},
|
||||
"header": {
|
||||
"title": {"tag": "plain_text", "content": msg["title"][:120]},
|
||||
"template": {"error": "red", "warning": "orange",
|
||||
"success": "green"}.get(msg["level"], "blue"),
|
||||
},
|
||||
"elements": [{"tag": "markdown",
|
||||
"content": cls._cut(msg["markdown"])}],
|
||||
},
|
||||
}
|
||||
return {"url": hook["url"], "json": body,
|
||||
"headers": dict(hook.get("headers") or {})}
|
||||
|
||||
@classmethod
|
||||
def sign_request(cls, url, secret, body):
|
||||
ts = str(int(time.time())) # 秒
|
||||
body = dict(body)
|
||||
body["timestamp"] = ts
|
||||
body["sign"] = _hmac_b64(f"{ts}\n{secret}".encode("utf-8"), b"")
|
||||
return url, body
|
||||
|
||||
|
||||
ADAPTERS = {WecomAdapter.name: WecomAdapter, JsonAdapter.name: JsonAdapter,
|
||||
BarkAdapter.name: BarkAdapter}
|
||||
BarkAdapter.name: BarkAdapter, DingtalkAdapter.name: DingtalkAdapter,
|
||||
FeishuAdapter.name: FeishuAdapter}
|
||||
|
||||
|
||||
# ================== 发送 ==================
|
||||
@@ -597,8 +707,20 @@ def _record(rec):
|
||||
("" if rec.get("ok") else " | " + scrub(rec.get("error", ""))))
|
||||
|
||||
|
||||
def _hmac_b64(key: bytes, message: bytes) -> str:
|
||||
"""HMAC-SHA256 → Base64（钉钉/飞书加签共用这一步）。"""
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac as _hmac
|
||||
return base64.b64encode(
|
||||
_hmac.new(key, message, digestmod=hashlib.sha256).digest()).decode()
|
||||
|
||||
|
||||
def _post_once(adapter, msg, hook, timeout):
|
||||
kw = adapter.render(msg, hook)
|
||||
if hook.get("secret"):
|
||||
kw["url"], kw["json"] = adapter.sign_request(kw["url"], hook["secret"],
|
||||
kw.get("json") or {})
|
||||
kw["timeout"] = timeout
|
||||
kw["allow_redirects"] = False # 防重定向把消息带到别处
|
||||
r = requests.post(**kw)
|
||||
|
||||
Reference in New Issue
Block a user