smallwebwaf/internal/anomaly/anomaly.go

// 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 netblock of
	// SWWAF_IPV6_GROUP_PREFIX.
	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 the IPv6 netblock of
	// SWWAF_IPV6_GROUP_PREFIX its address is in.
	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,
	}
}