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, as CrowdSec names it. ban = "ban" ) 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)}, {"Range", "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 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, 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", }, { "an answer that does not read", func(e *engine) { e.answer = "" }, "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 the engine's 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, log string, lists *reputation.Lists, queue *alerts.Queue, ) { t.Helper() 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{ {"Range", "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, // and 503 while failing. It counts the fetches. type engine struct { mu sync.Mutex key string decisions []decision answer string failing bool fetches int } // 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++ status, body := http.StatusOK, 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.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: http.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) } }