mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-13 11:40:53 +08:00
BaaRai
809436e111
The IM renderers cut the summary of a single push at 120 characters and listed at most 3 assets, as fixed constants. Both are now global settings in the settings table, read by the delivery engine when it renders: notify_summary_max_runes 0-4000, default 120 notify_assets_max_shown 0-50, default 3 0 means no truncation and an unset key keeps 120 / 3. PUT /api/settings answers 400 for a value outside the bounds. The limits travel on notify.Message, so the notify package still does not touch the database. With no truncation, nothing bounded a single item in a Feishu card, and Feishu rejects an oversized request as a whole. feishuCard now truncates an item's text as the final backstop, the role TruncateBytes plays in markdownBody for DingTalk and WeCom. feishuMaxCardBytes goes from 24000 to 16000: Feishu custom bots cap the request body at 20 KB, which the old value exceeded. Refs #168.
1140 lines
44 KiB
Go
1140 lines
44 KiB
Go
package server
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/Autumn-27/artex/db"
|
||
"github.com/Autumn-27/artex/notify"
|
||
)
|
||
|
||
// This file covers the end-to-end behavior of the notification delivery feature:
|
||
// vulnerability persisted → event → fan-out → real HTTP send.
|
||
//
|
||
// One safety note: these cases do **not** call the global Notifier.step(); they
|
||
// only call stepRealtime/stepDigest on channels they created themselves. The
|
||
// reason is that step() iterates over all enabled channels in the database — on
|
||
// a development database with real DingTalk/WeCom bots configured, a global step
|
||
// would push the vulnerabilities produced during the test into those groups for
|
||
// real. Per-channel calls strictly confine the blast radius to the fake receivers
|
||
// created by the tests.
|
||
//
|
||
// Cleanup: when a case finishes, the events it produced (with deliveries deleted
|
||
// by cascade) and its channels are removed, leaving no backlog for real channels.
|
||
//
|
||
// Assertion policy: stepRealtime/stepDigest return nothing and log internally, so
|
||
// what is asserted here is **observable external behavior** (what the fake
|
||
// receiver got, what state the delivery rows landed in), not function return
|
||
// values — this is closer to the real call path than stubbing return values.
|
||
|
||
// notifyFixture is the shared fixture for the cases in this file.
|
||
type notifyFixture struct {
|
||
s *Server
|
||
pg *db.DB
|
||
request func(method, path, body string) *httptest.ResponseRecorder
|
||
n *Notifier
|
||
// Self-created task/exploration: cases record vulnerabilities here, isolated
|
||
// from other cases' data.
|
||
taskID int64
|
||
expID int64
|
||
// Events produced after cleanupMark are deleted during cleanup.
|
||
cleanupMark int64
|
||
}
|
||
|
||
func newNotifyFixture(t *testing.T) *notifyFixture {
|
||
t.Helper()
|
||
// All fake receivers in this file run on 127.0.0.1, while delivery rejects
|
||
// loopback addresses by default (SSRF protection for same-host services and
|
||
// cloud metadata). The tests explicitly open this switch; the
|
||
// "reject by default" guard behavior is covered by ssrf_test.go in the notify
|
||
// package.
|
||
t.Setenv(notify.AllowLocalTargetsEnv, "1")
|
||
s, _, request := trafficEvidenceServer(t)
|
||
pg := s.m.pg
|
||
|
||
// Create our own task: the task made by the shared fixture
|
||
// trafficEvidenceServer has no exploration id available, which recording a
|
||
// vulnerability requires.
|
||
task, err := s.m.CreateTask("notification delivery test", "delivery behavior verification", nil, 0, 0)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
taskID, err := strconv.ParseInt(task.ID, 10, 64)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { pg.Exec(`DELETE FROM tasks WHERE id=$1`, taskID) })
|
||
|
||
var mark int64
|
||
if err := pg.QueryRow(`SELECT COALESCE(max(id),0) FROM notification_events`).Scan(&mark); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// Make the fixture self-contained: mark all events that existed before the
|
||
// fixture was built as already fanned out in one go.
|
||
//
|
||
// Why this is necessary: FanOutPendingEvents is **global** and expands every
|
||
// unfanned event in the database onto all matching channels. The shared
|
||
// fixture trafficEvidenceServer itself records one vulnerability (exactly the
|
||
// initial finding it returns), and other cases may have leftovers too. Without
|
||
// isolation, these stray events would be fanned out onto this case's channels,
|
||
// making assertions like "there should be N deliveries" intermittently wrong —
|
||
// and the way they fail would depend on case execution order, which is harder
|
||
// to debug than a straight failure.
|
||
if _, err := pg.Exec(`UPDATE notification_events SET fanned_out = true WHERE id <= $1 AND NOT fanned_out`, mark); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
f := ¬ifyFixture{s: s, pg: pg, request: request, n: newNotifier(s), taskID: taskID, expID: task.ExpID, cleanupMark: mark}
|
||
t.Cleanup(func() {
|
||
if _, err := pg.Exec(`DELETE FROM notification_events WHERE id > $1`, f.cleanupMark); err != nil {
|
||
t.Logf("failed to clean up notification events: %v", err)
|
||
}
|
||
})
|
||
// The master switch must be on (other cases may have turned it off).
|
||
if err := pg.SetBool(settingNotifyEnabled, true); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return f
|
||
}
|
||
|
||
// record persists one vulnerability via the real evidence write path and returns
|
||
// the finding id. That path registers the notification event in the **same
|
||
// transaction** — which is exactly the hook point of this feature.
|
||
func (f *notifyFixture) record(t *testing.T, vulnclass, severity string) int64 {
|
||
t.Helper()
|
||
out, err := f.s.evidenceStore().Record(context.Background(), db.RecordFindingInput{
|
||
TaskID: f.taskID,
|
||
ExplorationID: f.expID,
|
||
Worker: "test",
|
||
VulnClass: vulnclass,
|
||
Name: vulnclass,
|
||
Severity: severity,
|
||
Summary: vulnclass + " summary",
|
||
Evidence: "poc",
|
||
}, nil)
|
||
if err != nil {
|
||
t.Fatalf("failed to record finding: %v", err)
|
||
}
|
||
return out.FindingID
|
||
}
|
||
|
||
// channel reads back the channel configuration (for per-channel stepX calls).
|
||
func (f *notifyFixture) channel(t *testing.T, id int64) *db.NotificationChannel {
|
||
t.Helper()
|
||
ch, err := f.pg.NotificationChannelByID(context.Background(), id)
|
||
if err != nil {
|
||
t.Fatalf("failed to read channel: %v", err)
|
||
}
|
||
return ch
|
||
}
|
||
|
||
// deliver fans out events and runs one delivery round for the specified channel
|
||
// only.
|
||
func (f *notifyFixture) deliver(t *testing.T, chID int64, baseURL string) {
|
||
t.Helper()
|
||
ctx := context.Background()
|
||
if _, _, err := f.pg.FanOutPendingEvents(ctx, 500); err != nil {
|
||
t.Fatalf("fan-out failed: %v", err)
|
||
}
|
||
f.n.stepRealtime(ctx, f.channel(t, chID), 50, baseURL)
|
||
}
|
||
|
||
// createChannel creates a channel through the HTTP API, also covering the API's
|
||
// own validation paths along the way.
|
||
func (f *notifyFixture) createChannel(t *testing.T, payload map[string]any) int64 {
|
||
t.Helper()
|
||
raw, _ := json.Marshal(payload)
|
||
r := f.request("POST", "/api/notify/channels", string(raw))
|
||
if r.Code != 200 {
|
||
t.Fatalf("failed to create channel %d: %s", r.Code, r.Body)
|
||
}
|
||
var res struct {
|
||
ID int64 `json:"id"`
|
||
}
|
||
if err := json.Unmarshal(r.Body.Bytes(), &res); err != nil || res.ID == 0 {
|
||
t.Fatalf("create channel returned an abnormal response: %s (%v)", r.Body, err)
|
||
}
|
||
t.Cleanup(func() { f.pg.Exec(`DELETE FROM notification_channels WHERE id=$1`, res.ID) })
|
||
return res.ID
|
||
}
|
||
|
||
// fakeWebhook is a fake receiver that records the request bodies it gets.
|
||
type fakeWebhook struct {
|
||
*httptest.Server
|
||
mu sync.Mutex
|
||
bodies []map[string]any
|
||
}
|
||
|
||
func newFakeWebhook(t *testing.T) *fakeWebhook {
|
||
t.Helper()
|
||
f := &fakeWebhook{}
|
||
f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
raw, _ := io.ReadAll(r.Body)
|
||
var body map[string]any
|
||
_ = json.Unmarshal(raw, &body)
|
||
f.mu.Lock()
|
||
f.bodies = append(f.bodies, body)
|
||
f.mu.Unlock()
|
||
w.Header().Set("Content-Type", "application/json")
|
||
_, _ = io.WriteString(w, `{"errcode":0,"errmsg":"ok"}`)
|
||
}))
|
||
t.Cleanup(f.Close)
|
||
return f
|
||
}
|
||
|
||
func (f *fakeWebhook) count() int {
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
return len(f.bodies)
|
||
}
|
||
|
||
func (f *fakeWebhook) body(t *testing.T, i int) map[string]any {
|
||
t.Helper()
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
if i >= len(f.bodies) {
|
||
t.Fatalf("fake receiver only got %d requests, cannot get request %d", len(f.bodies), i)
|
||
}
|
||
return f.bodies[i]
|
||
}
|
||
|
||
func (f *fakeWebhook) last(t *testing.T) map[string]any {
|
||
t.Helper()
|
||
if f.count() == 0 {
|
||
t.Fatal("fake receiver received no requests")
|
||
}
|
||
return f.body(t, f.count()-1)
|
||
}
|
||
|
||
// markdownText extracts the body text from the request body, tolerating the
|
||
// field-name differences between vendors: DingTalk markdown uses `text`,
|
||
// ActionCard uses `text`, WeCom markdown uses `content`.
|
||
func markdownText(t *testing.T, body map[string]any) string {
|
||
t.Helper()
|
||
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
|
||
}
|
||
}
|
||
}
|
||
t.Fatalf("no recognizable body text in request body: %v", body)
|
||
return ""
|
||
}
|
||
|
||
// agePendingBatch ages the channel's pending deliveries, used to test digest
|
||
// batches becoming due.
|
||
func (f *notifyFixture) agePendingBatch(t *testing.T, chID int64) {
|
||
t.Helper()
|
||
if _, err := f.pg.Exec(`UPDATE notification_deliveries SET created_at = now() - interval '2 hours'
|
||
WHERE channel_id=$1 AND state=$2`, chID, db.NotifyStatePending); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
|
||
func TestNotifyEndToEndRealtimeDelivery(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "realtime delivery",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
f.record(t, "SQL injection", "high")
|
||
f.deliver(t, chID, "")
|
||
|
||
if hook.count() != 1 {
|
||
t.Fatalf("expected 1 message sent, got %d", hook.count())
|
||
}
|
||
text := markdownText(t, hook.last(t))
|
||
for _, want := range []string{"SQL injection", "High", "Summary"} {
|
||
if !strings.Contains(text, want) {
|
||
t.Fatalf("message body is missing %q:\n%s", want, text)
|
||
}
|
||
}
|
||
// The delivery should transition to sent.
|
||
var pending int
|
||
if err := f.pg.QueryRow(`SELECT count(*) FROM notification_deliveries WHERE channel_id=$1 AND state <> $2`,
|
||
chID, db.NotifyStateSent).Scan(&pending); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if pending != 0 {
|
||
t.Fatalf("%d deliveries still not marked sent after delivery", pending)
|
||
}
|
||
}
|
||
|
||
func TestNotifyChannelAPIMasksSecretsAndPreservesOnUpdate(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "masking case",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": "https://oapi.dingtalk.com/robot/send?access_token=abc123456", "secret": "SECabcdef123456"},
|
||
})
|
||
|
||
r := f.request("GET", "/api/notify/channels", "")
|
||
if r.Code != 200 {
|
||
t.Fatalf("failed to list channels %d: %s", r.Code, r.Body)
|
||
}
|
||
if strings.Contains(r.Body.String(), "abc123456") || strings.Contains(r.Body.String(), "SECabcdef123456") {
|
||
t.Fatalf("API response leaked credentials: %s", r.Body)
|
||
}
|
||
var listed struct {
|
||
Channels []struct {
|
||
ID int64 `json:"id"`
|
||
Config map[string]any `json:"config"`
|
||
SecretKeys []string `json:"secret_keys"`
|
||
} `json:"channels"`
|
||
}
|
||
if err := json.Unmarshal(r.Body.Bytes(), &listed); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var mine *struct {
|
||
ID int64 `json:"id"`
|
||
Config map[string]any `json:"config"`
|
||
SecretKeys []string `json:"secret_keys"`
|
||
}
|
||
for i := range listed.Channels {
|
||
if listed.Channels[i].ID == chID {
|
||
mine = &listed.Channels[i]
|
||
}
|
||
}
|
||
if mine == nil {
|
||
t.Fatal("newly created channel did not appear in the list")
|
||
}
|
||
if !notify.IsMasked(fmt.Sprint(mine.Config["webhook"])) || !notify.IsMasked(fmt.Sprint(mine.Config["secret"])) {
|
||
t.Fatalf("credential fields should be masked values: %v", mine.Config)
|
||
}
|
||
if len(mine.SecretKeys) == 0 {
|
||
t.Fatal("the API should tell the frontend which fields are credentials")
|
||
}
|
||
|
||
// PATCH with only a rename + echoing back the masked credentials: the real
|
||
// credentials must be preserved unchanged.
|
||
body, _ := json.Marshal(map[string]any{
|
||
"name": "renamed",
|
||
"config": map[string]any{"webhook": fmt.Sprint(mine.Config["webhook"]), "secret": fmt.Sprint(mine.Config["secret"])},
|
||
})
|
||
if r := f.request("PATCH", fmt.Sprintf("/api/notify/channels/%d", chID), string(body)); r.Code != 200 {
|
||
t.Fatalf("update failed %d: %s", r.Code, r.Body)
|
||
}
|
||
cfg := f.channelConfig(t, chID)
|
||
if cfg["webhook"] != "https://oapi.dingtalk.com/robot/send?access_token=abc123456" {
|
||
t.Fatalf("masked echo overwrote the real credential: %v", cfg["webhook"])
|
||
}
|
||
if cfg["secret"] != "SECabcdef123456" {
|
||
t.Fatalf("masked echo overwrote the secret: %v", cfg["secret"])
|
||
}
|
||
if f.channel(t, chID).Name != "renamed" {
|
||
t.Fatal("name was not updated")
|
||
}
|
||
|
||
// Explicitly clearing the secret should take effect (as opposed to
|
||
// "echoing back the mask = keep unchanged").
|
||
body, _ = json.Marshal(map[string]any{"config": map[string]any{"secret": ""}})
|
||
if r := f.request("PATCH", fmt.Sprintf("/api/notify/channels/%d", chID), string(body)); r.Code != 200 {
|
||
t.Fatalf("clearing secret failed %d: %s", r.Code, r.Body)
|
||
}
|
||
if _, still := f.channelConfig(t, chID)["secret"]; still {
|
||
t.Fatal("empty string should clear the secret")
|
||
}
|
||
}
|
||
|
||
func (f *notifyFixture) channelConfig(t *testing.T, id int64) map[string]any {
|
||
t.Helper()
|
||
var cfg map[string]any
|
||
if err := json.Unmarshal(f.channel(t, id).Config, &cfg); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return cfg
|
||
}
|
||
|
||
func TestNotifyChannelAPICreateValidation(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
cases := []struct {
|
||
name string
|
||
payload map[string]any
|
||
wantSub string
|
||
}{
|
||
{"invalid kind", map[string]any{"name": "x", "kind": "nope", "config": map[string]any{}}, "invalid channel type"},
|
||
{"missing name", map[string]any{"kind": notify.KindDingTalk, "config": map[string]any{"webhook": "https://e.com/h"}}, "channel name is required"},
|
||
{"missing webhook", map[string]any{"name": "x", "kind": notify.KindDingTalk, "config": map[string]any{}}, "Webhook"},
|
||
{"invalid webhook scheme", map[string]any{"name": "x", "kind": notify.KindDingTalk, "config": map[string]any{"webhook": "file:///etc/passwd"}}, "invalid Webhook address"},
|
||
{"invalid mode", map[string]any{"name": "x", "kind": notify.KindDingTalk, "mode": "sometimes", "config": map[string]any{"webhook": "https://e.com/h"}}, "invalid delivery mode"},
|
||
}
|
||
for _, tc := range cases {
|
||
t.Run(tc.name, func(t *testing.T) {
|
||
raw, _ := json.Marshal(tc.payload)
|
||
r := f.request("POST", "/api/notify/channels", string(raw))
|
||
if r.Code != 400 {
|
||
t.Fatalf("expected 400, got %d: %s", r.Code, r.Body)
|
||
}
|
||
if !strings.Contains(r.Body.String(), tc.wantSub) {
|
||
t.Fatalf("error message should mention %q, got %s", tc.wantSub, r.Body)
|
||
}
|
||
})
|
||
}
|
||
if r := f.request("DELETE", "/api/notify/channels/99999999", ""); r.Code != 404 {
|
||
t.Fatalf("deleting a nonexistent channel should return 404, got %d", r.Code)
|
||
}
|
||
}
|
||
|
||
func TestNotifyFilterBlocksBelowThreshold(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "critical only",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
"filter": map[string]any{"min_severity": "critical"},
|
||
})
|
||
f.record(t, "low-risk issue", "low")
|
||
if _, _, err := f.pg.FanOutPendingEvents(context.Background(), 500); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var n int
|
||
if err := f.pg.QueryRow(`SELECT count(*) FROM notification_deliveries WHERE channel_id=$1`, chID).Scan(&n); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if n != 0 {
|
||
t.Fatalf("vulnerabilities below the threshold should not produce deliveries, got %d", n)
|
||
}
|
||
f.n.stepRealtime(context.Background(), f.channel(t, chID), 50, "")
|
||
if hook.count() != 0 {
|
||
t.Fatal("filtered vulnerabilities should not send messages")
|
||
}
|
||
}
|
||
|
||
func TestNotifyDigestBatchesMultipleFindingsIntoOneMessage(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "digest delivery",
|
||
"kind": notify.KindDingTalk,
|
||
"mode": db.NotifyModeDigest,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
for i := 0; i < 3; i++ {
|
||
f.record(t, fmt.Sprintf("digest vuln %d", i+1), "high")
|
||
}
|
||
ctx := context.Background()
|
||
if _, _, err := f.pg.FanOutPendingEvents(ctx, 500); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ch := f.channel(t, chID)
|
||
|
||
// Not due yet: do not send.
|
||
f.n.stepDigest(ctx, ch, 50, "")
|
||
if hook.count() != 0 {
|
||
t.Fatal("digest batch sent before it was due")
|
||
}
|
||
|
||
// After aging the batch: three findings merge into one message.
|
||
f.agePendingBatch(t, chID)
|
||
f.n.stepDigest(ctx, ch, 50, "")
|
||
if got := hook.count(); got != 1 {
|
||
t.Fatalf("three findings should be digested into one message, actually sent %d", got)
|
||
}
|
||
text := markdownText(t, hook.last(t))
|
||
if !strings.Contains(text, "in the last 30 minutes") || !strings.Contains(text, "3 new vulnerabilities") {
|
||
t.Fatalf("digest message is missing the count/time-window wording:\n%s", text)
|
||
}
|
||
for i := 1; i <= 3; i++ {
|
||
if !strings.Contains(text, fmt.Sprintf("digest vuln %d", i)) {
|
||
t.Fatalf("digest message is missing entry %d:\n%s", i, text)
|
||
}
|
||
}
|
||
// The same batch should share one batch_id.
|
||
var distinct, total int
|
||
if err := f.pg.QueryRow(`SELECT count(DISTINCT batch_id), count(*) FROM notification_deliveries WHERE channel_id=$1`, chID).Scan(&distinct, &total); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if total != 3 || distinct != 1 {
|
||
t.Fatalf("three deliveries should share one batch_id, got distinct=%d total=%d", distinct, total)
|
||
}
|
||
}
|
||
|
||
func TestNotifyDisabledChannelDoesNotSend(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "disabled channel",
|
||
"kind": notify.KindDingTalk,
|
||
"enabled": false,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
f.record(t, "vulnerability while disabled", "critical")
|
||
if _, _, err := f.pg.FanOutPendingEvents(context.Background(), 500); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var n int
|
||
if err := f.pg.QueryRow(`SELECT count(*) FROM notification_deliveries WHERE channel_id=$1`, chID).Scan(&n); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if n != 0 {
|
||
t.Fatalf("a disabled channel should not produce deliveries, got %d", n)
|
||
}
|
||
}
|
||
|
||
func TestNotifyStatusChangeDelivery(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "status change subscription",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
"filter": map[string]any{"on_status_change": true},
|
||
})
|
||
finding := f.record(t, "status change case", "high")
|
||
r := f.request("PATCH", fmt.Sprintf("/api/exploration/findings/%d", finding), `{"status":"fixed"}`)
|
||
if r.Code != 200 {
|
||
t.Fatalf("failed to change status %d: %s", r.Code, r.Body)
|
||
}
|
||
f.deliver(t, chID, "")
|
||
|
||
// There should be two messages: the "fixed" one is the status change; the
|
||
// finding_created one may also go out in the same round. The status-change one
|
||
// is actually created later, but do not rely on ordering — search them all.
|
||
found := false
|
||
for i := 0; i < hook.count(); i++ {
|
||
text := markdownText(t, hook.body(t, i))
|
||
if strings.Contains(text, "Status change") && strings.Contains(text, "Fixed") {
|
||
found = true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("no message containing \"Status change → Fixed\" received (%d messages total)", hook.count())
|
||
}
|
||
}
|
||
|
||
func TestNotifyStatusChangeSuppressedByDefault(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "not subscribed to status changes",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
finding := f.record(t, "unsubscribed change", "high")
|
||
if r := f.request("PATCH", fmt.Sprintf("/api/exploration/findings/%d", finding), `{"status":"false_positive"}`); r.Code != 200 {
|
||
t.Fatalf("failed to change status %d: %s", r.Code, r.Body)
|
||
}
|
||
if _, _, err := f.pg.FanOutPendingEvents(context.Background(), 500); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var n int
|
||
if err := f.pg.QueryRow(`SELECT count(*) FROM notification_deliveries d
|
||
JOIN notification_events e ON e.id = d.event_id
|
||
WHERE d.channel_id=$1 AND e.kind=$2`, chID, notify.EventFindingStatusChanged).Scan(&n); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if n != 0 {
|
||
t.Fatalf("a channel not subscribed to status changes should not receive status-change deliveries, got %d", n)
|
||
}
|
||
}
|
||
|
||
func TestNotifyTestMessageEndpoint(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "test send",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
if r := f.request("POST", fmt.Sprintf("/api/notify/channels/%d/test", chID), ""); r.Code != 200 {
|
||
t.Fatalf("test send failed %d: %s", r.Code, r.Body)
|
||
}
|
||
if hook.count() != 1 {
|
||
t.Fatalf("fake receiver should have received 1 test message, got %d", hook.count())
|
||
}
|
||
// The test message must be recognizable as a test at a glance; it must not be
|
||
// mistaken for a real vulnerability.
|
||
if text := markdownText(t, hook.last(t)); !strings.Contains(text, "Test") {
|
||
t.Fatalf("the test message should identify itself as a test: %s", text)
|
||
}
|
||
// When the configuration is broken, the channel's raw error should be returned
|
||
// to the user as-is.
|
||
badID := f.createChannel(t, map[string]any{
|
||
"name": "bad address",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": "http://127.0.0.1:1/hook"},
|
||
})
|
||
if r := f.request("POST", fmt.Sprintf("/api/notify/channels/%d/test", badID), ""); r.Code != 502 {
|
||
t.Fatalf("failed delivery should return 502, got %d: %s", r.Code, r.Body)
|
||
}
|
||
}
|
||
|
||
func TestNotifyDeliveriesHistoryAndRetry(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
// Point at an address that always fails, to produce failed deliveries.
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "failing retry",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": "http://127.0.0.1:1/hook"},
|
||
})
|
||
f.record(t, "delivery that fails", "high")
|
||
ctx := context.Background()
|
||
if _, _, err := f.pg.FanOutPendingEvents(ctx, 500); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ch := f.channel(t, chID)
|
||
// Keep delivering until the retry budget is exhausted.
|
||
for i := 0; i < db.MaxNotifyAttempts; i++ {
|
||
f.n.stepRealtime(ctx, ch, 50, "")
|
||
if _, err := f.pg.Exec(`UPDATE notification_deliveries SET next_attempt_at = now() - interval '1 minute' WHERE channel_id=$1`, chID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
var state string
|
||
if err := f.pg.QueryRow(`SELECT state FROM notification_deliveries WHERE channel_id=$1`, chID).Scan(&state); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if state != db.NotifyStateFailed {
|
||
t.Fatalf("should be failed after retries are exhausted, got %s", state)
|
||
}
|
||
|
||
r := f.request("GET", fmt.Sprintf("/api/notify/deliveries?channel_id=%d&state=failed", chID), "")
|
||
if r.Code != 200 {
|
||
t.Fatalf("failed to query history %d: %s", r.Code, r.Body)
|
||
}
|
||
var hist struct {
|
||
Deliveries []struct {
|
||
ID int64 `json:"id"`
|
||
State string `json:"state"`
|
||
LastError string `json:"last_error"`
|
||
Attempts int `json:"attempts"`
|
||
Title string `json:"title"`
|
||
} `json:"deliveries"`
|
||
Total int `json:"total"`
|
||
}
|
||
if err := json.Unmarshal(r.Body.Bytes(), &hist); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if hist.Total != 1 || len(hist.Deliveries) != 1 {
|
||
t.Fatalf("should find 1 failed delivery, got total=%d len=%d", hist.Total, len(hist.Deliveries))
|
||
}
|
||
if hist.Deliveries[0].LastError == "" {
|
||
t.Fatal("the history should carry the failure reason, otherwise users cannot troubleshoot")
|
||
}
|
||
if hist.Deliveries[0].Attempts < db.MaxNotifyAttempts {
|
||
t.Fatalf("the attempt count should be recorded, got %d", hist.Deliveries[0].Attempts)
|
||
}
|
||
if hist.Deliveries[0].Title != "delivery that fails" {
|
||
t.Fatalf("the history should include the vulnerability title, got %q", hist.Deliveries[0].Title)
|
||
}
|
||
|
||
// Manual resend: it should go back to pending with the counter reset.
|
||
if r := f.request("POST", fmt.Sprintf("/api/notify/deliveries/%d/retry", hist.Deliveries[0].ID), ""); r.Code != 200 {
|
||
t.Fatalf("resend failed %d: %s", r.Code, r.Body)
|
||
}
|
||
var attempts int
|
||
if err := f.pg.QueryRow(`SELECT state, attempts FROM notification_deliveries WHERE id=$1`, hist.Deliveries[0].ID).Scan(&state, &attempts); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if state != db.NotifyStatePending || attempts != 0 {
|
||
t.Fatalf("after resend it should be pending with attempts=0, got %s/%d", state, attempts)
|
||
}
|
||
}
|
||
|
||
func TestNotifyMetaAndSettingsRoundTrip(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
r := f.request("GET", "/api/notify/meta", "")
|
||
if r.Code != 200 {
|
||
t.Fatalf("meta failed: %s", r.Body)
|
||
}
|
||
var meta struct {
|
||
Kinds []struct {
|
||
Kind string `json:"kind"`
|
||
SecretKeys []string `json:"secret_keys"`
|
||
} `json:"kinds"`
|
||
}
|
||
if err := json.Unmarshal(r.Body.Bytes(), &meta); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(meta.Kinds) != len(notify.Kinds()) {
|
||
t.Fatalf("meta should list all %d channels, got %d", len(notify.Kinds()), len(meta.Kinds))
|
||
}
|
||
for _, k := range meta.Kinds {
|
||
if len(k.SecretKeys) == 0 {
|
||
t.Errorf("channel %s did not report credential fields", k.Kind)
|
||
}
|
||
}
|
||
|
||
// Round trip of the three global settings. The trailing slash should be
|
||
// normalized away, otherwise back-links would be built as "//function/...".
|
||
if r := f.request("PUT", "/api/settings", `{"notify_public_base_url":"https://artex.example.com/","notify_digest_interval_min":15,"notify_enabled":true}`); r.Code != 200 {
|
||
t.Fatalf("failed to write settings %d: %s", r.Code, r.Body)
|
||
}
|
||
t.Cleanup(func() {
|
||
f.pg.Exec(`DELETE FROM settings WHERE key IN ($1,$2)`, settingNotifyPublicBaseURL, settingNotifyDigestMinutes)
|
||
})
|
||
payload := f.s.settingsPayload()
|
||
if payload["notify_public_base_url"] != "https://artex.example.com" {
|
||
t.Fatalf("back-link address was not normalized: %v", payload["notify_public_base_url"])
|
||
}
|
||
if payload["notify_digest_interval_min"] != 15 {
|
||
t.Fatalf("digest interval did not take effect: %v", payload["notify_digest_interval_min"])
|
||
}
|
||
|
||
// Invalid values should be rejected.
|
||
for _, body := range []string{
|
||
`{"notify_public_base_url":"ftp://x"}`,
|
||
`{"notify_digest_interval_min":0}`,
|
||
`{"notify_digest_interval_min":99999}`,
|
||
} {
|
||
if r := f.request("PUT", "/api/settings", body); r.Code != 400 {
|
||
t.Errorf("%s should return 400, got %d", body, r.Code)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestNotifyLimitSettingsRoundTrip covers the two render-limit keys through the
|
||
// settings API and the settings table: unset reads as the 120 / 3 defaults, a
|
||
// stored 0 reads back as 0 (no truncation) and not as the default, and
|
||
// out-of-range values are rejected without touching the stored value.
|
||
func TestNotifyLimitSettingsRoundTrip(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
clear := func() {
|
||
f.pg.Exec(`DELETE FROM settings WHERE key IN ($1,$2)`, settingNotifySummaryMaxRunes, settingNotifyAssetsMaxShown)
|
||
}
|
||
clear()
|
||
t.Cleanup(clear)
|
||
expect := func(step string, summary, assets int) {
|
||
t.Helper()
|
||
payload := f.s.settingsPayload()
|
||
if payload["notify_summary_max_runes"] != summary || payload["notify_assets_max_shown"] != assets {
|
||
t.Fatalf("%s: settings payload should report %d / %d, got %v / %v", step, summary, assets,
|
||
payload["notify_summary_max_runes"], payload["notify_assets_max_shown"])
|
||
}
|
||
if got := notifyLimits(f.pg); got != (notify.Limits{SummaryRunes: summary, AssetsShown: assets}) {
|
||
t.Fatalf("%s: the delivery engine should read %d / %d, got %+v", step, summary, assets, got)
|
||
}
|
||
}
|
||
expect("unset", notify.DefaultMaxSummaryRunes, notify.DefaultMaxAssetsShown)
|
||
|
||
for _, tc := range []struct{ summary, assets int }{{500, 10}, {0, 0}, {4000, 50}} {
|
||
body := fmt.Sprintf(`{"notify_summary_max_runes":%d,"notify_assets_max_shown":%d}`, tc.summary, tc.assets)
|
||
if r := f.request("PUT", "/api/settings", body); r.Code != 200 {
|
||
t.Fatalf("%s failed %d: %s", body, r.Code, r.Body)
|
||
}
|
||
expect(body, tc.summary, tc.assets)
|
||
for key, want := range map[string]int{settingNotifySummaryMaxRunes: tc.summary, settingNotifyAssetsMaxShown: tc.assets} {
|
||
if v, ok, err := f.pg.GetSetting(key); err != nil || !ok || v != strconv.Itoa(want) {
|
||
t.Fatalf("%s: settings row %s should hold %d, got %q (ok=%v err=%v)", body, key, want, v, ok, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
for _, body := range []string{
|
||
`{"notify_summary_max_runes":-1}`,
|
||
`{"notify_summary_max_runes":4001}`,
|
||
`{"notify_assets_max_shown":-1}`,
|
||
`{"notify_assets_max_shown":51}`,
|
||
} {
|
||
if r := f.request("PUT", "/api/settings", body); r.Code != 400 {
|
||
t.Errorf("%s should return 400, got %d", body, r.Code)
|
||
}
|
||
}
|
||
expect("after rejected writes", 4000, 50)
|
||
}
|
||
|
||
// TestNotifySummaryLimitReachesDelivery is the end-to-end evidence that the
|
||
// stored limit reaches the message actually sent: with the key unset the summary
|
||
// is cut at 120 characters as before, and once it is set to 0 the same summary
|
||
// goes out whole.
|
||
func TestNotifySummaryLimitReachesDelivery(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
clear := func() { f.pg.Exec(`DELETE FROM settings WHERE key=$1`, settingNotifySummaryMaxRunes) }
|
||
clear()
|
||
t.Cleanup(clear)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "summary limit",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
summary := strings.Repeat("Z", 300)
|
||
record := func(name string) {
|
||
t.Helper()
|
||
if _, err := f.s.evidenceStore().Record(context.Background(), db.RecordFindingInput{
|
||
TaskID: f.taskID,
|
||
ExplorationID: f.expID,
|
||
Worker: "test",
|
||
VulnClass: name,
|
||
Name: name,
|
||
Severity: "high",
|
||
Summary: summary,
|
||
Evidence: "poc",
|
||
}, nil); err != nil {
|
||
t.Fatalf("failed to record finding: %v", err)
|
||
}
|
||
}
|
||
|
||
record("summary limit unset")
|
||
f.deliver(t, chID, "")
|
||
text := markdownText(t, hook.last(t))
|
||
if !strings.Contains(text, strings.Repeat("Z", 119)+"…") || strings.Contains(text, strings.Repeat("Z", 120)) {
|
||
t.Fatalf("with the key unset the summary should be cut at 120 characters:\n%s", text)
|
||
}
|
||
|
||
if r := f.request("PUT", "/api/settings", `{"notify_summary_max_runes":0}`); r.Code != 200 {
|
||
t.Fatalf("failed to write settings %d: %s", r.Code, r.Body)
|
||
}
|
||
record("summary limit zero")
|
||
f.deliver(t, chID, "")
|
||
if hook.count() != 2 {
|
||
t.Fatalf("expected 2 messages sent, got %d", hook.count())
|
||
}
|
||
if text := markdownText(t, hook.last(t)); !strings.Contains(text, summary) {
|
||
t.Fatalf("with the limit at 0 the whole summary should be sent:\n%s", text)
|
||
}
|
||
}
|
||
|
||
// TestNotifyDeepLinkUsesPublicBaseURL covers back-link building: when
|
||
// public_base_url is configured, single messages must use the ActionCard with a
|
||
// button, and the link must point to the vulnerability detail page.
|
||
func TestNotifyDeepLinkUsesPublicBaseURL(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "deep link",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
finding := f.record(t, "vulnerability with deep link", "high")
|
||
f.deliver(t, chID, "https://artex.example.com")
|
||
|
||
body := hook.last(t)
|
||
card, _ := body["actionCard"].(map[string]any)
|
||
if card == nil {
|
||
t.Fatalf("with a back-link an ActionCard should be used, got msgtype=%v", body["msgtype"])
|
||
}
|
||
want := fmt.Sprintf("https://artex.example.com/function/findings/detail?id=%d", finding)
|
||
if card["singleURL"] != want {
|
||
t.Fatalf("wrong back-link\nwant %s\ngot %v", want, card["singleURL"])
|
||
}
|
||
}
|
||
|
||
// TestNotifyNoDeepLinkWithoutBaseURL is the reverse coverage: with no external
|
||
// address configured, no broken link should be produced (e.g. pointing at
|
||
// localhost or a relative path); it should fall back to plain markdown.
|
||
func TestNotifyNoDeepLinkWithoutBaseURL(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "no deep link",
|
||
"kind": notify.KindDingTalk,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
f.record(t, "vulnerability without deep link", "high")
|
||
f.deliver(t, chID, "")
|
||
|
||
body := hook.last(t)
|
||
if body["msgtype"] != "markdown" {
|
||
t.Fatalf("markdown should be sent when no external address is configured, got %v", body["msgtype"])
|
||
}
|
||
if text := markdownText(t, body); strings.Contains(text, "View details") {
|
||
t.Fatalf("no detail link should appear when no external address is configured:\n%s", text)
|
||
}
|
||
}
|
||
|
||
// TestNotifyDigestSegmentsAndDefersRemainder is the end-to-end evidence for the
|
||
// "silent loss" fix.
|
||
//
|
||
// A digest message is bounded by the channel's length limit (WeCom: 4096 bytes).
|
||
// When a batch does not fit, it must be split **on whole entries**: the ones that
|
||
// fit in this message are marked sent, and the rest go back to the queue for the
|
||
// next message. A previous implementation marked the entire batch as succeeded —
|
||
// the truncated entries were neither in the message nor in the failure list,
|
||
// while the delivery history still showed success; the vulnerabilities were just
|
||
// gone.
|
||
//
|
||
// Four things are asserted: ① only the number of entries actually included is
|
||
// marked; ② the rest remain pending; ③ the deferred entries **did not consume
|
||
// retry attempts**; ④ running another round sends the remainder (no deadlock).
|
||
func TestNotifyDigestSegmentsAndDefersRemainder(t *testing.T) {
|
||
f := newNotifyFixture(t)
|
||
hook := newFakeWebhook(t)
|
||
// Use WeCom: its 4096-byte markdown limit is the tightest of the six channels.
|
||
chID := f.createChannel(t, map[string]any{
|
||
"name": "segmented digest",
|
||
"kind": notify.KindWeCom,
|
||
"mode": db.NotifyModeDigest,
|
||
"config": map[string]any{"webhook": hook.URL},
|
||
})
|
||
const total = 60
|
||
// Make the titles long, so 60 entries far exceed 4096 bytes and segmentation
|
||
// is guaranteed.
|
||
longName := strings.Repeat("very long vulnerability name", 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("only one message should be sent, got %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("some entries should have been marked as sent")
|
||
}
|
||
if pending == 0 {
|
||
t.Fatalf("a batch of %d entries cannot all fit into 4096 bytes, some should remain pending; sent=%d", total, sent)
|
||
}
|
||
if sent+pending != total {
|
||
t.Fatalf("entry counts do not add up: sent=%d pending=%d total=%d (neither sent nor pending = lost)", sent, pending, total)
|
||
}
|
||
// The message body must state honestly how many entries are not included in
|
||
// this one.
|
||
if text := markdownText(t, hook.last(t)); !strings.Contains(text, "remaining") {
|
||
t.Fatalf("the message should state that more entries are not included in this one:\n%.400s", text)
|
||
}
|
||
|
||
// Deferred entries must not consume the retry budget: claiming already
|
||
// optimistically incremented attempts by +1, and deferring must undo it.
|
||
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("deferred entries must not consume retry attempts (otherwise they would be marked failed after a few rounds), got attempts=%d", maxAttempts)
|
||
}
|
||
|
||
// Keep running until convergence. What is asserted is **eventual full
|
||
// delivery** plus the fact that it really took multiple rounds — this is
|
||
// stronger than "the second round finishes it": it proves segmentation neither
|
||
// deadlocks nor drops the remaining entries.
|
||
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("segmented delivery does not converge: after %d rounds there are still %d entries outstanding", rounds, undelivered)
|
||
}
|
||
before := hook.count()
|
||
f.n.stepDigest(ctx, ch, 50, "")
|
||
if hook.count() == before {
|
||
t.Fatalf("round %d made no progress; the remaining %d entries would be stuck forever", rounds, undelivered)
|
||
}
|
||
}
|
||
if rounds < 2 {
|
||
t.Fatalf("one 4096-byte message cannot hold %d long-titled vulnerabilities; it should take multiple rounds, but only %d were used", total, rounds)
|
||
}
|
||
// Every round after the first should be a **pure continuation**, with no entry
|
||
// rejected by the channel.
|
||
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("the fake receiver always returns success, so there should be no failed entries, got %d", failed)
|
||
}
|
||
}
|
||
|
||
// TestNotifyBackoffTableMatchesAttemptBudget is a drift-prevention assertion.
|
||
//
|
||
// The retry budget (db.MaxNotifyAttempts) and the backoff sequence table
|
||
// (notifyBackoff) live in two different packages: the former is the state
|
||
// machine's policy, the latter the engine's execution cadence. If only one of
|
||
// them is changed — say the budget is raised to 5 while forgetting to add backoff
|
||
// tiers — the code does not error; the 4th and 5th retries would just reuse the
|
||
// last tier's interval, showing up as "retry cadence inexplicably slows down",
|
||
// and it would be hard to trace back here. Asserting the two lengths match
|
||
// surfaces this kind of drift in CI.
|
||
func TestNotifyBackoffTableMatchesAttemptBudget(t *testing.T) {
|
||
if len(notifyBackoff) != db.MaxNotifyAttempts {
|
||
t.Fatalf("the number of backoff tiers (%d) does not match the max attempt count (%d) — changing one requires changing the other",
|
||
len(notifyBackoff), db.MaxNotifyAttempts)
|
||
}
|
||
// Backoff intervals must be monotonically non-decreasing, otherwise retries
|
||
// would get more and more urgent, aggravating rate limiting instead.
|
||
for i := 1; i < len(notifyBackoff); i++ {
|
||
if notifyBackoff[i] < notifyBackoff[i-1] {
|
||
t.Fatalf("backoff intervals must be monotonically non-decreasing: tier %d %v < tier %d %v",
|
||
i, notifyBackoff[i], i-1, notifyBackoff[i-1])
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestNotifyRateLimitDoesNotConsumeRetryBudget pins down the "take tokens first,
|
||
// then claim" order. If reversed (claim first, then discard), deliveries blocked
|
||
// by rate limiting would already have counted one attempts increment, and the
|
||
// budget would be burned by pure waiting, ending in failed.
|
||
func TestNotifyRateLimitDoesNotConsumeRetryBudget(t *testing.T) {
|
||
// Only the token bucket itself is tested; no Server is needed (and none should
|
||
// be built for it).
|
||
n := &Notifier{buckets: map[int64]*notifyBucket{}}
|
||
now := time.Now()
|
||
// 1 per minute: with a full bucket, at most 1.
|
||
if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now); got != 1 {
|
||
t.Fatalf("with a full bucket at 1/min it should take 1 token, got %d", got)
|
||
}
|
||
if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(time.Millisecond)); got != 0 {
|
||
t.Fatalf("should return 0 immediately once tokens are exhausted, got %d", got)
|
||
}
|
||
if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(30*time.Second)); got != 0 {
|
||
t.Fatalf("half a period should not refill a full token, got %d", got)
|
||
}
|
||
if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(time.Minute)); got != 1 {
|
||
t.Fatalf("a full period should refill 1 token, got %d", got)
|
||
}
|
||
// A channel with no rate limit gets a finite cap, so a single tick is not
|
||
// dragged down by unlimited backlog.
|
||
if got := n.takeTokens(2, 0, notifyUnlimitedBurstPerTick+10, now); got != notifyUnlimitedBurstPerTick {
|
||
t.Fatalf("no rate limit should return the per-tick cap %d, got %d", notifyUnlimitedBurstPerTick, got)
|
||
}
|
||
// Token buckets are independent between channels.
|
||
if got := n.takeTokens(1, 1, notifyMaxSendsPerChannelPerTick, now.Add(time.Millisecond)); got != 0 {
|
||
t.Fatalf("channel 1's bucket should still be empty, got %d", got)
|
||
}
|
||
}
|
||
|
||
// TestNotifyTakeTokensKeepsUnusedTokens pins down the "take only want" semantics.
|
||
//
|
||
// A previous implementation drained the entire bucket and let the caller
|
||
// truncate, so a channel with rate=100/min would fill its bucket, use 5 in one
|
||
// round, and have the remaining 95 tokens simply discarded; a channel with no
|
||
// pending deliveries in a round was charged the same way. As a result, the
|
||
// "burst the full rate_per_min at once when there is a backlog" promised by the
|
||
// comment was impossible under any circumstances.
|
||
func TestNotifyTakeTokensKeepsUnusedTokens(t *testing.T) {
|
||
n := &Notifier{buckets: map[int64]*notifyBucket{}}
|
||
now := time.Now()
|
||
// The bucket starts full (100); this round only wants 5.
|
||
if got := n.takeTokens(1, 100, 5, now); got != 5 {
|
||
t.Fatalf("with want=5 exactly 5 tokens should be taken, got %d", got)
|
||
}
|
||
// The key assertion: the remaining 95 must still be in the bucket, not drained
|
||
// and discarded. Time is not advanced, ensuring what is taken can only come
|
||
// from the stock, not from refills.
|
||
if got := n.takeTokens(1, 100, 95, now); got != 95 {
|
||
t.Fatalf("the remaining tokens should still be available (want 95), got %d — the bucket was drained for the whole round", got)
|
||
}
|
||
if got := n.takeTokens(1, 100, 1, now); got != 0 {
|
||
t.Fatalf("the bucket is drained, should return 0, got %d", got)
|
||
}
|
||
// want<=0 must not deduct any tokens (an empty round is free).
|
||
n2 := &Notifier{buckets: map[int64]*notifyBucket{}}
|
||
if got := n2.takeTokens(1, 20, 0, now); got != 0 {
|
||
t.Fatalf("want=0 should return 0, got %d", got)
|
||
}
|
||
if got := n2.takeTokens(1, 20, 20, now); got != 20 {
|
||
t.Fatalf("the want=0 call should not have consumed any tokens, the full 20 should still be available, got %d", got)
|
||
}
|
||
}
|
||
|
||
// TestDigestTickPlanDecouplesBatchSizeFromSendBudget pins down the two dimensions
|
||
// of digest mode.
|
||
//
|
||
// Once the digest batch size is coupled to the per-tick request budget, a channel
|
||
// with rate_per_min=20 can only refill 1 token per 3-second tick, so every digest
|
||
// message carries just 1 vulnerability — functionally equivalent to having no
|
||
// digest at all, while the message header still says "1 new vulnerability in the
|
||
// last 30 minutes". This regression does not error, and the existing end-to-end
|
||
// cases cannot see it either (they pass a large enough limit to stepDigest
|
||
// manually, bypassing the allowance calculation in step), so the decision itself
|
||
// is asserted here directly.
|
||
func TestDigestTickPlanDecouplesBatchSizeFromSendBudget(t *testing.T) {
|
||
tokens, claimLimit := digestTickPlan()
|
||
// One batch = one message = one request = one token. Tokens are metered in
|
||
// messages, not vulnerabilities.
|
||
if tokens != 1 {
|
||
t.Fatalf("a digest batch sends only one message, so it should consume exactly 1 token, got %d", tokens)
|
||
}
|
||
if claimLimit != db.MaxDigestBatchSize {
|
||
t.Fatalf("the digest batch size should be the memory bound db.MaxDigestBatchSize=%d, got %d",
|
||
db.MaxDigestBatchSize, claimLimit)
|
||
}
|
||
// The key relationship: the batch size must be far larger than the per-tick
|
||
// request budget. If the two end up on the same scale, "how many messages to
|
||
// send" and "how many vulnerabilities a batch holds" have been conflated into
|
||
// one number again.
|
||
if claimLimit <= notifyMaxSendsPerChannelPerTick {
|
||
t.Fatalf("digest batch size %d must not be bound by the per-tick request budget %d — "+
|
||
"the request budget is \"how many requests to send\", derived backwards from the lease, a different dimension from \"how many vulnerabilities a batch holds\"",
|
||
claimLimit, notifyMaxSendsPerChannelPerTick)
|
||
}
|
||
}
|
||
|
||
// TestNotifyTickBudgetFitsWithinLease is another drift-prevention assertion.
|
||
//
|
||
// The per-tick delivery cap for a single channel
|
||
// (notifyMaxSendsPerChannelPerTick) is derived backwards from the lease duration:
|
||
// the worst-case duration of one tick's serial deliveries must be < the lease,
|
||
// otherwise the lease on the later rows expires before they are sent, and in a
|
||
// multi-instance deployment the peer would reclaim them and resend. These three
|
||
// constants live in different places, and changing any one of them could break
|
||
// the relationship without any error — so it is pinned down here.
|
||
func TestNotifyTickBudgetFitsWithinLease(t *testing.T) {
|
||
worst := time.Duration(notifyMaxSendsPerChannelPerTick) * notifySendTimeout
|
||
if worst >= notifyLease {
|
||
t.Fatalf("the worst-case duration of one tick for a single channel %v must not reach or exceed the lease %v"+
|
||
" (notifyMaxSendsPerChannelPerTick=%d × notifySendTimeout=%v) — "+
|
||
"changing any of these three constants requires checking the other two",
|
||
worst, notifyLease, notifyMaxSendsPerChannelPerTick, notifySendTimeout)
|
||
}
|
||
}
|
||
|
||
// TestNotifyLimitValue pins how a stored render limit is resolved. The case that
|
||
// matters is "0": it is a real value (no truncation) and must not be read as
|
||
// "unset", while anything unparseable or outside the bounds falls back to the
|
||
// default instead of reaching the renderer.
|
||
func TestNotifyLimitValue(t *testing.T) {
|
||
const def, max = 120, 4000
|
||
cases := []struct {
|
||
v string
|
||
ok bool
|
||
want int
|
||
}{
|
||
{"", false, def},
|
||
{"0", true, 0},
|
||
{" 500 ", true, 500},
|
||
{"4000", true, max},
|
||
{"4001", true, def},
|
||
{"-1", true, def},
|
||
{"abc", true, def},
|
||
{"", true, def},
|
||
}
|
||
for _, tc := range cases {
|
||
if got := notifyLimitValue(tc.v, tc.ok, def, max); got != tc.want {
|
||
t.Errorf("notifyLimitValue(%q, %v) = %d, want %d", tc.v, tc.ok, got, tc.want)
|
||
}
|
||
}
|
||
}
|