Alerts to a JSON webhook, with a cooldown and an hourly summary (closes #26)
check / check (push) Waiting to run

SWWAF_ALERT_WEBHOOK_URL gets one JSON POST per alert, in SPEC.md's
schema, with SWWAF_ALERT_WEBHOOK_HEADERS: ban and permanent_ban, with
the ban's notes; source_failure for GeoJS; file_error for a rule or
state file edit that does not parse and a failed state write.
SWWAF_ALERT_EVENTS chooses, SWWAF_ALERT_COOLDOWN holds back repeats,
and past SWWAF_ALERT_MAX_PER_HOUR the hour ends in one summary. A
bounded queue, retried with backoff, holds up no request; alerts.json
keeps it, the cooldowns and the hour. The ledger now reports whether it
made a ban, or made one permanent.

Judgement call: the summary's event is summary, which SPEC.md omits.
Judgement call: admin bans and observe mode raise no alert.

Model: opus-5-5
This commit is contained in:
2026-10-07 00:12:21 +00:00
parent 5d6f6ffaf9
commit a8e86c18db
24 changed files with 2768 additions and 209 deletions
+527
View File
@@ -0,0 +1,527 @@
// Package alerts sends alerts on bans, on a source that fails and on a
// file with an error to the webhook SWWAF_ALERT_WEBHOOK_URL names, each
// as one JSON object, as the "Alert webhook schema" section of SPEC.md
// describes. A repeat within SWWAF_ALERT_COOLDOWN is held back, and so is
// an alert past SWWAF_ALERT_MAX_PER_HOUR, for the hour's summary. The
// others wait in a bounded queue, so that a slow or unreachable webhook
// never holds up a request. The state is written to alerts.json and read
// from it by the state package.
package alerts
import (
"bytes"
"cmp"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"maps"
"net/http"
"net/netip"
"net/url"
"slices"
"sync"
"sync/atomic"
"time"
)
// The events an alert is for, as SWWAF_ALERT_EVENTS names them.
const (
// EventBan is a ban smallwebwaf made.
EventBan = "ban"
// 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.
EventWAFBlock = "waf_block"
EventAnomaly = "anomaly"
EventReputationHit = "reputation_hit"
// EventSourceFailure is GeoJS failing or refusing smallwebwaf.
EventSourceFailure = "source_failure"
// EventFileError is a rule file or state file edited while smallwebwaf
// runs that does not parse, or a state file that cannot be written.
EventFileError = "file_error"
// EventSummary is the summary of the alerts an hour held back past
// SWWAF_ALERT_MAX_PER_HOUR. SWWAF_ALERT_EVENTS does not name it.
EventSummary = "summary"
)
// Events returns every event SWWAF_ALERT_EVENTS can name, which is its
// default.
func Events() []string {
return []string{
EventBan, EventPermanentBan, EventWAFBlock, EventAnomaly,
EventReputationHit, EventSourceFailure, EventFileError,
}
}
const (
// queueSize is the most alerts that wait to be sent. Past it, the
// oldest is dropped.
queueSize = 1000
// sendTimeout bounds one request to the webhook.
sendTimeout = 10 * time.Second
// After a request to the webhook fails, the alert is sent again a
// second later, and retryDelayFactor times as long after each further
// failure in a row, up to a minute.
firstRetryDelay = time.Second
retryDelayFactor = 2
maxRetryDelay = time.Minute
// maxAnswerBytes is the most of the webhook's answer that is read.
maxAnswerBytes = 64 << 10
)
var errStatus = errors.New("the webhook answered")
// Params are what New needs.
type Params struct {
// WebhookURL is where each alert is posted (SWWAF_ALERT_WEBHOOK_URL),
// nil while it is unset and no alert is sent. WebhookHeaders are sent
// with each (SWWAF_ALERT_WEBHOOK_HEADERS).
WebhookURL *url.URL
WebhookHeaders http.Header
// Events are the events alerts are sent for (SWWAF_ALERT_EVENTS).
Events []string
// Cooldown is how long a repeat of an alert is held back
// (SWWAF_ALERT_COOLDOWN), 0 for no time. MaxPerHour is the most alerts
// sent in an hour (SWWAF_ALERT_MAX_PER_HOUR), 0 for no limit.
Cooldown time.Duration
MaxPerHour int
// Instance is SWWAF_INSTANCE_NAME, which every alert gives.
Instance string
// Now tells the time of an alert, normally time.Now in UTC.
Now func() time.Time
// ProcessLog receives the requests to the webhook that fail.
ProcessLog *slog.Logger
}
// Alert is one alert, as the webhook is sent it and alerts.json holds it,
// with the fields of the "Alert webhook schema" section of SPEC.md. ASN
// and ASName are empty until AS numbers are looked up.
//
//nolint:tagliatelle // SPEC.md's alert webhook schema names its fields in snake_case
type Alert struct {
Instance string `json:"instance"`
Time time.Time `json:"time"`
Event string `json:"event"`
Client netip.Addr `json:"client"`
Netblock netip.Prefix `json:"netblock"`
ASN string `json:"asn"`
ASName string `json:"as_name"`
Country string `json:"country"`
// Reason is a short sentence, and Detail what is particular to the
// event.
Reason string `json:"reason"`
Detail map[string]any `json:"detail"`
// SuppressedRepeats is how many repeats of the alert the cooldown
// held back since the last one let through.
SuppressedRepeats int `json:"suppressed_repeats"`
}
// Cooldown is, for an event on a netblock, or for an event without a
// netblock, 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 {
Event string `json:"event"`
Netblock netip.Prefix `json:"netblock"`
Sent time.Time `json:"sent"`
SuppressedRepeats int `json:"suppressed_repeats"`
}
// Hour is the hour under way, by the clock, as alerts.json holds it: when
// it started, how many alerts were let through in it, and how many were
// held back in it past MaxPerHour, by event, for its summary.
//
//nolint:tagliatelle // the state files use snake_case, as the request log does
type Hour struct {
Start time.Time `json:"start"`
Sent int `json:"sent"`
HeldBack map[string]int `json:"held_back"`
}
// State is what alerts.json holds: the cooldowns, the hour under way, and
// the alerts waiting to be sent, oldest first.
type State struct {
Cooldowns []Cooldown `json:"cooldowns"`
Hour Hour `json:"hour"`
Waiting []Alert `json:"waiting"`
}
// Queue takes the alerts raised, holds back those it must, and sends the
// others to the webhook. It is safe for concurrent use.
type Queue struct {
params Params
// httpClient follows no redirect: a redirect is a failure.
httpClient *http.Client
// queued receives a value when an alert joins the queue, unless one
// waits already, so that Run looks at the queue again.
queued chan struct{}
mu sync.Mutex
// cooldowns are the alerts last let through, by event and netblock.
cooldowns map[cooldownKey]*Cooldown
hour Hour
// waiting are the alerts waiting to be sent, oldest first.
waiting []*Alert
sent, failed, suppressed, dropped atomic.Int64
}
// cooldownKey is what makes an alert a repeat of another: the same event
// on the same netblock, which is none for an event without one.
type cooldownKey struct {
event string
netblock netip.Prefix
}
// New returns a Queue with no alert yet.
func New(params Params) *Queue {
return &Queue{
params: params,
httpClient: &http.Client{
CheckRedirect: func(*http.Request, []*http.Request) error {
return http.ErrUseLastResponse
},
},
queued: make(chan struct{}, 1),
cooldowns: map[cooldownKey]*Cooldown{},
hour: Hour{HeldBack: map[string]int{}},
}
}
// Raise sends alert, which names its event and what is particular to it,
// unless no webhook is set or SWWAF_ALERT_EVENTS leaves its event out. It
// gives alert the instance and the time. An alert that repeats the last
// one let through less than Cooldown before is held back and counted,
// and the next one let through gives that count. Past MaxPerHour alerts
// let through in the hour under way, by the clock, an alert is held back
// for that hour's summary instead, which is sent once the hour has ended.
// Raise never waits: an alert let through joins the queue, from which Run
// sends it, and with queueSize alerts waiting the oldest is dropped.
func (q *Queue) Raise(alert Alert) {
if q.params.WebhookURL == nil || !slices.Contains(q.params.Events, alert.Event) {
return
}
q.mu.Lock()
defer q.mu.Unlock()
now := q.params.Now()
alert.Instance = q.params.Instance
alert.Time = now
if q.repeat(&alert, now) {
q.suppressed.Add(1)
return
}
q.endHour(now)
if q.params.MaxPerHour > 0 && q.hour.Sent >= q.params.MaxPerHour {
q.hour.HeldBack[alert.Event]++
q.suppressed.Add(1)
return
}
q.hour.Sent++
q.queue(&alert)
}
// Run sends the alerts waiting, oldest first, until ctx is done. An alert
// stays in the queue until the webhook answers it with a 2xx status. A
// request that fails is logged, and the alert sent again
// firstRetryDelay later, retryDelayFactor times as long after each
// further failure in a row, up to maxRetryDelay. Run also ends each hour
// as Raise does, so that the hour's summary is sent as it ends. With no
// webhook set, it returns at once.
func (q *Queue) Run(ctx context.Context) {
if q.params.WebhookURL == nil {
return
}
var (
retryDelay time.Duration
retryAt time.Time
)
for {
alert, untilHourEnds := q.next()
hourEnds := time.NewTimer(untilHourEnds)
var due <-chan time.Time // nil while no alert waits
if alert != nil {
due = time.After(time.Until(retryAt))
}
select {
case <-ctx.Done():
hourEnds.Stop()
return
case <-q.queued:
case <-hourEnds.C:
q.mu.Lock()
q.endHour(q.params.Now())
q.mu.Unlock()
case <-due:
err := q.send(ctx, alert)
switch {
case err == nil:
q.remove(alert)
q.sent.Add(1)
retryDelay = 0
retryAt = time.Time{}
case ctx.Err() == nil: // not cut off as smallwebwaf stops
q.failed.Add(1)
retryDelay = min(max(retryDelayFactor*retryDelay, firstRetryDelay),
maxRetryDelay)
retryAt = time.Now().Add(retryDelay)
q.params.ProcessLog.Warn("sending an alert to SWWAF_ALERT_WEBHOOK_URL failed",
"error", err.Error(), "sending_again_in", retryDelay.String())
}
}
hourEnds.Stop()
}
}
// Sent is how many alerts the webhook has taken.
func (q *Queue) Sent() int64 {
return q.sent.Load()
}
// Failed is how many requests to the webhook have failed.
func (q *Queue) Failed() int64 {
return q.failed.Load()
}
// Suppressed is how many alerts were held back: by the cooldown, and past
// MaxPerHour.
func (q *Queue) Suppressed() int64 {
return q.suppressed.Load()
}
// Dropped is how many alerts were dropped from a full queue.
func (q *Queue) Dropped() int64 {
return q.dropped.Load()
}
// Snapshot returns the queue's state, as alerts.json holds it, with the
// cooldowns sorted by netblock, then by event.
func (q *Queue) Snapshot() State {
q.mu.Lock()
defer q.mu.Unlock()
state := State{
Cooldowns: make([]Cooldown, 0, len(q.cooldowns)),
Hour: q.hour,
Waiting: make([]Alert, 0, len(q.waiting)),
}
state.Hour.HeldBack = maps.Clone(q.hour.HeldBack)
for _, cooldown := range q.cooldowns {
state.Cooldowns = append(state.Cooldowns, *cooldown)
}
slices.SortFunc(state.Cooldowns, func(a, b Cooldown) int {
if order := a.Netblock.Compare(b.Netblock); order != 0 {
return order
}
return cmp.Compare(a.Event, b.Event)
})
for _, alert := range q.waiting {
state.Waiting = append(state.Waiting, *alert)
}
return state
}
// Load puts state, read from alerts.json, in place of the queue's state.
// Each cooldown's netblock is masked to its length, so that
// 203.0.113.9/24 is 203.0.113.0/24. Past queueSize alerts waiting, the
// oldest are dropped.
func (q *Queue) Load(state State) {
q.mu.Lock()
defer q.mu.Unlock()
q.cooldowns = map[cooldownKey]*Cooldown{}
for _, cooldown := range state.Cooldowns {
cooldown.Netblock = cooldown.Netblock.Masked()
q.cooldowns[cooldownKey{cooldown.Event, cooldown.Netblock}] = &cooldown
}
q.hour = state.Hour
q.hour.HeldBack = maps.Clone(state.Hour.HeldBack)
if q.hour.HeldBack == nil {
q.hour.HeldBack = map[string]int{}
}
q.waiting = nil
for _, alert := range state.Waiting {
q.queue(&alert)
}
}
// repeat reports whether alert, raised at now, repeats the last one let
// through less than Cooldown before, and counts it if it does. Otherwise
// it gives alert the count of the repeats held back since that one, and
// notes alert as the last one let through.
func (q *Queue) repeat(alert *Alert, now time.Time) bool {
if q.params.Cooldown == 0 {
return false
}
key := cooldownKey{alert.Event, alert.Netblock}
last, found := q.cooldowns[key]
if found && now.Sub(last.Sent) < q.params.Cooldown {
last.SuppressedRepeats++
return true
}
if found {
alert.SuppressedRepeats = last.SuppressedRepeats
}
q.cooldowns[key] = &Cooldown{Event: alert.Event, Netblock: alert.Netblock, Sent: now}
return false
}
// endHour ends the hour under way, if now is past it: it queues that
// hour's summary when alerts were held back in it past MaxPerHour, and
// forgets the cooldowns that have run out with no repeat held back, which
// no alert needs any more.
func (q *Queue) endHour(now time.Time) {
start := now.Truncate(time.Hour)
if !start.After(q.hour.Start) {
return
}
heldBack := 0
for _, count := range q.hour.HeldBack {
heldBack += count
}
if heldBack > 0 {
q.queue(&Alert{
Instance: q.params.Instance,
Time: now,
Event: EventSummary,
Reason: fmt.Sprintf("%d alerts held back in the hour from %s, past the %d "+
"an hour SWWAF_ALERT_MAX_PER_HOUR allows", heldBack,
q.hour.Start.Format(time.RFC3339), q.params.MaxPerHour),
Detail: map[string]any{
"hour": q.hour.Start, "count": heldBack, "events": q.hour.HeldBack,
},
})
}
q.hour = Hour{Start: start, HeldBack: map[string]int{}}
for key, cooldown := range q.cooldowns {
if now.Sub(cooldown.Sent) >= q.params.Cooldown && cooldown.SuppressedRepeats == 0 {
delete(q.cooldowns, key)
}
}
}
// queue adds alert to the alerts waiting, first dropping the oldest while
// queueSize wait, and has Run look at the queue again.
func (q *Queue) queue(alert *Alert) {
if len(q.waiting) == queueSize {
q.waiting = slices.Delete(q.waiting, 0, 1)
q.dropped.Add(1)
}
q.waiting = append(q.waiting, alert)
select {
case q.queued <- struct{}{}:
default: // a value waits already
}
}
// next returns the oldest alert waiting, nil when none waits, and how
// long it is until the hour under way ends.
func (q *Queue) next() (*Alert, time.Duration) {
q.mu.Lock()
defer q.mu.Unlock()
var oldest *Alert
if len(q.waiting) > 0 {
oldest = q.waiting[0]
}
return oldest, q.hour.Start.Add(time.Hour).Sub(q.params.Now())
}
// remove takes alert, which Run has sent, out of the queue, unless it has
// been dropped from it, or Load has replaced the queue, since Run took it.
// Only the oldest alert is ever dropped, so alert is the oldest if it is
// there at all.
func (q *Queue) remove(alert *Alert) {
q.mu.Lock()
defer q.mu.Unlock()
if len(q.waiting) > 0 && q.waiting[0] == alert {
q.waiting = slices.Delete(q.waiting, 0, 1)
}
}
// send posts alert to the webhook as JSON, with WebhookHeaders, and
// returns an error unless the webhook answers with a 2xx status.
func (q *Queue) send(ctx context.Context, alert *Alert) error {
body, err := json.Marshal(alert)
if err != nil {
return fmt.Errorf("encode the alert: %w", err)
}
ctx, cancel := context.WithTimeout(ctx, sendTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
q.params.WebhookURL.String(), bytes.NewReader(body))
if err != nil {
return fmt.Errorf("make the request: %w", err)
}
maps.Copy(req.Header, q.params.WebhookHeaders)
req.Header.Set("Content-Type", "application/json")
res, err := q.httpClient.Do(req)
if err != nil {
return err
}
defer func() {
_ = res.Body.Close()
}()
// Read, so that the connection can be used again.
_, _ = io.Copy(io.Discard, io.LimitReader(res.Body, maxAnswerBytes))
if res.StatusCode < http.StatusOK || res.StatusCode >= http.StatusMultipleChoices {
return fmt.Errorf("%w %s", errStatus, res.Status)
}
return nil
}
+645
View File
@@ -0,0 +1,645 @@
package alerts_test
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"net/netip"
"net/url"
"reflect"
"slices"
"strings"
"sync"
"testing"
"testing/synctest"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
)
// The tests run in a synctest bubble, where the time package runs on a
// clock of the test's own, which starts at 2000-01-01T00:00:00Z, the start
// of an hour: a wait lasts exactly as long as it should, however slowly
// the test process runs, and synctest.Wait returns once the queue has
// done all it can before time passes. The stand-in for the webhook
// answers without the network, since a request waiting on the network
// would keep that clock from moving on.
const (
// webhookURL is where the alerts are posted.
webhookURL = "https://alerts.example/smallwebwaf?team=ops"
// instance is the instance name every alert gives.
instance = "fsn1app1/gitea"
// started is when each test starts, as an alert gives it, and
// anHourOn an hour later.
started = "2000-01-01T00:00:00Z"
anHourOn = "2000-01-01T01:00:00Z"
// cooldown is the cooldown of most tests, the default.
cooldown = 15 * time.Minute
)
func TestAlertIsPostedAsJSONWithItsFieldsAndTheHeaders(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
params.WebhookHeaders = http.Header{
"Authorization": {"Bearer 0123456789abcdef"},
"X-Team": {"ops"},
}
webhook, q := start(t, params)
q.Raise(alerts.Alert{
Event: alerts.EventBan,
Client: netip.MustParseAddr("203.0.113.9"),
Netblock: netip.MustParsePrefix("203.0.113.0/24"),
Country: "DE",
Reason: "requests per minute over the limit of 1000",
Detail: map[string]any{"cause": "limit", "ban_expires": anHourOn},
})
synctest.Wait()
got := webhook.received()
if len(got) != 1 {
t.Fatalf("the webhook had %d requests, want 1", len(got))
}
if got[0].method != http.MethodPost || got[0].url != webhookURL {
t.Errorf("request %s %s, want POST %s", got[0].method, got[0].url, webhookURL)
}
for name, want := range map[string]string{
"Content-Type": "application/json",
"Authorization": "Bearer 0123456789abcdef",
"X-Team": "ops",
} {
if got[0].header.Get(name) != want {
t.Errorf("header %s is %q, want %q", name, got[0].header.Get(name), want)
}
}
wantAlert(t, got[0].alert, map[string]any{
"instance": instance,
"time": started,
"event": "ban",
"client": "203.0.113.9",
"netblock": "203.0.113.0/24",
"asn": "",
"as_name": "",
"country": "DE",
"reason": "requests per minute over the limit of 1000",
"detail": map[string]any{"cause": "limit", "ban_expires": anHourOn},
"suppressed_repeats": float64(0),
})
wantCounts(t, q, 1, 0, 0, 0)
})
}
func TestOnlyTheChosenEventsAreSent(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
params.Events = []string{alerts.EventSourceFailure, alerts.EventFileError}
webhook, q := start(t, params)
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
q.Raise(alerts.Alert{Event: alerts.EventFileError, Reason: "a file error"})
q.Raise(alerts.Alert{Event: alerts.EventPermanentBan, Netblock: netblock(1)})
synctest.Wait()
wantEvents(t, webhook, alerts.EventFileError)
wantCounts(t, q, 1, 0, 0, 0)
})
}
func TestNothingIsQueuedWithoutAWebhook(t *testing.T) {
t.Parallel()
params := newParams()
params.WebhookURL = nil
q := alerts.New(params)
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
if waiting := q.Snapshot().Waiting; len(waiting) != 0 {
t.Errorf("%d alerts wait, want none", len(waiting))
}
}
func TestRepeatWithinTheCooldownIsHeldBackAndCountedInTheNext(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
webhook, q := start(t, newParams())
raise := func(event string, n int) {
q.Raise(alerts.Alert{Event: event, Netblock: netblock(n)})
}
raise(alerts.EventBan, 1)
// The same event on the same netblock is a repeat; another netblock
// or another event is not.
time.Sleep(time.Minute)
raise(alerts.EventBan, 1)
raise(alerts.EventBan, 2)
raise(alerts.EventPermanentBan, 1)
time.Sleep(cooldown - time.Minute - time.Nanosecond)
raise(alerts.EventBan, 1)
// Once the cooldown has run out, the next one is sent with the
// count of those held back.
time.Sleep(time.Nanosecond)
raise(alerts.EventBan, 1)
// And starts the cooldown again.
time.Sleep(time.Minute)
raise(alerts.EventBan, 1)
synctest.Wait()
got := webhook.received()
wantEvents(t, webhook, alerts.EventBan, alerts.EventBan, alerts.EventPermanentBan,
alerts.EventBan)
for i, want := range []struct {
netblock int
repeats float64
}{{1, 0}, {2, 0}, {1, 0}, {1, 2}} {
alert := got[i].alert
if alert["netblock"] != netblock(want.netblock).String() ||
alert["suppressed_repeats"] != want.repeats {
t.Errorf("alert %d is for %v with %v repeats, want %s with %v", i,
alert["netblock"], alert["suppressed_repeats"], netblock(want.netblock),
want.repeats)
}
}
wantCounts(t, q, 4, 0, 3, 0)
})
}
func TestNoCooldownSendsEveryRepeat(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
params.Cooldown = 0
webhook, q := start(t, params)
for range 3 {
q.Raise(alerts.Alert{Event: alerts.EventFileError})
time.Sleep(time.Minute)
}
synctest.Wait()
wantEvents(t, webhook, alerts.EventFileError, alerts.EventFileError,
alerts.EventFileError)
wantCounts(t, q, 3, 0, 0, 0)
})
}
func TestAlertsPastTheHourlyLimitAreRolledIntoOneSummary(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
params.MaxPerHour = 2
webhook, q := start(t, params)
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(2)})
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(3)})
q.Raise(alerts.Alert{Event: alerts.EventPermanentBan, Netblock: netblock(4)})
q.Raise(alerts.Alert{Event: alerts.EventFileError})
// The summary is sent as the hour ends, and not before.
time.Sleep(time.Hour - time.Nanosecond)
synctest.Wait()
wantEvents(t, webhook, alerts.EventBan, alerts.EventBan)
time.Sleep(time.Nanosecond)
synctest.Wait()
wantEvents(t, webhook, alerts.EventBan, alerts.EventBan, alerts.EventSummary)
summary := webhook.received()[2].alert
wantAlert(t, summary, map[string]any{
"instance": instance,
"time": anHourOn,
"event": "summary",
"client": "",
"netblock": "",
"asn": "",
"as_name": "",
"country": "",
"reason": "3 alerts held back in the hour from 2000-01-01T00:00:00Z, " +
"past the 2 an hour SWWAF_ALERT_MAX_PER_HOUR allows",
"detail": map[string]any{
"hour": started,
"count": float64(3),
"events": map[string]any{
"ban": float64(1), "permanent_ban": float64(1), "file_error": float64(1),
},
},
"suppressed_repeats": float64(0),
})
// The next hour sends alerts again, and, with none held back, ends
// without a summary.
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(5)})
time.Sleep(time.Hour)
synctest.Wait()
wantEvents(t, webhook, alerts.EventBan, alerts.EventBan, alerts.EventSummary,
alerts.EventBan)
wantCounts(t, q, 4, 0, 3, 0)
})
}
func TestFailedRequestIsSentAgainWithBackoff(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
log := &lockedBuffer{}
params.ProcessLog = slog.New(slog.NewJSONHandler(log, nil))
webhook, q := start(t, params)
webhook.set(failing)
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
// A second after the first failure, then twice as long after each
// further one, up to a minute.
time.Sleep(200 * time.Second)
synctest.Wait()
after := make([]time.Duration, 0, len(webhook.received()))
for _, request := range webhook.received() {
after = append(after, request.at.Sub(midnight()))
}
want := []time.Duration{
0, time.Second, 3 * time.Second, 7 * time.Second, 15 * time.Second,
31 * time.Second, 63 * time.Second, 123 * time.Second, 183 * time.Second,
}
if !slices.Equal(after, want) {
t.Errorf("requests at %v, want %v", after, want)
}
wantCounts(t, q, 0, int64(len(want)), 0, 0)
if !strings.Contains(log.String(),
`"msg":"sending an alert to SWWAF_ALERT_WEBHOOK_URL failed"`) {
t.Errorf("process log %q names no failure", log.String())
}
// Once the webhook answers, the alert is sent, and leaves the
// queue.
webhook.set(answering)
time.Sleep(time.Minute)
synctest.Wait()
got := webhook.received()
if last := got[len(got)-1]; !last.answered ||
last.alert["netblock"] != netblock(1).String() {
t.Errorf("the last request was not the alert, answered")
}
wantCounts(t, q, 1, int64(len(want)), 0, 0)
if waiting := q.Snapshot().Waiting; len(waiting) != 0 {
t.Errorf("%d alerts still wait, want none", len(waiting))
}
})
}
func TestFullQueueDropsTheOldestAndRaiseNeverWaits(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
params.MaxPerHour = 0
webhook, q := start(t, params)
webhook.set(hanging)
// The webhook does not answer the first alert, while one more alert
// than the queue holds is raised: none waits, and the oldest, the
// one the webhook was sent, is dropped.
for n := range alerts.QueueSize + 1 {
q.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(n)})
if n == 0 {
synctest.Wait()
}
}
if took := time.Since(midnight()); took != 0 {
t.Errorf("raising the alerts took %s, want no time", took)
}
wantCounts(t, q, 0, 0, 0, 1)
waiting := q.Snapshot().Waiting
if len(waiting) != alerts.QueueSize || waiting[0].Netblock != netblock(1) {
t.Fatalf("%d alerts wait, the first for %s, want %d, the first for %s",
len(waiting), waiting[0].Netblock, alerts.QueueSize, netblock(1))
}
// The request is abandoned after 10 seconds, and the webhook, which
// answers again, is sent the others, in order, a second later.
webhook.set(answering)
time.Sleep(11 * time.Second)
synctest.Wait()
got := webhook.received()
if len(got) != alerts.QueueSize+1 ||
got[0].alert["netblock"] != netblock(0).String() {
t.Fatalf("the webhook had %d requests, want %d, the first for %s",
len(got), alerts.QueueSize+1, netblock(0))
}
for i, request := range got[1:] {
if request.alert["netblock"] != netblock(i+1).String() {
t.Fatalf("request %d is for %v, want %s", i+1, request.alert["netblock"],
netblock(i+1))
}
}
wantCounts(t, q, alerts.QueueSize, 1, 0, 1)
})
}
func TestStateLoadedIntoANewQueueCarriesOn(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
params := newParams()
params.MaxPerHour = 1
before := alerts.New(params)
// Not sent: Run is not running. The repeat is held back by the
// cooldown, and the file error past the hourly limit.
before.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
before.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
before.Raise(alerts.Alert{Event: alerts.EventFileError})
time.Sleep(time.Minute)
webhook, after := start(t, params)
after.Load(roundTrip(t, before.Snapshot()))
// The new queue sends the alert waiting, holds back the repeat as
// the cooldown still runs, and sends the summary of the hour.
after.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
synctest.Wait()
wantEvents(t, webhook, alerts.EventBan)
time.Sleep(time.Hour)
synctest.Wait()
wantEvents(t, webhook, alerts.EventBan, alerts.EventSummary)
detail, _ := webhook.received()[1].alert["detail"].(map[string]any)
if detail["count"] != float64(1) {
t.Errorf("the summary counts %v alerts, want 1", detail["count"])
}
// The cooldown has run out, and the next one gives both repeats.
after.Raise(alerts.Alert{Event: alerts.EventBan, Netblock: netblock(1)})
synctest.Wait()
got := webhook.received()
if repeats := got[len(got)-1].alert["suppressed_repeats"]; repeats != float64(2) {
t.Errorf("the last alert gives %v repeats, want 2", repeats)
}
})
}
// How the stand-in for the webhook answers.
const (
answering = iota // with 204
failing // with 503
hanging // not at all, until the request is abandoned
)
// standIn is a stand-in for the webhook. It notes each request it is
// sent.
type standIn struct {
mu sync.Mutex
answers int
requests []post
}
// post is a request the webhook was sent: when, its method, URL and
// headers, the alert it carried, and whether the webhook answered it with
// a 2xx status.
type post struct {
at time.Time
method string
url string
header http.Header
alert map[string]any
answered bool
}
// RoundTrip has the stand-in answer req, in place of the network. A
// request abandoned before the stand-in answers fails, as over the
// network.
func (s *standIn) RoundTrip(req *http.Request) (*http.Response, error) {
answer := httptest.NewRecorder()
s.ServeHTTP(answer, req)
_ = req.Body.Close()
err := req.Context().Err()
if err != nil {
return nil, err
}
return answer.Result(), nil
}
// ServeHTTP notes the request, and answers it as the stand-in is set to.
func (s *standIn) ServeHTTP(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
var alert map[string]any
_ = json.Unmarshal(body, &alert)
s.mu.Lock()
answers := s.answers
s.requests = append(s.requests, post{
at: time.Now(), method: r.Method, url: r.URL.String(), header: r.Header.Clone(),
alert: alert, answered: answers == answering,
})
s.mu.Unlock()
switch answers {
case failing:
w.WriteHeader(http.StatusServiceUnavailable)
case hanging:
<-r.Context().Done()
default:
w.WriteHeader(http.StatusNoContent)
}
}
// set sets how the stand-in answers.
func (s *standIn) set(answers int) {
s.mu.Lock()
defer s.mu.Unlock()
s.answers = answers
}
// received returns the requests the stand-in has been sent so far.
func (s *standIn) received() []post {
s.mu.Lock()
defer s.mu.Unlock()
return slices.Clone(s.requests)
}
// lockedBuffer is a buffer the process log can write to while the test
// reads it.
type lockedBuffer struct {
mu sync.Mutex
buf bytes.Buffer
}
// Write adds p to the buffer.
func (b *lockedBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.Write(p)
}
// String returns what was written.
func (b *lockedBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.String()
}
// newParams returns the Params of most tests: the webhook at webhookURL,
// every event, the default cooldown and hourly limit, and the bubble's
// clock in UTC.
func newParams() alerts.Params {
webhook, err := url.Parse(webhookURL)
if err != nil {
panic(err)
}
return alerts.Params{
WebhookURL: webhook,
Events: alerts.Events(),
Cooldown: cooldown,
MaxPerHour: 60,
Instance: instance,
Now: func() time.Time { return time.Now().UTC() },
ProcessLog: slog.New(slog.DiscardHandler),
}
}
// start returns a stand-in for the webhook that answers, and a Queue that
// sends to it, run until the test ends.
func start(t *testing.T, params alerts.Params) (*standIn, *alerts.Queue) {
t.Helper()
webhook := &standIn{}
q := alerts.New(params)
q.SetTransport(webhook)
ctx, stop := context.WithCancel(t.Context())
stopped := make(chan struct{})
go func() {
q.Run(ctx)
close(stopped)
}()
t.Cleanup(func() {
stop()
<-stopped
})
return webhook, q
}
// midnight is when each test starts.
func midnight() time.Time {
return time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
}
// netblock returns the n-th netblock of a test, counted from 0.
func netblock(n int) netip.Prefix {
return netip.MustParsePrefix(fmt.Sprintf("203.0.%d.%d/32", 113+n/256, n%256))
}
// roundTrip returns state once written as JSON and read back, as
// alerts.json carries it from one start to the next.
func roundTrip(t *testing.T, state alerts.State) alerts.State {
t.Helper()
data, err := json.Marshal(state)
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)
}
return read
}
// wantAlert checks every field of an alert the webhook was sent.
func wantAlert(t *testing.T, got, want map[string]any) {
t.Helper()
if !reflect.DeepEqual(got, want) {
t.Errorf("alert %v, want %v", got, want)
}
}
// wantEvents checks the events of the alerts the webhook was sent, in
// order.
func wantEvents(t *testing.T, webhook *standIn, want ...string) {
t.Helper()
got := make([]string, 0, len(webhook.received()))
for _, request := range webhook.received() {
event, _ := request.alert["event"].(string)
got = append(got, event)
}
if !slices.Equal(got, want) {
t.Errorf("the webhook was sent %v, want %v", got, want)
}
}
// wantCounts checks the alerts q counts as sent, the requests it counts as
// failed, and the alerts it counts as held back and as dropped.
func wantCounts(
t *testing.T, q *alerts.Queue, sent, failed, suppressed, dropped int64,
) {
t.Helper()
if q.Sent() != sent || q.Failed() != failed || q.Suppressed() != suppressed ||
q.Dropped() != dropped {
t.Errorf("counts sent %d, failed %d, suppressed %d and dropped %d, "+
"want %d, %d, %d and %d", q.Sent(), q.Failed(), q.Suppressed(), q.Dropped(),
sent, failed, suppressed, dropped)
}
}
+12
View File
@@ -0,0 +1,12 @@
package alerts
import "net/http"
// QueueSize is the most alerts that wait to be sent.
const QueueSize = queueSize
// SetTransport has q's requests to the webhook go through transport
// instead of the network.
func (q *Queue) SetTransport(transport http.RoundTripper) {
q.httpClient.Transport = transport
}
+8 -8
View File
@@ -65,9 +65,9 @@ func TestReasonOfTheBansSmallwebwafMakes(t *testing.T) {
ledger := bans.New(defaultRules())
limit := ledger.BanForLimit(netip.MustParsePrefix("203.0.113.1/32"), midnight(),
limit, _ := ledger.BanForLimit(netip.MustParsePrefix("203.0.113.1/32"), midnight(),
bans.Notes{Limit: 1000, Window: "minute"})
attack := ledger.BanForAttack(netip.MustParsePrefix("203.0.113.2/32"), midnight(),
attack, _ := ledger.BanForAttack(netip.MustParsePrefix("203.0.113.2/32"), midnight(),
bans.Notes{RuleID: "git-dir", Target: "path"})
for _, tc := range []struct{ got, want string }{
@@ -101,12 +101,12 @@ func TestLiftedBanForALimitRefusesNothingAndMakesNoBanLonger(t *testing.T) {
// kept, and counted among the earlier bans.
now := midnight().Add(30 * time.Minute)
_, banned := ledger.Check(netblock.Addr(), now)
_, banned, _ := ledger.Check(netblock.Addr(), now)
if banned {
t.Error("the lifted ban refuses")
}
ban := ledger.BanForLimit(netblock, now, bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, now, bans.Notes{})
if ban.Expires.Sub(ban.Start) != time.Hour ||
ban.Notes.EarlierBans != (bans.EarlierBans{Limit: 1}) {
t.Errorf("the next ban lasts %s with earlier bans %+v, want 1h and 1 for a limit",
@@ -145,7 +145,7 @@ func TestLiftedBanForAnAttackRefusesNothingAndMakesNoBanLonger(t *testing.T) {
}
// The next clear sign of attack bans for seven days, as a first does.
ban := ledger.BanForAttack(netblock, now, bans.Notes{})
ban, _ := ledger.BanForAttack(netblock, now, bans.Notes{})
if ban.Expires.Sub(ban.Start) != 7*day {
t.Errorf("the next ban for an attack ends at %s, want seven days on", ban.Expires)
}
@@ -155,7 +155,7 @@ func TestLoadEditCountsTheBansAnAdminMade(t *testing.T) {
t.Parallel()
ledger := bans.New(defaultRules())
made := ledger.BanForLimit(netip.MustParsePrefix("203.0.113.1/32"), midnight(),
made, _ := ledger.BanForLimit(netip.MustParsePrefix("203.0.113.1/32"), midnight(),
bans.Notes{})
atStart := bans.Ban{
Netblock: netip.MustParsePrefix("203.0.113.2/32"),
@@ -261,11 +261,11 @@ func TestLiftLiftsEveryActiveBanCoveringTheClient(t *testing.T) {
wantChanged(t, ledger, true)
if _, banned := ledger.Check(client, now); banned {
if _, banned, _ := ledger.Check(client, now); banned {
t.Error("the client is still banned")
}
if _, banned := ledger.Check(other.Addr(), now); !banned {
if _, banned, _ := ledger.Check(other.Addr(), now); !banned {
t.Error("the other client's ban was lifted")
}
+29 -21
View File
@@ -194,25 +194,27 @@ func (l *Ledger) Changed() <-chan struct{} {
// a ban on a netblock client is in is active, and returns that ban, with
// the request counted among those it refused. A ban for a clear sign of
// attack is made permanent by the request: the netblock is malicious.
func (l *Ledger) Check(client netip.Addr, now time.Time) (Ban, bool) {
// The last result reports whether the request made the ban permanent.
func (l *Ledger) Check(client netip.Addr, now time.Time) (Ban, bool, bool) {
l.mu.Lock()
defer l.mu.Unlock()
ban := l.active(client, now)
if ban == nil {
return Ban{}, false
return Ban{}, false, false
}
ban.Notes.Requests++
ban.Notes.Refused++
if ban.Cause == CauseAttack && !ban.Permanent() {
madePermanent := ban.Cause == CauseAttack && !ban.Permanent()
if madePermanent {
ban.Expires = time.Time{}
l.markChanged()
}
return *ban, true
return *ban, true, madePermanent
}
// Find is Check without counting the request among those the ban
@@ -244,16 +246,18 @@ func activeBan(bans []Ban, now time.Time) *Ban {
}
// BanForLimit bans netblock at now for a broken limit, with notes, and
// returns the ban. A first ban lasts LimitBanDuration. A ban made within
// LimitBanRepeatWindow after the netblock's ban that ended last, other
// than one for a clear sign of attack or a lifted one, lasts repeatFactor
// times as long as that one. A ban that would be longer than
// MaxBanDuration is permanent instead. If a ban on netblock is still
// returns the ban, and true. A first ban lasts LimitBanDuration. A ban
// made within LimitBanRepeatWindow after the netblock's ban that ended
// last, other than one for a clear sign of attack or a lifted one, lasts
// repeatFactor times as long as that one. A ban that would be longer
// than MaxBanDuration is permanent instead. If a ban on netblock is still
// active, as when two of its requests break a limit at once, that ban is
// returned and no other is made. The ledger fills in the notes' Refused
// and EarlierBans itself, and gives the ban the reason "requests per
// <Window> over the limit of <Limit>", from the notes.
func (l *Ledger) BanForLimit(netblock netip.Prefix, now time.Time, notes Notes) Ban {
// returned with false, and no other is made. The ledger fills in the
// notes' Refused and EarlierBans itself, and gives the ban the reason
// "requests per <Window> over the limit of <Limit>", from the notes.
func (l *Ledger) BanForLimit(
netblock netip.Prefix, now time.Time, notes Notes,
) (Ban, bool) {
reason := fmt.Sprintf("requests per %s over the limit of %d",
notes.Window, notes.Limit)
@@ -261,10 +265,13 @@ func (l *Ledger) BanForLimit(netblock netip.Prefix, now time.Time, notes Notes)
}
// BanForAttack bans netblock at now for a clear sign of attack, with
// notes, and returns the ban, as BanForLimit does. A first ban lasts
// AttackBanDuration; once the netblock has had one that was not lifted,
// the next is permanent. Its reason is "matched the rule <RuleID>".
func (l *Ledger) BanForAttack(netblock netip.Prefix, now time.Time, notes Notes) Ban {
// notes, and returns the ban, and whether it made it, as BanForLimit
// does. A first ban lasts AttackBanDuration; once the netblock has had
// one that was not lifted, the next is permanent. Its reason is "matched
// the rule <RuleID>".
func (l *Ledger) BanForAttack(
netblock netip.Prefix, now time.Time, notes Notes,
) (Ban, bool) {
return l.ban(netblock, now, CauseAttack, "matched the rule "+notes.RuleID, notes)
}
@@ -483,10 +490,11 @@ func (l *Ledger) holds(netblock netip.Prefix, start time.Time) bool {
}
// ban bans netblock at now for cause, with reason and notes, as
// BanForLimit and BanForAttack describe, and returns the ban.
// BanForLimit and BanForAttack describe, and returns the ban, and whether
// it made it.
func (l *Ledger) ban(
netblock netip.Prefix, now time.Time, cause, reason string, notes Notes,
) Ban {
) (Ban, bool) {
l.mu.Lock()
defer l.mu.Unlock()
@@ -497,7 +505,7 @@ func (l *Ledger) ban(
if found {
active := activeBan(*bans, now)
if active != nil {
return *active
return *active, false
}
held = *bans
@@ -517,7 +525,7 @@ func (l *Ledger) ban(
l.made[cause]++
l.markChanged()
return ban
return ban, true
}
// earlierBans returns how many bans a netblock with the bans held, oldest
+52 -38
View File
@@ -21,7 +21,7 @@ func TestRepeatsTripleUntilPermanent(t *testing.T) {
// Each ban is followed by another as soon as it ends: 1, 3, 9, 27 and
// 81 hours.
for i, hours := range []int{1, 3, 9, 27, 81} {
ban := ledger.BanForLimit(netblock, now, bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, now, bans.Notes{})
length := time.Duration(hours) * time.Hour
if !ban.Expires.Equal(now.Add(length)) ||
@@ -35,12 +35,12 @@ func TestRepeatsTripleUntilPermanent(t *testing.T) {
// The sixth would last 243 hours, more than seven days: it is
// permanent, and never ends.
ban := ledger.BanForLimit(netblock, now, bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, now, bans.Notes{})
if !ban.Permanent() {
t.Fatalf("sixth ban ends at %s, want a permanent one", ban.Expires)
}
_, banned := ledger.Check(netblock.Addr(), now.Add(100*365*day))
_, banned, _ := ledger.Check(netblock.Addr(), now.Add(100*365*day))
if !banned {
t.Error("a permanent ban ended")
}
@@ -64,8 +64,8 @@ func TestRepeatWindowRunsOut(t *testing.T) {
ledger := bans.New(defaultRules())
netblock := netip.MustParsePrefix("203.0.113.9/32")
first := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
second := ledger.BanForLimit(netblock, first.Expires.Add(tc.gap), bans.Notes{})
first, _ := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
second, _ := ledger.BanForLimit(netblock, first.Expires.Add(tc.gap), bans.Notes{})
if second.Expires.Sub(second.Start) != tc.want ||
second.Notes.EarlierBans != (bans.EarlierBans{Limit: 1}) {
@@ -83,7 +83,7 @@ func TestFirstBanLongerThanTheMaximumIsPermanent(t *testing.T) {
rules.LimitBanDuration = rules.MaxBanDuration + time.Hour
ledger := bans.New(rules)
ban := ledger.BanForLimit(netip.MustParsePrefix("203.0.113.9/32"), midnight(),
ban, _ := ledger.BanForLimit(netip.MustParsePrefix("203.0.113.9/32"), midnight(),
bans.Notes{})
if !ban.Permanent() {
t.Errorf("first ban ends at %s, want a permanent one", ban.Expires)
@@ -103,7 +103,7 @@ func TestLongestBanSetFarOffDoesNotOverflow(t *testing.T) {
now := midnight()
for i := range 14 {
ban := ledger.BanForLimit(netblock, now, bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, now, bans.Notes{})
if !ban.Expires.After(ban.Start) {
t.Fatalf("ban %d starts at %s and ends at %s", i+1, ban.Start, ban.Expires)
}
@@ -111,7 +111,7 @@ func TestLongestBanSetFarOffDoesNotOverflow(t *testing.T) {
now = ban.Expires
}
ban := ledger.BanForLimit(netblock, now, bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, now, bans.Notes{})
if !ban.Permanent() {
t.Errorf("15th ban ends at %s, want a permanent one", ban.Expires)
}
@@ -123,12 +123,22 @@ func TestBrokenLimitDuringABanMakesNoOther(t *testing.T) {
ledger := bans.New(defaultRules())
netblock := netip.MustParsePrefix("203.0.113.9/32")
first := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
again := ledger.BanForLimit(netblock, midnight().Add(time.Minute), bans.Notes{})
first, made := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
if !made {
t.Error("the first ban was not made")
}
if again != first || len(ledger.Bans(netblock)) != 1 {
t.Errorf("a limit broken during a ban gave %+v and %d bans, want %+v and 1",
again, len(ledger.Bans(netblock)), first)
again, made := ledger.BanForLimit(netblock, midnight().Add(time.Minute), bans.Notes{})
if made || again != first || len(ledger.Bans(netblock)) != 1 {
t.Errorf("a limit broken during a ban gave %+v, made %t, and %d bans, "+
"want %+v, not made, and 1", again, made, len(ledger.Bans(netblock)), first)
}
again, made = ledger.BanForAttack(netblock, midnight().Add(time.Minute), bans.Notes{})
if made || again != first {
t.Errorf("an attack during a ban gave %+v, made %t, want %+v, not made",
again, made, first)
}
}
@@ -137,21 +147,21 @@ func TestCheckRefusesWhileTheBanLastsAndCountsTheRefusals(t *testing.T) {
ledger := bans.New(defaultRules())
netblock := netip.MustParsePrefix("203.0.113.9/32")
ban := ledger.BanForLimit(netblock, midnight(), bans.Notes{Requests: 5})
ban, _ := ledger.BanForLimit(netblock, midnight(), bans.Notes{Requests: 5})
for range 3 {
got, banned := ledger.Check(netblock.Addr(), ban.Expires.Add(-time.Nanosecond))
got, banned, _ := ledger.Check(netblock.Addr(), ban.Expires.Add(-time.Nanosecond))
if !banned || got.Start != ban.Start {
t.Fatalf("check during the ban gives %+v and %t", got, banned)
}
}
_, banned := ledger.Check(netip.MustParseAddr("203.0.113.10"), midnight())
_, banned, _ := ledger.Check(netip.MustParseAddr("203.0.113.10"), midnight())
if banned {
t.Error("another netblock is banned")
}
_, banned = ledger.Check(netblock.Addr(), ban.Expires)
_, banned, _ = ledger.Check(netblock.Addr(), ban.Expires)
if banned {
t.Error("the ban did not end")
}
@@ -169,7 +179,7 @@ func TestFindCountsNothing(t *testing.T) {
ledger := bans.New(defaultRules())
netblock := netip.MustParsePrefix("203.0.113.9/32")
ban := ledger.BanForLimit(netblock, midnight(), bans.Notes{Requests: 5})
ban, _ := ledger.BanForLimit(netblock, midnight(), bans.Notes{Requests: 5})
got, banned := ledger.Find(netblock.Addr(), ban.Expires.Add(-time.Nanosecond))
if !banned || got != ban {
@@ -198,7 +208,7 @@ func TestMaxBansDropsTheEarliestBanOfTheNetblockSeenLongestAgo(t *testing.T) {
d := netip.MustParsePrefix("2001:db8::/64")
now := midnight()
first := ledger.BanForLimit(a, now, bans.Notes{})
first, _ := ledger.BanForLimit(a, now, bans.Notes{})
ledger.BanForLimit(b, now, bans.Notes{})
ledger.BanForLimit(c, now, bans.Notes{})
@@ -233,8 +243,8 @@ func TestFullLedgerDropsTheEarlierBanOfTheNetblockBannedAgain(t *testing.T) {
ledger := bans.New(rules)
netblock := netip.MustParsePrefix("203.0.113.9/32")
first := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
second := ledger.BanForLimit(netblock, first.Expires, bans.Notes{})
first, _ := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
second, _ := ledger.BanForLimit(netblock, first.Expires, bans.Notes{})
held := ledger.Bans(netblock)
if len(held) != 1 || held[0] != second ||
@@ -251,7 +261,7 @@ func TestRequestDuringAnAttackBanMakesItPermanent(t *testing.T) {
netblock := netip.MustParsePrefix("203.0.113.9/32")
notes := bans.Notes{RuleID: "env-file", Target: "path"}
ban := ledger.BanForAttack(netblock, midnight(), notes)
ban, _ := ledger.BanForAttack(netblock, midnight(), notes)
if !ban.Expires.Equal(midnight().Add(7*day)) || ban.Cause != bans.CauseAttack ||
ban.Notes.RuleID != "env-file" || ledger.Made(bans.CauseAttack) != 1 ||
ledger.Made(bans.CauseLimit) != 0 {
@@ -268,17 +278,21 @@ func TestRequestDuringAnAttackBanMakesItPermanent(t *testing.T) {
t.Fatal("a request found under the ban made it permanent")
}
// A request it refuses makes it permanent, and bans.json due.
got, _ = ledger.Check(netblock.Addr(), midnight().Add(time.Hour))
if !got.Permanent() || !ledger.Bans(netblock)[0].Permanent() {
t.Fatalf("after a request during the ban, it is %+v, want it permanent", got)
// A request it refuses makes it permanent, says so, and makes
// bans.json due.
got, _, madePermanent := ledger.Check(netblock.Addr(), midnight().Add(time.Hour))
if !madePermanent || !got.Permanent() || !ledger.Bans(netblock)[0].Permanent() {
t.Fatalf("after a request during the ban, it is %+v, made permanent %t, "+
"want it made permanent", got, madePermanent)
}
wantChanged(t, ledger, true)
_, banned := ledger.Check(netblock.Addr(), midnight().Add(100*365*day))
if !banned {
t.Error("the permanent ban ended")
// The next request finds it permanent already.
_, banned, madePermanent := ledger.Check(netblock.Addr(), midnight().Add(100*365*day))
if !banned || madePermanent {
t.Errorf("a later request is banned %t, and made the ban permanent %t, "+
"want banned by the permanent ban", banned, madePermanent)
}
}
@@ -289,8 +303,8 @@ func TestAttackAfterAnAttackBanHasEndedBansPermanently(t *testing.T) {
netblock := netip.MustParsePrefix("203.0.113.9/32")
// A ban for a broken limit before does not count.
first := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
second := ledger.BanForAttack(netblock, first.Expires, bans.Notes{})
first, _ := ledger.BanForLimit(netblock, midnight(), bans.Notes{})
second, _ := ledger.BanForAttack(netblock, first.Expires, bans.Notes{})
if second.Expires.Sub(second.Start) != 7*day {
t.Fatalf("the first ban for an attack lasts %s, want 7 days",
@@ -299,14 +313,14 @@ func TestAttackAfterAnAttackBanHasEndedBansPermanently(t *testing.T) {
// Once that has run out without a request, the netblock is served, and
// its next clear sign of attack bans it for good.
_, banned := ledger.Check(netblock.Addr(), second.Expires)
_, banned, _ := ledger.Check(netblock.Addr(), second.Expires)
if banned {
t.Fatal("the ban did not end")
}
// Its notes show the earlier ban for an attack that makes it permanent,
// beside the one for a limit.
third := ledger.BanForAttack(netblock, second.Expires.Add(30*day), bans.Notes{})
third, _ := ledger.BanForAttack(netblock, second.Expires.Add(30*day), bans.Notes{})
if !third.Permanent() ||
third.Notes.EarlierBans != (bans.EarlierBans{Limit: 1, Attack: 1}) {
t.Errorf("the next ban for an attack is %+v, want a permanent one, "+
@@ -322,16 +336,16 @@ func TestAttackBanDoesNotLengthenTheNextBanForALimit(t *testing.T) {
// Three times the seven days would be permanent; a limit broken as the
// ban for an attack ends bans for an hour, as a first broken limit does.
attack := ledger.BanForAttack(netblock, midnight(), bans.Notes{})
limit := ledger.BanForLimit(netblock, attack.Expires, bans.Notes{})
attack, _ := ledger.BanForAttack(netblock, midnight(), bans.Notes{})
limit, _ := ledger.BanForLimit(netblock, attack.Expires, bans.Notes{})
if limit.Expires.Sub(limit.Start) != time.Hour || limit.Cause != bans.CauseLimit {
t.Errorf("the ban for a limit is %+v, want one of an hour", limit)
}
// And a request during the ban for a limit leaves it as it is.
got, _ := ledger.Check(netblock.Addr(), limit.Start)
if got.Permanent() {
got, _, madePermanent := ledger.Check(netblock.Addr(), limit.Start)
if got.Permanent() || madePermanent {
t.Error("a request during a ban for a limit made it permanent")
}
}
@@ -346,7 +360,7 @@ func TestRequestTextsAreCutTo256Bytes(t *testing.T) {
Time: midnight(), Method: long, Host: long, Path: long, Status: 403, UserAgent: long,
}
ban := ledger.BanForLimit(netblock, midnight(), bans.Notes{Request: request})
ban, _ := ledger.BanForLimit(netblock, midnight(), bans.Notes{Request: request})
cut := long[:256]
want := bans.Request{
+11 -11
View File
@@ -42,7 +42,7 @@ func TestSnapshotListsEveryBanByNetblock(t *testing.T) {
high := netip.MustParsePrefix("203.0.113.10/32")
low := netip.MustParsePrefix("203.0.113.9/32")
first := ledger.BanForLimit(v6, midnight(), bans.Notes{})
first, _ := ledger.BanForLimit(v6, midnight(), bans.Notes{})
ledger.BanForLimit(high, midnight(), bans.Notes{})
ledger.BanForLimit(low, midnight(), bans.Notes{})
ledger.BanForLimit(v6, first.Expires, bans.Notes{})
@@ -68,7 +68,7 @@ func TestLoadedBansCarryOn(t *testing.T) {
before := bans.New(defaultRules())
netblock := netip.MustParsePrefix("203.0.113.9/32")
ban := before.BanForLimit(netblock, midnight(), bans.Notes{Limit: 1})
ban, _ := before.BanForLimit(netblock, midnight(), bans.Notes{Limit: 1})
// Loaded into a new ledger, as across a restart, the ban still refuses
// while it lasts, and once it has ended a broken limit bans for three
@@ -76,12 +76,12 @@ func TestLoadedBansCarryOn(t *testing.T) {
after := bans.New(defaultRules())
after.Load(before.Snapshot())
_, banned := after.Check(netblock.Addr(), ban.Expires.Add(-time.Second))
_, banned, _ := after.Check(netblock.Addr(), ban.Expires.Add(-time.Second))
if !banned {
t.Error("the loaded ban does not refuse")
}
again := after.BanForLimit(netblock, ban.Expires, bans.Notes{})
again, _ := after.BanForLimit(netblock, ban.Expires, bans.Notes{})
if again.Expires.Sub(again.Start) != 3*time.Hour ||
again.Notes.EarlierBans != (bans.EarlierBans{Limit: 1}) {
t.Errorf("the next ban lasts %s with earlier bans %+v, want 3h and 1 for a limit",
@@ -111,7 +111,7 @@ func TestLoadedBanRefusesEveryClientInItsNetblock(t *testing.T) {
"198.51.100.7": true,
"198.51.100.8": false,
} {
_, banned := ledger.Check(netip.MustParseAddr(client), midnight())
_, banned, _ := ledger.Check(netip.MustParseAddr(client), midnight())
if banned != want {
t.Errorf("%s is refused: %t, want %t", client, banned, want)
}
@@ -155,14 +155,14 @@ func TestPermanentBanStartedBeforeAnEndedOneRefuses(t *testing.T) {
t.Errorf("find gives %+v and %t, want the permanent ban", ban, banned)
}
ban, banned = ledger.Check(client, now)
ban, banned, _ = ledger.Check(client, now)
if !banned || !ban.Permanent() {
t.Errorf("the client is refused: %t, under %+v, want under the permanent ban",
banned, ban)
}
// A limit broken now makes no shorter ban over the permanent one.
ban = ledger.BanForLimit(netblock, now, bans.Notes{})
ban, _ = ledger.BanForLimit(netblock, now, bans.Notes{})
if !ban.Permanent() || len(ledger.Bans(netblock)) != 2 {
t.Errorf("a broken limit returned %+v and left the netblock %d bans, "+
"want the permanent ban and 2", ban, len(ledger.Bans(netblock)))
@@ -194,7 +194,7 @@ func TestNextBanWorkedOutFromTheBanThatEndedLast(t *testing.T) {
// Once both have ended, a limit broken within the repeat window bans
// for three times the 9 hours, and the notes count the two bans
// before the 9-hour one and it, for a limit, and the admin's.
ban := ledger.BanForLimit(netblock, nineHours.Expires.Add(time.Hour), bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, nineHours.Expires.Add(time.Hour), bans.Notes{})
if ban.Expires.Sub(ban.Start) != 27*time.Hour ||
ban.Notes.EarlierBans != (bans.EarlierBans{Limit: 3, Admin: 1}) {
t.Errorf("the next ban lasts %s with earlier bans %+v, "+
@@ -255,15 +255,15 @@ func TestLoadReplacesTheBansHeld(t *testing.T) {
// bans.json is taken in, that ban is lifted.
ledger.Load([]bans.Ban{kept})
_, banned := ledger.Check(netip.MustParseAddr("203.0.113.9"), midnight())
_, banned, _ := ledger.Check(netip.MustParseAddr("203.0.113.9"), midnight())
if banned {
t.Error("a ban left out of the second load still refuses")
}
// The ledger holds one ban, so it makes two more without dropping any.
first := ledger.BanForLimit(netip.MustParsePrefix("198.51.100.7/32"), midnight(),
first, _ := ledger.BanForLimit(netip.MustParsePrefix("198.51.100.7/32"), midnight(),
bans.Notes{})
second := ledger.BanForLimit(netip.MustParsePrefix("198.51.100.8/32"), midnight(),
second, _ := ledger.BanForLimit(netip.MustParsePrefix("198.51.100.8/32"), midnight(),
bans.Notes{})
want := []bans.Ban{first, second, kept}
+162 -10
View File
@@ -22,6 +22,7 @@ import (
"time"
"unicode/utf8"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/remotelog"
)
@@ -158,6 +159,19 @@ type Config struct {
LogRemoteBuffer int
LogRemoteFacility int
LogRemoteAppName string
// AlertWebhookURL is where each alert is posted as JSON
// (SWWAF_ALERT_WEBHOOK_URL), nil while it is unset and no alert is
// sent. AlertWebhookHeaders are sent with each
// (SWWAF_ALERT_WEBHOOK_HEADERS). AlertEvents are the events alerts are
// sent for (SWWAF_ALERT_EVENTS). A repeat of an alert within
// AlertCooldown is held back (SWWAF_ALERT_COOLDOWN), and so is an alert
// past AlertMaxPerHour in an hour, for the hour's summary
// (SWWAF_ALERT_MAX_PER_HOUR); 0 is off for both.
AlertWebhookURL *url.URL
AlertWebhookHeaders http.Header
AlertEvents []string
AlertCooldown time.Duration
AlertMaxPerHour int
// settings are the values read, as given or by default, and the
// files they were read from, for the log line at start.
@@ -230,7 +244,17 @@ var (
errNotFacility = errors.New("is not a syslog facility such as local0 or daemon")
errNotAppName = errors.New(
"is not 1 to 48 printable ASCII characters without a space, such as gitea")
errSetTwice = errors.New("set only one of them")
errSetTwice = errors.New("set only one of them")
errNotWebhookURL = errors.New(
"is not an http or https URL without a user or a fragment, " +
"such as https://alerts.example/smallwebwaf")
errNotWebhookHeader = errors.New(
"is not a header name followed by : and the header's value, " +
"such as Authorization:Bearer <token>")
errNotAlertEvent = errors.New(
"is not ban, permanent_ban, waf_block, anomaly, reputation_hit, " +
"source_failure or file_error")
errNotNumberOrOff = errors.New("is not a whole number above zero, such as 60, or off")
)
// FromEnvironment reads the settings with lookupEnv, normally
@@ -278,15 +302,21 @@ func FromEnvironment(lookupEnv func(string) (string, bool)) (*Config, error) {
StateCounterInterval: env.durationNotOff("SWWAF_STATE_COUNTER_INTERVAL", "15m"),
LogRequestHeaders: env.headerNames("SWWAF_LOG_REQUEST_HEADERS",
"accept,accept-language,accept-encoding,content-type,origin,range"),
AdminToken: env.token("SWWAF_ADMIN_TOKEN"),
MetricsToken: env.token("SWWAF_METRICS_TOKEN"),
MetricsTopN: env.numberNotOff("SWWAF_METRICS_TOP_N", "50"),
RulesDir: env.value("SWWAF_RULES_DIR", "/etc/smallwebwaf/rules.d"),
RulesEnabled: env.boolean("SWWAF_RULES_ENABLED", "true"),
LogRemoteURL: env.logRemoteURL("SWWAF_LOG_REMOTE_URL"),
LogRemoteTLSCAs: env.certificates("SWWAF_LOG_REMOTE_TLS_CA_FILE"),
LogRemoteBuffer: env.numberNotOff("SWWAF_LOG_REMOTE_BUFFER", "10000"),
LogRemoteFacility: env.facility("SWWAF_LOG_REMOTE_FACILITY", "local0"),
AdminToken: env.token("SWWAF_ADMIN_TOKEN"),
MetricsToken: env.token("SWWAF_METRICS_TOKEN"),
MetricsTopN: env.numberNotOff("SWWAF_METRICS_TOP_N", "50"),
RulesDir: env.value("SWWAF_RULES_DIR", "/etc/smallwebwaf/rules.d"),
RulesEnabled: env.boolean("SWWAF_RULES_ENABLED", "true"),
LogRemoteURL: env.logRemoteURL("SWWAF_LOG_REMOTE_URL"),
LogRemoteTLSCAs: env.certificates("SWWAF_LOG_REMOTE_TLS_CA_FILE"),
LogRemoteBuffer: env.numberNotOff("SWWAF_LOG_REMOTE_BUFFER", "10000"),
LogRemoteFacility: env.facility("SWWAF_LOG_REMOTE_FACILITY", "local0"),
AlertWebhookURL: env.webhookURL("SWWAF_ALERT_WEBHOOK_URL"),
AlertWebhookHeaders: env.webhookHeaders("SWWAF_ALERT_WEBHOOK_HEADERS"),
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"),
}
cfg.LogRemoteAppName = env.appName("SWWAF_LOG_REMOTE_APP_NAME",
@@ -636,6 +666,49 @@ func (e *environment) appName(name, instanceName string, sending bool) string {
return value
}
// webhookURL reads the setting that is where each alert is posted. Unset
// or empty, it is nil, and no alert is sent.
func (e *environment) webhookURL(name string) *url.URL {
value := e.value(name, "")
if value == "" {
return nil
}
webhook, err := parseWebhookURL(value)
e.check(name, err)
return webhook
}
// webhookHeaders reads the setting that is the headers sent with each
// alert. The log shows each header's value as ********, since a header
// such as Authorization carries a secret.
func (e *environment) webhookHeaders(name string) http.Header {
value, _ := e.lookup(name)
headers, logged, err := parseWebhookHeaders(value)
e.settings = append(e.settings, slog.String(name, logged))
e.check(name, err)
return headers
}
// alertEvents reads the setting that is the events alerts are sent for.
func (e *environment) alertEvents(name, defaultValue string) []string {
events, err := parseAlertEvents(e.value(name, defaultValue))
e.check(name, err)
return events
}
// numberOrOff reads a setting that is a whole number above zero, or off,
// which is 0.
func (e *environment) numberOrOff(name, defaultValue string) int {
number, err := parseNumberOrOff(e.value(name, defaultValue))
e.check(name, err)
return number
}
// parseDuration reads a duration in Go's syntax, such as 90s or 15m, a
// whole number of days such as 7d, or off.
func parseDuration(value string) (time.Duration, error) {
@@ -1051,6 +1124,85 @@ func parseFacility(value string) (int, error) {
return number, nil
}
// parseWebhookURL reads where each alert is posted: http or https, a
// host, and an optional port from 1 to 65535, path and query, without a
// user or a fragment.
func parseWebhookURL(value string) (*url.URL, error) {
webhook, err := url.Parse(value)
if err != nil {
return nil, fmt.Errorf("%q %w", value, errNotWebhookURL)
}
port, err := strconv.ParseUint(webhook.Port(), 10, 16)
valid := (webhook.Scheme == "http" || webhook.Scheme == "https") &&
webhook.Hostname() != "" && (webhook.Port() == "" || (err == nil && port != 0)) &&
webhook.User == nil && webhook.Opaque == "" && webhook.Fragment == ""
if !valid {
return nil, fmt.Errorf("%q %w", value, errNotWebhookURL)
}
return webhook, nil
}
// parseWebhookHeaders reads a comma-separated list of headers, each its
// name, :, and its value, and returns them, and how the log shows them,
// with each value as ********. An error names the item by its place in
// the list, so that it shows no value. An empty value is an empty list.
func parseWebhookHeaders(value string) (http.Header, string, error) {
headers := http.Header{}
if strings.TrimSpace(value) == "" {
return headers, "", nil
}
logged := []string{}
for i, item := range strings.Split(value, ",") {
name, headerValue, found := strings.Cut(item, ":")
name = strings.TrimSpace(name)
if !found || !IsHeaderName(name) || strings.ContainsAny(headerValue, "\r\n\x00") {
return nil, "", fmt.Errorf("item %d %w", i+1, errNotWebhookHeader)
}
headers.Add(name, strings.TrimSpace(headerValue))
logged = append(logged, name+":"+masked)
}
return headers, strings.Join(logged, ","), nil
}
// parseAlertEvents reads a comma-separated list of the events alerts can
// be sent for.
func parseAlertEvents(value string) ([]string, error) {
events, err := parseList(value)
if err != nil {
return nil, err
}
for _, event := range events {
if !slices.Contains(alerts.Events(), event) {
return nil, fmt.Errorf("%q %w", event, errNotAlertEvent)
}
}
return events, nil
}
// parseNumberOrOff reads a whole number above zero, or off, which is 0.
func parseNumberOrOff(value string) (int, error) {
if value == off {
return 0, nil
}
n, err := strconv.Atoi(value)
if err != nil || n <= 0 {
return 0, fmt.Errorf("%q %w", value, errNotNumberOrOff)
}
return n, nil
}
// appNameMaxLength is the most characters RFC 5424 allows in an
// APP-NAME.
const appNameMaxLength = 48
+126
View File
@@ -6,9 +6,11 @@ import (
"encoding/json"
"log/slog"
"maps"
"net/http"
"net/netip"
"os"
"path/filepath"
"reflect"
"slices"
"strings"
"testing"
@@ -62,6 +64,19 @@ const (
logRemoteBuffer = "SWWAF_LOG_REMOTE_BUFFER"
logRemoteFacility = "SWWAF_LOG_REMOTE_FACILITY"
logRemoteAppName = "SWWAF_LOG_REMOTE_APP_NAME"
alertWebhookURL = "SWWAF_ALERT_WEBHOOK_URL"
alertWebhookHeaders = "SWWAF_ALERT_WEBHOOK_HEADERS"
alertEvents = "SWWAF_ALERT_EVENTS"
alertCooldown = "SWWAF_ALERT_COOLDOWN"
alertMaxPerHour = "SWWAF_ALERT_MAX_PER_HOUR"
)
// defaultAlertEvents is the default of SWWAF_ALERT_EVENTS, and
// defaultAlertCooldown that of SWWAF_ALERT_COOLDOWN.
const (
defaultAlertEvents = "ban,permanent_ban,waf_block,anomaly,reputation_hit," +
"source_failure,file_error"
defaultAlertCooldown = "15m"
)
// defaultLogRequestHeaders is the default of SWWAF_LOG_REQUEST_HEADERS.
@@ -477,6 +492,112 @@ func TestAppNameSetStopsTheStartWhileSending(t *testing.T) {
}
}
func TestAlertSettingsDefaults(t *testing.T) {
t.Parallel()
cfg := fromEnvironment(t, environment{})
if cfg.AlertWebhookURL != nil || len(cfg.AlertWebhookHeaders) != 0 ||
strings.Join(cfg.AlertEvents, ",") != defaultAlertEvents ||
cfg.AlertCooldown != 15*time.Minute || cfg.AlertMaxPerHour != 60 {
t.Errorf("alert settings %v, %v, %v, %s and %d, want no URL, no headers, "+
"%s, 15m and 60", cfg.AlertWebhookURL, cfg.AlertWebhookHeaders,
cfg.AlertEvents, cfg.AlertCooldown, cfg.AlertMaxPerHour, defaultAlertEvents)
}
}
func TestAlertSettingsAsSet(t *testing.T) {
t.Parallel()
const webhook = "https://alerts.example:8443/hooks/waf?team=ops"
cfg := fromEnvironment(t, environment{
alertWebhookURL: webhook,
alertWebhookHeaders: "Authorization: Bearer abc:def , x-team:ops",
alertEvents: "ban, file_error",
alertCooldown: "1h",
alertMaxPerHour: "10",
})
headers := http.Header{"Authorization": {"Bearer abc:def"}, "X-Team": {"ops"}}
if cfg.AlertWebhookURL.String() != webhook ||
!reflect.DeepEqual(cfg.AlertWebhookHeaders, headers) ||
!slices.Equal(cfg.AlertEvents, []string{"ban", "file_error"}) ||
cfg.AlertCooldown != time.Hour || cfg.AlertMaxPerHour != 10 {
t.Errorf("alert settings %v, %v, %v, %s and %d", cfg.AlertWebhookURL,
cfg.AlertWebhookHeaders, cfg.AlertEvents, cfg.AlertCooldown,
cfg.AlertMaxPerHour)
}
cfg = fromEnvironment(t, environment{
alertWebhookURL: "", alertEvents: "", alertCooldown: off, alertMaxPerHour: off,
})
if cfg.AlertWebhookURL != nil || len(cfg.AlertEvents) != 0 ||
cfg.AlertCooldown != 0 || cfg.AlertMaxPerHour != 0 {
t.Errorf("set empty or off, alert settings %v, %v, %s and %d",
cfg.AlertWebhookURL, cfg.AlertEvents, cfg.AlertCooldown, cfg.AlertMaxPerHour)
}
}
func TestInvalidAlertSettingStopsTheStart(t *testing.T) {
t.Parallel()
wantStartStopped(t, []struct{ name, value string }{
{alertWebhookURL, "alerts.example/smallwebwaf"},
{alertWebhookURL, "ftp://alerts.example/"},
{alertWebhookURL, "https:///smallwebwaf"},
{alertWebhookURL, "https://user:password@alerts.example/"},
{alertWebhookURL, "https://alerts.example/#top"},
{alertWebhookURL, "https://alerts.example:0/"},
{alertWebhookURL, "https://alerts.example:65536/"},
{alertWebhookHeaders, "Authorization"},
{alertWebhookHeaders, "X Team:ops"},
{alertWebhookHeaders, ":ops"},
{alertWebhookHeaders, "X-Team:ops,"},
{alertWebhookHeaders, "X-Team:o\r\nps"},
{alertEvents, "bans"},
{alertEvents, "summary"},
{alertEvents, "ban,,file_error"},
{alertCooldown, "0"},
{alertCooldown, "soon"},
{alertMaxPerHour, "0"},
{alertMaxPerHour, "-1"},
{alertMaxPerHour, "1.5"},
})
}
func TestWebhookHeadersAreLoggedMaskedAndNeverShown(t *testing.T) {
t.Parallel()
const secret = "Bearer 0123456789abcdef"
cfg := fromEnvironment(t, environment{
alertWebhookHeaders: "Authorization:" + secret + ",X-Team:ops",
})
var out bytes.Buffer
slog.New(slog.NewJSONHandler(&out, nil)).Info("starting", "settings", cfg)
logged := out.String()
if strings.Contains(logged, secret) || strings.Contains(logged, "ops") ||
!strings.Contains(logged,
`"`+alertWebhookHeaders+`":"Authorization:********,X-Team:********"`) {
t.Errorf("the headers are not logged masked: %s", logged)
}
// An item that is not a header is named by its place, not shown.
_, err := config.FromEnvironment(environment{
alertWebhookHeaders: "X-Team:ops," + secret,
}.lookupEnv)
want := alertWebhookHeaders + ": item 2 is not a header name followed by : " +
"and the header's value, such as Authorization:Bearer <token>"
if err == nil || err.Error() != want {
t.Errorf("error %v, want %s", err, want)
}
}
func TestCodeOnBothCountryListsStopsTheStart(t *testing.T) {
t.Parallel()
@@ -926,6 +1047,11 @@ func TestLogsEachSettingWithItsValue(t *testing.T) {
logRemoteBuffer: "10000",
logRemoteFacility: "local0",
logRemoteAppName: hostname,
alertWebhookURL: "",
alertWebhookHeaders: "",
alertEvents: defaultAlertEvents,
alertCooldown: defaultAlertCooldown,
alertMaxPerHour: "60",
}
if !maps.Equal(line.Settings, want) {
t.Errorf("logged settings\n%v\nwant\n%v", line.Settings, want)
+13
View File
@@ -19,6 +19,7 @@ import (
"time"
"github.com/hashicorp/golang-lru/v2/simplelru"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/metrics"
)
@@ -68,6 +69,8 @@ type Params struct {
// Metrics count the requests to GeoJS, those that failed, and the
// clients that go without an answer.
Metrics *metrics.Metrics
// Alerts receive a source_failure alert each time GeoJS fails.
Alerts *alerts.Queue
}
// GeoJS looks up clients' countries through GeoJS. At most one request
@@ -78,6 +81,7 @@ type GeoJS struct {
now func() time.Time
processLog *slog.Logger
metrics *metrics.Metrics
alerts *alerts.Queue
// httpClient follows no redirect, so that visitors' addresses go to
// GeoJS alone: a redirect is a failure.
httpClient *http.Client
@@ -127,6 +131,7 @@ func New(params Params) *GeoJS {
now: params.Now,
processLog: params.ProcessLog,
metrics: params.Metrics,
alerts: params.Alerts,
httpClient: &http.Client{
CheckRedirect: func(*http.Request, []*http.Request) error {
return http.ErrUseLastResponse
@@ -386,6 +391,14 @@ func (g *GeoJS) keep(
g.processLog.Warn("asking GeoJS failed",
"error", err.Error(), "asking_again_in", g.retryDelay.String())
g.alerts.Raise(alerts.Alert{
Event: alerts.EventSourceFailure,
Reason: "asking GeoJS failed",
Detail: map[string]any{
"source": "geojs", "error": err.Error(),
"asking_again_in": g.retryDelay.String(),
},
})
return false
}
+61 -1
View File
@@ -6,6 +6,8 @@ import (
"net/http"
"net/http/httptest"
"net/netip"
"net/url"
"reflect"
"slices"
"strings"
"sync"
@@ -14,6 +16,7 @@ import (
"time"
"github.com/prometheus/client_golang/prometheus/testutil"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/lookup"
"sneak.berlin/go/smallwebwaf/internal/metrics"
)
@@ -197,6 +200,7 @@ func TestFailureIsLoggedWithoutTheAddressesAskedAbout(t *testing.T) {
Now: time.Now,
ProcessLog: slog.New(slog.NewTextHandler(&log, nil)),
Metrics: metrics.New(1),
Alerts: alerts.New(alerts.Params{}),
})
g.SetTransport(geojs)
@@ -211,6 +215,45 @@ func TestFailureIsLoggedWithoutTheAddressesAskedAbout(t *testing.T) {
})
}
func TestFailureRaisesASourceFailureAlertOncePerCooldown(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
geojs, clock, g, queue := startWithAlerts()
clients := newClients()
geojs.set(failing)
wantCountry(t, g, clients(), "")
want := alerts.Alert{
Time: clock.Now(),
Event: alerts.EventSourceFailure,
Reason: "asking GeoJS failed",
Detail: map[string]any{
"source": "geojs",
"error": "GeoJS answered 503 Service Unavailable",
"asking_again_in": "1s",
},
}
// The next failure, a second later, is a repeat within the
// cooldown.
clock.advance(time.Second)
wantCountry(t, g, clients(), "")
wantRequests(t, geojs, 2)
waiting := queue.Snapshot().Waiting
if len(waiting) != 1 || !reflect.DeepEqual(waiting[0], want) {
t.Errorf("alerts waiting %+v, want only %+v", waiting, want)
}
if queue.Suppressed() != 1 {
t.Errorf("%d alerts held back, want the repeat", queue.Suppressed())
}
})
}
func TestWaitingClientsAreAskedAboutInOneRequest(t *testing.T) {
t.Parallel()
@@ -361,6 +404,7 @@ func TestClientsWithoutAnAnswerAreCounted(t *testing.T) {
Now: time.Now,
ProcessLog: slog.New(slog.DiscardHandler),
Metrics: m,
Alerts: alerts.New(alerts.Params{}),
})
g.SetTransport(&standIn{answers: failing})
@@ -524,17 +568,33 @@ func (c *testClock) advance(d time.Duration) {
// start returns a stand-in for GeoJS that answers, a clock, and a GeoJS
// asking the stand-in by that clock.
func start() (*standIn, *testClock, *lookup.GeoJS) {
geojs, clock, g, _ := startWithAlerts()
return geojs, clock, g
}
// startWithAlerts is start, and returns the queue of the alerts GeoJS
// raises as well, for a webhook that is never sent them, with the default
// cooldown, by the same clock.
func startWithAlerts() (*standIn, *testClock, *lookup.GeoJS, *alerts.Queue) {
geojs := &standIn{}
clock := &testClock{now: time.Date(2026, 10, 4, 0, 0, 0, 0, time.UTC)}
queue := alerts.New(alerts.Params{
WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"},
Events: alerts.Events(),
Cooldown: 15 * time.Minute,
Now: clock.Now,
})
g := lookup.New(lookup.Params{
URL: lookup.URL,
Now: clock.Now,
ProcessLog: slog.New(slog.DiscardHandler),
Metrics: metrics.New(1),
Alerts: queue,
})
g.SetTransport(geojs)
return geojs, clock, g
return geojs, clock, g, queue
}
// newClients returns what returns a new IPv4 client each time it is
+41
View File
@@ -11,6 +11,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/collectors"
"github.com/prometheus/client_golang/prometheus/promhttp"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/bans"
"sneak.berlin/go/smallwebwaf/internal/ratelimit"
"sneak.berlin/go/smallwebwaf/internal/remotelog"
@@ -225,6 +226,46 @@ func (m *Metrics) AddRemoteLog(remote *remotelog.Sender) {
)
}
// AddAlerts adds the metrics of the alerts sent to
// SWWAF_ALERT_WEBHOOK_URL, read from queue as the metrics are asked for,
// with the destination webhook: the alerts sent, the requests to the
// webhook that failed, and the alerts held back and dropped.
func (m *Metrics) AddAlerts(queue *alerts.Queue) {
webhook := prometheus.Labels{"destination": "webhook"}
m.registry.MustRegister(
prometheus.NewCounterFunc(prometheus.CounterOpts{
Name: "smallwebwaf_alerts_sent_total",
Help: "Alerts the destination took.",
ConstLabels: webhook,
}, func() float64 {
return float64(queue.Sent())
}),
prometheus.NewCounterFunc(prometheus.CounterOpts{
Name: "smallwebwaf_alerts_failed_total",
Help: "Requests to the destination that failed.",
ConstLabels: webhook,
}, func() float64 {
return float64(queue.Failed())
}),
prometheus.NewCounterFunc(prometheus.CounterOpts{
Name: "smallwebwaf_alerts_suppressed_total",
Help: "Alerts held back: repeats within SWWAF_ALERT_COOLDOWN, and " +
"alerts past SWWAF_ALERT_MAX_PER_HOUR, for the hour's summary.",
ConstLabels: webhook,
}, func() float64 {
return float64(queue.Suppressed())
}),
prometheus.NewCounterFunc(prometheus.CounterOpts{
Name: "smallwebwaf_alerts_dropped_total",
Help: "Alerts dropped, the oldest first, from a full queue.",
ConstLabels: webhook,
}, func() float64 {
return float64(queue.Dropped())
}),
)
}
// ServeHTTP answers with the metrics in the Prometheus text format.
func (m *Metrics) ServeHTTP(w http.ResponseWriter, r *http.Request) {
m.handler.ServeHTTP(w, r)
+174
View File
@@ -0,0 +1,174 @@
package proxy_test
import (
"maps"
"net/http"
"net/netip"
"reflect"
"testing"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/bans"
"sneak.berlin/go/smallwebwaf/internal/proxy"
"sneak.berlin/go/smallwebwaf/internal/requestlog"
)
const (
alertWebhookURL = "SWWAF_ALERT_WEBHOOK_URL"
// alertInstance is the instance every alert of these tests gives.
alertInstance = "fsn1app1/gitea"
)
func TestBanForABrokenLimitRaisesABanAlertWithItsNotes(t *testing.T) {
t.Parallel()
s, clk, server, queue := startWithAlerts(t, map[string]string{
rateLimitPerMinute: "1",
banScopeV4Prefix: "24",
})
start := clk.Now()
s.get(client, http.StatusOK, requestlog.ActionForward)
s.get(client, http.StatusForbidden, requestlog.ActionRateLimited)
netblock := netip.MustParsePrefix("203.0.113.0/24")
ban := server.Ledger.Bans(netblock)[0]
// A request refused under the ban raises no other alert.
clk.advance(time.Minute)
s.get(client, http.StatusForbidden, requestlog.ActionBanned)
wantAlerts(t, queue, alerts.Alert{
Instance: alertInstance,
Time: start,
Event: alerts.EventBan,
Client: netip.MustParseAddr(client),
Netblock: netblock,
Reason: "requests per minute over the limit of 1",
Detail: map[string]any{
"cause": bans.CauseLimit,
"ban_expires": requestlog.FormatTime(start.Add(time.Hour)),
"notes": ban.Notes,
},
})
if ban.Notes.Limit != 1 || ban.Notes.Request.Path != "/" {
t.Errorf("the alert's notes are %+v, want those of the broken limit", ban.Notes)
}
}
func TestAttackBanRaisesABanAlertThenAPermanentBanAlert(t *testing.T) {
t.Parallel()
s, clk, server, queue := startWithAlerts(t, map[string]string{
rulesDir: writeRules(t, testRules),
})
start := clk.Now()
netblock := netip.MustParsePrefix(client + "/32")
other := netip.MustParsePrefix(otherClient + "/32")
// The probe bans the client for seven days, and its next request makes
// the ban permanent. The request after that changes nothing.
s.request(client, "/.env", http.StatusForbidden, requestlog.ActionBanned)
attackBan := server.Ledger.Bans(netblock)[0]
clk.advance(time.Minute)
s.get(client, http.StatusForbidden, requestlog.ActionBanned)
permanentBan := server.Ledger.Bans(netblock)[0]
s.get(client, http.StatusForbidden, requestlog.ActionBanned)
// Another client's probe after its first ban has run out without a
// request makes a permanent ban at once.
s.request(otherClient, "/.env", http.StatusForbidden, requestlog.ActionBanned)
clk.advance(7 * 24 * time.Hour)
s.request(otherClient, "/.env", http.StatusForbidden, requestlog.ActionBanned)
otherBans := server.Ledger.Bans(other)
wantAlerts(t, queue,
attackAlert(alerts.EventBan, start, client, attackBan,
requestlog.FormatTime(start.Add(7*24*time.Hour))),
attackAlert(alerts.EventPermanentBan, start.Add(time.Minute), client,
permanentBan, "permanent"),
attackAlert(alerts.EventBan, start.Add(time.Minute), otherClient, otherBans[0],
requestlog.FormatTime(start.Add(time.Minute+7*24*time.Hour))),
attackAlert(alerts.EventPermanentBan, start.Add(time.Minute+7*24*time.Hour),
otherClient, otherBans[1], "permanent"),
)
}
func TestObserveModeRaisesNoBanAlert(t *testing.T) {
t.Parallel()
s, _, _, queue := startWithAlerts(t, map[string]string{
mode: "observe",
rateLimitPerMinute: "1",
rulesDir: writeRules(t, testRules),
})
s.get(client, http.StatusOK, requestlog.ActionForward)
s.get(client, http.StatusOK, requestlog.ActionForward)
s.request(otherClient, "/.env", http.StatusOK, requestlog.ActionForward)
wantAlerts(t, queue)
}
// startWithAlerts is startWithClock with alerts to a webhook, which is
// never sent them, and returns the queue they wait in as well.
func startWithAlerts(
t *testing.T, env map[string]string,
) (*sender, *clock, *proxy.Server, *alerts.Queue) {
t.Helper()
app := startApp(t, func(http.ResponseWriter, *http.Request) {})
clk := &clock{now: time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)}
settings := map[string]string{
trustedProxies: trustLocalhost,
alertWebhookURL: "https://alerts.example/smallwebwaf",
instanceName: alertInstance,
}
maps.Copy(settings, env)
addr, out, server, queue := startProxyWithAlerts(t, app.URL, "", clk.Now, settings)
return &sender{t: t, addr: addr, out: out}, clk, server, queue
}
// attackAlert returns the alert for event, raised by a request from client
// at the time raised, for ban, a ban for the probe rule of testRules,
// which ends at expires, as the log line gives it.
func attackAlert(
event string, raised time.Time, client string, ban bans.Ban, expires string,
) alerts.Alert {
return alerts.Alert{
Instance: alertInstance,
Time: raised,
Event: event,
Client: netip.MustParseAddr(client),
Netblock: ban.Netblock,
Reason: "matched the rule probe",
Detail: map[string]any{
"cause": bans.CauseAttack, "ban_expires": expires, "notes": ban.Notes,
},
}
}
// wantAlerts checks the alerts waiting in queue, in order.
func wantAlerts(t *testing.T, queue *alerts.Queue, want ...alerts.Alert) {
t.Helper()
got := queue.Snapshot().Waiting
if len(got) != len(want) {
t.Fatalf("%d alerts wait, want %d: %+v", len(got), len(want), got)
}
for i := range want {
if !reflect.DeepEqual(got[i], want[i]) {
t.Errorf("alert %d is\n%+v\nwant\n%+v", i, got[i], want[i])
}
}
}
+47 -6
View File
@@ -4,6 +4,7 @@ import (
"net/netip"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/bans"
"sneak.berlin/go/smallwebwaf/internal/requestlog"
"sneak.berlin/go/smallwebwaf/internal/rules"
@@ -16,14 +17,25 @@ func (rq *request) banResponse(action string) *refusal {
}
// banned reports whether a ban on a netblock the client is in covers the
// request at now, and notes for the log line when that ban ends.
// request at now, and notes for the log line when that ban ends. A
// request that makes the ban permanent raises the alert for it.
func (rq *request) banned(now time.Time) bool {
check := rq.h.ledger.Check
var (
ban bans.Ban
banned bool
madePermanent bool
)
if rq.h.config.Observe {
check = rq.h.ledger.Find // in observe mode the ban refuses nothing
ban, banned = rq.h.ledger.Find(rq.client, now) // the ban refuses nothing
} else {
ban, banned, madePermanent = rq.h.ledger.Check(rq.client, now)
}
if madePermanent {
rq.alertBan(ban)
}
ban, banned := check(rq.client, now)
if banned {
rq.line.BanExpires = banExpires(ban)
}
@@ -54,7 +66,7 @@ func (rq *request) limitBroken(now time.Time) bool {
}
netblock := rq.h.netblock(rq.client)
ban := rq.h.ledger.BanForLimit(netblock, now, bans.Notes{
ban, made := rq.h.ledger.BanForLimit(netblock, now, bans.Notes{
Country: rq.line.Country,
Limit: hit.Limit,
Window: hit.Window,
@@ -65,6 +77,10 @@ func (rq *request) limitBroken(now time.Time) bool {
rq.h.limiter.Reset(group)
rq.line.BanExpires = banExpires(ban)
if made {
rq.alertBan(ban)
}
return true
}
@@ -72,7 +88,7 @@ func (rq *request) limitBroken(now time.Time) bool {
// attack, the match of rule, a ban rule.
func (rq *request) banForAttack(now time.Time, rule rules.Rule) {
netblock := rq.h.netblock(rq.client)
ban := rq.h.ledger.BanForAttack(netblock, now, bans.Notes{
ban, made := rq.h.ledger.BanForAttack(netblock, now, bans.Notes{
Country: rq.line.Country,
RuleID: rule.ID,
Target: rule.Target,
@@ -80,6 +96,31 @@ func (rq *request) banForAttack(now time.Time, rule rules.Rule) {
Requests: rq.netblockRequests(netblock),
})
rq.line.BanExpires = banExpires(ban)
if made {
rq.alertBan(ban)
}
}
// alertBan raises the alert for ban, which the request made, or made
// permanent: permanent_ban for a permanent ban, ban for another. Its
// detail gives the ban's cause, when it ends, and its notes.
func (rq *request) alertBan(ban bans.Ban) {
event := alerts.EventBan
if ban.Permanent() {
event = alerts.EventPermanentBan
}
rq.h.alerts.Raise(alerts.Alert{
Event: event,
Client: rq.client,
Netblock: ban.Netblock,
Country: ban.Notes.Country,
Reason: ban.Reason,
Detail: map[string]any{
"cause": ban.Cause, "ban_expires": banExpires(ban), "notes": ban.Notes,
},
})
}
// noted is the request, refused at now with SWWAF_BAN_RESPONSE, as the
+8 -1
View File
@@ -11,6 +11,7 @@ import (
"strings"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/bans"
"sneak.berlin/go/smallwebwaf/internal/config"
"sneak.berlin/go/smallwebwaf/internal/lookup"
@@ -63,6 +64,9 @@ type Params struct {
// Rules are the rule files' rules, which each request is checked
// against.
Rules *rules.Files
// Alerts receive the alert for each ban the proxy makes or makes
// permanent, and for GeoJS failing.
Alerts *alerts.Queue
}
// Server is the server smallwebwaf runs, with the parts of the proxy
@@ -110,8 +114,10 @@ func New(params Params) *Server {
Now: params.Now,
ProcessLog: params.ProcessLog,
Metrics: m,
Alerts: params.Alerts,
}),
rules: params.Rules,
rules: params.Rules,
alerts: params.Alerts,
}
m.AddBansAndClients(h.ledger, h.limiter, params.Now)
m.AddRules(params.Rules)
@@ -150,6 +156,7 @@ type handler struct {
ledger *bans.Ledger
geojs *lookup.GeoJS
rules *rules.Files
alerts *alerts.Queue
}
// newTransport returns what carries requests to the app. It never goes
+27 -1
View File
@@ -14,6 +14,7 @@ import (
"testing"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/config"
"sneak.berlin/go/smallwebwaf/internal/proxy"
"sneak.berlin/go/smallwebwaf/internal/requestlog"
@@ -223,6 +224,20 @@ func startProxyWithClock(
) (string, *output, *proxy.Server) {
t.Helper()
addr, out, server, _ := startProxyWithAlerts(t, appURL, geojsURL, now, env)
return addr, out, server
}
// startProxyWithAlerts is startProxyWithClock, and returns the queue of
// the alerts the proxy raises as well, as the settings in env make it. No
// alert is sent from it: they wait in it, for the test to look at.
func startProxyWithAlerts(
t *testing.T, appURL, geojsURL string, now func() time.Time,
env map[string]string,
) (string, *output, *proxy.Server, *alerts.Queue) {
t.Helper()
settings := map[string]string{"SWWAF_UPSTREAM_URL": appURL, rulesDir: t.TempDir()}
maps.Copy(settings, env)
@@ -245,6 +260,16 @@ func startProxyWithClock(
t.Fatalf("rule files: %v", err)
}
alertQueue := alerts.New(alerts.Params{
WebhookURL: cfg.AlertWebhookURL,
Events: cfg.AlertEvents,
Cooldown: cfg.AlertCooldown,
MaxPerHour: cfg.AlertMaxPerHour,
Instance: cfg.InstanceName,
Now: now,
ProcessLog: processLog,
})
server := proxy.New(proxy.Params{
Config: cfg,
RequestLog: out,
@@ -252,6 +277,7 @@ func startProxyWithClock(
GeoJSURL: geojsURL,
Now: now,
Rules: ruleFiles,
Alerts: alertQueue,
})
listener, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0")
@@ -267,7 +293,7 @@ func startProxyWithClock(
_ = server.Close()
})
return listener.Addr().String(), out, server
return listener.Addr().String(), out, server, alertQueue
}
// newClient returns an HTTP client that sends requests as they are made,
+15 -4
View File
@@ -22,6 +22,7 @@ import (
"github.com/fsnotify/fsnotify"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/config"
)
@@ -100,6 +101,8 @@ type Params struct {
// ProcessLog receives how many rules were read, and the error in a
// rule file edited while smallwebwaf runs.
ProcessLog *slog.Logger
// Alerts receive a file_error alert for that error.
Alerts *alerts.Queue
}
// Files are the rule files of a running smallwebwaf, and the rules read
@@ -223,13 +226,21 @@ func (f *Files) readAfterChanges(
}
// readAgain reads the rule files again, in place of the rules loaded, or
// logs the error that keeps the rules as they were.
// logs the error that keeps the rules as they were, and raises a
// file_error alert for it.
func (f *Files) readAgain() {
rules, err := read(f.params.Dir)
if err != nil {
f.params.ProcessLog.Error(
"a rule file has an error, and the rules stay as they were",
"error", err.Error())
const kept = "a rule file has an error, and the rules stay as they were"
// Raised before it is logged, so that the alert is there once the
// log line is.
f.params.Alerts.Raise(alerts.Alert{
Event: alerts.EventFileError,
Reason: kept,
Detail: map[string]any{"error": err.Error()},
})
f.params.ProcessLog.Error(kept, "error", err.Error())
return
}
+31 -7
View File
@@ -7,12 +7,15 @@ import (
"maps"
"net/http"
"net/http/httptest"
"net/url"
"os"
"path/filepath"
"slices"
"strconv"
"testing"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/rules"
)
@@ -337,7 +340,7 @@ func TestEditsTakenInWhileRunning(t *testing.T) {
t.Parallel()
dir := writeFiles(t, ruleFiles{firstFile: "first path block ^/first\n"})
files, lines := watch(t, dir)
files, lines, _ := watch(t, dir)
// matches reports whether path matches a rule.
matches := func(path string) bool { return len(files.Match(get(t, path))) == 1 }
@@ -366,7 +369,7 @@ func TestBrokenEditKeepsTheRulesAsTheyWere(t *testing.T) {
t.Parallel()
dir := writeFiles(t, ruleFiles{firstFile: "first path block ^/first\n"})
files, lines := watch(t, dir)
files, lines, queue := watch(t, dir)
// The edit's second line has an unknown action, so the rules stay as
// they were, the first line's earlier version included.
@@ -380,13 +383,26 @@ func TestBrokenEditKeepsTheRulesAsTheyWere(t *testing.T) {
t.Errorf("logged %v, want an error %q", line, want)
}
// The error is raised as a file_error alert too.
wantFileError := func() {
t.Helper()
waiting := queue.Snapshot().Waiting
if len(waiting) != 1 || waiting[0].Event != alerts.EventFileError ||
waiting[0].Reason != hasError || waiting[0].Detail["error"] != want {
t.Errorf("alerts waiting %+v, want one file_error alert for %q", waiting, want)
}
}
wantFileError()
wantMatched(t, files, get(t, "/first"), "first")
wantMatched(t, files, get(t, "/second"))
// Once mended, the file is read again.
// Once mended, the file is read again, and raises no alert.
save(t, dir, firstFile, "first path block ^/edited\nsecond path ban ^/second\n")
lines.waitUntil(t, func() bool { return len(files.Match(get(t, "/second"))) == 1 })
wantMatched(t, files, get(t, "/edited"), "first")
wantFileError()
}
func TestDefaultFileBansProbesAtTheSiteRootAlone(t *testing.T) {
@@ -504,7 +520,8 @@ func save(t *testing.T, dir, name, content string) {
}
// newParams returns Params for the rule files in dir, switched on, with
// the process log in the processLog returned.
// the process log in the processLog returned, and the alerts waiting in a
// queue for a webhook that is never sent them.
func newParams(dir string) (rules.Params, processLog) {
lines := make(processLog, maxLogLines)
@@ -512,6 +529,12 @@ func newParams(dir string) (rules.Params, processLog) {
Dir: dir,
Enabled: true,
ProcessLog: slog.New(slog.NewJSONHandler(lines, nil)),
Alerts: alerts.New(alerts.Params{
WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"},
Events: alerts.Events(),
Cooldown: 15 * time.Minute,
Now: time.Now,
}),
}, lines
}
@@ -531,8 +554,9 @@ func load(t *testing.T, files ruleFiles) *rules.Files {
}
// watch loads the rules in dir, runs their Watch until the test ends, and
// waits until it watches the directory.
func watch(t *testing.T, dir string) (*rules.Files, processLog) {
// waits until it watches the directory. It returns the alerts' queue as
// well.
func watch(t *testing.T, dir string) (*rules.Files, processLog, *alerts.Queue) {
t.Helper()
params, lines := newParams(dir)
@@ -557,7 +581,7 @@ func watch(t *testing.T, dir string) (*rules.Files, processLog) {
lines.waitFor(t, watching)
return files, lines
return files, lines, params.Alerts
}
// wantRefused checks that loading the rule files in dir fails with the
+3
View File
@@ -13,6 +13,8 @@ import (
"time"
"github.com/fsnotify/fsnotify"
"sneak.berlin/go/smallwebwaf/internal/alerts"
)
// The tests below run readAfterChanges in a synctest bubble, where time is
@@ -91,6 +93,7 @@ func load(t *testing.T, dir string) *Files {
files, err := Load(Params{
Dir: dir, Enabled: true, ProcessLog: slog.New(slog.DiscardHandler),
Alerts: alerts.New(alerts.Params{}),
})
if err != nil {
t.Fatalf("load: %v", err)
+77 -38
View File
@@ -15,6 +15,7 @@ import (
"syscall"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/config"
"sneak.berlin/go/smallwebwaf/internal/lookup"
"sneak.berlin/go/smallwebwaf/internal/proxy"
@@ -91,10 +92,16 @@ func Run(ctx context.Context, params Params) int {
defer stopSending()
}
// The state files and the alerts give times in UTC.
now := func() time.Time { return time.Now().UTC() }
alertQueue := newAlertQueue(cfg, now, processLog)
ruleFiles, err := rules.Load(rules.Params{
Dir: cfg.RulesDir,
Enabled: cfg.RulesEnabled,
ProcessLog: processLog,
Alerts: alertQueue,
})
if err != nil {
processLog.Error("cannot use the rule files", "error", err.Error())
@@ -102,9 +109,6 @@ func Run(ctx context.Context, params Params) int {
return 1
}
// The state files give times in UTC.
now := func() time.Time { return time.Now().UTC() }
server := proxy.New(proxy.Params{
Config: cfg,
RequestLog: stdout,
@@ -112,22 +116,17 @@ func Run(ctx context.Context, params Params) int {
GeoJSURL: lookup.URL,
Now: now,
Rules: ruleFiles,
Alerts: alertQueue,
})
if remote != nil {
server.Metrics.AddRemoteLog(remote)
}
files, err := state.Load(state.Params{
Dir: cfg.StateDir,
WriteDelay: cfg.StateWriteDelay,
CounterInterval: cfg.StateCounterInterval,
Ledger: server.Ledger,
Limiter: server.Limiter,
GeoJS: server.GeoJS,
Now: now,
ProcessLog: processLog,
Metrics: server.Metrics,
})
if cfg.AlertWebhookURL != nil {
server.Metrics.AddAlerts(alertQueue)
}
files, err := loadStateFiles(cfg, server, alertQueue, now, processLog)
if err != nil {
processLog.Error("cannot use the state files", "error", err.Error())
@@ -147,7 +146,44 @@ func Run(ctx context.Context, params Params) int {
"address", listener.Addr().String(),
"settings", cfg)
return serve(ctx, server.Server, listener, files, ruleFiles, processLog)
return serve(ctx, server.Server, listener, files, ruleFiles, alertQueue, processLog)
}
// loadStateFiles reads the state files into the parts of server and into
// alertQueue, as state.Load does.
func loadStateFiles(
cfg *config.Config, server *proxy.Server, alertQueue *alerts.Queue,
now func() time.Time, processLog *slog.Logger,
) (*state.Files, error) {
return state.Load(state.Params{
Dir: cfg.StateDir,
WriteDelay: cfg.StateWriteDelay,
CounterInterval: cfg.StateCounterInterval,
Ledger: server.Ledger,
Limiter: server.Limiter,
GeoJS: server.GeoJS,
Alerts: alertQueue,
Now: now,
ProcessLog: processLog,
Metrics: server.Metrics,
})
}
// newAlertQueue returns the queue of the alerts to SWWAF_ALERT_WEBHOOK_URL,
// with the settings for it.
func newAlertQueue(
cfg *config.Config, now func() time.Time, processLog *slog.Logger,
) *alerts.Queue {
return alerts.New(alerts.Params{
WebhookURL: cfg.AlertWebhookURL,
WebhookHeaders: cfg.AlertWebhookHeaders,
Events: cfg.AlertEvents,
Cooldown: cfg.AlertCooldown,
MaxPerHour: cfg.AlertMaxPerHour,
Instance: cfg.InstanceName,
Now: now,
ProcessLog: processLog,
})
}
// newRemoteLogSender returns a sender of the log lines to
@@ -188,12 +224,14 @@ func startSending(
}
// serve serves requests on listener, writes the state files as they are
// due, takes in an admin's edits of them, and reads the rule files again
// as they change, until ctx is done. Then it gives the requests in
// progress shutdownTimeout to finish, and writes every state file.
// due, takes in an admin's edits of them, reads the rule files again as
// they change, and sends the alerts, until ctx is done. Then it gives the
// requests in progress shutdownTimeout to finish, and writes every state
// file, alerts.json with the alerts still waiting.
func serve(
ctx context.Context, server *http.Server, listener net.Listener,
files *state.Files, ruleFiles *rules.Files, processLog *slog.Logger,
files *state.Files, ruleFiles *rules.Files, alertQueue *alerts.Queue,
processLog *slog.Logger,
) int {
served := make(chan error, 1)
@@ -204,24 +242,10 @@ func serve(
writing, stopWriting := context.WithCancel(ctx)
defer stopWriting()
written := make(chan struct{})
watched := make(chan struct{})
rulesWatched := make(chan struct{})
go func() {
files.Run(writing)
close(written)
}()
go func() {
files.Watch(writing)
close(watched)
}()
go func() {
ruleFiles.Watch(writing)
close(rulesWatched)
}()
written := inBackground(func() { files.Run(writing) })
watched := inBackground(func() { files.Watch(writing) })
rulesWatched := inBackground(func() { ruleFiles.Watch(writing) })
alertsSent := inBackground(func() { alertQueue.Run(writing) })
select {
case err := <-served:
@@ -253,7 +277,8 @@ func serve(
}
// Run and Watch have ended, so nothing else reads or writes the
// files. Every request has ended too, but for two kinds
// files, and no alert is being sent, so that alerts.json keeps every
// alert not yet sent. Every request has ended too, but for two kinds
// that Go's server does not wait for: one cut off because Shutdown
// timed out, and one whose connection switched protocols, such as a
// WebSocket. Such a request adds to its client's history only as it
@@ -262,6 +287,7 @@ func serve(
<-written
<-watched
<-rulesWatched
<-alertsSent
err = files.WriteAll()
if err != nil {
@@ -274,3 +300,16 @@ func serve(
return 0
}
// inBackground runs task on a goroutine of its own, and returns a channel
// that is closed once task has returned.
func inBackground(task func()) <-chan struct{} {
done := make(chan struct{})
go func() {
task()
close(done)
}()
return done
}
+166
View File
@@ -14,6 +14,7 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
@@ -547,6 +548,97 @@ func TestStalledRemoteLogEndpointHoldsUpNoRequest(t *testing.T) {
out.line(t, "type", "request")
}
func TestBanIsAlertedAndAnAlertNotSentIsKeptAcrossARestart(t *testing.T) {
t.Parallel()
webhook := startWebhook(t)
rules := t.TempDir()
err := os.WriteFile(filepath.Join(rules, "50-app.rules"),
[]byte(`probe path ban ^/\.env$`+"\n"), 0o600)
if err != nil {
t.Fatalf("write the rule file: %v", err)
}
dir := t.TempDir()
env := map[string]string{
listenAddr: localhost + ":0",
upstreamURL: startApp(t),
stateDir: dir,
rulesDir: rules,
"SWWAF_ALERT_WEBHOOK_URL": webhook.url,
"SWWAF_ALERT_WEBHOOK_HEADERS": "Authorization:Bearer " + adminSecret,
}
runUntilStopped(t, env, func(url string) {
// The probe bans the client, and the webhook is sent the alert.
wantRefused(t, url+".env")
post := webhook.waitFor(t, "ban", true)
if post.alert["client"] != localhost || post.alert["netblock"] != localhost+"/32" ||
post.authorization != "Bearer "+adminSecret {
t.Errorf("the webhook was sent %v, with Authorization %q", post.alert,
post.authorization)
}
// The webhook fails, so the alert for the ban made permanent by the
// client's next request waits.
webhook.failing.Store(true)
wantRefused(t, url)
webhook.waitFor(t, "permanent_ban", false)
})
// alerts.json keeps it as smallwebwaf stops, and once started again,
// smallwebwaf sends it.
var file struct {
Waiting []struct {
Event string `json:"event"`
} `json:"waiting"`
}
path := filepath.Join(dir, "alerts.json")
data, err := os.ReadFile(path) //nolint:gosec // a file in the test's directory
if err == nil {
err = json.Unmarshal(data, &file)
}
if err != nil || len(file.Waiting) != 1 || file.Waiting[0].Event != "permanent_ban" {
t.Fatalf("alerts.json holds %s (%v), want the permanent_ban alert waiting", data, err)
}
// It counts the alert sent in the metrics, read here from a client the
// ban does not cover.
const token = "0123456789abcdef0123456789abcdef"
webhook.failing.Store(false)
env["SWWAF_ALLOW_NETS"] = localhost
env["SWWAF_METRICS_TOKEN"] = token
runUntilStopped(t, env, func(url string) {
webhook.waitFor(t, "permanent_ban", true)
// As long as that takes, so that a slow test process cannot fail
// the test.
const sent = "\nsmallwebwaf_alerts_sent_total{destination=\"webhook\"} 1\n"
metrics := metricsText(t, url+"_smallwebwaf/metrics", token)
for !strings.Contains(metrics, sent) {
time.Sleep(pollInterval)
metrics = metricsText(t, url+"_smallwebwaf/metrics", token)
}
for _, series := range []string{"failed", "suppressed", "dropped"} {
zero := "\nsmallwebwaf_alerts_" + series + "_total{destination=\"webhook\"} 0\n"
if !strings.Contains(metrics, zero) {
t.Errorf("no %q in the metrics:\n%s", zero, metrics)
}
}
})
}
func TestStateFileThatDoesNotParseStopsTheStart(t *testing.T) {
t.Parallel()
@@ -917,6 +1009,80 @@ func saveUntilAnswered(t *testing.T, path, content, url, from string, status int
}
}
// webhook is a stand-in for SWWAF_ALERT_WEBHOOK_URL. It notes each alert
// it is sent, and answers 204, or 503 while failing.
type webhook struct {
url string
failing atomic.Bool
mu sync.Mutex
posts []webhookPost
}
// webhookPost is an alert the webhook was sent, with the Authorization
// header sent with it, and whether the webhook took it.
type webhookPost struct {
alert map[string]any
authorization string
answered bool
}
// startWebhook starts a webhook that takes every alert.
func startWebhook(t *testing.T) *webhook {
t.Helper()
w := &webhook{}
server := httptest.NewServer(http.HandlerFunc(
func(rw http.ResponseWriter, r *http.Request) {
var alert map[string]any
_ = json.NewDecoder(r.Body).Decode(&alert)
failing := w.failing.Load()
w.mu.Lock()
w.posts = append(w.posts, webhookPost{
alert: alert, authorization: r.Header.Get("Authorization"),
answered: !failing,
})
w.mu.Unlock()
if failing {
rw.WriteHeader(http.StatusServiceUnavailable)
return
}
rw.WriteHeader(http.StatusNoContent)
}))
t.Cleanup(server.Close)
w.url = server.URL + "/alerts"
return w
}
// waitFor waits until the webhook has been sent an alert for event that
// it took, or, unless answered, failed, and returns it. It waits as long
// as that takes, so that a slow test process cannot fail the test.
func (w *webhook) waitFor(t *testing.T, event string, answered bool) webhookPost {
t.Helper()
for {
w.mu.Lock()
for _, post := range w.posts {
if post.alert["event"] == event && post.answered == answered {
w.mu.Unlock()
return post
}
}
w.mu.Unlock()
time.Sleep(pollInterval)
}
}
// statusFrom returns the status a request to url from the client at
// from, as X-Forwarded-For names it, is answered with.
func statusFrom(t *testing.T, url, from string) int {
+101 -17
View File
@@ -1,7 +1,8 @@
// Package state keeps smallwebwaf's state in JSON files in
// 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, and lookups.json GeoJS's answers. Load reads them at start,
// history, lookups.json GeoJS's answers, and alerts.json the cooldowns,
// the hour under way and the alerts waiting. 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 snapshot or to put in what
@@ -24,6 +25,7 @@ import (
"time"
"github.com/fsnotify/fsnotify"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/bans"
"sneak.berlin/go/smallwebwaf/internal/lookup"
"sneak.berlin/go/smallwebwaf/internal/metrics"
@@ -42,6 +44,7 @@ const (
bansJSON = "bans.json"
clientsJSON = "clients.json"
lookupsJSON = "lookups.json"
alertsJSON = "alerts.json"
)
var (
@@ -60,10 +63,13 @@ type Params struct {
// is (SWWAF_STATE_COUNTER_INTERVAL).
WriteDelay time.Duration
CounterInterval time.Duration
// Ledger, Limiter and GeoJS hold the state.
// 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
// Now tells the time by which the counters' buckets run out, normally
// time.Now in UTC.
Now func() time.Time
@@ -121,6 +127,14 @@ type lookupsFile struct {
Lookups []lookup.Answer `json:"lookups"`
}
// alertsFile is alerts.json, indented for an admin to read and edit.
type alertsFile struct {
Version int `json:"version"`
Cooldowns []alerts.Cooldown `json:"cooldowns"`
Hour alerts.Hour `json:"hour"`
Waiting []alerts.Alert `json:"waiting"`
}
// stateFile is the struct of a state file. Once the file is decoded, its
// check refuses the first entry without a field it needs, which would
// otherwise be read as something the entry does not say. data is the
@@ -147,23 +161,25 @@ func Load(params Params) (*Files, error) {
bansRead, bansErr := f.read(bansJSON)
clientsRead, clientsErr := f.read(clientsJSON)
lookupsRead, lookupsErr := f.read(lookupsJSON)
alertsRead, alertsErr := f.read(alertsJSON)
err = errors.Join(bansErr, clientsErr, lookupsErr)
err = errors.Join(bansErr, clientsErr, lookupsErr, alertsErr)
if err != nil {
return nil, err
}
params.ProcessLog.Info("read the state files", "directory", params.Dir,
"bans", bansRead, "clients", clientsRead, "lookups", lookupsRead)
"bans", bansRead, "clients", clientsRead, "lookups", lookupsRead,
"alerts_waiting", alertsRead)
return f, nil
}
// Run writes bans.json WriteDelay after a ban is made, with every ban
// made in between, and every file every CounterInterval, until ctx is
// done. A write that fails is logged, and the file is written again at
// its next write. Each write takes in an admin's edit of its file first,
// as writeFile describes.
// done. A write that fails is logged, raised as a file_error alert, and
// the file is written again at its next write. Each write takes in an
// admin's edit of its file first, as writeFile describes.
func (f *Files) Run(ctx context.Context) {
interval := time.NewTicker(f.params.CounterInterval)
defer interval.Stop()
@@ -192,7 +208,7 @@ func (f *Files) Run(ctx context.Context) {
// fails does not keep the others from being written.
func (f *Files) WriteAll() error {
return errors.Join(f.writeFile(bansJSON), f.writeFile(clientsJSON),
f.writeFile(lookupsJSON))
f.writeFile(lookupsJSON), f.writeFile(alertsJSON))
}
// Watch watches Dir until ctx is done, and takes in an admin's edit of a
@@ -227,7 +243,7 @@ func (f *Files) Watch(ctx context.Context) {
return
case event := <-watcher.Events:
switch name := filepath.Base(event.Name); name {
case bansJSON, clientsJSON, lookupsJSON:
case bansJSON, clientsJSON, lookupsJSON, alertsJSON:
f.fileChanged(name)
}
case err = <-watcher.Errors:
@@ -237,11 +253,20 @@ func (f *Files) Watch(ctx context.Context) {
}
}
// logFailure logs a write that failed.
// logFailure logs a write that failed, and raises a file_error alert for
// it.
func (f *Files) logFailure(err error) {
if err != nil {
f.params.ProcessLog.Error("writing the state files failed",
"error", err.Error())
const failed = "writing the state files failed"
// Raised before it is logged, so that the alert is there once the
// log line is.
f.params.Alerts.Raise(alerts.Alert{
Event: alerts.EventFileError,
Reason: failed,
Detail: map[string]any{"error": err.Error()},
})
f.params.ProcessLog.Error(failed, "error", err.Error())
}
}
@@ -362,6 +387,18 @@ func (f *Files) takeIn(name string, data []byte, edit bool) (int, error) {
f.params.GeoJS.Load(file.Lookups)
entries = len(file.Lookups)
case alertsJSON:
var file alertsFile
err := parse(path, data, &file)
if err != nil {
return 0, err
}
f.params.Alerts.Load(alerts.State{
Cooldowns: file.Cooldowns, Hour: file.Hour, Waiting: file.Waiting,
})
entries = len(file.Waiting)
}
f.sums[name] = sha256.Sum256(data)
@@ -413,8 +450,9 @@ func (f *Files) writeFile(name string) error {
// setAside renames the state file name, an edit that does not parse with
// parseErr, to name.bad, for the admin to mend, and logs it with where in
// the file the error is. If the rename fails, the edit is left as it is,
// and the error returned is parseErr joined with the rename's.
// the file the error is, and raises a file_error alert for it. If the
// rename fails, the edit is left as it is, and the error returned is
// parseErr joined with the rename's.
func (f *Files) setAside(name string, parseErr error) error {
path := filepath.Join(f.params.Dir, name)
@@ -423,8 +461,16 @@ func (f *Files) setAside(name string, parseErr error) error {
return errors.Join(parseErr, err)
}
f.params.ProcessLog.Error("set aside an edit of a state file that does not parse",
"file", path+".bad", "error", parseErr.Error())
const setAside = "set aside an edit of a state file that does not parse"
// Raised before it is logged, so that the alert is there once the log
// line is.
f.params.Alerts.Raise(alerts.Alert{
Event: alerts.EventFileError,
Reason: setAside,
Detail: map[string]any{"file": path + ".bad", "error": parseErr.Error()},
})
f.params.ProcessLog.Error(setAside, "file", path+".bad", "error", parseErr.Error())
f.params.Metrics.StateFileEditSetAside(name)
return nil
@@ -445,8 +491,21 @@ func (f *Files) encode(name string) ([]byte, error) {
return append(data, '\n'), nil
case clientsJSON:
return encodeOnePerLine("clients", f.params.Limiter.Snapshot())
default: // lookups.json
case lookupsJSON:
return encodeOnePerLine("lookups", f.params.GeoJS.Snapshot())
default: // alerts.json
held := f.params.Alerts.Snapshot()
file := alertsFile{
Version: version, Cooldowns: held.Cooldowns, Hour: held.Hour,
Waiting: held.Waiting,
}
data, err := json.MarshalIndent(file, "", " ")
if err != nil {
return nil, err
}
return append(data, '\n'), nil
}
}
@@ -581,6 +640,31 @@ func (f *lookupsFile) check(data []byte) error {
return nil
}
// check refuses a cooldown without its event or when its alert was sent,
// which would hold back no repeat, and an alert waiting without its event
// or its time.
func (f *alertsFile) check([]byte) error {
for i, cooldown := range f.Cooldowns {
switch {
case cooldown.Event == "":
return fmt.Errorf("cooldowns %w", missing(i, "event"))
case cooldown.Sent.IsZero():
return fmt.Errorf("cooldowns %w", missing(i, "sent"))
}
}
for i, alert := range f.Waiting {
switch {
case alert.Event == "":
return fmt.Errorf("waiting %w", missing(i, "event"))
case alert.Time.IsZero():
return fmt.Errorf("waiting %w", missing(i, "time"))
}
}
return nil
}
// countsWithoutStart reports whether b holds requests but no start, which
// places them in time.
func countsWithoutStart(b ratelimit.Buckets) bool {
+257 -18
View File
@@ -10,8 +10,10 @@ import (
"net/http"
"net/http/httptest"
"net/netip"
"net/url"
"os"
"path/filepath"
"reflect"
"slices"
"strconv"
"strings"
@@ -19,6 +21,7 @@ import (
"testing/synctest"
"time"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/bans"
"sneak.berlin/go/smallwebwaf/internal/lookup"
"sneak.berlin/go/smallwebwaf/internal/metrics"
@@ -31,6 +34,7 @@ const (
bansJSON = "bans.json"
clientsJSON = "clients.json"
lookupsJSON = "lookups.json"
alertsJSON = "alerts.json"
// What the process log says once Watch watches the directory, and as
// it takes in an edit.
watching = "watching the state files for edits"
@@ -85,6 +89,71 @@ const liftedBansJSON = `{"version": 1, "bans": [{"netblock": "203.0.113.9/32", `
`"start": "2026-10-06T00:00:00Z", "expires": "2026-10-06T01:00:00Z", ` +
`"cause": "limit", "lifted": "2026-10-06T00:10:00Z"}]}`
// filledAlertsJSON is alerts.json holding the alerts of fill.
const filledAlertsJSON = `{
"version": 1,
"cooldowns": [
{
"event": "file_error",
"netblock": "",
"sent": "2026-10-06T00:00:00Z",
"suppressed_repeats": 0
},
{
"event": "source_failure",
"netblock": "",
"sent": "2026-10-06T00:00:00Z",
"suppressed_repeats": 0
},
{
"event": "ban",
"netblock": "203.0.113.9/32",
"sent": "2026-10-06T00:00:00Z",
"suppressed_repeats": 1
}
],
"hour": {
"start": "2026-10-06T00:00:00Z",
"sent": 2,
"held_back": {
"source_failure": 1
}
},
"waiting": [
{
"instance": "fsn1app1/gitea",
"time": "2026-10-06T00:00:00Z",
"event": "ban",
"client": "203.0.113.9",
"netblock": "203.0.113.9/32",
"asn": "",
"as_name": "",
"country": "DE",
"reason": "requests per minute over the limit of 1",
"detail": {
"cause": "limit"
},
"suppressed_repeats": 0
},
{
"instance": "fsn1app1/gitea",
"time": "2026-10-06T00:00:00Z",
"event": "file_error",
"client": "",
"netblock": "",
"asn": "",
"as_name": "",
"country": "",
"reason": "writing the state files failed",
"detail": {
"error": "no space left on device"
},
"suppressed_repeats": 0
}
]
}
`
func TestFilesWrittenAndReadBack(t *testing.T) {
t.Parallel()
@@ -111,13 +180,38 @@ func TestFilesWrittenAndReadBack(t *testing.T) {
wantEqual(t, clientsJSON, after.Limiter.Snapshot(), before.Limiter.Snapshot())
wantEqual(t, lookupsJSON, after.GeoJS.Snapshot(), before.GeoJS.Snapshot())
if got, want := after.Alerts.Snapshot(), before.Alerts.Snapshot(); !reflect.DeepEqual(
got, want) {
t.Errorf("%s read back\n%+v\nwant\n%+v", alertsJSON, got, want)
}
// Each one-per-line file lists its entries by client, and nothing
// but the three files is left in the directory.
// but the four files is left in the directory.
wantEntries(t, filepath.Join(dir, clientsJSON), "clients",
"192.0.2.1/32", "203.0.113.9/32", "2001:db8::/64")
wantEntries(t, filepath.Join(dir, lookupsJSON), "lookups",
"192.0.2.1/32", "203.0.113.9/32")
wantFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
wantFiles(t, dir, alertsJSON, bansJSON, clientsJSON, lookupsJSON)
}
func TestAlertsJSONIsIndentedWithTheCooldownsTheHourAndTheAlertsWaiting(t *testing.T) {
t.Parallel()
dir := t.TempDir()
params := newParams(dir)
fill(params)
files := load(t, params)
err := files.WriteAll()
if err != nil {
t.Fatalf("write: %v", err)
}
got := readFile(t, filepath.Join(dir, alertsJSON))
if got != filledAlertsJSON {
t.Errorf("alerts.json\n%s\nwant\n%s", got, filledAlertsJSON)
}
}
func TestBansJSONIsIndentedWithNullForAPermanentBan(t *testing.T) {
@@ -146,8 +240,10 @@ func TestMissingFilesAreEmptyState(t *testing.T) {
params := newParams(t.TempDir())
load(t, params)
held := params.Alerts.Snapshot()
if len(params.Ledger.Snapshot()) != 0 || len(params.Limiter.Snapshot()) != 0 ||
len(params.GeoJS.Snapshot()) != 0 {
len(params.GeoJS.Snapshot()) != 0 || len(held.Cooldowns) != 0 ||
len(held.Waiting) != 0 || held.Hour.Sent != 0 {
t.Error("state from no files")
}
}
@@ -189,6 +285,11 @@ func TestFileThatDoesNotParseStopsTheStart(t *testing.T) {
`{"version": 1, "bans": [{"netblock": "203.0.113.300/32"}]}`,
`: netip.ParsePrefix("203.0.113.300/32")`,
},
{
"an unknown field of an alert waiting", alertsJSON,
`{"version": 1, "waiting": [{"event": "ban", "evnet": "ban"}]}`,
`: json: unknown field "evnet"`,
},
} {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
@@ -278,6 +379,43 @@ func TestEntryWithoutAFieldItNeedsStopsTheStart(t *testing.T) {
}
}
func TestAlertsJSONEntryWithoutAFieldItNeedsStopsTheStart(t *testing.T) {
t.Parallel()
for _, tc := range []struct {
name, content string
// want is what the error says after the file's path.
want string
}{
{
"a cooldown without its event",
`{"version": 1, "cooldowns": [{"sent": "2026-10-06T00:00:00Z"}]}`,
`: cooldowns entry 1 has no "event"`,
},
{
"a cooldown without when it was sent",
`{"version": 1, "cooldowns": [{"event": "ban"}]}`,
`: cooldowns entry 1 has no "sent"`,
},
{
"an alert waiting without its event",
`{"version": 1, "waiting": [{"time": "2026-10-06T00:00:00Z"}]}`,
`: waiting entry 1 has no "event"`,
},
{
"an alert waiting without its time",
`{"version": 1, "waiting": [{"event": "ban"}]}`,
`: waiting entry 1 has no "time"`,
},
} {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
wantRefused(t, alertsJSON, tc.content, tc.want)
})
}
}
func TestBanWithAnotherCauseStopsTheStart(t *testing.T) {
t.Parallel()
@@ -294,7 +432,7 @@ func TestBanWithAnotherCauseStopsTheStart(t *testing.T) {
func TestUnknownVersionStopsTheStart(t *testing.T) {
t.Parallel()
for _, file := range []string{bansJSON, clientsJSON, lookupsJSON} {
for _, file := range []string{bansJSON, clientsJSON, lookupsJSON, alertsJSON} {
for _, content := range []string{`{"version": 2}`, `{}`} {
t.Run(file+" "+content, func(t *testing.T) {
t.Parallel()
@@ -344,12 +482,12 @@ func TestBansWrittenOnceWriteDelayAfterABan(t *testing.T) {
// A second ban, made while the first waits to be written, puts the
// write off no further, and is written with it.
first := params.Ledger.BanForLimit(netip.MustParsePrefix("203.0.113.9/32"),
first, _ := params.Ledger.BanForLimit(netip.MustParsePrefix("203.0.113.9/32"),
midnight(), bans.Notes{})
time.Sleep(5 * time.Second)
second := params.Ledger.BanForLimit(netip.MustParsePrefix("203.0.113.10/32"),
second, _ := params.Ledger.BanForLimit(netip.MustParsePrefix("203.0.113.10/32"),
midnight(), bans.Notes{})
time.Sleep(5*time.Second - time.Nanosecond)
@@ -396,8 +534,8 @@ func TestEveryFileWrittenEveryCounterInterval(t *testing.T) {
time.Sleep(time.Nanosecond)
synctest.Wait()
wantFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
removeFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
wantFiles(t, dir, alertsJSON, bansJSON, clientsJSON, lookupsJSON)
removeFiles(t, dir, alertsJSON, bansJSON, clientsJSON, lookupsJSON)
}
})
}
@@ -440,6 +578,54 @@ func TestEditJustBeforeAScheduledWriteSurvivesIt(t *testing.T) {
})
}
func TestWriteThatFailsWhileRunningRaisesAFileErrorAlertOncePerCooldown(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
dir := t.TempDir()
params := newParams(dir)
params.CounterInterval = time.Minute
run(t, load(t, params).Run)
// A directory in the way of bans.json's temporary file fails each of
// its writes. It holds a file, so that the write cannot remove it.
err := os.Mkdir(filepath.Join(dir, bansJSON+".tmp"), 0o700)
if err == nil {
err = os.WriteFile(filepath.Join(dir, bansJSON+".tmp", "kept"), nil, 0o600)
}
if err != nil {
t.Fatalf("put a directory in the way: %v", err)
}
time.Sleep(time.Minute)
synctest.Wait()
waiting := params.Alerts.Snapshot().Waiting
if len(waiting) != 1 {
t.Fatalf("%d alerts wait, want 1", len(waiting))
}
message, _ := waiting[0].Detail["error"].(string)
if waiting[0].Event != alerts.EventFileError ||
waiting[0].Reason != "writing the state files failed" ||
!strings.Contains(message, bansJSON+".tmp") {
t.Fatalf("alerts waiting %+v, want a file_error alert naming bans.json's "+
"temporary file", waiting)
}
// The next write fails too, within the cooldown, which holds it back.
time.Sleep(time.Minute)
synctest.Wait()
if len(params.Alerts.Snapshot().Waiting) != 1 || params.Alerts.Suppressed() != 1 {
t.Errorf("%d alerts wait and %d are held back, want 1 and 1",
len(params.Alerts.Snapshot().Waiting), params.Alerts.Suppressed())
}
})
}
func TestFailedWriteLeavesTheFileAsItWas(t *testing.T) {
t.Parallel()
@@ -565,7 +751,7 @@ func TestFileThatCannotBeReadIsNotWrittenOver(t *testing.T) {
t.Errorf("bans.json is now %v (%v), want the socket", info, err)
}
wantFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
wantFiles(t, dir, alertsJSON, bansJSON, clientsJSON, lookupsJSON)
wantWriteFailed(t, params, bansJSON)
}
@@ -637,6 +823,25 @@ func TestEditOfEachFileTakenIn(t *testing.T) {
wantTakenIn(t, lines, dir, lookupsJSON)
wantEqual(t, lookupsJSON, params.GeoJS.Snapshot(),
[]lookup.Answer{{Client: client, Country: "FR", Answered: midnight()}})
// A netblock with bits past its length is read as the netblock it is
// in.
edit(t, dir, alertsJSON, `{"version": 1, "cooldowns": [{"event": "ban", `+
`"netblock": "198.51.100.9/24", "sent": "2026-10-06T00:00:00Z"}], `+
`"waiting": [{"event": "file_error", "time": "2026-10-06T00:00:00Z"}]}`)
wantTakenIn(t, lines, dir, alertsJSON)
want := alerts.State{
Cooldowns: []alerts.Cooldown{{
Event: alerts.EventBan, Netblock: netip.MustParsePrefix("198.51.100.0/24"),
Sent: midnight(),
}},
Hour: alerts.Hour{HeldBack: map[string]int{}},
Waiting: []alerts.Alert{{Event: alerts.EventFileError, Time: midnight()}},
}
if got := params.Alerts.Snapshot(); !reflect.DeepEqual(got, want) {
t.Errorf("%s taken in as\n%+v\nwant\n%+v", alertsJSON, got, want)
}
}
func TestOwnWritesAreNotTakenIn(t *testing.T) {
@@ -708,7 +913,7 @@ func TestBanAddedAndLiftedThroughBansJSON(t *testing.T) {
`"start": "2026-10-06T00:00:00Z", "expires": null}]}`)
wantTakenIn(t, lines, dir, bansJSON)
_, banned := params.Ledger.Check(client, midnight())
_, banned, _ := params.Ledger.Check(client, midnight())
if !banned {
t.Error("the ban added to bans.json does not refuse")
}
@@ -717,7 +922,7 @@ func TestBanAddedAndLiftedThroughBansJSON(t *testing.T) {
edit(t, dir, bansJSON, `{"version": 1, "bans": []}`)
wantTakenIn(t, lines, dir, bansJSON)
_, banned = params.Ledger.Check(client, midnight())
_, banned, _ = params.Ledger.Check(client, midnight())
if banned {
t.Error("the ban removed from bans.json still refuses")
}
@@ -844,7 +1049,7 @@ func TestBrokenEditSetAsideAtTheNextWrite(t *testing.T) {
edit(t, dir, bansJSON, broken)
edit(t, dir, clientsJSON, `{"version": 1, "clients": []}`)
wantTakenIn(t, lines, dir, clientsJSON)
wantFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
wantFiles(t, dir, alertsJSON, bansJSON, clientsJSON, lookupsJSON)
// The next write sets it aside, logged with where the error is, and
// writes bans.json again from what smallwebwaf still holds.
@@ -861,7 +1066,14 @@ func TestBrokenEditSetAsideAtTheNextWrite(t *testing.T) {
t.Errorf("set aside with %v", line)
}
wantFiles(t, dir, bansJSON, bansJSON+".bad", clientsJSON, lookupsJSON)
// It is raised as a file_error alert, with the same file and error.
waiting := params.Alerts.Snapshot().Waiting
if len(waiting) != 1 || waiting[0].Event != alerts.EventFileError ||
waiting[0].Detail["file"] != path+".bad" || waiting[0].Detail["error"] != message {
t.Errorf("alerts waiting %+v, want a file_error alert for %s", waiting, path+".bad")
}
wantFiles(t, dir, alertsJSON, bansJSON, bansJSON+".bad", clientsJSON, lookupsJSON)
if got := readFile(t, path+".bad"); got != broken {
t.Errorf("bans.json.bad holds\n%s\nwant the edit", got)
@@ -990,7 +1202,8 @@ func midnight() time.Time {
}
// newParams returns Params for the state files in dir, with parts that
// hold nothing yet. GeoJS is never asked.
// hold nothing yet. GeoJS is never asked, and the alerts, at most two an
// hour, are never sent.
func newParams(dir string) state.Params {
discard := slog.New(slog.DiscardHandler)
m := metrics.New(1)
@@ -1010,6 +1223,14 @@ 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,
}),
Now: midnight,
ProcessLog: discard,
Metrics: m,
@@ -1017,8 +1238,9 @@ func newParams(dir string) state.Params {
}
// 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, and
// GeoJS answers into the parts of params.
// 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.
func fill(params state.Params) {
now := midnight()
client := netip.MustParsePrefix("203.0.113.9/32")
@@ -1043,6 +1265,23 @@ func fill(params state.Params) {
Answered: now.Add(-time.Hour), Used: now.Add(-time.Minute),
},
})
// An alert waiting, a repeat of it the cooldown holds back, another
// alert waiting, and one past the two an hour, for the hour's summary.
ban := alerts.Alert{
Event: alerts.EventBan, Client: client.Addr(), Netblock: client, Country: "DE",
Reason: "requests per minute over the limit of 1",
Detail: map[string]any{"cause": "limit"},
}
params.Alerts.Raise(ban)
params.Alerts.Raise(ban)
params.Alerts.Raise(alerts.Alert{
Event: alerts.EventFileError, Reason: "writing the state files failed",
Detail: map[string]any{"error": "no space left on device"},
})
params.Alerts.Raise(alerts.Alert{
Event: alerts.EventSourceFailure, Reason: "asking GeoJS failed",
})
}
// permanentBan is the ban permanentBansJSON holds.
@@ -1098,13 +1337,13 @@ func wantLiftedBanKept(
netblock := netip.MustParsePrefix(liftedClient + "/32")
_, banned := ledger.Check(netblock.Addr(), afterLifting())
_, banned, _ := ledger.Check(netblock.Addr(), afterLifting())
if banned {
t.Error("the lifted ban refuses")
}
// Were the lifted ban counted, the next would last three hours.
ban := ledger.BanForLimit(netblock, afterLifting(), bans.Notes{})
ban, _ := ledger.BanForLimit(netblock, afterLifting(), bans.Notes{})
if ban.Expires.Sub(ban.Start) != time.Hour {
t.Errorf("the next ban lasts %s, want 1h", ban.Expires.Sub(ban.Start))
}