From 87ec799e99f095f5052ef00c9be3911ffd503827 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Wed, 7 Oct 2026 12:49:18 +0000 Subject: [PATCH] Anomaly thresholds: alerts for unusual traffic, nothing refused (closes #101) Work in progress. Model: opus-5-5 --- internal/alerts/alerts.go | 58 ++-- internal/anomaly/anomaly.go | 406 ++++++++++++++++++++++++++++ internal/anomaly/anomaly_test.go | 238 ++++++++++++++++ internal/config/config.go | 120 +++++++- internal/config/config_test.go | 222 ++++++++++++++- internal/proxy/anomaly_test.go | 332 +++++++++++++++++++++++ internal/proxy/bans.go | 64 ++--- internal/proxy/biased_test.go | 27 +- internal/proxy/proxy.go | 19 +- internal/proxy/request.go | 63 +++-- internal/ratelimit/ratelimit.go | 28 +- internal/smallwebwaf/smallwebwaf.go | 1 + internal/state/state.go | 88 +++++- internal/state/state_test.go | 172 +++++++++++- 14 files changed, 1705 insertions(+), 133 deletions(-) create mode 100644 internal/anomaly/anomaly.go create mode 100644 internal/anomaly/anomaly_test.go create mode 100644 internal/proxy/anomaly_test.go diff --git a/internal/alerts/alerts.go b/internal/alerts/alerts.go index 5ce532a..3ee9f6b 100644 --- a/internal/alerts/alerts.go +++ b/internal/alerts/alerts.go @@ -1,5 +1,6 @@ -// Package alerts sends alerts on bans, on a source that fails and on a -// file with an error to each destination set: to the webhook +// Package alerts sends alerts on bans, on traffic over an anomaly +// threshold, on a source that fails and on a file with an error to each +// destination set: to the webhook // SWWAF_ALERT_WEBHOOK_URL names, each as one JSON object, as the "Alert // webhook schema" section of SPEC.md describes, to the Slack incoming // webhook SWWAF_ALERT_SLACK_WEBHOOK_URL names, as a message, and to the @@ -40,11 +41,12 @@ const ( // EventPermanentBan is a permanent ban smallwebwaf made, or a ban it // made permanent. EventPermanentBan = "permanent_ban" - // EventWAFBlock, EventAnomaly and EventReputationHit come with the - // Core Rule Set, the anomaly thresholds and the reputation sources; - // nothing raises them yet. + // EventAnomaly is a count of requests or bytes over an anomaly + // threshold. + EventAnomaly = "anomaly" + // EventWAFBlock and EventReputationHit come with the Core Rule Set and + // the reputation sources; nothing raises them yet. EventWAFBlock = "waf_block" - EventAnomaly = "anomaly" EventReputationHit = "reputation_hit" // EventSourceFailure is GeoJS failing or refusing smallwebwaf. EventSourceFailure = "source_failure" @@ -153,8 +155,9 @@ type Alert struct { ASName string `json:"as_name"` Country string `json:"country"` // Reason is a short sentence, and Detail what is particular to the - // event: for a file_error, its "file", and for a source_failure, its - // "source", which the cooldown tells repeats by. + // event: for a file_error, its "file", for a source_failure, its + // "source", and for an anomaly, its "scope", with the "asn" or the + // "name" of some scopes, which the cooldown tells repeats by. Reason string `json:"reason"` Detail map[string]any `json:"detail"` // SuppressedRepeats is how many repeats of the alert the cooldown @@ -162,9 +165,10 @@ type Alert struct { SuppressedRepeats int `json:"suppressed_repeats"` } -// Cooldown is, for an event on a netblock, or about a file or a source, -// when the last alert let through was raised, and how many repeats the -// cooldown has held back since, as alerts.json holds it. +// Cooldown is, for an event on a netblock, about a file or a source, or +// for an anomaly in a scope, when the last alert let through was raised, +// and how many repeats the cooldown has held back since, as alerts.json +// holds it. // //nolint:tagliatelle // the state files use snake_case, as the request log does type Cooldown struct { @@ -172,6 +176,9 @@ type Cooldown struct { Netblock netip.Prefix `json:"netblock"` File string `json:"file,omitempty"` Source string `json:"source,omitempty"` + Scope string `json:"scope,omitempty"` + ASN string `json:"asn,omitempty"` + Name string `json:"name,omitempty"` Sent time.Time `json:"sent"` SuppressedRepeats int `json:"suppressed_repeats"` } @@ -214,7 +221,7 @@ type Queue struct { mu sync.Mutex // cooldowns are the alerts last let through, by event and netblock, - // file or source. + // file, source or scope. cooldowns map[cooldownKey]*Cooldown hour Hour @@ -248,21 +255,28 @@ type destination struct { } // cooldownKey is what makes an alert a repeat of another: the same event -// on the same netblock, and about the same file or source, as its detail -// names them. Each is empty for an alert without one. +// on the same netblock, and about the same file or source, or in the same +// scope with the same AS number or name, as its detail names them. Each +// is empty for an alert without one. type cooldownKey struct { event string netblock netip.Prefix file string source string + scope string + asn string + name string } // cooldownKeyOf returns what makes another alert a repeat of alert. func cooldownKeyOf(alert *Alert) cooldownKey { file, _ := alert.Detail["file"].(string) source, _ := alert.Detail["source"].(string) + scope, _ := alert.Detail["scope"].(string) + asn, _ := alert.Detail["asn"].(string) + name, _ := alert.Detail["name"].(string) - return cooldownKey{alert.Event, alert.Netblock, file, source} + return cooldownKey{alert.Event, alert.Netblock, file, source, scope, asn, name} } // New returns a Queue with no alert yet. @@ -425,7 +439,8 @@ func (q *Queue) Suppressed() int64 { } // Snapshot returns the queue's state, as alerts.json holds it, with the -// cooldowns sorted by netblock, then by event, file and source. +// cooldowns sorted by netblock, then by event, file, source, scope, AS +// number and name. func (q *Queue) Snapshot() State { q.mu.Lock() defer q.mu.Unlock() @@ -443,7 +458,9 @@ func (q *Queue) Snapshot() State { slices.SortFunc(state.Cooldowns, func(a, b Cooldown) int { return cmp.Or(a.Netblock.Compare(b.Netblock), cmp.Compare(a.Event, b.Event), - cmp.Compare(a.File, b.File), cmp.Compare(a.Source, b.Source)) + cmp.Compare(a.File, b.File), cmp.Compare(a.Source, b.Source), + cmp.Compare(a.Scope, b.Scope), cmp.Compare(a.ASN, b.ASN), + cmp.Compare(a.Name, b.Name)) }) for _, d := range q.destinations { @@ -466,7 +483,10 @@ func (q *Queue) Load(state State) { for _, cooldown := range state.Cooldowns { cooldown.Netblock = cooldown.Netblock.Masked() - key := cooldownKey{cooldown.Event, cooldown.Netblock, cooldown.File, cooldown.Source} + key := cooldownKey{ + cooldown.Event, cooldown.Netblock, cooldown.File, cooldown.Source, + cooldown.Scope, cooldown.ASN, cooldown.Name, + } q.cooldowns[key] = &cooldown } @@ -538,7 +558,7 @@ func (q *Queue) startCooldown(alert *Alert, now time.Time) { q.cooldowns[key] = &Cooldown{ Event: alert.Event, Netblock: alert.Netblock, File: key.file, Source: key.source, - Sent: now, + Scope: key.scope, ASN: key.asn, Name: key.name, Sent: now, } } diff --git a/internal/anomaly/anomaly.go b/internal/anomaly/anomaly.go new file mode 100644 index 0000000..de54cfb --- /dev/null +++ b/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, + } +} diff --git a/internal/anomaly/anomaly_test.go b/internal/anomaly/anomaly_test.go new file mode 100644 index 0000000..ddb71ae --- /dev/null +++ b/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 + }) +} diff --git a/internal/config/config.go b/internal/config/config.go index f99aa51..b8f4ec6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,6 +24,7 @@ import ( "unicode/utf8" "sneak.berlin/go/smallwebwaf/internal/alerts" + "sneak.berlin/go/smallwebwaf/internal/anomaly" "sneak.berlin/go/smallwebwaf/internal/remotelog" ) @@ -220,6 +221,23 @@ type Config struct { AlertEvents []string AlertCooldown time.Duration AlertMaxPerHour int + // The anomaly thresholds, which only raise alerts: the most requests + // and bytes a minute and an hour per client (SWWAF_ANOMALY_CLIENT_*), + // per netblock around a client (SWWAF_ANOMALY_NET_*), per AS number + // (SWWAF_ANOMALY_ASN_*), for the whole service (SWWAF_ANOMALY_TOTAL_*) + // and per named netblock (SWWAF_WATCH_*), each 0 while it is off. + // AnomalyNetV4Prefix and AnomalyNetV6Prefix are the lengths of the + // netblock around a client (SWWAF_ANOMALY_NET_V4_PREFIX and + // SWWAF_ANOMALY_NET_V6_PREFIX), and WatchNets the named netblocks + // (SWWAF_WATCH_NETS). + AnomalyClient anomaly.Thresholds + AnomalyNet anomaly.Thresholds + AnomalyASN anomaly.Thresholds + AnomalyTotal anomaly.Thresholds + AnomalyWatch anomaly.Thresholds + AnomalyNetV4Prefix int + AnomalyNetV6Prefix int + WatchNets []anomaly.NamedNetblock // settings are the values read, as given or by default, and the // files they were read from, for the log line at start. @@ -240,6 +258,7 @@ const ( mebibyte = 1 << 20 gibibyte = 1 << 30 ipv4Bits = 32 + ipv6Bits = 128 // minTokenLength is the fewest characters a token may have. minTokenLength = 32 // masked is what the log shows for a token that is set, and in place of @@ -287,6 +306,10 @@ var ( errNotBanResponse = errors.New("is not 403, 429 or close") errNotV4Prefix = errors.New( "is not the length of an IPv4 netblock, from 0 to 32, such as 24") + errNotV6Prefix = errors.New( + "is not the length of an IPv6 netblock, from 0 to 128, such as 48") + errNotNamedNetblock = errors.New( + "is not a name, = and a netblock, such as office=203.0.113.0/24") errNotAbsolutePath = errors.New( "is not an absolute path, such as /var/lib/smallwebwaf") errShortToken = errors.New("is shorter than 32 characters") @@ -398,8 +421,16 @@ func FromEnvironment(lookupEnv func(string) (string, bool)) (*Config, error) { AlertNtfyToken: env.secret("SWWAF_ALERT_NTFY_TOKEN"), AlertEvents: env.alertEvents("SWWAF_ALERT_EVENTS", strings.Join(alerts.Events(), ",")), - AlertCooldown: env.duration("SWWAF_ALERT_COOLDOWN", "15m"), - AlertMaxPerHour: env.numberOrOff("SWWAF_ALERT_MAX_PER_HOUR", "60"), + AlertCooldown: env.duration("SWWAF_ALERT_COOLDOWN", "15m"), + AlertMaxPerHour: env.numberOrOff("SWWAF_ALERT_MAX_PER_HOUR", "60"), + AnomalyClient: env.thresholds("SWWAF_ANOMALY_CLIENT_"), + AnomalyNet: env.thresholds("SWWAF_ANOMALY_NET_"), + AnomalyASN: env.thresholds("SWWAF_ANOMALY_ASN_"), + AnomalyTotal: env.thresholds("SWWAF_ANOMALY_TOTAL_"), + AnomalyWatch: env.thresholds("SWWAF_WATCH_"), + AnomalyNetV4Prefix: env.v4Prefix("SWWAF_ANOMALY_NET_V4_PREFIX", "24"), + AnomalyNetV6Prefix: env.v6Prefix("SWWAF_ANOMALY_NET_V6_PREFIX", "48"), + WatchNets: env.namedNetblocks("SWWAF_WATCH_NETS"), } cfg.LogRemoteAppName = env.appName("SWWAF_LOG_REMOTE_APP_NAME", @@ -666,9 +697,9 @@ func (e *environment) checkLookupDBPath(cfg *Config) { // checkCountriesAndLookups refuses a country on both country lists, and, // while SWWAF_LOOKUP_SOURCE is off, each setting that needs clients looked -// up: the country lists, SWWAF_ADD_LOOKUP_HEADERS, and the biased -// thresholds, of which SWWAF_UNKNOWN_LIMIT_PERCENT needs them only below -// 100, where it lowers a limit. +// up: the country lists, SWWAF_ADD_LOOKUP_HEADERS, the biased thresholds, +// of which SWWAF_UNKNOWN_LIMIT_PERCENT needs them only below 100, where it +// lowers a limit, and the anomaly thresholds per AS number. func (e *environment) checkCountriesAndLookups(cfg *Config) { for _, country := range cfg.ExclusivelyAllowedCountries { if slices.Contains(cfg.DeniedCountries, country) { @@ -693,6 +724,10 @@ func (e *environment) checkCountriesAndLookups(cfg *Config) { {"SWWAF_ASN_BYTES_PERCENT", len(cfg.ASNBytesPercent) > 0}, {"SWWAF_COUNTRY_BYTES_PERCENT", len(cfg.CountryBytesPercent) > 0}, {"SWWAF_UNKNOWN_LIMIT_PERCENT", cfg.UnknownLimitPercent < 100}, + {"SWWAF_ANOMALY_ASN_REQUESTS_PER_MINUTE", cfg.AnomalyASN.RequestsPerMinute > 0}, + {"SWWAF_ANOMALY_ASN_REQUESTS_PER_HOUR", cfg.AnomalyASN.RequestsPerHour > 0}, + {"SWWAF_ANOMALY_ASN_BYTES_PER_MINUTE", cfg.AnomalyASN.BytesPerMinute > 0}, + {"SWWAF_ANOMALY_ASN_BYTES_PER_HOUR", cfg.AnomalyASN.BytesPerHour > 0}, } { if setting.set { e.check(setting.name, fmt.Errorf("is set while SWWAF_LOOKUP_SOURCE is off; %w", @@ -744,6 +779,35 @@ func (e *environment) v4Prefix(name, defaultValue string) int { return length } +// v6Prefix reads a setting that is the length of an IPv6 netblock. +func (e *environment) v6Prefix(name, defaultValue string) int { + length, err := parseV6Prefix(e.value(name, defaultValue)) + e.check(name, err) + + return length +} + +// thresholds reads the four anomaly thresholds whose settings' names +// start with prefix: requests and bytes per minute and per hour. Each is +// off by default. +func (e *environment) thresholds(prefix string) anomaly.Thresholds { + return anomaly.Thresholds{ + RequestsPerMinute: e.count(prefix+"REQUESTS_PER_MINUTE", off), + RequestsPerHour: e.count(prefix+"REQUESTS_PER_HOUR", off), + BytesPerMinute: e.size(prefix+"BYTES_PER_MINUTE", off), + BytesPerHour: e.size(prefix+"BYTES_PER_HOUR", off), + } +} + +// namedNetblocks reads a setting that is a list of named netblocks. It is +// empty by default. +func (e *environment) namedNetblocks(name string) []anomaly.NamedNetblock { + named, err := parseNamedNetblocks(e.value(name, "")) + e.check(name, err) + + return named +} + // absolutePath reads a setting that is an absolute path. func (e *environment) absolutePath(name, defaultValue string) string { path := e.value(name, defaultValue) @@ -1085,6 +1149,16 @@ func parseV4Prefix(value string) (int, error) { return n, nil } +// parseV6Prefix reads the length of an IPv6 netblock, from 0 to 128. +func parseV6Prefix(value string) (int, error) { + n, err := strconv.Atoi(value) + if err != nil || n < 0 || n > ipv6Bits { + return 0, fmt.Errorf("%q %w", value, errNotV6Prefix) + } + + return n, nil +} + // parseList splits a comma-separated list and trims the spaces around // each item. An empty value is an empty list. func parseList(value string) ([]string, error) { @@ -1144,6 +1218,42 @@ func parseNetblock(value string) (netip.Prefix, error) { return netip.PrefixFrom(addr, addr.BitLen()), nil } +// parseNamedNetblocks reads a comma-separated list of named netblocks, +// each a name, = and a netblock, such as office=203.0.113.0/24. An empty +// value is an empty list. A name listed twice is an error. +func parseNamedNetblocks(value string) ([]anomaly.NamedNetblock, error) { + items, err := parseList(value) + if err != nil { + return nil, err + } + + named := make([]anomaly.NamedNetblock, 0, len(items)) + + for _, item := range items { + name, netblockText, found := strings.Cut(item, "=") + name = strings.TrimSpace(name) + + if !found || name == "" { + return nil, fmt.Errorf("%q %w", item, errNotNamedNetblock) + } + + netblock, err := parseNetblock(strings.TrimSpace(netblockText)) + if err != nil { + return nil, err + } + + if slices.ContainsFunc(named, func(n anomaly.NamedNetblock) bool { + return n.Name == name + }) { + return nil, fmt.Errorf("%q %w", name, errListedTwice) + } + + named = append(named, anomaly.NamedNetblock{Name: name, Netblock: netblock}) + } + + return named, nil +} + // parsePathPrefixes reads a comma-separated list of path prefixes, each // starting with /. func parsePathPrefixes(value string) ([]string, error) { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index f52919b..1467277 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -12,10 +12,12 @@ import ( "path/filepath" "reflect" "slices" + "strconv" "strings" "testing" "time" + "sneak.berlin/go/smallwebwaf/internal/anomaly" "sneak.berlin/go/smallwebwaf/internal/config" ) @@ -85,8 +87,59 @@ const ( alertEvents = "SWWAF_ALERT_EVENTS" alertCooldown = "SWWAF_ALERT_COOLDOWN" alertMaxPerHour = "SWWAF_ALERT_MAX_PER_HOUR" + anomalyNetV4Prefix = "SWWAF_ANOMALY_NET_V4_PREFIX" + anomalyNetV6Prefix = "SWWAF_ANOMALY_NET_V6_PREFIX" + watchNets = "SWWAF_WATCH_NETS" ) +// The anomaly thresholds: each of the prefixes below, which name a scope, +// followed by each of the four ends. +const ( + anomalyClient = "SWWAF_ANOMALY_CLIENT_" + anomalyNet = "SWWAF_ANOMALY_NET_" + anomalyASN = "SWWAF_ANOMALY_ASN_" + anomalyTotal = "SWWAF_ANOMALY_TOTAL_" + watch = "SWWAF_WATCH_" + + requestsPerMinute = "REQUESTS_PER_MINUTE" + requestsPerHour = "REQUESTS_PER_HOUR" + bytesPerMinute = "BYTES_PER_MINUTE" + bytesPerHour = "BYTES_PER_HOUR" +) + +// anomalyScopes returns the prefixes of the anomaly thresholds, one for +// each scope. +func anomalyScopes() []string { + return []string{anomalyClient, anomalyNet, anomalyASN, anomalyTotal, watch} +} + +// anomalyThresholds returns the names of the twenty anomaly thresholds. +func anomalyThresholds() []string { + ends := []string{requestsPerMinute, requestsPerHour, bytesPerMinute, bytesPerHour} + names := make([]string, 0, len(anomalyScopes())*len(ends)) + + for _, scope := range anomalyScopes() { + for _, end := range ends { + names = append(names, scope+end) + } + } + + return names +} + +// loggedAnomalyDefaults returns the anomaly settings as the settings +// logged at start give them by default. +func loggedAnomalyDefaults() map[string]string { + logged := map[string]string{ + anomalyNetV4Prefix: "24", anomalyNetV6Prefix: "48", watchNets: "", + } + for _, name := range anomalyThresholds() { + logged[name] = off + } + + return logged +} + // defaultAlertEvents is the default of SWWAF_ALERT_EVENTS, and // defaultAlertCooldown that of SWWAF_ALERT_COOLDOWN. const ( @@ -897,14 +950,18 @@ func TestSettingNeedingLookupsStopsTheStartWhileTheyAreOff(t *testing.T) { t.Parallel() for name, value := range map[string]string{ - deniedCountries: "kp", - allowedCountries: "de", - addLookupHeaders: enabled, - asnLimitPercent: "AS64496:50", - countryLimitPercent: "cn:25", - asnBytesPercent: "AS64496:50", - countryBytesPercent: "cn:25", - unknownLimitPercent: "99", + deniedCountries: "kp", + allowedCountries: "de", + addLookupHeaders: enabled, + asnLimitPercent: "AS64496:50", + countryLimitPercent: "cn:25", + asnBytesPercent: "AS64496:50", + countryBytesPercent: "cn:25", + unknownLimitPercent: "99", + anomalyASN + requestsPerMinute: "1000", + anomalyASN + requestsPerHour: "10000", + anomalyASN + bytesPerMinute: "1G", + anomalyASN + bytesPerHour: "10G", } { t.Run(name, func(t *testing.T) { t.Parallel() @@ -920,12 +977,153 @@ func TestSettingNeedingLookupsStopsTheStartWhileTheyAreOff(t *testing.T) { } // Set empty, the lists need nothing looked up, and nor does - // SWWAF_UNKNOWN_LIMIT_PERCENT at 100, which lowers no limit. - fromEnvironment(t, environment{ + // SWWAF_UNKNOWN_LIMIT_PERCENT at 100, which lowers no limit, an anomaly + // threshold per AS number that is off, or any other anomaly threshold. + env := environment{ lookupSource: off, deniedCountries: "", allowedCountries: "", asnLimitPercent: "", countryLimitPercent: "", asnBytesPercent: "", countryBytesPercent: "", unknownLimitPercent: "100", - }) + } + for _, name := range anomalyThresholds() { + env[name] = "1000" + if strings.HasPrefix(name, anomalyASN) { + env[name] = off + } + } + + fromEnvironment(t, env) +} + +func TestAnomalySettingsDefaults(t *testing.T) { + t.Parallel() + + cfg := fromEnvironment(t, environment{}) + wantAllOff(t, cfg) + + if cfg.AnomalyNetV4Prefix != 24 || cfg.AnomalyNetV6Prefix != 48 || + len(cfg.WatchNets) != 0 { + t.Errorf("%s, %s and %s gave %d, %d and %v, want 24, 48 and none", + anomalyNetV4Prefix, anomalyNetV6Prefix, watchNets, cfg.AnomalyNetV4Prefix, + cfg.AnomalyNetV6Prefix, cfg.WatchNets) + } +} + +func TestAnomalySettingsAsSet(t *testing.T) { + t.Parallel() + + // Each threshold of a scope its own value; bytes are sizes. + env := environment{ + anomalyNetV4Prefix: "16", + anomalyNetV6Prefix: "56", + // Spaces around a name or a netblock, and a bare address. + watchNets: "office = 203.0.113.0/24, scraper-x=198.51.100.7,v6=2001:db8::/32", + } + want := map[string]anomaly.Thresholds{} + + for i, scope := range anomalyScopes() { + n := int64(i + 1) + env[scope+requestsPerMinute] = strconv.FormatInt(n, 10) + env[scope+requestsPerHour] = strconv.FormatInt(10*n, 10) + env[scope+bytesPerMinute] = strconv.FormatInt(n, 10) + "K" + env[scope+bytesPerHour] = strconv.FormatInt(n, 10) + "G" + want[scope] = anomaly.Thresholds{ + RequestsPerMinute: n, RequestsPerHour: 10 * n, + BytesPerMinute: n << 10, BytesPerHour: n << 30, + } + } + + cfg := fromEnvironment(t, env) + + if got := thresholdsByScope(cfg); !maps.Equal(got, want) { + t.Errorf("thresholds by scope\n%+v\nwant\n%+v", got, want) + } + + wantNamed := []anomaly.NamedNetblock{ + {Name: "office", Netblock: netip.MustParsePrefix("203.0.113.0/24")}, + {Name: "scraper-x", Netblock: netip.MustParsePrefix("198.51.100.7/32")}, + {Name: "v6", Netblock: netip.MustParsePrefix("2001:db8::/32")}, + } + if cfg.AnomalyNetV4Prefix != 16 || cfg.AnomalyNetV6Prefix != 56 || + !slices.Equal(cfg.WatchNets, wantNamed) { + t.Errorf("%s, %s and %s gave %d, %d and %v, want 16, 56 and %v", + anomalyNetV4Prefix, anomalyNetV6Prefix, watchNets, cfg.AnomalyNetV4Prefix, + cfg.AnomalyNetV6Prefix, cfg.WatchNets, wantNamed) + } + + // off switches each threshold off. + for _, name := range anomalyThresholds() { + env[name] = off + } + + wantAllOff(t, fromEnvironment(t, env)) +} + +// thresholdsByScope returns cfg's anomaly thresholds, each by the prefix +// of its scope's settings. +func thresholdsByScope(cfg *config.Config) map[string]anomaly.Thresholds { + return map[string]anomaly.Thresholds{ + anomalyClient: cfg.AnomalyClient, anomalyNet: cfg.AnomalyNet, + anomalyASN: cfg.AnomalyASN, anomalyTotal: cfg.AnomalyTotal, + watch: cfg.AnomalyWatch, + } +} + +// wantAllOff checks that every anomaly threshold of cfg is off. +func wantAllOff(t *testing.T, cfg *config.Config) { + t.Helper() + + for scope, thresholds := range thresholdsByScope(cfg) { + if thresholds != (anomaly.Thresholds{}) { + t.Errorf("%s* gave %+v, want every one off", scope, thresholds) + } + } +} + +func TestInvalidAnomalySettingStopsTheStartSayingWhatIsWrong(t *testing.T) { + t.Parallel() + + const ( + notCount = " is not a whole number of requests such as 1000, or off" + notSize = " is not a size such as 512K, 100M or 5G, or off" + notPositive = " must be more than zero, or off" + notNamed = " is not a name, = and a netblock, such as office=203.0.113.0/24" + notNetblock = " is not a netblock such as 10.0.0.0/8, or an address" + notV4Prefix = " is not the length of an IPv4 netblock, from 0 to 32, such as 24" + notV6Prefix = " is not the length of an IPv6 netblock, from 0 to 128, such as 48" + officeNetblock = "office=203.0.113.0/24" + scraperNetblock = "scraper=198.51.100.0/24" + ) + + for _, tc := range []struct{ name, value, want string }{ + {anomalyClient + requestsPerMinute, "1K", `"1K"` + notCount}, + {anomalyNet + requestsPerHour, "0", `"0"` + notPositive}, + {anomalyTotal + bytesPerMinute, "1T", `"1T"` + notSize}, + {watch + bytesPerHour, "-1G", `"-1G"` + notPositive}, + {anomalyNetV4Prefix, "33", `"33"` + notV4Prefix}, + {anomalyNetV4Prefix, off, `"off"` + notV4Prefix}, + {anomalyNetV6Prefix, "129", `"129"` + notV6Prefix}, + {anomalyNetV6Prefix, "/48", `"/48"` + notV6Prefix}, + {watchNets, "office", `"office"` + notNamed}, + {watchNets, "=203.0.113.0/24", `"=203.0.113.0/24"` + notNamed}, + {watchNets, "office=203.0.113.300/24", `"203.0.113.300/24"` + notNetblock}, + {watchNets, officeNetblock + ",", `"` + officeNetblock + `," has an empty item ` + + `in its list`}, + { + watchNets, officeNetblock + "," + scraperNetblock + ",office=192.0.2.0/24", + `"office" is listed twice`, + }, + } { + t.Run(tc.name+"="+tc.value, func(t *testing.T) { + t.Parallel() + + _, err := config.FromEnvironment(environment{tc.name: tc.value}.lookupEnv) + + want := tc.name + ": " + tc.want + if err == nil || err.Error() != want { + t.Errorf("error %v, want %s", err, want) + } + }) + } } func TestBiasedThresholdsAsSet(t *testing.T) { @@ -1461,6 +1659,8 @@ func TestLogsEachSettingWithItsValue(t *testing.T) { alertCooldown: defaultAlertCooldown, alertMaxPerHour: "60", } + maps.Copy(want, loggedAnomalyDefaults()) + if got := loggedSettings(t, cfg); !maps.Equal(got, want) { t.Errorf("logged settings\n%v\nwant\n%v", got, want) } diff --git a/internal/proxy/anomaly_test.go b/internal/proxy/anomaly_test.go new file mode 100644 index 0000000..169dd9f --- /dev/null +++ b/internal/proxy/anomaly_test.go @@ -0,0 +1,332 @@ +package proxy_test + +import ( + "maps" + "net/http" + "net/netip" + "reflect" + "slices" + "strconv" + "testing" + "time" + + "sneak.berlin/go/smallwebwaf/internal/alerts" + "sneak.berlin/go/smallwebwaf/internal/anomaly" + "sneak.berlin/go/smallwebwaf/internal/ratelimit" + "sneak.berlin/go/smallwebwaf/internal/requestlog" +) + +// The anomaly thresholds: the prefix of a scope followed by the end of a +// count. +const ( + anomalyClient = "SWWAF_ANOMALY_CLIENT_" + anomalyNet = "SWWAF_ANOMALY_NET_" + anomalyASN = "SWWAF_ANOMALY_ASN_" + anomalyTotal = "SWWAF_ANOMALY_TOTAL_" + anomalyWatch = "SWWAF_WATCH_" + + requestsPerMinute = "REQUESTS_PER_MINUTE" + requestsPerHour = "REQUESTS_PER_HOUR" + bytesPerMinute = "BYTES_PER_MINUTE" + bytesPerHour = "BYTES_PER_HOUR" +) + +// The other anomaly settings. +const ( + anomalyNetV4Prefix = "SWWAF_ANOMALY_NET_V4_PREFIX" + anomalyNetV6Prefix = "SWWAF_ANOMALY_NET_V6_PREFIX" + watchNets = "SWWAF_WATCH_NETS" +) + +const ( + // clientsNet is the netblock around client at the default length, and + // office a named netblock of the same. + clientsNet = "203.0.113.0/24" + office = "office=" + clientsNet + // aLot is a threshold no test reaches. + aLot = "1000" + // hour is the window an alert names for a threshold per hour. + hour = "hour" +) + +func TestEachScopeAndWindowOverItsThresholdAlertsOncePerCooldown(t *testing.T) { + t.Parallel() + + for _, scope := range []struct { + prefix, scope string + // netblock is the alert's, and counted what its reason names. extra + // is what its detail gives besides what every anomaly alert's does. + netblock netip.Prefix + counted string + extra map[string]any + }{ + { + anomalyClient, anomaly.ScopeClient, netip.MustParsePrefix(client + "/32"), + "the client " + client + "/32", nil, + }, + { + anomalyNet, anomaly.ScopeNet, netip.MustParsePrefix(clientsNet), + "the netblock " + clientsNet, nil, + }, + {anomalyASN, anomaly.ScopeASN, netip.Prefix{}, asnDE, map[string]any{"asn": asnDE}}, + {anomalyTotal, anomaly.ScopeTotal, netip.Prefix{}, "the whole service", nil}, + { + anomalyWatch, anomaly.ScopeWatch, netip.MustParsePrefix(clientsNet), + "the named netblock office, " + clientsNet, map[string]any{"name": "office"}, + }, + } { + for _, threshold := range []struct { + end, kind, window string + // value is the threshold, which the third upload of 100 bytes + // takes the count over, to count. + value int64 + count float64 + }{ + {requestsPerMinute, ratelimit.KindRequests, minute, 2, 3}, + {requestsPerHour, ratelimit.KindRequests, hour, 2, 3}, + {bytesPerMinute, ratelimit.KindBytes, minute, 250, 300}, + {bytesPerHour, ratelimit.KindBytes, hour, 250, 300}, + } { + setting := scope.prefix + threshold.end + value := strconv.FormatInt(threshold.value, 10) + + t.Run(setting, func(t *testing.T) { + t.Parallel() + + s, clk, server, queue := startWithLookupsAndClock(t, map[string]string{ + setting: value, watchNets: office, + }) + start := clk.Now() + + // The third upload takes the count over the threshold, and the + // fourth, within the cooldown, is held back. Each is passed to + // the app. + for range 4 { + s.uploadFrom(client) + } + + detail := map[string]any{ + "scope": scope.scope, "window": threshold.window, "kind": threshold.kind, + "count": threshold.count, "threshold": threshold.value, + } + maps.Copy(detail, scope.extra) + + wantAlerts(t, queue, alerts.Alert{ + Instance: alertInstance, + Time: start, + Event: alerts.EventAnomaly, + Client: netip.MustParseAddr(client), + Netblock: scope.netblock, + ASN: asnDE, + ASName: asNameDE, + Country: "DE", + Reason: threshold.kind + " per " + threshold.window + " of " + + scope.counted + " over the threshold of " + value, + Detail: detail, + }) + + wantAlertedAgainOnceTheCooldownHasRunOut(t, s, clk, queue) + + if held := server.Ledger.Snapshot(); len(held) != 0 { + t.Errorf("the ledger holds %+v, want no ban", held) + } + }) + } + } +} + +// wantAlertedAgainOnceTheCooldownHasRunOut checks that, once the cooldown +// has run out after a first alert, which held back one repeat, the next +// count over the threshold, at the latest three uploads from client on, +// raises another alert, giving that repeat. +func wantAlertedAgainOnceTheCooldownHasRunOut( + t *testing.T, s *sender, clk *clock, queue *alerts.Queue, +) { + t.Helper() + + clk.advance(15 * time.Minute) + + for range 3 { + s.uploadFrom(client) + } + + waiting := queue.Snapshot().Waiting[alerts.DestinationWebhook] + if len(waiting) != 2 || !waiting[1].Time.Equal(clk.Now()) || + waiting[1].SuppressedRepeats != 1 { + t.Errorf("alerts wait %+v, want the first and another, with 1 repeat", waiting) + } +} + +func TestEveryRequestIsCountedWhateverIsDoneWithIt(t *testing.T) { + t.Parallel() + + const ( + allowed = "192.0.2.7" // in SWWAF_ALLOW_NETS + exempt = "192.0.2.10" // in SWWAF_RATE_LIMIT_EXEMPT_NETS + denied = "192.0.2.20" // in SWWAF_DENY_NETS + ) + + s, _, _, queue := startAppWithAlerts(t, readAndAnswer, map[string]string{ + anomalyClient + requestsPerMinute: "2", + allowNets: allowed, + rateLimitExemptNets: exempt, + rateLimitExemptPaths: "/static/", + denyNets: denied, + }) + + // The third request of each takes its client's count over the threshold + // of 2. + for _, sent := range []struct { + from, path string + status int + action string + }{ + {allowed, "/", http.StatusOK, requestlog.ActionForward}, + {exempt, "/", http.StatusOK, requestlog.ActionForward}, + {client, "/static/app.js", http.StatusOK, requestlog.ActionForward}, + {denied, "/", http.StatusForbidden, requestlog.ActionDenied}, + } { + for range 3 { + s.request(sent.from, sent.path, sent.status, sent.action) + } + } + + waiting := queue.Snapshot().Waiting[alerts.DestinationWebhook] + + got := make([]string, 0, len(waiting)) + for _, alert := range waiting { + got = append(got, alert.Client.String()) + } + + if want := []string{allowed, exempt, client, denied}; !slices.Equal(got, want) { + t.Errorf("alerts for the clients %v, want %v", got, want) + } +} + +func TestThresholdsOffCountNothingAndAlertNothing(t *testing.T) { + t.Parallel() + + // With every threshold off, nothing is counted. + s, server, queue := startWithLookups(t, map[string]string{watchNets: office}) + + for range 5 { + s.uploadFrom(client) + } + + if counters := server.Anomalies.Snapshot(); len(counters) != 0 { + t.Errorf("counters %+v, want none", counters) + } + + wantAlerts(t, queue) + + // With one set, its count alone is counted, in its scope alone. + s, clk, server, queue := startWithLookupsAndClock(t, map[string]string{ + anomalyNet + requestsPerMinute: aLot, watchNets: office, + }) + + for range 5 { + s.uploadFrom(client) + } + + want := []anomaly.Counter{{ + Scope: anomaly.ScopeNet, + Netblock: netip.MustParsePrefix(clientsNet), + Minute: ratelimit.Buckets{Start: clk.Now(), Current: 5}, + }} + if got := server.Anomalies.Snapshot(); !reflect.DeepEqual(got, want) { + t.Errorf("counters\n%+v\nwant\n%+v", got, want) + } + + wantAlerts(t, queue) +} + +func TestNetblockAroundAClientIsAsLongAsTheSettingsSay(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + env map[string]string + // Each client of sent sends one request, and want gives the + // netblocks they are counted in, each with its requests. + sent []string + want map[string]int64 + }{ + { + "by default", nil, + []string{client, "203.0.113.200", "192.0.2.7", ipv6Client, "2001:db8:0:ffff::1"}, + map[string]int64{clientsNet: 2, "192.0.2.0/24": 1, "2001:db8::/48": 2}, + }, + { + "as set", map[string]string{anomalyNetV4Prefix: "16", anomalyNetV6Prefix: "32"}, + []string{client, "203.0.200.1", ipv6Client, "2001:db8:ffff::1"}, + map[string]int64{"203.0.0.0/16": 2, "2001:db8::/32": 2}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + env := map[string]string{anomalyNet + requestsPerMinute: aLot} + maps.Copy(env, tc.env) + + s, _, server, _ := startAppWithAlerts(t, readAndAnswer, env) + + for _, from := range tc.sent { + s.get(from, http.StatusOK, requestlog.ActionForward) + } + + got := map[string]int64{} + for _, counter := range server.Anomalies.Snapshot() { + got[counter.Netblock.String()] = counter.Minute.Current + } + + if !maps.Equal(got, tc.want) { + t.Errorf("requests by netblock %v, want %v", got, tc.want) + } + }) + } +} + +func TestClientIsCountedForItsASNumberOnceTheLookupGivesOne(t *testing.T) { + t.Parallel() + + s, server, _ := startWithLookups(t, map[string]string{ + anomalyASN + requestsPerMinute: aLot, + }) + + // The lookup database does not hold unplaced. + for _, from := range []string{fromDE, fromDE, fromKP, noCountry, unplaced} { + s.uploadFrom(from) + } + + got := map[string]int64{} + for _, counter := range server.Anomalies.Snapshot() { + got[counter.ASN] = counter.Minute.Current + } + + if want := map[string]int64{asnDE: 2, asnKP: 1, "AS64500": 1}; !maps.Equal(got, want) { + t.Errorf("requests by AS number %v, want %v", got, want) + } +} + +func TestEachNamedNetblockCountsTheClientsInIt(t *testing.T) { + t.Parallel() + + s, _, server, _ := startAppWithAlerts(t, readAndAnswer, map[string]string{ + anomalyWatch + requestsPerMinute: aLot, + watchNets: office + ",wide=203.0.0.0/16,other=198.51.100.0/25", + }) + + // client is in office and in wide. + for _, from := range []string{client, "203.0.200.1", "192.0.2.7"} { + s.get(from, http.StatusOK, requestlog.ActionForward) + } + + got := map[string]int64{} + for _, counter := range server.Anomalies.Snapshot() { + got[counter.Name] = counter.Minute.Current + } + + if want := map[string]int64{"office": 1, "wide": 2}; !maps.Equal(got, want) { + t.Errorf("requests by named netblock %v, want %v", got, want) + } +} diff --git a/internal/proxy/bans.go b/internal/proxy/bans.go index 0ecaf13..95f889a 100644 --- a/internal/proxy/bans.go +++ b/internal/proxy/bans.go @@ -56,44 +56,23 @@ func (rq *request) limitBroken(now time.Time) bool { return over } -// countBytes counts the request's bytes for the byte limits, once its -// response has ended, and notes the client's byte totals for the log line; -// its requests stay there as the rate limits counted them. The bytes are -// the response's body bytes, the request's, or both, as SWWAF_BYTES_COUNT -// says; for an upgraded connection, such as a WebSocket, which has closed -// by then, what it carried from the app counts with the response's and -// what it carried from the client with the request's. Only a request -// passed to the app has them counted, and only one the rate limits -// counted; in observe mode, not one that enforce mode would have refused. -// Bytes that take the client over a byte limit, as its limit percentage -// for the byte limits lowers it, break it; the response was passed on -// whole. +// countBytes counts the request's bytes, as countedBytes gives them, for +// the byte limits, once its response has ended, and notes the client's +// byte totals for the log line; its requests stay there as the rate limits +// counted them. Only a request passed to the app has them counted, and +// only one the rate limits counted; in observe mode, not one that enforce +// mode would have refused. Bytes that take the client over a byte limit, +// as its limit percentage for the byte limits lowers it, break it; the +// response was passed on whole. func (rq *request) countBytes() { if !rq.counted || rq.line.WouldAction != "" { return } - response, request := rq.out.bytes, rq.requestBytes() - if rq.upgraded != nil { - response += rq.upgraded.fromApp.Load() - request += rq.upgraded.toApp.Load() - } - - var bytes int64 - - switch rq.h.config.BytesCount { - case "response": - bytes = response - case "request": - bytes = request - default: // both - bytes = response + request - } - now := rq.h.now() - counts, hit, over := rq.h.limiter.CountBytes(clientGroup(rq.client), now, bytes, - rq.bytesPercent.percent) + counts, hit, over := rq.h.limiter.CountBytes(clientGroup(rq.client), now, + rq.countedBytes(), rq.bytesPercent.percent) rq.line.Counts.MinuteBytes = counts.MinuteBytes rq.line.Counts.HourBytes = counts.HourBytes rq.line.Counts.DayBytes = counts.DayBytes @@ -103,6 +82,29 @@ func (rq *request) countBytes() { } } +// countedBytes returns the request's bytes, once it has ended, as the +// byte limits and the anomaly thresholds count them: the response's body +// bytes, the request's, or both, as SWWAF_BYTES_COUNT says. For an +// upgraded connection, such as a WebSocket, which has closed by then, what +// it carried from the app counts with the response's and what it carried +// from the client with the request's. +func (rq *request) countedBytes() int64 { + response, request := rq.out.bytes, rq.requestBytes() + if rq.upgraded != nil { + response += rq.upgraded.fromApp.Load() + request += rq.upgraded.toApp.Load() + } + + switch rq.h.config.BytesCount { + case "response": + return response + case "request": + return request + default: // both + return response + request + } +} + // banForLimit bans the client's netblock at now for a broken limit, the // one hit names, and notes the offence for the log line. status is what // the client was sent, or is sent: SWWAF_BAN_RESPONSE for a request over diff --git a/internal/proxy/biased_test.go b/internal/proxy/biased_test.go index b74c695..9f4505f 100644 --- a/internal/proxy/biased_test.go +++ b/internal/proxy/biased_test.go @@ -421,17 +421,28 @@ func TestBanForALoweredLimitGivesThePercentageInItsNotesAndItsAlert(t *testing.T } } -// startWithLookups is startAppWithAlerts in front of readAndAnswer, with -// the settings in env on top of clients looked up in a lookup database, -// which places fromDE and fromKP in the AS numbers and countries the -// stand-in for GeoJS gives them, noCountry in AS64500 and no country, and -// no other address. It returns the sender, the server and the queue of -// the alerts. +// startWithLookups is startWithLookupsAndClock for a test that needs no +// clock. func startWithLookups( t *testing.T, env map[string]string, ) (*sender, *proxy.Server, *alerts.Queue) { t.Helper() + s, _, server, queue := startWithLookupsAndClock(t, env) + + return s, server, queue +} + +// startWithLookupsAndClock is startAppWithAlerts in front of +// readAndAnswer, with the settings in env on top of clients looked up in a +// lookup database, which places fromDE and fromKP in the AS numbers and +// countries the stand-in for GeoJS gives them, noCountry in AS64500 and no +// country, and no other address. +func startWithLookupsAndClock( + t *testing.T, env map[string]string, +) (*sender, *clock, *proxy.Server, *alerts.Queue) { + t.Helper() + path := filepath.Join(t.TempDir(), "ipinfo_lite.mmdb") lookuptest.Write(t, path, map[string]lookuptest.Network{ fromDE + "/32": {ASN: asnDE, ASName: asNameDE, Country: "DE"}, @@ -442,9 +453,7 @@ func startWithLookups( settings := map[string]string{lookupSource: fileSource, lookupDBPath: path} maps.Copy(settings, env) - s, _, server, queue := startAppWithAlerts(t, readAndAnswer, settings) - - return s, server, queue + return startAppWithAlerts(t, readAndAnswer, settings) } // uploadFrom is upload from the client at from. diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index f2b5671..4df438f 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -12,6 +12,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/config" "sneak.berlin/go/smallwebwaf/internal/lookup" @@ -68,7 +69,8 @@ type Params struct { // against. Rules *rules.Files // Alerts receive the alert for each ban the proxy makes or makes - // permanent, and for GeoJS failing. + // permanent, for each count over an anomaly threshold, and for GeoJS + // failing. Alerts *alerts.Queue } @@ -81,6 +83,7 @@ type Server struct { Ledger *bans.Ledger Limiter *ratelimit.Limiter GeoJS *lookup.GeoJS + Anomalies *anomaly.Counters LookupFile *lookup.File Metrics *metrics.Metrics } @@ -117,6 +120,17 @@ func New(params Params) *Server { AttackBanDuration: params.Config.AttackBanDuration, MaxBans: params.Config.MaxBans, }), + anomalies: anomaly.New(anomaly.Params{ + Client: params.Config.AnomalyClient, + Net: params.Config.AnomalyNet, + ASN: params.Config.AnomalyASN, + Total: params.Config.AnomalyTotal, + Watch: params.Config.AnomalyWatch, + NetV4Prefix: params.Config.AnomalyNetV4Prefix, + NetV6Prefix: params.Config.AnomalyNetV6Prefix, + NamedNetblocks: params.Config.WatchNets, + Alerts: params.Alerts, + }), lookupFile: params.LookupFile, rules: params.Rules, alerts: params.Alerts, @@ -154,6 +168,7 @@ func New(params Params) *Server { Ledger: h.ledger, Limiter: h.limiter, GeoJS: h.geojs, + Anomalies: h.anomalies, LookupFile: h.lookupFile, Metrics: m, } @@ -172,6 +187,7 @@ type handler struct { limiter *ratelimit.Limiter ledger *bans.Ledger geojs *lookup.GeoJS + anomalies *anomaly.Counters lookupFile *lookup.File rules *rules.Files alerts *alerts.Queue @@ -211,6 +227,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Once the request has ended, before its log line is written. defer rq.addToHistory() + defer rq.countAnomalies() refused := rq.check(r.Context()) rq.checked = time.Now() diff --git a/internal/proxy/request.go b/internal/proxy/request.go index d5452c4..cb1bee5 100644 --- a/internal/proxy/request.go +++ b/internal/proxy/request.go @@ -16,6 +16,7 @@ import ( "sync/atomic" "time" + "sneak.berlin/go/smallwebwaf/internal/anomaly" "sneak.berlin/go/smallwebwaf/internal/lookup" "sneak.berlin/go/smallwebwaf/internal/ratelimit" "sneak.berlin/go/smallwebwaf/internal/requestlog" @@ -513,16 +514,14 @@ func timing(start, end time.Time) *float64 { } // addToHistory adds the request, which has ended, to its client's -// history, and then, for a client that was looked up, the lookup -// database's answer about it, or the answer from GeoJS kept about it, to -// that history and to the notes of the bans on its netblock: an answer -// may have come before either was there, and one from GeoJS that comes -// later is added when it comes. +// history, and then the lookup's answer about the client, as +// answerAtTheEnd gives it, to that history and to the notes of the bans +// on its netblock: an answer may have come before either was there, and +// one from GeoJS that comes later is added when it comes. func (rq *request) addToHistory() { forwarded := !rq.upstreamStart.IsZero() - group := clientGroup(rq.client) - rq.h.limiter.AddToHistory(group, rq.h.now(), ratelimit.Request{ + rq.h.limiter.AddToHistory(clientGroup(rq.client), rq.h.now(), ratelimit.Request{ Forwarded: forwarded, Refused: !forwarded && rq.refused.Load() != nil, Status: rq.out.status, @@ -531,23 +530,47 @@ func (rq *request) addToHistory() { BrokeLimit: rq.line.Offence == requestlog.OffenceLimit, }) - if !rq.lookedUp { - return - } - - // The lookup database's answer was there at once. - if rq.h.config.LookupSource == "file" { - rq.h.addLookup(rq.lookupAnswer) - - return - } - - answer, kept := rq.h.geojs.Kept(group) - if kept { + answer, found := rq.answerAtTheEnd() + if found { rq.h.addLookup(answer) } } +// countAnomalies counts the request, which has ended, and its bytes, as +// countedBytes gives them, for the anomaly thresholds, whatever was done +// with it: a request refused, one from a client in SWWAF_ALLOW_NETS or +// SWWAF_RATE_LIMIT_EXEMPT_NETS, and one for a path in +// SWWAF_RATE_LIMIT_EXEMPT_PATHS are counted too. It is counted for its +// client's AS number when answerAtTheEnd gives one. +func (rq *request) countAnomalies() { + answer, _ := rq.answerAtTheEnd() + + rq.h.anomalies.Count(rq.h.now(), anomaly.Request{ + Client: rq.client, + ClientGroup: clientGroup(rq.client), + ASN: answer.ASN, + ASName: answer.ASName, + Country: answer.Country, + Bytes: rq.countedBytes(), + }) +} + +// answerAtTheEnd returns, for a client that was looked up, the lookup's +// answer about it as the request ends, and whether there is one: the +// lookup database's, which was there at once, or the one GeoJS has given +// by then, which a request does not wait for unless a setting needs it. +func (rq *request) answerAtTheEnd() (lookup.Answer, bool) { + if !rq.lookedUp { + return lookup.Answer{}, false + } + + if rq.h.config.LookupSource == "file" { + return rq.lookupAnswer, true + } + + return rq.h.geojs.Kept(clientGroup(rq.client)) +} + // requestBytes is how many bytes of the request's body have been read. func (rq *request) requestBytes() int64 { if rq.body == nil { diff --git a/internal/ratelimit/ratelimit.go b/internal/ratelimit/ratelimit.go index f184063..fa91f97 100644 --- a/internal/ratelimit/ratelimit.go +++ b/internal/ratelimit/ratelimit.go @@ -361,9 +361,7 @@ func (l *Limiter) Load(clients []Client, now time.Time) { for _, c := range clients { for i, w := range l.windows { for _, b := range []*Buckets{c.buckets()[i], c.byteBuckets()[i]} { - // The window that ends at now covers neither bucket once it - // begins after the bucket under way has ended. - if !now.Add(-w.length).Before(b.Start.Add(w.length)) { + if b.Passed(now, w.length) { *b = Buckets{} } } @@ -393,8 +391,8 @@ func (l *Limiter) count( ) for i, w := range l.windows { - requestCounts[i] = requestBuckets[i].add(now, w.length, requests) - byteCounts[i] = byteBuckets[i].add(now, w.length, bytes) + requestCounts[i] = requestBuckets[i].Add(now, w.length, requests) + byteCounts[i] = byteBuckets[i].Add(now, w.length, bytes) limit, byteLimit := percentOf(w.limit, percent), percentOf(w.byteLimit, percent) switch { @@ -460,18 +458,18 @@ func percentOf(limit, percent int64) int64 { return limit/hundred*percent + limit%hundred*percent/hundred } -// add counts n requests, or n bytes, at now in a window of length, and -// returns the client's count in the window that ends at now: what is in -// the bucket under way, and what is in the bucket before it weighted by -// how much of that bucket the window still covers. With n zero it counts -// nothing, and returns the count. +// Add counts n requests, or n bytes, at now in a window of length, and +// returns the count in the window that ends at now: what is in the bucket +// under way, and what is in the bucket before it weighted by how much of +// that bucket the window still covers. With n zero it counts nothing, and +// returns the count. The anomaly counters count in Buckets too. // // Concurrent requests can be counted out of order, so now can be a moment // before the bucket under way began; such a request is counted in that // bucket. A request dated more than a second before it means the clock // was set back, and the buckets start afresh: otherwise the bucket before // would keep its full weight until the clock caught up. -func (b *Buckets) add(now time.Time, length time.Duration, n int64) float64 { +func (b *Buckets) Add(now time.Time, length time.Duration, n int64) float64 { if now.Before(b.Start.Add(-time.Second)) { *b = Buckets{} } @@ -496,6 +494,14 @@ func (b *Buckets) add(now time.Time, length time.Duration, n int64) float64 { return float64(b.Previous)*covered + float64(b.Current) } +// Passed reports whether the window of length that ends at now covers +// neither of b's buckets: it begins after the bucket under way has ended. +// What they hold then counts no more, and a state file read at now drops +// it. +func (b *Buckets) Passed(now time.Time, length time.Duration) bool { + return !now.Add(-length).Before(b.Start.Add(length)) +} + // add counts a response with status in its class. A status of 0, for // nothing sent, is not a response. func (r *Responses) add(status int) { diff --git a/internal/smallwebwaf/smallwebwaf.go b/internal/smallwebwaf/smallwebwaf.go index ee027bf..be9c446 100644 --- a/internal/smallwebwaf/smallwebwaf.go +++ b/internal/smallwebwaf/smallwebwaf.go @@ -201,6 +201,7 @@ func loadStateFiles( Limiter: server.Limiter, GeoJS: server.GeoJS, Alerts: alertQueue, + Anomalies: server.Anomalies, Now: now, ProcessLog: processLog, Metrics: server.Metrics, diff --git a/internal/state/state.go b/internal/state/state.go index eb1d805..6daf5c6 100644 --- a/internal/state/state.go +++ b/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 { diff --git a/internal/state/state_test.go b/internal/state/state_test.go index e92b54f..62ad793 100644 --- a/internal/state/state_test.go +++ b/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.