CrowdSec decision list fetched, kept, and its clients banned until the decision ends (closes #106)

SWWAF_CROWDSEC_LAPI_URL and SWWAF_CROWDSEC_LAPI_KEY name an engine whose
decision list, <url>/v1/decisions, is fetched every minute with the key in
X-Api-Key, following no redirect, and kept as a blocklist is: used while a
fetch fails, and across restarts through reputation.json. Ban decisions on an
Ip or a Range end at the fetch time plus their duration. A listed client's
request is refused and bans its netblock with the cause crowdsec until the
decision ends; bans.json, ban notes and metrics take the cause.

Judgement call: fetched every minute, not a setting.
Judgement call: a crowdsec ban never lengthens a limit ban.
Judgement call: a lifted crowdsec ban is remade while its decision lasts.

Model: opus-5-5

internal/reputation/crowdsec_test.go

@@ -0,0 +1,513 @@
+package reputation_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/netip"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/reputation"
+)
+
+// The tests run in a synctest bubble, as those of the blocklists do, and
+// fetch the decision list from engine, a stand-in for a CrowdSec engine
+// that answers without the network.
+
+const (
+	// decisionsURL is the decision list of the tests' engine, and engineKey
+	// the key it answers.
+	decisionsURL = "http://crowdsec.example:8080/v1/decisions"
+	engineKey    = "crowdsec-key-0123456789abcdef"
+	// sshBF and probing are scenarios of the engine's decisions.
+	sshBF   = "crowdsecurity/ssh-bf"
+	probing = "crowdsecurity/http-probing"
+	// ban is the type of a decision to ban, and rangeScope the scope of a
+	// decision on a netblock, as CrowdSec names them.
+	ban        = "ban"
+	rangeScope = "Range"
+)
+
+func TestCrowdSecDecisionBansItsNetblockUntilItEndsEvenWithTheEngineDown(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		began := time.Now()
+		manual := "manual 'ban' from 'localhost'"
+		e := &engine{key: engineKey, decisions: []decision{
+			{"Ip", suspect, ban, manual, began.Add(6 * time.Hour)},
+			// A shorter decision on the same address, which is not the one
+			// used.
+			{"Ip", suspect, ban, sshBF, began.Add(4 * time.Hour)},
+			{rangeScope, "198.51.100.0/24", ban, probing, began.Add(time.Hour)},
+			{"Ip", "2001:db8::1", ban, sshBF, began.Add(2 * time.Hour)},
+			// Left out: a decision to show a captcha, and one on a country.
+			{"Ip", "192.0.2.50", "captcha", probing, began.Add(time.Hour)},
+			{"Country", "KP", ban, manual, began.Add(time.Hour)},
+		}}
+		lists := start(t, e, crowdSecParams())
+
+		for addr, want := range map[string]reputation.Decision{
+			suspect:          {Expires: began.Add(6 * time.Hour), Scenario: manual},
+			"198.51.100.0":   {Expires: began.Add(time.Hour), Scenario: probing},
+			"198.51.100.255": {Expires: began.Add(time.Hour), Scenario: probing},
+			"2001:db8::1":    {Expires: began.Add(2 * time.Hour), Scenario: sshBF},
+			"203.0.113.10":   {},
+			"198.51.101.0":   {},
+			"2001:db8::2":    {},
+			"192.0.2.50":     {},
+		} {
+			wantDecision(t, lists, addr, want)
+		}
+
+		// With the engine down, the copy kept still holds the decision on
+		// 198.51.100.0/24, which no longer bans once it has ended.
+		e.set(func(e *engine) { e.failing = true })
+		time.Sleep(time.Hour - time.Nanosecond)
+		synctest.Wait()
+		wantDecision(t, lists, "198.51.100.7",
+			reputation.Decision{Expires: began.Add(time.Hour), Scenario: probing})
+
+		time.Sleep(time.Nanosecond)
+		synctest.Wait()
+		wantDecision(t, lists, "198.51.100.7", reputation.Decision{})
+		wantDecision(t, lists, suspect,
+			reputation.Decision{Expires: began.Add(6 * time.Hour), Scenario: manual})
+	})
+}
+
+func TestCrowdSecDecisionListFetchedAgainEveryMinute(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		began := time.Now()
+		e := &engine{key: engineKey, decisions: []decision{
+			{"Ip", suspect, ban, sshBF, began.Add(4 * time.Hour)},
+		}}
+		lists := start(t, e, crowdSecParams())
+		wantEngineFetches(t, e, 1)
+
+		added := reputation.Decision{Expires: began.Add(2 * time.Hour), Scenario: probing}
+
+		e.set(func(e *engine) {
+			e.decisions = append(e.decisions,
+				decision{"Ip", "203.0.113.10", ban, probing, added.Expires})
+		})
+
+		time.Sleep(time.Minute - time.Nanosecond)
+		wantEngineFetches(t, e, 1)
+		wantDecision(t, lists, "203.0.113.10", reputation.Decision{})
+
+		time.Sleep(time.Nanosecond)
+		wantEngineFetches(t, e, 2)
+		wantDecision(t, lists, "203.0.113.10", added)
+	})
+}
+
+func TestCrowdSecDecisionOnAClientIsTheOneThatEndsLast(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		began := time.Now()
+		e := &engine{key: engineKey, decisions: []decision{
+			// Two decisions on one address, the shorter listed first.
+			{"Ip", suspect, ban, sshBF, began.Add(2 * time.Hour)},
+			{"Ip", suspect, ban, probing, began.Add(4 * time.Hour)},
+			// 198.51.100.130 is held by a decision on its address that ends
+			// after the one on its netblock, and 192.0.2.20 by one that ends
+			// before.
+			{rangeScope, "198.51.100.128/25", ban, sshBF, began.Add(time.Hour)},
+			{"Ip", "198.51.100.130", ban, probing, began.Add(3 * time.Hour)},
+			{rangeScope, "192.0.2.0/24", ban, probing, began.Add(5 * time.Hour)},
+			{"Ip", "192.0.2.20", ban, sshBF, began.Add(2 * time.Hour)},
+		}}
+		lists := start(t, e, crowdSecParams())
+
+		wantDecision(t, lists, suspect,
+			reputation.Decision{Expires: began.Add(4 * time.Hour), Scenario: probing})
+		wantDecision(t, lists, "198.51.100.130",
+			reputation.Decision{Expires: began.Add(3 * time.Hour), Scenario: probing})
+		wantDecision(t, lists, "192.0.2.20",
+			reputation.Decision{Expires: began.Add(5 * time.Hour), Scenario: probing})
+	})
+}
+
+func TestCrowdSecAnswerOfNoDecisionIsAGoodCopyThatListsNoClient(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		began := time.Now()
+		e := &engine{key: engineKey, decisions: []decision{
+			{"Ip", suspect, ban, sshBF, began.Add(4 * time.Hour)},
+		}}
+		lists := start(t, e, crowdSecParams())
+
+		// With its decision deleted, the engine answers null.
+		e.set(func(e *engine) { e.decisions = nil })
+		time.Sleep(time.Minute)
+		wantEngineFetches(t, e, 2)
+
+		want := []reputation.List{{
+			URL: decisionsURL, Tried: time.Now(), Fetched: time.Now(), Lines: []string{"null"},
+		}}
+		if got := lists.Snapshot(); !reflect.DeepEqual(got, want) {
+			t.Errorf("lists %+v, want %+v", got, want)
+		}
+
+		if lists.Failures(decisionsURL) != 0 {
+			t.Errorf("%d failures, want 0", lists.Failures(decisionsURL))
+		}
+
+		wantDecision(t, lists, suspect, reputation.Decision{})
+	})
+}
+
+func TestCrowdSecFailureKeepsTheLastGoodCopyAlertsOncePerCooldownAndHidesTheKey(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	for _, tc := range crowdSecFailures() {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			synctest.Test(t, func(t *testing.T) {
+				var log bytes.Buffer
+
+				began := time.Now()
+				e := &engine{key: engineKey, decisions: []decision{
+					{"Ip", suspect, ban, sshBF, began.Add(4 * time.Hour)},
+				}}
+				queue := newQueue()
+				p := crowdSecParams()
+				p.ProcessLog = slog.New(slog.NewJSONHandler(&log, nil))
+				p.Alerts = queue
+				lists := start(t, e, p)
+				kept := lists.Snapshot()
+
+				e.set(tc.fail)
+
+				// Each failure is tried again a minute after it.
+				for range 2 {
+					time.Sleep(time.Minute)
+					synctest.Wait()
+				}
+
+				wantEngineFetches(t, e, 3)
+				wantDecision(t, lists, suspect,
+					reputation.Decision{Expires: began.Add(4 * time.Hour), Scenario: sshBF})
+
+				want := kept[0]
+				want.Tried = time.Now()
+
+				if got := lists.Snapshot(); !reflect.DeepEqual(got, []reputation.List{want}) {
+					t.Errorf("lists %+v, want the first copy, last tried now, %+v", got, want)
+				}
+
+				if lists.Failures(decisionsURL) != 2 {
+					t.Errorf("%d failures, want 2", lists.Failures(decisionsURL))
+				}
+
+				// One alert for the first failure; the cooldown holds back the
+				// second.
+				wantAlert(t, queue,
+					fetchFailure(time.Now().Add(-time.Minute), decisionsURL, tc.error))
+
+				if !strings.Contains(log.String(), `"msg":"fetching a list failed",`+
+					`"url":"`+decisionsURL+`","error":"`+tc.error) {
+					t.Errorf("logged\n%s\nwant the failures", log.String())
+				}
+
+				wantKeyNotShown(t, e, log.String(), lists, queue)
+			})
+		})
+	}
+}
+
+// crowdSecFailure is a way for the engine to fail: fail has it answer the
+// fetches after the first so that they fail with error.
+type crowdSecFailure struct {
+	name  string
+	fail  func(e *engine)
+	error string
+}
+
+// crowdSecFailures returns the ways the engine can fail.
+func crowdSecFailures() []crowdSecFailure {
+	const notDecision = " does not give an address or a netblock and a duration, " +
+		"such as 4h0m0s"
+
+	return []crowdSecFailure{
+		{
+			"an answer other than 200",
+			func(e *engine) { e.failing = true },
+			"the server answered 503 Service Unavailable",
+		},
+		{
+			"a key the engine refuses",
+			func(e *engine) { e.key = "another-key-0123456789abcdef" },
+			"the server answered 403 Forbidden",
+		},
+		{
+			"a redirect",
+			func(e *engine) { e.redirect = "http://elsewhere.example/v1/decisions" },
+			"the server answered 302 Found",
+		},
+		{
+			"an answer that does not read",
+			func(e *engine) { e.answer = "<html>" },
+			"read the answer: invalid character '<' looking for beginning of value",
+		},
+		{
+			"a decision to ban whose value does not read",
+			func(e *engine) {
+				e.answer = `[{"duration": "4h", "scenario": "` + sshBF + `", ` +
+					`"scope": "Ip", "type": "ban", "value": "203.0.113.300"}]`
+			},
+			"decision 1" + notDecision,
+		},
+		{
+			"a decision to ban whose duration does not read",
+			func(e *engine) {
+				e.answer = `[{"duration": "4h", "scope": "Country", "type": "ban", ` +
+					`"value": "KP"}, {"duration": "four hours", "scope": "Range", ` +
+					`"type": "ban", "value": "198.51.100.0/24"}]`
+			},
+			"decision 2" + notDecision,
+		},
+	}
+}
+
+// wantKeyNotShown checks that no fetch carried the engine's key to a URL
+// other than its decision list, such as the one a redirect names, and that
+// the key is in none of what the fetches leave behind: log, the process
+// log, the alerts waiting in queue, and the copies of lists, which
+// reputation.json keeps.
+func wantKeyNotShown(
+	t *testing.T, e *engine, log string, lists *reputation.Lists, queue *alerts.Queue,
+) {
+	t.Helper()
+
+	e.mu.Lock()
+	keySentTo := e.keySentTo
+	e.mu.Unlock()
+
+	if len(keySentTo) != 0 {
+		t.Errorf("the key was sent to %v", keySentTo)
+	}
+
+	shown, err := json.Marshal([]any{lists.Snapshot(), waiting(queue)})
+	if err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+
+	if strings.Contains(log+string(shown), engineKey) {
+		t.Errorf("the key is shown in\n%s\n%s", log, shown)
+	}
+}
+
+func TestCrowdSecDecisionListKeptAcrossARestartEndsWhenItsDecisionsDo(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		began := time.Now()
+		e := &engine{key: engineKey, decisions: []decision{
+			{rangeScope, "198.51.100.0/24", ban, probing, began.Add(time.Hour)},
+		}}
+		lists := start(t, e, crowdSecParams())
+		kept := lists.Snapshot()
+
+		// Restarted half an hour later with what reputation.json keeps, and
+		// the engine down, the decision still bans, until the end it had at
+		// the fetch, half an hour on.
+		time.Sleep(30 * time.Minute)
+
+		down := &engine{key: engineKey, failing: true}
+		again := reputation.New(crowdSecParams())
+		again.SetTransport(down)
+
+		err := again.Load(kept)
+		if err != nil {
+			t.Fatalf("load: %v", err)
+		}
+
+		run(t, again)
+
+		want := reputation.Decision{Expires: began.Add(time.Hour), Scenario: probing}
+		wantDecision(t, again, "198.51.100.7", want)
+
+		time.Sleep(30*time.Minute - time.Nanosecond)
+		synctest.Wait()
+		wantDecision(t, again, "198.51.100.7", want)
+
+		time.Sleep(time.Nanosecond)
+		synctest.Wait()
+		wantDecision(t, again, "198.51.100.7", reputation.Decision{})
+	})
+}
+
+func TestLoadTakesACrowdSecListNeverFetchedAndRefusesACopyThatDoesNotRead(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	now := time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)
+	lists := reputation.New(crowdSecParams())
+
+	// Tried, and never fetched: there is no copy to read.
+	err := lists.Load([]reputation.List{{URL: decisionsURL, Tried: now}})
+	if err != nil {
+		t.Errorf("load the list never fetched: %v", err)
+	}
+
+	err = lists.Load([]reputation.List{{
+		URL: decisionsURL, Tried: now, Fetched: now, Lines: []string{
+			`[{"duration": "4h", "scope": "Range", "type": "ban", ` +
+				`"value": "198.51.100.0/33"}]`,
+		},
+	}})
+
+	const want = "the copy of " + decisionsURL + ": decision 1 does not give an " +
+		"address or a netblock and a duration, such as 4h0m0s"
+	if err == nil || err.Error() != want {
+		t.Errorf("error %v, want %s", err, want)
+	}
+}
+
+// engine is a stand-in for the local API of a CrowdSec engine. It answers
+// a fetch of the decision list that carries its key in X-Api-Key with its
+// decisions still in force, each with the time it has left as it answers,
+// by the bubble's clock, as an engine does, or with answer while that is
+// not "". It answers 403 to a fetch without its key, as an engine does,
+// with a redirect to redirect while that is not "", and 503 while failing.
+// It counts the fetches, and notes in keySentTo the URL of each fetch of
+// another URL that carries a key, as one following a redirect would.
+type engine struct {
+	mu        sync.Mutex
+	key       string
+	decisions []decision
+	answer    string
+	redirect  string
+	failing   bool
+	fetches   int
+	keySentTo []string
+}
+
+// decision is a decision of the engine, which ends at expires.
+type decision struct {
+	scope, value, kind, scenario string
+	expires                      time.Time
+}
+
+// RoundTrip has the engine answer req, in place of the network.
+func (e *engine) RoundTrip(req *http.Request) (*http.Response, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	e.fetches++
+
+	if req.URL.String() != decisionsURL && req.Header.Get("X-Api-Key") != "" {
+		e.keySentTo = append(e.keySentTo, req.URL.String())
+	}
+
+	status, header, body := http.StatusOK, http.Header{}, e.answer
+
+	switch {
+	case req.URL.String() != decisionsURL || req.Header.Get("X-Api-Key") != e.key:
+		status, body = http.StatusForbidden, `{"message":"access forbidden"}`
+	case e.redirect != "":
+		status, header = http.StatusFound, http.Header{"Location": {e.redirect}}
+	case e.failing:
+		status, body = http.StatusServiceUnavailable, ""
+	case body == "":
+		body = e.inForce(time.Now())
+	}
+
+	return &http.Response{
+		StatusCode: status,
+		Status:     fmt.Sprintf("%d %s", status, http.StatusText(status)),
+		Header:     header,
+		Body:       io.NopCloser(strings.NewReader(body)),
+		Request:    req,
+	}, nil
+}
+
+// inForce returns the decisions in force at now, as the engine answers
+// them: a JSON list, null for none.
+func (e *engine) inForce(now time.Time) string {
+	var answer []map[string]string
+
+	for _, d := range e.decisions {
+		if now.Before(d.expires) {
+			answer = append(answer, map[string]string{
+				"duration": d.expires.Sub(now).String(), "origin": "crowdsec",
+				"scenario": d.scenario, "scope": d.scope, "type": d.kind, "value": d.value,
+			})
+		}
+	}
+
+	body, err := json.Marshal(answer)
+	if err != nil {
+		panic(err) // a list of maps of strings always encodes
+	}
+
+	return string(body)
+}
+
+// set changes the engine with change.
+func (e *engine) set(change func(e *engine)) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	change(e)
+}
+
+// crowdSecParams returns the Params of the decision list of the tests'
+// engine, fetched with its key, by the bubble's clock, with alerts to a
+// queue that sends none.
+func crowdSecParams() reputation.Params {
+	p := params()
+	p.CrowdSecDecisionsURL = decisionsURL
+	p.CrowdSecKey = engineKey
+
+	return p
+}
+
+// wantEngineFetches waits until Run has made the fetches due, and checks
+// how many the engine has had.
+func wantEngineFetches(t *testing.T, e *engine, want int) {
+	t.Helper()
+
+	synctest.Wait()
+
+	e.mu.Lock()
+	got := e.fetches
+	e.mu.Unlock()
+
+	if got != want {
+		t.Errorf("%d fetches, want %d", got, want)
+	}
+}
+
+// wantDecision checks the decision lists says is in force on addr now,
+// the zero Decision for none.
+func wantDecision(
+	t *testing.T, lists *reputation.Lists, addr string, want reputation.Decision,
+) {
+	t.Helper()
+
+	got, listed := lists.CrowdSecDecision(netip.MustParseAddr(addr), time.Now())
+	if listed != !want.Expires.IsZero() ||
+		listed && (!got.Expires.Equal(want.Expires) || got.Scenario != want.Scenario) {
+		t.Errorf("%s has the decision %+v in force %t, want %+v", addr, got, listed, want)
+	}
+}