smallwebwaf/internal/lookup/file.go

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
	// (SWW_LOOKUP_DB_PATH).
	Path string
	// Now tells the time, normally time.Now in UTC.
	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
// sww 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("SWW_LOOKUP_DB_PATH cannot be read: %w", err)
	}

	reader, err := maxminddb.OpenBytes(data)
	if err != nil {
		return nil, fmt.Errorf("SWW_LOOKUP_DB_PATH %s is not a .mmdb file: %w", path, err)
	}

	return reader, nil
}