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/anomaly/anomaly.go

@@ -0,0 +1,406 @@
+// Package anomaly counts requests and bytes over a minute and an hour, per
+// client, per surrounding netblock, per AS number, for the whole service
+// and per named netblock, and raises an anomaly alert for a count over its
+// threshold, as "Anomaly thresholds" under "Configuration surface" in
+// SPEC.md describes. It refuses and bans nothing. At most 20,000 counters
+// are kept, in memory, and written to alerts.json and read from it by the
+// state package.
+package anomaly
+
+import (
+	"cmp"
+	"fmt"
+	"net/netip"
+	"slices"
+	"sync"
+	"time"
+
+	"github.com/hashicorp/golang-lru/v2/simplelru"
+	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
+)
+
+// maxCounters is how many counters are kept. Past it, the counter counted
+// least recently is dropped, and starts afresh if it is counted again.
+const maxCounters = 20000
+
+// The scopes, what a counter counts, as the settings, alerts.json and the
+// alerts name them.
+const (
+	// ScopeClient is one client: an IPv4 address, or an IPv6 /64.
+	ScopeClient = "client"
+	// ScopeNet is the netblock around a client, SWWAF_ANOMALY_NET_V4_PREFIX
+	// or SWWAF_ANOMALY_NET_V6_PREFIX long.
+	ScopeNet = "net"
+	// ScopeASN is an AS number.
+	ScopeASN = "asn"
+	// ScopeTotal is the whole service.
+	ScopeTotal = "total"
+	// ScopeWatch is a named netblock of SWWAF_WATCH_NETS.
+	ScopeWatch = "watch"
+)
+
+// Scopes returns every scope.
+func Scopes() []string {
+	return []string{ScopeClient, ScopeNet, ScopeASN, ScopeTotal, ScopeWatch}
+}
+
+// The windows a counter counts in, as the alerts name them.
+const (
+	minute = "minute"
+	hour   = "hour"
+)
+
+// Thresholds are the most requests and the most bytes a scope may have
+// counted in a minute and in an hour before an alert is raised. Zero is
+// off.
+type Thresholds struct {
+	RequestsPerMinute int64
+	RequestsPerHour   int64
+	BytesPerMinute    int64
+	BytesPerHour      int64
+}
+
+// NamedNetblock is a netblock SWWAF_WATCH_NETS names.
+type NamedNetblock struct {
+	Name     string
+	Netblock netip.Prefix
+}
+
+// Params are what New needs.
+type Params struct {
+	// The thresholds of each scope: SWWAF_ANOMALY_CLIENT_*,
+	// SWWAF_ANOMALY_NET_*, SWWAF_ANOMALY_ASN_*, SWWAF_ANOMALY_TOTAL_* and
+	// SWWAF_WATCH_*.
+	Client, Net, ASN, Total, Watch Thresholds
+	// NetV4Prefix and NetV6Prefix are the lengths of the netblock around a
+	// client (SWWAF_ANOMALY_NET_V4_PREFIX and SWWAF_ANOMALY_NET_V6_PREFIX).
+	NetV4Prefix, NetV6Prefix int
+	// NamedNetblocks are SWWAF_WATCH_NETS.
+	NamedNetblocks []NamedNetblock
+	// Alerts receive the anomaly alerts.
+	Alerts *alerts.Queue
+}
+
+// Counter is one scope's counts, as alerts.json holds them: the scope,
+// with the netblock, the AS number or the name that tells it from the
+// others in that scope, and its two buckets of requests and of bytes in
+// the minute and in the hour. A bucket whose threshold is off counts
+// nothing, and is left out.
+//
+//nolint:tagliatelle // the state files use snake_case, as the request log does
+type Counter struct {
+	Scope       string            `json:"scope"`
+	Netblock    netip.Prefix      `json:"netblock,omitzero"`
+	ASN         string            `json:"asn,omitempty"`
+	Name        string            `json:"name,omitempty"`
+	Minute      ratelimit.Buckets `json:"minute,omitzero"`
+	Hour        ratelimit.Buckets `json:"hour,omitzero"`
+	MinuteBytes ratelimit.Buckets `json:"minute_bytes,omitzero"`
+	HourBytes   ratelimit.Buckets `json:"hour_bytes,omitzero"`
+}
+
+// Request is a request that has ended, as the counters count it.
+type Request struct {
+	// Client is the client's address, and ClientGroup the client it is
+	// counted as: its IPv4 address, or its IPv6 /64.
+	Client      netip.Addr
+	ClientGroup netip.Prefix
+	// ASN, ASName and Country are the client's as looked up, each "" when
+	// unknown.
+	ASN, ASName, Country string
+	// Bytes are the request's bytes, as SWWAF_BYTES_COUNT counts them.
+	Bytes int64
+}
+
+// Counters counts each request in the scopes it is in. It is safe for
+// concurrent use.
+type Counters struct {
+	params Params
+
+	mu       sync.Mutex
+	counters *simplelru.LRU[key, *Counter]
+}
+
+// key is what tells a counter from the others: its scope, with its
+// netblock, AS number or name.
+type key struct {
+	scope    string
+	netblock netip.Prefix
+	asn      string
+	name     string
+}
+
+// New returns Counters for params, with nothing counted yet.
+func New(params Params) *Counters {
+	counters, err := simplelru.NewLRU[key, *Counter](maxCounters, nil)
+	if err != nil {
+		panic(err) // NewLRU fails only for a size below one
+	}
+
+	return &Counters{params: params, counters: counters}
+}
+
+// Count counts r, a request that has ended, at now, in each scope it is
+// in whose thresholds are not all off: its client, the netblock around
+// it, its AS number once known, the whole service, and each named
+// netblock it is in. Only the counts whose threshold is set are counted.
+// For each scope whose count is over a threshold, it raises an anomaly
+// alert, for the first such count in the order requests and bytes in the
+// minute, then in the hour; the alert queue's cooldown holds back the
+// repeats. Nothing is refused or banned.
+func (c *Counters) Count(now time.Time, r Request) {
+	var raised []alerts.Alert
+
+	c.mu.Lock()
+
+	for _, scope := range c.scopesOf(r) {
+		counter, found := c.counters.Get(scope.key)
+		if !found {
+			counter = scope.key.counter()
+			c.counters.Add(scope.key, counter)
+		}
+
+		over, passed := counter.add(now, r.Bytes, scope.thresholds)
+		if passed {
+			raised = append(raised, alertFor(r, scope.key, over))
+		}
+	}
+
+	c.mu.Unlock()
+
+	for _, alert := range raised {
+		c.params.Alerts.Raise(alert)
+	}
+}
+
+// Snapshot returns every counter, sorted by scope, then by netblock, AS
+// number and name, as alerts.json lists them.
+func (c *Counters) Snapshot() []Counter {
+	c.mu.Lock()
+
+	counters := make([]Counter, 0, c.counters.Len())
+	for _, counter := range c.counters.Values() {
+		counters = append(counters, *counter)
+	}
+
+	c.mu.Unlock()
+
+	slices.SortFunc(counters, func(a, b Counter) int {
+		return cmp.Or(cmp.Compare(a.Scope, b.Scope), a.Netblock.Compare(b.Netblock),
+			cmp.Compare(a.ASN, b.ASN), cmp.Compare(a.Name, b.Name))
+	})
+
+	return counters
+}
+
+// Load puts counters, read from alerts.json, in place of those held, in
+// the order they were last counted, as the starts of their buckets tell,
+// so that the one counted least recently is dropped first. Each netblock
+// is masked to its length, so that 203.0.113.9/24 is 203.0.113.0/24.
+// Buckets whose time has passed at now are emptied, and a counter left
+// with every bucket empty is dropped.
+func (c *Counters) Load(counters []Counter, now time.Time) {
+	counters = slices.Clone(counters)
+	slices.SortStableFunc(counters, func(a, b Counter) int {
+		return a.lastStart().Compare(b.lastStart())
+	})
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.counters.Purge()
+
+	for _, counter := range counters {
+		counter.Netblock = counter.Netblock.Masked()
+		empty := true
+
+		for _, count := range counter.counts() {
+			if count.buckets.Passed(now, count.length) {
+				*count.buckets = ratelimit.Buckets{}
+			}
+
+			empty = empty && *count.buckets == ratelimit.Buckets{}
+		}
+
+		if !empty {
+			c.counters.Add(counter.key(), &counter)
+		}
+	}
+}
+
+// scope is a scope a request is counted in, and its thresholds.
+type scope struct {
+	key        key
+	thresholds Thresholds
+}
+
+// scopesOf returns the scopes r is in whose thresholds are not all off.
+func (c *Counters) scopesOf(r Request) []scope {
+	p := c.params
+	client := r.Client.Unmap()
+
+	all := []scope{
+		{key{scope: ScopeClient, netblock: r.ClientGroup}, p.Client},
+		{key{scope: ScopeNet, netblock: c.netAround(client)}, p.Net},
+		{key{scope: ScopeTotal}, p.Total},
+	}
+
+	if r.ASN != "" {
+		all = append(all, scope{key{scope: ScopeASN, asn: r.ASN}, p.ASN})
+	}
+
+	for _, named := range p.NamedNetblocks {
+		if named.Netblock.Contains(client) {
+			all = append(all, scope{
+				key{scope: ScopeWatch, netblock: named.Netblock, name: named.Name}, p.Watch,
+			})
+		}
+	}
+
+	return slices.DeleteFunc(all, func(s scope) bool {
+		return s.thresholds == Thresholds{}
+	})
+}
+
+// netAround returns the netblock around client that ScopeNet counts it
+// in: NetV4Prefix or NetV6Prefix long.
+func (c *Counters) netAround(client netip.Addr) netip.Prefix {
+	length := c.params.NetV6Prefix
+	if client.Is4() {
+		length = c.params.NetV4Prefix
+	}
+
+	return netip.PrefixFrom(client, length).Masked()
+}
+
+// overThreshold is a count over its threshold: what it counts, requests or
+// bytes, its window, the count and the threshold.
+type overThreshold struct {
+	kind, window string
+	count        float64
+	threshold    int64
+}
+
+// add counts a request of bytes at now in each of c's counts whose
+// threshold, in thresholds, is set, and returns the first count over its
+// threshold, and whether there is one.
+func (c *Counter) add(
+	now time.Time, bytes int64, thresholds Thresholds,
+) (overThreshold, bool) {
+	// In the order of counts.
+	inOrder := [4]int64{
+		thresholds.RequestsPerMinute, thresholds.BytesPerMinute,
+		thresholds.RequestsPerHour, thresholds.BytesPerHour,
+	}
+
+	var (
+		first  overThreshold
+		passed bool
+	)
+
+	for i, count := range c.counts() {
+		threshold := inOrder[i]
+		if threshold == 0 {
+			continue
+		}
+
+		n := int64(1)
+		if count.kind == ratelimit.KindBytes {
+			n = bytes
+		}
+
+		counted := count.buckets.Add(now, count.length, n)
+		if !passed && counted > float64(threshold) {
+			first = overThreshold{count.kind, count.window, counted, threshold}
+			passed = true
+		}
+	}
+
+	return first, passed
+}
+
+// bucketCount is one of a counter's four counts: requests or bytes, in a
+// window of length, and the buckets they are counted in.
+type bucketCount struct {
+	kind, window string
+	length       time.Duration
+	buckets      *ratelimit.Buckets
+}
+
+// counts returns c's counts: requests and bytes in the minute, then in
+// the hour.
+func (c *Counter) counts() [4]bucketCount {
+	return [4]bucketCount{
+		{ratelimit.KindRequests, minute, time.Minute, &c.Minute},
+		{ratelimit.KindBytes, minute, time.Minute, &c.MinuteBytes},
+		{ratelimit.KindRequests, hour, time.Hour, &c.Hour},
+		{ratelimit.KindBytes, hour, time.Hour, &c.HourBytes},
+	}
+}
+
+// lastStart returns the start of c's latest bucket, which tells, to the
+// minute or to the hour, when c was last counted.
+func (c *Counter) lastStart() time.Time {
+	var latest time.Time
+
+	for _, count := range c.counts() {
+		if count.buckets.Start.After(latest) {
+			latest = count.buckets.Start
+		}
+	}
+
+	return latest
+}
+
+// key returns what tells c from the other counters.
+func (c *Counter) key() key {
+	return key{scope: c.Scope, netblock: c.Netblock, asn: c.ASN, name: c.Name}
+}
+
+// counter returns a counter for k, with nothing counted yet.
+func (k key) counter() *Counter {
+	return &Counter{Scope: k.scope, Netblock: k.netblock, ASN: k.asn, Name: k.name}
+}
+
+// alertFor returns the anomaly alert for o, a count over its threshold in
+// the scope k, which r took over it. It gives r's client, with its AS
+// number, AS name and country, and the netblock counted, of a client, the
+// netblock around it or a named netblock. Its detail gives the scope, the
+// AS number or the name of a scope that has one, the window, what is
+// counted, the count and the threshold.
+func alertFor(r Request, k key, o overThreshold) alerts.Alert {
+	detail := map[string]any{
+		"scope": k.scope, "window": o.window, "kind": o.kind, "count": o.count,
+		"threshold": o.threshold,
+	}
+
+	var counted string
+
+	switch k.scope {
+	case ScopeClient:
+		counted = "the client " + k.netblock.String()
+	case ScopeNet:
+		counted = "the netblock " + k.netblock.String()
+	case ScopeASN:
+		counted = k.asn
+		detail["asn"] = k.asn
+	case ScopeTotal:
+		counted = "the whole service"
+	default: // watch
+		counted = "the named netblock " + k.name + ", " + k.netblock.String()
+		detail["name"] = k.name
+	}
+
+	return alerts.Alert{
+		Event:    alerts.EventAnomaly,
+		Client:   r.Client,
+		Netblock: k.netblock,
+		ASN:      r.ASN,
+		ASName:   r.ASName,
+		Country:  r.Country,
+		Reason: fmt.Sprintf("%s per %s of %s over the threshold of %d", o.kind, o.window,
+			counted, o.threshold),
+		Detail: detail,
+	}
+}

internal/anomaly/anomaly_test.go

@@ -0,0 +1,238 @@
+package anomaly_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/netip"
+	"net/url"
+	"reflect"
+	"slices"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/anomaly"
+	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
+)
+
+// maxCounters is how many counters are kept.
+const maxCounters = 20000
+
+func TestEachScopeHasACooldownOfItsOwn(t *testing.T) {
+	t.Parallel()
+
+	queue := newQueue()
+	office := netip.MustParsePrefix("203.0.113.0/24")
+	overAtTheSecond := anomaly.Thresholds{RequestsPerMinute: 1}
+	counters := anomaly.New(anomaly.Params{
+		Client: overAtTheSecond, Net: overAtTheSecond, ASN: overAtTheSecond,
+		Total: overAtTheSecond, Watch: overAtTheSecond,
+		// The netblock around a client is the client's own, and two names
+		// name one netblock.
+		NetV4Prefix: 32,
+		NamedNetblocks: []anomaly.NamedNetblock{
+			{Name: "office", Netblock: office}, {Name: "hq", Netblock: office},
+		},
+		Alerts: queue,
+	})
+
+	// The first client's second request is over the threshold in the six
+	// scopes it is in. The other client's two are both over it in the whole
+	// service and in each named netblock, three repeats each, and its
+	// second is over it in the scopes of its own, its client, its netblock
+	// and its AS number, which are no repeats.
+	for _, r := range []anomaly.Request{
+		{Client: netip.MustParseAddr("203.0.113.9"), ASN: "AS64496"},
+		{Client: netip.MustParseAddr("203.0.113.10"), ASN: "AS64511"},
+	} {
+		r.ClientGroup = netip.PrefixFrom(r.Client, 32)
+
+		for range 2 {
+			counters.Count(midnight(), r)
+		}
+	}
+
+	waiting := queue.Snapshot().Waiting[alerts.DestinationWebhook]
+	if len(waiting) != 9 || queue.Suppressed() != 6 {
+		t.Fatalf("%d alerts wait and %d are held back, want 9 and 6: %+v",
+			len(waiting), queue.Suppressed(), waiting)
+	}
+
+	// alerts.json keeps each scope's cooldown: each alert raised again
+	// after a restart is a repeat.
+	data, err := json.Marshal(queue.Snapshot())
+	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)
+	}
+
+	after := newQueue()
+	after.Load(read)
+
+	for _, alert := range read.Waiting[alerts.DestinationWebhook] {
+		after.Raise(alert)
+	}
+
+	if after.Suppressed() != 9 {
+		t.Errorf("after loading, %d alerts are held back, want 9", after.Suppressed())
+	}
+}
+
+func TestKeepsAtMost20000CountersDroppingTheLeastRecentlyCounted(t *testing.T) {
+	t.Parallel()
+
+	counters := newCounters(anomaly.Params{
+		Client: anomaly.Thresholds{RequestsPerMinute: 1000},
+	})
+
+	for i := range maxCounters {
+		counters.Count(midnight(), request(i))
+	}
+
+	// Counted again, the first client is the most recently counted, and
+	// the second is dropped for a new one.
+	counters.Count(midnight(), request(0))
+	counters.Count(midnight(), request(maxCounters))
+
+	got := counters.Snapshot()
+	if len(got) != maxCounters || !holds(got, 0) || holds(got, 1) ||
+		!holds(got, maxCounters) {
+		t.Errorf("%d counters, holding the first client %v, the second %v and the "+
+			"new one %v, want %d, the first and the new one", len(got), holds(got, 0),
+			holds(got, 1), holds(got, maxCounters), maxCounters)
+	}
+}
+
+func TestLoadEmptiesBucketsWhoseTimeHasPassedAndDropsEmptyCounters(t *testing.T) {
+	t.Parallel()
+
+	counters := newCounters(anomaly.Params{
+		Net:         anomaly.Thresholds{RequestsPerMinute: 1000, RequestsPerHour: 1000},
+		Total:       anomaly.Thresholds{RequestsPerMinute: 1000},
+		NetV4Prefix: 24,
+	})
+	halfAnHourOn := midnight().Add(30 * time.Minute)
+
+	// Half an hour on, the hour's buckets count still, and the minute's
+	// do not.
+	counters.Load([]anomaly.Counter{
+		{
+			Scope:    anomaly.ScopeNet,
+			Netblock: netip.MustParsePrefix("203.0.113.9/24"),
+			Minute:   ratelimit.Buckets{Start: midnight(), Current: 5},
+			Hour:     ratelimit.Buckets{Start: midnight(), Current: 7},
+		},
+		{
+			Scope:  anomaly.ScopeTotal,
+			Minute: ratelimit.Buckets{Start: midnight(), Current: 1},
+		},
+	}, halfAnHourOn)
+
+	// The whole service's counter, left empty, is dropped, and the
+	// netblock read is masked to its length.
+	netblock := anomaly.Counter{
+		Scope:    anomaly.ScopeNet,
+		Netblock: netip.MustParsePrefix("203.0.113.0/24"),
+		Hour:     ratelimit.Buckets{Start: midnight(), Current: 7},
+	}
+	if got, want := counters.Snapshot(), []anomaly.Counter{netblock}; !reflect.DeepEqual(
+		got, want) {
+		t.Errorf("counters read\n%+v\nwant\n%+v", got, want)
+	}
+
+	// A request from the netblock is counted with the requests read.
+	counters.Count(halfAnHourOn, anomaly.Request{
+		Client:      netip.MustParseAddr("203.0.113.9"),
+		ClientGroup: netip.MustParsePrefix("203.0.113.9/32"),
+	})
+
+	netblock.Minute = ratelimit.Buckets{Start: halfAnHourOn, Current: 1}
+	netblock.Hour.Current = 8
+	want := []anomaly.Counter{netblock, {
+		Scope:  anomaly.ScopeTotal,
+		Minute: ratelimit.Buckets{Start: halfAnHourOn, Current: 1},
+	}}
+
+	if got := counters.Snapshot(); !reflect.DeepEqual(got, want) {
+		t.Errorf("counters after a request\n%+v\nwant\n%+v", got, want)
+	}
+}
+
+func TestLoadDropsTheLeastRecentlyCountedFirst(t *testing.T) {
+	t.Parallel()
+
+	counters := newCounters(anomaly.Params{
+		Client: anomaly.Thresholds{RequestsPerMinute: 1000},
+	})
+	now := midnight().Add(time.Minute)
+
+	// The second half of the file was counted in the minute before the
+	// first half.
+	read := make([]anomaly.Counter, 0, maxCounters)
+	for i := range maxCounters {
+		start := now
+		if i >= maxCounters/2 {
+			start = midnight()
+		}
+
+		read = append(read, anomaly.Counter{
+			Scope: anomaly.ScopeClient, Netblock: request(i).ClientGroup,
+			Minute: ratelimit.Buckets{Start: start, Current: 1},
+		})
+	}
+
+	counters.Load(read, now)
+	counters.Count(now, request(maxCounters))
+
+	got := counters.Snapshot()
+	if !holds(got, 0) || holds(got, maxCounters/2) {
+		t.Errorf("holding the first client of the file %v, and the first counted in "+
+			"the minute before %v, want only the first", holds(got, 0),
+			holds(got, maxCounters/2))
+	}
+}
+
+// midnight is the time of the tests' requests.
+func midnight() time.Time {
+	return time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)
+}
+
+// newCounters returns Counters for params, whose alerts go nowhere.
+func newCounters(params anomaly.Params) *anomaly.Counters {
+	params.Alerts = alerts.New(alerts.Params{})
+
+	return anomaly.New(params)
+}
+
+// newQueue returns a queue of alerts to a webhook, with the default
+// cooldown, which keeps them waiting, since it is never run.
+func newQueue() *alerts.Queue {
+	return alerts.New(alerts.Params{
+		WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"},
+		Events:     alerts.Events(),
+		Cooldown:   15 * time.Minute,
+		MaxPerHour: 60,
+		Now:        midnight,
+	})
+}
+
+// request returns a request from client number i, an address in
+// 10.0.0.0/8.
+func request(i int) anomaly.Request {
+	client := netip.MustParseAddr(fmt.Sprintf("10.%d.%d.%d", i>>16, i>>8&255, i&255))
+
+	return anomaly.Request{Client: client, ClientGroup: netip.PrefixFrom(client, 32)}
+}
+
+// holds reports whether counters hold the counter of client number i.
+func holds(counters []anomaly.Counter, i int) bool {
+	return slices.ContainsFunc(counters, func(counter anomaly.Counter) bool {
+		return counter.Netblock == request(i).ClientGroup
+	})
+}