From c77c1b38409de3f3faf16309c108d0040e9e32cb Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Fri, 31 Jul 2026 17:23:19 +0200 Subject: [PATCH] fix(evmrpc): stream request-body budget charging to close slowloris gap (PLT-780) Charge the global request-byte budget incrementally as bytes are read instead of reserving Content-Length up front, so a stalled/slow body can no longer pin the shared budget while barely sending data. Adds a per-chunk body read idle timeout (408) as a backstop independent of the overall ReadTimeout, and moves JWT ahead of the byte limiter so unauthenticated clients can't touch the budget at all. Also fixes two issues found while validating the fix: seiLegacyHTTPGate (always present in production) was writing its own response on a failed body read, masking the limiter's 429/408 behind a 400 and leaking internal error text; and the gate closing the body right after buffering it was releasing the budget before the request finished processing instead of holding it for the whole request. Co-Authored-By: Claude Sonnet 5 --- evmrpc/AGENTS.md | 21 +++ evmrpc/config/config.go | 32 +++- evmrpc/config/config_fuzz_test.go | 1 + evmrpc/config/config_test.go | 5 + evmrpc/config/testdata/evm.golden | 1 + evmrpc/metrics.go | 7 +- evmrpc/request_limiter.go | 240 ++++++++++++++++++++++--- evmrpc/request_limiter_test.go | 290 ++++++++++++++++++++++++++++-- evmrpc/rpcstack.go | 13 +- evmrpc/sei_legacy_test.go | 2 +- evmrpc/server.go | 1 + 11 files changed, 558 insertions(+), 55 deletions(-) diff --git a/evmrpc/AGENTS.md b/evmrpc/AGENTS.md index 315eef3f5b..5e9037d5b5 100644 --- a/evmrpc/AGENTS.md +++ b/evmrpc/AGENTS.md @@ -1,6 +1,27 @@ # EVM RPC SPECS EVM RPCs live under `evmrpc/` folder. +## HTTP middleware order (JSON-RPC) + +When JWT is configured, unauthenticated requests are rejected before the byte +budget is touched: + +``` +jwt → requestSizeLimiter → seiLegacyHTTPGate → gzip → vhost → cors → rpc.Server +``` + +Without JWT: + +``` +requestSizeLimiter → seiLegacyHTTPGate → gzip → vhost → cors → rpc.Server +``` + +`requestSizeLimiter` caps each body with `http.MaxBytesReader`, charges the +global `max_concurrent_request_bytes` budget incrementally as body bytes are +read (64 KiB batches), and enforces `body_read_idle_timeout` between body +chunks via `http.ResponseController.SetReadDeadline` (HTTP 408 on stall, HTTP +429 on mid-read budget exhaustion). + EVM RPCs prefixed by `eth_` and `debug_` on Sei generally follows [Ethereum's spec](https://www.quicknode.com/docs/ethereum/api-overview). However, there are some notable distinctions. - **Pending** - Sei has instant finality and thus has no concept of `pending` blocks. However, the RPCs still accept `pending` for applicable parameters, and will treat it equivalent to `final`/`safe`/`latest`. diff --git a/evmrpc/config/config.go b/evmrpc/config/config.go index e645bc57f9..bfea7664ac 100644 --- a/evmrpc/config/config.go +++ b/evmrpc/config/config.go @@ -257,12 +257,17 @@ type Config struct { MaxRequestBodyBytes int64 `mapstructure:"max_request_body_bytes"` // MaxConcurrentRequestBytes bounds the total size, in bytes, of HTTP - // JSON-RPC request bodies admitted for processing concurrently, weighted by - // each request's Content-Length. Requests that would exceed the budget are - // rejected fast (HTTP 429) before decode, capping peak memory under load. - // Set to 0 to disable the limit. + // JSON-RPC request bodies admitted for processing concurrently, charged + // incrementally as body bytes are read. Requests that would exceed the + // budget mid-read are rejected (HTTP 429). Set to 0 to disable the limit. MaxConcurrentRequestBytes int64 `mapstructure:"max_concurrent_request_bytes"` + // BodyReadIdleTimeout is the maximum idle time allowed between body chunks + // while reading an HTTP JSON-RPC request. Stalled body reads are cut with + // HTTP 408 and release any byte budget held so far. Zero disables the + // per-chunk idle guard (http.Server ReadTimeout remains the backstop). + BodyReadIdleTimeout time.Duration `mapstructure:"body_read_idle_timeout"` + // MaxOpenConnections caps the number of simultaneously accepted connections // on the EVM HTTP and WebSocket listeners. The limit is applied per listener // (HTTP and WS each get their own budget). Excess connections block in the @@ -327,6 +332,7 @@ var DefaultConfig = Config{ MaxRequestBodyBytes: 5 * 1024 * 1024, // 5 MiB (matches go-ethereum rpc default body limit) MaxConcurrentRequestBytes: 128 * 1024 * 1024, // 128 MiB of request bodies admitted concurrently MaxOpenConnections: 2000, + BodyReadIdleTimeout: 10 * time.Second, } const ( @@ -382,6 +388,7 @@ const ( flagMaxRequestBodyBytes = "evm.max_request_body_bytes" flagMaxConcurrentRequestBytes = "evm.max_concurrent_request_bytes" flagMaxOpenConnections = "evm.max_open_connections" + flagBodyReadIdleTimeout = "evm.body_read_idle_timeout" ) func ReadConfig(opts servertypes.AppOptions) (Config, error) { @@ -659,6 +666,11 @@ func ReadConfig(opts servertypes.AppOptions) (Config, error) { return cfg, fmt.Errorf("%s must be >= 0 (0 disables the limit), got %d", flagMaxOpenConnections, cfg.MaxOpenConnections) } } + if v := opts.Get(flagBodyReadIdleTimeout); v != nil { + if cfg.BodyReadIdleTimeout, err = cast.ToDurationE(v); err != nil { + return cfg, err + } + } return cfg, nil } @@ -922,11 +934,17 @@ batch_response_max_size = {{ .EVM.BatchResponseMaxSize }} max_request_body_bytes = {{ .EVM.MaxRequestBodyBytes }} # max_concurrent_request_bytes bounds the total size, in bytes, of HTTP JSON-RPC -# request bodies admitted for processing concurrently (weighted by each request's -# Content-Length). Requests that would exceed the budget are rejected fast -# (HTTP 429) before decode, capping peak memory under load. Set to 0 to disable. +# request bodies admitted for processing concurrently, charged incrementally as +# body bytes are read. Requests that would exceed the budget mid-read are +# rejected (HTTP 429). Set to 0 to disable. max_concurrent_request_bytes = {{ .EVM.MaxConcurrentRequestBytes }} +# body_read_idle_timeout is the maximum idle time allowed between body chunks +# while reading an HTTP JSON-RPC request. Stalled body reads return HTTP 408 and +# release any byte budget held so far. Set to 0 to disable (ReadTimeout remains +# the whole-request backstop). +body_read_idle_timeout = "{{ .EVM.BodyReadIdleTimeout }}" + # max_open_connections caps the number of simultaneously accepted connections on # the EVM HTTP and WebSocket listeners. Set to 0 to disable the limit. max_open_connections = {{ .EVM.MaxOpenConnections }} diff --git a/evmrpc/config/config_fuzz_test.go b/evmrpc/config/config_fuzz_test.go index 1b21dcf8da..7632afae46 100644 --- a/evmrpc/config/config_fuzz_test.go +++ b/evmrpc/config/config_fuzz_test.go @@ -93,6 +93,7 @@ var evmKeys = []configtest.KeySpec{ {Key: "evm.batch_response_max_size", Path: "BatchResponseMaxSize", Cast: configtest.CastInt, Checked: true}, {Key: "evm.max_request_body_bytes", Path: "MaxRequestBodyBytes", Cast: configtest.CastInt64, Checked: true}, {Key: "evm.max_concurrent_request_bytes", Path: "MaxConcurrentRequestBytes", Cast: configtest.CastInt64, Checked: true}, + {Key: "evm.body_read_idle_timeout", Path: "BodyReadIdleTimeout", Cast: configtest.CastDuration, Checked: true}, } func readEVM(opts configtest.AppOpts) (any, error) { return config.ReadConfig(opts) } diff --git a/evmrpc/config/config_test.go b/evmrpc/config/config_test.go index baa061f58f..b6737b72b0 100644 --- a/evmrpc/config/config_test.go +++ b/evmrpc/config/config_test.go @@ -49,6 +49,7 @@ type opts struct { maxRequestBodyBytes interface{} maxConcurrentRequestBytes interface{} maxOpenConnections interface{} + bodyReadIdleTimeout interface{} maxTraceStructLogBytes interface{} traceAllowedTracers interface{} traceAllowJSTracers interface{} @@ -192,6 +193,9 @@ func (o *opts) Get(k string) interface{} { if k == "evm.max_open_connections" { return o.maxOpenConnections } + if k == "evm.body_read_idle_timeout" { + return o.bodyReadIdleTimeout + } if k == "evm.max_trace_struct_log_bytes" { return o.maxTraceStructLogBytes } @@ -253,6 +257,7 @@ func getDefaultOpts() opts { int64(5 * 1024 * 1024), int64(128 * 1024 * 1024), 2000, + 10 * time.Second, uint64(256 * 1024 * 1024), []string{"callTracer", "prestateTracer"}, false, diff --git a/evmrpc/config/testdata/evm.golden b/evmrpc/config/testdata/evm.golden index d75b19fb1a..bcbef7bf37 100644 --- a/evmrpc/config/testdata/evm.golden +++ b/evmrpc/config/testdata/evm.golden @@ -56,4 +56,5 @@ BatchRequestLimit = int(1000) BatchResponseMaxSize = int(25000000) MaxRequestBodyBytes = int64(5242880) MaxConcurrentRequestBytes = int64(134217728) +BodyReadIdleTimeout = time.Duration(10s) MaxOpenConnections = int(2000) diff --git a/evmrpc/metrics.go b/evmrpc/metrics.go index ead49300e2..2e81dceaf9 100644 --- a/evmrpc/metrics.go +++ b/evmrpc/metrics.go @@ -19,8 +19,9 @@ const ( jsonrpcCodeKey = "jsonrpc_code" rejectReasonKey = "reason" // reject reason values for requestRejectedCount. - rejectReasonOversize = "oversize" // body exceeded max_request_body_bytes - rejectReasonBusy = "busy" // max_concurrent_request_bytes budget exhausted + rejectReasonOversize = "oversize" // body exceeded max_request_body_bytes + rejectReasonBudgetMidread = "budget_midread" // global byte budget exhausted mid-body read + rejectReasonSlowBody = "slow_body" // body read idle timeout exceeded // error_class values; empty string ("") means success. errorClassPanic = "panic" errorClassExecutionReverted = "execution_reverted" @@ -167,7 +168,7 @@ func recordHistoricalDebugTraceAttempt(ctx context.Context, endpoint, connection } // recordRequestRejected counts an HTTP JSON-RPC request dropped by pre-decode -// admission control. reason is one of rejectReasonOversize / rejectReasonBusy. +// admission control. reason is one of rejectReasonOversize / rejectReasonBudgetMidread / rejectReasonSlowBody. // No endpoint dimension is recorded: the rejection happens before the JSON-RPC // method is decoded, so it is not yet known. func recordRequestRejected(ctx context.Context, reason string) { diff --git a/evmrpc/request_limiter.go b/evmrpc/request_limiter.go index b138df93ff..03ab9f2c6c 100644 --- a/evmrpc/request_limiter.go +++ b/evmrpc/request_limiter.go @@ -1,7 +1,12 @@ package evmrpc import ( + "errors" + "io" + "net" "net/http" + "os" + "time" "golang.org/x/sync/semaphore" ) @@ -11,21 +16,26 @@ import ( // max_request_body_bytes is left at 0 ("use the default"). const defaultMaxRequestBodyBytes int64 = 5 * 1024 * 1024 +// budgetAcquireBatch is the byte step for incremental global-budget accounting. +// Batching limits semaphore contention on large uploads while still bounding what +// a slow/stalled body pins (at most one batch, not the declared Content-Length). +const budgetAcquireBatch int64 = 64 * 1024 + // requestSizeLimiter is an HTTP middleware that bounds peak decode-time memory by -// admitting JSON-RPC requests *before* the body is buffered or decoded. It enforces: +// admitting JSON-RPC request bodies incrementally as bytes are read. It enforces: // // - maxBody: per-request body cap. Over-Content-Length requests get 413; the body is // also wrapped in http.MaxBytesReader so chunked / mis-declared bodies can't exceed it. -// - budget: a size-weighted global semaphore bounding bytes admitted concurrently; -// over-budget requests get 429. The reservation is held for the full inner request -// (not just decode) and trusts the declared Content-Length — conservative by design, -// so slow or trickled requests can exhaust the budget even when little is buffered. -// Actual bytes stay hard-capped by MaxBytesReader; per-IP rate limiting will be -// wired in front of this handler. +// - budget: a global semaphore charged in batches as body bytes arrive; over-budget +// reads get 429 mid-body. Stalled uploads hold at most one batch, not declared size. +// Charged bytes stay reserved for the whole inner request, not just the read. +// - bodyReadIdleTimeout: per-chunk idle guard via ResponseController.SetReadDeadline; +// stalled body reads get 408 and release any budget held so far. type requestSizeLimiter struct { - inner http.Handler - maxBody int64 // always > 0 after construction - budget *semaphore.Weighted // nil when the global budget is disabled + inner http.Handler + maxBody int64 // always > 0 after construction + budget *semaphore.Weighted + bodyReadIdleTimeout time.Duration } // newRequestSizeLimiter wraps inner with pre-decode admission control. maxBody <= 0 @@ -33,11 +43,15 @@ type requestSizeLimiter struct { // 0 means "use the default", never "no cap"). maxConcurrentBytes <= 0 disables the // global budget. If a positive budget is smaller than maxBody it is raised to maxBody // so that a single maximum-size request can always be admitted. -func newRequestSizeLimiter(inner http.Handler, maxBody, maxConcurrentBytes int64) http.Handler { +func newRequestSizeLimiter(inner http.Handler, maxBody, maxConcurrentBytes int64, bodyReadIdleTimeout time.Duration) http.Handler { if maxBody <= 0 { maxBody = defaultMaxRequestBodyBytes } - l := &requestSizeLimiter{inner: inner, maxBody: maxBody} + l := &requestSizeLimiter{ + inner: inner, + maxBody: maxBody, + bodyReadIdleTimeout: bodyReadIdleTimeout, + } if maxConcurrentBytes > 0 { if maxConcurrentBytes < maxBody { maxConcurrentBytes = maxBody @@ -57,23 +71,195 @@ func (l *requestSizeLimiter) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Backstop for chunked / mis-declared bodies: cap the bytes actually readable. r.Body = http.MaxBytesReader(w, r.Body, l.maxBody) - if l.budget != nil { - // For requests with a known length Go's HTTP server enforces Content-Length, - // so it is a sound upper bound on bytes read (and 0 means an empty body that - // reserves nothing). Chunked / unknown-length requests report ContentLength - // == -1; reserve the worst case (maxBody) for them. weight is therefore always - // in [0, maxBody], and the budget is >= maxBody, so the request can always fit. - weight := r.ContentLength - if weight < 0 { - weight = l.maxBody + var outcome limiterOutcome + var budgetWrapped *budgetBody + if l.budget != nil || l.bodyReadIdleTimeout > 0 { + rc := http.NewResponseController(w) + budgetWrapped = &budgetBody{ + inner: r.Body, + budget: l.budget, + rc: rc, + idleTimeout: l.bodyReadIdleTimeout, + outcome: &outcome, + } + r.Body = budgetWrapped + } + + // cw suppresses the inner handler's own response once outcome is set, so the + // status/message below always wins over whatever the inner handler wrote. + cw := &captureResponseWriter{ResponseWriter: w, outcome: &outcome} + l.inner.ServeHTTP(cw, r) + + if budgetWrapped != nil { + // Release only after the inner handler chain fully returns, so the budget + // stays held for the whole request even if an inner handler (e.g. the + // sei-legacy gate) closes the body early after buffering it. + _ = budgetWrapped.Close() + budgetWrapped.release() + } + + if outcome.status != 0 && !cw.wroteHeader { + recordRequestRejected(r.Context(), outcome.reason) + http.Error(w, outcome.message, outcome.status) + } +} + +type limiterOutcome struct { + status int + message string + reason string +} + +// budgetBody wraps r.Body to charge the global byte budget incrementally and to +// enforce a per-chunk body read idle timeout via ResponseController. +type budgetBody struct { + inner io.ReadCloser + budget *semaphore.Weighted + rc *http.ResponseController + idleTimeout time.Duration + outcome *limiterOutcome + + reserved int64 // bytes charged to the global semaphore + unbilled int64 // bytes read since the last batch charge +} + +func (b *budgetBody) Read(p []byte) (int, error) { + if b.idleTimeout > 0 && b.rc != nil { + if err := b.rc.SetReadDeadline(time.Now().Add(b.idleTimeout)); err != nil { + // ResponseController may be unavailable on exotic ResponseWriters; proceed + // without the idle guard rather than failing the request. + } + } + + n, err := b.inner.Read(p) + if n > 0 && b.budget != nil { + if chargeErr := b.charge(int64(n)); chargeErr != nil { + b.fail(rejectReasonBudgetMidread, http.StatusTooManyRequests, "server busy", chargeErr) + return n, chargeErr } - if !l.budget.TryAcquire(weight) { - recordRequestRejected(r.Context(), rejectReasonBusy) - http.Error(w, "server busy", http.StatusTooManyRequests) - return + } + if err != nil { + if errors.Is(err, io.EOF) { + if flushErr := b.flush(); flushErr != nil { + b.fail(rejectReasonBudgetMidread, http.StatusTooManyRequests, "server busy", flushErr) + return n, flushErr + } + return n, err + } + if isReadIdleTimeout(err) { + b.fail(rejectReasonSlowBody, http.StatusRequestTimeout, "request timeout", err) + return n, err } - defer l.budget.Release(weight) + b.release() } + return n, err +} + +// Close stops the body and charges any trailing unflushed bytes. It does not release +// the budget; requestSizeLimiter does that once, after the inner handler returns. +func (b *budgetBody) Close() error { + if b.inner == nil { + return nil + } + inner := b.inner + b.inner = nil + if b.budget != nil { + _ = b.flush() + } + return inner.Close() +} + +func (b *budgetBody) charge(n int64) error { + b.unbilled += n + for b.unbilled >= budgetAcquireBatch { + if !b.budget.TryAcquire(budgetAcquireBatch) { + return errBudgetExhausted + } + b.reserved += budgetAcquireBatch + b.unbilled -= budgetAcquireBatch + } + return nil +} + +func (b *budgetBody) flush() error { + if b.unbilled == 0 { + return nil + } + if !b.budget.TryAcquire(b.unbilled) { + return errBudgetExhausted + } + b.reserved += b.unbilled + b.unbilled = 0 + return nil +} + +func (b *budgetBody) release() { + if b.budget != nil && b.reserved > 0 { + b.budget.Release(b.reserved) + b.reserved = 0 + } + b.unbilled = 0 +} + +func (b *budgetBody) fail(reason string, status int, message string, readErr error) { + b.release() + if b.inner != nil { + _ = b.inner.Close() + b.inner = nil + } + if b.outcome != nil && b.outcome.status == 0 { + b.outcome.status = status + b.outcome.message = message + b.outcome.reason = reason + } + _ = readErr +} + +var errBudgetExhausted = errors.New("request byte budget exhausted") + +func isReadIdleTimeout(err error) bool { + if errors.Is(err, os.ErrDeadlineExceeded) { + return true + } + var netErr net.Error + return errors.As(err, &netErr) && netErr.Timeout() +} + +// captureResponseWriter tracks whether the inner handler wrote a response, and once +// outcome is set, suppresses further inner-handler writes so its response can't +// leak past the caller's canonical status/message. +type captureResponseWriter struct { + http.ResponseWriter + outcome *limiterOutcome + wroteHeader bool +} + +func (w *captureResponseWriter) suppressed() bool { + return w.outcome != nil && w.outcome.status != 0 +} + +func (w *captureResponseWriter) WriteHeader(statusCode int) { + if w.suppressed() { + return + } + w.wroteHeader = true + w.ResponseWriter.WriteHeader(statusCode) +} + +func (w *captureResponseWriter) Write(b []byte) (int, error) { + if w.suppressed() { + return len(b), nil + } + w.wroteHeader = true + return w.ResponseWriter.Write(b) +} + +func (w *captureResponseWriter) Flush() { + if f, ok := w.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} - l.inner.ServeHTTP(w, r) +func (w *captureResponseWriter) Unwrap() http.ResponseWriter { + return w.ResponseWriter } diff --git a/evmrpc/request_limiter_test.go b/evmrpc/request_limiter_test.go index 9f5af56248..598233cab0 100644 --- a/evmrpc/request_limiter_test.go +++ b/evmrpc/request_limiter_test.go @@ -1,12 +1,17 @@ package evmrpc import ( + "bufio" + "fmt" "io" + "net" "net/http" "net/http/httptest" "strings" "sync" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -34,7 +39,7 @@ func TestRequestSizeLimiter(t *testing.T) { h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ran = true w.WriteHeader(http.StatusOK) - }), 1024, 4096) + }), 1024, 4096, 0) rec := httptest.NewRecorder() h.ServeHTTP(rec, newSizedRequest("hello", 5)) @@ -48,7 +53,7 @@ func TestRequestSizeLimiter(t *testing.T) { h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { b, _ := io.ReadAll(r.Body) bodyRead = len(b) > 0 - }), 100, 0) + }), 100, 0, 0) rec := httptest.NewRecorder() h.ServeHTTP(rec, newSizedRequest(strings.Repeat("x", 200), 200)) @@ -58,7 +63,7 @@ func TestRequestSizeLimiter(t *testing.T) { }) t.Run("zero maxBody uses default cap", func(t *testing.T) { - l, ok := newRequestSizeLimiter(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), 0, 0).(*requestSizeLimiter) + l, ok := newRequestSizeLimiter(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), 0, 0, 0).(*requestSizeLimiter) require.True(t, ok) require.Equal(t, defaultMaxRequestBodyBytes, l.maxBody) }) @@ -69,7 +74,7 @@ func TestRequestSizeLimiter(t *testing.T) { h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { innerRan = true _, readErr = io.ReadAll(r.Body) - }), 100, 0) + }), 100, 0, 0) rec := httptest.NewRecorder() h.ServeHTTP(rec, newSizedRequest(strings.Repeat("x", 500), -1)) @@ -80,15 +85,16 @@ func TestRequestSizeLimiter(t *testing.T) { t.Run("raises misconfigured budget to maxBody", func(t *testing.T) { const maxBody int64 = 1000 - l, ok := newRequestSizeLimiter(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), maxBody, 500).(*requestSizeLimiter) + l, ok := newRequestSizeLimiter(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), maxBody, 500, 0).(*requestSizeLimiter) require.True(t, ok) require.NotNil(t, l.budget) ran := false h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ran = true + _, _ = io.ReadAll(r.Body) w.WriteHeader(http.StatusOK) - }), maxBody, 500) + }), maxBody, 500, 0) rec := httptest.NewRecorder() h.ServeHTTP(rec, newSizedRequest(strings.Repeat("x", int(maxBody)), maxBody)) @@ -96,23 +102,46 @@ func TestRequestSizeLimiter(t *testing.T) { require.True(t, ran) require.Equal(t, http.StatusOK, rec.Code) }) + + t.Run("known Content-Length zero body reserves nothing", func(t *testing.T) { + ran := false + h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ran = true + _, err := io.ReadAll(r.Body) + require.NoError(t, err) + w.WriteHeader(http.StatusOK) + }), 1024, 1024, 0) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, newSizedRequest("", 0)) + + require.True(t, ran) + require.Equal(t, http.StatusOK, rec.Code) + }) } func TestRequestSizeLimiter_budgetExhaustionAndRelease(t *testing.T) { const maxBody = 1000 - const budget = 1500 // room for exactly one max-size request at a time + const budget = 1500 // room for exactly one max-size body at a time release := make(chan struct{}) admitted := make(chan struct{}, 1) var innerCalls int h := newRequestSizeLimiter( - blockUntilRelease(release, func() { + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { innerCalls++ + _, err := io.ReadAll(r.Body) + if err != nil { + return + } admitted <- struct{}{} + <-release + w.WriteHeader(http.StatusOK) }), maxBody, budget, + 0, ) oversizeBody := strings.Repeat("x", maxBody) makeRequest := func() *httptest.ResponseRecorder { @@ -121,22 +150,22 @@ func TestRequestSizeLimiter_budgetExhaustionAndRelease(t *testing.T) { return rec } - // First request acquires weight=1000, leaving 500 < 1000. + // First request charges 1000 bytes from the budget, leaving 500 free. firstDone := make(chan int, 1) go func() { firstDone <- makeRequest().Code }() - <-admitted //wait until the first request is processed + <-admitted - // Second request needs 1000 but only 500 remains. + // Second max-size request fails mid-read once it tries to charge the full body. rec2 := makeRequest() require.Equal(t, http.StatusTooManyRequests, rec2.Code) - require.Equal(t, 1, innerCalls, "rejected request must not reach inner handler") + require.Equal(t, 2, innerCalls, "rejected request reaches inner handler but fails on body read") - close(release) // closed channel unblocks the first request and any later admitted ones + close(release) require.Equal(t, http.StatusOK, <-firstDone) - // Budget was released; a new request fits again. + // Budget was released when the first handler returned; a new request fits again. require.Equal(t, http.StatusOK, makeRequest().Code) - require.Equal(t, 2, innerCalls) + require.Equal(t, 3, innerCalls) } func TestRequestSizeLimiter_budgetDisabled(t *testing.T) { @@ -149,6 +178,7 @@ func TestRequestSizeLimiter_budgetDisabled(t *testing.T) { blockUntilRelease(release, admitted.Done), maxBody, 0, // budget disabled + 0, ) oversizeBody := strings.Repeat("x", maxBody) @@ -166,3 +196,233 @@ func TestRequestSizeLimiter_budgetDisabled(t *testing.T) { require.Equal(t, http.StatusOK, <-codes) require.Equal(t, http.StatusOK, <-codes) } + +func TestRequestSizeLimiter_slowlorisDoesNotPinDeclaredSize(t *testing.T) { + const maxBody = 1000 + const budget = 1500 + const stallers = 2 // each would have pinned maxBody under the old design + + stallRelease := make(chan struct{}) + var stallersAdmitted int32 + + h := newRequestSizeLimiter( + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := atomic.AddInt32(&stallersAdmitted, 1) + if n <= stallers { + buf := make([]byte, 32) + _, _ = r.Body.Read(buf) // trickle a few bytes, then stall + <-stallRelease + return + } + _, _ = io.ReadAll(r.Body) + w.WriteHeader(http.StatusOK) + }), + maxBody, + budget, + 0, + ) + + startStaller := func() { + go func() { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, newSizedRequest(strings.Repeat("x", maxBody), maxBody)) + }() + } + for range stallers { + startStaller() + } + require.Eventually(t, func() bool { + return atomic.LoadInt32(&stallersAdmitted) == stallers + }, time.Second, 10*time.Millisecond) + + // Small request should still fit: stallers hold at most 32 bytes each, not maxBody. + rec := httptest.NewRecorder() + h.ServeHTTP(rec, newSizedRequest("ok", 2)) + require.Equal(t, http.StatusOK, rec.Code) + + close(stallRelease) +} + +func TestRequestSizeLimiter_incrementalBudgetOnLargeBody(t *testing.T) { + const maxBody = 200 * 1024 + const budget = maxBody + + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Len(t, body, int(maxBody)) + w.WriteHeader(http.StatusOK) + }) + + rec := httptest.NewRecorder() + h := newRequestSizeLimiter(inner, maxBody, budget, 0) + h.ServeHTTP(rec, newSizedRequest(strings.Repeat("b", int(maxBody)), maxBody)) + require.Equal(t, http.StatusOK, rec.Code) +} + +func TestRequestSizeLimiter_bodyReadIdleTimeout(t *testing.T) { + idle := 50 * time.Millisecond + var completed atomic.Bool + + srv := httptest.NewServer(newRequestSizeLimiter( + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, err := io.ReadAll(r.Body) + if err == nil { + completed.Store(true) + w.WriteHeader(http.StatusOK) + } + }), + 1024, + 4096, + idle, + )) + t.Cleanup(srv.Close) + + conn, err := net.Dial("tcp", srv.Listener.Addr().String()) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + host := srv.Listener.Addr().String() + req := fmt.Sprintf("POST / HTTP/1.1\r\nHost: %s\r\nContent-Length: 100\r\n\r\nx", host) + _, err = conn.Write([]byte(req)) + require.NoError(t, err) + + time.Sleep(idle + 150*time.Millisecond) + + resp, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodPost}) + require.NoError(t, err) + t.Cleanup(func() { _ = resp.Body.Close() }) + require.Equal(t, http.StatusRequestTimeout, resp.StatusCode) + require.False(t, completed.Load()) +} + +func TestRequestSizeLimiter_steadySlowUploadSucceeds(t *testing.T) { + idle := 200 * time.Millisecond + body := strings.Repeat("z", 512) + + srv := httptest.NewServer(newRequestSizeLimiter( + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got, err := io.ReadAll(r.Body) + require.NoError(t, err) + require.Equal(t, body, string(got)) + w.WriteHeader(http.StatusOK) + }), + 1024, + 4096, + idle, + )) + t.Cleanup(srv.Close) + + pr, pw := io.Pipe() + go func() { + defer pw.Close() + chunk := []byte(body) + for len(chunk) > 0 { + n := min(64, len(chunk)) + _, _ = pw.Write(chunk[:n]) + chunk = chunk[n:] + time.Sleep(idle / 4) + } + }() + + req, err := http.NewRequest(http.MethodPost, srv.URL, pr) + require.NoError(t, err) + req.ContentLength = int64(len(body)) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + t.Cleanup(func() { _ = resp.Body.Close() }) + require.Equal(t, http.StatusOK, resp.StatusCode) +} + +func TestJWTBeforeRequestSizeLimiter(t *testing.T) { + var innerCalls int + secret := []byte("secret") + + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + innerCalls++ + w.WriteHeader(http.StatusOK) + }) + limiter := newRequestSizeLimiter(inner, 1024, 1024, 0) + stack := newJWTHandler(secret, limiter) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("{}")) + stack.ServeHTTP(rec, req) + + require.Equal(t, http.StatusUnauthorized, rec.Code) + require.Equal(t, 0, innerCalls, "unauthenticated request must not reach the byte limiter or inner handler") +} + +// TestRequestSizeLimiter_budgetOutcomeSurvivesInnerHandlerWrite runs the limiter through +// seiLegacyHTTPGate (always present in production) to check two things the gate can +// break: the limiter's own 429/408 must win over the gate's response, and the budget +// must stay held through the first request's processing, not just its body read. +func TestRequestSizeLimiter_budgetOutcomeSurvivesInnerHandlerWrite(t *testing.T) { + const maxBody = 1000 + const budget = 1500 + + release := make(chan struct{}) + admitted := make(chan struct{}, 1) + + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + admitted <- struct{}{} + <-release + w.WriteHeader(http.StatusOK) + }) + gated := wrapSeiLegacyHTTP(inner, map[string]struct{}{}, maxBody) + h := newRequestSizeLimiter(gated, maxBody, budget, 0) + + makeRequest := func() *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, newSizedRequest(strings.Repeat("x", maxBody), maxBody)) + return rec + } + + firstDone := make(chan int, 1) + go func() { firstDone <- makeRequest().Code }() + <-admitted + + rec := makeRequest() + require.Equal(t, http.StatusTooManyRequests, rec.Code, "gate's own 400 must not mask the limiter's 429") + require.NotContains(t, rec.Body.String(), "budget exhausted", "internal error text must not leak to the client") + + close(release) + require.Equal(t, http.StatusOK, <-firstDone) +} + +func BenchmarkRequestSizeLimiter_smallPOST(b *testing.B) { + body := `{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":1}` + h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.ReadAll(r.Body) + w.WriteHeader(http.StatusOK) + }), defaultMaxRequestBodyBytes, 128*1024*1024, 10*time.Second) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, newSizedRequest(body, int64(len(body)))) + if rec.Code != http.StatusOK { + b.Fatalf("unexpected status %d", rec.Code) + } + } +} + +func BenchmarkRequestSizeLimiter_maxBodyUpload(b *testing.B) { + payload := strings.Repeat("x", int(defaultMaxRequestBodyBytes)) + h := newRequestSizeLimiter(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.ReadAll(r.Body) + w.WriteHeader(http.StatusOK) + }), defaultMaxRequestBodyBytes, 128*1024*1024, 10*time.Second) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, newSizedRequest(payload, defaultMaxRequestBodyBytes)) + if rec.Code != http.StatusOK { + b.Fatalf("unexpected status %d", rec.Code) + } + } +} diff --git a/evmrpc/rpcstack.go b/evmrpc/rpcstack.go index 1fd9e6bead..b8dc30f596 100644 --- a/evmrpc/rpcstack.go +++ b/evmrpc/rpcstack.go @@ -68,6 +68,8 @@ type RPCEndpointConfig struct { maxRequestBodyBytes int64 // maxConcurrentRequestBytes bounds total request bytes admitted concurrently; 0 disables. maxConcurrentRequestBytes int64 + // bodyReadIdleTimeout caps idle time between body chunks; 0 disables. + bodyReadIdleTimeout time.Duration } type rpcHandler struct { @@ -344,15 +346,22 @@ func (h *HTTPServer) EnableRPC(apis []rpc.API, config HTTPConfig) error { srv.RegisterDenyList(method) } h.HTTPConfig = config - base := NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts, config.JwtSecret) + // JWT runs before the byte limiter so unauthenticated clients get 401 without + // touching the global body budget. The inner stack omits JWT when configured here. + base := NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts, nil) // maxRequestBodyBytes feeds all three body-cap layers (requestSizeLimiter, the gate, and // srv.SetHTTPBodyLimit above) so they agree; change the cap via the config value, not one layer. - handler := newRequestSizeLimiter( + limiter := newRequestSizeLimiter( wrapSeiLegacyHTTP(base, config.SeiLegacyAllowlist, config.maxRequestBodyBytes), config.maxRequestBodyBytes, config.maxConcurrentRequestBytes, + config.bodyReadIdleTimeout, ) + handler := limiter + if len(config.JwtSecret) != 0 { + handler = newJWTHandler(config.JwtSecret, limiter) + } h.httpHandler.Store(&rpcHandler{ Handler: handler, server: srv, diff --git a/evmrpc/sei_legacy_test.go b/evmrpc/sei_legacy_test.go index 0e1f113c3b..c9c8cca810 100644 --- a/evmrpc/sei_legacy_test.go +++ b/evmrpc/sei_legacy_test.go @@ -296,7 +296,7 @@ func TestComposedStack_OverLimitRejectedConsistently(t *testing.T) { _, _ = io.ReadAll(r.Body) _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":"0x1"}`)) }) - stack := newRequestSizeLimiter(wrapSeiLegacyHTTP(base, enabled, maxBody), maxBody, 0) + stack := newRequestSizeLimiter(wrapSeiLegacyHTTP(base, enabled, maxBody), maxBody, 0, 0) req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(mkBody(tc.bodyLen))) req.Header.Set("Content-Type", "application/json") diff --git a/evmrpc/server.go b/evmrpc/server.go index 58ad5d9055..d58b30a402 100644 --- a/evmrpc/server.go +++ b/evmrpc/server.go @@ -233,6 +233,7 @@ func NewEVMHTTPServer( httpConfig.batchResponseSizeLimit = config.BatchResponseMaxSize httpConfig.maxRequestBodyBytes = config.MaxRequestBodyBytes httpConfig.maxConcurrentRequestBytes = config.MaxConcurrentRequestBytes + httpConfig.bodyReadIdleTimeout = config.BodyReadIdleTimeout if err := httpServer.EnableRPC(apis, httpConfig); err != nil { return nil, err }