Core Rule Set reads request bodies up to SWWAF_WAF_BODY_LIMIT (closes #116)

SWWAF_WAF_BODY_LIMIT (default off, at most 1G) has the Core Rule Set read
form data and multipart up to the limit, the rest streaming on, and JSON
and XML no larger than it, with text/json and the application and text
types ending in +json or +xml. The part read is held for the app. A size
or time limit met while reading ends the request. Content-Encoding is
refused again on these kinds. A body Coraza cannot parse, or a multipart
body failing its strict checks, adds 5, as does a multipart body the limit
cuts in a part's headers before a colon or a line feed. Coraza is built
with no_fs_access, so writes no file. Rule 900300 moves to phase 2.

Judgement call: Content-Encoding is refused on a JSON or XML body too
large to read, as SPEC.md allows.

Model: opus-5-5

internal/proxy/bodies.go

@@ -21,6 +21,9 @@ type requestBody struct {
 	// SWWAF_REQUEST_MAX_BYTES.
 	body io.ReadCloser
 	rq   *request
+	// readByCoreRuleSet is what the Core Rule Set read of the body before
+	// the request went to the app, and Read gives first.
+	readByCoreRuleSet []byte
 	// waiting is true while a Read waits for the client to send more.
 	waiting atomic.Bool
 	// received is true once the client has sent the whole body.
@@ -29,8 +32,16 @@ type requestBody struct {
 	bytes atomic.Int64
 }
 
-// Read reads from the client's body.
+// Read reads from the client's body, after what the Core Rule Set read of
+// it, which has been counted already.
 func (b *requestBody) Read(p []byte) (int, error) {
+	if len(b.readByCoreRuleSet) > 0 {
+		n := copy(p, b.readByCoreRuleSet)
+		b.readByCoreRuleSet = b.readByCoreRuleSet[n:]
+
+		return n, nil
+	}
+
 	b.waiting.Store(true)
 	n, err := b.body.Read(p)
 	b.waiting.Store(false)

internal/proxy/coreruleset.go

@@ -1,6 +1,9 @@
 package proxy
 
 import (
+	"errors"
+	"net/http"
+	"os"
 	"time"
 
 	"sneak.berlin/go/smallwebwaf/internal/alerts"
@@ -16,7 +19,8 @@ import (
 // SWWAF_WAF_ANOMALY_THRESHOLD is a match: it raises the waf_block alert,
 // and in block mode refuses the request, which is an offence its client's
 // history counts, and so returns ActionWAFBlocked. It returns "" for a
-// request it does not refuse.
+// request it does not refuse, and for one whose body meets a size or time
+// limit while the Core Rule Set reads it, which it notes nothing of.
 func (rq *request) checkCoreRuleSet() string {
 	cfg := rq.h.config
 	if cfg.WAFMode == config.WAFModeOff || pathExempt(rq.in.URL, cfg.WAFExemptPaths) {
@@ -24,7 +28,12 @@ func (rq *request) checkCoreRuleSet() string {
 	}
 
 	start := time.Now()
-	result := rq.h.coreRuleSet.Inspect(rq.in, rq.client)
+
+	result := rq.inspect()
+	if rq.refused.Load() != nil {
+		return "" // the refusal for that limit, which check returns
+	}
+
 	rq.line.DurationWAF = new(requestlog.Milliseconds(time.Since(start)))
 	rq.line.WAFRuleIDs = result.RuleIDs
 	rq.line.WAFScore = &result.Score
@@ -49,6 +58,39 @@ func (rq *request) checkCoreRuleSet() string {
 	return requestlog.ActionWAFBlocked
 }
 
+// inspect runs the Core Rule Set on the request, which reads the part of
+// its body it inspects within SWWAF_CLIENT_REQUEST_TIMEOUT, and keeps that
+// part for the app. A client that runs out of time is refused with 408
+// here, and a body over SWWAF_REQUEST_MAX_BYTES with 413 as it is read;
+// check returns the refusal. A body that breaks off for any other reason
+// is passed on as far as it came, and the request to the app fails there,
+// as it would have without the Core Rule Set.
+func (rq *request) inspect() waf.Result {
+	if rq.body == nil {
+		// Nothing is read of no body, so nothing can go wrong reading it.
+		result, _, _ := rq.h.coreRuleSet.Inspect(rq.in, rq.client, http.NoBody)
+
+		return result
+	}
+
+	_ = rq.rc.SetReadDeadline(rq.clientRequestDeadline())
+	result, read, err := rq.h.coreRuleSet.Inspect(rq.in, rq.client, rq.body)
+	// The timeouts that run while the request goes to the app take over.
+	_ = rq.rc.SetReadDeadline(time.Time{})
+
+	rq.body.readByCoreRuleSet = read
+
+	if errors.Is(err, os.ErrDeadlineExceeded) {
+		rq.refuse(refusal{
+			status: http.StatusRequestTimeout,
+			action: requestlog.ActionTimedOut,
+			limit:  "SWWAF_CLIENT_REQUEST_TIMEOUT",
+		})
+	}
+
+	return result
+}
+
 // alertWAFBlock raises the waf_block alert for the request, which the Core
 // Rule Set scored at result, at or over SWWAF_WAF_ANOMALY_THRESHOLD. Its
 // detail gives the rule ids, the score, the method and the path with the

internal/proxy/coreruleset_test.go

@@ -1,11 +1,14 @@
 package proxy_test
 
 import (
+	"io"
 	"net/http"
 	"net/netip"
 	"slices"
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"sneak.berlin/go/smallwebwaf/internal/alerts"
 	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
@@ -18,10 +21,14 @@ const (
 	wafAnomalyThreshold = "SWWAF_WAF_ANOMALY_THRESHOLD"
 	wafDisabledRules    = "SWWAF_WAF_DISABLED_RULES"
 	wafExemptPaths      = "SWWAF_WAF_EXEMPT_PATHS"
+	wafBodyLimit        = "SWWAF_WAF_BODY_LIMIT"
 	block               = "block"
 	detect              = "detect"
 )
 
+// formData is the type of a form's body.
+const formData = "application/x-www-form-urlencoded"
+
 // sqlInjection asks for / with an SQL injection in its query, which only
 // the Core Rule Set's rule 942100 matches, with a score of 5, the default
 // SWWAF_WAF_ANOMALY_THRESHOLD.
@@ -224,6 +231,197 @@ func TestDisabledRulesSwitchOffWhatGiteaWouldBeRefused(t *testing.T) {
 	}
 }
 
+func TestAttackInAFormBodyIsRefusedOnlyWhileBodiesAreRead(t *testing.T) {
+	t.Parallel()
+
+	const body = "id=1'%20OR%20'1'='1"
+
+	header := "Content-Type: " + formData + "\r\nContent-Length: " +
+		strconv.Itoa(len(body))
+
+	s, _, _ := startWithClock(t, "", map[string]string{wafMode: block})
+	line, _ := s.requestWithBody(http.MethodPost, client, "/", header, body,
+		http.StatusOK, requestlog.ActionForward)
+	wantWAF(t, line, new(0))
+
+	s, _, _ = startWithClock(t, "", map[string]string{
+		wafMode: block, wafBodyLimit: sizeLimitSetting,
+	})
+	line, _ = s.requestWithBody(http.MethodPost, client, "/", header, body,
+		http.StatusForbidden, requestlog.ActionWAFBlocked)
+	wantWAF(t, line, new(5), 942100)
+}
+
+func TestBodiesReachTheAppAsSentWhileBodiesAreRead(t *testing.T) {
+	t.Parallel()
+
+	// The app answers with the body it was sent, once it has the whole of
+	// it: Go's server reads no more of a body once the answer has begun.
+	app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		_, _ = w.Write(body)
+	})
+	addr, out := startProxy(t, app.URL, map[string]string{
+		wafMode: block, wafBodyLimit: sizeLimitSetting,
+	})
+	longer := "a=" + strings.Repeat("b", 64*sizeLimit)
+
+	for i, tc := range []struct {
+		name, contentType, body string
+		// announced sends the body's length in Content-Length; otherwise
+		// the body is sent in chunks with no length given.
+		announced bool
+	}{
+		{"form data within the limit", formData, "a=b", true},
+		{"form data longer than the limit", formData, longer, true},
+		{"form data longer than the limit, not announced", formData, longer, false},
+		{
+			"JSON larger than the limit", "application/json",
+			`{"a":"` + strings.Repeat("b", 2*sizeLimit) + `"}`, true,
+		},
+		{
+			"a binary body", "application/octet-stream",
+			strings.Repeat("\x00\xff", sizeLimit), true,
+		},
+	} {
+		// A reader whose length the client cannot tell is sent in chunks.
+		var body io.Reader = strings.NewReader(tc.body)
+		if !tc.announced {
+			body = io.MultiReader(body)
+		}
+
+		req := newRequest(t, http.MethodPost, addr, "/", body)
+		req.Header.Set("Content-Type", tc.contentType)
+
+		got := do(t, req)
+		if got.status != http.StatusOK || string(got.body) != tc.body {
+			t.Errorf("%s: the app got %d bytes, answered %d, want the %d sent, 200",
+				tc.name, len(got.body), got.status, len(tc.body))
+		}
+
+		line := out.requestLines(t, i+1)[i]
+		wantLine(t, line, http.StatusOK, requestlog.ActionForward)
+
+		if line.RequestBytes != int64(len(tc.body)) {
+			t.Errorf("%s: log line has request_bytes %d, want %d", tc.name,
+				line.RequestBytes, len(tc.body))
+		}
+	}
+}
+
+func TestFormBodyLongerThanTheLimitStreamsOnToTheApp(t *testing.T) {
+	t.Parallel()
+
+	const (
+		first = "a=" // and twice the limit of b's, then the rest
+		rest  = 64 * sizeLimit
+	)
+
+	// past is closed once the app has received twice what the Core Rule
+	// Set reads, and got is the length of the whole body it received.
+	past := make(chan struct{})
+	got := make(chan int64, 1)
+	app := startApp(t, func(_ http.ResponseWriter, r *http.Request) {
+		n, _ := io.CopyN(io.Discard, r.Body, 2*sizeLimit)
+
+		close(past)
+
+		m, _ := io.Copy(io.Discard, r.Body)
+		got <- n + m
+	})
+	addr, out := startProxy(t, app.URL, map[string]string{
+		wafMode: block, wafBodyLimit: sizeLimitSetting,
+	})
+
+	// The client sends the rest only once the app has received the first
+	// part: were smallwebwaf to hold the body until the end, it would
+	// never come.
+	body, sender := io.Pipe()
+
+	go func() {
+		_, _ = io.WriteString(sender, first+strings.Repeat("b", 2*sizeLimit))
+
+		select {
+		case <-past:
+		case <-time.After(waitLimit):
+			t.Error("the app got no more than the Core Rule Set reads " +
+				"before the whole body was sent")
+
+			_ = sender.CloseWithError(io.ErrUnexpectedEOF)
+
+			return
+		}
+
+		_, _ = io.WriteString(sender, strings.Repeat("b", rest))
+		_ = sender.Close()
+	}()
+
+	req := newRequest(t, http.MethodPost, addr, "/", body)
+	req.Header.Set("Content-Type", formData)
+	wantStatus(t, do(t, req), http.StatusOK)
+
+	want := int64(len(first) + 2*sizeLimit + rest)
+	if n := <-got; n != want {
+		t.Errorf("the app got %d bytes, want %d", n, want)
+	}
+
+	wantLine(t, out.requestLine(t), http.StatusOK, requestlog.ActionForward)
+}
+
+func TestClientTooSlowToSendWhatTheCoreRuleSetReads(t *testing.T) {
+	t.Parallel()
+
+	app := startApp(t, func(http.ResponseWriter, *http.Request) {})
+	addr, out := startProxy(t, app.URL, map[string]string{
+		wafMode: block, wafBodyLimit: sizeLimitSetting,
+		clientRequestTimeout: shortTimeoutSetting, metricsToken: token,
+	})
+
+	conn := dial(t, addr)
+	send(t, conn, "POST /comment HTTP/1.1\r\nHost: app\r\nContent-Type: "+formData+
+		"\r\nContent-Length: 100\r\n\r\ncontent=the first bytes")
+
+	wantStatus(t, readResponse(t, conn), http.StatusRequestTimeout)
+
+	line := out.requestLine(t)
+	wantLine(t, line, http.StatusRequestTimeout, requestlog.ActionTimedOut)
+	wantNotSentToTheApp(t, line)
+	wantLimitHits(t, addr, clientRequestTimeout, 1)
+}
+
+func TestBodyOverTheSizeLimitWhileTheCoreRuleSetReadsIt(t *testing.T) {
+	t.Parallel()
+
+	app := startApp(t, func(http.ResponseWriter, *http.Request) {})
+	addr, out := startProxy(t, app.URL, map[string]string{
+		wafMode: block, wafBodyLimit: "4K",
+		requestMaxBytes: sizeLimitSetting, metricsToken: token,
+	})
+
+	// Sent in chunks, its length is not announced, and is found to be over
+	// the limit as the Core Rule Set reads it.
+	body := io.MultiReader(strings.NewReader("a=" + strings.Repeat("b", 2*sizeLimit)))
+	req := newRequest(t, http.MethodPost, addr, "/", body)
+	req.Header.Set("Content-Type", formData)
+	wantStatus(t, do(t, req), http.StatusRequestEntityTooLarge)
+
+	line := out.requestLine(t)
+	wantLine(t, line, http.StatusRequestEntityTooLarge, requestlog.ActionTooLarge)
+	wantNotSentToTheApp(t, line)
+	wantLimitHits(t, addr, requestMaxBytes, 1)
+}
+
+// wantNotSentToTheApp checks that the request of line was not sent to the
+// app at all.
+func wantNotSentToTheApp(t *testing.T, line logLine) {
+	t.Helper()
+
+	_, sent := line.fields["duration_upstream_total"]
+	if sent {
+		t.Error("log line has duration_upstream_total, for a request sent to the app")
+	}
+}
+
 func TestResponsesAreNotInspected(t *testing.T) {
 	t.Parallel()
 

internal/proxy/proxy.go

@@ -232,8 +232,8 @@ func newReputation(
 }
 
 // newCoreRuleSet returns the Core Rule Set at SWWAF_WAF_PARANOIA_LEVEL,
-// without the rules SWWAF_WAF_DISABLED_RULES switches off, or nil while
-// SWWAF_WAF_MODE is off.
+// without the rules SWWAF_WAF_DISABLED_RULES switches off, reading bodies
+// up to SWWAF_WAF_BODY_LIMIT, or nil while SWWAF_WAF_MODE is off.
 func newCoreRuleSet(cfg *config.Config) *waf.CoreRuleSet {
 	if cfg.WAFMode == config.WAFModeOff {
 		return nil
@@ -241,11 +241,12 @@ func newCoreRuleSet(cfg *config.Config) *waf.CoreRuleSet {
 
 	coreRuleSet, err := waf.New(waf.Params{
 		ParanoiaLevel: cfg.WAFParanoiaLevel, DisabledRules: cfg.WAFDisabledRules,
+		BodyLimit: cfg.WAFBodyLimit,
 	})
 	if err != nil {
 		// The Core Rule Set is built in, and the settings cannot break it:
-		// the paranoia level is from 1 to 4, and the id of no rule switches
-		// nothing off.
+		// the paranoia level is from 1 to 4, the body limit at most 1G, and
+		// the id of no rule switches nothing off.
 		panic(err)
 	}
 

internal/proxy/request.go

@@ -188,16 +188,23 @@ func requestHeaders(r *http.Request, names []string) map[string]string {
 }
 
 // check is the one place where a request can be refused once its client
-// is known, before its body is read or anything reaches the app. It
-// returns nil to let the request through. The checks of checkClient come
-// first, answered with SWWAF_BAN_RESPONSE, or 403 for a block rule or the
-// Core Rule Set, and then the size limit, so that a request the rate
-// limits count is counted even when it is refused for its size. In
-// observe mode a request checkClient refuses goes on to the size limit
-// like any other. ctx is the request's own context.
+// is known, before anything reaches the app, and before its body is read,
+// but for the part the Core Rule Set reads. It returns nil to let the
+// request through. The checks of checkClient come first, answered with
+// SWWAF_BAN_RESPONSE, or 403 for a block rule or the Core Rule Set, and
+// then the size limit, so that a request the rate limits count is counted
+// even when it is refused for its size. In observe mode a request
+// checkClient refuses goes on to the size limit like any other. A size or
+// time limit the Core Rule Set's reading of the body meets ends the
+// request in either mode. ctx is the request's own context.
 func (rq *request) check(ctx context.Context) *refusal {
 	action := rq.checkClient(ctx)
 
+	refused := rq.refused.Load()
+	if refused != nil {
+		return refused
+	}
+
 	switch {
 	case action == "":
 	case rq.h.config.Observe: