package alerts_test import ( "bytes" "context" "encoding/json" "fmt" "io" "log/slog" "net/http" "net/http/httptest" "net/netip" "net/url" "reflect" "slices" "strings" "sync" "testing" "testing/synctest" "time" "sneak.berlin/go/smallwebwaf/internal/alerts" ) // The tests run in a synctest bubble, where the time package runs on a // clock of the test's own, which starts at 2000-01-01T00:00:00Z, the start // of an hour: a wait lasts exactly as long as it should, however slowly // the test process runs, and synctest.Wait returns once the queue has // done all it can before time passes. The stand-ins for the webhook, // Slack and ntfy answer without the network, since a request waiting on // the network would keep that clock from moving on. const ( // webhookURL is where the alerts are posted, slackURL the Slack // incoming webhook, and ntfyURL the ntfy topic. ntfyToken is the ntfy // token of the tests that set one. webhookURL = "https://alerts.example/smallwebwaf?team=ops" slackURL = "https://hooks.slack.example/services/T0123/B4567/abcdef" ntfyURL = "https://ntfy.example/smallwebwaf-alerts" ntfyToken = "tk_0123456789abcdefghijklmnopq" // instance is the instance name every alert gives. instance = "fsn1app1/gitea" // started is when each test starts, as an alert gives it, and // anHourOn an hour later. started = "2000-01-01T00:00:00Z" anHourOn = "2000-01-01T01:00:00Z" // cooldown is the cooldown of most tests, the default. cooldown = 15 * time.Minute ) func TestAlertIsPostedAsJSONWithItsFieldsAndTheHeaders(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.WebhookHeaders = http.Header{ "Authorization": {"Bearer 0123456789abcdef"}, "X-Team": {"ops"}, } webhook, q := start(t, params) q.Raise(alerts.Alert{ Event: alerts.EventBan, Client: netip.MustParseAddr("203.0.113.9"), Netblock: netip.MustParsePrefix("203.0.113.0/24"), Country: "DE", Reason: "requests per minute over the limit of 1000", Detail: map[string]any{"cause": "limit", "ban_expires": anHourOn}, }) synctest.Wait() got := webhook.received() if len(got) != 1 { t.Fatalf("the webhook had %d requests, want 1", len(got)) } if got[0].method != http.MethodPost || got[0].url != webhookURL { t.Errorf("request %s %s, want POST %s", got[0].method, got[0].url, webhookURL) } for name, want := range map[string]string{ "Content-Type": "application/json", "Authorization": "Bearer 0123456789abcdef", "X-Team": "ops", } { if got[0].header.Get(name) != want { t.Errorf("header %s is %q, want %q", name, got[0].header.Get(name), want) } } wantAlert(t, got[0].alert, map[string]any{ "instance": instance, "time": started, "event": "ban", "client": "203.0.113.9", "netblock": "203.0.113.0/24", "asn": "", "as_name": "", "country": "DE", "reason": "requests per minute over the limit of 1000", "detail": map[string]any{"cause": "limit", "ban_expires": anHourOn}, "suppressed_repeats": float64(0), }) wantCounts(t, q, 1, 0, 0, 0) }) } func TestOnlyTheChosenEventsAreSent(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.Events = []string{alerts.EventSourceFailure, alerts.EventFileError} webhook, q := start(t, params) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) q.Raise(alerts.Alert{Event: alerts.EventFileError, Reason: "a file error"}) q.Raise(alerts.Alert{Event: alerts.EventPermanentBan, Netblock: netblock(1)}) synctest.Wait() wantEvents(t, webhook, alerts.EventFileError) wantCounts(t, q, 1, 0, 0, 0) }) } func TestWouldSendOnlyForTheChosenEvents(t *testing.T) { t.Parallel() params := newParams() params.Events = []string{alerts.EventSourceFailure, alerts.EventFileError} q := alerts.New(params) if q.WouldSend(alerts.EventBan, netblock(1)) || q.WouldSend(alerts.EventPermanentBan, netblock(1)) { t.Error("a ban alert would be sent, though SWWAF_ALERT_EVENTS leaves it out") } if !q.WouldSend(alerts.EventFileError, netip.Prefix{}) { t.Error("a file_error alert would not be sent") } } func TestRaiseDoesNothingWithoutADestination(t *testing.T) { t.Parallel() params := newParams() params.WebhookURL = nil q := alerts.New(params) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) // No alert waits, no cooldown has started, and the hour counts none. want := alerts.State{ Cooldowns: []alerts.Cooldown{}, Hour: alerts.Hour{HeldBack: map[string]int{}}, Waiting: map[string][]alerts.Alert{}, } if got := q.Snapshot(); !reflect.DeepEqual(got, want) { t.Errorf("state %+v, want %+v", got, want) } } func TestWouldSendNothingWithoutADestination(t *testing.T) { t.Parallel() params := newParams() params.WebhookURL = nil q := alerts.New(params) if q.WouldSend(alerts.EventBan, netblock(1)) { t.Error("an alert would be sent with no webhook set") } } func TestWouldSendWithOnlySlackOrOnlyNtfySet(t *testing.T) { t.Parallel() onlySlack := newParams() onlySlack.WebhookURL = nil onlySlack.SlackURL = parseURL(slackURL) onlyNtfy := newParams() onlyNtfy.WebhookURL = nil onlyNtfy.NtfyURL = parseURL(ntfyURL) for setting, params := range map[string]alerts.Params{ "SWWAF_ALERT_SLACK_WEBHOOK_URL": onlySlack, "SWWAF_ALERT_NTFY_URL": onlyNtfy, } { if !alerts.New(params).WouldSend(alerts.EventBan, netblock(1)) { t.Errorf("a ban alert would not be sent with only %s set", setting) } } } func TestRepeatWithinTheCooldownIsHeldBackAndCountedInTheNext(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { webhook, q := start(t, newParams()) raise := func(event string, n int) { q.Raise(alerts.Alert{Event: event, Netblock: netblock(n)}) } raise(alerts.EventBan, 1) // The same event on the same netblock is a repeat; another netblock // or another event is not. time.Sleep(time.Minute) raise(alerts.EventBan, 1) raise(alerts.EventBan, 2) raise(alerts.EventPermanentBan, 1) time.Sleep(cooldown - time.Minute - time.Nanosecond) raise(alerts.EventBan, 1) // Once the cooldown has run out, the next one is sent with the // count of those held back. time.Sleep(time.Nanosecond) raise(alerts.EventBan, 1) // And starts the cooldown again. time.Sleep(time.Minute) raise(alerts.EventBan, 1) synctest.Wait() got := webhook.received() wantEvents(t, webhook, alerts.EventBan, alerts.EventBan, alerts.EventPermanentBan, alerts.EventBan) for i, want := range []struct { netblock int repeats float64 }{{1, 0}, {2, 0}, {1, 0}, {1, 2}} { alert := got[i].alert if alert["netblock"] != netblock(want.netblock).String() || alert["suppressed_repeats"] != want.repeats { t.Errorf("alert %d is for %v with %v repeats, want %s with %v", i, alert["netblock"], alert["suppressed_repeats"], netblock(want.netblock), want.repeats) } } wantCounts(t, q, 4, 0, 3, 0) }) } func TestFileErrorAndSourceFailureRepeatOnlyForTheSameFileOrSource(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() webhook, q := start(t, params) fileError := func(file string) alerts.Alert { return alerts.Alert{ Event: alerts.EventFileError, Detail: map[string]any{"file": file, "error": "line 2: an error"}, } } sourceFailure := func(source string) alerts.Alert { return alerts.Alert{ Event: alerts.EventSourceFailure, Detail: map[string]any{"source": source}, } } // Another file, or another source, is no repeat. q.Raise(fileError("/rules.d/50-a.rules")) q.Raise(fileError("/rules.d/50-b.rules")) q.Raise(fileError("/rules.d/50-a.rules")) q.Raise(sourceFailure("geojs")) q.Raise(sourceFailure("abuseipdb")) q.Raise(sourceFailure("geojs")) synctest.Wait() // Each alert is named by its file, or its source. got := make([]string, 0, len(webhook.received())) for _, request := range webhook.received() { detail, _ := request.alert["detail"].(map[string]any) file, _ := detail["file"].(string) source, _ := detail["source"].(string) got = append(got, file+source) } want := []string{ "/rules.d/50-a.rules", "/rules.d/50-b.rules", "geojs", "abuseipdb", } if !slices.Equal(got, want) { t.Errorf("the webhook was sent alerts for %v, want %v", got, want) } wantCounts(t, q, 4, 0, 2, 0) // alerts.json keeps each file's cooldown: a new queue holds back // the next for the first file, and sends the one for a third. after := alerts.New(params) after.Load(roundTrip(t, q.Snapshot())) after.Raise(fileError("/rules.d/50-a.rules")) after.Raise(fileError("/rules.d/50-c.rules")) waiting := after.Snapshot().Waiting[alerts.DestinationWebhook] if len(waiting) != 1 || waiting[0].Detail["file"] != "/rules.d/50-c.rules" { t.Errorf("after loading, alerts wait %+v, want the one for 50-c.rules", waiting) } }) } func TestNoCooldownSendsEveryRepeat(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.Cooldown = 0 webhook, q := start(t, params) for range 3 { q.Raise(alerts.Alert{Event: alerts.EventFileError}) time.Sleep(time.Minute) } synctest.Wait() wantEvents(t, webhook, alerts.EventFileError, alerts.EventFileError, alerts.EventFileError) wantCounts(t, q, 3, 0, 0, 0) }) } func TestAlertsPastTheHourlyLimitAreRolledIntoOneSummary(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.MaxPerHour = 2 webhook, q := start(t, params) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(2)}) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(3)}) q.Raise(alerts.Alert{Event: alerts.EventPermanentBan, Netblock: netblock(4)}) q.Raise(alerts.Alert{Event: alerts.EventFileError}) // The summary is sent as the hour ends, and not before. time.Sleep(time.Hour - time.Nanosecond) synctest.Wait() wantEvents(t, webhook, alerts.EventBan, alerts.EventBan) time.Sleep(time.Nanosecond) synctest.Wait() wantEvents(t, webhook, alerts.EventBan, alerts.EventBan, alerts.EventSummary) summary := webhook.received()[2].alert wantAlert(t, summary, map[string]any{ "instance": instance, "time": anHourOn, "event": "summary", "client": "", "netblock": "", "asn": "", "as_name": "", "country": "", "reason": "3 alerts held back in the hour from 2000-01-01T00:00:00Z, " + "past the 2 an hour SWWAF_ALERT_MAX_PER_HOUR allows", "detail": map[string]any{ "hour": started, "count": float64(3), "events": map[string]any{ "ban": float64(1), "permanent_ban": float64(1), "file_error": float64(1), }, }, "suppressed_repeats": float64(0), }) // The next hour sends alerts again, and, with none held back, ends // without a summary. q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(5)}) time.Sleep(time.Hour) synctest.Wait() wantEvents(t, webhook, alerts.EventBan, alerts.EventBan, alerts.EventSummary, alerts.EventBan) wantCounts(t, q, 4, 0, 3, 0) }) } func TestRepeatsBeforeAnAlertPastTheHourlyLimitAreGivenByTheNextSent(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.MaxPerHour = 1 webhook, q := start(t, params) raise := func() { q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) } // The hour's one alert, and two repeats the cooldown holds back. raise() raise() raise() // Once the cooldown has run out, the next is past the hourly limit. time.Sleep(cooldown) raise() // The next hour's first alert gives the two repeats, and the summary // the alert past the limit. time.Sleep(time.Hour - cooldown) synctest.Wait() raise() synctest.Wait() wantEvents(t, webhook, alerts.EventBan, alerts.EventSummary, alerts.EventBan) got := webhook.received() if len(got) == 3 { detail, _ := got[1].alert["detail"].(map[string]any) repeats := got[2].alert["suppressed_repeats"] if detail["count"] != float64(1) || repeats != float64(2) { t.Errorf("the summary counts %v alerts, and the last alert gives %v "+ "repeats, want 1 and 2", detail["count"], repeats) } } wantCounts(t, q, 3, 0, 3, 0) }) } func TestFailedRequestIsSentAgainWithBackoff(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() log := &lockedBuffer{} params.ProcessLog = slog.New(slog.NewJSONHandler(log, nil)) webhook, q := start(t, params) webhook.set(failing) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) // A second after the first failure, then twice as long after each // further one, up to a minute. time.Sleep(200 * time.Second) synctest.Wait() after := make([]time.Duration, 0, len(webhook.received())) for _, request := range webhook.received() { after = append(after, request.at.Sub(midnight())) } want := []time.Duration{ 0, time.Second, 3 * time.Second, 7 * time.Second, 15 * time.Second, 31 * time.Second, 63 * time.Second, 123 * time.Second, 183 * time.Second, } if !slices.Equal(after, want) { t.Errorf("requests at %v, want %v", after, want) } wantCounts(t, q, 0, int64(len(want)), 0, 0) if !strings.Contains(log.String(), `"msg":"sending an alert to SWWAF_ALERT_WEBHOOK_URL failed"`) { t.Errorf("process log %q names no failure", log.String()) } // Once the webhook answers, the alert is sent, and leaves the // queue. webhook.set(answering) time.Sleep(time.Minute) synctest.Wait() got := webhook.received() if last := got[len(got)-1]; !last.answered || last.alert["netblock"] != netblock(1).String() { t.Errorf("the last request was not the alert, answered") } wantCounts(t, q, 1, int64(len(want)), 0, 0) waiting := q.Snapshot().Waiting[alerts.DestinationWebhook] if len(waiting) != 0 { t.Errorf("%d alerts still wait, want none", len(waiting)) } }) } func TestRefusedAlertIsGivenUpAndTheNextSent(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() log := &lockedBuffer{} params.ProcessLog = slog.New(slog.NewJSONHandler(log, nil)) webhook, q := start(t, params) // 429 and 408 are failures, and the alert is sent again; 400 refuses // it, and it is given up. webhook.set(http.StatusTooManyRequests) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) synctest.Wait() webhook.set(http.StatusRequestTimeout) time.Sleep(time.Second) synctest.Wait() webhook.set(refusing) time.Sleep(2 * time.Second) synctest.Wait() // The next alert is sent at once. webhook.set(answering) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(2)}) time.Sleep(time.Minute) synctest.Wait() // Each request, by when it was sent, and the netblock of its alert. got := make([]string, 0, len(webhook.received())) for _, request := range webhook.received() { block, _ := request.alert["netblock"].(string) got = append(got, request.at.Sub(midnight()).String()+" "+block) } want := []string{ "0s " + netblock(1).String(), "1s " + netblock(1).String(), "3s " + netblock(1).String(), "3s " + netblock(2).String(), } if !slices.Equal(got, want) { t.Errorf("requests %v, want %v", got, want) } wantCounts(t, q, 1, 3, 0, 1) if !strings.Contains(log.String(), `"msg":"gave up an alert SWWAF_ALERT_WEBHOOK_URL refused"`) { t.Errorf("process log %q names no alert given up", log.String()) } }) } func TestFailedRequestIsLoggedWithoutTheURL(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() log := &lockedBuffer{} params.ProcessLog = slog.New(slog.NewJSONHandler(log, nil)) webhook, q := start(t, params) webhook.set(hanging) // The request is abandoned after 10 seconds, with an error from the // HTTP client, which names the URL. q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) time.Sleep(11 * time.Second) synctest.Wait() logged := log.String() if !strings.Contains(logged, `"msg":"sending an alert to SWWAF_ALERT_WEBHOOK_URL failed"`) || strings.Contains(logged, "alerts.example") || strings.Contains(logged, "team=ops") { t.Errorf("process log %q names no failure, or names the URL", logged) } }) } func TestFullQueueDropsTheOldestAndRaiseNeverWaits(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.MaxPerHour = 0 webhook, q := start(t, params) webhook.set(hanging) // The webhook does not answer the first alert, while one more alert // than the queue holds is raised: none waits, and the oldest, the // one the webhook was sent, is dropped. for n := range alerts.QueueSize + 1 { q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(n)}) if n == 0 { synctest.Wait() } } if took := time.Since(midnight()); took != 0 { t.Errorf("raising the alerts took %s, want no time", took) } wantCounts(t, q, 0, 0, 0, 1) waiting := q.Snapshot().Waiting[alerts.DestinationWebhook] if len(waiting) != alerts.QueueSize || waiting[0].Netblock != netblock(1) { t.Fatalf("%d alerts wait, the first for %s, want %d, the first for %s", len(waiting), waiting[0].Netblock, alerts.QueueSize, netblock(1)) } // The request is abandoned after 10 seconds, and the webhook, which // answers again, is sent the others, in order, a second later. webhook.set(answering) time.Sleep(11 * time.Second) synctest.Wait() got := webhook.received() if len(got) != alerts.QueueSize+1 || got[0].alert["netblock"] != netblock(0).String() { t.Fatalf("the webhook had %d requests, want %d, the first for %s", len(got), alerts.QueueSize+1, netblock(0)) } for i, request := range got[1:] { if request.alert["netblock"] != netblock(i+1).String() { t.Fatalf("request %d is for %v, want %s", i+1, request.alert["netblock"], netblock(i+1)) } } wantCounts(t, q, alerts.QueueSize, 1, 0, 1) }) } func TestStateLoadedIntoANewQueueCarriesOn(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := newParams() params.MaxPerHour = 1 before := alerts.New(params) // Not sent: Run is not running. The repeat is held back by the // cooldown, and the file error past the hourly limit. before.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) before.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) before.Raise(alerts.Alert{Event: alerts.EventFileError}) time.Sleep(time.Minute) webhook, after := start(t, params) after.Load(roundTrip(t, before.Snapshot())) // The new queue sends the alert waiting, holds back the repeat as // the cooldown still runs, and sends the summary of the hour. after.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) synctest.Wait() wantEvents(t, webhook, alerts.EventBan) time.Sleep(time.Hour) synctest.Wait() wantEvents(t, webhook, alerts.EventBan, alerts.EventSummary) detail, _ := webhook.received()[1].alert["detail"].(map[string]any) if detail["count"] != float64(1) { t.Errorf("the summary counts %v alerts, want 1", detail["count"]) } // The cooldown has run out, and the next one gives both repeats. after.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) synctest.Wait() got := webhook.received() if repeats := got[len(got)-1].alert["suppressed_repeats"]; repeats != float64(2) { t.Errorf("the last alert gives %v repeats, want 2", repeats) } }) } func TestSlackAndNtfyAreSentAMessageForEachEvent(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) params.WebhookURL = nil standIns, q := startAll(t, params) for _, each := range anAlertForEachEvent() { q.Raise(each.alert) } synctest.Wait() want := anAlertForEachEvent() slack := standIns[alerts.DestinationSlack].received() ntfy := standIns[alerts.DestinationNtfy].received() if len(slack) != len(want) || len(ntfy) != len(want) { t.Fatalf("Slack was sent %d messages and ntfy %d, want %d each", len(slack), len(ntfy), len(want)) } for i, each := range want { wantSlackMessage(t, slack[i], "*"+each.title+"*\n"+each.text) wantNtfyMessage(t, ntfy[i], each.title, each.priorityAndTag, each.text) } }) } func TestSlackAndNtfyAreSentTheSummaryAndTheRepeatsHeldBack(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) params.WebhookURL = nil params.MaxPerHour = 1 standIns, q := startAll(t, params) ban := alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1), Reason: "a ban"} // The hour's one alert, a repeat of it the cooldown holds back, and // an alert past the limit; once the hour has ended, its summary, and // the next alert, which gives the repeat. q.Raise(ban) q.Raise(ban) q.Raise(alerts.Alert{Event: alerts.EventFileError, Reason: "a file error"}) time.Sleep(time.Hour) synctest.Wait() q.Raise(ban) synctest.Wait() slack := standIns[alerts.DestinationSlack].received() ntfy := standIns[alerts.DestinationNtfy].received() if len(slack) != 3 || len(ntfy) != 3 { t.Fatalf("Slack was sent %d messages and ntfy %d, want 3 each", len(slack), len(ntfy)) } const summary = "1 alerts held back in the hour from 2000-01-01T00:00:00Z, " + "past the 1 an hour SWWAF_ALERT_MAX_PER_HOUR allows" wantSlackMessage(t, slack[1], "*"+instance+": summary*\n"+summary) wantNtfyMessage(t, ntfy[1], instance+": summary", "default bar_chart", summary) wantSlackMessage(t, slack[2], "*"+instance+": ban*\na ban\nnetblock: 203.0.113.1/32\nsuppressed repeats: 1") wantNtfyMessage(t, ntfy[2], instance+": ban", "default no_entry", "a ban\nnetblock: 203.0.113.1/32\nsuppressed repeats: 1") }) } func TestSlackMessageEscapesAmpersandsAndAngleBrackets(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) params.Instance = "app<1>" standIns, q := startAll(t, params) q.Raise(alerts.Alert{ Event: alerts.EventFileError, Reason: " & ", }) synctest.Wait() got := standIns[alerts.DestinationSlack].received() if len(got) != 1 { t.Fatalf("Slack was sent %d messages, want 1", len(got)) } wantSlackMessage(t, got[0], "*app<1>: file_error*\n"+ "<!channel> & <https://x.example|y>") wantBodies(t, standIns[alerts.DestinationNtfy], " & ") }) } func TestNtfyTokenIsSentToNtfyAlone(t *testing.T) { t.Parallel() for _, tc := range []struct { name, token string want []string }{ {"with a token", ntfyToken, []string{"Bearer " + ntfyToken}}, {"without one", "", nil}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) params.NtfyToken = tc.token standIns, q := startAll(t, params) q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)}) synctest.Wait() for destination, want := range map[string][]string{ alerts.DestinationWebhook: nil, alerts.DestinationSlack: nil, alerts.DestinationNtfy: tc.want, } { got := standIns[destination].received() if len(got) != 1 { t.Fatalf("%s was sent %d requests, want 1", destination, len(got)) } authorization := got[0].header.Values("Authorization") if !slices.Equal(authorization, want) { t.Errorf("%s was sent Authorization %q, want %q", destination, authorization, want) } } }) }) } } func TestDestinationThatDoesNotAnswerHoldsUpNeitherOther(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) log := &lockedBuffer{} params.ProcessLog = slog.New(slog.NewJSONHandler(log, nil)) standIns, q := startAll(t, params) standIns[alerts.DestinationSlack].set(hanging) for n := range 3 { q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(n)}) } synctest.Wait() // With no time passed, the webhook and ntfy have taken every alert, // while Slack has not answered the first. for destination, want := range map[string]alerts.Counts{ alerts.DestinationWebhook: {Sent: 3}, alerts.DestinationSlack: {}, alerts.DestinationNtfy: {Sent: 3}, } { wantDestinationCounts(t, q, destination, want) } if got := standIns[alerts.DestinationSlack].received(); len(got) != 1 { t.Errorf("Slack was sent %d requests, want 1", len(got)) } // Slack's request is abandoned after 10 seconds, and logged without // its URL; Slack, which answers again, is sent every alert a second // later. standIns[alerts.DestinationSlack].set(answering) time.Sleep(11 * time.Second) synctest.Wait() wantDestinationCounts(t, q, alerts.DestinationSlack, alerts.Counts{Sent: 3, Failed: 1}) logged := log.String() if !strings.Contains(logged, `"msg":"sending an alert to SWWAF_ALERT_SLACK_WEBHOOK_URL failed"`) || strings.Contains(logged, "hooks.slack.example") || strings.Contains(logged, "T0123") { t.Errorf("process log %q names no failure, or names the URL", logged) } }) } func TestDropsAreCountedForEachDestination(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) params.MaxPerHour = 0 standIns, q := startAll(t, params) standIns[alerts.DestinationSlack].set(hanging) standIns[alerts.DestinationNtfy].set(refusing) // Each alert is raised once each destination has done all it can // with those before it. for n := range alerts.QueueSize + 1 { q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(n)}) synctest.Wait() } // Slack, which does not answer the first alert, drops it from its // full queue for the last; ntfy refuses each, which is given up; and // the webhook takes every one. const all = alerts.QueueSize + 1 for destination, want := range map[string]alerts.Counts{ alerts.DestinationWebhook: {Sent: all}, alerts.DestinationSlack: {Dropped: 1}, alerts.DestinationNtfy: {Failed: all, Dropped: all}, } { wantDestinationCounts(t, q, destination, want) } }) } func TestEachDestinationIsSentOnlyTheAlertsWaitingForIt(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { params := withSlackAndNtfy(newParams()) params.WebhookURL = nil standIns, q := startAll(t, params) waiting := func(reason string) alerts.Alert { return alerts.Alert{ Instance: instance, Time: midnight(), Event: alerts.EventFileError, Reason: reason, } } // As read from alerts.json. The alert waiting for the webhook, which // is not set, is dropped. q.Load(roundTrip(t, alerts.State{Waiting: map[string][]alerts.Alert{ alerts.DestinationWebhook: {waiting("first")}, alerts.DestinationSlack: {waiting("second"), waiting("third")}, alerts.DestinationNtfy: {waiting("fourth")}, }})) synctest.Wait() wantBodies(t, standIns[alerts.DestinationSlack], `{"text":"*fsn1app1/gitea: file_error*\nsecond"}`, `{"text":"*fsn1app1/gitea: file_error*\nthird"}`) wantBodies(t, standIns[alerts.DestinationNtfy], "fourth") // alerts.json then lists Slack and ntfy alone, with no alert waiting. want := map[string][]alerts.Alert{ alerts.DestinationSlack: {}, alerts.DestinationNtfy: {}, } if got := q.Snapshot().Waiting; !reflect.DeepEqual(got, want) { t.Errorf("alerts waiting %v, want %v", got, want) } }) } // How a stand-in for a destination answers: with a status, or, hanging, // not at all, until the request is abandoned. const ( answering = http.StatusNoContent failing = http.StatusServiceUnavailable refusing = http.StatusBadRequest hanging = 0 ) // standIn is a stand-in for a destination. It notes each request it is // sent. type standIn struct { mu sync.Mutex answers int requests []post } // post is a request a destination was sent: when, its method, URL and // headers, its body, and that body read as a JSON object, which for the // webhook is the alert it carried, and whether the destination answered // it with a 2xx status. type post struct { at time.Time method string url string header http.Header body string alert map[string]any answered bool } // RoundTrip has the stand-in answer req, in place of the network. A // request abandoned before the stand-in answers fails, as over the // network. func (s *standIn) RoundTrip(req *http.Request) (*http.Response, error) { answer := httptest.NewRecorder() s.ServeHTTP(answer, req) _ = req.Body.Close() err := req.Context().Err() if err != nil { return nil, err } return answer.Result(), nil } // ServeHTTP notes the request, and answers it as the stand-in is set to. func (s *standIn) ServeHTTP(w http.ResponseWriter, r *http.Request) { body, _ := io.ReadAll(r.Body) var alert map[string]any _ = json.Unmarshal(body, &alert) s.mu.Lock() answers := s.answers s.requests = append(s.requests, post{ at: time.Now(), method: r.Method, url: r.URL.String(), header: r.Header.Clone(), body: string(body), alert: alert, answered: answers == answering, }) s.mu.Unlock() if answers == hanging { <-r.Context().Done() } else { w.WriteHeader(answers) } } // set sets how the stand-in answers: with the status answers, or hanging. func (s *standIn) set(answers int) { s.mu.Lock() defer s.mu.Unlock() s.answers = answers } // received returns the requests the stand-in has been sent so far. func (s *standIn) received() []post { s.mu.Lock() defer s.mu.Unlock() return slices.Clone(s.requests) } // lockedBuffer is a buffer the process log can write to while the test // reads it. type lockedBuffer struct { mu sync.Mutex buf bytes.Buffer } // Write adds p to the buffer. func (b *lockedBuffer) Write(p []byte) (int, error) { b.mu.Lock() defer b.mu.Unlock() return b.buf.Write(p) } // String returns what was written. func (b *lockedBuffer) String() string { b.mu.Lock() defer b.mu.Unlock() return b.buf.String() } // newParams returns the Params of most tests: the webhook at webhookURL, // every event, the default cooldown and hourly limit, and the bubble's // clock in UTC. func newParams() alerts.Params { return alerts.Params{ WebhookURL: parseURL(webhookURL), Events: alerts.Events(), Cooldown: cooldown, MaxPerHour: 60, Instance: instance, Now: func() time.Time { return time.Now().UTC() }, ProcessLog: slog.New(slog.DiscardHandler), } } // withSlackAndNtfy returns params with Slack at slackURL and ntfy at // ntfyURL set as well. func withSlackAndNtfy(params alerts.Params) alerts.Params { params.SlackURL = parseURL(slackURL) params.NtfyURL = parseURL(ntfyURL) return params } // parseURL returns rawURL, parsed. func parseURL(rawURL string) *url.URL { parsed, err := url.Parse(rawURL) if err != nil { panic(err) } return parsed } // start returns a stand-in for the webhook that answers, and a Queue that // sends to it, run until the test ends. func start(t *testing.T, params alerts.Params) (*standIn, *alerts.Queue) { t.Helper() standIns, q := startAll(t, params) return standIns[alerts.DestinationWebhook], q } // startAll returns, by destination, a stand-in that answers for each // destination params sets, and a Queue that sends to them, run until the // test ends. func startAll(t *testing.T, params alerts.Params) (map[string]*standIn, *alerts.Queue) { t.Helper() q := alerts.New(params) standIns := map[string]*standIn{} for _, destination := range q.DestinationsSet() { standIns[destination] = &standIn{answers: answering} q.SetTransport(destination, standIns[destination]) } ctx, stop := context.WithCancel(t.Context()) stopped := make(chan struct{}) go func() { q.Run(ctx) close(stopped) }() t.Cleanup(func() { stop() <-stopped }) return standIns, q } // midnight is when each test starts. func midnight() time.Time { return time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC) } // netblock returns the n-th netblock of a test, counted from 0. func netblock(n int) netip.Prefix { return netip.MustParsePrefix(fmt.Sprintf("203.0.%d.%d/32", 113+n/256, n%256)) } // roundTrip returns state once written as JSON and read back, as // alerts.json carries it from one start to the next. func roundTrip(t *testing.T, state alerts.State) alerts.State { t.Helper() data, err := json.Marshal(state) if err != nil { t.Fatalf("encode: %v", err) } var read alerts.State err = json.Unmarshal(data, &read) if err != nil { t.Fatalf("decode: %v", err) } return read } // wantAlert checks every field of an alert the webhook was sent. func wantAlert(t *testing.T, got, want map[string]any) { t.Helper() if !reflect.DeepEqual(got, want) { t.Errorf("alert %v, want %v", got, want) } } // wantEvents checks the events of the alerts the webhook was sent, in // order. func wantEvents(t *testing.T, webhook *standIn, want ...string) { t.Helper() got := make([]string, 0, len(webhook.received())) for _, request := range webhook.received() { event, _ := request.alert["event"].(string) got = append(got, event) } if !slices.Equal(got, want) { t.Errorf("the webhook was sent %v, want %v", got, want) } } // wantCounts checks the alerts q counts as sent to the webhook, the // requests to it that it counts as failed, the alerts it counts as held // back, and those it counts as dropped for the webhook. func wantCounts( t *testing.T, q *alerts.Queue, sent, failed, suppressed, dropped int64, ) { t.Helper() wantDestinationCounts(t, q, alerts.DestinationWebhook, alerts.Counts{Sent: sent, Failed: failed, Dropped: dropped}) if q.Suppressed() != suppressed { t.Errorf("%d alerts held back, want %d", q.Suppressed(), suppressed) } } // wantDestinationCounts checks q's counts for destination. func wantDestinationCounts( t *testing.T, q *alerts.Queue, destination string, want alerts.Counts, ) { t.Helper() if got := q.Counts(destination); got != want { t.Errorf("%s counts %+v, want %+v", destination, got, want) } } // eventMessage is an alert, and the message Slack and ntfy are sent for // it: its title, the priority and the tag ntfy is sent, with a space // between them, and its text. type eventMessage struct { alert alerts.Alert title, priorityAndTag, text string } // anAlertForEachEvent returns an alert for each event SWWAF_ALERT_EVENTS // names, the first in observe mode, each with its message. func anAlertForEachEvent() []eventMessage { return []eventMessage{ { alerts.Alert{ Event: alerts.EventBan, Client: netip.MustParseAddr("203.0.113.9"), Netblock: netip.MustParsePrefix("203.0.113.0/24"), Country: "DE", Reason: "requests per hour over the limit of 10000", Detail: map[string]any{"mode": "observe"}, }, "fsn1app1/gitea: ban", "default no_entry", "requests per hour over the limit of 10000\nclient: 203.0.113.9\n" + "netblock: 203.0.113.0/24\ncountry: DE\nmode: observe", }, { alerts.Alert{ Event: alerts.EventPermanentBan, Client: netip.MustParseAddr("198.51.100.7"), Netblock: netip.MustParsePrefix("198.51.100.7/32"), Country: "FR", Reason: "matched the rule env-file", }, "fsn1app1/gitea: permanent_ban", "high no_entry", "matched the rule env-file\nclient: 198.51.100.7\n" + "netblock: 198.51.100.7/32\ncountry: FR", }, { alerts.Alert{ Event: alerts.EventWAFBlock, Client: netip.MustParseAddr("192.0.2.1"), Netblock: netip.MustParsePrefix("192.0.2.1/32"), Reason: "refused by the Core Rule Set", }, "fsn1app1/gitea: waf_block", "default shield", "refused by the Core Rule Set\nclient: 192.0.2.1\nnetblock: 192.0.2.1/32", }, { alerts.Alert{ Event: alerts.EventAnomaly, Netblock: netip.MustParsePrefix("192.0.2.0/24"), Reason: "requests per minute over the threshold of 5000", }, "fsn1app1/gitea: anomaly", "high chart_with_upwards_trend", "requests per minute over the threshold of 5000\nnetblock: 192.0.2.0/24", }, { alerts.Alert{ Event: alerts.EventReputationHit, Client: netip.MustParseAddr("192.0.2.2"), Netblock: netip.MustParsePrefix("192.0.2.2/32"), Reason: "listed by a DNS blocklist", }, "fsn1app1/gitea: reputation_hit", "low label", "listed by a DNS blocklist\nclient: 192.0.2.2\nnetblock: 192.0.2.2/32", }, { alerts.Alert{ Event: alerts.EventSourceFailure, Reason: "asking GeoJS failed", Detail: map[string]any{"source": "geojs"}, }, "fsn1app1/gitea: source_failure", "high warning", "asking GeoJS failed\nsource: geojs", }, { alerts.Alert{ Event: alerts.EventFileError, Reason: "a rule file has an error, and the rules stay as they were", Detail: map[string]any{ "file": "/etc/smallwebwaf/rules.d/50-app.rules", "error": "line 2: no action", }, }, "fsn1app1/gitea: file_error", "high warning", "a rule file has an error, and the rules stay as they were\n" + "file: /etc/smallwebwaf/rules.d/50-app.rules\nerror: line 2: no action", }, } } // wantSlackMessage checks that request, one Slack was sent, posted the // message text, as JSON. func wantSlackMessage(t *testing.T, request post, text string) { t.Helper() if request.method != http.MethodPost || request.url != slackURL || request.header.Get("Content-Type") != "application/json" || !reflect.DeepEqual(request.alert, map[string]any{"text": text}) { t.Errorf("Slack was sent %s %s, Content-Type %q, %s, want POST %s, "+ "application/json, the text %q", request.method, request.url, request.header.Get("Content-Type"), request.body, slackURL, text) } } // wantNtfyMessage checks that request, one ntfy was sent, posted the // message text with the title, and with the priority and the tag // priorityAndTag gives, with a space between them. func wantNtfyMessage(t *testing.T, request post, title, priorityAndTag, text string) { t.Helper() got := []string{ request.method, request.url, request.header.Get("Title"), request.header.Get("Priority") + " " + request.header.Get("Tags"), request.body, } want := []string{http.MethodPost, ntfyURL, title, priorityAndTag, text} if !slices.Equal(got, want) { t.Errorf("ntfy was sent the method, URL, title, priority and tag, and text "+ "%q, want %q", got, want) } } // wantBodies checks the bodies of the requests the stand-in was sent, in // order. func wantBodies(t *testing.T, s *standIn, want ...string) { t.Helper() got := make([]string, 0, len(want)) for _, request := range s.received() { got = append(got, request.body) } if !slices.Equal(got, want) { t.Errorf("bodies %q, want %q", got, want) } }