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/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 {