Take in an admin's edits of the state files while running (closes #68)

smallwebwaf watches SWWAF_STATE_DIR with fsnotify and takes in a saved edit of a state file in place of what it held. It tells its own writes from an admin's by the SHA-256 of what it last read or wrote; each write first takes in an edit made since. An edit that does not parse is renamed to <name>.bad at the file's next write. Every ban on a netblock is checked, and the next ban is worked out from the one that ended last. Two metrics count the edits taken in and set aside. README.md says how to add and lift a ban. Judgement call: a broken edit is set aside at the next write, since an editor's file can be read half written. Model: opus-5-5
2026-10-06 09:48:06 +00:00
parent 234c5eac60
commit 34ebf1abb9
13 changed files with 995 additions and 176 deletions
@@ -3,7 +3,9 @@ package state_test
 import (
 	"context"
 	"encoding/json"
+	"io/fs"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/netip"
@@ -28,6 +30,13 @@ const (
 	bansJSON    = "bans.json"
 	clientsJSON = "clients.json"
 	lookupsJSON = "lookups.json"
+	// What the process log says once Watch watches the directory, and as
+	// it takes in an edit.
+	watching = "watching the state files for edits"
+	tookIn   = "took in an edit of a state file"
+	// maxLogLines is how many lines of the process log wait for a test to
+	// read them.
+	maxLogLines = 64
 )

 // permanentBansJSON is bans.json holding permanentBan.
@@ -290,7 +299,7 @@ func TestUnwritableDirectoryStopsTheStart(t *testing.T) {
 	}
 }

-// The two tests below run Run in a synctest bubble, where time is a clock
+// The three tests below run Run 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 Run waits for its next write, so that every write due by
 // then is on disk.
@@ -302,7 +311,7 @@ func TestBansWrittenOnceWriteDelayAfterABan(t *testing.T) {
 		dir := t.TempDir()
 		params := newParams(dir)
 		params.WriteDelay = 10 * time.Second
-		run(t, load(t, params))
+		run(t, load(t, params).Run)

 		// A second ban, made while the first waits to be written, puts the
 		// write off no further, and is written with it.
@@ -347,7 +356,7 @@ func TestEveryFileWrittenEveryCounterInterval(t *testing.T) {
 		dir := t.TempDir()
 		params := newParams(dir)
 		params.CounterInterval = time.Minute
-		run(t, load(t, params))
+		run(t, load(t, params).Run)

 		// The files are removed once written, so that each interval shows
 		// them written again.
@@ -364,6 +373,44 @@ func TestEveryFileWrittenEveryCounterInterval(t *testing.T) {
 	})
 }

+func TestEditJustBeforeAScheduledWriteSurvivesIt(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		dir := t.TempDir()
+		params := newParams(dir)
+		params.WriteDelay = 10 * time.Second
+		run(t, load(t, params).Run)
+
+		// A ban, and bans.json written with it.
+		params.Ledger.BanForLimit(netip.MustParsePrefix("203.0.113.9/32"),
+			midnight(), bans.Notes{})
+		time.Sleep(params.WriteDelay)
+		synctest.Wait()
+
+		// A second ban is to be written WriteDelay later. Just before
+		// then, an admin saves bans.json with the first ban lifted and
+		// another added.
+		params.Ledger.BanForLimit(netip.MustParsePrefix("203.0.113.10/32"),
+			midnight(), bans.Notes{})
+		time.Sleep(params.WriteDelay - time.Nanosecond)
+		synctest.Wait()
+		edit(t, dir, bansJSON, permanentBansJSON)
+
+		// The write takes the edit in first, and writes it back. The second
+		// ban, made after the admin opened the file, is lost, as "Edits
+		// while running" in SPEC.md says.
+		time.Sleep(time.Nanosecond)
+		synctest.Wait()
+
+		if got := readFile(t, filepath.Join(dir, bansJSON)); got != permanentBansJSON {
+			t.Errorf("bans.json holds\n%s\nwant the edit", got)
+		}
+
+		wantEqual(t, bansJSON, params.Ledger.Snapshot(), []bans.Ban{permanentBan()})
+	})
+}
+
 func TestFailedWriteLeavesTheFileAsItWas(t *testing.T) {
 	t.Parallel()

@@ -459,26 +506,248 @@ func TestWritesAreCountedInTheMetrics(t *testing.T) {
 		float64(len(permanentBansJSON)))
 }

-func TestFailedRenameLeavesNoTemporaryFile(t *testing.T) {
+func TestFileThatCannotBeReadIsNotWrittenOver(t *testing.T) {
 	t.Parallel()

 	dir := t.TempDir()
+	path := filepath.Join(dir, bansJSON)
 	files := load(t, newParams(dir))

-	// A directory named bans.json cannot be renamed over.
-	err := os.Mkdir(filepath.Join(dir, bansJSON), 0o700)
+	// bans.json is a socket, which cannot be opened as a file, even by
+	// root, as the tests run in Docker, but which a rename could replace.
+	// Whether it holds an edit cannot be told, so it is left as it is.
+	socket, err := (&net.ListenConfig{}).Listen(t.Context(), "unix", path)
 	if err != nil {
-		t.Fatalf("mkdir: %v", err)
+		t.Fatalf("listen: %v", err)
 	}

+	defer func() {
+		_ = socket.Close()
+	}()
+
 	err = files.WriteAll()
 	if err == nil {
-		t.Error("writing over a directory did not fail")
+		t.Error("writing with bans.json unreadable did not fail")
+	}
+
+	info, err := os.Lstat(path)
+	if err != nil || info.Mode().Type() != fs.ModeSocket {
+		t.Errorf("bans.json is now %v (%v), want the socket", info, err)
 	}

 	wantFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
 }

+func TestEditOfEachFileTakenIn(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	params := newParams(dir)
+	lines := logInto(&params)
+	fill(params)
+	files := load(t, params)
+
+	err := files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	watch(t, files, lines)
+
+	// Each edit holds one entry, for a client the parts did not hold, and
+	// takes the place of everything the part held.
+	client := netip.MustParsePrefix("198.51.100.7/32")
+
+	edit(t, dir, bansJSON, `{"version": 1, "bans": [{"netblock": "198.51.100.7/32", `+
+		`"start": "2026-10-06T00:00:00Z", "expires": null}]}`)
+	wantTakenIn(t, lines, dir, bansJSON)
+	wantEqual(t, bansJSON, params.Ledger.Snapshot(),
+		[]bans.Ban{{Netblock: client, Start: midnight()}})
+
+	edit(t, dir, clientsJSON, `{"version": 1, "clients": [`+
+		`{"client": "198.51.100.7/32", "history": {"requests": 7}}]}`)
+	wantTakenIn(t, lines, dir, clientsJSON)
+	wantEqual(t, clientsJSON, params.Limiter.Snapshot(),
+		[]ratelimit.Client{{Client: client, History: ratelimit.History{Requests: 7}}})
+
+	edit(t, dir, lookupsJSON, `{"version": 1, "lookups": [{"client": "198.51.100.7/32", `+
+		`"country": "FR", "answered": "2026-10-06T00:00:00Z"}]}`)
+	wantTakenIn(t, lines, dir, lookupsJSON)
+	wantEqual(t, lookupsJSON, params.GeoJS.Snapshot(),
+		[]lookup.Answer{{Client: client, Country: "FR", Answered: midnight()}})
+}
+
+func TestOwnWritesAreNotTakenIn(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	params := newParams(dir)
+	lines := logInto(&params)
+	fill(params)
+	files := load(t, params)
+	watch(t, files, lines)
+
+	// Every file is written while watched, and then lookups.json edited:
+	// the first edit taken in is that one.
+	err := files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	edit(t, dir, lookupsJSON, `{"version": 1, "lookups": []}`)
+	wantTakenIn(t, lines, dir, lookupsJSON)
+}
+
+func TestBanAddedAndLiftedThroughBansJSON(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	params := newParams(dir)
+	lines := logInto(&params)
+	watch(t, load(t, params), lines)
+
+	client := netip.MustParseAddr("203.0.113.9")
+
+	// An entry added, as an admin writes it, bans its netblock.
+	edit(t, dir, bansJSON, `{"version": 1, "bans": [{"netblock": "203.0.113.0/24", `+
+		`"start": "2026-10-06T00:00:00Z", "expires": null}]}`)
+	wantTakenIn(t, lines, dir, bansJSON)
+
+	_, banned := params.Ledger.Check(client, midnight())
+	if !banned {
+		t.Error("the ban added to bans.json does not refuse")
+	}
+
+	// The entry removed lifts the ban.
+	edit(t, dir, bansJSON, `{"version": 1, "bans": []}`)
+	wantTakenIn(t, lines, dir, bansJSON)
+
+	_, banned = params.Ledger.Check(client, midnight())
+	if banned {
+		t.Error("the ban removed from bans.json still refuses")
+	}
+}
+
+func TestBrokenEditSetAsideAtTheNextWrite(t *testing.T) {
+	t.Parallel()
+
+	// It ends a ban's entry with a comma.
+	const broken = "{\n    \"version\": 1,\n    \"bans\": [\n" +
+		"        {\"netblock\": \"203.0.113.9/32\",}\n    ]\n}\n"
+
+	dir := t.TempDir()
+	path := filepath.Join(dir, bansJSON)
+	params := newParams(dir)
+	lines := logInto(&params)
+	params.Ledger.Load([]bans.Ban{permanentBan()})
+	files := load(t, params)
+
+	err := files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	watch(t, files, lines)
+
+	// While smallwebwaf runs, the broken edit is left as it is: an edit
+	// of clients.json, made after it and taken in, shows that it has been
+	// seen.
+	edit(t, dir, bansJSON, broken)
+	edit(t, dir, clientsJSON, `{"version": 1, "clients": []}`)
+	wantTakenIn(t, lines, dir, clientsJSON)
+	wantFiles(t, dir, bansJSON, clientsJSON, lookupsJSON)
+
+	// The next write sets it aside, logged with where the error is, and
+	// writes bans.json again from what smallwebwaf still holds.
+	err = files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	line := lines.waitFor(t, "set aside an edit of a state file that does not parse")
+	message, _ := line["error"].(string)
+
+	if line["file"] != path+".bad" ||
+		!strings.HasPrefix(message, path+", line 4, column 39: ") {
+		t.Errorf("set aside with %v", line)
+	}
+
+	wantFiles(t, dir, bansJSON, bansJSON+".bad", clientsJSON, lookupsJSON)
+
+	if got := readFile(t, path+".bad"); got != broken {
+		t.Errorf("bans.json.bad holds\n%s\nwant the edit", got)
+	}
+
+	if got := readFile(t, path); got != permanentBansJSON {
+		t.Errorf("bans.json holds\n%s\nwant\n%s", got, permanentBansJSON)
+	}
+}
+
+func TestEditsTakenInAreCountedInTheMetrics(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	params := newParams(dir)
+	lines := logInto(&params)
+	files := load(t, params)
+
+	// One edit is taken in by the write of its file, before Watch runs,
+	// and one by Watch.
+	edit(t, dir, bansJSON, `{"version": 1, "bans": []}`)
+
+	err := files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	watch(t, files, lines)
+	edit(t, dir, bansJSON, permanentBansJSON)
+	wantTakenIn(t, lines, dir, bansJSON)
+
+	wantMetric(t, scrape(t, params),
+		`smallwebwaf_state_file_edits_taken_in_total{file="bans.json"}`, 2)
+}
+
+func TestEditsSetAsideAreCountedInTheMetrics(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	params := newParams(dir)
+	files := load(t, params)
+
+	edit(t, dir, bansJSON, `{"version": 1, "bans": [`)
+
+	err := files.WriteAll()
+	if err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	wantMetric(t, scrape(t, params),
+		`smallwebwaf_state_file_edits_set_aside_total{file="bans.json"}`, 1)
+}
+
+func TestDirectoryThatCannotBeWatchedIsLogged(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	params := newParams(dir)
+	lines := logInto(&params)
+	files := load(t, params)
+
+	err := os.Remove(dir)
+	if err != nil {
+		t.Fatalf("remove: %v", err)
+	}
+
+	// Watch returns at once.
+	files.Watch(t.Context())
+
+	line := lines.waitFor(t, "cannot watch the state files for edits")
+	if line["level"] != "ERROR" {
+		t.Errorf("logged as %v", line)
+	}
+}
+
 // midnight is the time of the tests' clock.
 func midnight() time.Time {
 	return time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)
@@ -573,15 +842,16 @@ func load(t *testing.T, params state.Params) *state.Files {
 	return files
 }

-// run runs files' writes until the test ends.
-func run(t *testing.T, files *state.Files) {
+// run runs task, the Run or the Watch of state files, until the test
+// ends.
+func run(t *testing.T, task func(context.Context)) {
 	t.Helper()

 	ctx, stop := context.WithCancel(t.Context())
 	stopped := make(chan struct{})

 	go func() {
-		files.Run(ctx)
+		task(ctx)
 		close(stopped)
 	}()

@@ -591,6 +861,79 @@ func run(t *testing.T, files *state.Files) {
 	})
 }

+// watch runs files' Watch until the test ends, and waits until it
+// watches the directory.
+func watch(t *testing.T, files *state.Files, lines processLog) {
+	t.Helper()
+
+	run(t, files.Watch)
+	lines.waitFor(t, watching)
+}
+
+// processLog receives the lines of a process log, each a JSON object, for
+// a test to wait for.
+type processLog chan string
+
+// logInto has params' process log write its lines into a new processLog,
+// and returns that.
+func logInto(params *state.Params) processLog {
+	lines := make(processLog, maxLogLines)
+	params.ProcessLog = slog.New(slog.NewJSONHandler(lines, nil))
+
+	return lines
+}
+
+// Write receives a line of the process log.
+func (l processLog) Write(line []byte) (int, error) {
+	l <- string(line)
+
+	return len(line), nil
+}
+
+// waitFor returns the next line of the process log whose message is msg,
+// passing over the lines before it. It waits as long as that takes, so
+// that a slow test process cannot fail the test.
+func (l processLog) waitFor(t *testing.T, msg string) map[string]any {
+	t.Helper()
+
+	for line := range l {
+		var fields map[string]any
+
+		err := json.Unmarshal([]byte(line), &fields)
+		if err != nil {
+			t.Fatalf("process log line %q is not JSON: %v", line, err)
+		}
+
+		if fields["msg"] == msg {
+			return fields
+		}
+	}
+
+	return nil // never reached: nothing closes the log
+}
+
+// wantTakenIn waits for the next edit taken in, and checks that it is of
+// the state file name in dir.
+func wantTakenIn(t *testing.T, lines processLog, dir, name string) {
+	t.Helper()
+
+	line := lines.waitFor(t, tookIn)
+	if line["file"] != filepath.Join(dir, name) {
+		t.Fatalf("took in %v, want an edit of %s", line, name)
+	}
+}
+
+// edit writes content to the state file name in dir, as an admin saves an
+// edit of it.
+func edit(t *testing.T, dir, name, content string) {
+	t.Helper()
+
+	err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600)
+	if err != nil {
+		t.Fatalf("write %s: %v", name, err)
+	}
+}
+
 // wantEqual checks that the entries read back from file are those
 // written.
 func wantEqual[E comparable](t *testing.T, file string, got, want []E) {