Blocklists and an AS percentage file fetched by URL (closes #29)

SWWAF_BLOCKLIST_URLS names lists of addresses and netblocks, fetched every SWWAF_BLOCKLIST_REFRESH (24h, never under 1h). The last good copy is kept whole in reputation.json and used while a fetch fails and across restarts. SWWAF_BLOCKLIST_ACTION denies, limits or only logs a listed client; the log line names the lists, each raises reputation_hit, and a failed fetch raises source_failure. SWWAF_ASN_LIMIT_PERCENT_URL is fetched the same way and counts as SWWAF_ASN_LIMIT_PERCENT does, the lower winning. Judgement call: a restart fetches only lists whose copy is due. Judgement call: a failed fetch is retried after the refresh, not sooner. Not done: ban notes do not name the lists yet. Model: opus-5-5
2026-10-07 15:10:41 +00:00
parent 82e20e0cb5
commit 0b2ff56417
19 changed files with 2403 additions and 322 deletions
@@ -0,0 +1,503 @@
+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 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<html>\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)
+
+				if got := lists.Snapshot(); !reflect.DeepEqual(got, kept) {
+					t.Errorf("copies %+v, want the first %+v", got, kept)
+				}
+
+				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, alerts.Alert{
+					Time:   time.Now().Add(-refresh),
+					Event:  alerts.EventSourceFailure,
+					Reason: "fetching a list failed",
+					Detail: map[string]any{"source": dropURL, "error": 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 TestFetchCutOffAsItStopsIsNoFailure(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		servers := &standIn{lists: map[string]string{}, hanging: true}
+		queue := newQueue()
+		p := params(dropURL)
+		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)
+		}()
+
+		synctest.Wait()
+		stop()
+		<-stopped
+
+		if lists.Failures(dropURL) != 0 || len(waiting(queue)) != 0 {
+			t.Errorf("%d failures and alerts %+v, want none", lists.Failures(dropURL),
+				waiting(queue))
+		}
+	})
+}
+
+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, Fetched: fetched, Lines: []string{"203.0.113.9"}}
+	lists := reputation.New(params(dropURL))
+
+	err := lists.Load([]reputation.List{
+		kept, {URL: torURL, 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 *standIn, 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]
+}
+
+// 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)
+	}
+}