Anomaly thresholds: alerts for unusual traffic, nothing refused (closes #101)
check / check (push) Waiting to run

Work in progress.

Model: opus-5-5
This commit is contained in:
2026-10-07 12:49:18 +00:00
parent 2421cdc273
commit 87ec799e99
14 changed files with 1705 additions and 133 deletions
+39 -19
View File
@@ -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,
}
}
+406
View File
@@ -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,
}
}
+238
View File
@@ -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
})
}
+115 -5
View File
@@ -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) {
+211 -11
View File
@@ -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)
}
+332
View File
@@ -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)
}
}
+33 -31
View File
@@ -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
+18 -9
View File
@@ -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.
+18 -1
View File
@@ -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()
+43 -20
View File
@@ -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 {
+17 -11
View File
@@ -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) {
+1
View File
@@ -201,6 +201,7 @@ func loadStateFiles(
Limiter: server.Limiter,
GeoJS: server.GeoJS,
Alerts: alertQueue,
Anomalies: server.Anomalies,
Now: now,
ProcessLog: processLog,
Metrics: server.Metrics,
+73 -15
View File
@@ -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 {
+161 -11
View File
@@ -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.