Anomaly thresholds: alerts for unusual traffic, nothing refused (closes #101)
check / check (push) Canceled after 0s

SWWAF_ANOMALY_CLIENT_*, _NET_*, _ASN_*, _TOTAL_* and SWWAF_WATCH_* with
SWWAF_WATCH_NETS: requests and bytes per minute and per hour, each off by
default. Every request but the health check is counted, allow-listed and
exempt ones included, in the buckets the rate limits use; a count over its
threshold raises an anomaly alert, with a cooldown per scope. At most
20,000 counters, kept in alerts.json. A per-AS-number threshold with
lookups off, or a malformed SWWAF_WATCH_NETS, stops the start.

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

Model: opus-5-5
This commit is contained in:
2026-10-07 13:04:15 +00:00
parent 2421cdc273
commit 860940246d
15 changed files with 1939 additions and 237 deletions
+377
View File
@@ -0,0 +1,377 @@
package proxy_test
import (
"maps"
"net/http"
"net/netip"
"reflect"
"slices"
"strconv"
"sync/atomic"
"testing"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/anomaly"
"sneak.berlin/go/smallwebwaf/internal/proxy"
"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 TestRequestCountsForTheASNumberGeoJSGivesBeforeItEnds(t *testing.T) {
t.Parallel()
// The stand-in for GeoJS answers only once released, which the app
// does as it answers the request, and then waits until the answer is
// kept.
geojsURL, _, release := startHeldGeoJS(t)
var server atomic.Pointer[proxy.Server]
app := startApp(t, func(http.ResponseWriter, *http.Request) {
release()
waitUntil(func() bool {
_, kept := server.Load().GeoJS.Kept(netip.MustParsePrefix(fromDE + "/32"))
return kept
})
})
clk := &clock{now: time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)}
addr, out, started := startProxyWithClock(t, app.URL, geojsURL, clk.Now,
map[string]string{
trustedProxies: trustLocalhost,
lookupTimeout: "1h",
anomalyASN + requestsPerMinute: aLot,
})
server.Store(started)
// The request went on without the answer, and is counted for the AS
// number it gives.
s := &sender{t: t, addr: addr, out: out}
if line := s.get(fromDE, http.StatusOK, requestlog.ActionForward); line.ASN != "" {
t.Errorf("log line has AS number %q, want none: the request waited", line.ASN)
}
want := []anomaly.Counter{{
Scope: anomaly.ScopeASN, ASN: asnDE,
Minute: ratelimit.Buckets{Start: clk.Now(), Current: 1},
}}
if got := started.Anomalies.Snapshot(); !reflect.DeepEqual(got, want) {
t.Errorf("counters\n%+v\nwant\n%+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 {