Add asn= and prefix= tags from a routewatch instance (closes #204)
check / check (push) Successful in 4m37s
check / check (push) Successful in 4m37s
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
This commit is contained in:
1 parent
d7c62f2854
commit
790f5df069
14 files changed
+946
-13
No files matched your search
@@ -0,0 +1,242 @@
|
||||
// Package routewatch asks a routewatch instance about client addresses, in the
|
||||
// background, and keeps its answers in memory. A routewatch instance knows, for
|
||||
// an address, the netblock it is announced in and the AS that originates it.
|
||||
package routewatch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"net/url"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/golang-lru/v2/simplelru"
|
||||
)
|
||||
|
||||
const (
|
||||
// answerKept is how long an answer is used after it came.
|
||||
answerKept = time.Hour
|
||||
// failureKept is how long routewatch is not asked again after a query
|
||||
// fails, so that a dead routewatch is not asked on every request.
|
||||
failureKept = time.Minute
|
||||
// queryTimeout is how long one query may take.
|
||||
queryTimeout = 2 * time.Second
|
||||
// maxAnswers is how many answers are kept. Past it, the one that came
|
||||
// longest ago is dropped.
|
||||
maxAnswers = 100000
|
||||
// maxQueries is how many queries may be under way at once. Past it, a
|
||||
// client is not asked about until its next request, so that a swarm of
|
||||
// new addresses cannot fill the memory.
|
||||
maxQueries = 100
|
||||
// maxAnswerBytes is the most of an answer that is read.
|
||||
maxAnswerBytes = 64 << 10
|
||||
)
|
||||
|
||||
// Missing is why Route gives no route. The request log's field
|
||||
// routewatch_tags_missing gives these values.
|
||||
type Missing string
|
||||
|
||||
const (
|
||||
// NoAnswerYet is a client routewatch has not answered about yet.
|
||||
NoAnswerYet Missing = "no_answer_yet"
|
||||
// RequestFailed is no answer because a recent query failed.
|
||||
RequestFailed Missing = "request_failed"
|
||||
// NoRoute is routewatch knowing no live route for the address.
|
||||
NoRoute Missing = "no_route"
|
||||
)
|
||||
|
||||
var (
|
||||
errStatus = errors.New("routewatch answered")
|
||||
errNoResult = errors.New("the answer holds no result")
|
||||
errNotNetblock = errors.New("the answer's netblock is not a netblock")
|
||||
)
|
||||
|
||||
// Route is what routewatch knows of an address: the netblock it is announced
|
||||
// in, and the AS that originates it.
|
||||
type Route struct {
|
||||
Netblock netip.Prefix
|
||||
ASN uint32
|
||||
}
|
||||
|
||||
// Params are what New needs.
|
||||
type Params struct {
|
||||
// URL is where routewatch is (SWW_ROUTEWATCH_URL).
|
||||
URL string
|
||||
// Now tells the time, normally time.Now in UTC.
|
||||
Now func() time.Time
|
||||
// ProcessLog receives the query that fails, and why.
|
||||
ProcessLog *slog.Logger
|
||||
}
|
||||
|
||||
// Client asks a routewatch instance about clients, in the background, and keeps
|
||||
// its answers. It is safe for concurrent use.
|
||||
type Client struct {
|
||||
params Params
|
||||
httpClient *http.Client
|
||||
|
||||
mu sync.Mutex
|
||||
// answers are by client. Each is added as it comes and never moved up,
|
||||
// so that the one that came longest ago is the first dropped.
|
||||
answers *simplelru.LRU[netip.Prefix, answer]
|
||||
// asking are the clients whose query is under way.
|
||||
asking map[netip.Prefix]bool
|
||||
// retryAt is when routewatch may be asked again after a query failed.
|
||||
retryAt time.Time
|
||||
}
|
||||
|
||||
// answer is one kept answer: the route, or why there is none, and when it came.
|
||||
type answer struct {
|
||||
route Route
|
||||
missing Missing
|
||||
came time.Time
|
||||
}
|
||||
|
||||
// New returns a Client with no answer yet.
|
||||
func New(params Params) *Client {
|
||||
answers, err := simplelru.NewLRU[netip.Prefix, answer](maxAnswers, nil)
|
||||
if err != nil {
|
||||
panic(err) // NewLRU fails only for a size below one
|
||||
}
|
||||
|
||||
return &Client{
|
||||
params: params,
|
||||
httpClient: &http.Client{},
|
||||
answers: answers,
|
||||
asking: map[netip.Prefix]bool{},
|
||||
}
|
||||
}
|
||||
|
||||
// Route returns what routewatch answered about client, an IPv4 address or an
|
||||
// IPv6 group, or why it has no route. It never waits for a query. A client
|
||||
// without an answer is asked about in the background, by addr, the address its
|
||||
// request came from, unless its query is under way, a query failed less than
|
||||
// failureKept ago, or maxQueries are under way. ctx is the context of the
|
||||
// client's request, and a query goes on after the request ends.
|
||||
func (c *Client) Route(
|
||||
ctx context.Context, client netip.Prefix, addr netip.Addr,
|
||||
) (Route, Missing) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
now := c.params.Now()
|
||||
|
||||
kept, found := c.answers.Peek(client)
|
||||
if found && now.Sub(kept.came) < answerKept {
|
||||
return kept.route, kept.missing
|
||||
}
|
||||
|
||||
if now.Before(c.retryAt) {
|
||||
return Route{}, RequestFailed
|
||||
}
|
||||
|
||||
if !c.asking[client] && len(c.asking) < maxQueries {
|
||||
c.asking[client] = true
|
||||
|
||||
go c.ask(context.WithoutCancel(ctx), client, addr)
|
||||
}
|
||||
|
||||
return Route{}, NoAnswerYet
|
||||
}
|
||||
|
||||
// ask asks routewatch about addr, keeps the answer as client's, and notes the
|
||||
// query as no longer under way. A query that fails gives no answer, and
|
||||
// routewatch is not asked again for failureKept. Only the first failure of
|
||||
// those under way is logged.
|
||||
func (c *Client) ask(ctx context.Context, client netip.Prefix, addr netip.Addr) {
|
||||
route, missing, err := c.query(ctx, addr)
|
||||
now := c.params.Now()
|
||||
|
||||
c.mu.Lock()
|
||||
|
||||
delete(c.asking, client)
|
||||
|
||||
firstFailure := err != nil && !now.Before(c.retryAt)
|
||||
|
||||
if err == nil {
|
||||
c.answers.Add(client, answer{route: route, missing: missing, came: now})
|
||||
} else {
|
||||
c.retryAt = now.Add(failureKept)
|
||||
}
|
||||
|
||||
c.mu.Unlock()
|
||||
|
||||
if firstFailure {
|
||||
c.params.ProcessLog.Warn("asking routewatch failed", "error", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// query asks routewatch's IP info API about addr. An answer other than 200, one
|
||||
// that cannot be read, and none within queryTimeout, fail. A result with no
|
||||
// netblock is an address without a live route.
|
||||
func (c *Client) query(ctx context.Context, addr netip.Addr) (Route, Missing, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, queryTimeout)
|
||||
defer cancel()
|
||||
|
||||
endpoint, err := url.JoinPath(c.params.URL, "api/v1/ip", addr.String())
|
||||
if err != nil {
|
||||
return Route{}, "", fmt.Errorf("make the URL: %w", err)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, http.NoBody)
|
||||
if err != nil {
|
||||
return Route{}, "", fmt.Errorf("make the request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
res, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
// Do's error names the URL, which holds the client's address, which
|
||||
// is not to be logged: only what went wrong is kept.
|
||||
return Route{}, "", fmt.Errorf("ask routewatch: %w", errors.Unwrap(err))
|
||||
}
|
||||
|
||||
defer func() {
|
||||
_ = res.Body.Close()
|
||||
}()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
return Route{}, "", fmt.Errorf("%w %s", errStatus, res.Status)
|
||||
}
|
||||
|
||||
return readAnswer(io.LimitReader(res.Body, maxAnswerBytes))
|
||||
}
|
||||
|
||||
// readAnswer reads routewatch's answer about one address.
|
||||
func readAnswer(body io.Reader) (Route, Missing, error) {
|
||||
var decoded struct {
|
||||
Data struct {
|
||||
Results []struct {
|
||||
Netblock string `json:"netblock"`
|
||||
ASN uint32 `json:"asn"`
|
||||
} `json:"results"`
|
||||
} `json:"data"`
|
||||
}
|
||||
|
||||
err := json.NewDecoder(body).Decode(&decoded)
|
||||
if err != nil {
|
||||
return Route{}, "", fmt.Errorf("read the answer: %w", err)
|
||||
}
|
||||
|
||||
if len(decoded.Data.Results) == 0 {
|
||||
return Route{}, "", errNoResult
|
||||
}
|
||||
|
||||
result := decoded.Data.Results[0]
|
||||
if result.Netblock == "" {
|
||||
return Route{}, NoRoute, nil
|
||||
}
|
||||
|
||||
netblock, err := netip.ParsePrefix(result.Netblock)
|
||||
if err != nil {
|
||||
return Route{}, "", fmt.Errorf("%w: %q", errNotNetblock, result.Netblock)
|
||||
}
|
||||
|
||||
return Route{Netblock: netblock.Masked(), ASN: result.ASN}, "", nil
|
||||
}
|
||||
@@ -0,0 +1,290 @@
|
||||
package routewatch_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/netip"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/smallwebwaf/internal/routewatch"
|
||||
)
|
||||
|
||||
const (
|
||||
clientAddr = "192.0.2.1"
|
||||
routedAnswer = `{"status":"ok","data":{"results":[` +
|
||||
`{"ip":"192.0.2.1","netblock":"192.0.2.0/24","asn":64496}]}}`
|
||||
noRouteAnswer = `{"status":"ok","data":{"results":[{"ip":"192.0.2.1"}]}}`
|
||||
waitLimit = 5 * time.Second
|
||||
)
|
||||
|
||||
// clock is a time the tests move by hand.
|
||||
type clock struct {
|
||||
mu sync.Mutex
|
||||
now time.Time
|
||||
}
|
||||
|
||||
func (c *clock) Now() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
return c.now
|
||||
}
|
||||
|
||||
func (c *clock) advance(by time.Duration) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.now = c.now.Add(by)
|
||||
}
|
||||
|
||||
// fixture is a Client asking a stand-in for routewatch. queries counts what the
|
||||
// stand-in was asked, paths holds the last path it was asked for, and log holds
|
||||
// the process log.
|
||||
type fixture struct {
|
||||
t *testing.T
|
||||
client *routewatch.Client
|
||||
clk *clock
|
||||
queries *atomic.Int32
|
||||
path *atomic.Value
|
||||
log *syncBuffer
|
||||
}
|
||||
|
||||
// syncBuffer is a buffer the background queries can log to while a test reads it.
|
||||
type syncBuffer struct {
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *syncBuffer) Write(p []byte) (int, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
return b.buf.Write(p)
|
||||
}
|
||||
|
||||
func (b *syncBuffer) String() string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
return b.buf.String()
|
||||
}
|
||||
|
||||
func newFixture(t *testing.T, standIn http.HandlerFunc) *fixture {
|
||||
t.Helper()
|
||||
|
||||
f := &fixture{
|
||||
t: t,
|
||||
clk: &clock{now: time.Date(2026, 11, 9, 0, 0, 0, 0, time.UTC)},
|
||||
queries: &atomic.Int32{},
|
||||
path: &atomic.Value{},
|
||||
log: &syncBuffer{},
|
||||
}
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, r *http.Request) {
|
||||
f.queries.Add(1)
|
||||
f.path.Store(r.URL.Path)
|
||||
standIn(w, r)
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
f.client = routewatch.New(routewatch.Params{
|
||||
URL: server.URL + "/", Now: f.clk.Now,
|
||||
ProcessLog: slog.New(slog.NewTextHandler(f.log, nil)),
|
||||
})
|
||||
|
||||
return f
|
||||
}
|
||||
|
||||
func answering(body string) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = w.Write([]byte(body))
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fixture) route() (routewatch.Route, routewatch.Missing) {
|
||||
addr := netip.MustParseAddr(clientAddr)
|
||||
client := netip.PrefixFrom(addr, addr.BitLen())
|
||||
|
||||
return f.client.Route(context.Background(), client, addr)
|
||||
}
|
||||
|
||||
// waitFor asks until the answer is why, or, while why is "", until there is a
|
||||
// route, and returns the route.
|
||||
func (f *fixture) waitFor(why routewatch.Missing) routewatch.Route {
|
||||
f.t.Helper()
|
||||
|
||||
deadline := time.Now().Add(waitLimit)
|
||||
|
||||
for {
|
||||
route, missing := f.route()
|
||||
if missing == why {
|
||||
return route
|
||||
}
|
||||
|
||||
if time.Now().After(deadline) {
|
||||
f.t.Fatalf("still %q after %s, want %q", missing, waitLimit, why)
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouteGivesRoutewatchsAnswerAndAsksByAddress(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newFixture(t, answering(routedAnswer))
|
||||
|
||||
if _, missing := f.route(); missing != routewatch.NoAnswerYet {
|
||||
t.Errorf("before the answer: %q, want %q", missing, routewatch.NoAnswerYet)
|
||||
}
|
||||
|
||||
route := f.waitFor("")
|
||||
|
||||
want := routewatch.Route{Netblock: netip.MustParsePrefix("192.0.2.0/24"), ASN: 64496}
|
||||
if route != want {
|
||||
t.Errorf("route %+v, want %+v", route, want)
|
||||
}
|
||||
|
||||
if got := f.path.Load(); got != "/api/v1/ip/"+clientAddr {
|
||||
t.Errorf("asked for path %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddressWithoutARouteIsKeptAsNoRoute(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newFixture(t, answering(noRouteAnswer))
|
||||
|
||||
f.route()
|
||||
f.waitFor(routewatch.NoRoute)
|
||||
|
||||
if got := f.queries.Load(); got != 1 {
|
||||
t.Errorf("%d queries, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnswerRoutewatchCannotBeTakenFromFails(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := map[string]http.HandlerFunc{
|
||||
"500": func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
},
|
||||
"not JSON": answering("no"),
|
||||
"no results": answering(`{"status":"ok","data":{"results":[]}}`),
|
||||
"bad netblock": answering(
|
||||
`{"status":"ok","data":{"results":[{"netblock":"nowhere","asn":1}]}}`),
|
||||
}
|
||||
for name, standIn := range tests {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newFixture(t, standIn)
|
||||
|
||||
f.route()
|
||||
f.waitFor(routewatch.RequestFailed)
|
||||
|
||||
// The failure is noted before it is logged.
|
||||
deadline := time.Now().Add(waitLimit)
|
||||
for !strings.Contains(f.log.String(), "asking routewatch failed") {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("process log %q names no failure", f.log.String())
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
|
||||
if strings.Contains(f.log.String(), clientAddr) {
|
||||
t.Errorf("process log %q names the client", f.log.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFailureIsKeptForAMinuteSoRoutewatchIsNotAskedPerRequest(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newFixture(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
})
|
||||
|
||||
f.route()
|
||||
f.waitFor(routewatch.RequestFailed)
|
||||
f.route()
|
||||
|
||||
if got := f.queries.Load(); got != 1 {
|
||||
t.Errorf("%d queries while the failure is kept, want 1", got)
|
||||
}
|
||||
|
||||
f.clk.advance(time.Minute)
|
||||
f.route()
|
||||
|
||||
deadline := time.Now().Add(waitLimit)
|
||||
for f.queries.Load() != 2 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("%d queries after the failure expired, want 2", f.queries.Load())
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHangingRoutewatchNeverDelaysAndFailsTheQuery(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
f := newFixture(t, func(_ http.ResponseWriter, r *http.Request) {
|
||||
<-r.Context().Done()
|
||||
})
|
||||
|
||||
start := time.Now()
|
||||
|
||||
f.route()
|
||||
|
||||
if elapsed := time.Since(start); elapsed > time.Second {
|
||||
t.Errorf("Route took %s while routewatch hangs", elapsed)
|
||||
}
|
||||
|
||||
f.waitFor(routewatch.RequestFailed)
|
||||
}
|
||||
|
||||
func TestRepeatedRequestsMakeOneQueryAndTheAnswerExpires(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
release := make(chan struct{})
|
||||
f := newFixture(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
<-release
|
||||
|
||||
_, _ = w.Write([]byte(routedAnswer))
|
||||
})
|
||||
|
||||
for range 50 {
|
||||
f.route()
|
||||
}
|
||||
|
||||
close(release)
|
||||
f.waitFor("")
|
||||
|
||||
for range 50 {
|
||||
f.route()
|
||||
}
|
||||
|
||||
if got := f.queries.Load(); got != 1 {
|
||||
t.Errorf("%d queries for repeated requests, want 1", got)
|
||||
}
|
||||
|
||||
f.clk.advance(time.Hour)
|
||||
f.waitFor(routewatch.NoAnswerYet)
|
||||
f.waitFor("")
|
||||
|
||||
if got := f.queries.Load(); got != 2 {
|
||||
t.Errorf("%d queries after the answer expired, want 2", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user