Anomaly thresholds: alerts for unusual traffic, nothing refused (closes #101)

SWWAF_ANOMALY_CLIENT_*, _NET_*, _ASN_*, _TOTAL_* and SWWAF_WATCH_* with
SWWAF_WATCH_NETS: requests and bytes per minute and per hour, each off by
default; with all off, nothing is counted. Otherwise every request but the
health check is counted, allow-listed and exempt ones included; a count
over its threshold raises an anomaly alert, with a cooldown per scope. At
most 20,000 counters, kept in alerts.json. A per-AS-number threshold with
lookups off, or a malformed SWWAF_WATCH_NETS, stops the start. A cooldown
that has run out is dropped as the hour ends, whatever it held back; the
hour's summary gives its repeats.

Judgement call: refused requests are counted too.
Judgement call: per-client counters are kept in alerts.json, which SPEC.md does not list.
Judgement call: a request counts for an AS number only if the lookup answered before it ended.

Model: opus-5-5

internal/state/state.go

@@ -2,7 +2,8 @@
 // SWWAF_STATE_DIR, as the "Persistent state" section of SPEC.md describes:
 // bans.json holds the bans, clients.json each client's counters and
 // history, lookups.json GeoJS's answers, and alerts.json the cooldowns,
-// the hour under way and the alerts waiting for each destination. Load
+// the hour under way, the alerts waiting for each destination and the
+// anomaly counters. Load
 // reads them at start, Watch takes in an admin's edit of one while
 // smallwebwaf runs, and Run and WriteAll write them. The disk is read and
 // written outside the parts' locks, which are held only to take a
@@ -29,6 +30,7 @@ import (
 
 	"github.com/fsnotify/fsnotify"
 	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/anomaly"
 	"sneak.berlin/go/smallwebwaf/internal/bans"
 	"sneak.berlin/go/smallwebwaf/internal/lookup"
 	"sneak.berlin/go/smallwebwaf/internal/metrics"
@@ -56,6 +58,7 @@ var (
 	errMissing     = errors.New("has no")
 	errCause       = errors.New("is not limit, attack or admin")
 	errDestination = errors.New("is not webhook, slack or ntfy")
+	errScope       = errors.New("is not client, net, asn, total or watch")
 	errWaitingList = errors.New(`waiting is a list, but now lists the alerts by ` +
 		`destination: put the list under "webhook", as "waiting": {"webhook": [...]}, ` +
 		`or remove the file`)
@@ -70,13 +73,14 @@ type Params struct {
 	// is (SWWAF_STATE_COUNTER_INTERVAL).
 	WriteDelay      time.Duration
 	CounterInterval time.Duration
-	// Ledger, Limiter, GeoJS and Alerts hold the state. Alerts also
-	// receive a file_error alert for an edit set aside, and for a write
-	// that fails while smallwebwaf runs.
-	Ledger  *bans.Ledger
-	Limiter *ratelimit.Limiter
-	GeoJS   *lookup.GeoJS
-	Alerts  *alerts.Queue
+	// Ledger, Limiter, GeoJS, Alerts and Anomalies hold the state. Alerts
+	// also receive a file_error alert for an edit set aside, and for a
+	// write that fails while smallwebwaf runs.
+	Ledger    *bans.Ledger
+	Limiter   *ratelimit.Limiter
+	GeoJS     *lookup.GeoJS
+	Alerts    *alerts.Queue
+	Anomalies *anomaly.Counters
 	// Now tells the time by which the counters' buckets run out, normally
 	// time.Now in UTC.
 	Now func() time.Time
@@ -135,11 +139,14 @@ type lookupsFile struct {
 }
 
 // alertsFile is alerts.json, indented for an admin to read and edit.
+//
+//nolint:tagliatelle // the state files use snake_case, as the request log does
 type alertsFile struct {
-	Version   int                       `json:"version"`
-	Cooldowns []alerts.Cooldown         `json:"cooldowns"`
-	Hour      alerts.Hour               `json:"hour"`
-	Waiting   map[string][]alerts.Alert `json:"waiting"`
+	Version         int                       `json:"version"`
+	Cooldowns       []alerts.Cooldown         `json:"cooldowns"`
+	Hour            alerts.Hour               `json:"hour"`
+	Waiting         map[string][]alerts.Alert `json:"waiting"`
+	AnomalyCounters []anomaly.Counter         `json:"anomaly_counters"`
 }
 
 // stateFile is the struct of a state file. Once the file is decoded, its
@@ -420,6 +427,7 @@ func (f *Files) takeIn(name string, data []byte, edit bool) (int, error) {
 		f.params.Alerts.Load(alerts.State{
 			Cooldowns: file.Cooldowns, Hour: file.Hour, Waiting: file.Waiting,
 		})
+		f.params.Anomalies.Load(file.AnomalyCounters, f.params.Now())
 
 		for _, waiting := range file.Waiting {
 			entries += len(waiting)
@@ -522,7 +530,7 @@ func (f *Files) encode(name string) ([]byte, error) {
 		held := f.params.Alerts.Snapshot()
 		file := alertsFile{
 			Version: version, Cooldowns: held.Cooldowns, Hour: held.Hour,
-			Waiting: held.Waiting,
+			Waiting: held.Waiting, AnomalyCounters: f.params.Anomalies.Snapshot(),
 		}
 
 		data, err := json.MarshalIndent(file, "", "    ")
@@ -673,8 +681,9 @@ func (f *lookupsFile) check(data []byte) error {
 
 // check refuses a cooldown without its event or when its alert was sent,
 // which would hold back no repeat, alerts waiting for a destination with
-// another name than webhook, slack or ntfy, most likely misspelt, and an
-// alert waiting without its event or its time.
+// another name than webhook, slack or ntfy, most likely misspelt, an
+// alert waiting without its event or its time, and an anomaly counter as
+// checkAnomalyCounters does.
 func (f *alertsFile) check([]byte) error {
 	for i, cooldown := range f.Cooldowns {
 		switch {
@@ -700,9 +709,58 @@ func (f *alertsFile) check([]byte) error {
 		}
 	}
 
+	return checkAnomalyCounters(f.AnomalyCounters)
+}
+
+// checkAnomalyCounters refuses an anomaly counter whose scope is not
+// client, net, asn, total or watch, most likely misspelt, and one without
+// a field it needs, as missingFromCounter tells.
+func checkAnomalyCounters(counters []anomaly.Counter) error {
+	for i, counter := range counters {
+		if !slices.Contains(anomaly.Scopes(), counter.Scope) {
+			return fmt.Errorf("anomaly_counters entry %d's scope %q %w", i+1,
+				counter.Scope, errScope)
+		}
+
+		field := missingFromCounter(counter)
+		if field != "" {
+			return fmt.Errorf("anomaly_counters %w", missing(i, field))
+		}
+	}
+
 	return nil
 }
 
+// missingFromCounter returns the first field counter, an anomaly counter,
+// needs and has not, or "" when it has them all: what tells it from the
+// others in its scope, without which it would never be counted again, the
+// netblock of a client, net or watch counter, the AS number of an asn one
+// and the name of a watch one; and the start of a window in which it has
+// requests or bytes, without which they would be dropped.
+func missingFromCounter(counter anomaly.Counter) string {
+	scope := counter.Scope
+
+	switch {
+	case scope != anomaly.ScopeASN && scope != anomaly.ScopeTotal &&
+		!counter.Netblock.IsValid():
+		return "netblock"
+	case scope == anomaly.ScopeASN && counter.ASN == "":
+		return "asn"
+	case scope == anomaly.ScopeWatch && counter.Name == "":
+		return "name"
+	case countsWithoutStart(counter.Minute):
+		return "minute.start"
+	case countsWithoutStart(counter.Hour):
+		return "hour.start"
+	case countsWithoutStart(counter.MinuteBytes):
+		return "minute_bytes.start"
+	case countsWithoutStart(counter.HourBytes):
+		return "hour_bytes.start"
+	default:
+		return ""
+	}
+}
+
 // countsWithoutStart reports whether b holds requests, or bytes, but no
 // start, which places them in time.
 func countsWithoutStart(b ratelimit.Buckets) bool {

internal/state/state_test.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/anomaly"
 	"sneak.berlin/go/smallwebwaf/internal/bans"
 	"sneak.berlin/go/smallwebwaf/internal/lookup"
 	"sneak.berlin/go/smallwebwaf/internal/metrics"
@@ -156,7 +157,50 @@ const filledAlertsJSON = `{
                 "suppressed_repeats": 0
             }
         ]
-    }
+    },
+    "anomaly_counters": [
+        {
+            "scope": "asn",
+            "asn": "AS64496",
+            "hour_bytes": {
+                "start": "2026-10-06T00:00:00Z",
+                "current": 8,
+                "previous": 0
+            }
+        },
+        {
+            "scope": "net",
+            "netblock": "203.0.113.0/24",
+            "minute": {
+                "start": "2026-10-06T00:00:00Z",
+                "current": 1,
+                "previous": 0
+            }
+        },
+        {
+            "scope": "total",
+            "minute": {
+                "start": "2026-10-06T00:00:00Z",
+                "current": 1,
+                "previous": 0
+            },
+            "minute_bytes": {
+                "start": "2026-10-06T00:00:00Z",
+                "current": 8,
+                "previous": 0
+            }
+        },
+        {
+            "scope": "watch",
+            "netblock": "203.0.113.0/24",
+            "name": "office",
+            "hour": {
+                "start": "2026-10-06T00:00:00Z",
+                "current": 1,
+                "previous": 0
+            }
+        }
+    ]
 }
 `
 
@@ -191,6 +235,8 @@ func TestFilesWrittenAndReadBack(t *testing.T) {
 		t.Errorf("%s read back\n%+v\nwant\n%+v", alertsJSON, got, want)
 	}
 
+	wantEqual(t, alertsJSON, after.Anomalies.Snapshot(), before.Anomalies.Snapshot())
+
 	// Each one-per-line file lists its entries by client, and nothing
 	// but the four files is left in the directory.
 	wantEntries(t, filepath.Join(dir, clientsJSON), "clients",
@@ -251,6 +297,49 @@ func TestSourceFailureCooldownKeptInAlertsJSONAcrossARestart(t *testing.T) {
 	}
 }
 
+func TestAnomalyCountersKeptInAlertsJSONAcrossARestart(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	request := anomaly.Request{
+		Client:      netip.MustParseAddr("203.0.113.9"),
+		ClientGroup: netip.MustParsePrefix("203.0.113.9/32"),
+	}
+
+	// The whole service may have two requests a minute.
+	withThreshold := func() state.Params {
+		params := newParams(dir)
+		params.Anomalies = anomaly.New(anomaly.Params{
+			Total: anomaly.Thresholds{RequestsPerMinute: 2}, Alerts: params.Alerts,
+		})
+
+		return params
+	}
+
+	before := withThreshold()
+	files := load(t, before)
+
+	for range 2 {
+		before.Anomalies.Count(midnight(), request)
+	}
+
+	err := files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	// After the restart, the third request in the minute is over it.
+	after := withThreshold()
+	load(t, after)
+	after.Anomalies.Count(midnight(), request)
+
+	waiting := after.Alerts.Snapshot().Waiting[alerts.DestinationWebhook]
+	if len(waiting) != 1 || waiting[0].Event != alerts.EventAnomaly ||
+		waiting[0].Detail["count"] != float64(3) {
+		t.Errorf("alerts wait %+v, want one for 3 requests", waiting)
+	}
+}
+
 func TestBansJSONIsIndentedWithNullForAPermanentBan(t *testing.T) {
 	t.Parallel()
 
@@ -332,6 +421,13 @@ func TestFileThatDoesNotParseStopsTheStart(t *testing.T) {
 			`{"version": 1, "waiting": {"webhook": [], "slak": []}}`,
 			`: waiting "slak" is not webhook, slack or ntfy`,
 		},
+		{
+			"an anomaly counter of an unknown scope", alertsJSON,
+			`{"version": 1, "anomaly_counters": [{"scope": "total"}, ` +
+				`{"scope": "nett", "netblock": "203.0.113.0/24"}]}`,
+			`: anomaly_counters entry 2's scope "nett" is not client, net, asn, total ` +
+				`or watch`,
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
@@ -455,6 +551,39 @@ func TestAlertsJSONEntryWithoutAFieldItNeedsStopsTheStart(t *testing.T) {
 			`{"version": 1, "waiting": {"slack": [{"event": "ban"}]}}`,
 			`: waiting slack entry 1 has no "time"`,
 		},
+		{
+			// The whole service's counter needs nothing to tell it apart.
+			"an anomaly counter of a netblock without it",
+			`{"version": 1, "anomaly_counters": [{"scope": "total"}, {"scope": "net"}]}`,
+			`: anomaly_counters entry 2 has no "netblock"`,
+		},
+		{
+			"an anomaly counter of a client without its netblock",
+			`{"version": 1, "anomaly_counters": [{"scope": "client"}]}`,
+			`: anomaly_counters entry 1 has no "netblock"`,
+		},
+		{
+			"an anomaly counter of an AS number without it",
+			`{"version": 1, "anomaly_counters": [{"scope": "asn"}]}`,
+			`: anomaly_counters entry 1 has no "asn"`,
+		},
+		{
+			"an anomaly counter of a named netblock without its name",
+			`{"version": 1, "anomaly_counters": [` +
+				`{"scope": "watch", "netblock": "203.0.113.0/24"}]}`,
+			`: anomaly_counters entry 1 has no "name"`,
+		},
+		{
+			"an anomaly counter of a named netblock without its netblock",
+			`{"version": 1, "anomaly_counters": [{"scope": "watch", "name": "office"}]}`,
+			`: anomaly_counters entry 1 has no "netblock"`,
+		},
+		{
+			"an anomaly counter with bytes in a window without its start",
+			`{"version": 1, "anomaly_counters": [` +
+				`{"scope": "total", "hour_bytes": {"current": 5}}]}`,
+			`: anomaly_counters entry 1 has no "hour_bytes.start"`,
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
@@ -1273,10 +1402,19 @@ func midnight() time.Time {
 
 // newParams returns Params for the state files in dir, with parts that
 // hold nothing yet. GeoJS is never asked, and the alerts, at most two an
-// hour, are never sent.
+// hour, are never sent. The anomaly counters count the scopes fill
+// counts, with thresholds fill does not reach.
 func newParams(dir string) state.Params {
 	discard := slog.New(slog.DiscardHandler)
 	m := metrics.New(1, "app")
+	queue := alerts.New(alerts.Params{
+		WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"},
+		Events:     alerts.Events(),
+		Cooldown:   15 * time.Minute,
+		MaxPerHour: 2,
+		Instance:   "fsn1app1/gitea",
+		Now:        midnight,
+	})
 
 	return state.Params{
 		Dir:             dir,
@@ -1293,13 +1431,16 @@ func newParams(dir string) state.Params {
 		GeoJS: lookup.New(lookup.Params{
 			Now: midnight, ProcessLog: discard, Metrics: m,
 		}),
-		Alerts: alerts.New(alerts.Params{
-			WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"},
-			Events:     alerts.Events(),
-			Cooldown:   15 * time.Minute,
-			MaxPerHour: 2,
-			Instance:   "fsn1app1/gitea",
-			Now:        midnight,
+		Alerts: queue,
+		Anomalies: anomaly.New(anomaly.Params{
+			Net:            anomaly.Thresholds{RequestsPerMinute: 1000},
+			ASN:            anomaly.Thresholds{BytesPerHour: 1 << 30},
+			Total:          anomaly.Thresholds{RequestsPerMinute: 1000, BytesPerMinute: 1 << 30},
+			Watch:          anomaly.Thresholds{RequestsPerHour: 1000},
+			NetV4Prefix:    24,
+			NetV6Prefix:    48,
+			NamedNetblocks: []anomaly.NamedNetblock{{Name: "office", Netblock: office()}},
+			Alerts:         queue,
 		}),
 		Now:        midnight,
 		ProcessLog: discard,
@@ -1307,10 +1448,15 @@ func newParams(dir string) state.Params {
 	}
 }
 
+// office is the named netblock of the anomaly counters of newParams.
+func office() netip.Prefix {
+	return netip.MustParsePrefix("203.0.113.0/24")
+}
+
 // fill puts a permanent ban an admin made, a ban for a broken limit and
 // one for a clear sign of attack, clients with counts and histories,
-// GeoJS answers, and alerts, as filledAlertsJSON holds them, into the
-// parts of params.
+// GeoJS answers, and alerts and anomaly counters, as filledAlertsJSON
+// holds them, into the parts of params.
 func fill(params state.Params) {
 	now := midnight()
 	client := netip.MustParsePrefix("203.0.113.9/32")
@@ -1361,6 +1507,10 @@ func fill(params state.Params) {
 	params.Alerts.Raise(alerts.Alert{
 		Event: alerts.EventSourceFailure, Reason: "asking GeoJS failed",
 	})
+
+	params.Anomalies.Count(now, anomaly.Request{
+		Client: client.Addr(), ClientGroup: client, ASN: asn, Bytes: 8,
+	})
 }
 
 // permanentBan is the ban permanentBansJSON holds.