Look clients up in the IPinfo Lite file with SWWAF_LOOKUP_SOURCE=file (closes #22)
check / check (push) Waiting to run

SWWAF_LOOKUP_SOURCE=file looks every client up in the file
SWWAF_LOOKUP_DB_PATH names, without GeoJS. file without the path, the
path with another source, or a file that cannot be read stops the start.
The file is read whole into memory, so overwriting it in place cannot
disturb a lookup, and read again 2 seconds after its last change; a
replacement that cannot be read is logged, counted and sent as a
file_error alert, and the old one stays in use. Metrics give when it
was read and the failed reads. Tests write their databases through
internal/lookup/lookuptest.

Deviation: go.mod and go.sum written by hand; go runs only through make.
Judgement call: the 2-second wait, as the rule files have.

Model: opus-5-5
This commit was merged in pull request #98.
This commit is contained in:
2026-10-07 10:04:10 +02:00
parent 26f4abef7f
commit c80753c56e
21 changed files with 1291 additions and 168 deletions
+233
View File
@@ -0,0 +1,233 @@
package lookup
import (
"context"
"fmt"
"log/slog"
"net/netip"
"os"
"path/filepath"
"sync"
"time"
"github.com/fsnotify/fsnotify"
"github.com/oschwald/maxminddb-golang/v2"
"sneak.berlin/go/smallwebwaf/internal/alerts"
)
// quietTime is how long the lookup database must go without a change
// before it is read again, so that a file still being copied in is read
// only once whole.
const quietTime = 2 * time.Second
// FileParams are what OpenFile needs.
type FileParams struct {
// Path is the lookup database, the IPinfo Lite file in its .mmdb form
// (SWWAF_LOOKUP_DB_PATH).
Path string
// Now tells the time, normally time.Now.
Now func() time.Time
// ProcessLog receives each reading of the file, and why a replacement
// of it cannot be read.
ProcessLog *slog.Logger
// Alerts receive a file_error alert for each replacement that cannot
// be read.
Alerts *alerts.Queue
}
// File looks up clients' AS numbers and countries in the lookup database,
// held in memory, and reads it again when it is replaced. It is safe for
// concurrent use.
type File struct {
params FileParams
mu sync.Mutex
// reader is the database in use, and lastRead when it was read.
// readFailures are the replacements that could not be read.
reader *maxminddb.Reader
lastRead time.Time
readFailures int
}
// record is what the lookup database holds about a network, of the fields
// smallwebwaf reads.
type record struct {
ASN string `maxminddb:"asn"`
ASName string `maxminddb:"as_name"`
CountryCode string `maxminddb:"country_code"`
}
// OpenFile reads the lookup database. A file that cannot be read, or that
// is not a .mmdb file, is an error.
func OpenFile(params FileParams) (*File, error) {
reader, err := read(params.Path)
if err != nil {
return nil, err
}
f := &File{params: params}
f.use(reader)
return f, nil
}
// LookUp returns what the lookup database says about client: its AS
// number, such as AS64496, the AS's name, and its country, such as DE,
// each "" when the database does not give it, as for an address missing
// from it. The database is asked about the client's first address, as
// GeoJS is.
func (f *File) LookUp(client netip.Prefix) Answer {
f.mu.Lock()
reader := f.reader
f.mu.Unlock()
var found record
// A record that cannot be decoded places the client nowhere, as a
// missing one does.
err := reader.Lookup(client.Addr()).Decode(&found)
if err != nil {
found = record{}
}
return Answer{
Client: client,
ASN: found.ASN,
ASName: found.ASName,
Country: found.CountryCode,
Answered: f.params.Now(),
}
}
// LastRead returns when the lookup database in use was read.
func (f *File) LastRead() time.Time {
f.mu.Lock()
defer f.mu.Unlock()
return f.lastRead
}
// ReadFailures returns how many replacements of the lookup database could
// not be read.
func (f *File) ReadFailures() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.readFailures
}
// Watch watches the directory of the lookup database until ctx is done,
// and reads the file again once it has gone without a change for
// quietTime, after it is replaced, written or removed, and after Watch
// starts watching. If the directory cannot be watched, that is logged, and
// the database read at start stays in use.
func (f *File) Watch(ctx context.Context) {
watcher, err := fsnotify.NewWatcher()
if err == nil {
defer func() {
_ = watcher.Close()
}()
err = watcher.Add(filepath.Dir(f.params.Path))
}
if err != nil {
f.params.ProcessLog.Error("cannot watch the lookup database for replacements",
"error", err.Error())
return
}
f.params.ProcessLog.Info("watching the lookup database for replacements",
"file", f.params.Path)
f.readAfterChanges(ctx, watcher.Events, watcher.Errors)
}
// readAfterChanges reads the lookup database again once quietTime has
// passed without a change to it from events, until ctx is done, and logs
// the errors from errs. A change to another file in its directory does not
// count. The wait starts at once, as if for a change, so that a file
// replaced after OpenFile read it, and before its directory was watched,
// is read too.
func (f *File) readAfterChanges(
ctx context.Context, events <-chan fsnotify.Event, errs <-chan error,
) {
path := filepath.Clean(f.params.Path)
quiet := time.NewTimer(quietTime)
defer quiet.Stop()
for {
select {
case <-ctx.Done():
return
case event := <-events:
if filepath.Clean(event.Name) == path {
quiet.Reset(quietTime)
}
case <-quiet.C:
f.readAgain()
case err := <-errs:
f.params.ProcessLog.Warn("watching the lookup database failed",
"error", err.Error())
}
}
}
// readAgain reads the lookup database again, in place of the one in use,
// or, if it cannot be read, counts that, raises a file_error alert for it
// and logs it, and the one in use stays in use.
func (f *File) readAgain() {
reader, err := read(f.params.Path)
if err != nil {
const kept = "the lookup database cannot be read, " +
"and the one read before stays in use"
f.mu.Lock()
f.readFailures++
f.mu.Unlock()
// Raised before it is logged, so that the alert is there once the
// log line is.
f.params.Alerts.Raise(alerts.Alert{
Event: alerts.EventFileError,
Reason: kept,
Detail: map[string]any{"file": f.params.Path, "error": err.Error()},
})
f.params.ProcessLog.Error(kept, "error", err.Error())
return
}
f.use(reader)
}
// use puts reader in use, in place of the database read before, and logs
// that the file was read.
func (f *File) use(reader *maxminddb.Reader) {
f.mu.Lock()
f.reader = reader
f.lastRead = f.params.Now()
f.mu.Unlock()
f.params.ProcessLog.Info("read the lookup database", "file", f.params.Path)
}
// read reads the lookup database at path. The whole file is read into
// memory, rather than mapped into it as the reader can, so that a file
// overwritten in place cannot change, or end, under a lookup.
func read(path string) (*maxminddb.Reader, error) {
data, err := os.ReadFile(path) //nolint:gosec // the file the admin names
if err != nil {
return nil, fmt.Errorf("SWWAF_LOOKUP_DB_PATH cannot be read: %w", err)
}
reader, err := maxminddb.OpenBytes(data)
if err != nil {
return nil, fmt.Errorf("SWWAF_LOOKUP_DB_PATH %s is not a .mmdb file: %w", path, err)
}
return reader, nil
}
+379
View File
@@ -0,0 +1,379 @@
package lookup
import (
"context"
"log/slog"
"net/netip"
"net/url"
"os"
"path/filepath"
"reflect"
"testing"
"testing/synctest"
"time"
"github.com/fsnotify/fsnotify"
"github.com/maxmind/mmdbwriter/mmdbtype"
"sneak.berlin/go/smallwebwaf/internal/alerts"
"sneak.berlin/go/smallwebwaf/internal/lookup/lookuptest"
)
// testNetblock is the netblock the tests' lookup databases place, and
// testClient a client in it.
const (
testNetblock = "203.0.113.0/24"
testClient = "203.0.113.9/32"
)
func TestFilePlacesClientsAndCountsAnAddressMissingFromItAsUnknown(t *testing.T) {
t.Parallel()
germany := lookuptest.Network{ASN: "AS64496", ASName: "Example Net", Country: "DE"}
northKorea := lookuptest.Network{ASN: "AS64511", ASName: "Other Net", Country: "KP"}
path := filepath.Join(t.TempDir(), "ipinfo_lite.mmdb")
lookuptest.Write(t, path, map[string]lookuptest.Network{
testNetblock: germany,
"2001:db8::/32": northKorea,
})
now := time.Date(2026, 10, 7, 0, 0, 0, 0, time.UTC)
f, err := OpenFile(FileParams{
Path: path,
Now: func() time.Time { return now },
ProcessLog: slog.New(slog.DiscardHandler),
Alerts: newQueue(),
})
if err != nil {
t.Fatalf("open %s: %v", path, err)
}
for client, want := range map[string]lookuptest.Network{
testClient: germany,
// An IPv6 client is its /64.
"2001:db8:1:2::/64": northKorea,
"198.51.100.7/32": {},
} {
prefix := netip.MustParsePrefix(client)
got := f.LookUp(prefix)
if got != (Answer{
Client: prefix, ASN: want.ASN, ASName: want.ASName, Country: want.Country,
Answered: now,
}) {
t.Errorf("%s has the answer %+v, want %+v, answered %s", client, got, want, now)
}
}
}
func TestRecordThatCannotBeReadPlacesTheClientNowhere(t *testing.T) {
t.Parallel()
// The AS number is a number, where a string belongs. The writer writes
// a record's fields in the order of their names, so as_name is read
// before the AS number fails.
path := filepath.Join(t.TempDir(), "ipinfo_lite.mmdb")
lookuptest.WriteRecords(t, path, map[string]mmdbtype.Map{
testNetblock: {
"asn": mmdbtype.Uint32(64496),
"as_name": mmdbtype.String("Example Net"),
"country_code": mmdbtype.String("DE"),
},
})
f := openFile(t, path, newQueue())
answer := f.LookUp(netip.MustParsePrefix(testClient))
if answer.ASN != "" || answer.ASName != "" || answer.Country != "" {
t.Errorf("%s is placed %+v, want nowhere", testClient, answer)
}
}
func TestFileThatCannotBeReadIsAnError(t *testing.T) {
t.Parallel()
dir := t.TempDir()
missing := filepath.Join(dir, "missing.mmdb")
notDatabase := filepath.Join(dir, "not.mmdb")
writeFile(t, notDatabase, "not a lookup database\n")
for path, want := range map[string]string{
missing: "SWWAF_LOOKUP_DB_PATH cannot be read: open " + missing +
": no such file or directory",
notDatabase: "SWWAF_LOOKUP_DB_PATH " + notDatabase +
" is not a .mmdb file: error opening database: invalid MaxMind DB file",
} {
_, err := OpenFile(FileParams{
Path: path,
Now: time.Now,
ProcessLog: slog.New(slog.DiscardHandler),
Alerts: newQueue(),
})
if err == nil || err.Error() != want {
t.Errorf("opening %s failed with %v, want %s", path, err, want)
}
}
}
// The tests below run readAfterChanges in a synctest bubble, where time is
// a clock of the test's own: time.Sleep moves it on at once, and
// synctest.Wait returns once readAfterChanges waits again, so that every
// reading due by then is done. The test sends the changes itself, as the
// watch of a directory cannot run in a bubble.
func TestReplacementCopiedOverTheFileInTwoPartsIsReadOnlyWhole(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "ipinfo_lite.mmdb")
writeDatabase(t, path, "DE")
queue := newQueue()
f := openFile(t, path, queue)
changes := watch(t, f)
other := filepath.Join(dir, "replacement.mmdb")
writeDatabase(t, other, "KP")
replacement, err := os.ReadFile(other) //nolint:gosec // a file the test wrote
if err != nil {
t.Fatalf("read %s: %v", other, err)
}
// The file in use is overwritten in place, and keeps giving what
// it gave. Its first part alone is not a .mmdb file.
file, err := os.Create(path) //nolint:gosec // a file the test wrote
if err != nil {
t.Fatalf("create %s: %v", path, err)
}
defer func() {
_ = file.Close()
}()
half := len(replacement) / 2
write(t, file, replacement[:half])
changes <- fsnotify.Event{Name: path, Op: fsnotify.Write}
time.Sleep(quietTime - time.Nanosecond)
synctest.Wait()
wantCountry(t, f, "DE")
// The second part starts the wait again.
write(t, file, replacement[half:])
changes <- fsnotify.Event{Name: path, Op: fsnotify.Write}
time.Sleep(quietTime - time.Nanosecond)
synctest.Wait()
wantCountry(t, f, "DE")
time.Sleep(time.Nanosecond)
synctest.Wait()
wantCountry(t, f, "KP")
if !f.LastRead().Equal(time.Now()) || f.ReadFailures() != 0 {
t.Errorf("read at %s, with %d failures; want read now, with none",
f.LastRead(), f.ReadFailures())
}
wantAlerts(t, queue)
})
}
func TestReplacementThatCannotBeReadLeavesTheFileInUseWithOneAlert(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
path := filepath.Join(t.TempDir(), "ipinfo_lite.mmdb")
writeDatabase(t, path, "DE")
queue := newQueue()
f := openFile(t, path, queue)
read := f.LastRead()
changes := watch(t, f)
writeFile(t, path, "not a lookup database\n")
changes <- fsnotify.Event{Name: path, Op: fsnotify.Write}
// Long after, the replacement has been read once.
time.Sleep(time.Hour)
synctest.Wait()
wantCountry(t, f, "DE")
if !f.LastRead().Equal(read) || f.ReadFailures() != 1 {
t.Errorf("read at %s, with %d failures; want read at %s, with one",
f.LastRead(), f.ReadFailures(), read)
}
wantAlerts(t, queue, alerts.Alert{
Time: read.Add(quietTime),
Event: alerts.EventFileError,
Reason: "the lookup database cannot be read, and the one read before stays in use",
Detail: map[string]any{
"file": path,
"error": "SWWAF_LOOKUP_DB_PATH " + path + " is not a .mmdb file: " +
"error opening database: invalid MaxMind DB file",
},
})
})
}
func TestChangeOfAnotherFileInTheDirectoryIsNoReplacement(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "ipinfo_lite.mmdb")
writeDatabase(t, path, "DE")
f := openFile(t, path, newQueue())
changes := watch(t, f)
// The wait that starts with the watch ends with a reading.
time.Sleep(quietTime)
synctest.Wait()
writeDatabase(t, path, "KP")
changes <- fsnotify.Event{Name: filepath.Join(dir, "other.mmdb"), Op: fsnotify.Create}
time.Sleep(quietTime)
synctest.Wait()
wantCountry(t, f, "DE")
changes <- fsnotify.Event{Name: path, Op: fsnotify.Write}
time.Sleep(quietTime)
synctest.Wait()
wantCountry(t, f, "KP")
})
}
func TestReplacementSavedBeforeTheWatchStartsIsRead(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
path := filepath.Join(t.TempDir(), "ipinfo_lite.mmdb")
writeDatabase(t, path, "DE")
f := openFile(t, path, newQueue())
// Saved after OpenFile read the file, and before its directory was
// watched, so that no change is seen for it.
writeDatabase(t, path, "KP")
watch(t, f)
time.Sleep(quietTime)
synctest.Wait()
wantCountry(t, f, "KP")
})
}
// newQueue returns a queue of alerts for a webhook that is never sent
// them, so that they wait in it for the test to look at.
func newQueue() *alerts.Queue {
return alerts.New(alerts.Params{
WebhookURL: &url.URL{Scheme: "https", Host: "alerts.example"},
Events: alerts.Events(),
Cooldown: 15 * time.Minute,
Now: time.Now,
})
}
// writeDatabase writes a lookup database at path that places testNetblock
// in country, and no other address.
func writeDatabase(t *testing.T, path, country string) {
t.Helper()
lookuptest.Write(t, path, map[string]lookuptest.Network{
testNetblock: {ASN: "AS64496", ASName: "Example Net", Country: country},
})
}
// openFile opens the lookup database at path, which raises its alerts to
// queue.
func openFile(t *testing.T, path string, queue *alerts.Queue) *File {
t.Helper()
f, err := OpenFile(FileParams{
Path: path,
Now: time.Now,
ProcessLog: slog.New(slog.DiscardHandler),
Alerts: queue,
})
if err != nil {
t.Fatalf("open %s: %v", path, err)
}
return f
}
// watch runs f's readAfterChanges until the test ends, and returns the
// channel that sends it changes.
func watch(t *testing.T, f *File) chan<- fsnotify.Event {
t.Helper()
changes := make(chan fsnotify.Event)
ctx, stop := context.WithCancel(t.Context())
stopped := make(chan struct{})
go func() {
f.readAfterChanges(ctx, changes, nil)
close(stopped)
}()
t.Cleanup(func() {
stop()
<-stopped
})
return changes
}
// wantCountry checks the country f gives testClient.
func wantCountry(t *testing.T, f *File, want string) {
t.Helper()
got := f.LookUp(netip.MustParsePrefix(testClient)).Country
if got != want {
t.Errorf("%s is in %q, want %q", testClient, got, want)
}
}
// wantAlerts checks the alerts waiting in queue, and that it held none
// back.
func wantAlerts(t *testing.T, queue *alerts.Queue, want ...alerts.Alert) {
t.Helper()
waiting := queue.Snapshot().Waiting[alerts.DestinationWebhook]
if len(waiting) != len(want) || (len(want) > 0 && !reflect.DeepEqual(waiting, want)) {
t.Errorf("alerts waiting %+v, want %+v", waiting, want)
}
if queue.Suppressed() != 0 {
t.Errorf("%d alerts held back, want none", queue.Suppressed())
}
}
// writeFile writes content to the file at path.
func writeFile(t *testing.T, path, content string) {
t.Helper()
err := os.WriteFile(path, []byte(content), 0o600)
if err != nil {
t.Fatalf("write %s: %v", path, err)
}
}
// write writes data to the end of file.
func write(t *testing.T, file *os.File, data []byte) {
t.Helper()
_, err := file.Write(data)
if err != nil {
t.Fatalf("write: %v", err)
}
}
+10 -8
View File
@@ -1,6 +1,7 @@
// Package lookup looks up each client's AS number and country through
// the GeoJS web service, and keeps the answers in memory, for at most
// 100,000 clients and for 7 days each. The answers are written to
// Package lookup looks up each client's AS number and country, through
// the GeoJS web service or in the lookup database, the IPinfo Lite file
// SWWAF_LOOKUP_DB_PATH names. GeoJS's answers are kept in memory, for at
// most 100,000 clients and for 7 days each, and are written to
// lookups.json and read from it by the state package.
package lookup
@@ -113,11 +114,12 @@ type GeoJS struct {
retryAt time.Time
}
// Answer is what GeoJS said about a client, as lookups.json holds it: its
// AS number, such as AS64496, and the AS's name, both "" when GeoJS knows
// no AS number for it; its country, "" when GeoJS cannot place it; when
// GeoJS said so, and when the answer was last used. The zero Answer is
// that of a client with no answer.
// Answer is what GeoJS or the lookup database said about a client: its AS
// number, such as AS64496, and the AS's name, both "" when the source knows
// no AS number for it; its country, "" when the source cannot place it;
// when the source said so; and, for GeoJS's answers, which lookups.json
// holds, when the answer was last used. The zero Answer is that of a
// client with no answer.
//
//nolint:tagliatelle // the state files use snake_case, as the request log does
type Answer struct {
+83
View File
@@ -0,0 +1,83 @@
// Package lookuptest writes lookup databases, IPinfo Lite files in their
// .mmdb form, for the tests of the packages that read them.
package lookuptest
import (
"bytes"
"net"
"os"
"testing"
"github.com/maxmind/mmdbwriter"
"github.com/maxmind/mmdbwriter/mmdbtype"
)
// fileMode is the mode of the files written: read and written by their
// owner alone.
const fileMode = 0o600
// Network is what a lookup database holds about a netblock, of the fields
// smallwebwaf reads: its AS number, such as AS64496, the AS's name, and
// its country, such as DE.
type Network struct {
ASN string
ASName string
Country string
}
// Write writes a lookup database at path that places each netblock in
// networks, such as 203.0.113.0/24, as its Network says, and no other
// address.
func Write(tb testing.TB, path string, networks map[string]Network) {
tb.Helper()
records := make(map[string]mmdbtype.Map, len(networks))
for netblock, network := range networks {
records[netblock] = mmdbtype.Map{
"asn": mmdbtype.String(network.ASN),
"as_name": mmdbtype.String(network.ASName),
"country_code": mmdbtype.String(network.Country),
}
}
WriteRecords(tb, path, records)
}
// WriteRecords writes a lookup database at path that holds each record in
// records for its netblock, and nothing for any other address.
func WriteRecords(tb testing.TB, path string, records map[string]mmdbtype.Map) {
tb.Helper()
tree, err := mmdbwriter.New(mmdbwriter.Options{
DatabaseType: "ipinfo_lite",
// The tests' clients are in the netblocks kept for documentation.
IncludeReservedNetworks: true,
})
if err != nil {
tb.Fatalf("new lookup database: %v", err)
}
for netblock, record := range records {
_, network, err := net.ParseCIDR(netblock)
if err != nil {
tb.Fatalf("netblock %q: %v", netblock, err)
}
err = tree.Insert(network, record)
if err != nil {
tb.Fatalf("insert %s: %v", netblock, err)
}
}
var database bytes.Buffer
_, err = tree.WriteTo(&database)
if err != nil {
tb.Fatalf("write the lookup database: %v", err)
}
err = os.WriteFile(path, database.Bytes(), fileMode)
if err != nil {
tb.Fatalf("write %s: %v", path, err)
}
}