Byte limits per client over a minute, an hour and a day (closes #20)

SWWAF_BYTES_LIMIT_PER_MINUTE, _PER_HOUR and _PER_DAY (10G, 20G, 50G)
and SWWAF_BYTES_COUNT (both). A request's bytes are counted once its
answer has ended, for a request passed to the app that the rate limits
count; what a WebSocket carries each way, once it closes. Bytes over a
limit ban the client as a broken rate limit does, and cut nothing
short. clients.json keeps the byte buckets, the log line's counts carry
the byte totals, ban notes say what the limit is on, and the limit hits
metric is labelled by kind.

Judgement call: limit_hit names a byte window minute_bytes, hour_bytes
or day_bytes, as counts names the byte totals.
Judgement call: in observe mode, the bytes of a request enforce mode
would have refused are not counted.

Model: opus-5-5

internal/proxy/alerts_test.go

@@ -203,7 +203,16 @@ func startWithAlerts(
 ) (*sender, *clock, *proxy.Server, *alerts.Queue) {
 	t.Helper()
 
-	app := startApp(t, func(http.ResponseWriter, *http.Request) {})
+	return startAppWithAlerts(t, func(http.ResponseWriter, *http.Request) {}, env)
+}
+
+// startAppWithAlerts is startWithAlerts in front of the app handler.
+func startAppWithAlerts(
+	t *testing.T, handler http.HandlerFunc, env map[string]string,
+) (*sender, *clock, *proxy.Server, *alerts.Queue) {
+	t.Helper()
+
+	app := startApp(t, handler)
 	clk := &clock{now: time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC)}
 	settings := map[string]string{
 		trustedProxies:  trustLocalhost,

internal/proxy/bans.go

@@ -6,6 +6,7 @@ import (
 
 	"sneak.berlin/go/smallwebwaf/internal/alerts"
 	"sneak.berlin/go/smallwebwaf/internal/bans"
+	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
 	"sneak.berlin/go/smallwebwaf/internal/requestlog"
 	"sneak.berlin/go/smallwebwaf/internal/rules"
 )
@@ -41,36 +42,92 @@ func (rq *request) banned(now time.Time) bool {
 
 // limitBroken counts the request for the rate limits at now, notes the
 // client's counts for the log line, and reports whether the request takes
-// the client over a limit. In enforce mode such a request bans the
-// client's netblock, and sets the client's counters back to zero; in
-// observe mode it does neither, and raises the alert for the ban it would
-// have made, if that alert would be sent.
+// the client over a rate limit, which breaks it.
 func (rq *request) limitBroken(now time.Time) bool {
-	group := clientGroup(rq.client)
-
-	counts, hit, over := rq.h.limiter.Count(group, now)
+	counts, hit, over := rq.h.limiter.Count(clientGroup(rq.client), now)
 	rq.line.Counts = counts
 
-	if !over {
-		return false
+	if over {
+		rq.banForLimit(now, hit, rq.h.config.BanResponse)
 	}
 
+	return over
+}
+
+// countBytes counts the request's bytes for the byte limits, once its
+// response has ended, and notes the client's byte totals for the log line;
+// its requests stay there as the rate limits counted them. The bytes are
+// the response's body bytes, the request's, or both, as SWWAF_BYTES_COUNT
+// says; for an upgraded connection, such as a WebSocket, which has closed
+// by then, what it carried from the app counts with the response's and
+// what it carried from the client with the request's. Only a request
+// passed to the app has them counted, and only one the rate limits
+// counted; in observe mode, not one that enforce mode would have refused.
+// Bytes that take the client over a byte limit break it; the response was
+// passed on whole.
+func (rq *request) countBytes() {
+	if !rq.counted || rq.line.WouldAction != "" {
+		return
+	}
+
+	response, request := rq.out.bytes, rq.requestBytes()
+	if rq.upgraded != nil {
+		response += rq.upgraded.fromApp.Load()
+		request += rq.upgraded.toApp.Load()
+	}
+
+	var bytes int64
+
+	switch rq.h.config.BytesCount {
+	case "response":
+		bytes = response
+	case "request":
+		bytes = request
+	default: // both
+		bytes = response + request
+	}
+
+	now := rq.h.now()
+
+	counts, hit, over := rq.h.limiter.CountBytes(clientGroup(rq.client), now, bytes)
+	rq.line.Counts.MinuteBytes = counts.MinuteBytes
+	rq.line.Counts.HourBytes = counts.HourBytes
+	rq.line.Counts.DayBytes = counts.DayBytes
+
+	if over {
+		rq.banForLimit(now, hit, rq.out.status)
+	}
+}
+
+// banForLimit bans the client's netblock at now for a broken limit, the
+// one hit names, and notes the offence for the log line. status is what
+// the client was sent, or is sent: SWWAF_BAN_RESPONSE for a request over
+// a rate limit, the app's answer for one whose bytes broke a byte limit.
+// The ban sets the client's counters back to zero. In observe mode it
+// makes no ban and sets nothing back, and raises the alert for the ban it
+// would have made, if that alert would be sent.
+func (rq *request) banForLimit(now time.Time, hit ratelimit.Hit, status int) {
 	rq.line.LimitHit = hit.Window
+	if hit.Kind == ratelimit.KindBytes {
+		rq.line.LimitHit += "_bytes" // as counts names the byte totals
+	}
+
 	rq.line.Offence = requestlog.OffenceLimit
 
 	netblock := rq.h.netblock(rq.client)
 	if rq.h.config.Observe && !rq.wouldAlertBan(netblock, now, bans.CauseLimit) {
-		return true
+		return
 	}
 
 	notes := bans.Notes{
 		ASN:      rq.line.ASN,
 		ASName:   rq.line.ASName,
 		Country:  rq.line.Country,
+		Kind:     hit.Kind,
 		Limit:    hit.Limit,
 		Window:   hit.Window,
-		Count:    hit.Requests,
-		Request:  rq.noted(now),
+		Count:    hit.Count,
+		Request:  rq.noted(now, status),
 		Requests: rq.netblockRequests(netblock),
 	}
 
@@ -80,18 +137,16 @@ func (rq *request) limitBroken(now time.Time) bool {
 			rq.alertBan(ban)
 		}
 
-		return true
+		return
 	}
 
 	ban, made := rq.h.ledger.BanForLimit(netblock, now, notes)
-	rq.h.limiter.Reset(group)
+	rq.h.limiter.Reset(clientGroup(rq.client))
 	rq.line.BanExpires = banExpires(ban)
 
 	if made {
 		rq.alertBan(ban)
 	}
-
-	return true
 }
 
 // banForAttack bans the client's netblock at now for a clear sign of
@@ -110,7 +165,7 @@ func (rq *request) banForAttack(now time.Time, rule rules.Rule) {
 		Country:  rq.line.Country,
 		RuleID:   rule.ID,
 		Target:   rule.Target,
-		Request:  rq.noted(now),
+		Request:  rq.noted(now, rq.h.config.BanResponse),
 		Requests: rq.netblockRequests(netblock),
 	}
 
@@ -177,16 +232,16 @@ func (rq *request) alertBan(ban bans.Ban) {
 	})
 }
 
-// noted is the request, refused at now with SWWAF_BAN_RESPONSE, or in
-// observe mode as it would have been, as the notes of the ban it makes
-// keep it.
-func (rq *request) noted(now time.Time) bans.Request {
+// noted is the request, at now, with status, what the client was sent, or
+// in observe mode would have been, as the notes of the ban it makes keep
+// it.
+func (rq *request) noted(now time.Time, status int) bans.Request {
 	return bans.Request{
 		Time:      now,
 		Method:    rq.in.Method,
 		Host:      rq.in.Host,
 		Path:      rq.in.URL.RequestURI(),
-		Status:    rq.h.config.BanResponse,
+		Status:    status,
 		UserAgent: rq.in.UserAgent(),
 	}
 }

internal/proxy/bans_test.go

@@ -284,6 +284,7 @@ func TestBanNotes(t *testing.T) {
 			ASN:     asnDE,
 			ASName:  asNameDE,
 			Country: "DE",
+			Kind:    "requests",
 			Limit:   1,
 			Window:  minute,
 			Count:   2,

internal/proxy/bodies.go

@@ -103,6 +103,48 @@ func (b *responseBody) Close() error {
 	return b.body.Close()
 }
 
+// upgradedConn is the connection to the app once the app has switched
+// protocols, as for a WebSocket. ReverseProxy writes to it what the client
+// sends and reads from it what the app sends, on goroutines of its own,
+// until the connection closes; it counts the bytes each way, for the byte
+// limits.
+type upgradedConn struct {
+	io.ReadWriteCloser
+
+	// fromApp is how many bytes the app has sent, and toApp how many the
+	// client has.
+	fromApp atomic.Int64
+	toApp   atomic.Int64
+}
+
+// Read reads what the app sends.
+func (c *upgradedConn) Read(p []byte) (int, error) {
+	n, err := c.ReadWriteCloser.Read(p)
+	c.fromApp.Add(int64(n))
+
+	return n, err
+}
+
+// Write sends the app what the client sent.
+func (c *upgradedConn) Write(p []byte) (int, error) {
+	n, err := c.ReadWriteCloser.Write(p)
+	c.toApp.Add(int64(n))
+
+	return n, err
+}
+
+// CloseWrite tells the app that the client sends no more, while what the
+// app sends still passes. ReverseProxy calls it once the client has
+// stopped sending, and closes the connection there if it is not supported.
+func (c *upgradedConn) CloseWrite() error {
+	conn, ok := c.ReadWriteCloser.(interface{ CloseWrite() error })
+	if !ok {
+		return http.ErrNotSupported
+	}
+
+	return conn.CloseWrite()
+}
+
 // limitBody returns body, cut off with an *http.MaxBytesError after
 // maxBytes, or unchanged if maxBytes is zero, which is off.
 func limitBody(body io.ReadCloser, maxBytes int64) io.ReadCloser {

internal/proxy/bytelimits_test.go

@@ -0,0 +1,508 @@
+package proxy_test
+
+import (
+	"bufio"
+	"io"
+	"net/http"
+	"net/netip"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/bans"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+)
+
+// The byte limit settings.
+const (
+	bytesLimitPerMinute = "SWWAF_BYTES_LIMIT_PER_MINUTE"
+	bytesLimitPerHour   = "SWWAF_BYTES_LIMIT_PER_HOUR"
+	bytesLimitPerDay    = "SWWAF_BYTES_LIMIT_PER_DAY"
+	bytesCount          = "SWWAF_BYTES_COUNT"
+)
+
+// The values of SWWAF_BYTES_COUNT.
+const (
+	countResponse = "response"
+	countRequest  = "request"
+	countBoth     = "both"
+)
+
+const (
+	// bodyBytes is the size of the body of each request these tests send
+	// with one, and answerBytes that of each answer of the app.
+	bodyBytes   = 30
+	answerBytes = 70
+	// byteLimit is the byte limit these tests set, as a setting: a request
+	// with a body and its answer, 100 bytes, go over it.
+	byteLimit = "99"
+	// minuteBytes is limit_hit for SWWAF_BYTES_LIMIT_PER_MINUTE.
+	minuteBytes = "minute_bytes"
+)
+
+func TestEachByteLimitBansOnceTheResponseHasEnded(t *testing.T) {
+	t.Parallel()
+
+	const scraper = "192.0.2.200"
+
+	for _, tc := range []struct {
+		setting, window string
+		// apart is the time between the two requests, which the window
+		// still covers.
+		apart time.Duration
+	}{
+		{bytesLimitPerMinute, minute, 0},
+		{bytesLimitPerHour, "hour", 2 * time.Minute},
+		{bytesLimitPerDay, "day", 2 * time.Hour},
+	} {
+		t.Run(tc.setting, func(t *testing.T) {
+			t.Parallel()
+
+			s, clk := startWithAnswers(t, map[string]string{
+				tc.setting: byteLimit, metricsToken: token,
+			})
+
+			// 70 bytes are within the limit of 99.
+			line, _ := s.download()
+			if line.LimitHit != "" || line.Offence != "" {
+				t.Errorf("log line has limit_hit %q and offence %q, want neither",
+					line.LimitHit, line.Offence)
+			}
+
+			// 140 bytes are over it. The response is passed on whole, and
+			// then bans the client for an hour.
+			clk.advance(tc.apart)
+			expires := requestlog.FormatTime(clk.Now().Add(time.Hour))
+
+			line, got := s.download()
+			if got.err != nil || len(got.body) != answerBytes ||
+				line.ResponseBytes != answerBytes {
+				t.Errorf("got %d bytes (%v), and the log line has response_bytes %d, "+
+					"want %d", len(got.body), got.err, line.ResponseBytes, answerBytes)
+			}
+
+			if line.LimitHit != tc.window+"_bytes" || line.Offence != requestlog.OffenceLimit ||
+				line.BanExpires != expires {
+				t.Errorf("log line has limit_hit %q, offence %q and ban_expires %q, "+
+					"want %s_bytes, limit and %s", line.LimitHit, line.Offence,
+					line.BanExpires, tc.window, expires)
+			}
+
+			s.get(client, http.StatusForbidden, requestlog.ActionBanned)
+
+			wantMetric(t, s.scrape(scraper), `smallwebwaf_rate_limit_hits_total{`+
+				`instance="`+alertInstance+`",kind="bytes",window="`+tc.window+`"}`, 1)
+		})
+	}
+}
+
+func TestResponseOverAByteLimitByItselfIsPassedOnWhole(t *testing.T) {
+	t.Parallel()
+
+	s, _ := startWithAnswers(t, map[string]string{bytesLimitPerMinute: "50"})
+
+	// The answer's 70 bytes are over the limit of 50 on their own.
+	line, got := s.download()
+	if got.err != nil || len(got.body) != answerBytes || line.LimitHit != minuteBytes {
+		t.Errorf("got %d bytes (%v), and the log line has limit_hit %q, want %d and %s",
+			len(got.body), got.err, line.LimitHit, answerBytes, minuteBytes)
+	}
+
+	s.get(client, http.StatusForbidden, requestlog.ActionBanned)
+}
+
+func TestBytesOfAnAnswerThatBreaksOffAreCounted(t *testing.T) {
+	t.Parallel()
+
+	s, clk, _, _ := startAppWithAlerts(t, breakOff, map[string]string{
+		bytesLimitPerMinute: "50",
+	})
+	expires := requestlog.FormatTime(clk.Now().Add(time.Hour))
+
+	// The 70 bytes passed on before the app broke off are over the limit of
+	// 50, and ban the client for an hour.
+	line, got := s.requestWithBody(http.MethodGet, client, "/", "", "", http.StatusOK,
+		requestlog.ActionUpstreamError)
+	if len(got.body) != answerBytes || line.LimitHit != minuteBytes ||
+		line.BanExpires != expires {
+		t.Errorf("got %d bytes, and the log line has limit_hit %q and ban_expires %q, "+
+			"want %d, %s and %s", len(got.body), line.LimitHit, line.BanExpires,
+			answerBytes, minuteBytes, expires)
+	}
+
+	s.get(client, http.StatusForbidden, requestlog.ActionBanned)
+}
+
+func TestWebSocketBytesAreCountedOnceItCloses(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		setting string
+		counted float64
+	}{
+		{countResponse, answerBytes},
+		{countRequest, bodyBytes},
+		{countBoth, bodyBytes + answerBytes},
+	} {
+		t.Run(tc.setting, func(t *testing.T) {
+			t.Parallel()
+
+			s, _, _, _ := startAppWithAlerts(t, answerAfterUpgrade, map[string]string{
+				bytesLimitPerMinute: "29", bytesCount: tc.setting,
+			})
+
+			// The client sends 30 bytes and the app 70, each over the limit
+			// of 29, which bans the client once the WebSocket has closed.
+			line := s.webSocket()
+			if line.LimitHit != minuteBytes || line.Counts.MinuteBytes != tc.counted {
+				t.Errorf("log line has limit_hit %q and minute_bytes %v, want %s and %v",
+					line.LimitHit, line.Counts.MinuteBytes, minuteBytes, tc.counted)
+			}
+
+			s.get(client, http.StatusForbidden, requestlog.ActionBanned)
+		})
+	}
+}
+
+func TestBytesCountSaysWhichBytesCount(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		setting string
+		// each is the bytes each request counts, and breaking the request
+		// that goes over the limit of 99.
+		each     float64
+		breaking int
+	}{
+		{countResponse, answerBytes, 2},
+		{countRequest, bodyBytes, 4},
+		{countBoth, bodyBytes + answerBytes, 1},
+	} {
+		t.Run(tc.setting, func(t *testing.T) {
+			t.Parallel()
+
+			s, _ := startWithAnswers(t, map[string]string{
+				bytesLimitPerMinute: byteLimit, bytesCount: tc.setting,
+			})
+
+			for i := 1; i <= tc.breaking; i++ {
+				line := s.upload()
+
+				want := ""
+				if i == tc.breaking {
+					want = minuteBytes
+				}
+
+				counted := float64(i) * tc.each
+				if line.LimitHit != want || line.Counts.MinuteBytes != counted {
+					t.Errorf("request %d: log line has limit_hit %q and minute_bytes %v, "+
+						"want %q and %v", i, line.LimitHit, line.Counts.MinuteBytes,
+						want, counted)
+				}
+			}
+
+			s.get(client, http.StatusForbidden, requestlog.ActionBanned)
+		})
+	}
+}
+
+func TestByteLimitsLeaveOutWhatTheRateLimitsLeaveOut(t *testing.T) {
+	t.Parallel()
+
+	const (
+		allowed = "192.0.2.7"  // in SWWAF_ALLOW_NETS
+		exempt  = "192.0.2.10" // in SWWAF_RATE_LIMIT_EXEMPT_NETS
+	)
+
+	s, _ := startWithAnswers(t, map[string]string{
+		bytesLimitPerMinute:  byteLimit,
+		allowNets:            allowed,
+		rateLimitExemptNets:  exempt,
+		rateLimitExemptPaths: "/assets/",
+	})
+
+	// Each sends 200 bytes, none of which is counted.
+	for _, sent := range []struct{ from, path string }{
+		{allowed, "/"}, {exempt, "/"}, {client, "/assets/app.js"},
+	} {
+		for range 2 {
+			line, _ := s.requestWithBody(http.MethodPost, sent.from, sent.path,
+				uploadHeader, uploadBody, http.StatusOK, requestlog.ActionForward)
+			if _, counted := line.fields["counts"]; counted || line.LimitHit != "" {
+				t.Errorf("%s %s: log line has counts %v and limit_hit %q, want neither",
+					sent.from, sent.path, line.fields["counts"], line.LimitHit)
+			}
+		}
+	}
+
+	// A path that is not exempt is counted, and breaks the limit.
+	line := s.upload()
+	if line.LimitHit != minuteBytes {
+		t.Errorf("log line has limit_hit %q, want %s", line.LimitHit, minuteBytes)
+	}
+}
+
+func TestByteLimitsOffCountTheBytesAndBanNoOne(t *testing.T) {
+	t.Parallel()
+
+	const off = "off"
+
+	s, _ := startWithAnswers(t, map[string]string{
+		bytesLimitPerMinute: off, bytesLimitPerHour: off, bytesLimitPerDay: off,
+	})
+
+	for i := 1; i <= 3; i++ {
+		line := s.upload()
+
+		counted := float64(i * (bodyBytes + answerBytes))
+		if line.LimitHit != "" || line.Counts.MinuteBytes != counted ||
+			line.Counts.HourBytes != counted || line.Counts.DayBytes != counted {
+			t.Errorf("request %d: log line has limit_hit %q and counts %+v, "+
+				"want none and %v bytes in each window", i, line.LimitHit,
+				line.Counts, counted)
+		}
+	}
+}
+
+func TestBanForABrokenByteLimitHasItsNotesAndItsAlert(t *testing.T) {
+	t.Parallel()
+
+	s, clk, server, queue := startAppWithAlerts(t, readAndAnswer, map[string]string{
+		bytesLimitPerMinute: byteLimit,
+	})
+	start := clk.Now()
+
+	s.requestWithBody(http.MethodPost, client, "/upload?part=1", uploadHeader,
+		uploadBody, http.StatusOK, requestlog.ActionForward)
+
+	netblock := netip.MustParsePrefix(client + "/32")
+	want := bans.Ban{
+		Netblock: netblock,
+		Start:    start,
+		Expires:  start.Add(time.Hour),
+		Cause:    bans.CauseLimit,
+		Reason:   "bytes per minute over the limit of " + byteLimit,
+		Notes: bans.Notes{
+			Kind:   "bytes",
+			Limit:  99,
+			Window: minute,
+			Count:  bodyBytes + answerBytes,
+			// The request as it was answered, by the app.
+			Request: bans.Request{
+				Time:      start,
+				Method:    http.MethodPost,
+				Host:      appHost,
+				Path:      "/upload?part=1",
+				Status:    http.StatusOK,
+				UserAgent: userAgent,
+			},
+			Requests: 1,
+		},
+	}
+
+	got := server.Ledger.Bans(netblock)
+	if len(got) != 1 || got[0] != want {
+		t.Fatalf("bans\n%+v\nwant\n%+v", got, want)
+	}
+
+	wantAlerts(t, queue, banAlert(alerts.EventBan, start, client, want,
+		requestlog.FormatTime(want.Expires)))
+
+	if offences := historyOf(t, server, client).Offences.Limit; offences != 1 {
+		t.Errorf("history counts %d offences for a limit, want 1", offences)
+	}
+}
+
+func TestObserveModeLogsAndAlertsAByteLimitAndBansNoOne(t *testing.T) {
+	t.Parallel()
+
+	s, clk, server, queue := startAppWithAlerts(t, readAndAnswer, map[string]string{
+		mode:                observe,
+		bytesLimitPerMinute: byteLimit,
+	})
+	start := clk.Now()
+
+	// No ban sets the client's counters back to zero, so each request
+	// breaks the limit again. The answer is the app's either way, and the
+	// alert for the ban is not sent twice within the cooldown.
+	for range 2 {
+		line := s.upload()
+		wantWouldAction(t, line, "")
+
+		if line.LimitHit != minuteBytes || line.Offence != requestlog.OffenceLimit ||
+			line.BanExpires != "" {
+			t.Errorf("log line has limit_hit %q, offence %q and ban_expires %q, "+
+				"want %s, limit and none", line.LimitHit, line.Offence, line.BanExpires,
+				minuteBytes)
+		}
+	}
+
+	if held := server.Ledger.Snapshot(); len(held) != 0 {
+		t.Errorf("the ledger holds %+v, want no ban", held)
+	}
+
+	waiting := queue.Snapshot().Waiting[alerts.DestinationWebhook]
+	if len(waiting) != 1 {
+		t.Fatalf("%d alerts wait, want 1: %+v", len(waiting), waiting)
+	}
+
+	notes, _ := waiting[0].Detail["notes"].(bans.Notes)
+	alert := banAlert(alerts.EventBan, start, client, bans.Ban{
+		Netblock: netip.MustParsePrefix(client + "/32"), Cause: bans.CauseLimit,
+		Reason: "bytes per minute over the limit of " + byteLimit, Notes: notes,
+	}, requestlog.FormatTime(start.Add(time.Hour)))
+	alert.Detail["mode"] = observe
+	wantAlerts(t, queue, alert)
+}
+
+func TestObserveModeLeavesOutTheBytesOfARequestEnforceModeRefuses(t *testing.T) {
+	t.Parallel()
+
+	s, _ := startWithAnswers(t, map[string]string{
+		mode:                observe,
+		rateLimitPerMinute:  "1",
+		bytesLimitPerMinute: "150",
+	})
+
+	s.upload()
+
+	// The second request breaks the rate limit, which in enforce mode would
+	// refuse it before the app sent anything, so its 100 bytes are not
+	// counted, and the byte limit is not broken. Its line gives the bytes
+	// counted before it.
+	line := s.upload()
+	wantWouldAction(t, line, requestlog.ActionRateLimited)
+
+	if line.LimitHit != minute || line.Counts.MinuteBytes != bodyBytes+answerBytes {
+		t.Errorf("log line has limit_hit %q and minute_bytes %v, want minute and %d",
+			line.LimitHit, line.Counts.MinuteBytes, bodyBytes+answerBytes)
+	}
+}
+
+// uploadHeader and uploadBody are the header and the body of a request
+// with a body of bodyBytes.
+//
+//nolint:gochecknoglobals // a constant cannot call strings.Repeat
+var (
+	uploadHeader = "Content-Length: " + strconv.Itoa(bodyBytes)
+	uploadBody   = strings.Repeat("u", bodyBytes)
+)
+
+// readAndAnswer is the app of these tests: it reads each request's whole
+// body and answers with answerBytes bytes.
+func readAndAnswer(w http.ResponseWriter, r *http.Request) {
+	_, _ = io.Copy(io.Discard, r.Body)
+	_, _ = io.WriteString(w, strings.Repeat("a", answerBytes))
+}
+
+// breakOff is an app that announces an answer of twice answerBytes, and
+// breaks off after answerBytes.
+func breakOff(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Content-Length", strconv.Itoa(2*answerBytes))
+	_, _ = io.WriteString(w, strings.Repeat("a", answerBytes))
+}
+
+// answerAfterUpgrade is an app that switches protocols, as for a
+// WebSocket, and then answers each line it receives with a line of
+// answerBytes.
+func answerAfterUpgrade(w http.ResponseWriter, _ *http.Request) {
+	conn, buffered, err := http.NewResponseController(w).Hijack()
+	if err != nil {
+		return
+	}
+
+	defer func() {
+		_ = conn.Close()
+	}()
+
+	_, _ = buffered.WriteString("HTTP/1.1 101 Switching Protocols\r\n" +
+		"Connection: Upgrade\r\nUpgrade: websocket\r\n\r\n")
+	_ = buffered.Flush()
+
+	for {
+		_, err := buffered.ReadString('\n')
+		if err != nil {
+			return
+		}
+
+		_, _ = buffered.WriteString(strings.Repeat("a", answerBytes-1) + "\n")
+		_ = buffered.Flush()
+	}
+}
+
+// webSocket opens a WebSocket from client to answerAfterUpgrade, sends a
+// line of bodyBytes on it, reads the answer, and closes it. It checks the
+// answer, and the log line as request does, and returns the log line.
+func (s *sender) webSocket() logLine {
+	s.t.Helper()
+
+	conn := dial(s.t, s.addr)
+	send(s.t, conn, "GET /socket HTTP/1.1\r\nHost: "+appHost+"\r\n"+forwardedFor+
+		": "+client+"\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n")
+
+	err := conn.SetReadDeadline(time.Now().Add(waitLimit))
+	if err != nil {
+		s.t.Fatalf("set read deadline: %v", err)
+	}
+
+	reader := bufio.NewReader(conn)
+
+	res, err := http.ReadResponse(reader, nil)
+	if err != nil {
+		s.t.Fatalf("read the answer to the upgrade: %v", err)
+	}
+
+	_ = res.Body.Close()
+
+	if res.StatusCode != http.StatusSwitchingProtocols {
+		s.t.Fatalf("status %d, want %d", res.StatusCode, http.StatusSwitchingProtocols)
+	}
+
+	send(s.t, conn, strings.Repeat("u", bodyBytes-1)+"\n")
+
+	got, err := reader.ReadString('\n')
+	if err != nil || len(got) != answerBytes {
+		s.t.Errorf("got %d bytes (%v), want %d", len(got), err, answerBytes)
+	}
+
+	_ = conn.Close()
+
+	line := s.out.requestLines(s.t, s.sent+1)[s.sent]
+	s.sent++
+	wantLine(s.t, line, http.StatusSwitchingProtocols, requestlog.ActionForward)
+
+	return line
+}
+
+// startWithAnswers is startAppWithAlerts in front of readAndAnswer, for a
+// test that looks at neither the server nor the alerts.
+func startWithAnswers(t *testing.T, env map[string]string) (*sender, *clock) {
+	t.Helper()
+
+	s, clk, _, _ := startAppWithAlerts(t, readAndAnswer, env)
+
+	return s, clk
+}
+
+// download sends a GET request for / from client, and checks that the
+// app's answer is passed on, as request does. It returns the log line and
+// the answer.
+func (s *sender) download() (logLine, answer) {
+	s.t.Helper()
+
+	return s.requestWithBody(http.MethodGet, client, "/", "", "", http.StatusOK,
+		requestlog.ActionForward)
+}
+
+// upload is download for a POST request with a body of bodyBytes, and
+// returns the log line.
+func (s *sender) upload() logLine {
+	s.t.Helper()
+
+	line, _ := s.requestWithBody(http.MethodPost, client, "/", uploadHeader,
+		uploadBody, http.StatusOK, requestlog.ActionForward)
+
+	return line
+}

internal/proxy/metrics_test.go

@@ -222,8 +222,8 @@ func TestMetricsCountLimitsAndBans(t *testing.T) {
 	metrics := s.scrape(scraper)
 	wantMetric(t, metrics,
 		`smallwebwaf_requests_total{action="denied",instance="app",status_class="none"}`, 1)
-	wantMetric(t, metrics,
-		`smallwebwaf_rate_limit_hits_total{instance="app",window="minute"}`, 1)
+	wantMetric(t, metrics, `smallwebwaf_rate_limit_hits_total{instance="app",`+
+		`kind="requests",window="minute"}`, 1)
 	wantMetric(t, metrics, `smallwebwaf_offences_total{instance="app",kind="limit"}`, 1)
 	wantMetric(t, metrics, `smallwebwaf_bans_made_total{cause="limit",instance="app"}`, 1)
 	wantMetric(t, metrics, `smallwebwaf_active_bans{instance="app"}`, 1)
@@ -238,8 +238,8 @@ func TestMetricsCountLimitsAndBans(t *testing.T) {
 	s.get(client, 0, requestlog.ActionRateLimited)
 
 	metrics = s.scrape(scraper)
-	wantMetric(t, metrics,
-		`smallwebwaf_rate_limit_hits_total{instance="app",window="minute"}`, 2)
+	wantMetric(t, metrics, `smallwebwaf_rate_limit_hits_total{instance="app",`+
+		`kind="requests",window="minute"}`, 2)
 	wantMetric(t, metrics, `smallwebwaf_offences_total{instance="app",kind="limit"}`, 2)
 	wantMetric(t, metrics, `smallwebwaf_bans_made_total{cause="limit",instance="app"}`, 2)
 	wantMetric(t, metrics, `smallwebwaf_active_bans{instance="app"}`, 1)

internal/proxy/passthrough_test.go

@@ -122,6 +122,8 @@ func wantAnswer(t *testing.T, got answer, body []byte) {
 func wantRequestFields(t *testing.T, line logLine, host string, sent, received int) {
 	t.Helper()
 
+	bytes := float64(sent + received)
+
 	want := withTimings(line, requestlog.Line{
 		Type: requestType, Time: line.Time, Instance: "app",
 		ClientIP: localhost, Method: http.MethodPatch, Scheme: plain, Host: host,
@@ -131,7 +133,10 @@ func wantRequestFields(t *testing.T, line logLine, host string, sent, received i
 		RequestID: line.RequestID, PeerIP: localhost, ClientGroup: localhost + "/32",
 		ContentLength: int64(sent), ResponseContentType: "text/plain; charset=utf-8",
 		UpstreamStatus: http.StatusTeapot, Action: requestlog.ActionForward,
-		Counts: ratelimit.Counts{Minute: 1, Hour: 1, Day: 1},
+		Counts: ratelimit.Counts{
+			Minute: 1, Hour: 1, Day: 1,
+			MinuteBytes: bytes, HourBytes: bytes, DayBytes: bytes,
+		},
 	})
 	if !reflect.DeepEqual(line.Line, want) {
 		t.Errorf("log line\n%+v\nwant\n%+v", line.Line, want)

internal/proxy/proxy.go

@@ -103,9 +103,12 @@ func New(params Params) *Server {
 		now:        params.Now,
 		metrics:    m,
 		limiter: ratelimit.New(ratelimit.Limits{
-			PerMinute: params.Config.RateLimitPerMinute,
-			PerHour:   params.Config.RateLimitPerHour,
-			PerDay:    params.Config.RateLimitPerDay,
+			PerMinute:      params.Config.RateLimitPerMinute,
+			PerHour:        params.Config.RateLimitPerHour,
+			PerDay:         params.Config.RateLimitPerDay,
+			BytesPerMinute: params.Config.BytesLimitPerMinute,
+			BytesPerHour:   params.Config.BytesLimitPerHour,
+			BytesPerDay:    params.Config.BytesLimitPerDay,
 		}),
 		ledger: bans.New(bans.Rules{
 			LimitBanDuration:     params.Config.LimitBanDuration,
@@ -226,5 +229,10 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Once the response has ended, before the request is added to its
+	// client's history. Deferred, since ReverseProxy panics to end a
+	// response it cannot finish.
+	defer rq.countBytes()
+
 	rq.forward(r.Context())
 }

internal/proxy/request.go

@@ -3,6 +3,7 @@ package proxy
 import (
 	"context"
 	"errors"
+	"io"
 	"net/http"
 	"net/http/httptrace"
 	"net/http/httputil"
@@ -54,7 +55,10 @@ type request struct {
 	// what the lookup gave then, the zero Answer while GeoJS had given none.
 	lookedUp     bool
 	lookupAnswer lookup.Answer
-	start        time.Time
+	// counted is true for a request the rate limits counted, whose bytes
+	// the byte limits count once it has ended.
+	counted bool
+	start   time.Time
 	// checked is when the checks were done, and upstreamStart when the
 	// request was handed to the app.
 	checked       time.Time
@@ -65,6 +69,9 @@ type request struct {
 	refused atomic.Pointer[refusal]
 	// complete is true once the app's whole answer has been passed on.
 	complete bool
+	// upgraded is the connection to the app once the app has switched
+	// protocols, as for a WebSocket, and nil otherwise.
+	upgraded *upgradedConn
 
 	// mu guards what follows. The timeouts run on goroutines of their
 	// own, and the transport starts and stops them, and notes the times
@@ -205,8 +212,9 @@ func (rq *request) check(ctx context.Context) *refusal {
 // them refuses is not counted for the rate limits. Then come the rate
 // limits, unless the client is in SWWAF_RATE_LIMIT_EXEMPT_NETS or the
 // request's path is exempt under SWWAF_RATE_LIMIT_EXEMPT_PATHS, so that
-// every other request is counted, and last the rule files. ctx is the
-// request's own context.
+// every other request is counted, and last the rule files. A request
+// exempt from the rate limits is exempt from the byte limits too. ctx is
+// the request's own context.
 func (rq *request) checkClient(ctx context.Context) string {
 	cfg := rq.h.config
 	if isInside(rq.client, cfg.AllowNets) {
@@ -229,9 +237,9 @@ func (rq *request) checkClient(ctx context.Context) string {
 		return requestlog.ActionCountryDenied
 	}
 
-	exempt := isInside(rq.client, cfg.RateLimitExemptNets) ||
-		pathExempt(rq.in.URL, cfg.RateLimitExemptPaths)
-	if !exempt && rq.limitBroken(now) {
+	rq.counted = !isInside(rq.client, cfg.RateLimitExemptNets) &&
+		!pathExempt(rq.in.URL, cfg.RateLimitExemptPaths)
+	if rq.counted && rq.limitBroken(now) {
 		return requestlog.ActionRateLimited
 	}
 
@@ -324,11 +332,18 @@ func (rq *request) modifyResponse(res *http.Response) error {
 	if res.StatusCode == http.StatusSwitchingProtocols {
 		// An upgraded connection, such as a WebSocket, is not cut by the
 		// timeouts. ReverseProxy writes this answer straight to the
-		// connection it takes over, not through rq.out.
+		// connection it takes over, not through rq.out, and then copies
+		// what passes each way through res.Body, the connection to the app.
 		rq.stopTimers()
 		rq.out.status = res.StatusCode
 		rq.line.Websocket = true
 
+		conn, ok := res.Body.(io.ReadWriteCloser)
+		if ok {
+			rq.upgraded = &upgradedConn{ReadWriteCloser: conn}
+			res.Body = rq.upgraded
+		}
+
 		return nil
 	}
 
@@ -431,10 +446,7 @@ func (rq *request) finish() {
 	line.ResponseContentType = header.Get("Content-Type")
 	line.CacheControl = header.Get("Cache-Control")
 	line.Location = header.Get("Location")
-
-	if rq.body != nil {
-		line.RequestBytes = rq.body.bytes.Load()
-	}
+	line.RequestBytes = rq.requestBytes()
 
 	// limit is the setting whose size or time limit the request passed.
 	var limit string
@@ -497,11 +509,6 @@ func timing(start, end time.Time) *float64 {
 // may have come before either was there, and one from GeoJS that comes
 // later is added when it comes.
 func (rq *request) addToHistory() {
-	var requestBytes int64
-	if rq.body != nil {
-		requestBytes = rq.body.bytes.Load()
-	}
-
 	forwarded := !rq.upstreamStart.IsZero()
 	group := clientGroup(rq.client)
 
@@ -509,7 +516,7 @@ func (rq *request) addToHistory() {
 		Forwarded:     forwarded,
 		Refused:       !forwarded && rq.refused.Load() != nil,
 		Status:        rq.out.status,
-		RequestBytes:  requestBytes,
+		RequestBytes:  rq.requestBytes(),
 		ResponseBytes: rq.out.bytes,
 		BrokeLimit:    rq.line.Offence == requestlog.OffenceLimit,
 	})
@@ -531,6 +538,15 @@ func (rq *request) addToHistory() {
 	}
 }
 
+// requestBytes is how many bytes of the request's body have been read.
+func (rq *request) requestBytes() int64 {
+	if rq.body == nil {
+		return 0
+	}
+
+	return rq.body.bytes.Load()
+}
+
 // 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 {

internal/proxy/requestlog_test.go

@@ -119,7 +119,10 @@ func wantFullLine(t *testing.T, line logLine) {
 		ResponseContentType: "text/html", UpstreamStatus: http.StatusFound,
 		CacheControl: "no-store", Location: "/elsewhere",
 		Action: requestlog.ActionForward,
-		Counts: ratelimit.Counts{Minute: 1, Hour: 1, Day: 1},
+		// Its 3 bytes in and 5 out, each way counted by default.
+		Counts: ratelimit.Counts{
+			Minute: 1, Hour: 1, Day: 1, MinuteBytes: 8, HourBytes: 8, DayBytes: 8,
+		},
 	})
 	if !reflect.DeepEqual(line.Line, want) {
 		t.Errorf("log line\n%+v\nwant\n%+v", line.Line, want)