Blocklists and an AS percentage file fetched by URL (closes #29)
check / check (push) Waiting to run
check / check (push) Waiting to run
SWWAF_BLOCKLIST_URLS names lists of addresses and netblocks, fetched every SWWAF_BLOCKLIST_REFRESH (24h, never under 1h); an IPv4-mapped line stands for its IPv4 address or netblock. reputation.json keeps each list's last try, failed or not, even one cut off by a stop, which a restart waits on as a running instance does, and its last good copy, whole, used while a fetch fails. 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 failed fetch is retried after the refresh, not sooner. Not done: ban notes do not name the lists yet. Model: opus-5-5
This commit was merged in pull request #108.
This commit is contained in:
@@ -0,0 +1,9 @@
|
||||
package reputation
|
||||
|
||||
import "net/http"
|
||||
|
||||
// SetTransport has l's fetches go through transport instead of the
|
||||
// network.
|
||||
func (l *Lists) SetTransport(transport http.RoundTripper) {
|
||||
l.httpClient.Transport = transport
|
||||
}
|
||||
@@ -0,0 +1,499 @@
|
||||
// Package reputation fetches the lists the settings name by URL: the
|
||||
// blocklists of SWWAF_BLOCKLIST_URLS, and the file of AS:percent lines
|
||||
// SWWAF_ASN_LIMIT_PERCENT_URL names. It keeps the last good copy of each,
|
||||
// whole, comment lines included, which is used while a fetch fails, and
|
||||
// when each was last tried, which the state package writes to
|
||||
// reputation.json and reads from it, so that a restart keeps them too.
|
||||
package reputation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/smallwebwaf/internal/alerts"
|
||||
"sneak.berlin/go/smallwebwaf/internal/config"
|
||||
)
|
||||
|
||||
const (
|
||||
// maxListBytes is the most of a list that is read. A longer one is a
|
||||
// failure, so that a wrong URL cannot fill the memory.
|
||||
maxListBytes = 16 << 20
|
||||
// fetchTimeout bounds one fetch of a list.
|
||||
fetchTimeout = time.Minute
|
||||
// mappedBits is the length of ::ffff:0.0.0.0/96, the netblock of every
|
||||
// IPv4-mapped address.
|
||||
mappedBits = 96
|
||||
)
|
||||
|
||||
var (
|
||||
errStatus = errors.New("the server answered")
|
||||
errTooLong = errors.New("the list is longer than 16 MiB")
|
||||
errNotNetblock = errors.New("is not an address or a netblock, such as 192.0.2.0/24")
|
||||
errNotASNPercent = errors.New(
|
||||
"is not an AS number, : and a percentage, such as AS64496:50")
|
||||
)
|
||||
|
||||
// List is a list as reputation.json holds it: the URL it is fetched from,
|
||||
// when it was last tried, the fetch failed or not, and its last good copy:
|
||||
// when that was fetched, and its lines, as fetched, comment lines
|
||||
// included, both left out while no fetch of it has succeeded.
|
||||
type List struct {
|
||||
URL string `json:"url"`
|
||||
Tried time.Time `json:"tried"`
|
||||
Fetched time.Time `json:"fetched,omitzero"`
|
||||
Lines []string `json:"lines,omitzero"`
|
||||
}
|
||||
|
||||
// Params are what New needs.
|
||||
type Params struct {
|
||||
// BlocklistURLs are the blocklists (SWWAF_BLOCKLIST_URLS), and
|
||||
// ASNLimitPercentURL the file of AS:percent lines
|
||||
// (SWWAF_ASN_LIMIT_PERCENT_URL), "" while it is unset.
|
||||
BlocklistURLs []string
|
||||
ASNLimitPercentURL string
|
||||
// Refresh is how long after a list was last fetched or tried it is
|
||||
// fetched again (SWWAF_BLOCKLIST_REFRESH).
|
||||
Refresh time.Duration
|
||||
// Now tells the time, normally time.Now in UTC.
|
||||
Now func() time.Time
|
||||
// ProcessLog receives each fetch of a list, and why one failed.
|
||||
ProcessLog *slog.Logger
|
||||
// Alerts receive a source_failure alert for each fetch that fails.
|
||||
Alerts *alerts.Queue
|
||||
}
|
||||
|
||||
// Lists are the lists Params names, each with its last good copy. They
|
||||
// are safe for concurrent use.
|
||||
type Lists struct {
|
||||
params Params
|
||||
httpClient *http.Client
|
||||
|
||||
mu sync.Mutex
|
||||
// lists are by URL, one for each URL Params names.
|
||||
lists map[string]*list
|
||||
}
|
||||
|
||||
// list is one list: what reputation.json keeps of it, its last try, zero
|
||||
// before the first, and its last good copy, what that copy says, and how
|
||||
// many fetches of it failed.
|
||||
type list struct {
|
||||
kept List
|
||||
entries entries
|
||||
failures int
|
||||
}
|
||||
|
||||
// entries are what the lines of a copy say: for a blocklist, the netblocks
|
||||
// it names, with the lengths among them, and for the file of AS:percent
|
||||
// lines, the percentage it gives each AS number.
|
||||
type entries struct {
|
||||
netblocks map[netip.Prefix]bool
|
||||
lengths []int
|
||||
percents map[string]int64
|
||||
}
|
||||
|
||||
// New returns the lists, without a copy of any yet.
|
||||
func New(params Params) *Lists {
|
||||
l := &Lists{params: params, httpClient: &http.Client{}, lists: map[string]*list{}}
|
||||
|
||||
for _, listURL := range l.URLs() {
|
||||
l.lists[listURL] = &list{kept: List{URL: listURL}}
|
||||
}
|
||||
|
||||
return l
|
||||
}
|
||||
|
||||
// URLs returns the URL of every list: the blocklists' in the order
|
||||
// SWWAF_BLOCKLIST_URLS names them, then SWWAF_ASN_LIMIT_PERCENT_URL.
|
||||
func (l *Lists) URLs() []string {
|
||||
urls := slices.Clone(l.params.BlocklistURLs)
|
||||
if l.params.ASNLimitPercentURL != "" {
|
||||
urls = append(urls, l.params.ASNLimitPercentURL)
|
||||
}
|
||||
|
||||
return urls
|
||||
}
|
||||
|
||||
// ListedBy returns the URLs of the blocklists whose copy lists addr, in
|
||||
// the order SWWAF_BLOCKLIST_URLS names them.
|
||||
func (l *Lists) ListedBy(addr netip.Addr) []string {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
var listedBy []string
|
||||
|
||||
for _, listURL := range l.params.BlocklistURLs {
|
||||
if l.lists[listURL].entries.contain(addr) {
|
||||
listedBy = append(listedBy, listURL)
|
||||
}
|
||||
}
|
||||
|
||||
return listedBy
|
||||
}
|
||||
|
||||
// ASNLimitPercent returns the percentage the copy of the file of
|
||||
// AS:percent lines gives asn, and whether it lists asn.
|
||||
func (l *Lists) ASNLimitPercent(asn string) (int64, bool) {
|
||||
if l.params.ASNLimitPercentURL == "" {
|
||||
return 0, false
|
||||
}
|
||||
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
percent, listed := l.lists[l.params.ASNLimitPercentURL].entries.percents[asn]
|
||||
|
||||
return percent, listed
|
||||
}
|
||||
|
||||
// Fetched returns when the copy in use of the list at listURL was
|
||||
// fetched, or zero while there is none.
|
||||
func (l *Lists) Fetched(listURL string) time.Time {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
return l.lists[listURL].kept.Fetched
|
||||
}
|
||||
|
||||
// Failures returns how many fetches of the list at listURL failed.
|
||||
func (l *Lists) Failures(listURL string) int {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
return l.lists[listURL].failures
|
||||
}
|
||||
|
||||
// Run fetches each list once Refresh has passed since it was last fetched
|
||||
// or tried, the later of the two, until ctx is done. A list never tried is
|
||||
// fetched at once, and so is one whose last try or copy, read from
|
||||
// reputation.json, is that old.
|
||||
func (l *Lists) Run(ctx context.Context) {
|
||||
if len(l.lists) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
for ctx.Err() == nil {
|
||||
next := l.fetchDue(ctx)
|
||||
timer := time.NewTimer(next.Sub(l.params.Now()))
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-timer.C:
|
||||
}
|
||||
|
||||
timer.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshot returns each list that has been tried, with its copy, if it
|
||||
// has one, sorted by URL, as reputation.json lists them.
|
||||
func (l *Lists) Snapshot() []List {
|
||||
l.mu.Lock()
|
||||
|
||||
tried := make([]List, 0, len(l.lists))
|
||||
|
||||
for _, held := range l.lists {
|
||||
if !held.kept.Tried.IsZero() {
|
||||
tried = append(tried, held.kept)
|
||||
}
|
||||
}
|
||||
|
||||
l.mu.Unlock()
|
||||
|
||||
slices.SortFunc(tried, func(a, b List) int {
|
||||
return strings.Compare(a.URL, b.URL)
|
||||
})
|
||||
|
||||
return tried
|
||||
}
|
||||
|
||||
// Load puts lists, read from reputation.json, in place of the last tries
|
||||
// and copies held. A list Params does not name is dropped. A copy with a
|
||||
// line that parse refuses is an error, and then nothing changes.
|
||||
func (l *Lists) Load(lists []List) error {
|
||||
found := make(map[string]entries, len(lists))
|
||||
|
||||
for _, kept := range lists {
|
||||
if _, named := l.lists[kept.URL]; !named {
|
||||
continue
|
||||
}
|
||||
|
||||
read, err := l.parse(kept.URL, kept.Lines)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the copy of %s: %w", kept.URL, err)
|
||||
}
|
||||
|
||||
found[kept.URL] = read
|
||||
}
|
||||
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
for listURL, held := range l.lists {
|
||||
held.kept, held.entries = List{URL: listURL}, entries{}
|
||||
}
|
||||
|
||||
for _, kept := range lists {
|
||||
read, named := found[kept.URL]
|
||||
if named {
|
||||
l.lists[kept.URL].kept, l.lists[kept.URL].entries = kept, read
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// fetchDue fetches each list that is due, one after another, and returns
|
||||
// when the next is due. Once ctx has ended, it starts none, since a fetch
|
||||
// cut off is noted as a try.
|
||||
func (l *Lists) fetchDue(ctx context.Context) time.Time {
|
||||
var next time.Time
|
||||
|
||||
for _, listURL := range l.URLs() {
|
||||
due := l.due(listURL)
|
||||
if ctx.Err() == nil && !l.params.Now().Before(due) {
|
||||
l.fetch(ctx, listURL)
|
||||
due = l.due(listURL)
|
||||
}
|
||||
|
||||
if next.IsZero() || due.Before(next) {
|
||||
next = due
|
||||
}
|
||||
}
|
||||
|
||||
return next
|
||||
}
|
||||
|
||||
// due returns when the list at listURL is to be fetched: Refresh after it
|
||||
// was last fetched or tried, the later of the two.
|
||||
func (l *Lists) due(listURL string) time.Time {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
held := l.lists[listURL]
|
||||
|
||||
last := held.kept.Fetched
|
||||
if held.kept.Tried.After(last) {
|
||||
last = held.kept.Tried
|
||||
}
|
||||
|
||||
return last.Add(l.params.Refresh)
|
||||
}
|
||||
|
||||
// fetch fetches the list at listURL, and notes the try. A good copy takes
|
||||
// the place of the one held. A failure leaves that in use, and is counted,
|
||||
// logged and raised as a source_failure alert. A fetch cut off as ctx
|
||||
// ends, as smallwebwaf stops, is no failure, but is still noted as a try,
|
||||
// so that a restart waits for it: the server may have had its request.
|
||||
func (l *Lists) fetch(ctx context.Context, listURL string) {
|
||||
lines, err := l.get(ctx, listURL)
|
||||
|
||||
var found entries
|
||||
if err == nil {
|
||||
found, err = l.parse(listURL, lines)
|
||||
}
|
||||
|
||||
cutOff := err != nil && ctx.Err() != nil
|
||||
now := l.params.Now()
|
||||
|
||||
l.mu.Lock()
|
||||
|
||||
held := l.lists[listURL]
|
||||
held.kept.Tried = now
|
||||
|
||||
if err == nil {
|
||||
held.kept.Fetched, held.kept.Lines = now, lines
|
||||
held.entries = found
|
||||
} else if !cutOff {
|
||||
held.failures++
|
||||
}
|
||||
|
||||
l.mu.Unlock()
|
||||
|
||||
if cutOff {
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
const failed = "fetching a list failed"
|
||||
|
||||
// Raised before it is logged, so that the alert is there once the
|
||||
// log line is.
|
||||
l.params.Alerts.Raise(alerts.Alert{
|
||||
Event: alerts.EventSourceFailure,
|
||||
Reason: failed,
|
||||
Detail: map[string]any{"source": listURL, "error": err.Error()},
|
||||
})
|
||||
l.params.ProcessLog.Warn(failed, "url", listURL, "error", err.Error())
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
l.params.ProcessLog.Info("fetched a list", "url", listURL, "lines", len(lines))
|
||||
}
|
||||
|
||||
// get fetches the list at listURL, and returns its lines. An answer other
|
||||
// than 200, or a list longer than maxListBytes, is a failure.
|
||||
func (l *Lists) get(ctx context.Context, listURL string) ([]string, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, fetchTimeout)
|
||||
defer cancel()
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, listURL, http.NoBody)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("make the request: %w", err)
|
||||
}
|
||||
|
||||
res, err := l.httpClient.Do(req)
|
||||
if err != nil {
|
||||
// Do's error names the URL, which the log line and the alert name
|
||||
// already: only what went wrong is kept.
|
||||
return nil, fmt.Errorf("fetch the list: %w", errors.Unwrap(err))
|
||||
}
|
||||
|
||||
defer func() {
|
||||
_ = res.Body.Close()
|
||||
}()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("%w %s", errStatus, res.Status)
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(res.Body, maxListBytes+1))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read the list: %w", err)
|
||||
}
|
||||
|
||||
if len(body) > maxListBytes {
|
||||
return nil, errTooLong
|
||||
}
|
||||
|
||||
lines := []string{}
|
||||
for line := range strings.Lines(string(body)) {
|
||||
lines = append(lines, strings.TrimSuffix(line, "\n"))
|
||||
}
|
||||
|
||||
return lines, nil
|
||||
}
|
||||
|
||||
// parse reads the lines of the list at listURL: those of a blocklist, or
|
||||
// of the file of AS:percent lines. Anything after a ; or a # on a line is
|
||||
// left out, and so is a line left blank. Any other line that does not read
|
||||
// is an error naming it by its number.
|
||||
func (l *Lists) parse(listURL string, lines []string) (entries, error) {
|
||||
if listURL == l.params.ASNLimitPercentURL {
|
||||
return parsePercents(lines)
|
||||
}
|
||||
|
||||
return parseNetblocks(lines)
|
||||
}
|
||||
|
||||
// parseNetblocks reads a blocklist's lines, each an address or a netblock
|
||||
// as the settings take them.
|
||||
func parseNetblocks(lines []string) (entries, error) {
|
||||
found := entries{netblocks: map[netip.Prefix]bool{}}
|
||||
|
||||
for i, line := range lines {
|
||||
text := withoutComment(line)
|
||||
if text == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
netblock, ok := parseNetblock(text)
|
||||
if !ok {
|
||||
return entries{}, fmt.Errorf("line %d %w", i+1, errNotNetblock)
|
||||
}
|
||||
|
||||
found.netblocks[netblock] = true
|
||||
|
||||
if !slices.Contains(found.lengths, netblock.Bits()) {
|
||||
found.lengths = append(found.lengths, netblock.Bits())
|
||||
}
|
||||
}
|
||||
|
||||
return found, nil
|
||||
}
|
||||
|
||||
// parseNetblock reads text, a line of a blocklist, and reports whether it
|
||||
// is an address or a netblock as the settings take them. A client's IPv4
|
||||
// address is checked as IPv4, never IPv4-mapped, so an IPv4-mapped line,
|
||||
// such as ::ffff:192.0.2.0/120, is read as the IPv4 address or netblock it
|
||||
// stands for, 192.0.2.0/24, and a mapped netblock shorter than /96, which
|
||||
// stands for none, is refused.
|
||||
func parseNetblock(text string) (netip.Prefix, bool) {
|
||||
netblock, err := config.ParseNetblock(text)
|
||||
if err != nil {
|
||||
return netip.Prefix{}, false
|
||||
}
|
||||
|
||||
// The address as written: ParseNetblock's has the bits past the
|
||||
// netblock's length cleared, the ::ffff among them below /96.
|
||||
written, _, _ := strings.Cut(text, "/")
|
||||
if addr, _ := netip.ParseAddr(written); !addr.Is4In6() {
|
||||
return netblock, true
|
||||
}
|
||||
|
||||
if netblock.Bits() < mappedBits {
|
||||
return netip.Prefix{}, false
|
||||
}
|
||||
|
||||
return netip.PrefixFrom(netblock.Addr().Unmap(), netblock.Bits()-mappedBits), true
|
||||
}
|
||||
|
||||
// parsePercents reads the lines of the file of AS:percent lines, each an
|
||||
// AS number, : and a percentage, as SWWAF_ASN_LIMIT_PERCENT takes them. An
|
||||
// AS number listed more than once gets the lowest of its percentages.
|
||||
func parsePercents(lines []string) (entries, error) {
|
||||
found := entries{percents: map[string]int64{}}
|
||||
|
||||
for i, line := range lines {
|
||||
text := withoutComment(line)
|
||||
if text == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
asnText, percentText, _ := strings.Cut(text, ":")
|
||||
asn, asnErr := config.ParseASN(asnText)
|
||||
|
||||
percent, percentErr := config.ParsePercent(percentText)
|
||||
if asnErr != nil || percentErr != nil {
|
||||
return entries{}, fmt.Errorf("line %d %w", i+1, errNotASNPercent)
|
||||
}
|
||||
|
||||
earlier, listed := found.percents[asn]
|
||||
if !listed || percent < earlier {
|
||||
found.percents[asn] = percent
|
||||
}
|
||||
}
|
||||
|
||||
return found, nil
|
||||
}
|
||||
|
||||
// withoutComment returns line without anything after a ; or a #, and
|
||||
// without the spaces around what is left.
|
||||
func withoutComment(line string) string {
|
||||
text, _, _ := strings.Cut(line, ";")
|
||||
text, _, _ = strings.Cut(text, "#")
|
||||
|
||||
return strings.TrimSpace(text)
|
||||
}
|
||||
|
||||
// contain reports whether the netblocks of a blocklist's copy hold addr:
|
||||
// whether addr, cut to one of their lengths, is one of them.
|
||||
func (e entries) contain(addr netip.Addr) bool {
|
||||
for _, length := range e.lengths {
|
||||
netblock, err := addr.Prefix(length)
|
||||
if err == nil && e.netblocks[netblock] {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,610 @@
|
||||
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<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)
|
||||
|
||||
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, 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 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<html>\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 *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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user