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 }