Merge pull request #7980 from kamilikamil/request-body-budget

Bound request body memory held by concurrent HTTP requests

README.md

@@ -256,6 +256,7 @@ CONFIGURATIONS:
    -rsr, -response-size-read int         max response size to read in bytes
    -rss, -response-size-save int         max response size to read in bytes (default 1048576)
    -rqss, -request-size-save int         max request body size to save in bytes (0 to save the whole body) (default 32768)
+   -mrbi, -max-request-body-inflight int max request body bytes held by in-flight http requests; requests with bodies of 1MB or more queue past it (0 disables)
    -reset                                reset removes all nuclei configuration and data files (including nuclei-templates)
    -tlsi, -tls-impersonate               enable experimental client hello (ja3) tls randomization
    -hae, -http-api-endpoint string       experimental http api endpoint

cmd/nuclei/main.go

@@ -407,6 +407,7 @@ on extensive configurability, massive extensibility and ease of use.`)
 		flagSet.IntVarP(&options.ResponseReadSize, "response-size-read", "rsr", 0, "max response size to read in bytes"),
 		flagSet.IntVarP(&options.ResponseSaveSize, "response-size-save", "rss", unitutils.Mega, "max response size to read in bytes"),
 		flagSet.IntVarP(&options.RequestSaveSize, "request-size-save", "rqss", 32*unitutils.Kilo, "max request body size to save in bytes (0 to save the whole body)"),
+		flagSet.IntVarP(&options.MaxRequestBodyInFlight, "max-request-body-inflight", "mrbi", 0, "max request body bytes held by in-flight http requests; requests with bodies of 1MB or more queue past it (0 disables)"),
 		flagSet.CallbackVar(func() { resetRequested = true }, "reset", "reset removes all nuclei configuration and data files (including nuclei-templates)"),
 		flagSet.BoolVarP(&options.TlsImpersonate, "tls-impersonate", "tlsi", false, "enable experimental client hello (ja3) tls randomization"),
 		flagSet.StringVarP(&options.HttpApiEndpoint, "http-api-endpoint", "hae", "", "experimental http api endpoint"),

internal/runner/runner.go

@@ -73,6 +73,7 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/templates"
 	"github.com/projectdiscovery/nuclei/v3/pkg/types"
 	"github.com/projectdiscovery/nuclei/v3/pkg/utils"
+	"github.com/projectdiscovery/nuclei/v3/pkg/utils/httpbody"
 	"github.com/projectdiscovery/nuclei/v3/pkg/utils/stats"
 	"github.com/projectdiscovery/nuclei/v3/pkg/utils/yaml"
 	"github.com/projectdiscovery/retryablehttp-go"
@@ -676,6 +677,7 @@ func (r *Runner) RunEnumeration() error {
 		Catalog:             r.catalog,
 		IssuesClient:        r.issuesClient,
 		RateLimiter:         r.rateLimiter,
+		RequestBodyBudget:   httpbody.NewBudget(int64(r.options.MaxRequestBodyInFlight)),
 		Interactsh:          r.interactsh,
 		ProjectFile:         r.projectFile,
 		Browser:             r.browser,

lib/multi.go

@@ -39,12 +39,14 @@ func createEphemeralObjects(ctx context.Context, base *NucleiEngine, opts *types
 		Catalog:      base.catalog,
 		IssuesClient: base.rc,
 		RateLimiter:  base.rateLimiter,
-		Interactsh:   base.interactshClient,
-		Colorizer:    aurora.New(aurora.WithColors(true)),
-		ResumeCfg:    types.NewResumeCfg(),
-		Parser:       base.parser,
-		Browser:      base.browserInstance,
-		DoNotCache:   opts.DoNotCacheTemplates,
+
+		RequestBodyBudget: base.executerOpts.RequestBodyBudget,
+		Interactsh:        base.interactshClient,
+		Colorizer:         aurora.New(aurora.WithColors(true)),
+		ResumeCfg:         types.NewResumeCfg(),
+		Parser:            base.parser,
+		Browser:           base.browserInstance,
+		DoNotCache:        opts.DoNotCacheTemplates,
 	}
 	if base.interactshClient != nil {
 		u.executerOpts.InteractshScope = base.interactshClient.NewRequestScope()

lib/sdk_private.go

@@ -36,6 +36,7 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/templates"
 	"github.com/projectdiscovery/nuclei/v3/pkg/types"
 	nucleiUtils "github.com/projectdiscovery/nuclei/v3/pkg/utils"
+	"github.com/projectdiscovery/nuclei/v3/pkg/utils/httpbody"
 	"github.com/projectdiscovery/ratelimit"
 	pdcpauth "github.com/projectdiscovery/utils/auth/pdcp"
 )
@@ -279,6 +280,7 @@ func (e *NucleiEngine) init(ctx context.Context) error {
 		Catalog:            e.catalog,
 		IssuesClient:       e.rc,
 		RateLimiter:        e.rateLimiter,
+		RequestBodyBudget:  httpbody.NewBudget(int64(e.opts.MaxRequestBodyInFlight)),
 		Interactsh:         e.interactshClient,
 		Colorizer:          aurora.New(aurora.WithColors(true)),
 		ResumeCfg:          types.NewResumeCfg(),

pkg/protocols/http/body_budget.go

@@ -0,0 +1,103 @@
+package http
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
+	"github.com/projectdiscovery/nuclei/v3/pkg/utils/httpbody"
+)
+
+// bodySizeEstimate is the largest request body a template has built so far.
+type bodySizeEstimate struct {
+	known    atomic.Bool
+	max      atomic.Int64
+	learning sync.Mutex
+}
+
+type bodyReservation struct {
+	budget  *httpbody.Budget
+	size    int64
+	release func()
+	unlock  func()
+}
+
+func noop() {}
+
+// reserveBody reserves the template's largest body size before the body is
+// built. Until a size is known, builds run one at a time.
+func (request *Request) reserveBody(ctx context.Context) (*bodyReservation, error) {
+	r := &bodyReservation{budget: request.options.RequestBodyBudget, release: noop, unlock: noop}
+	estimate := request.bodySize
+	if r.budget == nil || estimate == nil {
+		return r, nil
+	}
+	if !estimate.known.Load() {
+		estimate.learning.Lock()
+		if !estimate.known.Load() {
+			r.unlock = estimate.learning.Unlock
+			return r, nil
+		}
+		estimate.learning.Unlock()
+	}
+	r.size = estimate.max.Load()
+	release, err := r.budget.Acquire(ctx, r.size)
+	if err != nil {
+		return nil, err
+	}
+	r.release = release
+	return r, nil
+}
+
+// makeReserved builds the next request inside a body reservation, waiting
+// for budget until waitCtx ends. On success the caller owns the reservation
+// and calls done once the send ends.
+func (request *Request) makeReserved(waitCtx context.Context, generator *requestGenerator, input *contextargs.Context, data string, payloads, dynamicValues map[string]interface{}) (*generatedRequest, *bodyReservation, error) {
+	reservation, err := request.reserveBody(waitCtx)
+	if err != nil {
+		return nil, nil, err
+	}
+	generated, err := generator.Make(input.Context(), input, data, payloads, dynamicValues)
+	if err == nil {
+		err = reservation.built(waitCtx, request, generated.bodyLen())
+	}
+	if err != nil {
+		reservation.done()
+		return nil, nil, err
+	}
+	generated.bodyReserved = true
+	return generated, reservation, nil
+}
+
+// built grows the reservation to the body's real size. It releases before
+// acquiring again: waiting while holding part of the budget can deadlock.
+func (r *bodyReservation) built(ctx context.Context, request *Request, size int64) error {
+	if estimate := request.bodySize; estimate != nil {
+		for {
+			seen := estimate.max.Load()
+			if size <= seen || estimate.max.CompareAndSwap(seen, size) {
+				break
+			}
+		}
+		estimate.known.Store(true)
+	}
+	r.unlock()
+	r.unlock = noop
+	if size <= r.size {
+		return nil
+	}
+	r.release()
+	r.release = noop
+	release, err := r.budget.Acquire(ctx, size)
+	if err != nil {
+		return err
+	}
+	r.release, r.size = release, size
+	return nil
+}
+
+func (r *bodyReservation) done() {
+	r.unlock()
+	r.release()
+}

pkg/protocols/http/body_budget_memory_test.go

@@ -0,0 +1,125 @@
+//go:build !race
+
+package http
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"runtime"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/projectdiscovery/nuclei/v3/internal/tests/testutils"
+	"github.com/projectdiscovery/nuclei/v3/pkg/model"
+	"github.com/projectdiscovery/nuclei/v3/pkg/model/types/severity"
+	"github.com/projectdiscovery/nuclei/v3/pkg/output"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
+	"github.com/projectdiscovery/nuclei/v3/pkg/utils/httpbody"
+)
+
+// Threads and pipeline build bodies ahead of the send. Built bodies must stay
+// within the budget while the server holds the first request.
+func TestRequestBodyBudgetBoundsBodiesBuiltAhead(t *testing.T) {
+	const bodySize = 8 << 20
+	const payloads = 12
+
+	for _, mode := range []string{"threads", "pipeline"} {
+		t.Run(mode, func(t *testing.T) {
+			if mode == "pipeline" {
+				t.Skip("needs projectdiscovery/rawhttp#551: the pipeline client hangs after 10s idle")
+			}
+			arrived := make(chan struct{}, payloads)
+			unblock := make(chan struct{})
+			var received atomic.Int32
+			// net/http drops pipelined connections with large bodies, so serve
+			// Content-Length framed requests in order by hand.
+			listener, err := net.Listen("tcp", "127.0.0.1:0")
+			require.NoError(t, err)
+			defer func() { _ = listener.Close() }()
+			go func() {
+				for {
+					conn, err := listener.Accept()
+					if err != nil {
+						return
+					}
+					go func() {
+						defer func() { _ = conn.Close() }()
+						reader := bufio.NewReader(conn)
+						for {
+							req, err := http.ReadRequest(reader)
+							if err != nil {
+								return
+							}
+							_, _ = io.Copy(io.Discard, req.Body)
+							received.Add(1)
+							arrived <- struct{}{}
+							<-unblock
+							if _, err := io.WriteString(conn, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"); err != nil {
+								return
+							}
+						}
+					}()
+				}
+			}()
+			target := "http://" + listener.Addr().String()
+
+			options := testutils.DefaultOptions.Copy()
+			testutils.Init(options)
+
+			values := make([]string, payloads)
+			for i := range values {
+				values[i] = fmt.Sprint(i)
+			}
+			request := &Request{
+				ID:       "body-budget-" + mode,
+				Payloads: map[string]interface{}{"v": values},
+			}
+			body := `{{repeat("A", 8388608)}}`
+			if mode == "threads" {
+				request.Threads = 8
+				request.Path = []string{"{{BaseURL}}/?v={{v}}"}
+				request.Method = HTTPMethodTypeHolder{MethodType: HTTPPost}
+				request.Body = body
+			} else {
+				request.Pipeline = true
+				request.PipelineConcurrentConnections = 1
+				request.PipelineRequestsPerConnection = payloads
+				request.Raw = []string{"POST /?v={{v}} HTTP/1.1\nHost: {{Hostname}}\nContent-Length: 8388608\n\n" + body}
+			}
+			executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+				ID:   request.ID,
+				Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+			})
+			executerOpts.RequestBodyBudget = httpbody.NewBudget(bodySize)
+			require.NoError(t, request.Compile(executerOpts))
+
+			var before runtime.MemStats
+			runtime.GC()
+			runtime.ReadMemStats(&before)
+
+			done := make(chan error, 1)
+			go func() {
+				done <- request.ExecuteWithResults(contextargs.NewWithInput(context.Background(), target), make(output.InternalEvent), make(output.InternalEvent), func(*output.InternalWrappedEvent) {})
+			}()
+			<-arrived
+			time.Sleep(300 * time.Millisecond)
+			var held runtime.MemStats
+			runtime.GC()
+			runtime.ReadMemStats(&held)
+			close(unblock)
+			require.NoError(t, <-done)
+
+			grown := int64(held.HeapAlloc) - int64(before.HeapAlloc)
+			t.Logf("heap held: %d MB", grown>>20)
+			require.Less(t, grown, int64(4*bodySize), "bodies built ahead of the send must stay within the budget")
+			require.EqualValues(t, payloads, received.Load())
+		})
+	}
+}

pkg/protocols/http/body_budget_test.go

@@ -0,0 +1,200 @@
+package http
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/projectdiscovery/nuclei/v3/internal/tests/testutils"
+	"github.com/projectdiscovery/nuclei/v3/pkg/model"
+	"github.com/projectdiscovery/nuclei/v3/pkg/model/types/severity"
+	"github.com/projectdiscovery/nuclei/v3/pkg/operators"
+	"github.com/projectdiscovery/nuclei/v3/pkg/operators/matchers"
+	"github.com/projectdiscovery/nuclei/v3/pkg/output"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
+	"github.com/projectdiscovery/nuclei/v3/pkg/types"
+	"github.com/projectdiscovery/nuclei/v3/pkg/utils/httpbody"
+)
+
+// The third request queues ~2.4s, so it only beats the 3s timeout if the
+// timeout starts after admission.
+func TestRequestBodyBudgetQueuesHeavyRequests(t *testing.T) {
+	const bodySize = 1 << 20
+	const requests = 3
+
+	var inFlight, maxInFlight atomic.Int32
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		n := inFlight.Add(1)
+		defer inFlight.Add(-1)
+		for {
+			seen := maxInFlight.Load()
+			if n <= seen || maxInFlight.CompareAndSwap(seen, n) {
+				break
+			}
+		}
+		body, _ := io.ReadAll(r.Body)
+		time.Sleep(1200 * time.Millisecond)
+		if len(body) == bodySize {
+			_, _ = io.WriteString(w, "full body")
+		}
+	}))
+	defer ts.Close()
+
+	options := testutils.DefaultOptions.Copy()
+	options.SetTimeouts(&types.Timeouts{DialTimeout: time.Second, HttpTimeout: 3 * time.Second})
+	testutils.Init(options)
+
+	request := &Request{
+		ID:     "body-budget",
+		Path:   []string{"{{BaseURL}}"},
+		Method: HTTPMethodTypeHolder{MethodType: HTTPPost},
+		Body:   `{{repeat("A", 1048576)}}`,
+		Operators: operators.Operators{
+			Matchers: []*matchers.Matcher{{
+				Part:  "body",
+				Type:  matchers.MatcherTypeHolder{MatcherType: matchers.WordsMatcher},
+				Words: []string{"full body"},
+			}},
+		},
+	}
+	executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOpts.RequestBodyBudget = httpbody.NewBudget(bodySize)
+	require.NoError(t, request.Compile(executerOpts))
+
+	var matched atomic.Int32
+	var wg sync.WaitGroup
+	for range requests {
+		wg.Go(func() {
+			err := request.ExecuteWithResults(contextargs.NewWithInput(context.Background(), ts.URL), make(output.InternalEvent), make(output.InternalEvent), func(event *output.InternalWrappedEvent) {
+				if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+					matched.Add(1)
+				}
+			})
+			require.NoError(t, err)
+		})
+	}
+	wg.Wait()
+
+	require.EqualValues(t, 1, maxInFlight.Load(), "budget must hold heavy requests to one in flight")
+	require.EqualValues(t, requests, matched.Load(), "queued requests must not time out while waiting")
+}
+
+func TestReserveBodyQueuesBuildsOnLearnedSize(t *testing.T) {
+	const bodySize = 4 << 20
+	request := &Request{
+		options:  &protocols.ExecutorOptions{RequestBodyBudget: httpbody.NewBudget(6 << 20)},
+		bodySize: &bodySizeEstimate{},
+	}
+
+	first, err := request.reserveBody(t.Context())
+	require.NoError(t, err)
+	blocked := make(chan struct{})
+	go func() {
+		second, err := request.reserveBody(t.Context())
+		require.NoError(t, err)
+		second.done()
+		close(blocked)
+	}()
+	select {
+	case <-blocked:
+		t.Fatal("second build started before the first recorded its size")
+	case <-time.After(20 * time.Millisecond):
+	}
+	require.NoError(t, first.built(t.Context(), request, bodySize))
+	first.done()
+	<-blocked
+
+	held, err := request.reserveBody(t.Context())
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	_, err = request.reserveBody(ctx)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	held.done()
+	next, err := request.reserveBody(t.Context())
+	require.NoError(t, err)
+	next.done()
+}
+
+func TestReserveBodyGrowsToLargerBody(t *testing.T) {
+	request := &Request{
+		options:  &protocols.ExecutorOptions{RequestBodyBudget: httpbody.NewBudget(6 << 20)},
+		bodySize: &bodySizeEstimate{},
+	}
+	learn, err := request.reserveBody(t.Context())
+	require.NoError(t, err)
+	require.NoError(t, learn.built(t.Context(), request, 1<<20))
+	learn.done()
+
+	grown, err := request.reserveBody(t.Context())
+	require.NoError(t, err)
+	require.NoError(t, grown.built(t.Context(), request, 5<<20))
+
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	_, err = request.reserveBody(ctx)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	grown.done()
+}
+
+// Stop-at-first-match ends the workers with tasks still queued. Their
+// reservations must go back to the shared budget.
+func TestRequestBodyBudgetStopAtFirstMatchReleasesQueued(t *testing.T) {
+	const budget = 16 << 20
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		time.Sleep(100 * time.Millisecond)
+		_, _ = io.WriteString(w, "match")
+	}))
+	defer ts.Close()
+
+	options := testutils.DefaultOptions.Copy()
+	testutils.Init(options)
+
+	values := make([]string, 64)
+	for i := range values {
+		values[i] = fmt.Sprint(i)
+	}
+	request := &Request{
+		ID:               "body-budget-stop-first",
+		Method:           HTTPMethodTypeHolder{MethodType: HTTPPost},
+		Path:             []string{"{{BaseURL}}/?v={{v}}"},
+		Body:             `{{repeat("A", 1048576)}}`,
+		Threads:          4,
+		Payloads:         map[string]interface{}{"v": values},
+		StopAtFirstMatch: true,
+		Operators: operators.Operators{Matchers: []*matchers.Matcher{{
+			Part:  "body",
+			Type:  matchers.MatcherTypeHolder{MatcherType: matchers.WordsMatcher},
+			Words: []string{"match"},
+		}}},
+	}
+	executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOpts.RequestBodyBudget = httpbody.NewBudget(budget)
+	require.NoError(t, request.Compile(executerOpts))
+
+	err := request.ExecuteWithResults(contextargs.NewWithInput(context.Background(), ts.URL), make(output.InternalEvent), make(output.InternalEvent), func(*output.InternalWrappedEvent) {})
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	release, err := executerOpts.RequestBodyBudget.Acquire(ctx, budget)
+	require.NoError(t, err, "every reservation must be released")
+	release()
+}

pkg/protocols/http/build_request.go

@@ -82,6 +82,9 @@ type generatedRequest struct {
 	dynamicValues        map[string]interface{}
 	interactshURLs       []string
 	customCancelFunction context.CancelFunc
+	// sendTimeout starts after the body budget admits the request.
+	sendTimeout  time.Duration
+	bodyReserved bool
 	// requestURLPattern tracks unmodified request url pattern without values ( it is used for constant vuln_hash)
 	// ex: {{BaseURL}}/api/exp?param={{randstr}}
 	requestURLPattern string

pkg/protocols/http/http.go

@@ -147,6 +147,7 @@ type Request struct {
 	rawhttpClient     *rawhttp.Client
 	replayClient      *http.Client // set when -replay-proxy is given
 	dialer            *fastdialer.Dialer
+	bodySize          *bodySizeEstimate
 
 	// description: |
 	//   SelfContained specifies if the request is self-contained.
@@ -416,6 +417,7 @@ func (request *Request) Compile(options *protocols.ExecutorOptions) error {
 		return errors.Wrap(err, "could not get dialer")
 	}
 	request.dialer = dialer
+	request.bodySize = &bodySizeEstimate{}
 
 	request.options = options
 	for _, option := range request.options.Options.CustomHeaders {

pkg/protocols/http/operators.go

@@ -15,6 +15,7 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/helpers/responsehighlighter"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/utils"
 	"github.com/projectdiscovery/nuclei/v3/pkg/types"
+	convUtil "github.com/projectdiscovery/utils/conversion"
 )
 
 // Match matches a generic data response again a given matcher
@@ -225,7 +226,7 @@ func (request *Request) savedRequest(dump string) string {
 	if request.options == nil || request.options.Options == nil {
 		return dump
 	}
-	return string(capDumpBody([]byte(dump), request.options.Options.RequestSaveSize))
+	return string(capDumpBody(convUtil.Bytes(dump), request.options.Options.RequestSaveSize))
 }
 
 // eventResponse returns the full response of event. Events matched after the

pkg/protocols/http/request.go

@@ -331,6 +331,7 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 	// bounded worker-pool to avoid spawning one goroutine per payload
 	type task struct {
 		req                *generatedRequest
+		body               *bodyReservation
 		updatedInput       *contextargs.Context
 		hasInteractMarkers bool
 	}
@@ -338,72 +339,87 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 	var workersWg sync.WaitGroup
 	currentWorkers := maxWorkers
 	tasks := make(chan task, maxWorkers)
+	// Workers exit on cancel without draining, and the budget is shared, so
+	// release what they left queued.
+	stopWorkers := func() {
+		close(tasks)
+		workersWg.Wait()
+		for t := range tasks {
+			t.body.done()
+		}
+	}
 	spawnWorker := func(ctx context.Context) {
 		workersWg.Add(1)
 		go func() {
 			defer workersWg.Done()
 			for t := range tasks {
-				select {
-				case <-ctx.Done():
-					return
-				default:
-				}
-				if spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(t.updatedInput) || spmHandler.Cancelled() {
-					continue
-				}
-				if err := spmHandler.Acquire(); err != nil {
-					continue
-				}
-				if spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(t.updatedInput) || spmHandler.Cancelled() {
-					spmHandler.Release()
-					continue
-				}
-				// Full URL, so https and an explicit port stay on the same key the observation uses.
-				// The handler context ends the wait when stop-at-first-match fires.
-				waitCtx, cancelWait := stopWithHandler(ctx, spmHandler)
-				err := request.rateLimitTake(waitCtx, hostBackoffKey(t.req, t.updatedInput))
-				cancelWait()
-				if err != nil {
+				if func() (exit bool) {
+					defer t.body.done()
+					select {
+					case <-ctx.Done():
+						return true
+					default:
+					}
+					if spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(t.updatedInput) || spmHandler.Cancelled() {
+						return false
+					}
+					if err := spmHandler.Acquire(); err != nil {
+						return false
+					}
+					if spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(t.updatedInput) || spmHandler.Cancelled() {
+						spmHandler.Release()
+						return false
+					}
+					// Full URL, so https and an explicit port stay on the same key the observation uses.
+					// The handler context ends the wait when stop-at-first-match fires.
+					waitCtx, cancelWait := stopWithHandler(ctx, spmHandler)
+					err := request.rateLimitTake(waitCtx, hostBackoffKey(t.req, t.updatedInput))
+					cancelWait()
+					if err != nil {
+						select {
+						case <-spmHandler.Done():
+							spmHandler.Release()
+							return false
+						case <-ctx.Done():
+							spmHandler.Release()
+							return false
+						case spmHandler.ResultChan <- err:
+							spmHandler.Release()
+							return false
+						}
+					}
+					if ctx.Err() != nil || spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(t.updatedInput) || spmHandler.Cancelled() {
+						spmHandler.Release()
+						return false
+					}
+					hasInteractMatchers := interactsh.HasMatchers(request.CompiledOperators)
+					needsRequestEvent := hasInteractMatchers && request.NeedsRequestCondition()
+					err = request.executeRequest(t.updatedInput, t.req, make(map[string]interface{}), hasInteractMatchers, func(event *output.InternalWrappedEvent) {
+						if (t.hasInteractMarkers || needsRequestEvent) && request.options.Interactsh != nil {
+							requestData := &interactsh.RequestData{
+								MakeResultFunc: request.MakeResultEvent,
+								Event:          event,
+								Operators:      request.CompiledOperators,
+								MatchFunc:      request.Match,
+								ExtractFunc:    request.Extract,
+							}
+							allOASTUrls := httputils.GetInteractshURLSFromEvent(event.InternalEvent)
+							allOASTUrls = append(allOASTUrls, t.req.interactshURLs...)
+							request.options.RegisterInteractshRequest(sliceutil.Dedupe(allOASTUrls), requestData)
+						}
+						wrappedCallback(event)
+					}, 0)
 					select {
 					case <-spmHandler.Done():
 						spmHandler.Release()
-						continue
-					case <-ctx.Done():
-						spmHandler.Release()
-						continue
+						return false
 					case spmHandler.ResultChan <- err:
+						request.recordHostResultAndCancelIfUnresponsive(t.updatedInput, err, spmHandler.Cancel)
 						spmHandler.Release()
-						continue
 					}
-				}
-				if ctx.Err() != nil || spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(t.updatedInput) || spmHandler.Cancelled() {
-					spmHandler.Release()
-					continue
-				}
-				hasInteractMatchers := interactsh.HasMatchers(request.CompiledOperators)
-				needsRequestEvent := hasInteractMatchers && request.NeedsRequestCondition()
-				err = request.executeRequest(t.updatedInput, t.req, make(map[string]interface{}), hasInteractMatchers, func(event *output.InternalWrappedEvent) {
-					if (t.hasInteractMarkers || needsRequestEvent) && request.options.Interactsh != nil {
-						requestData := &interactsh.RequestData{
-							MakeResultFunc: request.MakeResultEvent,
-							Event:          event,
-							Operators:      request.CompiledOperators,
-							MatchFunc:      request.Match,
-							ExtractFunc:    request.Extract,
-						}
-						allOASTUrls := httputils.GetInteractshURLSFromEvent(event.InternalEvent)
-						allOASTUrls = append(allOASTUrls, t.req.interactshURLs...)
-						request.options.RegisterInteractshRequest(sliceutil.Dedupe(allOASTUrls), requestData)
-					}
-					wrappedCallback(event)
-				}, 0)
-				select {
-				case <-spmHandler.Done():
-					spmHandler.Release()
-					continue
-				case spmHandler.ResultChan <- err:
-					request.recordHostResultAndCancelIfUnresponsive(t.updatedInput, err, spmHandler.Cancel)
-					spmHandler.Release()
+					return false
+				}() {
+					return
 				}
 			}
 		}()
@@ -422,8 +438,7 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 
 		select {
 		case <-input.Context().Done():
-			close(tasks)
-			workersWg.Wait()
+			stopWorkers()
 			return input.Context().Err()
 		default:
 		}
@@ -431,8 +446,7 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 		// resize check point - nop if there are no changes
 		if shouldFollowGlobal && spmHandler.Size() != request.options.Options.PayloadConcurrency {
 			if err := spmHandler.Resize(input.Context(), request.options.Options.PayloadConcurrency); err != nil {
-				close(tasks)
-				workersWg.Wait()
+				stopWorkers()
 				return err
 			}
 			// if payload concurrency increased, add more workers
@@ -449,15 +463,22 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 			break
 		}
 
-		ctx := input.Context()
-		generatedHttpRequest, err := generator.Make(ctx, input, inputData, payloads, dynamicValues)
+		// Queued tasks hold budget and are only released once the workers
+		// stop, so the wait has to end with them.
+		generatedHttpRequest, reservation, err := request.makeReserved(workerCtx, generator, input, inputData, payloads, dynamicValues)
 		if err != nil {
 			if err == types.ErrNoMoreRequests {
 				break
 			}
+			if workerCtx.Err() != nil {
+				stopWorkers()
+				if spmHandler.FoundFirstMatch() {
+					return nil
+				}
+				return context.Cause(workerCtx)
+			}
 			request.options.Progress.IncrementFailedRequestsBy(int64(generator.Total()))
-			close(tasks)
-			workersWg.Wait()
+			stopWorkers()
 			return err
 		}
 		hasInteractMarkers := interactsh.HasMarkers(inputData) || len(generatedHttpRequest.interactshURLs) > 0
@@ -467,33 +488,32 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 		updatedInput := contextargs.GetCopyIfHostOutdated(input, generatedHttpRequest.URL())
 		if request.isUnresponsiveAddress(updatedInput) {
 			// skip on unresponsive host no need to continue
+			reservation.done()
 			spmHandler.Cancel()
-			close(tasks)
-			workersWg.Wait()
+			stopWorkers()
 			return nil
 		}
 		select {
 		case <-spmHandler.Done():
-			close(tasks)
-			workersWg.Wait()
+			reservation.done()
+			stopWorkers()
 			spmHandler.Wait()
 			if spmHandler.FoundFirstMatch() {
 				return nil
 			}
 			return multierr.Combine(spmHandler.CombinedResults()...)
 		case <-workerCtx.Done():
-			close(tasks)
-			workersWg.Wait()
+			reservation.done()
+			stopWorkers()
 			if spmHandler.FoundFirstMatch() {
 				return nil
 			}
 			return context.Cause(workerCtx)
-		case tasks <- task{req: generatedHttpRequest, updatedInput: updatedInput, hasInteractMarkers: hasInteractMarkers}:
+		case tasks <- task{req: generatedHttpRequest, body: reservation, updatedInput: updatedInput, hasInteractMarkers: hasInteractMarkers}:
 		}
 		request.options.Progress.IncrementRequests()
 	}
-	close(tasks)
-	workersWg.Wait()
+	stopWorkers()
 	spmHandler.Wait()
 	if spmHandler.FoundFirstMatch() {
 		// ignore any context cancellation and in-transit execution errors
@@ -582,7 +602,7 @@ func (request *Request) executeTurboHTTP(input *contextargs.Context, dynamicValu
 		}
 
 		ctx := input.Context()
-		generatedHttpRequest, err := generator.Make(ctx, input, inputData, payloads, dynamicValues)
+		generatedHttpRequest, reservation, err := request.makeReserved(ctx, generator, input, inputData, payloads, dynamicValues)
 		if err != nil {
 			request.options.Progress.IncrementFailedRequestsBy(int64(generator.Total()))
 			return err
@@ -593,15 +613,18 @@ func (request *Request) executeTurboHTTP(input *contextargs.Context, dynamicValu
 		updatedInput := contextargs.GetCopyIfHostOutdated(input, generatedHttpRequest.URL())
 		if request.isUnresponsiveAddress(updatedInput) {
 			// skip on unresponsive host no need to continue
+			reservation.done()
 			spmHandler.Cancel()
 			return nil
 		}
 		generatedHttpRequest.pipelinedClient = pipeClient
 		if err := spmHandler.Acquire(); err != nil {
+			reservation.done()
 			break
 		}
-		go func(httpRequest *generatedRequest, requestInput *contextargs.Context) {
+		go func(httpRequest *generatedRequest, body *bodyReservation, requestInput *contextargs.Context) {
 			defer spmHandler.Release()
+			defer body.done()
 			if spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(requestInput) {
 				// skip if first match is found
 				return
@@ -626,7 +649,7 @@ func (request *Request) executeTurboHTTP(input *contextargs.Context, dynamicValu
 				request.recordHostResultAndCancelIfUnresponsive(requestInput, err, spmHandler.Cancel)
 				return
 			}
-		}(generatedHttpRequest, updatedInput)
+		}(generatedHttpRequest, reservation, updatedInput)
 		request.options.Progress.IncrementRequests()
 	}
 	spmHandler.Wait()
@@ -676,16 +699,17 @@ executionLoop:
 			hasInteractMatchers := interactsh.HasMatchers(request.CompiledOperators)
 
 			ctx := input.Context()
-			ctxWithTimeout, cancel := context.WithTimeoutCause(ctx, request.options.Options.GetTimeouts().HttpTimeout, ErrHttpEngineRequestDeadline)
-			defer cancel()
-
-			generatedHttpRequest, err := generator.Make(ctxWithTimeout, input, data, payloads, dynamicValue)
+			generatedHttpRequest, reservation, err := request.makeReserved(ctx, generator, input, data, payloads, dynamicValue)
 			if err != nil {
 				if err == types.ErrNoMoreRequests {
 					return false, true, nil, nil
 				}
 				return false, true, nil, err
 			}
+			defer reservation.done()
+			if generatedHttpRequest.customCancelFunction == nil {
+				generatedHttpRequest.sendTimeout = request.options.Options.GetTimeouts().HttpTimeout
+			}
 			// ideally if http template used a custom port or hostname
 			// we would want to update it in input but currently templateCtx logic
 			// is closely tied to contextargs.Context so we are temporarily creating
@@ -1060,6 +1084,20 @@ func (request *Request) executeRequest(input *contextargs.Context, generatedRequ
 		return fmt.Errorf("hostErrorsCache : host %s is unresponsive", input.MetaInput.Input)
 	}
 
+	// Race requests must all be in flight together to reach the open gate.
+	if !generatedRequest.original.Race && !generatedRequest.bodyReserved {
+		releaseBody, err := request.options.RequestBodyBudget.Acquire(input.Context(), generatedRequest.bodyLen())
+		if err != nil {
+			return err
+		}
+		defer releaseBody()
+	}
+	if generatedRequest.sendTimeout > 0 && generatedRequest.request != nil {
+		ctx, cancel := context.WithTimeoutCause(generatedRequest.request.Context(), generatedRequest.sendTimeout, ErrHttpEngineRequestDeadline)
+		defer cancel()
+		generatedRequest.request = generatedRequest.request.WithContext(ctx)
+	}
+
 	// wrap one more callback for validation and fixing event
 	callback := func(event *output.InternalWrappedEvent) {
 		// validateNFixEvent performs necessary validation on generated event

pkg/protocols/http/utils.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"slices"
 	"strings"
 
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/generators"
@@ -82,6 +83,9 @@ func dumpRequest(req *retryablehttp.Request, maxBody int) (requestDump, error) {
 	}
 
 	capture := &cappedWriter{headerEnd: -1, maxBody: maxBody}
+	if maxBody <= 0 {
+		capture.buf = make([]byte, 0, len(body)+dumpHeaderRoom)
+	}
 	if err := clone.Write(capture); err != nil {
 		return requestDump{}, err
 	}
@@ -89,7 +93,7 @@ func dumpRequest(req *retryablehttp.Request, maxBody int) (requestDump, error) {
 	// http.Transport asks for gzip unless the caller set an encoding.
 	if clone.Header.Get("Accept-Encoding") == "" && clone.Header.Get("Range") == "" && clone.Method != http.MethodHead && capture.headerEnd >= 2 {
 		end := capture.headerEnd - 2
-		data = append(data[:end:end], append([]byte("Accept-Encoding: gzip\r\n"), data[end:]...)...)
+		data = slices.Insert(data, end, acceptEncodingGzip...)
 	}
 	return requestDump{data: appendTruncationMarker(data, capture.dropped), truncated: capture.dropped}, nil
 }
@@ -123,6 +127,10 @@ func capDumpBody(dump []byte, maxBody int) []byte {
 
 var headerTerminator = []byte("\r\n\r\n")
 
+var acceptEncodingGzip = []byte("Accept-Encoding: gzip\r\n")
+
+const dumpHeaderRoom = 8 * 1024
+
 // cappedWriter keeps the request line and headers in full and at most
 // maxBody bytes of the body that follows, counting the rest.
 type cappedWriter struct {
@@ -177,6 +185,19 @@ func getHTTPProjectCacheScope(requestDump requestDump, body []byte, scheme, host
 	return scoped.Bytes()
 }
 
+func (g *generatedRequest) bodyLen() int64 {
+	if g.rawRequest != nil {
+		return int64(len(g.rawRequest.Data))
+	}
+	if g.request == nil || g.request.Body == nil {
+		return 0
+	}
+	if body, ok := g.request.Body.(*httpbody.Body); ok {
+		return int64(body.Len())
+	}
+	return g.request.ContentLength
+}
+
 // body returns the request body without copying it.
 func (g *generatedRequest) body() []byte {
 	if g.request != nil {

pkg/protocols/protocols.go

@@ -44,6 +44,7 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/scan"
 	templateTypes "github.com/projectdiscovery/nuclei/v3/pkg/templates/types"
 	"github.com/projectdiscovery/nuclei/v3/pkg/types"
+	"github.com/projectdiscovery/nuclei/v3/pkg/utils/httpbody"
 	unitutils "github.com/projectdiscovery/utils/unit"
 )
 
@@ -106,6 +107,8 @@ type ExecutorOptions struct {
 	Progress progress.Progress
 	// RateLimiter is a rate-limiter for limiting sent number of requests.
 	RateLimiter *ratelimit.Limiter
+	// RequestBodyBudget bounds request-body bytes in flight; nil is no limit.
+	RequestBodyBudget *httpbody.Budget
 	// Catalog is a template catalog implementation for nuclei
 	Catalog catalog.Catalog
 	// ProjectFile is the project file for nuclei
@@ -395,6 +398,7 @@ func (e *ExecutorOptions) Copy() *ExecutorOptions {
 		IssuesClient:                 e.IssuesClient,
 		Progress:                     e.Progress,
 		RateLimiter:                  e.RateLimiter,
+		RequestBodyBudget:            e.RequestBodyBudget,
 		Catalog:                      e.Catalog,
 		ProjectFile:                  e.ProjectFile,
 		Browser:                      e.Browser,
@@ -706,6 +710,7 @@ func (e *ExecutorOptions) ApplyNewEngineOptions(n *ExecutorOptions) {
 	e.IssuesClient = n.IssuesClient
 	e.Progress = n.Progress
 	e.RateLimiter = n.RateLimiter
+	e.RequestBodyBudget = n.RequestBodyBudget
 	e.Catalog = n.Catalog
 	e.ProjectFile = n.ProjectFile
 	e.Browser = n.Browser

pkg/types/types.go

@@ -370,6 +370,8 @@ type Options struct {
 	// RequestSaveSize is the maximum request body kept in saved output and
 	// debug dumps. Matchers see the full request. 0 keeps the whole body.
 	RequestSaveSize int
+	// MaxRequestBodyInFlight bounds request-body bytes in flight; 0 disables it.
+	MaxRequestBodyInFlight int
 	// Health Check
 	HealthCheck bool
 	// Time to wait between each input read operation before closing the stream
@@ -765,6 +767,7 @@ func (options *Options) Copy() *Options {
 		ResponseReadSize:               options.ResponseReadSize,
 		ResponseSaveSize:               options.ResponseSaveSize,
 		RequestSaveSize:                options.RequestSaveSize,
+		MaxRequestBodyInFlight:         options.MaxRequestBodyInFlight,
 		HealthCheck:                    options.HealthCheck,
 		InputReadTimeout:               options.InputReadTimeout,
 		DisableStdin:                   options.DisableStdin,

pkg/utils/httpbody/budget.go

@@ -0,0 +1,38 @@
+package httpbody
+
+import (
+	"context"
+
+	"golang.org/x/sync/semaphore"
+)
+
+// MinBudgetedBody is the smallest body that waits on a Budget, so small
+// requests never queue behind a heavy one.
+const MinBudgetedBody = 1 << 20
+
+// Budget bounds the request-body bytes held by requests in flight.
+type Budget struct {
+	size int64
+	sem  *semaphore.Weighted
+}
+
+// NewBudget returns nil when size <= 0; a nil budget never blocks.
+func NewBudget(size int64) *Budget {
+	if size <= 0 {
+		return nil
+	}
+	return &Budget{size: size, sem: semaphore.NewWeighted(size)}
+}
+
+// Acquire blocks until n bytes fit or ctx is done. n larger than the budget
+// is clamped, so that request runs alone.
+func (b *Budget) Acquire(ctx context.Context, n int64) (release func(), err error) {
+	if b == nil || n < MinBudgetedBody {
+		return func() {}, nil
+	}
+	n = min(n, b.size)
+	if err := b.sem.Acquire(ctx, n); err != nil {
+		return nil, err
+	}
+	return func() { b.sem.Release(n) }, nil
+}

pkg/utils/httpbody/budget_test.go

@@ -0,0 +1,68 @@
+package httpbody
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBudgetQueuesBodiesOverSize(t *testing.T) {
+	budget := NewBudget(48 << 20)
+
+	first, err := budget.Acquire(t.Context(), 20<<20)
+	require.NoError(t, err)
+	second, err := budget.Acquire(t.Context(), 20<<20)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	_, err = budget.Acquire(ctx, 20<<20)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	first()
+	third, err := budget.Acquire(t.Context(), 20<<20)
+	require.NoError(t, err)
+	third()
+	second()
+}
+
+func TestBudgetRunsBodyLargerThanBudgetAlone(t *testing.T) {
+	budget := NewBudget(8 << 20)
+
+	huge, err := budget.Acquire(t.Context(), 64<<20)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	_, err = budget.Acquire(ctx, 2<<20)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	huge()
+	next, err := budget.Acquire(t.Context(), 2<<20)
+	require.NoError(t, err)
+	next()
+}
+
+func TestBudgetSkipsSmallBodies(t *testing.T) {
+	budget := NewBudget(MinBudgetedBody)
+	full, err := budget.Acquire(t.Context(), MinBudgetedBody)
+	require.NoError(t, err)
+	defer full()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	release, err := budget.Acquire(ctx, MinBudgetedBody-1)
+	require.NoError(t, err)
+	release()
+}
+
+func TestNilBudgetNeverBlocks(t *testing.T) {
+	require.Nil(t, NewBudget(0))
+
+	var budget *Budget
+	release, err := budget.Acquire(t.Context(), 1<<30)
+	require.NoError(t, err)
+	release()
+}