Merge pull request #187 from Autumn-27/fix/notify-digest-and-mask

fix(notify): 汇总模式退化、目的地留空误判、SMTP 缺 SSRF 守卫

db/notification_delivery.go

@@ -171,9 +171,15 @@ func (d *DB) ClaimDigestBatch(ctx context.Context, channelID int64, limit int, l
 	if limit <= 0 {
 		return nil, nil
 	}
-	// 单批同时受两个上界约束：调用方的限流额度（本轮还能发几条）与内存上界
-	// MaxDigestBatchSize。此前只用了后者，于是 digest 渠道的 rate_per_min
-	// 完全不生效——令牌被 takeTokens 扣掉却没人用，等于白扣。
+	// limit 是**内存上界**，调用方传 MaxDigestBatchSize；这里再夹一道，
+	// 防止调用方传进一个更大的值。
+	//
+	// 刻意不接受「限流额度」充当批次大小：限流的单位是消息条数——一个批次只发
+	// 一条消息、消耗一个令牌，由 server 层的 takeTokens 扣除——与「一批装几条
+	// 漏洞」是两个不同的量纲。曾经为了让 rate_per_min 对 digest 生效而把每轮
+	// 请求预算传进来当批次大小，结果 rate=20/min 的渠道每批只装 1 条漏洞，
+	// digest 退化成带汇总文案的实时推送。要改限流请改 takeTokens 的 want，
+	// 不要动这里。
 	if limit > MaxDigestBatchSize {
 		limit = MaxDigestBatchSize
 	}

notify/email.go

@@ -130,8 +130,13 @@ func (c emailChannel) Send(ctx context.Context, cfg map[string]any, m Message) (
 // 会话期限在**建连处**就设好（而非事后补设），因为 net/smtp 的 Client 把底层
 // 连接藏在未导出字段里，外部拿不到它；连接一旦交出去就只能靠预先设置的 deadline
 // 兜底。这也顺带覆盖了握手阶段的阻塞。
+// Control 挂 blockInternalDial 与 HTTP 系渠道共用同一道守卫。不挂的话 SMTP
+// 就是整套 SSRF 防护的缺口：host 填 169.254.169.254 或 127.0.0.1 能直接连上，
+// 而 smtp.NewClient 握手失败时会把对端返回的那一行包进错误、经 last_error
+// 由投递历史接口回显，构成半盲读原语；「连接被拒 vs 超时」的耗时差异还能
+// 用来探测端口。拨号阶段是最终生效点，也覆盖 DNS 重绑定。
 func emailDial(ctx context.Context, addr, host string, implicitTLS bool) (*smtp.Client, error) {
-	d := &net.Dialer{Timeout: emailDialTimeout}
+	d := &net.Dialer{Timeout: emailDialTimeout, Control: blockInternalDial}
 	var conn net.Conn
 	var err error
 	if implicitTLS {

notify/mask.go

@@ -183,8 +183,19 @@ func rejectMaskedInContainers(incoming map[string]any, secretKeys []string) erro
 
 // sameConfigValue 比较两个配置值是否等价。用 JSON 序列化比较是为了顺带处理
 // 类型差异——前端提交的端口是 number，而库里读回来的是 float64，直接 == 会误判。
+//
+// 「空」必须先归一化再比较：空串与「键不存在」在这个配置模型里是同一个状态，
+// 因为 MergeConfig 把空串当显式清空、直接 delete 掉该键。不归一化的话，一个
+// 始终留空的可选目的地字段（Telegram 的 base_url 是唯一这样的字段：留空即用
+// 官方地址）会走成这条路径——
+//
+//	新建时存下 base_url:""  →  第一次保存被 MergeConfig 删键
+//	→ 第二次保存时 incoming 是 ""、stored 缺键，被判成「地址变了」
+//	→ 凭据是掩码值 → 400「目标地址已变更，请同时重新填写凭据字段」
+//
+// 此后每次保存都失败，除非用户重新粘贴一遍 Bot Token，而他什么都没改。
 func sameConfigValue(a, b any) bool {
-	if a == nil && b == nil {
+	if isBlankConfigValue(a) && isBlankConfigValue(b) {
 		return true
 	}
 	ra, errA := json.Marshal(a)
@@ -195,6 +206,17 @@ func sameConfigValue(a, b any) bool {
 	return string(ra) == string(rb)
 }
 
+// isBlankConfigValue 判定一个配置值是否为「空」。
+// 口径必须与 MergeConfig 的清空判定一致（strings.TrimSpace(s) == ""），
+// 否则会出现「MergeConfig 认为该删、sameConfigValue 认为有值」的夹缝。
+func isBlankConfigValue(v any) bool {
+	if v == nil {
+		return true
+	}
+	s, ok := v.(string)
+	return ok && strings.TrimSpace(s) == ""
+}
+
 // MergeConfig 把 incoming 合并到 stored 之上，用于更新渠道配置。
 //
 // 规则：

notify/mask_test.go

@@ -277,6 +277,78 @@ func TestPrepareConfigUpdatePortTypeTolerance(t *testing.T) {
 	}
 }
 
+// TestPrepareConfigUpdateSurvivesRepeatedSaveWithBlankDestination 覆盖「可留空的
+// 目的地字段」这条路径：Telegram 的 base_url 留空表示用官方 API 地址。
+//
+// 曾经这里会让渠道从第二次保存起永久保存失败：
+//
+//	新建时库里存下 base_url:""（创建路径直接存前端提交的 config，不走 MergeConfig）
+//	→ 第一次保存，MergeConfig 把空串当显式清空、delete 掉该键
+//	→ 第二次保存，incoming 仍是 ""、而 stored 里已经没这个键，被判成「地址变了」
+//	→ bot_token 是掩码回显值 → 400「目标地址已变更，请同时重新填写凭据字段」
+//
+// 用户什么都没改，却从此再也存不上，除非重新粘贴一遍 Bot Token。
+func TestPrepareConfigUpdateSurvivesRepeatedSaveWithBlankDestination(t *testing.T) {
+	stored := map[string]any{"bot_token": "123:ABC", "chat_id": "-100", "base_url": ""}
+
+	// 前端 buildConfig() 对该渠道的每个字段定义都提交一个值：凭据回填掩码，
+	// 空文本框提交空串。这里完整复现它的输出，而不是只提交「改动的键」。
+	submit := func() map[string]any {
+		return map[string]any{
+			"bot_token": MaskedValue("123:ABC"),
+			"chat_id":   "-100",
+			"base_url":  "",
+		}
+	}
+
+	// 第一次保存：只改了渠道名字，config 原样回传。
+	merged, err := PrepareConfigUpdate(KindTelegram, stored, submit())
+	if err != nil {
+		t.Fatalf("第一次保存被误拦: %v", err)
+	}
+	if _, ok := merged["base_url"]; ok {
+		t.Fatal("前提已变：空串应被 MergeConfig 删除——本用例要覆盖的正是「键消失之后」那一步")
+	}
+
+	// 第二次保存：提交内容与上次完全一致，用户什么都没改。
+	merged2, err := PrepareConfigUpdate(KindTelegram, merged, submit())
+	if err != nil {
+		t.Fatalf("第二次保存被误拦（用户什么都没改）: %v", err)
+	}
+	// 第三次，确认不是「只错一次」而是稳定可保存。
+	if _, err := PrepareConfigUpdate(KindTelegram, merged2, submit()); err != nil {
+		t.Fatalf("第三次保存被误拦: %v", err)
+	}
+	// 凭据必须一路保留下来，没有被空串逻辑连带清掉。
+	if got := merged2["bot_token"]; got != "123:ABC" {
+		t.Fatalf("Bot Token 应沿用原值，得到 %v", got)
+	}
+}
+
+// TestPrepareConfigUpdateStillGuardsBlankDestinationChanges 是上一条用例的配对
+// 断言：把空串与「键不存在」视为等价，**不能**连带放过真正的地址变更。
+// 这两个方向都是真实的凭据外发路径——Telegram 的 Bot Token 走在 URL 路径里，
+// 换了 base_url 就等于把 Token 送给新地址。
+func TestPrepareConfigUpdateStillGuardsBlankDestinationChanges(t *testing.T) {
+	// 方向一：从「空」（官方地址）换到自建地址。
+	official := map[string]any{"bot_token": "123:ABC", "chat_id": "-100"}
+	if _, err := PrepareConfigUpdate(KindTelegram, official, map[string]any{
+		"bot_token": MaskedValue("123:ABC"),
+		"base_url":  "https://tg-proxy.attacker.tld",
+	}); err == nil {
+		t.Fatal("从官方地址换到自建地址必须要求重新填写 Token")
+	}
+
+	// 方向二：把自建地址清空（= 换回官方 API）同样是地址变更。
+	proxied := map[string]any{"bot_token": "123:ABC", "base_url": "https://proxy.internal/bot"}
+	if _, err := PrepareConfigUpdate(KindTelegram, proxied, map[string]any{
+		"bot_token": MaskedValue("123:ABC"),
+		"base_url":  "",
+	}); err == nil {
+		t.Fatal("清空自建地址（换回官方 API）同样是地址变更，必须要求重新填写 Token")
+	}
+}
+
 func TestDestinationKeysDeclaredForEveryKind(t *testing.T) {
 	// 与 SecretKeys 同理：渠道若忘记声明目的地键，PrepareConfigUpdate 就保护不到它。
 	for kind, ch := range registry {

notify/ssrf_test.go

@@ -147,3 +147,47 @@ func TestValidateHTTPURLErrorNeverLeaksCredentials(t *testing.T) {
 	}
 	assertNoSecret(t, err.Error(), leakProbeToken)
 }
+
+// TestEmailDialGuardRejectsLoopbackByDefault 覆盖 SMTP 渠道的拨号守卫。
+//
+// 邮件渠道曾经用的是裸 net.Dialer，是整套 SSRF 防护里唯一的缺口：host 填成
+// 169.254.169.254 或 127.0.0.1 能直接连上，而 smtp.NewClient 握手失败时会把
+// 对端返回的那一行包进错误、经 last_error 由投递历史接口回显——正是其它渠道
+// 已经关掉的半盲读原语；「连接被拒 vs 超时」的耗时差异还能用来探测端口。
+//
+// 本包的 TestMain 全局打开了 AllowLocalTargetsEnv（大量用例用 127.0.0.1 上的
+// 假接收端），所以这个用例必须自己把它清掉——否则守卫在不在都会通过，
+// 这也正是缺口当初没被任何测试发现的原因。
+func TestEmailDialGuardRejectsLoopbackByDefault(t *testing.T) {
+	f := newFakeSMTP(t)
+	cfg := emailCfg(t, f, nil)
+
+	t.Setenv(AllowLocalTargetsEnv, "") // 关掉逃生口 = 默认行为
+	_, err := (emailChannel{}).Send(context.Background(), cfg, singleMsg())
+	if err == nil {
+		t.Fatal("默认不应允许把邮件投递到环回地址")
+	}
+	// 连接根本不该建立：守卫在 Control 钩子里拦下，EHLO 永远发不出去。
+	if f.sawCommand("EHLO") || f.sawCommand("HELO") {
+		t.Fatal("SMTP 会话已经建立——守卫没生效")
+	}
+	// 错误信息要能指导用户怎么放开（本机 postfix 中继是合法配置）。
+	if !strings.Contains(err.Error(), AllowLocalTargetsEnv) {
+		t.Errorf("拒绝信息应说明如何显式放开: %v", err)
+	}
+}
+
+// TestEmailDialGuardAllowsLoopbackWhenOptedIn 是配对的反向用例：显式打开后
+// 必须能正常投递。内网自建 SMTP / 本机中继是非常常见的部署，守卫不能一刀切。
+func TestEmailDialGuardAllowsLoopbackWhenOptedIn(t *testing.T) {
+	f := newFakeSMTP(t)
+	cfg := emailCfg(t, f, nil)
+
+	t.Setenv(AllowLocalTargetsEnv, "1")
+	if _, err := (emailChannel{}).Send(context.Background(), cfg, singleMsg()); err != nil {
+		t.Fatalf("显式放开后本机 SMTP 应可投递: %v", err)
+	}
+	if !f.sawCommand("EHLO") {
+		t.Fatal("未看到 EHLO——会话没真正建立")
+	}
+}

server/notifier.go

@@ -136,24 +136,50 @@ func (n *Notifier) step(ctx context.Context) {
 		if !ch.IsEnabled() {
 			continue
 		}
-		// 先问令牌桶这一轮还能发几条，再按这个数量去领——顺序不能反，
-		// 否则被限流挡下的投递已经消耗过重试次数。
-		allow := n.takeTokens(ch.ID, ch.RatePerMin, time.Now())
-		if allow > notifyMaxSendsPerChannelPerTick {
-			allow = notifyMaxSendsPerChannelPerTick
-		}
-		if allow <= 0 {
+		// 令牌桶的计量单位是**消息条数**（等价于 HTTP 请求数），不是漏洞条数。
+		// 实时模式下两者相同（一条漏洞一条消息）；汇总模式下一整批漏洞合成
+		// 一条消息，所以只消耗一个令牌。
+		//
+		// 两种模式都先问令牌桶、再按额度去领——顺序不能反，否则被限流挡下的
+		// 投递已经消耗过重试次数。
+		now := time.Now()
+		if ch.Mode == db.NotifyModeDigest {
+			tokens, claimLimit := digestTickPlan()
+			if n.takeTokens(ch.ID, ch.RatePerMin, tokens, now) <= 0 {
+				continue
+			}
+			n.stepDigest(ctx, ch, claimLimit, baseURL)
 			continue
 		}
-		if ch.Mode == db.NotifyModeDigest {
-			// 汇总也受同一份额度约束：领到的条数不能超过本轮令牌数。
-			n.stepDigest(ctx, ch, allow, baseURL)
+		allow := n.takeTokens(ch.ID, ch.RatePerMin, notifyMaxSendsPerChannelPerTick, now)
+		if allow <= 0 {
 			continue
 		}
 		n.stepRealtime(ctx, ch, allow, baseURL)
 	}
 }
 
+// digestTickPlan 返回汇总渠道本轮的令牌消耗与批次大小上界。
+//
+// 两个返回值是**两个不同的量纲**，这正是独立成函数的理由：
+//
+//   - tokens 是消息条数。一批漏洞合成一条消息、发一次 HTTP 请求，所以恒为 1。
+//     rate_per_min 因此仍然对 digest 生效（每分钟最多这么多条汇总消息）。
+//   - claimLimit 是这一批最多装几条漏洞。它只受内存上界约束，与请求预算无关。
+//
+// 曾经为了让 rate_per_min 对 digest 生效，把每轮请求预算
+// （notifyMaxSendsPerChannelPerTick，由租约倒推而来）直接当批次大小传下去。
+// 后果是 rate_per_min=20 的渠道在 3 秒的 tick 里只补到 1 个令牌，于是每条汇总
+// 消息只装 1 个漏洞——digest 退化成「带汇总文案的实时推送」，读者收到的是一串
+// 「近 30 分钟新增 1 个漏洞」，而 db.MaxDigestBatchSize 永不可达。
+//
+// 这个症状在端到端测试里不容易发现（现有用例都手动传一个够大的 limit 给
+// stepDigest，绕过了 step 里的额度计算），所以把决策收在这里由
+// TestDigestTickPlanDecouplesBatchSizeFromSendBudget 直接钉住。
+func digestTickPlan() (tokens, claimLimit int) {
+	return 1, db.MaxDigestBatchSize
+}
+
 // stepRealtime 领取并投递某渠道的实时任务，一条漏洞一条消息。
 func (n *Notifier) stepRealtime(ctx context.Context, ch *db.NotificationChannel, allow int, baseURL string) {
 	deliveries, err := n.pg.ClaimRealtimeDeliveries(ctx, ch.ID, allow, notifyLease)
@@ -422,14 +448,23 @@ func (n *Notifier) itemFor(ctx context.Context, snap notify.Snapshot, baseURL st
 	return item, nil
 }
 
-// takeTokens 从渠道令牌桶里取出本轮可投递的条数。
+// takeTokens 从渠道令牌桶里取走**最多 want 个**令牌，返回实际取到的数量。
 //
-// 桶容量为该渠道每分钟上限：积压时最多一次性冲这么多（平台规则允许），
-// 之后按恒定速率补充。ratePerMin<=0 表示不限流，返回一个有限但足够大的值，
-// 防止单轮循环被无限积压拖住。
-func (n *Notifier) takeTokens(channelID int64, ratePerMin int, now time.Time) int {
+// 一个令牌 = 一条消息（一次 HTTP 请求）。实时模式下调用方要几条就传几条；
+// 汇总模式下一整批漏洞只发一条消息，传 1。
+//
+// 桶容量为该渠道每分钟上限，按恒定速率补充。ratePerMin<=0 表示不限流，
+// 返回一个有限但足够大的值，防止单轮循环被无限积压拖住。
+//
+// want 这个上限是必需的：没有它就只能把桶整个抽空，而调用方自己还有每轮上限，
+// 多取的令牌既用不上、又在下次补充前凭空消失——攒下来的突发容量永远不可达，
+// 连「这一轮没有任何待发投递」都会照扣一笔。
+func (n *Notifier) takeTokens(channelID int64, ratePerMin, want int, now time.Time) int {
+	if want <= 0 {
+		return 0
+	}
 	if ratePerMin <= 0 {
-		return notifyUnlimitedBurstPerTick
+		return min(want, notifyUnlimitedBurstPerTick)
 	}
 	n.mu.Lock()
 	defer n.mu.Unlock()
@@ -446,7 +481,7 @@ func (n *Notifier) takeTokens(channelID int64, ratePerMin int, now time.Time) in
 	// 加一个极小 epsilon 再取整：令牌数是浮点累加出来的，分两次补满时
 	// 0.5 + 0.5 可能得到 0.9999999999，直接 int() 会被截成 0——
 	// 数学上已满的桶却取不出令牌。1e-9 远小于一个令牌，不会放过真正的欠额。
-	take := int(b.tokens + 1e-9)
+	take := min(int(b.tokens+1e-9), want)
 	if take <= 0 {
 		return 0
 	}

server/notify_api_test.go

@@ -856,28 +856,84 @@ func TestNotifyRateLimitDoesNotConsumeRetryBudget(t *testing.T) {
 	n := &Notifier{buckets: map[int64]*notifyBucket{}}
 	now := time.Now()
 	// 每分钟 1 条：满桶时最多 1 条。
-	if got := n.takeTokens(1, 1, now); got != 1 {
+	if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now); got != 1 {
 		t.Fatalf("满桶时每分钟 1 条应取 1 个令牌，得到 %d", got)
 	}
-	if got := n.takeTokens(1, 1, now.Add(time.Millisecond)); got != 0 {
+	if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(time.Millisecond)); got != 0 {
 		t.Fatalf("令牌耗尽后应立即返回 0，得到 %d", got)
 	}
-	if got := n.takeTokens(1, 1, now.Add(30*time.Second)); got != 0 {
+	if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(30*time.Second)); got != 0 {
 		t.Fatalf("半程不应补满一个令牌，得到 %d", got)
 	}
-	if got := n.takeTokens(1, 1, now.Add(time.Minute)); got != 1 {
+	if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(time.Minute)); got != 1 {
 		t.Fatalf("满一个周期应补回 1 个令牌，得到 %d", got)
 	}
 	// 不限流渠道走有限上限，避免单轮被无限积压拖住。
-	if got := n.takeTokens(2, 0, now); got != notifyUnlimitedBurstPerTick {
+	if got := n.takeTokens(2, 0, notifyUnlimitedBurstPerTick+10, now); got != notifyUnlimitedBurstPerTick {
 		t.Fatalf("不限流应返回每轮上限 %d，得到 %d", notifyUnlimitedBurstPerTick, got)
 	}
 	// 渠道之间的令牌桶互相独立。
-	if got := n.takeTokens(1, 1, now.Add(time.Millisecond)); got != 0 {
+	if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(time.Millisecond)); got != 0 {
 		t.Fatalf("渠道 1 的桶应仍然为空，得到 %d", got)
 	}
 }
 
+// TestNotifyTakeTokensKeepsUnusedTokens 锁住「只取 want 个」的语义。
+//
+// 曾经的实现把桶整个抽空后才由调用方截断，于是 rate=100/min 的渠道攒满桶、
+// 一轮只用 5 条，剩下 95 个令牌直接丢弃；渠道这一轮没有待发投递时同样照扣。
+// 结果是注释声称的「积压时可以一次性冲 rate_per_min 条」在任何情况下都做不到。
+func TestNotifyTakeTokensKeepsUnusedTokens(t *testing.T) {
+	n := &Notifier{buckets: map[int64]*notifyBucket{}}
+	now := time.Now()
+	// 桶初始为满（100），本轮只要 5 个。
+	if got := n.takeTokens(1, 100, 5, now); got != 5 {
+		t.Fatalf("want=5 时应恰好取 5 个令牌，得到 %d", got)
+	}
+	// 关键断言：余下的 95 个必须还在桶里，而不是被抽空丢弃。
+	// 不推进时间，确保取到的只可能来自存量而非补充。
+	if got := n.takeTokens(1, 100, 95, now); got != 95 {
+		t.Fatalf("剩余令牌应仍可取用（期望 95），得到 %d——桶被整轮抽空了", got)
+	}
+	if got := n.takeTokens(1, 100, 1, now); got != 0 {
+		t.Fatalf("桶已取尽，应返回 0，得到 %d", got)
+	}
+	// want<=0 不应扣减任何令牌（空轮不收费）。
+	n2 := &Notifier{buckets: map[int64]*notifyBucket{}}
+	if got := n2.takeTokens(1, 20, 0, now); got != 0 {
+		t.Fatalf("want=0 应返回 0，得到 %d", got)
+	}
+	if got := n2.takeTokens(1, 20, 20, now); got != 20 {
+		t.Fatalf("want=0 的那次不该消耗令牌，应仍可取满 20，得到 %d", got)
+	}
+}
+
+// TestDigestTickPlanDecouplesBatchSizeFromSendBudget 钉住汇总模式的两个量纲。
+//
+// 汇总批次的大小一旦跟每轮请求预算挂上，rate_per_min=20 的渠道就只能在每个
+// 3 秒 tick 里补到 1 个令牌，于是每条汇总消息只装 1 个漏洞——功能上等于没有
+// 汇总，而消息头部还写着「近 30 分钟新增 1 个漏洞」。这个退化不会报错，
+// 现有的端到端用例也看不出来（它们手动给 stepDigest 传一个够大的 limit，
+// 绕过了 step 里的额度计算），所以在这里直接断言决策本身。
+func TestDigestTickPlanDecouplesBatchSizeFromSendBudget(t *testing.T) {
+	tokens, claimLimit := digestTickPlan()
+	// 一批 = 一条消息 = 一次请求 = 一个令牌。令牌的单位是消息，不是漏洞。
+	if tokens != 1 {
+		t.Fatalf("汇总一批只发一条消息，应恰好消耗 1 个令牌，得到 %d", tokens)
+	}
+	if claimLimit != db.MaxDigestBatchSize {
+		t.Fatalf("汇总批次大小应为内存上界 db.MaxDigestBatchSize=%d，得到 %d",
+			db.MaxDigestBatchSize, claimLimit)
+	}
+	// 关键关系：批次大小必须远大于每轮请求预算。两者一旦同量级，
+	// 说明又把「发几条消息」和「一批装几条漏洞」混成了一个数。
+	if claimLimit <= notifyMaxSendsPerChannelPerTick {
+		t.Fatalf("汇总批次大小 %d 不应受每轮请求预算 %d 约束——"+
+			"请求预算是由租约倒推的「发几次请求」，与「一批装几条漏洞」是两个量纲",
+			claimLimit, notifyMaxSendsPerChannelPerTick)
+	}
+}
+
 // TestNotifyTickBudgetFitsWithinLease 是又一条防漂移断言。
 //
 // 单渠道每轮的投递条数上限（notifyMaxSendsPerChannelPerTick）是从租约时长倒推的：