smallwebwaf/internal/reputation/crowdsec_test.go

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)
	}
}