fix(notify): 安全审计修复 —— 掩码绕过、SSRF、静默丢失等 12 项

对推送功能做了一轮安全审计,每条都先复现、再修、再补回归测试。
以下按严重程度排列。

【严重】掩码可被「改目标地址、保留凭据」绕过
目标地址与凭据是两套独立字段,而配置合并对「未提及的键」保留原值。于是
只改地址、对凭据避而不谈,服务端就会把库里的真凭据发到任意地址。已实测
复现四个渠道:通用 Webhook 的 Authorization 头、Telegram 的 Bot Token
(进请求路径)、邮件的 SMTP 密码(STARTTLS 后交给对端)、钉钉/飞书的加签
密钥(该项不可利用,一并列出仅为说明面)。完全静默、不依赖重定向,所以
上一轮加的「拒绝跨主机跳转」挡不住它。

修法:新增 Channel.DestinationKeys() 声明「消息发往哪」,PrepareConfigUpdate
在地址变化时要求对**每一个**凭据字段显式表态(给新值,或显式留空表示不再
需要)——原样回传掩码值等于「沿用旧凭据」,正是攻击形状,一并拒绝。
刻意不做「自动丢弃凭据」:对可选字段(如 headers)那会变成「鉴权静默没了
但接口返回成功」,比报错更难排查。

【中】markdown 系三渠道对不可信内容零转义
钉钉/企微/飞书此前完全不转义(Telegram 与邮件都做了),而标题与摘要来自
模型输出、资产 url 来自被测目标。一条标题里的 `[文字](http://evil)` 会在
工程师的钉钉里变成可点击外链;`![](http://evil/beacon)` 会被客户端拉取,
泄露阅读者 IP 并暴露「这条已被看过」。现统一单行化 + markdown 元字符转义。

修的过程中还纠了自己一个设计错误:转义一度被放进 markdownTitle,而它被
markdown 正文、Telegram 的 HTML、飞书卡片、Webhook 的 JSON 与邮件主题四种
语境共用,markdown 的反斜杠泄漏到了 Telegram 消息里(用户会看到 `\(1\)`)。
转义必须发生在各自的渲染出口——测试直接抓到了这个回归。

【中】投递地址的 SSRF 面
此前只校验 scheme 与 host,云元数据 169.254.169.254、环回、内网一律可投递;
而失败响应体前 200 字节会进 last_error 并被投递历史接口回显,构成半盲读
原语。现在在**拨号阶段**设防(最终生效点,同时覆盖 DNS 重绑定与同主机
重定向)。环回/链路本地需显式 ARTEX_NOTIFY_ALLOW_LOCAL=1 才放行(本机
SMTP 中继是合法配置);RFC1918 私网**刻意放行**——内网自建 Mattermost /
SMTP 很常见,做到那个程度会把正常部署一起废掉,这条取舍写进了测试断言。

【中】汇总消息被截断后整批标记已送达 = 静默丢失
渠道都有长度上限(企微 4096 字节最紧)。此前渲染完整篇再截断、然后整批
标记成功:被截掉的那些既不在消息里、也不在失败列表里,投递历史还显示
成功。现改为按**整条**打包,只标记真正装下的,其余回队续发,消息头部
如实写明还有多少条在下一条。被推迟的条目**不消耗重试次数**(领取时乐观
+1 的那次减回去),否则 500 条的积压会在第三段就把尾部判成失败。

【中】其余修复
- min_severity 打错字让过滤器静默失效(未知级别序数为 0 → rank>=0 恒真,
  用户以为「仅高危」实际全推)。改为写入时校验取值。
- rate_per_min=0(不限流)不可达:文档/UI/令牌桶都按 0=不限流解释,唯独
  写库那层 `if <=0 取默认值` 把它改成 20 或 100。默认值改到接口层填。
- digest 渠道完全绕过令牌桶(allow 被扣却没人用)。
- 失败处置按整批 max(attempts) 判断,让老投递连坐同批的新投递。
- 汇总批次里快照无法解析的投递被静默标记成功。
- url.Parse 失败时错误文本仍带完整地址(上一轮只修了 http.Client.Do 那侧;
  且当时的用例全走 scheme 分支、没覆盖解析失败路径,是假保证)。
- 单渠道每轮投递条数可能超出租约时长(多实例下重复发送)。加了防漂移
  断言,写的时候立刻发现原取值 6 恰好用满租约、零余量,已改 5。
- Telegram 截断可能切断 HTML 实体,导致整条被拒收。
- 邮件把 SMTP 4xx(灰名单等临时失败)判成永久失败,最该重试的场景从不重试。
- 提交嵌套结构时会把掩码字面量真的写进库,导致鉴权静默失效。

测试:notify 包从 82% 提到 86%,新增 SSRF 守卫、掩码绕过(用攻击形状的
输入而非「防御逻辑的正确输入」)、分段投递收敛、markdown 转义、SMTP 4xx/5xx
分类等回归用例;端到端补了一条「60 条长标题跨多轮发完且不卡死」的收敛测试。

Co-Authored-By: Claude Code <[email protected]>
This commit is contained in:
luckoneandClaude Code committed 2026-09-30 16:25:06 +08:00
1 parent f3fb37f70d
commit a049f09b97
26 files changed
+1586 -229

No files matched your search

+141 -45
View File
@@ -42,6 +42,22 @@ const (
// 存在的意义是防止「一个渠道配成不限流 + 一次扫出上千条漏洞」把
// 单轮循环拖成长时间阻塞。
notifyUnlimitedBurstPerTick = 50
// notifyMaxSendsPerChannelPerTick 是单渠道每轮最多投递几条。
//
// 这个上限由**租约时长**倒推:领取时给行打的是租约(notifyLease = 3 分钟),
// 若一轮里串行投递的条数多到最坏耗时超过租约,后几条还没发完租约就过期了。
// 单进程内无所谓(Run 是单个 goroutine 串行跑,tick 不会重入),但**两个
// 进程连同一个库**时,对端会把租约过期的行重新领走并重复发送,还会把
// attempts 双份递增、在原进程仍在投递时就判成失败。
//
// 取值:3 分钟租约 / 30 秒单次超时 = 6 是**刚好用满租约**、零余量,
// 不能取;取 5 让最坏耗时 150 秒留出 30 秒余量。这个关系由
// TestNotifyTickBudgetFitsWithinLease 钉死——改 notifyLease、
// notifySendTimeout 或本值中的任意一个都会让那条断言失败。
notifyMaxSendsPerChannelPerTick = 5
// notifySendTimeout 是单次投递的超时。它同时决定上一条常量的取值,
// 两者相乘不能超过 notifyLease,见 TestNotifyTickBudgetFitsWithinLease。
notifySendTimeout = 30 * time.Second
)
// notifyBackoff 是失败重试的退避序列,下标为已尝试次数。
@@ -123,11 +139,15 @@ func (n *Notifier) step(ctx context.Context) {
// 先问令牌桶这一轮还能发几条,再按这个数量去领——顺序不能反,
// 否则被限流挡下的投递已经消耗过重试次数。
allow := n.takeTokens(ch.ID, ch.RatePerMin, time.Now())
if allow > notifyMaxSendsPerChannelPerTick {
allow = notifyMaxSendsPerChannelPerTick
}
if allow <= 0 {
continue
}
if ch.Mode == db.NotifyModeDigest {
n.stepDigest(ctx, ch, baseURL)
// 汇总也受同一份额度约束:领到的条数不能超过本轮令牌数。
n.stepDigest(ctx, ch, allow, baseURL)
continue
}
n.stepRealtime(ctx, ch, allow, baseURL)
@@ -161,7 +181,7 @@ func (n *Notifier) stepRealtime(ctx context.Context, ch *db.NotificationChannel,
}
// stepDigest 在批次到期时把某渠道的待发投递聚合成一条消息发出。
func (n *Notifier) stepDigest(ctx context.Context, ch *db.NotificationChannel, baseURL string) {
func (n *Notifier) stepDigest(ctx context.Context, ch *db.NotificationChannel, allow int, baseURL string) {
window := n.digestInterval()
due, err := n.pg.DigestBatchDue(ctx, ch.ID, window)
if err != nil {
@@ -171,7 +191,7 @@ func (n *Notifier) stepDigest(ctx context.Context, ch *db.NotificationChannel, b
if !due {
return
}
deliveries, err := n.pg.ClaimDigestBatch(ctx, ch.ID, notifyLease)
deliveries, err := n.pg.ClaimDigestBatch(ctx, ch.ID, allow, notifyLease)
if err != nil {
log.Printf("[notify] 领取汇总批次失败 channel=%d: %v", ch.ID, err)
return
@@ -184,49 +204,127 @@ func (n *Notifier) stepDigest(ctx context.Context, ch *db.NotificationChannel, b
_ = n.pg.FailDeliveries(ctx, deliveryIDs(deliveries), fmt.Sprintf("渠道类型 %q 未注册", ch.Kind))
return
}
msg, err := n.renderBatch(ctx, deliveries, baseURL, int(window.Minutes()))
msg, included, err := n.renderBatch(ctx, deliveries, baseURL, int(window.Minutes()))
if err != nil {
_ = n.pg.FailDeliveries(ctx, deliveryIDs(deliveries), err.Error())
return
}
n.send(ctx, channel, cfg, msg, deliveries)
// 快照坏掉、没能进消息的那些投递要显式判失败。不这么做的话它们会留在
// included 之外、既不进消息也不进失败列表——发送成功时它们的状态会被
// 之后的批量标记漏掉,永远停在 sending 直到租约过期被反复领取。
if skipped := excludeDeliveries(deliveries, included); len(skipped) > 0 {
reason := "事件快照无法解析,本条漏洞无法渲染成消息"
if fErr := n.pg.FailDeliveries(ctx, deliveryIDs(skipped), reason); fErr != nil {
log.Printf("[notify] 标记坏快照投递失败 channel=%s ids=%v: %v", ch.Kind, deliveryIDs(skipped), fErr)
}
log.Printf("[notify] 跳过 %d 条快照无法解析的投递 channel=%d", len(skipped), ch.ID)
}
// 只把进了消息的那些交给 send:included[i] 与 msg.Items[i] 严格对应,
// send 依赖这个对应关系把「渠道回报装下了前 K 条」落到正确的投递行上。
n.send(ctx, channel, cfg, msg, included)
}
// send 投递并按结果流转状态。
//
// 同一批投递(汇总模式下可能几十条)共享一个发送结果:要么整批送达、要么整批
// 重试。不做逐条重试——汇总消息是一条,重发其中一部分会让批次语义错乱。
// 同一批投递(汇总模式下可能几十条)共享一个发送结果:要么送达、要么整批重试。
// 不做逐条重试——汇总消息是一条,重发其中一部分会让批次语义错乱。
//
// 唯一的例外是**渠道长度上限导致的分段**:渠道回报实际只装下了前 K 条,
// 那么第 K+1 条起必须留到下一批,而不是跟着一起被标记成功。否则被截掉的
// 那些漏洞既不在消息里、也不在失败列表里,彻底消失。
func (n *Notifier) send(ctx context.Context, channel notify.Channel, cfg map[string]any, msg notify.Message, deliveries []*db.NotificationDelivery) {
ids := deliveryIDs(deliveries)
attempts := maxAttempts(deliveries)
// 单次投递设上限,避免某个渠道卡住把这一轮剩余渠道全部拖住。
sendCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
sendCtx, cancel := context.WithTimeout(ctx, notifySendTimeout)
defer cancel()
err := channel.Send(sendCtx, cfg, msg)
delivered, err := channel.Send(sendCtx, cfg, msg)
if err == nil && delivered > 0 {
if delivered > len(deliveries) {
// 渠道回报的条数不可能超过投递数;真发生了说明渲染层算错了,
// 按全部送达处理并把问题记下来,总好过把记录写乱。
log.Printf("[notify] 渠道回报送达条数 %d 超过投递数 %d channel=%s,按全部送达处理",
delivered, len(deliveries), channel.Kind())
delivered = len(deliveries)
}
sent, rest := deliveries[:delivered], deliveries[delivered:]
if err := n.pg.MarkDeliveriesSent(ctx, deliveryIDs(sent)); err != nil {
log.Printf("[notify] 标记已送达失败 channel=%s ids=%v: %v", channel.Kind(), deliveryIDs(sent), err)
}
if len(rest) > 0 {
// 本条消息已达渠道长度上限:剩下的立刻回队,由下一个 tick 续发。
// 用 DeferDeliveries 而非 RescheduleDeliveries —— 这不是失败,
// 不该消耗重试预算(领取时已经乐观 +1 了,那里会减回去)。
if err := n.pg.DeferDeliveries(ctx, deliveryIDs(rest),
fmt.Sprintf("本条消息已达渠道长度上限,仅送达前 %d 条,其余留待下一批", delivered)); err != nil {
log.Printf("[notify] 分段续发排队失败 channel=%s ids=%v: %v", channel.Kind(), deliveryIDs(rest), err)
}
}
return
}
if err == nil {
if err := n.pg.MarkDeliveriesSent(ctx, ids); err != nil {
log.Printf("[notify] 标记已送达失败 channel=%s ids=%v: %v", channel.Kind(), ids, err)
}
return
// 渠道既没报错也没说送达了多少条。按失败处理(走退避),
// 免得这条投递被反复领取却永远标记不掉。
err = fmt.Errorf("渠道未报告送达条数(delivered=%d)", delivered)
}
// 永久失败与「重试次数用尽」都直接落 failed,等人工在投递历史里重发;
// 其余按退避序列重排。
if notify.IsPermanent(err) || attempts >= db.MaxNotifyAttempts {
reason := err.Error()
if !notify.IsPermanent(err) {
reason = fmt.Sprintf("重试 %d 次后仍失败: %s", attempts, err)
// 失败处置**逐条**决定,而不是拿整批的最大尝试次数做判断。
//
// 曾经是 `if maxAttempts(deliveries) >= MaxNotifyAttempts` 整批判死,但批次里
// 各条的尝试次数并不相同:一个已经重试两次的老投递(attempts=2)会把同一批里
// 全新的投递(attempts=1)一起拖进 failed——新漏洞一条重试都没用上就永久丢了,
// 与「不让老行拖新行下水」的初衷正好相反。
permanent := notify.IsPermanent(err)
var failIDs, exhaustedIDs []int64
byDelay := map[time.Duration][]int64{}
for _, dl := range deliveries {
switch {
case permanent:
failIDs = append(failIDs, dl.ID)
case dl.Attempts >= db.MaxNotifyAttempts:
exhaustedIDs = append(exhaustedIDs, dl.ID)
default:
delay := notifyBackoff[min(dl.Attempts, len(notifyBackoff)-1)]
byDelay[delay] = append(byDelay[delay], dl.ID)
}
if failErr := n.pg.FailDeliveries(ctx, ids, reason); failErr != nil {
log.Printf("[notify] 标记失败状态出错 channel=%s ids=%v: %v", channel.Kind(), ids, failErr)
}
if len(failIDs) > 0 {
if fErr := n.pg.FailDeliveries(ctx, failIDs, err.Error()); fErr != nil {
log.Printf("[notify] 标记失败状态出错 channel=%s ids=%v: %v", channel.Kind(), failIDs, fErr)
}
log.Printf("[notify] 投递失败 channel=%d kind=%s ids=%v: %s", deliveries[0].ChannelID, channel.Kind(), ids, reason)
return
}
delay := notifyBackoff[min(attempts, len(notifyBackoff)-1)]
if rErr := n.pg.RescheduleDeliveries(ctx, ids, delay, err.Error()); rErr != nil {
log.Printf("[notify] 重排投递失败 channel=%s ids=%v: %v", channel.Kind(), ids, rErr)
if len(exhaustedIDs) > 0 {
reason := fmt.Sprintf("重试 %d 次后仍失败: %s", db.MaxNotifyAttempts, err)
if fErr := n.pg.FailDeliveries(ctx, exhaustedIDs, reason); fErr != nil {
log.Printf("[notify] 标记失败状态出错 channel=%s ids=%v: %v", channel.Kind(), exhaustedIDs, fErr)
}
}
// 按延迟分组重排:只有 3 档退避,分组数天然很小,不必为每条单独发一次
// UPDATE(那会让一个 500 条的批次产生 500 次往返)。
for delay, group := range byDelay {
if rErr := n.pg.RescheduleDeliveries(ctx, group, delay, err.Error()); rErr != nil {
log.Printf("[notify] 重排投递失败 channel=%s ids=%v: %v", channel.Kind(), group, rErr)
}
}
if len(failIDs)+len(exhaustedIDs) > 0 {
log.Printf("[notify] 投递失败 channel=%d kind=%s 永久失败=%d 重试耗尽=%d 待重试=%d: %s",
deliveries[0].ChannelID, channel.Kind(), len(failIDs), len(exhaustedIDs), len(byDelay), err)
}
}
// excludeDeliveries 返回 all 中不在 keep 里的那些(按指针身份比较)。
// 用于找出「没能进消息」的投递——它们必须被显式处置,不能留在灰色地带。
func excludeDeliveries(all, keep []*db.NotificationDelivery) []*db.NotificationDelivery {
inKeep := make(map[*db.NotificationDelivery]bool, len(keep))
for _, dl := range keep {
inKeep[dl] = true
}
var out []*db.NotificationDelivery
for _, dl := range all {
if !inKeep[dl] {
out = append(out, dl)
}
}
return out
}
// adapt 取渠道实现并解析其配置。
@@ -263,29 +361,39 @@ func (n *Notifier) renderSingle(ctx context.Context, dl *db.NotificationDelivery
// renderBatch 渲染汇总消息。逐条解析快照——单条坏了只跳过那一条,
// 不让它把整批汇总拖没。
func (n *Notifier) renderBatch(ctx context.Context, deliveries []*db.NotificationDelivery, baseURL string, windowMinutes int) (notify.Message, error) {
//
// 返回值 included 与 msg.Items **严格一一对应**(第 i 个投递 ↔ 第 i 个条目)。
// 这个对应关系是硬要求:调用方按「渠道回报装下了前 K 条」来决定前 K 个投递
// 标记已送达。若这里跳过了坏快照却不把跳过的投递从 included 里剔除,
// 下标就会错位——本该失败的坏条目会被标成已送达,而好条目被误判为未送达。
// 坏掉的那些由调用方显式标记失败,见 stepDigest。
func (n *Notifier) renderBatch(ctx context.Context, deliveries []*db.NotificationDelivery, baseURL string, windowMinutes int) (notify.Message, []*db.NotificationDelivery, error) {
items := make([]notify.Item, 0, len(deliveries))
included := make([]*db.NotificationDelivery, 0, len(deliveries))
for _, dl := range deliveries {
snap, err := parseSnapshot(dl)
if err != nil {
// 坏快照不进消息,也不进 included——它的处置由调用方负责
// (显式标记失败,而不是混在「已送达」里蒙混过关)。
log.Printf("[notify] 汇总批次中跳过无法解析的快照 delivery=%d: %v", dl.ID, err)
continue
}
item, err := n.itemFor(ctx, snap, baseURL)
if err != nil {
return notify.Message{}, err
return notify.Message{}, nil, err
}
items = append(items, item)
included = append(included, dl)
}
if len(items) == 0 {
return notify.Message{}, fmt.Errorf("汇总批次 %d 条投递全部无法解析", len(deliveries))
return notify.Message{}, nil, fmt.Errorf("汇总批次 %d 条投递全部无法解析", len(deliveries))
}
return notify.Message{
Items: items,
Batch: true,
WindowMinutes: windowMinutes,
HomeURL: baseURL,
}, nil
}, included, nil
}
// itemFor 把事件快照渲染成待推送条目,顺带解析资产名与详情回链。
@@ -397,18 +505,6 @@ func deliveryIDs(deliveries []*db.NotificationDelivery) []int64 {
return out
}
// maxAttempts 取批次里最大的已尝试次数。批次会一起成败,用最大值决定是否
// 还有重试预算,避免新加入的行被老行的次数拖下水(反之亦然)。
func maxAttempts(deliveries []*db.NotificationDelivery) int {
m := 0
for _, dl := range deliveries {
if dl.Attempts > m {
m = dl.Attempts
}
}
return m
}
func trimTrailingSlash(s string) string {
for len(s) > 0 && s[len(s)-1] == '/' {
s = s[:len(s)-1]
+17 -2
View File
@@ -217,12 +217,19 @@ func (s *Server) notifyCreateChannel(w http.ResponseWriter, r *http.Request) {
ch.Mode = *req.Mode
}
if req.RatePerMin != nil {
// 显式给值就照用——包括 0,它表示「不限流」,是合法配置。
if *req.RatePerMin < 0 {
writeErr(w, 400, "限流值不能为负")
return
}
ch.RatePerMin = *req.RatePerMin
}
// 只有「字段缺省」才套用渠道默认值。默认值必须在这里决定而不是在 db 层:
// 只有请求体能区分「没传这个字段」与「显式传了 0」,而两者的含义完全不同
// (前者=用默认,后者=不限流)。db 层把 0 也当未指定,会让不限流配置不可达。
if req.RatePerMin == nil {
ch.RatePerMin = channel.DefaultRatePerMin()
}
if req.Filter != nil {
// 写入时校验取值受限的过滤字段(如 min_severity)。详见 notify.Filter.Validate:
// 门槛打错字会让过滤器静默失效变成全推,必须在入口拦掉。
@@ -285,7 +292,13 @@ func (s *Server) notifyUpdateChannel(w http.ResponseWriter, r *http.Request) {
if stored == nil {
stored = map[string]any{}
}
merged := notify.MergeConfig(stored, req.Config)
// 用 PrepareConfigUpdate 而不是裸的 MergeConfig:目标地址变更时必须让操作者
// 对凭据字段重新表态,否则「只改地址、凭据沿用」会把库里的真凭据发到新地址。
merged, err := notify.PrepareConfigUpdate(kind, stored, req.Config)
if err != nil {
writeErr(w, 400, err.Error())
return
}
if err := channel.Validate(merged); err != nil {
writeErr(w, 400, err.Error())
return
@@ -413,7 +426,9 @@ func (s *Server) notifyTestChannel(w http.ResponseWriter, r *http.Request) {
}
msg := notifyTestMessage(s.notifierBaseURL(pg))
start := time.Now()
if err := channel.Send(r.Context(), cfg, msg); err != nil {
// 测试消息只有一条,送达条数这里不需要(渠道长度上限对单条消息而言
// 由截断兜底,不涉及分段)。
if _, err := channel.Send(r.Context(), cfg, msg); err != nil {
// 把渠道返回的原始错误如实回给用户——这是他们调试配置的唯一线索。
writeErr(w, 502, err.Error())
return
+141 -11
View File
@@ -45,6 +45,10 @@ type notifyFixture struct {
func newNotifyFixture(t *testing.T) *notifyFixture {
t.Helper()
// 本文件的所有假接收端都跑在 127.0.0.1 上,而投递默认拒绝环回地址
// (防 SSRF 打到同机服务与云元数据)。测试显式打开这个开关;
// 守卫「默认拒绝」的行为由 notify 包的 ssrf_test.go 覆盖。
t.Setenv(notify.AllowLocalTargetsEnv, "1")
s, _, request := trafficEvidenceServer(t)
pg := s.m.pg
@@ -194,18 +198,22 @@ func (f *fakeWebhook) last(t *testing.T) map[string]any {
return f.body(t, f.count()-1)
}
// markdownText 从钉钉 markdown/actionCard 请求体里取出正文。
// markdownText 从请求体里取出正文,兼容各家的字段名差异:
// 钉钉 markdown 用 `text`、ActionCard 用 `text`、企业微信 markdown 用 `content`。
func markdownText(t *testing.T, body map[string]any) string {
t.Helper()
if md, ok := body["markdown"].(map[string]any); ok {
s, _ := md["text"].(string)
return s
for _, key := range []string{"markdown", "actionCard"} {
section, ok := body[key].(map[string]any)
if !ok {
continue
}
for _, field := range []string{"text", "content"} {
if s, ok := section[field].(string); ok && s != "" {
return s
}
}
}
if card, ok := body["actionCard"].(map[string]any); ok {
s, _ := card["text"].(string)
return s
}
t.Fatalf("请求体里没有 markdown/actionCard 正文: %v", body)
t.Fatalf("请求体里没有可识别的正文: %v", body)
return ""
}
@@ -407,14 +415,14 @@ func TestNotifyDigestBatchesMultipleFindingsIntoOneMessage(t *testing.T) {
ch := f.channel(t, chID)
// 未到期:不发。
f.n.stepDigest(ctx, ch, "")
f.n.stepDigest(ctx, ch, 50, "")
if hook.count() != 0 {
t.Fatal("汇总批次未到期就发了")
}
// 催老批次后:三条合成一条消息。
f.agePendingBatch(t, chID)
f.n.stepDigest(ctx, ch, "")
f.n.stepDigest(ctx, ch, 50, "")
if got := hook.count(); got != 1 {
t.Fatalf("三条应汇总成一条消息,实际发了 %d 条", got)
}
@@ -713,6 +721,112 @@ func TestNotifyNoDeepLinkWithoutBaseURL(t *testing.T) {
}
}
// TestNotifyDigestSegmentsAndDefersRemainder 是「静默丢失」修复的端到端证据。
//
// 汇总消息受渠道长度上限约束(企微 4096 字节),一批装不下时必须**按整条**切分:
// 装进本条的那些标记已送达,其余回到队列等下一条。曾经的实现是把整批标记
// 成功——被截掉的那些既不在消息里、也不在失败列表里,投递历史还显示成功,
// 漏洞就这么没了。
//
// 断言四件事:① 只标记了实际装下的条数 ② 其余仍是待发 ③ 被推迟的条目
// **没有消耗重试次数** ④ 再跑一轮能把剩下的发出去(不会卡死)。
func TestNotifyDigestSegmentsAndDefersRemainder(t *testing.T) {
f := newNotifyFixture(t)
hook := newFakeWebhook(t)
// 用企业微信:markdown 上限 4096 字节,是六个渠道里最紧的。
chID := f.createChannel(t, map[string]any{
"name": "分段汇总",
"kind": notify.KindWeCom,
"mode": db.NotifyModeDigest,
"config": map[string]any{"webhook": hook.URL},
})
const total = 60
// 标题取长一点,保证 60 条远超 4096 字节,必然分段。
longName := strings.Repeat("超长漏洞名称", 6)
for i := 0; i < total; i++ {
f.record(t, longName+strconv.Itoa(i+1), "high")
}
ctx := context.Background()
if _, _, err := f.pg.FanOutPendingEvents(ctx, 500); err != nil {
t.Fatal(err)
}
f.agePendingBatch(t, chID)
ch := f.channel(t, chID)
f.n.stepDigest(ctx, ch, 50, "")
if hook.count() != 1 {
t.Fatalf("应只发出一条消息,得到 %d", hook.count())
}
var sent, pending int
if err := f.pg.QueryRow(`SELECT
count(*) FILTER (WHERE state=$2),
count(*) FILTER (WHERE state=$3)
FROM notification_deliveries WHERE channel_id=$1`, chID, db.NotifyStateSent, db.NotifyStatePending).
Scan(&sent, &pending); err != nil {
t.Fatal(err)
}
if sent == 0 {
t.Fatal("应有条目被标记为已送达")
}
if pending == 0 {
t.Fatalf("一批 %d 条不可能全装进 4096 字节,应有剩余待发;sent=%d", total, sent)
}
if sent+pending != total {
t.Fatalf("条目数对不上:sent=%d pending=%d total=%d(既没送达也没待发=丢失)", sent, pending, total)
}
// 消息正文必须如实告知还有多少条没包含在本条里。
if text := markdownText(t, hook.last(t)); !strings.Contains(text, "其余") {
t.Fatalf("消息应说明还有条目未包含在本条:\n%.400s", text)
}
// 被推迟的条目不得消耗重试预算:领取时 attempts 已乐观 +1,推迟时要减回去。
var maxAttempts int
if err := f.pg.QueryRow(`SELECT COALESCE(max(attempts),0) FROM notification_deliveries
WHERE channel_id=$1 AND state=$2`, chID, db.NotifyStatePending).Scan(&maxAttempts); err != nil {
t.Fatal(err)
}
if maxAttempts > 0 {
t.Fatalf("被推迟的条目不该消耗重试次数(否则几条之后就会被判失败),得到 attempts=%d", maxAttempts)
}
// 反复跑直到收敛。断言的是**最终全部送达**且中途确实分了多轮——
// 这比「第二轮发完」更强:它证明分段不会卡死、也不会把剩余条目丢掉。
rounds := 0
for {
var undelivered int
if err := f.pg.QueryRow(`SELECT count(*) FROM notification_deliveries
WHERE channel_id=$1 AND state <> $2 AND state <> $3`, chID, db.NotifyStateSent, db.NotifyStateFailed).
Scan(&undelivered); err != nil {
t.Fatal(err)
}
if undelivered == 0 {
break
}
rounds++
if rounds > total+5 {
t.Fatalf("分段投递不收敛:跑了 %d 轮仍有 %d 条悬而未决", rounds, undelivered)
}
before := hook.count()
f.n.stepDigest(ctx, ch, 50, "")
if hook.count() == before {
t.Fatalf("第 %d 轮没有任何进展,剩余 %d 条会永久卡住", rounds, undelivered)
}
}
if rounds < 2 {
t.Fatalf("一条 4096 字节的消息装不下 %d 条长标题漏洞,应分多轮发出,实际只用了 %d 轮", total, rounds)
}
// 首轮之后的每一轮都应是**纯续发**,不存在被渠道拒绝的条目。
var failed int
if err := f.pg.QueryRow(`SELECT count(*) FROM notification_deliveries WHERE channel_id=$1 AND state=$2`,
chID, db.NotifyStateFailed).Scan(&failed); err != nil {
t.Fatal(err)
}
if failed != 0 {
t.Fatalf("假接收端始终返回成功,不该有失败条目,得到 %d", failed)
}
}
// TestNotifyBackoffTableMatchesAttemptBudget 是防漂移断言。
//
// 重试预算(db.MaxNotifyAttempts)与退避序列表(notifyBackoff)分居两个包:
@@ -763,3 +877,19 @@ func TestNotifyRateLimitDoesNotConsumeRetryBudget(t *testing.T) {
t.Fatalf("渠道 1 的桶应仍然为空,得到 %d", got)
}
}
// TestNotifyTickBudgetFitsWithinLease 是又一条防漂移断言。
//
// 单渠道每轮的投递条数上限(notifyMaxSendsPerChannelPerTick)是从租约时长倒推的:
// 一轮里串行投递的最坏耗时必须 < 租约,否则后几条还没发完租约就过期,
// 多实例部署时对端会把它们重新领走、重复发送。这三个常量分处不同位置,
// 改任意一个都可能打破关系而不会有任何报错——所以在这里钉死。
func TestNotifyTickBudgetFitsWithinLease(t *testing.T) {
worst := time.Duration(notifyMaxSendsPerChannelPerTick) * notifySendTimeout
if worst >= notifyLease {
t.Fatalf("单渠道一轮的最坏耗时 %v 不应达到或超过租约 %v"+
"(notifyMaxSendsPerChannelPerTick=%d × notifySendTimeout=%v)——"+
"改这三个常量中的任意一个都要同步检查另外两个",
worst, notifyLease, notifyMaxSendsPerChannelPerTick, notifySendTimeout)
}
}