smallwebwaf/internal/reputation/dnsbl.go

package reputation

import (
	"cmp"
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/netip"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/hashicorp/golang-lru/v2/simplelru"
	"sneak.berlin/go/smallwebwaf/internal/alerts"
)

const (
	// maxVerdicts is how many verdicts are kept. Past it, the one fetched
	// longest ago is dropped.
	maxVerdicts = 100000
	// maxQueries is how many queries may be under way at once. Past it, a
	// zone is not asked about a client until the client's next request, so
	// that a swarm of new addresses cannot fill the memory.
	maxQueries = 1000
	// failureDelay is how long a zone is not asked again after a query to
	// it fails, so that a zone refusing queries is not asked on every
	// request.
	failureDelay = time.Minute
)

var (
	errAsk        = errors.New("ask the zone")
	errRefused    = errors.New("the zone refused the query")
	errNotListing = errors.New("the answer is outside 127.0.0.0/8")
)

// Verdict is what a zone said about a client, as reputation.json holds
// it: the zone, the client's address, whether the zone lists it, and when
// the zone answered.
type Verdict struct {
	Zone    string     `json:"zone"`
	Client  netip.Addr `json:"client"`
	Listed  bool       `json:"listed"`
	Fetched time.Time  `json:"fetched"`
}

// DNSBLParams are what NewDNSBL needs.
type DNSBLParams struct {
	// Zones are the DNSBL zones clients are asked about in
	// (SWWAF_DNSBL_ZONES).
	Zones []string
	// Resolver is the resolver they are asked through
	// (SWWAF_DNSBL_RESOLVER), or, while it is the zero AddrPort, the
	// host's, as /etc/resolv.conf names it.
	Resolver netip.AddrPort
	// CacheTTL is how long a verdict is used after it was fetched
	// (SWWAF_REPUTATION_CACHE_TTL), and Timeout how long a query may take
	// (SWWAF_REPUTATION_TIMEOUT).
	CacheTTL time.Duration
	Timeout  time.Duration
	// Now tells the time, normally time.Now in UTC.
	Now func() time.Time
	// ProcessLog receives each query that fails, and why.
	ProcessLog *slog.Logger
	// Alerts receive a source_failure alert for each query that fails.
	Alerts *alerts.Queue
}

// DNSBL asks the DNSBL zones about clients, in the background, and keeps
// their verdicts. It is safe for concurrent use.
type DNSBL struct {
	params   DNSBLParams
	resolver *net.Resolver

	mu sync.Mutex
	// verdicts are by query. Each is added as it is fetched and never moved
	// up, so that the one fetched longest ago is the first dropped.
	verdicts *simplelru.LRU[query, Verdict]
	// asking are the queries under way.
	asking map[query]bool
	// queries and failures count, by zone, the queries made and those that
	// failed, and retryAt is when a zone whose last query failed may be
	// asked again.
	queries  map[string]int
	failures map[string]int
	retryAt  map[string]time.Time
}

// query is a client's address, to ask a zone about.
type query struct {
	zone   string
	client netip.Addr
}

// NewDNSBL returns a DNSBL with no verdict yet.
func NewDNSBL(params DNSBLParams) *DNSBL {
	verdicts, err := simplelru.NewLRU[query, Verdict](maxVerdicts, nil)
	if err != nil {
		panic(err) // NewLRU fails only for a size below one
	}

	resolver := &net.Resolver{}
	if params.Resolver.IsValid() {
		// Dial is used by Go's own resolver alone.
		resolver.PreferGo = true
		resolver.Dial = func(ctx context.Context, network, _ string) (net.Conn, error) {
			var dialer net.Dialer

			return dialer.DialContext(ctx, network, params.Resolver.String())
		}
	}

	return &DNSBL{
		params:   params,
		resolver: resolver,
		verdicts: verdicts,
		asking:   map[query]bool{},
		queries:  map[string]int{},
		failures: map[string]int{},
		retryAt:  map[string]time.Time{},
	}
}

// Zones returns the zones, in the order SWWAF_DNSBL_ZONES names them.
func (d *DNSBL) Zones() []string {
	return slices.Clone(d.params.Zones)
}

// ListedBy returns the zones whose verdict on addr, a client's address,
// lists it, in the order SWWAF_DNSBL_ZONES names them. A verdict is used
// until CacheTTL has passed since it was fetched. Each zone without one is
// asked about addr in the background, unless a query about addr to it is
// under way, the zone is left alone after a failure, or maxQueries are
// under way; ListedBy never waits for a query. ctx is the context of the
// client's request, and a query goes on after the request ends.
func (d *DNSBL) ListedBy(ctx context.Context, addr netip.Addr) []string {
	d.mu.Lock()
	defer d.mu.Unlock()

	now := d.params.Now()

	var listedBy []string

	for _, zone := range d.params.Zones {
		q := query{zone: zone, client: addr}

		kept, found := d.verdicts.Peek(q)

		switch {
		case found && now.Sub(kept.Fetched) < d.params.CacheTTL:
			if kept.Listed {
				listedBy = append(listedBy, zone)
			}
		case !d.asking[q] && !now.Before(d.retryAt[zone]) && len(d.asking) < maxQueries:
			d.asking[q] = true
			d.queries[zone]++

			go d.ask(context.WithoutCancel(ctx), q)
		}
	}

	return listedBy
}

// Queries returns how many queries were made to zone.
func (d *DNSBL) Queries(zone string) int {
	d.mu.Lock()
	defer d.mu.Unlock()

	return d.queries[zone]
}

// Failures returns how many queries to zone failed.
func (d *DNSBL) Failures(zone string) int {
	d.mu.Lock()
	defer d.mu.Unlock()

	return d.failures[zone]
}

// Snapshot returns every verdict still in use, sorted by client, then by
// zone, as reputation.json lists them.
func (d *DNSBL) Snapshot() []Verdict {
	d.mu.Lock()

	now := d.params.Now()
	verdicts := make([]Verdict, 0, d.verdicts.Len())

	for _, kept := range d.verdicts.Values() {
		if now.Sub(kept.Fetched) < d.params.CacheTTL {
			verdicts = append(verdicts, kept)
		}
	}

	d.mu.Unlock()

	slices.SortFunc(verdicts, func(a, b Verdict) int {
		return cmp.Or(a.Client.Compare(b.Client), strings.Compare(a.Zone, b.Zone))
	})

	return verdicts
}

// Load keeps verdicts, read from reputation.json, in place of those it
// keeps. A verdict of a zone SWWAF_DNSBL_ZONES does not name, or one
// fetched CacheTTL ago or more, is dropped, and past maxVerdicts, those
// fetched longest ago.
func (d *DNSBL) Load(verdicts []Verdict) {
	verdicts = slices.Clone(verdicts)
	slices.SortStableFunc(verdicts, func(a, b Verdict) int {
		return a.Fetched.Compare(b.Fetched)
	})

	d.mu.Lock()
	defer d.mu.Unlock()

	d.verdicts.Purge()

	now := d.params.Now()

	for _, kept := range verdicts {
		if slices.Contains(d.params.Zones, kept.Zone) &&
			now.Sub(kept.Fetched) < d.params.CacheTTL {
			d.verdicts.Add(query{zone: kept.Zone, client: kept.Client}, kept)
		}
	}
}

// ask asks q's zone about q's client, keeps the verdict, and notes the
// query as no longer under way. A query that fails gives no verdict: it
// is counted, logged and raised as a source_failure alert, and the zone is
// not asked again for failureDelay.
func (d *DNSBL) ask(ctx context.Context, q query) {
	listed, err := d.lookUp(ctx, q)
	now := d.params.Now()

	d.mu.Lock()

	delete(d.asking, q)

	if err == nil {
		d.verdicts.Add(q, Verdict{
			Zone: q.zone, Client: q.client, Listed: listed, Fetched: now,
		})
	} else {
		d.failures[q.zone]++
		d.retryAt[q.zone] = now.Add(failureDelay)
	}

	d.mu.Unlock()

	if err != nil {
		const failed = "asking a DNSBL zone failed"

		// Raised before it is logged, so that the alert is there once the
		// log line is.
		d.params.Alerts.Raise(alerts.Alert{
			Event:  alerts.EventSourceFailure,
			Reason: failed,
			Detail: map[string]any{"source": q.zone, "error": err.Error()},
		})
		d.params.ProcessLog.Warn(failed, "zone", q.zone, "error", err.Error())
	}
}

// lookUp asks q's zone about q's client through the resolver, and returns
// whether the zone lists it, as readAnswer reads the answer. No such name
// is a client the zone does not list. A query not answered within Timeout
// fails.
func (d *DNSBL) lookUp(ctx context.Context, q query) (bool, error) {
	ctx, cancel := context.WithTimeout(ctx, d.params.Timeout)
	defer cancel()

	answer, err := d.resolver.LookupNetIP(ctx, "ip4", queryName(q.zone, q.client))

	var dnsErr *net.DNSError

	switch {
	case err == nil:
		return readAnswer(answer)
	case errors.As(err, &dnsErr) && dnsErr.IsNotFound:
		return false, nil
	case errors.As(err, &dnsErr):
		// The error names the name asked about, which holds the client's
		// address, which is not to be logged: only what went wrong is kept.
		return false, fmt.Errorf("%w: %s", errAsk, dnsErr.Err)
	default:
		return false, fmt.Errorf("%w: %w", errAsk, err)
	}
}

// queryName returns the name a zone is asked about addr by, as RFC 5782
// builds it: the four numbers of an IPv4 address, or the 32 hex digits of
// an IPv6 address, in reverse order, each followed by a dot, then the zone
// and a dot, which makes it a full name, to which the resolver adds no
// search domain of /etc/resolv.conf.
func queryName(zone string, addr netip.Addr) string {
	parts := strings.Split(addr.String(), ".")
	if addr.Is6() {
		parts = strings.Split(hex.EncodeToString(addr.AsSlice()), "")
	}

	slices.Reverse(parts)

	return strings.Join(parts, ".") + "." + zone + "."
}

// readAnswer reads the addresses a zone answered with. An address in
// 127.0.0.0/8 lists the client, as RFC 5782 has zones answer, but one in
// 127.255.255.0/24 is how Spamhaus refuses a query, such as one sent
// through a public resolver or one past its limit, and is a failure. So is
// an address outside 127.0.0.0/8, such as a resolver gives that answers
// even for names that do not exist.
func readAnswer(answer []netip.Addr) (bool, error) {
	listing := netip.MustParsePrefix("127.0.0.0/8")
	refusal := netip.MustParsePrefix("127.255.255.0/24")

	for _, addr := range answer {
		switch {
		case refusal.Contains(addr):
			return false, fmt.Errorf("%w: %s", errRefused, addr)
		case !listing.Contains(addr):
			return false, fmt.Errorf("%w: %s", errNotListing, addr)
		}
	}

	return len(answer) > 0, nil
}