Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
387744ab50 | ||
|
|
78047e52a9 |
No files matched your search
@@ -12,6 +12,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -807,8 +808,10 @@ func TestAVisitorThatStopsSendingAfterTheAppHasAnsweredGetsTheAnswer(t *testing.
|
|||||||
app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
|
app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||||
_, _ = io.WriteString(w, greeting)
|
_, _ = 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{
|
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
|
// 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)
|
standIn := startWAF(t, allow)
|
||||||
app := startApp(t, answerBeforeTheBody)
|
app := startApp(t, answerBeforeTheBody)
|
||||||
// The request time limit is longer than the 10 seconds the status and headers
|
// Both time limits are far longer than dial's deadline, so that the answer
|
||||||
// of an answer have to be sent, and than dial's deadline.
|
// 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{
|
addr, _ := startRProxy(t, standIn.URL, app.URL, map[string]string{
|
||||||
maxRequestTime: "15000",
|
maxRequestTime: "120000",
|
||||||
maxResponseTime: "300",
|
maxResponseTime: "60000",
|
||||||
})
|
})
|
||||||
|
|
||||||
conn := dial(t, addr)
|
conn := dial(t, addr)
|
||||||
@@ -910,6 +913,63 @@ func TestAVisitorThatStopsALittleShortIsReadUntilTheRequestTimeLimit(t *testing.
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAVisitorThatStopsALittleShortHasTheConnectionClosedWhenTheAnswerIsCutOff(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const limit = 300 * time.Millisecond
|
||||||
|
|
||||||
|
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.
|
||||||
|
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)
|
||||||
|
|
||||||
|
_, _ = io.WriteString(w, greeting)
|
||||||
|
_ = 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) {
|
func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -195,6 +195,20 @@ func (rq *request) forward(ctx context.Context) {
|
|||||||
rq.startRequestTimer()
|
rq.startRequestTimer()
|
||||||
rq.upstreamStart = time.Now()
|
rq.upstreamStart = time.Now()
|
||||||
reverseProxy.ServeHTTP(rq.out, out)
|
reverseProxy.ServeHTTP(rq.out, out)
|
||||||
|
|
||||||
|
// ReverseProxy ignores a failed flush, so it misses a short answer cut off when
|
||||||
|
// SWRP_MAX_RESPONSE_TIME_MS runs out; once a flush has failed, every later one
|
||||||
|
// fails too. Go's server would then keep the connection open while it reads
|
||||||
|
// the rest of a body that has not arrived whole. So that reading ends now, and
|
||||||
|
// the refusal keeps finish from extending it to SWRP_MAX_REQUEST_TIME_MS. The
|
||||||
|
// panic has Go's server close the connection, and log nothing; the deferred
|
||||||
|
// finish still runs.
|
||||||
|
if rq.out.status != http.StatusSwitchingProtocols &&
|
||||||
|
errors.Is(rq.rc.Flush(), os.ErrDeadlineExceeded) {
|
||||||
|
rq.refuse(http.StatusGatewayTimeout)
|
||||||
|
rq.stopPassingOnBody(time.Now())
|
||||||
|
panic(http.ErrAbortHandler)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// rewrite makes the request the app receives: the visitor's request, unchanged,
|
// rewrite makes the request the app receives: the visitor's request, unchanged,
|
||||||
@@ -385,7 +399,7 @@ func (rq *request) finish(ctx context.Context) {
|
|||||||
rq.stopTimers()
|
rq.stopTimers()
|
||||||
|
|
||||||
if rq.refused.Load() == 0 {
|
if rq.refused.Load() == 0 {
|
||||||
rq.stopPassingOnBody()
|
rq.stopPassingOnBody(rq.requestDeadline())
|
||||||
}
|
}
|
||||||
|
|
||||||
requestEnd := rq.endRequest()
|
requestEnd := rq.endRequest()
|
||||||
@@ -473,10 +487,10 @@ func (rq *request) stopReadingBody(deadline time.Time) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// stopPassingOnBody ends the transport's reading of a visitor's body that has
|
// 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
|
// not arrived whole, and lets Go's server read the rest until deadline. Go's
|
||||||
// SWRP_MAX_REQUEST_TIME_MS runs out. Go's server reads it once the handler
|
// server reads it once the handler returns, but with no deadline at all if
|
||||||
// returns, but with no deadline at all if another read of it is still waiting.
|
// another read of it is still waiting.
|
||||||
func (rq *request) stopPassingOnBody() {
|
func (rq *request) stopPassingOnBody(deadline time.Time) {
|
||||||
if rq.body == nil || rq.body.received.Load() {
|
if rq.body == nil || rq.body.received.Load() {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -489,7 +503,7 @@ func (rq *request) stopPassingOnBody() {
|
|||||||
rq.body.stopped = true
|
rq.body.stopped = true
|
||||||
rq.body.mu.Unlock()
|
rq.body.mu.Unlock()
|
||||||
|
|
||||||
rq.stopReadingBody(rq.requestDeadline())
|
rq.stopReadingBody(deadline)
|
||||||
}
|
}
|
||||||
|
|
||||||
// startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the
|
// startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the
|
||||||
|
|||||||
Reference in new issue
Block a user