diff --git a/internal/rproxy/limits_test.go b/internal/rproxy/limits_test.go index f52c27f..34b5a9d 100644 --- a/internal/rproxy/limits_test.go +++ b/internal/rproxy/limits_test.go @@ -26,6 +26,10 @@ import ( // post is the start of a POST whose headers end with what follows it. const post = "POST / HTTP/1.1\r\nHost: site.example\r\n" +// limitPastFirstPart is a SWW_RPROXY_MAX_REQUEST_SIZE past the first part of a +// body, the 64 KiB rproxy reads before it asks waf. +const limitPastFirstPart = "128K" + // announcedLength is the header that announces a body of size bytes, and the // end of the headers. func announcedLength(size int) string { @@ -75,7 +79,10 @@ func TestABodyAnnouncedOverItsLimitIsRefusedOnceWAFAllowsIt(t *testing.T) { size, asked int }{ {"with the limit within the first part", "1K", 1025, 1024}, - {"with the limit past the first part", "128K", 129 << 10, questionBodyMaxBytes}, + { + "with the limit past the first part", limitPastFirstPart, 129 << 10, + questionBodyMaxBytes, + }, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() @@ -136,6 +143,106 @@ func TestADenyOutranksTheBodySizeLimit(t *testing.T) { } } +func TestAVisitorSendingConnectionCloseReadsAnAnswerBeforeTheRestOfItsBody( + t *testing.T, +) { + t.Parallel() + + for _, tc := range []struct { + name string + waf http.HandlerFunc + rest int + status int + }{ + {"over the size limit", allow, 65 << 10, http.StatusRequestEntityTooLarge}, + {"denied", deny, 65 << 10, http.StatusForbidden}, + // rproxy reads 256 KiB of this rest. What it leaves unread must fit in the + // connection's buffers. Otherwise the visitor is still sending when the + // connection closes, and gets a reset. + { + "with more than 256 KiB left", allow, 272 << 10, + http.StatusRequestEntityTooLarge, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + standIn := startWAF(t, tc.waf) + app := startApp(t, func(http.ResponseWriter, *http.Request) { + t.Error("the app was asked") + }) + addr, _ := startRProxy(t, standIn.URL, app.URL, + map[string]string{maxRequestSize: limitPastFirstPart}) + + conn := dial(t, addr) + send(t, conn, post+"Connection: close\r\n"+ + announcedLength(questionBodyMaxBytes+tc.rest)+ + strings.Repeat("x", questionBodyMaxBytes)) + + reader := bufio.NewReader(conn) + + res, err := http.ReadResponse(reader, nil) + if err != nil { + t.Fatalf("read the answer: %v", err) + } + + _, _ = io.Copy(io.Discard, res.Body) + _ = res.Body.Close() + + if res.StatusCode != tc.status { + t.Errorf("status %d, want %d", res.StatusCode, tc.status) + } + + // The rest goes in small parts, so that a reset fails a later one. + for sent := 0; sent < tc.rest; sent += 1 << 10 { + send(t, conn, strings.Repeat("x", 1<<10)) + } + + got, err := io.ReadAll(reader) + if err != nil || len(got) != 0 { + t.Errorf( + "after the answer, read %q and %v, want the end", + got, + err, + ) + } + }) + } +} + +func TestCloseDoesNotWaitForTheRestOfTheBodyOfAVisitorSendingConnectionClose( + t *testing.T, +) { + t.Parallel() + + standIn := startWAF(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set(dispositionHeader, dispositionClose) + w.WriteHeader(http.StatusOK) + }) + app := startApp(t, func(http.ResponseWriter, *http.Request) { + t.Error("the app was asked") + }) + // The request time limit is far longer than dial's deadline. + addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{ + maxRequestTime: "120000", + }) + + conn := dial(t, addr) + send(t, conn, post+"Connection: close\r\n"+ + announcedLength(2*questionBodyMaxBytes)+ + strings.Repeat("x", questionBodyMaxBytes)) + + // If the connection stays open, dial's deadline ends this read with a timeout. + text, err := io.ReadAll(conn) + if err != nil || len(text) != 0 { + t.Errorf( + "read %q (%v), want the connection closed without an answer", + text, + err, + ) + } +} + func TestABodyNotAnnouncedOverItsLimitIsRefused(t *testing.T) { t.Parallel() @@ -148,7 +255,7 @@ func TestABodyNotAnnouncedOverItsLimitIsRefused(t *testing.T) { mayReachApp bool }{ {"within the first part", "1K", 2 << 10, false}, - {"past the first part", "128K", 129 << 10, true}, + {"past the first part", limitPastFirstPart, 129 << 10, true}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() diff --git a/internal/rproxy/request.go b/internal/rproxy/request.go index bdd59af..eb0d619 100644 --- a/internal/rproxy/request.go +++ b/internal/rproxy/request.go @@ -38,6 +38,10 @@ const plainText = "text/plain; charset=utf-8" // and the visitor then gets a closed connection and no answer. const minSendTime = 10 * time.Second +// maxBodyReadAfterAnswer is the most of a body left unread that Go's server +// reads once the handler returns, for a visitor without Connection: close. +const maxBodyReadAfterAnswer = 256 << 10 + // errResponseTooLate refuses an app's answer that arrives once // SWW_RPROXY_MAX_RESPONSE_TIME_MS has run out. var errResponseTooLate = errors.New( @@ -338,8 +342,8 @@ func (rq *request) answer(status int) { } // A visitor found too slow is read no more; any other may go on sending until - // its time is up, so that Go's server can read the rest of the body and end - // the request cleanly. + // its time is up, so that the rest of the body is read and the request ends + // cleanly: by Go's server, or by readRestOfBody. deadline := rq.requestDeadline() if status == http.StatusRequestTimeout { deadline = time.Now() @@ -523,6 +527,29 @@ func (rq *request) stopReadingBody(deadline time.Time) { _ = rq.rc.SetReadDeadline(deadline) } +// readRestOfBody reads up to 256 KiB of the rest of the body once the visitor +// has been answered, until the read deadline already set on the connection. It +// does this only for a body with a declared length from a visitor that sent +// Connection: close, because Go's server reads none of that rest. Without the +// read, the visitor, still sending, gets a reset and may never read the answer. +// Past 256 KiB, the MaxBytesReader made with Go's own ResponseWriter has Go's +// server tell the visitor it sends no more, then wait 500 ms before it closes +// the connection. +func (rq *request) readRestOfBody() { + if !rq.in.Close || rq.in.ContentLength <= 0 || rq.body.received.Load() || + rq.out.status == 0 || + rq.out.status == http.StatusSwitchingProtocols { + return + } + + rest := http.MaxBytesReader( + rq.out.ResponseWriter, + rq.in.Body, + maxBodyReadAfterAnswer, + ) + _, _ = io.Copy(io.Discard, rest) +} + // stopPassingOnBody ends the transport's reading of a visitor's body that has // not arrived whole, and lets Go's server read the rest until deadline. Go's // server reads it once the handler returns, but with no deadline at all if diff --git a/internal/rproxy/rproxy.go b/internal/rproxy/rproxy.go index 4e86c0a..32bb67d 100644 --- a/internal/rproxy/rproxy.go +++ b/internal/rproxy/rproxy.go @@ -169,6 +169,7 @@ func newWAFTransport() *http.Transport { // it writes the request's log line and sends its report. func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { rq := h.newRequest(w, r) + defer rq.readRestOfBody() defer rq.finish(r.Context()) if r.Method == http.MethodGet && r.URL.Path == waf.HealthPath {