Alerts to a JSON webhook, with a cooldown and an hourly summary (closes #26)

SWWAF_ALERT_WEBHOOK_URL gets one JSON POST per alert, in SPEC.md's
schema, with SWWAF_ALERT_WEBHOOK_HEADERS: ban and permanent_ban, with
the ban's notes, in observe mode too, marked mode observe and worked
out only when the alert would be sent; source_failure for GeoJS;
file_error for a rule or state file with an error. SWWAF_ALERT_EVENTS
chooses; SWWAF_ALERT_COOLDOWN holds back repeats by netblock, file or
source; past SWWAF_ALERT_MAX_PER_HOUR the hour ends in one summary. A
bounded queue, retried with backoff, holds up no request; a 4xx other
than 408 and 429 gives the alert up. alerts.json keeps the queue, the
cooldowns and the hour. Nothing shows the URL's path or query.

Judgement call: the summary's event is summary, which SPEC.md omits.
Judgement call: an admin's ban raises no alert.

Model: opus-5-5

internal/smallwebwaf/smallwebwaf.go

@@ -15,6 +15,7 @@ import (
 	"syscall"
 	"time"
 
+	"sneak.berlin/go/smallwebwaf/internal/alerts"
 	"sneak.berlin/go/smallwebwaf/internal/config"
 	"sneak.berlin/go/smallwebwaf/internal/lookup"
 	"sneak.berlin/go/smallwebwaf/internal/proxy"
@@ -91,10 +92,16 @@ func Run(ctx context.Context, params Params) int {
 		defer stopSending()
 	}
 
+	// The state files and the alerts give times in UTC.
+	now := func() time.Time { return time.Now().UTC() }
+
+	alertQueue := newAlertQueue(cfg, now, processLog)
+
 	ruleFiles, err := rules.Load(rules.Params{
 		Dir:        cfg.RulesDir,
 		Enabled:    cfg.RulesEnabled,
 		ProcessLog: processLog,
+		Alerts:     alertQueue,
 	})
 	if err != nil {
 		processLog.Error("cannot use the rule files", "error", err.Error())
@@ -102,9 +109,6 @@ func Run(ctx context.Context, params Params) int {
 		return 1
 	}
 
-	// The state files give times in UTC.
-	now := func() time.Time { return time.Now().UTC() }
-
 	server := proxy.New(proxy.Params{
 		Config:     cfg,
 		RequestLog: stdout,
@@ -112,22 +116,17 @@ func Run(ctx context.Context, params Params) int {
 		GeoJSURL:   lookup.URL,
 		Now:        now,
 		Rules:      ruleFiles,
+		Alerts:     alertQueue,
 	})
 	if remote != nil {
 		server.Metrics.AddRemoteLog(remote)
 	}
 
-	files, err := state.Load(state.Params{
-		Dir:             cfg.StateDir,
-		WriteDelay:      cfg.StateWriteDelay,
-		CounterInterval: cfg.StateCounterInterval,
-		Ledger:          server.Ledger,
-		Limiter:         server.Limiter,
-		GeoJS:           server.GeoJS,
-		Now:             now,
-		ProcessLog:      processLog,
-		Metrics:         server.Metrics,
-	})
+	if cfg.AlertWebhookURL != nil {
+		server.Metrics.AddAlerts(alertQueue)
+	}
+
+	files, err := loadStateFiles(cfg, server, alertQueue, now, processLog)
 	if err != nil {
 		processLog.Error("cannot use the state files", "error", err.Error())
 
@@ -147,7 +146,44 @@ func Run(ctx context.Context, params Params) int {
 		"address", listener.Addr().String(),
 		"settings", cfg)
 
-	return serve(ctx, server.Server, listener, files, ruleFiles, processLog)
+	return serve(ctx, server.Server, listener, files, ruleFiles, alertQueue, processLog)
+}
+
+// loadStateFiles reads the state files into the parts of server and into
+// alertQueue, as state.Load does.
+func loadStateFiles(
+	cfg *config.Config, server *proxy.Server, alertQueue *alerts.Queue,
+	now func() time.Time, processLog *slog.Logger,
+) (*state.Files, error) {
+	return state.Load(state.Params{
+		Dir:             cfg.StateDir,
+		WriteDelay:      cfg.StateWriteDelay,
+		CounterInterval: cfg.StateCounterInterval,
+		Ledger:          server.Ledger,
+		Limiter:         server.Limiter,
+		GeoJS:           server.GeoJS,
+		Alerts:          alertQueue,
+		Now:             now,
+		ProcessLog:      processLog,
+		Metrics:         server.Metrics,
+	})
+}
+
+// newAlertQueue returns the queue of the alerts to SWWAF_ALERT_WEBHOOK_URL,
+// with the settings for it.
+func newAlertQueue(
+	cfg *config.Config, now func() time.Time, processLog *slog.Logger,
+) *alerts.Queue {
+	return alerts.New(alerts.Params{
+		WebhookURL:     cfg.AlertWebhookURL,
+		WebhookHeaders: cfg.AlertWebhookHeaders,
+		Events:         cfg.AlertEvents,
+		Cooldown:       cfg.AlertCooldown,
+		MaxPerHour:     cfg.AlertMaxPerHour,
+		Instance:       cfg.InstanceName,
+		Now:            now,
+		ProcessLog:     processLog,
+	})
 }
 
 // newRemoteLogSender returns a sender of the log lines to
@@ -188,12 +224,14 @@ func startSending(
 }
 
 // serve serves requests on listener, writes the state files as they are
-// due, takes in an admin's edits of them, and reads the rule files again
-// as they change, until ctx is done. Then it gives the requests in
-// progress shutdownTimeout to finish, and writes every state file.
+// due, takes in an admin's edits of them, reads the rule files again as
+// they change, and sends the alerts, until ctx is done. Then it gives the
+// requests in progress shutdownTimeout to finish, and writes every state
+// file, alerts.json with the alerts still waiting.
 func serve(
 	ctx context.Context, server *http.Server, listener net.Listener,
-	files *state.Files, ruleFiles *rules.Files, processLog *slog.Logger,
+	files *state.Files, ruleFiles *rules.Files, alertQueue *alerts.Queue,
+	processLog *slog.Logger,
 ) int {
 	served := make(chan error, 1)
 
@@ -204,24 +242,10 @@ func serve(
 	writing, stopWriting := context.WithCancel(ctx)
 	defer stopWriting()
 
-	written := make(chan struct{})
-	watched := make(chan struct{})
-	rulesWatched := make(chan struct{})
-
-	go func() {
-		files.Run(writing)
-		close(written)
-	}()
-
-	go func() {
-		files.Watch(writing)
-		close(watched)
-	}()
-
-	go func() {
-		ruleFiles.Watch(writing)
-		close(rulesWatched)
-	}()
+	written := inBackground(func() { files.Run(writing) })
+	watched := inBackground(func() { files.Watch(writing) })
+	rulesWatched := inBackground(func() { ruleFiles.Watch(writing) })
+	alertsSent := inBackground(func() { alertQueue.Run(writing) })
 
 	select {
 	case err := <-served:
@@ -253,7 +277,8 @@ func serve(
 	}
 
 	// Run and Watch have ended, so nothing else reads or writes the
-	// files. Every request has ended too, but for two kinds
+	// files, and no alert is being sent, so that alerts.json keeps every
+	// alert not yet sent. Every request has ended too, but for two kinds
 	// that Go's server does not wait for: one cut off because Shutdown
 	// timed out, and one whose connection switched protocols, such as a
 	// WebSocket. Such a request adds to its client's history only as it
@@ -262,6 +287,7 @@ func serve(
 	<-written
 	<-watched
 	<-rulesWatched
+	<-alertsSent
 
 	err = files.WriteAll()
 	if err != nil {
@@ -274,3 +300,16 @@ func serve(
 
 	return 0
 }
+
+// inBackground runs task on a goroutine of its own, and returns a channel
+// that is closed once task has returned.
+func inBackground(task func()) <-chan struct{} {
+	done := make(chan struct{})
+
+	go func() {
+		task()
+		close(done)
+	}()
+
+	return done
+}

internal/smallwebwaf/smallwebwaf_test.go

@@ -14,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -547,6 +548,97 @@ func TestStalledRemoteLogEndpointHoldsUpNoRequest(t *testing.T) {
 	out.line(t, "type", "request")
 }
 
+func TestBanIsAlertedAndAnAlertNotSentIsKeptAcrossARestart(t *testing.T) {
+	t.Parallel()
+
+	webhook := startWebhook(t)
+	rules := t.TempDir()
+
+	err := os.WriteFile(filepath.Join(rules, "50-app.rules"),
+		[]byte(`probe path ban ^/\.env$`+"\n"), 0o600)
+	if err != nil {
+		t.Fatalf("write the rule file: %v", err)
+	}
+
+	dir := t.TempDir()
+	env := map[string]string{
+		listenAddr:                    localhost + ":0",
+		upstreamURL:                   startApp(t),
+		stateDir:                      dir,
+		rulesDir:                      rules,
+		"SWWAF_ALERT_WEBHOOK_URL":     webhook.url,
+		"SWWAF_ALERT_WEBHOOK_HEADERS": "Authorization:Bearer " + adminSecret,
+	}
+
+	runUntilStopped(t, env, func(url string) {
+		// The probe bans the client, and the webhook is sent the alert.
+		wantRefused(t, url+".env")
+
+		post := webhook.waitFor(t, "ban", true)
+		if post.alert["client"] != localhost || post.alert["netblock"] != localhost+"/32" ||
+			post.authorization != "Bearer "+adminSecret {
+			t.Errorf("the webhook was sent %v, with Authorization %q", post.alert,
+				post.authorization)
+		}
+
+		// The webhook fails, so the alert for the ban made permanent by the
+		// client's next request waits.
+		webhook.failing.Store(true)
+		wantRefused(t, url)
+		webhook.waitFor(t, "permanent_ban", false)
+	})
+
+	// alerts.json keeps it as smallwebwaf stops, and once started again,
+	// smallwebwaf sends it.
+	var file struct {
+		Waiting []struct {
+			Event string `json:"event"`
+		} `json:"waiting"`
+	}
+
+	path := filepath.Join(dir, "alerts.json")
+
+	data, err := os.ReadFile(path) //nolint:gosec // a file in the test's directory
+	if err == nil {
+		err = json.Unmarshal(data, &file)
+	}
+
+	if err != nil || len(file.Waiting) != 1 || file.Waiting[0].Event != "permanent_ban" {
+		t.Fatalf("alerts.json holds %s (%v), want the permanent_ban alert waiting", data, err)
+	}
+
+	// It counts the alert sent in the metrics, read here from a client the
+	// ban does not cover.
+	const token = "0123456789abcdef0123456789abcdef"
+
+	webhook.failing.Store(false)
+
+	env["SWWAF_ALLOW_NETS"] = localhost
+	env["SWWAF_METRICS_TOKEN"] = token
+
+	runUntilStopped(t, env, func(url string) {
+		webhook.waitFor(t, "permanent_ban", true)
+
+		// As long as that takes, so that a slow test process cannot fail
+		// the test.
+		const sent = "\nsmallwebwaf_alerts_sent_total{destination=\"webhook\"} 1\n"
+
+		metrics := metricsText(t, url+"_smallwebwaf/metrics", token)
+		for !strings.Contains(metrics, sent) {
+			time.Sleep(pollInterval)
+
+			metrics = metricsText(t, url+"_smallwebwaf/metrics", token)
+		}
+
+		for _, series := range []string{"failed", "suppressed", "dropped"} {
+			zero := "\nsmallwebwaf_alerts_" + series + "_total{destination=\"webhook\"} 0\n"
+			if !strings.Contains(metrics, zero) {
+				t.Errorf("no %q in the metrics:\n%s", zero, metrics)
+			}
+		}
+	})
+}
+
 func TestStateFileThatDoesNotParseStopsTheStart(t *testing.T) {
 	t.Parallel()
 
@@ -917,6 +1009,80 @@ func saveUntilAnswered(t *testing.T, path, content, url, from string, status int
 	}
 }
 
+// webhook is a stand-in for SWWAF_ALERT_WEBHOOK_URL. It notes each alert
+// it is sent, and answers 204, or 503 while failing.
+type webhook struct {
+	url     string
+	failing atomic.Bool
+
+	mu    sync.Mutex
+	posts []webhookPost
+}
+
+// webhookPost is an alert the webhook was sent, with the Authorization
+// header sent with it, and whether the webhook took it.
+type webhookPost struct {
+	alert         map[string]any
+	authorization string
+	answered      bool
+}
+
+// startWebhook starts a webhook that takes every alert.
+func startWebhook(t *testing.T) *webhook {
+	t.Helper()
+
+	w := &webhook{}
+	server := httptest.NewServer(http.HandlerFunc(
+		func(rw http.ResponseWriter, r *http.Request) {
+			var alert map[string]any
+
+			_ = json.NewDecoder(r.Body).Decode(&alert)
+			failing := w.failing.Load()
+
+			w.mu.Lock()
+			w.posts = append(w.posts, webhookPost{
+				alert: alert, authorization: r.Header.Get("Authorization"),
+				answered: !failing,
+			})
+			w.mu.Unlock()
+
+			if failing {
+				rw.WriteHeader(http.StatusServiceUnavailable)
+
+				return
+			}
+
+			rw.WriteHeader(http.StatusNoContent)
+		}))
+	t.Cleanup(server.Close)
+
+	w.url = server.URL + "/alerts"
+
+	return w
+}
+
+// waitFor waits until the webhook has been sent an alert for event that
+// it took, or, unless answered, failed, and returns it. It waits as long
+// as that takes, so that a slow test process cannot fail the test.
+func (w *webhook) waitFor(t *testing.T, event string, answered bool) webhookPost {
+	t.Helper()
+
+	for {
+		w.mu.Lock()
+
+		for _, post := range w.posts {
+			if post.alert["event"] == event && post.answered == answered {
+				w.mu.Unlock()
+
+				return post
+			}
+		}
+
+		w.mu.Unlock()
+		time.Sleep(pollInterval)
+	}
+}
+
 // statusFrom returns the status a request to url from the client at
 // from, as X-Forwarded-For names it, is answered with.
 func statusFrom(t *testing.T, url, from string) int {