Keep the bans, the clients and GeoJS's answers in state files (closes #17)

smallwebwaf now copies its state to bans.json, clients.json and
lookups.json in SWWAF_STATE_DIR, as "Persistent state" in SPEC.md
describes, and reads them back at start, so a restart lifts no ban and
gives no client a fresh allowance. Each client gains a history, and a
ban's notes count the netblock's requests. bans.json is written
SWWAF_STATE_WRITE_DELAY after a ban, and every file every
SWWAF_STATE_COUNTER_INTERVAL and at the stop. A ban read back is masked
to its netblock and refuses every client in it. A file that does not
parse, an unknown version, an entry without a field it needs, or an
unwritable directory stops the start.

Deviation: no AS number or name, and no ban cause, reason or lifting yet.

Model: opus-5-5

internal/proxy/bans.go

@@ -14,10 +14,10 @@ func (rq *request) banResponse(action string) *refusal {
 	return &refusal{status: rq.h.config.BanResponse, action: action}
 }
 
-// banned reports whether a ban on the client's netblock refuses the
-// request at now, and notes for the log line when that ban ends.
+// banned reports whether a ban on a netblock the client is in refuses
+// the request at now, and notes for the log line when that ban ends.
 func (rq *request) banned(now time.Time) bool {
-	ban, banned := rq.h.ledger.Check(rq.netblock(), now)
+	ban, banned := rq.h.ledger.Check(rq.client, now)
 	if banned {
 		rq.line.BanExpires = banExpires(ban)
 	}
@@ -36,7 +36,8 @@ func (rq *request) limitBroken(now time.Time) bool {
 		return false
 	}
 
-	ban := rq.h.ledger.BanForLimit(rq.netblock(), now, bans.Notes{
+	netblock := rq.netblock()
+	ban := rq.h.ledger.BanForLimit(netblock, now, bans.Notes{
 		Country: rq.line.Country,
 		Limit:   hit.Limit,
 		Window:  hit.Window,
@@ -49,6 +50,8 @@ func (rq *request) limitBroken(now time.Time) bool {
 			Status:    rq.h.config.BanResponse,
 			UserAgent: rq.in.UserAgent(),
 		},
+		// The histories count this request only once it has ended.
+		Requests: rq.h.limiter.Requests(netblock) + 1,
 	})
 	rq.h.limiter.Reset(group)
 

internal/proxy/bans_test.go

@@ -133,7 +133,7 @@ func TestBanIsNotCountedAndResetsTheCounters(t *testing.T) {
 	s.get(client, http.StatusOK, requestlog.ActionForward)
 	s.get(client, http.StatusForbidden, requestlog.ActionRateLimited)
 
-	banned := proxy.LedgerOf(server).Bans(netip.MustParsePrefix(client + "/32"))
+	banned := server.Ledger.Bans(netip.MustParsePrefix(client + "/32"))
 	if len(banned) != 2 || banned[0].Notes.Refused != 3 {
 		t.Errorf("bans %+v, want two, the first with 3 requests refused", banned)
 	}
@@ -291,12 +291,15 @@ func TestBanNotes(t *testing.T) {
 				Status:    http.StatusForbidden,
 				UserAgent: userAgent,
 			},
+			// The one let through, the one that broke the limit and the two
+			// refused under the ban.
+			Requests:    4,
 			Refused:     2,
 			EarlierBans: 0,
 		},
 	}
 
-	ledger := proxy.LedgerOf(server)
+	ledger := server.Ledger
 
 	got := ledger.Bans(netblock)
 	if len(got) != 1 || got[0] != want {
@@ -361,7 +364,7 @@ func (c *clock) advance(d time.Duration) {
 // set to midnight, the start of a bucket in every window.
 func startWithClock(
 	t *testing.T, geojsURL string, env map[string]string,
-) (*sender, *clock, *http.Server) {
+) (*sender, *clock, *proxy.Server) {
 	t.Helper()
 
 	app := startApp(t, func(http.ResponseWriter, *http.Request) {})

internal/proxy/export_test.go

@@ -1,15 +0,0 @@
-package proxy
-
-import (
-	"net/http"
-
-	"sneak.berlin/go/smallwebwaf/internal/bans"
-)
-
-// LedgerOf returns the ban ledger of a server New returned, so that the
-// tests can read the bans' notes.
-func LedgerOf(server *http.Server) *bans.Ledger {
-	h, _ := server.Handler.(*handler)
-
-	return h.ledger
-}

internal/proxy/history_test.go

@@ -0,0 +1,103 @@
+package proxy_test
+
+import (
+	"io"
+	"net/http"
+	"net/netip"
+	"strings"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/proxy"
+	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+)
+
+func TestHistoryKeepsEachRequestOfTheClient(t *testing.T) {
+	t.Parallel()
+
+	geojsURL, _ := startGeoJS(t)
+	s, clk, server := startWithClock(t, geojsURL, map[string]string{
+		rateLimitPerMinute: "2",
+		deniedCountries:    "kp",
+	})
+	start := clk.Now()
+
+	// Two let through, one over the limit, which bans the client, and one
+	// refused under that ban, for which the country is not looked up.
+	s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+	clk.advance(time.Second)
+	s.get(fromDE, http.StatusOK, requestlog.ActionForward)
+	s.get(fromDE, http.StatusForbidden, requestlog.ActionRateLimited)
+	clk.advance(time.Second)
+	s.get(fromDE, http.StatusForbidden, requestlog.ActionBanned)
+
+	want := ratelimit.History{
+		FirstSeen: start,
+		LastSeen:  start.Add(2 * time.Second),
+		Country:   "DE",
+		LookedUp:  start.Add(time.Second),
+		Requests:  4,
+		Forwarded: 2,
+		Refused:   2,
+		// The app answers with no body, smallwebwaf with its status text.
+		ResponseBytes: 2 * int64(len("Forbidden\n")),
+		Responses:     ratelimit.Responses{Status2xx: 2, Status4xx: 2},
+		Offences:      ratelimit.Offences{Limit: 1},
+	}
+
+	got := historyOf(t, server, fromDE)
+	if got != want {
+		t.Errorf("history\n%+v\nwant\n%+v", got, want)
+	}
+}
+
+func TestHistoryCountsTheBodiesEachWay(t *testing.T) {
+	t.Parallel()
+
+	app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		_, _ = io.WriteString(w, "hello")
+	})
+	addr, out, server := startProxyWithClock(t, app.URL, "", time.Now, nil)
+
+	got := do(t, newRequest(t, http.MethodPost, addr, "/", strings.NewReader("abc")))
+	wantStatus(t, got, http.StatusOK)
+	out.requestLine(t)
+
+	history := historyOf(t, server, localhost)
+	if history.RequestBytes != 3 || history.ResponseBytes != 5 {
+		t.Errorf("history counts %d bytes in and %d out, want 3 and 5",
+			history.RequestBytes, history.ResponseBytes)
+	}
+}
+
+func TestHealthEndpointIsNotInTheHistory(t *testing.T) {
+	t.Parallel()
+
+	app := startApp(t, func(http.ResponseWriter, *http.Request) {})
+	addr, out, server := startProxyWithClock(t, app.URL, "", time.Now, nil)
+
+	wantStatus(t, get(t, addr, proxy.HealthPath), http.StatusOK)
+	out.requestLine(t)
+
+	if clients := server.Limiter.Snapshot(); len(clients) != 0 {
+		t.Errorf("the table holds %+v, want no client", clients)
+	}
+}
+
+// historyOf returns the history of the client at addr.
+func historyOf(t *testing.T, server *proxy.Server, addr string) ratelimit.History {
+	t.Helper()
+
+	client := netip.MustParsePrefix(addr + "/32")
+	for _, c := range server.Limiter.Snapshot() {
+		if c.Client == client {
+			return c.History
+		}
+	}
+
+	t.Fatalf("%s is not in the table", client)
+
+	return ratelimit.History{}
+}

internal/proxy/proxy.go

@@ -38,53 +38,70 @@ type Params struct {
 	// lookup.URL. GeoJS is asked only while a country list is set.
 	GeoJSURL string
 	// Now tells the time by which requests are counted for the rate
-	// limits and bans are made and run out, normally time.Now.
+	// limits, bans are made and run out, and GeoJS's answers are kept,
+	// normally time.Now in UTC, the time the state files give.
 	Now func() time.Time
 }
 
+// Server is the server smallwebwaf runs, with the parts of the proxy
+// whose state the state files keep.
+type Server struct {
+	*http.Server
+
+	Ledger  *bans.Ledger
+	Limiter *ratelimit.Limiter
+	GeoJS   *lookup.GeoJS
+}
+
 // New returns the server smallwebwaf runs: each request it reads passes
 // through the proxy. Go's server itself refuses a request line and
 // headers over SWWAF_CLIENT_REQUEST_HEADER_MAX_BYTES, with 431, closes a
 // connection idle for SWWAF_CLIENT_IDLE_TIMEOUT, and applies
 // SWWAF_CLIENT_REQUEST_TIMEOUT while the headers arrive; the proxy
 // applies the timeouts and size limits from then on.
-func New(params Params) *http.Server {
+func New(params Params) *Server {
 	errorLog := slog.NewLogLogger(params.ProcessLog.Handler(), slog.LevelWarn)
+	h := &handler{
+		config:     params.Config,
+		requestLog: params.RequestLog,
+		processLog: params.ProcessLog,
+		errorLog:   errorLog,
+		transport:  newTransport(),
+		now:        params.Now,
+		limiter: ratelimit.New(ratelimit.Limits{
+			PerMinute: params.Config.RateLimitPerMinute,
+			PerHour:   params.Config.RateLimitPerHour,
+			PerDay:    params.Config.RateLimitPerDay,
+		}),
+		ledger: bans.New(bans.Rules{
+			LimitBanDuration:     params.Config.LimitBanDuration,
+			LimitBanRepeatWindow: params.Config.LimitBanRepeatWindow,
+			MaxBanDuration:       params.Config.MaxBanDuration,
+			MaxBans:              params.Config.MaxBans,
+		}),
+		geojs: lookup.New(lookup.Params{
+			URL:        params.GeoJSURL,
+			Now:        params.Now,
+			ProcessLog: params.ProcessLog,
+		}),
+	}
 
-	return &http.Server{
-		Addr: params.Config.ListenAddr,
-		Handler: &handler{
-			config:     params.Config,
-			requestLog: params.RequestLog,
-			processLog: params.ProcessLog,
-			errorLog:   errorLog,
-			transport:  newTransport(),
-			now:        params.Now,
-			limiter: ratelimit.New(ratelimit.Limits{
-				PerMinute: params.Config.RateLimitPerMinute,
-				PerHour:   params.Config.RateLimitPerHour,
-				PerDay:    params.Config.RateLimitPerDay,
-			}),
-			ledger: bans.New(bans.Rules{
-				LimitBanDuration:     params.Config.LimitBanDuration,
-				LimitBanRepeatWindow: params.Config.LimitBanRepeatWindow,
-				MaxBanDuration:       params.Config.MaxBanDuration,
-				MaxBans:              params.Config.MaxBans,
-			}),
-			geojs: lookup.New(lookup.Params{
-				URL:        params.GeoJSURL,
-				Now:        time.Now,
-				ProcessLog: params.ProcessLog,
-			}),
+	return &Server{
+		Server: &http.Server{
+			Addr:              params.Config.ListenAddr,
+			Handler:           h,
+			ReadHeaderTimeout: params.Config.ClientRequestTimeout,
+			// Off is an IdleTimeout of 0, which Go's server replaces with
+			// ReadTimeout: no limit, as long as ReadTimeout stays unset.
+			IdleTimeout: params.Config.ClientIdleTimeout,
+			// Go's server reads 4 KiB past MaxHeaderBytes before it
+			// refuses, so the limit a client meets is the setting.
+			MaxHeaderBytes: int(params.Config.ClientRequestHeaderMaxBytes - 4<<10),
+			ErrorLog:       errorLog,
 		},
-		ReadHeaderTimeout: params.Config.ClientRequestTimeout,
-		// Off is an IdleTimeout of 0, which Go's server replaces with
-		// ReadTimeout: no limit, as long as ReadTimeout stays unset.
-		IdleTimeout: params.Config.ClientIdleTimeout,
-		// Go's server reads 4 KiB past MaxHeaderBytes before it refuses,
-		// so the limit a client meets is the setting.
-		MaxHeaderBytes: int(params.Config.ClientRequestHeaderMaxBytes - 4<<10),
-		ErrorLog:       errorLog,
+		Ledger:  h.ledger,
+		Limiter: h.limiter,
+		GeoJS:   h.geojs,
 	}
 }
 
@@ -130,6 +147,9 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Once the request has ended, before its log line is written.
+	defer rq.addToHistory()
+
 	refused := rq.check(r.Context())
 	if refused != nil {
 		rq.answer(*refused)

internal/proxy/proxy_test.go

@@ -200,7 +200,7 @@ func startProxyWithGeoJS(
 func startProxyWithClock(
 	t *testing.T, appURL, geojsURL string, now func() time.Time,
 	env map[string]string,
-) (string, *output, *http.Server) {
+) (string, *output, *proxy.Server) {
 	t.Helper()
 
 	settings := map[string]string{"SWWAF_UPSTREAM_URL": appURL}

internal/proxy/request.go

@@ -12,6 +12,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
 	"sneak.berlin/go/smallwebwaf/internal/requestlog"
 )
 
@@ -318,6 +319,24 @@ func (rq *request) finish() {
 	}
 }
 
+// addToHistory adds the request, which has ended, to its client's
+// history.
+func (rq *request) addToHistory() {
+	var requestBytes int64
+	if rq.body != nil {
+		requestBytes = rq.body.bytes.Load()
+	}
+
+	rq.h.limiter.AddToHistory(clientGroup(rq.client), rq.h.now(), ratelimit.Request{
+		Country:       rq.line.Country,
+		Forwarded:     !rq.upstreamStart.IsZero(),
+		Status:        rq.out.status,
+		RequestBytes:  requestBytes,
+		ResponseBytes: rq.out.bytes,
+		BrokeLimit:    rq.line.Offence == requestlog.OffenceLimit,
+	})
+}
+
 // clientRequestDeadline is when the client must have sent its whole
 // request, or zero when SWWAF_CLIENT_REQUEST_TIMEOUT is off.
 func (rq *request) clientRequestDeadline() time.Time {