package reputation_test import ( "bytes" "context" "fmt" "io" "log/slog" "net/http" "net/netip" "net/url" "reflect" "slices" "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, where the time package runs on a // clock of the test's own: time.Sleep moves it on at once, and // synctest.Wait returns once Run waits for the next list to be due, so // that every fetch due by then has been made. The stand-in for the // servers the lists are fetched from answers without the network, since a // fetch waiting on the network would keep that clock from moving on. const ( // dropURL and torURL are the blocklists, and asnURL the file of // AS:percent lines. dropURL = "https://lists.example/drop.txt" torURL = "https://lists.example/tor.txt" asnURL = "https://lists.example/asn.txt" // refresh is the tests' SWWAF_BLOCKLIST_REFRESH, and cooldown their // SWWAF_ALERT_COOLDOWN, longer than it. refresh = 24 * time.Hour cooldown = 48 * time.Hour // drop is a blocklist as the Spamhaus DROP list is written, with an // address and a netblock in each of its comments, which list nothing. drop = "; Spamhaus DROP List 2026/10/07 - (c) 2026 The Spamhaus Project SLL\n" + "; Last-Modified: Wed, 07 Oct 2026 00:00:00 GMT ; 192.0.2.1\n" + "# 198.51.100.0/24\n" + "\n" + "203.0.113.0/24 ; SBL1\n" + " 192.0.2.9 # one address\n" + "2001:db8:1::/48 ; SBL2\n" ) func TestListedAddressesAndNetblocksWithTheCommentsLeftOut(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { servers := &standIn{lists: map[string]string{dropURL: drop}} lists := start(t, servers, params(dropURL)) for addr, want := range map[string][]string{ "203.0.113.0": {dropURL}, "203.0.113.255": {dropURL}, "192.0.2.9": {dropURL}, "2001:db8:1::7": {dropURL}, "203.0.114.0": nil, "192.0.2.8": nil, "192.0.2.1": nil, "198.51.100.7": nil, "2001:db8:2::7": nil, } { wantListedBy(t, lists, addr, want...) } }) } func TestIPv4MappedLineListsTheIPv4AddressOrNetblockItStandsFor(t *testing.T) { t.Parallel() now := time.Now() lists := reputation.New(params(dropURL)) err := lists.Load([]reputation.List{{ URL: dropURL, Tried: now, Fetched: now, Lines: []string{"::ffff:192.0.2.9", "::ffff:203.0.113.0/120"}, }}) if err != nil { t.Fatalf("load: %v", err) } for addr, want := range map[string][]string{ "192.0.2.9": {dropURL}, "203.0.113.255": {dropURL}, "192.0.2.8": nil, "203.0.114.0": nil, } { wantListedBy(t, lists, addr, want...) } // A mapped netblock shorter than /96 stands for no IPv4 one. err = lists.Load([]reputation.List{{ URL: dropURL, Tried: now, Fetched: now, Lines: []string{"::ffff:198.51.100.0/88"}, }}) const want = "the copy of " + dropURL + ": line 1 is not an address or a netblock, such as 192.0.2.0/24" if err == nil || err.Error() != want { t.Errorf("error %v, want %s", err, want) } } func TestClientIsListedByEachBlocklistThatListsIt(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { servers := &standIn{lists: map[string]string{ dropURL: "203.0.113.0/24\n", torURL: "203.0.113.9\n", }} lists := start(t, servers, params(torURL, dropURL)) // In the order SWWAF_BLOCKLIST_URLS names them. wantListedBy(t, lists, "203.0.113.9", torURL, dropURL) wantListedBy(t, lists, "203.0.113.8", dropURL) }) } func TestListFetchedAgainOnceRefreshHasPassed(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { servers := &standIn{lists: map[string]string{dropURL: "203.0.113.9\n"}} lists := start(t, servers, params(dropURL)) began := time.Now() wantFetches(t, servers, 1) servers.set(dropURL, "203.0.113.10\n") time.Sleep(refresh - time.Nanosecond) wantFetches(t, servers, 1) wantListedBy(t, lists, "203.0.113.9", dropURL) time.Sleep(time.Nanosecond) wantFetches(t, servers, 2) wantListedBy(t, lists, "203.0.113.9") wantListedBy(t, lists, "203.0.113.10", dropURL) if fetched := lists.Fetched(dropURL); !fetched.Equal(began.Add(refresh)) { t.Errorf("the copy in use was fetched at %s, want %s", fetched, began.Add(refresh)) } }) } func TestFailedFetchKeepsTheLastGoodCopyAndAlertsOncePerCooldown(t *testing.T) { t.Parallel() for _, tc := range []struct { name string // fail has the stand-in answer the fetches after the first so that // they fail with error. fail func(servers *standIn) error string }{ { "an answer other than 200", func(servers *standIn) { servers.set(dropURL, "") }, "the server answered 503 Service Unavailable", }, { "a line that does not read", func(servers *standIn) { servers.set(dropURL, "203.0.113.10\n\n") }, "line 2 is not an address or a netblock, such as 192.0.2.0/24", }, { "a list longer than 16 MiB", func(servers *standIn) { servers.set(dropURL, strings.Repeat("#\n", 8<<20+1)) }, "the list is longer than 16 MiB", }, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { var log bytes.Buffer servers := &standIn{lists: map[string]string{dropURL: drop}} queue := newQueue() p := params(dropURL) p.ProcessLog = slog.New(slog.NewJSONHandler(&log, nil)) p.Alerts = queue lists := start(t, servers, p) kept := lists.Snapshot() tc.fail(servers) // Each failure is tried again once refresh has passed since it. for range 2 { time.Sleep(refresh) synctest.Wait() } wantFetches(t, servers, 3) wantListedBy(t, lists, "203.0.113.9", dropURL) 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(dropURL) != 2 { t.Errorf("%d failures, want 2", lists.Failures(dropURL)) } // One alert for the first failure; the cooldown holds back the // second. wantAlert(t, queue, fetchFailure(time.Now().Add(-refresh), dropURL, tc.error)) if !strings.Contains(log.String(), `"msg":"fetching a list failed",`+ `"url":"`+dropURL+`","error":"`+tc.error) { t.Errorf("logged\n%s\nwant the failures", log.String()) } }) }) } } func TestFetchNotDoneWithinAMinuteFails(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { servers := &standIn{lists: map[string]string{}, hanging: true} lists := start(t, servers, params(dropURL)) time.Sleep(time.Minute - time.Nanosecond) synctest.Wait() if lists.Failures(dropURL) != 0 { t.Errorf("%d failures before a minute, want none", lists.Failures(dropURL)) } time.Sleep(time.Nanosecond) synctest.Wait() if lists.Failures(dropURL) != 1 { t.Errorf("%d failures after a minute, want 1", lists.Failures(dropURL)) } }) } func TestKeptCopyIsFetchedAgainOnceRefreshHasPassedSinceItWasFetched(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { servers := &standIn{lists: map[string]string{ dropURL: "203.0.113.10\n", torURL: "198.51.100.10\n", }} lists := reputation.New(params(dropURL, torURL)) lists.SetTransport(servers) // drop.txt was fetched an hour ago, and tor.txt a refresh ago, as // reputation.json says at start. err := lists.Load([]reputation.List{ {URL: dropURL, Fetched: time.Now().Add(-time.Hour), Lines: []string{"203.0.113.9"}}, {URL: torURL, Fetched: time.Now().Add(-refresh), Lines: []string{"198.51.100.9"}}, }) if err != nil { t.Fatalf("load: %v", err) } run(t, lists) wantFetches(t, servers, 1) wantListedBy(t, lists, "203.0.113.9", dropURL) wantListedBy(t, lists, "198.51.100.10", torURL) time.Sleep(refresh - time.Hour - time.Nanosecond) wantFetches(t, servers, 1) time.Sleep(time.Nanosecond) wantFetches(t, servers, 2) wantListedBy(t, lists, "203.0.113.10", dropURL) }) } func TestRestartWaitsRefreshAfterTheLastTryEvenOneThatFailed(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { // drop.txt is fetched, and a refresh later the fetch downloads it // whole but fails on a line that does not read. servers := &standIn{lists: map[string]string{dropURL: "198.51.100.1\n"}} lists := start(t, servers, params(dropURL)) servers.set(dropURL, "198.51.100.2\n\n") time.Sleep(refresh) wantFetches(t, servers, 2) // Restarted with what reputation.json keeps, it waits a refresh // after the failed try, as it does while it runs. restarted := &standIn{lists: map[string]string{dropURL: "198.51.100.2\n"}} again := reputation.New(params(dropURL)) again.SetTransport(restarted) err := again.Load(lists.Snapshot()) if err != nil { t.Fatalf("load: %v", err) } run(t, again) wantFetches(t, restarted, 0) wantListedBy(t, again, "198.51.100.1", dropURL) time.Sleep(refresh - time.Nanosecond) wantFetches(t, restarted, 0) time.Sleep(time.Nanosecond) wantFetches(t, restarted, 1) wantListedBy(t, again, "198.51.100.2", dropURL) }) } func TestFetchCutOffAsItStopsIsNoFailureButARestartWaitsForIt(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { // Stopped 30 seconds into the fetch of drop.txt, before tor.txt's. servers := &standIn{lists: map[string]string{}, hanging: true} queue := newQueue() p := params(dropURL, torURL) p.Alerts = queue lists := reputation.New(p) lists.SetTransport(servers) ctx, stop := context.WithCancel(t.Context()) stopped := make(chan struct{}) go func() { lists.Run(ctx) close(stopped) }() time.Sleep(30 * time.Second) stop() <-stopped wantFetches(t, servers, 1) if lists.Failures(dropURL) != 0 || len(waiting(queue)) != 0 { t.Errorf("%d failures and alerts %+v, want none", lists.Failures(dropURL), waiting(queue)) } // Restarted an hour later with what reputation.json keeps, it fetches // tor.txt, never tried, at once, and drop.txt a refresh after its // cut-off try. time.Sleep(time.Hour) restarted := &standIn{lists: map[string]string{ dropURL: "203.0.113.7\n", torURL: "198.51.100.7\n", }} again := reputation.New(params(dropURL, torURL)) again.SetTransport(restarted) err := again.Load(lists.Snapshot()) if err != nil { t.Fatalf("load: %v", err) } run(t, again) wantFetches(t, restarted, 1) wantListedBy(t, again, "198.51.100.7", torURL) time.Sleep(refresh - time.Hour - time.Nanosecond) wantFetches(t, restarted, 1) time.Sleep(time.Nanosecond) wantFetches(t, restarted, 2) wantListedBy(t, again, "203.0.113.7", dropURL) }) } func TestASNLimitPercentFileGivesEachASNumberItsLowestPercentage(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { servers := &standIn{lists: map[string]string{ asnURL: "# hosting networks\nAS14061:50 ; DigitalOcean\nas16276:25\n\n" + "AS14061:10\nAS14061:30\n", }} p := params() p.ASNLimitPercentURL = asnURL lists := start(t, servers, p) for asn, want := range map[string]int64{"AS14061": 10, "AS16276": 25} { percent, listed := lists.ASNLimitPercent(asn) if !listed || percent != want { t.Errorf("%s has %d (listed %t), want %d", asn, percent, listed, want) } } if _, listed := lists.ASNLimitPercent("AS64496"); listed { t.Error("AS64496 is listed") } // A line that does not read fails the fetch. servers.set(asnURL, "AS14061:50\nAS16276\n") time.Sleep(refresh) synctest.Wait() if lists.Failures(asnURL) != 1 { t.Errorf("%d failures, want 1", lists.Failures(asnURL)) } }) } func TestLoadDropsCopiesOfListsNotNamedAndRefusesOnesThatDoNotRead(t *testing.T) { t.Parallel() fetched := time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC) kept := reputation.List{ URL: dropURL, Tried: fetched, Fetched: fetched, Lines: []string{"203.0.113.9"}, } lists := reputation.New(params(dropURL)) err := lists.Load([]reputation.List{kept, { URL: torURL, Tried: fetched, Fetched: fetched, Lines: []string{"198.51.100.9"}, }}) if err != nil { t.Fatalf("load: %v", err) } if got := lists.Snapshot(); !reflect.DeepEqual(got, []reputation.List{kept}) { t.Errorf("copies %+v, want only %+v", got, kept) } err = lists.Load([]reputation.List{{URL: dropURL, Fetched: fetched, Lines: []string{ "; DROP", "203.0.113.300", }}}) const want = "the copy of " + dropURL + ": line 2 is not an address or a netblock, such as 192.0.2.0/24" if err == nil || err.Error() != want { t.Errorf("error %v, want %s", err, want) } if got := lists.Snapshot(); !reflect.DeepEqual(got, []reputation.List{kept}) { t.Errorf("copies %+v after the error, want %+v still", got, kept) } } // standIn is a stand-in for the servers the lists are fetched from. It // notes the URL of each fetch. type standIn struct { mu sync.Mutex // lists are what it answers with, by URL; it answers a URL it has no // list for with 503, and none at all while hanging. lists map[string]string hanging bool fetches []string } // RoundTrip has the stand-in answer req, in place of the network. A fetch // abandoned before the stand-in answers fails, as over the network. func (s *standIn) RoundTrip(req *http.Request) (*http.Response, error) { s.mu.Lock() s.fetches = append(s.fetches, req.URL.String()) list, found := s.lists[req.URL.String()] hanging := s.hanging s.mu.Unlock() if hanging { <-req.Context().Done() return nil, req.Context().Err() } status := http.StatusOK if !found { status = http.StatusServiceUnavailable } return &http.Response{ StatusCode: status, Status: fmt.Sprintf("%d %s", status, http.StatusText(status)), Header: http.Header{}, Body: io.NopCloser(strings.NewReader(list)), Request: req, }, nil } // set has the stand-in answer listURL with list, or with 503 for "". func (s *standIn) set(listURL, list string) { s.mu.Lock() defer s.mu.Unlock() if list == "" { delete(s.lists, listURL) return } s.lists[listURL] = list } // params returns the Params of the blocklists at urls, refreshed every // refresh, by the bubble's clock, with alerts to a queue that sends none. func params(urls ...string) reputation.Params { return reputation.Params{ BlocklistURLs: urls, Refresh: refresh, Now: time.Now, ProcessLog: slog.New(slog.DiscardHandler), Alerts: newQueue(), } } // newQueue returns a queue of alerts to a webhook that is never sent // them, with a cooldown of cooldown. func newQueue() *alerts.Queue { return alerts.New(alerts.Params{ WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"}, Events: alerts.Events(), Cooldown: cooldown, Now: time.Now, }) } // start returns the lists of p, fetched through servers by Run, which runs // until the test ends, once Run has fetched those due at start. func start( t *testing.T, servers http.RoundTripper, p reputation.Params, ) *reputation.Lists { t.Helper() lists := reputation.New(p) lists.SetTransport(servers) run(t, lists) return lists } // run runs lists' Run until the test ends, and waits until it has fetched // the lists due. func run(t *testing.T, lists *reputation.Lists) { t.Helper() ctx, stop := context.WithCancel(t.Context()) stopped := make(chan struct{}) go func() { lists.Run(ctx) close(stopped) }() t.Cleanup(func() { stop() <-stopped }) synctest.Wait() } // wantFetches waits until Run has made the fetches due, and checks how // many the servers have had. func wantFetches(t *testing.T, servers *standIn, want int) { t.Helper() synctest.Wait() servers.mu.Lock() got := len(servers.fetches) servers.mu.Unlock() if got != want { t.Errorf("%d fetches, want %d", got, want) } } // wantListedBy checks the URLs of the blocklists lists says list addr. func wantListedBy(t *testing.T, lists *reputation.Lists, addr string, want ...string) { t.Helper() got := lists.ListedBy(netip.MustParseAddr(addr)) if !slices.Equal(got, want) { t.Errorf("%s is listed by %v, want %v", addr, got, want) } } // waiting returns the alerts waiting in queue. func waiting(queue *alerts.Queue) []alerts.Alert { return queue.Snapshot().Waiting[alerts.DestinationWebhook] } // fetchFailure is the source_failure alert raised at the time raised for // a fetch of the list at listURL that failed with err. func fetchFailure(raised time.Time, listURL, err string) alerts.Alert { return alerts.Alert{ Time: raised, Event: alerts.EventSourceFailure, Reason: "fetching a list failed", Detail: map[string]any{"source": listURL, "error": err}, } } // wantAlert checks that want is the one alert waiting in queue, and that // the cooldown has held back one repeat of it. func wantAlert(t *testing.T, queue *alerts.Queue, want alerts.Alert) { t.Helper() got := waiting(queue) if len(got) != 1 || !reflect.DeepEqual(got[0], want) || queue.Suppressed() != 1 { t.Errorf("alerts waiting %+v, %d held back, want only %+v and 1", got, queue.Suppressed(), want) } }