diff --git a/internal/rproxy/limits_test.go b/internal/rproxy/limits_test.go index 2af3b1f..d44c8c0 100644 --- a/internal/rproxy/limits_test.go +++ b/internal/rproxy/limits_test.go @@ -12,7 +12,6 @@ import ( "os" "strconv" "strings" - "sync" "sync/atomic" "testing" "time" @@ -383,6 +382,19 @@ type deadlineRecorder struct { // stalled, if set, sends the part of the request body that never arrives. A // read deadline that has passed ends a read of it, as on a connection. stalled *io.PipeWriter + // flushError, if set, is what every flush returns, and nothing is flushed. + flushError error +} + +// FlushError returns flushError if it is set, and otherwise flushes. +func (w *deadlineRecorder) FlushError() error { + if w.flushError != nil { + return w.flushError + } + + w.Flush() + + return nil } // SetReadDeadline notes deadline. @@ -448,6 +460,45 @@ func TestABodyLeftUnreadIsReadOnlyUntilTheRequestTimeLimit(t *testing.T) { } } +func TestABodyLeftUnreadIsReadNoMoreOnceTheTimeToSendTheAnswerRunsOut(t *testing.T) { + t.Parallel() + + standIn := startWAF(t, allow) + app := startApp(t, answerBeforeTheBody) + _, _, server := startServer(t, standIn.URL, app.URL, nil) + + // The visitor sends the first part and a little more, and then nothing. + stalled, stop := io.Pipe() + + t.Cleanup(func() { _ = stop.Close() }) + + req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/", + io.MultiReader(bytes.NewReader(make([]byte, questionBodyMaxBytes+1)), stalled)) + req.ContentLength = 2 * questionBodyMaxBytes + req.Header.Set(contentLengthHeader, strconv.Itoa(2*questionBodyMaxBytes)) + + // Every flush fails, as once the time to send the answer has run out. The + // default time limits are far longer than the test, so rproxy records no + // refusal. + w := &deadlineRecorder{ + ResponseRecorder: httptest.NewRecorder(), + stalled: stop, + flushError: os.ErrDeadlineExceeded, + } + + before := time.Now() + + server.Handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Errorf("status %d, want the app's 200", w.Code) + } + + if w.readDeadline.Before(before) || w.readDeadline.After(time.Now()) { + t.Errorf("the reading of the body ends at %s, want now", w.readDeadline) + } +} + func TestAnAnswerRProxyGivesItselfHasTheResponseTimeToBeSent(t *testing.T) { t.Parallel() @@ -918,49 +969,45 @@ func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOf ) { t.Parallel() - const limit = 300 * time.Millisecond + // rproxy passes on a part sent at once well within this limit, even on a busy + // host. + const limit = 2 * time.Second - // With two parts, rproxy fails to send the first and then waits for the - // second, which the app sends only once rproxy has ended the request to it. + // The app sends the parts in atOnce with its status and headers. It sends the + // parts in late only once rproxy has ended the request to it, which rproxy + // does when it cuts off the answer, so that those never reach the visitor. for _, tc := range []struct { - name string - parts []string + name string + atOnce, late []string }{ - {"with the body in one part", []string{greeting}}, - {"with the body in two parts", []string{greeting[:5], greeting[5:]}}, + {"with the body in one part", nil, []string{greeting}}, + {"with the body in two parts", []string{greeting[:5]}, []string{greeting[5:]}}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() - headersRead := make(chan struct{}) - // letAppSendBody also runs when the test ends, as the app's server waits - // for the app. - letAppSendBody := sync.OnceFunc(func() { close(headersRead) }) standIn := startWAF(t, allow) - // The app sends its body once the visitor has the status and headers and - // the response time limit has run out since, so that rproxy cannot send - // it. After each part, it reads the request's body, which ends when rproxy - // ends the request to it. app := startApp(t, func(w http.ResponseWriter, r *http.Request) { rc := http.NewResponseController(w) _ = rc.EnableFullDuplex() w.Header().Set(contentLengthHeader, strconv.Itoa(len(greeting))) + for _, part := range tc.atOnce { + _, _ = io.WriteString(w, part) + } + _ = rc.Flush() - <-headersRead - time.Sleep(limit) + // The request's body ends when rproxy ends the request to the app. + _, _ = io.Copy(io.Discard, r.Body) - for _, part := range tc.parts { + for _, part := range tc.late { _, _ = io.WriteString(w, part) _ = rc.Flush() - _, _ = io.Copy(io.Discard, r.Body) } }) - t.Cleanup(letAppSendBody) - // The request time limit is far longer than dial's deadline. addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{ maxRequestTime: "120000", @@ -975,15 +1022,14 @@ func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOf t.Fatalf("read the answer: %v", err) } - letAppSendBody() - // If the connection stays open, dial's deadline ends this read with a // timeout. body, err := io.ReadAll(res.Body) _ = res.Body.Close() - if len(body) != 0 || !errors.Is(err, io.ErrUnexpectedEOF) { - t.Errorf("read %q (%v), want the connection closed", body, err) + want := strings.Join(tc.atOnce, "") + if string(body) != want || !errors.Is(err, io.ErrUnexpectedEOF) { + t.Errorf("read %q (%v), want %q and the connection closed", body, err, want) } }) }