diff --git a/caddytest/integration/request_body_timeout_test.go b/caddytest/integration/request_body_timeout_test.go new file mode 100644 index 000000000..81bb58862 --- /dev/null +++ b/caddytest/integration/request_body_timeout_test.go @@ -0,0 +1,133 @@ +package integration + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/caddyserver/caddy/v2/caddytest" +) + +// A timed-out upload must get an error response without sending the remainder +// of its body. HTTP/1 drains unread bodies smaller than 256 KiB before replying; +// clearing the expired read deadline makes that drain wait for the client. +func TestReverseProxyRequestBodyTimeout(t *testing.T) { + for _, proto := range []int{1, 2} { + for _, tc := range []struct { + name string + timeouts string + trickle bool + handler string + }{ + {"hard stalled", `"read_timeout": 300000000, "read_idle_timeout": 60000000000`, false, ""}, + {"hard trickled", `"read_timeout": 300000000, "read_idle_timeout": 60000000000`, true, ""}, + {"idle stalled", `"read_idle_timeout": 300000000`, false, ""}, + {"minimum rate trickled", `"read_idle_timeout": 300000000, "read_min_rate": 65536`, true, ""}, + {"route stalled", `"read_idle_timeout": 60000000000`, false, `{"handler": "timeouts", "read_timeout": 300000000},`}, + } { + t.Run(fmt.Sprintf("HTTP%d/%s", proto, tc.name), func(t *testing.T) { + type uploadResult struct { + n int64 + err error + } + upstreamRead := make(chan uploadResult, 1) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n, err := io.Copy(io.Discard, r.Body) + upstreamRead <- uploadResult{n, err} + w.WriteHeader(http.StatusNoContent) + })) + defer upstream.Close() + tester := caddytest.NewTester(t) + tester.InitServer(fmt.Sprintf(`{ + "admin": {"listen": "localhost:2999", "config": {"persist": false}}, + "apps": { + "tls": {"certificates": {"load_files": [{ + "certificate": "/caddy.localhost.crt", "key": "/caddy.localhost.key" + }]}}, + "http": {"grace_period": 1, "servers": {"srv0": { + "listen": ["127.0.0.1:9443"], + "automatic_https": {"disable": true}, + "tls_connection_policies": [{}], + %s, + "routes": [{"handle": [%s{"handler": "reverse_proxy", + "upstreams": [{"dial": %q}] + }]}] + }}} + } + }`, tc.timeouts, tc.handler, upstream.Listener.Addr().String()), "json") + if t.Failed() { + t.FailNow() + } + transport := tester.Client.Transport.(*http.Transport) + if proto == 1 { + transport.ForceAttemptHTTP2 = false + } + defer transport.CloseIdleConnections() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + pr, pw := io.Pipe() + defer pr.Close() + defer pw.Close() + stopClose := context.AfterFunc(ctx, func() { pr.CloseWithError(ctx.Err()) }) + defer stopClose() + writerDone := make(chan struct{}) + go func() { + defer close(writerDone) + if _, err := pw.Write(make([]byte, 1024)); err != nil { + return + } + if !tc.trickle { + return // Leave the pipe open so the request body stalls. + } + ticker := time.NewTicker(25 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if _, err := pw.Write([]byte("x")); err != nil { + return + } + } + } + }() + defer func() { + cancel() + pr.Close() + <-writerDone + }() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://a.caddy.localhost:9443", pr) + if err != nil { + t.Fatal(err) + } + req.ContentLength = 100 << 10 // Below net/http's post-handler drain limit. + resp, err := tester.Client.Do(req) + if err != nil { + t.Fatalf("did not receive a response to the timed-out upload: %v", err) + } + defer resp.Body.Close() + if resp.ProtoMajor != proto { + t.Fatalf("got HTTP/%d, want HTTP/%d", resp.ProtoMajor, proto) + } + // A read timeout can also cancel the request context, so the proxy + // can classify it as a canceled request rather than a gateway error. + if resp.StatusCode != 499 && resp.StatusCode != http.StatusBadGateway && resp.StatusCode != http.StatusGatewayTimeout { + t.Errorf("unexpected timeout response status: %d", resp.StatusCode) + } + select { + case result := <-upstreamRead: + if result.err == nil || result.n < 1024 || result.n >= req.ContentLength { + t.Errorf("upstream upload was not interrupted: read %d bytes, error %v", result.n, result.err) + } + case <-ctx.Done(): + t.Fatal("upstream upload was not interrupted") + } + }) + } + } +} diff --git a/modules/caddyhttp/idletimeout.go b/modules/caddyhttp/idletimeout.go index eca2c35b6..2b12fa6ae 100644 --- a/modules/caddyhttp/idletimeout.go +++ b/modules/caddyhttp/idletimeout.go @@ -16,8 +16,10 @@ package caddyhttp import ( "context" + "errors" "io" "net/http" + "os" "sync" "time" @@ -106,8 +108,10 @@ func (d *IdleDeadline) restore(o idleOverride) { // IdleTimeoutReader wraps a request body with IdleDeadline, resetting // the read deadline before every Read call instead of bounding the // whole body transfer with a single hard deadline. -// The deadline is cleared after an error (most likely io.EOF) is encountered when reading the body -// to prevent context canceled error for h1 requests. +// The deadline is cleared after a non-timeout error (most likely io.EOF) is +// encountered when reading the body to prevent context canceled errors for h1 +// requests. An expired deadline is kept when DrainDeadline is set to bound +// net/http's post-handler drain. // see: https://github.com/caddyserver/caddy/issues/8103 type IdleTimeoutReader struct { io.ReadCloser @@ -136,6 +140,7 @@ type IdleTimeoutReader struct { mu sync.Mutex unsupported bool armed armedDeadline + installed time.Time // Last deadline successfully installed through Ctrl. finished bool terminalErr error } @@ -186,8 +191,9 @@ func (r *IdleTimeoutReader) Read(p []byte) (int, error) { // HandlerDone prevents later body reads from using the response controller. // If the body is not finished and DrainDeadline is set, it leaves an idle // deadline armed for net/http's post-handler drain. Otherwise, it clears a -// deadline left from a finished body or put back between reads. It must be -// called before the installing handler returns. +// deadline left from a finished body or put back between reads, unless the +// body timed out with an expired drain deadline. It must be called before the +// installing handler returns. func (r *IdleTimeoutReader) HandlerDone() { r.mu.Lock() defer r.mu.Unlock() @@ -227,7 +233,7 @@ func (r *IdleTimeoutReader) Override(timeout time.Duration, minRate int64) (rest // rearmLocked moves an armed deadline to what the current settings give. func (r *IdleTimeoutReader) rearmLocked() { - if r.armed == deadlineIdle && !r.finished && !r.unsupported { + if r.armed == deadlineIdle && !r.finished && !r.unsupported && r.terminalErr == nil { r.setDeadlineLocked("could not set read deadline") } } @@ -249,6 +255,7 @@ func (r *IdleTimeoutReader) setReadDeadlineLocked(deadline time.Time, logMessage } return false } + r.installed = deadline return true } @@ -278,6 +285,13 @@ func (r *IdleTimeoutReader) clearDeadlineLocked() { if r.armed == deadlineNone { return } + if r.DrainDeadline && !r.installed.IsZero() && !time.Now().Before(r.installed) && + errors.Is(r.terminalErr, os.ErrDeadlineExceeded) { + // HTTP/1 drains a small unread body before writing the response. + // Clearing its expired deadline would let a stalled client block + // that drain, and therefore the error response, indefinitely. + return + } if !r.setReadDeadlineLocked(time.Time{}, "could not clear read deadline") { return } diff --git a/modules/caddyhttp/idletimeout_test.go b/modules/caddyhttp/idletimeout_test.go index 4e3d4aef2..ab45e5ced 100644 --- a/modules/caddyhttp/idletimeout_test.go +++ b/modules/caddyhttp/idletimeout_test.go @@ -26,6 +26,7 @@ import ( "strings" "sync" "testing" + "testing/iotest" "time" "github.com/stretchr/testify/assert" @@ -146,6 +147,133 @@ func TestIdleTimeoutReaderClearsDeadlineAtEOF(t *testing.T) { assert.True(t, deadlines[len(deadlines)-1].IsZero()) } +func TestIdleTimeoutReaderTerminalError(t *testing.T) { + for _, tc := range []struct { + name string + err error + keepDeadline bool + }{ + {"timeout", os.ErrDeadlineExceeded, true}, + {"wrapped timeout", fmt.Errorf("read body: %w", os.ErrDeadlineExceeded), true}, + {"EOF", io.EOF, false}, + {"other error", errors.New("read failed"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + deadline := time.Now().Add(-time.Second) + w := &readDeadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + r := &IdleTimeoutReader{ + ReadCloser: io.NopCloser(iotest.ErrReader(tc.err)), + Ctrl: http.NewResponseController(w), + Deadline: IdleDeadline{Timeout: time.Hour, HardDeadline: deadline}, + Logger: zap.NewNop(), + DrainDeadline: true, + } + _, err := r.Read(make([]byte, 1)) + require.ErrorIs(t, err, tc.err) + want := time.Time{} + if tc.keepDeadline { + want = deadline + } + deadlines := w.snapshot() + require.NotEmpty(t, deadlines) + assert.Equal(t, want, deadlines[len(deadlines)-1], "after Read") + r.HandlerDone() + deadlines = w.snapshot() + assert.Equal(t, want, deadlines[len(deadlines)-1], "after HandlerDone") + + _, err = r.Read(make([]byte, 1)) + require.ErrorIs(t, err, tc.err) + r.HandlerDone() + assert.Equal(t, deadlines, w.snapshot(), "late reads and repeated completion must not change deadlines") + }) + } +} + +// Only an expired deadline successfully installed for a draining request may +// survive a timeout; a failed replacement must not change that decision. +func TestIdleTimeoutReaderTimeoutDeadlineCleanup(t *testing.T) { + for _, tc := range []struct { + name string + drain bool + initial time.Duration + replacement time.Duration + failAt int + wrapped bool + wantClear bool + wantCalls int + }{ + {name: "early timeout", drain: true, initial: time.Hour, wantClear: true, wantCalls: 2}, + {name: "wrapped early timeout", drain: true, initial: time.Hour, wrapped: true, wantClear: true, wantCalls: 2}, + {name: "expired without drain", initial: -time.Hour, wantClear: true, wantCalls: 2}, + {name: "initial installation fails", drain: true, initial: -time.Hour, failAt: 1, wantCalls: 1}, + {name: "failed expired replacement", drain: true, initial: time.Hour, replacement: -time.Hour, failAt: 2, wantClear: true, wantCalls: 3}, + {name: "failed future replacement", drain: true, initial: -time.Hour, replacement: time.Hour, failAt: 2, wantCalls: 2}, + {name: "failed clear retries at completion", drain: true, initial: time.Hour, failAt: 2, wantClear: true, wantCalls: 3}, + } { + t.Run(tc.name, func(t *testing.T) { + w := &readDeadlineRecorder{ResponseRecorder: httptest.NewRecorder(), failAt: tc.failAt} + terminalErr := error(os.ErrDeadlineExceeded) + if tc.wrapped { + terminalErr = fmt.Errorf("read body: %w", terminalErr) + } + r := &IdleTimeoutReader{ + ReadCloser: io.NopCloser(iotest.ErrReader(terminalErr)), + Ctrl: http.NewResponseController(w), + Deadline: IdleDeadline{Timeout: tc.initial}, + Logger: zap.NewNop(), DrainDeadline: tc.drain, + } + // Install once, then exercise a failed replacement during Read. + if tc.replacement != 0 { + r.setDeadlineLocked("initial deadline") + r.Deadline.Timeout = tc.replacement + } + _, err := r.Read(make([]byte, 1)) + require.ErrorIs(t, err, terminalErr) + afterRead := w.snapshot() + if tc.wantClear { + require.True(t, afterRead[len(afterRead)-1].IsZero(), "must attempt clearing after Read") + } + r.HandlerDone() + deadlines := w.snapshot() + require.Len(t, deadlines, tc.wantCalls) + if tc.wantClear { + assert.True(t, deadlines[len(deadlines)-1].IsZero()) + } else if tc.failAt != 1 { + assert.True(t, deadlines[0].Before(time.Now()), "the successfully installed deadline expired") + } + _, err = r.Read(make([]byte, 1)) + require.ErrorIs(t, err, terminalErr) + r.HandlerDone() + assert.Equal(t, deadlines, w.snapshot(), "late reads and completion must leave deadlines alone") + }) + } +} + +func TestIdleTimeoutReaderTimeoutAfterOverride(t *testing.T) { + w := &readDeadlineRecorder{ResponseRecorder: httptest.NewRecorder()} + r := &IdleTimeoutReader{ + ReadCloser: io.NopCloser(iotest.ErrReader(os.ErrDeadlineExceeded)), + Ctrl: http.NewResponseController(w), + Deadline: IdleDeadline{Timeout: time.Hour}, + Logger: zap.NewNop(), + DrainDeadline: true, + } + restore := r.Override(-time.Second, 0) + _, err := r.Read(make([]byte, 1)) + require.ErrorIs(t, err, os.ErrDeadlineExceeded) + deadlines := w.snapshot() + require.NotEmpty(t, deadlines) + expired := deadlines[0] + require.True(t, expired.Before(time.Now())) + assert.Equal(t, expired, deadlines[len(deadlines)-1]) + + // Unwinding a route-level timeout must not rearm the longer server timeout. + restore() + r.HandlerDone() + deadlines = w.snapshot() + assert.Equal(t, expired, deadlines[len(deadlines)-1]) +} + func TestIdleTimeoutReaderHandlerDone(t *testing.T) { body := &blockingReadCloser{ started: make(chan struct{}),