Add asn= and prefix= tags from a routewatch instance (closes #204)

SWW_ROUTEWATCH_URL names the instance. The new heuristic routewatch_tags
asks it in the background about each client and tags the request. Answers
are kept in memory for an hour. A request without tags gives the reason in
routewatch_tags_missing, and it is never delayed or refused.

Model: opus-5-5

internal/discriminator/discriminator.go

@@ -123,6 +123,7 @@ func New(params Params) *Server {
 		Lists:      h.lists,
 		DNSBL:      h.dnsbl,
 		AbuseIPDB:  h.abuseIPDB,
+		RouteWatch: h.routeWatch,
 		Metrics:    h.metrics,
 	}
 }

internal/discriminator/heuristics.go

@@ -44,6 +44,7 @@ const (
 func beforeLookup() []heuristic {
 	return []heuristic{
 		tagLists{},
+		routeWatchTags{},
 		denyNets{},
 		activeBan{},
 	}

internal/discriminator/proxy.go

@@ -20,6 +20,7 @@ import (
 	"sneak.berlin/go/smallwebwaf/internal/metrics"
 	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
 	"sneak.berlin/go/smallwebwaf/internal/reputation"
+	"sneak.berlin/go/smallwebwaf/internal/routewatch"
 	"sneak.berlin/go/smallwebwaf/internal/rules"
 	"sneak.berlin/go/smallwebwaf/internal/waf"
 )
@@ -85,7 +86,8 @@ type Params struct {
 // Server is the server discriminator runs, with the parts whose state the state
 // files keep, the lookup database, nil unless SWW_LOOKUP_SOURCE is file, the
 // lists fetched from URLs, which its Run fetches, the DNSBL zones' verdicts,
-// AbuseIPDB's scores and checks spent, and the metrics.
+// AbuseIPDB's scores and checks spent, and the metrics. RouteWatch, nil while
+// SWW_ROUTEWATCH_URL is unset, keeps its answers in memory only.
 type Server struct {
 	*http.Server
 
@@ -97,6 +99,7 @@ type Server struct {
 	Lists      *reputation.Lists
 	DNSBL      *reputation.DNSBL
 	AbuseIPDB  *reputation.AbuseIPDB
+	RouteWatch *routewatch.Client
 	Metrics    *metrics.Metrics
 }
 
@@ -152,6 +155,12 @@ func newHandler(params Params) *handler {
 		coreRuleSet: params.CoreRuleSet,
 		alerts:      params.Alerts,
 	}
+	if params.Config.RouteWatchURL != "" {
+		h.routeWatch = routewatch.New(routewatch.Params{
+			URL: params.Config.RouteWatchURL, Now: params.Now, ProcessLog: params.ProcessLog,
+		})
+	}
+
 	h.geojs = lookup.New(lookup.Params{
 		URL:     params.GeoJSURL,
 		Timeout: params.Config.LookupTimeout,
@@ -255,6 +264,7 @@ type handler struct {
 	lists       *reputation.Lists
 	dnsbl       *reputation.DNSBL
 	abuseIPDB   *reputation.AbuseIPDB
+	routeWatch  *routewatch.Client
 	rules       *rules.Files
 	coreRuleSet *waf.CoreRuleSet
 	alerts      *alerts.Queue

internal/discriminator/routewatchtags.go

@@ -0,0 +1,42 @@
+package discriminator
+
+// The heuristic routewatch_tags. It reads SWW_ROUTEWATCH_URL.
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/config"
+)
+
+// routeWatchTags adds to the request the tags asn=<number> and prefix=<netblock>
+// from what routewatch answered about the client. It refuses nothing, and the
+// request does not wait for routewatch. A request that gets no tags from it has
+// the reason in the log line's routewatch_tags_missing. While SWW_ROUTEWATCH_URL
+// is unset it does nothing. Answers are kept by client and asked for by the
+// request's address, and a route without an origin AS gives only the prefix= tag.
+type routeWatchTags struct{}
+
+func (routeWatchTags) name() string { return "routewatch_tags" }
+
+func (routeWatchTags) check(ctx context.Context, rq *request, _ time.Time) *verdict {
+	if rq.h.routeWatch == nil {
+		return nil
+	}
+
+	route, missing := rq.h.routeWatch.Route(ctx, rq.h.clientGroup(rq.client), rq.client)
+	if missing != "" {
+		rq.line.RouteWatchTagsMissing = string(missing)
+
+		return nil
+	}
+
+	rq.addTag(config.PrefixTagPrefix + route.Netblock.String())
+
+	if route.ASN != 0 {
+		rq.addTag(config.ASNTagPrefix + strconv.FormatUint(uint64(route.ASN), 10))
+	}
+
+	return nil
+}

internal/discriminator/routewatchtags_test.go

@@ -0,0 +1,245 @@
+package discriminator_test
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"strings"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/discriminator"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+)
+
+// routeWatchURL is SWW_ROUTEWATCH_URL, and tagsMissing the log line's field
+// that says why a request got no tags from routewatch.
+const (
+	routeWatchURL = "SWW_ROUTEWATCH_URL"
+	tagsMissing   = "routewatch_tags_missing"
+	fromV6        = "2001:db8:1::5"
+	prefixV6      = "2001:db8::/32"
+	prefixDE      = "203.0.113.0/24"
+	prefixKP      = "198.51.100.0/24"
+)
+
+// standInRoute is what the stand-in for routewatch answers about an address.
+type standInRoute struct {
+	Netblock string `json:"netblock"`
+	ASN      int    `json:"asn"`
+}
+
+// startRouteWatch starts a stand-in for routewatch's IP info API that knows the
+// routes given by address, and returns its URL.
+func startRouteWatch(t *testing.T, routes map[string]standInRoute) string {
+	t.Helper()
+
+	server := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			address := strings.TrimPrefix(r.URL.Path, "/api/v1/ip/")
+
+			result := map[string]any{"ip": address}
+			if route, known := routes[address]; known {
+				result["netblock"], result["asn"] = route.Netblock, route.ASN
+			}
+
+			_ = json.NewEncoder(w).Encode(map[string]any{
+				"status": "ok",
+				"data":   map[string]any{"results": []any{result}},
+			})
+		}))
+	t.Cleanup(server.Close)
+
+	return server.URL
+}
+
+// waitForRouteWatch waits until routewatch's answer about from, in the client
+// group of the given length, is kept, so that the next request gets its tags.
+func waitForRouteWatch(
+	t *testing.T, server *discriminator.Server, from string, bits int,
+) {
+	t.Helper()
+
+	addr := netip.MustParseAddr(from)
+	client := netip.PrefixFrom(addr, bits).Masked()
+	deadline := time.Now().Add(5 * time.Second)
+
+	for {
+		_, missing := server.RouteWatch.Route(context.Background(), client, addr)
+		if missing == "" {
+			return
+		}
+
+		if time.Now().After(deadline) {
+			t.Fatalf("routewatch's answer about %s not kept: %q", from, missing)
+		}
+
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func wantTagsMissing(t *testing.T, line logLine, want string) {
+	t.Helper()
+
+	got, _ := line.fields[tagsMissing].(string)
+	if got != want {
+		t.Errorf("log line has %s %q, want %q", tagsMissing, got, want)
+	}
+}
+
+func TestRouteWatchTagsAreOnTheLogLineWithTheLookupsTags(t *testing.T) {
+	t.Parallel()
+
+	url := startRouteWatch(t, map[string]standInRoute{fromDE: {prefixDE, 64496}})
+	s, server, _ := startWithLookups(t, map[string]string{routeWatchURL: url})
+
+	waitForRouteWatch(t, server, fromDE, 32)
+
+	line := s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+	wantTags(t, line, "asn=64496", "country=DE", "prefix="+prefixDE)
+	wantTagsMissing(t, line, "")
+}
+
+func TestRouteWatchTagsAreOnALineDenyNetsRefused(t *testing.T) {
+	t.Parallel()
+
+	url := startRouteWatch(t, map[string]standInRoute{fromKP: {prefixKP, 64511}})
+	s, _, server := startWithClock(t, "", map[string]string{
+		lookupSource: off, routeWatchURL: url, denyNets: fromKP,
+	})
+
+	waitForRouteWatch(t, server, fromKP, 32)
+
+	line := s.get(fromKP, http.StatusForbidden, requestlog.ActionDenied)
+	wantHeuristic(t, line, "deny_nets", `{"netblock": "`+fromKP+`/32"}`)
+	wantTags(t, line, "asn=64511", "prefix="+prefixKP)
+}
+
+func TestRouteWithoutAnOriginASGivesOnlyThePrefixTag(t *testing.T) {
+	t.Parallel()
+
+	url := startRouteWatch(t, map[string]standInRoute{fromDE: {prefixDE, 0}})
+	s, _, server := startWithClock(t, "", map[string]string{
+		lookupSource: off, routeWatchURL: url,
+	})
+
+	waitForRouteWatch(t, server, fromDE, 32)
+
+	wantTags(t, s.get(fromDE, http.StatusOK, requestlog.ActionForward),
+		"prefix="+prefixDE)
+}
+
+func TestUnsetRouteWatchChangesNothing(t *testing.T) {
+	t.Parallel()
+
+	s, _, server := startWithClock(t, "", map[string]string{lookupSource: off})
+
+	if server.RouteWatch != nil {
+		t.Error("RouteWatch is set while SWW_ROUTEWATCH_URL is unset")
+	}
+
+	line := s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+	if _, present := line.fields[tagsMissing]; present {
+		t.Errorf("log line has %s", tagsMissing)
+	}
+
+	if _, present := line.fields["tags"]; present {
+		t.Errorf("log line has tags %v", line.fields["tags"])
+	}
+}
+
+func TestUnreachableRouteWatchLetsTheRequestThroughAndSaysWhy(t *testing.T) {
+	t.Parallel()
+
+	// A stand-in that is closed at once, so that the port refuses connections.
+	dead := httptest.NewServer(http.NotFoundHandler())
+	url := dead.URL
+
+	dead.Close()
+
+	s, _, _ := startWithClock(t, "", map[string]string{
+		lookupSource: off, routeWatchURL: url,
+	})
+
+	wantTagsMissing(t, s.get(fromDE, http.StatusOK, requestlog.ActionForward),
+		"no_answer_yet")
+
+	deadline := time.Now().Add(5 * time.Second)
+
+	for {
+		line := s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+
+		missing, _ := line.fields[tagsMissing].(string)
+		if missing == "request_failed" {
+			return
+		}
+
+		if time.Now().After(deadline) {
+			t.Fatalf("still %q, want request_failed", missing)
+		}
+
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func TestAddressWithoutARouteHasNoTagsAndTheLineSaysSo(t *testing.T) {
+	t.Parallel()
+
+	s, _, _ := startWithClock(t, "", map[string]string{
+		lookupSource: off, routeWatchURL: startRouteWatch(t, nil),
+	})
+
+	deadline := time.Now().Add(5 * time.Second)
+
+	for {
+		line := s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+
+		missing, _ := line.fields[tagsMissing].(string)
+		if missing == "no_route" {
+			return
+		}
+
+		if time.Now().After(deadline) {
+			t.Fatalf("still %q, want no_route", missing)
+		}
+
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func TestTagDenyRefusesOnARouteWatchPrefix(t *testing.T) {
+	t.Parallel()
+
+	url := startRouteWatch(t, map[string]standInRoute{
+		fromDE: {prefixDE, 64496}, fromKP: {prefixKP, 64511},
+	})
+	s, _, server := startWithClock(t, "", map[string]string{
+		lookupSource: off, routeWatchURL: url, tagDeny: "prefix=" + prefixKP,
+	})
+
+	waitForRouteWatch(t, server, fromKP, 32)
+	waitForRouteWatch(t, server, fromDE, 32)
+
+	wantHeuristic(t, s.get(fromKP, http.StatusForbidden, requestlog.ActionDenied),
+		"tag_deny", `{"tag": "prefix=`+prefixKP+`"}`)
+	s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+}
+
+func TestTagLimitPercentWorksWithAnIPv6Prefix(t *testing.T) {
+	t.Parallel()
+
+	url := startRouteWatch(t, map[string]standInRoute{fromV6: {prefixV6, 64496}})
+	s, _, server := startWithClock(t, "", map[string]string{
+		lookupSource: off, routeWatchURL: url, rateLimitPerMinute: "2",
+		tagLimitPercent: "prefix=" + prefixV6 + ":50",
+	})
+
+	waitForRouteWatch(t, server, fromV6, 64)
+
+	// Half of two is one: the second request breaks the limit.
+	line := s.get(fromV6, http.StatusOK, requestlog.ActionForward)
+	wantTags(t, line, "asn=64496", "prefix="+prefixV6)
+	s.get(fromV6, http.StatusForbidden, requestlog.ActionRateLimited)
+}