diff --git a/internal/rproxy/limits_test.go b/internal/rproxy/limits_test.go index 685f05b..2af3b1f 100644 --- a/internal/rproxy/limits_test.go +++ b/internal/rproxy/limits_test.go @@ -12,6 +12,7 @@ import ( "os" "strconv" "strings" + "sync" "sync/atomic" "testing" "time" @@ -807,8 +808,10 @@ func TestAVisitorThatStopsSendingAfterTheAppHasAnsweredGetsTheAnswer(t *testing. app := startApp(t, func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, greeting) }) + // The request time limit is far longer than dial's deadline, so that a busy + // host cannot use it up before the app's answer arrives. addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{ - maxRequestTime: "3000", + maxRequestTime: "120000", }) // Far more announced than sent, so that Go's server answers for the app @@ -869,11 +872,11 @@ func TestAVisitorThatStopsALittleShortGetsTheAppsAnswerAtOnce(t *testing.T) { standIn := startWAF(t, allow) app := startApp(t, answerBeforeTheBody) - // The request time limit is longer than the 10 seconds the status and headers - // of an answer have to be sent, and than dial's deadline. + // Both time limits are far longer than dial's deadline, so that the answer + // has to come before either runs out, and a busy host cannot use them up. addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{ - maxRequestTime: "15000", - maxResponseTime: "300", + maxRequestTime: "120000", + maxResponseTime: "60000", }) conn := dial(t, addr) @@ -910,6 +913,82 @@ func TestAVisitorThatStopsALittleShortIsReadUntilTheRequestTimeLimit(t *testing. } } +func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOff( + t *testing.T, +) { + t.Parallel() + + const limit = 300 * time.Millisecond + + // 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. + for _, tc := range []struct { + name string + parts []string + }{ + {"with the body in one part", []string{greeting}}, + {"with the body in two parts", []string{greeting[:5], 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))) + + _ = rc.Flush() + + <-headersRead + time.Sleep(limit) + + for _, part := range tc.parts { + _, _ = 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", + maxResponseTime: strconv.FormatInt(limit.Milliseconds(), 10), + }) + + conn := dial(t, addr) + sendAllButALittle(t, conn) + + res, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + 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) + } + }) + } +} + func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) { t.Parallel() diff --git a/internal/rproxy/request.go b/internal/rproxy/request.go index c346dae..170b14e 100644 --- a/internal/rproxy/request.go +++ b/internal/rproxy/request.go @@ -384,8 +384,19 @@ func (rq *request) refuse(status int) { func (rq *request) finish(ctx context.Context) { rq.stopTimers() - if rq.refused.Load() == 0 { - rq.stopPassingOnBody() + // After a cut-off answer, the reading of a body that has not arrived whole + // ends now. Go's server would otherwise keep the connection open while it + // reads the rest. A flush fails if the time to send the answer ran out, and + // every flush after a failed one fails too. ReverseProxy ignores the failure, + // but panics if it was still waiting for more of the app's answer; finish runs + // either way. Before a status, a flush would send a 200; after a 101, + // ReverseProxy has taken over the connection. + switch { + case rq.out.status != 0 && rq.out.status != http.StatusSwitchingProtocols && + errors.Is(rq.rc.Flush(), os.ErrDeadlineExceeded): + rq.stopPassingOnBody(time.Now()) + case rq.refused.Load() == 0: + rq.stopPassingOnBody(rq.requestDeadline()) } requestEnd := rq.endRequest() @@ -473,10 +484,10 @@ func (rq *request) stopReadingBody(deadline time.Time) { } // 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 -// SWRP_MAX_REQUEST_TIME_MS runs out. Go's server reads it once the handler -// returns, but with no deadline at all if another read of it is still waiting. -func (rq *request) stopPassingOnBody() { +// 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 +// another read of it is still waiting. +func (rq *request) stopPassingOnBody(deadline time.Time) { if rq.body == nil || rq.body.received.Load() { return } @@ -489,7 +500,7 @@ func (rq *request) stopPassingOnBody() { rq.body.stopped = true rq.body.mu.Unlock() - rq.stopReadingBody(rq.requestDeadline()) + rq.stopReadingBody(deadline) } // startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the