test(http): cover replay body bytes across read sizes

go.mod

@@ -53,7 +53,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/graphql-go/graphql v0.8.1
 	github.com/h2non/filetype v1.1.3
-	github.com/hashicorp/golang-lru/v2 v2.0.7
+	github.com/hashicorp/golang-lru/v2 v2.0.8-0.20261012062354-235099c106fc
 	github.com/hdm/jarm-go v0.0.8
 	github.com/invopop/jsonschema v0.13.0
 	github.com/itchyny/gojq v0.12.17

go.sum

@@ -673,8 +673,8 @@ github.com/hashicorp/go-version v1.9.0 h1:CeOIz6k+LoN3qX9Z0tyQrPtiB1DFYRPfCIBtaX
 github.com/hashicorp/go-version v1.9.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
-github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
-github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
+github.com/hashicorp/golang-lru/v2 v2.0.8-0.20261012062354-235099c106fc h1:yan3sbChZtrBNRSqJVcBgifbDHpv6lPYFQZiUSklqhg=
+github.com/hashicorp/golang-lru/v2 v2.0.8-0.20261012062354-235099c106fc/go.mod h1:qnbgnNzfydwuHjSCApF4bdul+tZ8T3y1MkZG/OFczLA=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hashicorp/vault/api v1.15.0 h1:O24FYQCWwhwKnF7CuSqP30S51rTV7vz1iACXE/pj5DA=

pkg/protocols/common/hostratelimit/pool.go

@@ -378,6 +378,7 @@ func (p *PerHostRateLimitPool) Stats() RateLimitPoolStats {
 
 func (p *PerHostRateLimitPool) Close() {
 	p.EvictAll()
+	p.cache.Close()
 }
 
 type RateLimitPoolStats struct {

pkg/protocols/common/hostratelimit/pool_close_test.go

@@ -9,8 +9,18 @@ import (
 
 	"github.com/projectdiscovery/nuclei/v3/pkg/types"
 	"github.com/stretchr/testify/require"
+	"github.com/tarunKoyalwar/goleak"
 )
 
+func TestPerHostRateLimitPoolCloseStopsExpirationWorker(t *testing.T) {
+	ignoreCurrent := goleak.IgnoreCurrent()
+	for range 20 {
+		pool := NewPerHostRateLimitPool(1, time.Hour, time.Hour, &types.Options{})
+		pool.Close()
+	}
+	goleak.VerifyNone(t, ignoreCurrent)
+}
+
 // TestPerHostRateLimitPoolDoesNotCreateGoroutinePerHost ensures a large target
 // set does not create one fixed-window refill goroutine for every host.
 func TestPerHostRateLimitPoolDoesNotCreateGoroutinePerHost(t *testing.T) {

pkg/protocols/http/body_admission.go

@@ -0,0 +1,98 @@
+package http
+
+import (
+	"context"
+	"sync"
+
+	"github.com/projectdiscovery/gologger"
+	"github.com/projectdiscovery/utils/memguardian"
+	"golang.org/x/sync/semaphore"
+)
+
+const (
+	largeBodyThreshold           = 1 << 20
+	largeBodyMemoryAmplification = 64
+)
+
+type bodyAdmission struct {
+	limit int64
+	slots *semaphore.Weighted
+}
+
+var (
+	sharedBodyAdmission     *bodyAdmission
+	sharedBodyAdmissionOnce sync.Once
+	sharedFlowAdmission     *bodyAdmission
+	sharedFlowAdmissionOnce sync.Once
+)
+
+func newBodyAdmission(totalMemory uint64) *bodyAdmission {
+	limit := int64(totalMemory / largeBodyMemoryAmplification)
+	if limit < largeBodyThreshold {
+		limit = largeBodyThreshold
+	}
+	return &bodyAdmission{limit: limit, slots: semaphore.NewWeighted(limit)}
+}
+
+func systemBodyAdmission() *bodyAdmission {
+	availableMemory := uint64(4 << 30)
+	info, err := memguardian.GetSysInfo()
+	if err != nil {
+		gologger.Warning().Msgf("could not read memory limit for large HTTP body admission: %s", err)
+	} else {
+		availableMemory = info.TotalRam()
+	}
+	if limit, limitErr := cgroupMemoryLimit("/proc/self/cgroup", "/sys/fs/cgroup"); limitErr != nil {
+		gologger.Warning().Msgf("could not read cgroup memory limit for large HTTP body admission: %s", limitErr)
+	} else if limit > 0 && limit < availableMemory {
+		availableMemory = limit
+	}
+	return newBodyAdmission(availableMemory)
+}
+
+func bodySize(req *generatedRequest) int64 {
+	if req.rawRequest != nil {
+		return int64(len(req.rawRequest.Data))
+	}
+	if req.request != nil {
+		return req.request.ContentLength
+	}
+	return 0
+}
+
+func (a *bodyAdmission) acquire(ctx context.Context, size int64) (func(), error) {
+	if size < largeBodyThreshold {
+		return func() {}, nil
+	}
+	weight := min(size, a.limit)
+	if err := a.slots.Acquire(ctx, weight); err != nil {
+		return nil, err
+	}
+	return func() { a.slots.Release(weight) }, nil
+}
+
+func admitLargeBody(ctx context.Context, req *generatedRequest) (func(), error) {
+	size := bodySize(req)
+	return admitBodySize(ctx, size)
+}
+
+func admitBodySize(ctx context.Context, size int64) (func(), error) {
+	if size < largeBodyThreshold {
+		return func() {}, nil
+	}
+	sharedBodyAdmissionOnce.Do(func() {
+		sharedBodyAdmission = systemBodyAdmission()
+	})
+	return sharedBodyAdmission.acquire(ctx, size)
+}
+
+// AdmitDeclaredBody bounds Flow executions before a declared large body is built.
+func AdmitDeclaredBody(ctx context.Context, size int64) (func(), error) {
+	if size < largeBodyThreshold {
+		return func() {}, nil
+	}
+	sharedFlowAdmissionOnce.Do(func() {
+		sharedFlowAdmission = systemBodyAdmission()
+	})
+	return sharedFlowAdmission.acquire(ctx, size)
+}

pkg/protocols/http/body_admission_memory.go

@@ -0,0 +1,73 @@
+package http
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+)
+
+func cgroupMemoryLimit(procPath, root string) (uint64, error) {
+	contents, err := os.ReadFile(procPath)
+	if errors.Is(err, os.ErrNotExist) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+
+	var lowest uint64
+	for line := range strings.SplitSeq(string(contents), "\n") {
+		fields := strings.SplitN(line, ":", 3)
+		if len(fields) != 3 {
+			continue
+		}
+		controllerRoot := root
+		limitFile := "memory.max"
+		if fields[1] != "" {
+			if !strings.Contains(","+fields[1]+",", ",memory,") {
+				continue
+			}
+			controllerRoot = filepath.Join(root, "memory")
+			limitFile = "memory.limit_in_bytes"
+		}
+		limit, err := memoryLimitInHierarchy(controllerRoot, fields[2], limitFile)
+		if err != nil {
+			return 0, err
+		}
+		if limit > 0 && (lowest == 0 || limit < lowest) {
+			lowest = limit
+		}
+	}
+	return lowest, nil
+}
+
+func memoryLimitInHierarchy(root, cgroupPath, limitFile string) (uint64, error) {
+	root = filepath.Clean(root)
+	current := filepath.Join(root, strings.TrimPrefix(filepath.Clean("/"+cgroupPath), "/"))
+	var lowest uint64
+	for {
+		contents, err := os.ReadFile(filepath.Join(current, limitFile))
+		if err == nil {
+			value := strings.TrimSpace(string(contents))
+			if value != "max" {
+				limit, parseErr := strconv.ParseUint(value, 10, 64)
+				if parseErr != nil || limit == 0 {
+					return 0, fmt.Errorf("invalid cgroup memory limit %q in %s", value, current)
+				}
+				if lowest == 0 || limit < lowest {
+					lowest = limit
+				}
+			}
+		} else if !errors.Is(err, os.ErrNotExist) {
+			return 0, err
+		}
+		if current == root {
+			break
+		}
+		current = filepath.Dir(current)
+	}
+	return lowest, nil
+}

pkg/protocols/http/body_admission_test.go

@@ -0,0 +1,155 @@
+package http
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"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/output"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
+	"github.com/projectdiscovery/nuclei/v3/pkg/types"
+	"github.com/projectdiscovery/utils/memguardian"
+)
+
+func TestLargeBodyAdmissionUsesMemoryBudget(t *testing.T) {
+	admission := newBodyAdmission(3 << 30)
+	require.EqualValues(t, 48<<20, admission.limit)
+
+	releases := make([]func(), 0, 2)
+	for range 2 {
+		release, err := admission.acquire(t.Context(), 20<<20)
+		require.NoError(t, err)
+		releases = append(releases, release)
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	_, err := admission.acquire(ctx, 20<<20)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	releases[0]()
+	release, err := admission.acquire(t.Context(), 20<<20)
+	require.NoError(t, err)
+	release()
+	releases[1]()
+}
+
+func TestSystemBodyAdmissionRespectsCgroupLimit(t *testing.T) {
+	raw, err := os.ReadFile("/sys/fs/cgroup/memory.max")
+	if err != nil {
+		t.Skipf("cgroup v2 memory limit unavailable: %v", err)
+	}
+	limit, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64)
+	if err != nil {
+		t.Skipf("cgroup v2 has no finite memory limit: %v", err)
+	}
+	info, err := memguardian.GetSysInfo()
+	require.NoError(t, err)
+	expected := newBodyAdmission(min(limit, info.TotalRam())).limit
+	actual := systemBodyAdmission().limit
+	require.Equalf(t, expected, actual, "body admission must use the lower of host RAM and %d-byte cgroup limit", limit)
+}
+
+func TestCgroupMemoryLimit(t *testing.T) {
+	for _, test := range []struct {
+		name, membership string
+		files            map[string]string
+		want             uint64
+		wantError        bool
+	}{
+		{name: "v2 namespaced root", membership: "0::/\n", files: map[string]string{"memory.max": "536870912\n"}, want: 512 << 20},
+		{name: "v2 parent bound", membership: "0::/tenant/pod\n", files: map[string]string{"memory.max": "max\n", "tenant/memory.max": "1073741824\n", "tenant/pod/memory.max": "2147483648\n"}, want: 1 << 30},
+		{name: "v2 unlimited", membership: "0::/\n", files: map[string]string{"memory.max": "max\n"}},
+		{name: "v1 memory controller", membership: "5:cpu,memory:/pod\n4:cpu:/pod\n", files: map[string]string{"memory/memory.limit_in_bytes": "max\n", "memory/pod/memory.limit_in_bytes": "268435456\n"}, want: 256 << 20},
+		{name: "invalid bound", membership: "0::/\n", files: map[string]string{"memory.max": "invalid\n"}, wantError: true},
+		{name: "no membership", membership: ""},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			root := t.TempDir()
+			membershipPath := filepath.Join(root, "self-cgroup")
+			require.NoError(t, os.WriteFile(membershipPath, []byte(test.membership), 0o600))
+			for name, contents := range test.files {
+				path := filepath.Join(root, name)
+				require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700))
+				require.NoError(t, os.WriteFile(path, []byte(contents), 0o600))
+			}
+			limit, err := cgroupMemoryLimit(membershipPath, root)
+			if test.wantError {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, test.want, limit)
+		})
+	}
+}
+
+func TestSmallBodyBypassesLargeBodyAdmission(t *testing.T) {
+	admission := newBodyAdmission(3 << 30)
+	fullRelease, err := admission.acquire(t.Context(), admission.limit)
+	require.NoError(t, err)
+	defer fullRelease()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond)
+	defer cancel()
+	release, err := admission.acquire(ctx, largeBodyThreshold-1)
+	require.NoError(t, err)
+	release()
+}
+
+func TestHTTPTimeoutStartsAfterLargeBodyAdmission(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.SetTimeouts(&types.Timeouts{HttpTimeout: time.Second})
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+
+	var sent atomic.Bool
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		sent.Store(true)
+		writer.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID:     "large-body-timeout",
+		Path:   []string{"{{BaseURL}}"},
+		Method: HTTPMethodTypeHolder{MethodType: HTTPPost},
+		Body:   "{{probe_body}}",
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: request.ID},
+	})
+	require.NoError(t, request.Compile(executerOptions))
+	input := contextargs.NewWithInput(t.Context(), server.URL)
+	body, err := NewReplayBody([]ReplayPart{{Pattern: "x", Count: 2 << 20}}, 0)
+	require.NoError(t, err)
+	executerOptions.GetTemplateCtx(input.MetaInput).Set("probe_body", body)
+
+	sharedBodyAdmissionOnce.Do(func() { sharedBodyAdmission = systemBodyAdmission() })
+	previousAdmission := sharedBodyAdmission
+	sharedBodyAdmission = newBodyAdmission(64 << 20)
+	t.Cleanup(func() { sharedBodyAdmission = previousAdmission })
+	hold, err := sharedBodyAdmission.acquire(t.Context(), body.Len())
+	require.NoError(t, err)
+	release := sync.OnceFunc(hold)
+	defer release()
+	time.AfterFunc(1500*time.Millisecond, release)
+
+	require.NoError(t, request.ExecuteWithResults(input, nil, nil, func(*output.InternalWrappedEvent) {}))
+	require.True(t, sent.Load())
+}

pkg/protocols/http/build_request.go

@@ -9,7 +9,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/pkg/errors"
 	"github.com/projectdiscovery/useragent"
 
 	"github.com/projectdiscovery/gologger"
@@ -83,6 +82,7 @@ type generatedRequest struct {
 	dynamicValues        map[string]interface{}
 	interactshURLs       []string
 	customCancelFunction context.CancelFunc
+	startRequestTimeout  bool
 	// requestURLPattern tracks unmodified request url pattern without values ( it is used for constant vuln_hash)
 	// ex: {{BaseURL}}/api/exp?param={{randstr}}
 	requestURLPattern string
@@ -233,14 +233,18 @@ func (r *requestGenerator) Make(ctx context.Context, input *contextargs.Context,
 	// Note: If possible any changes to current logic (i.e evaluate -> then parse URL)
 	// should be avoided since it is dependent on `urlutil` core logic
 
-	// Evaluate (replace) variable with final values
-	reqData, err = r.renderText(reqData, finalVars)
+	var replayBody *ReplayBody
+	if isRawRequest {
+		reqData, replayBody, err = r.renderRawRequest(reqData, finalVars)
+	} else {
+		reqData, err = r.renderText(reqData, finalVars)
+	}
 	if err != nil {
 		return nil, errkit.Wrap(err, "could not evaluate helper expressions")
 	}
 
 	if isRawRequest {
-		return r.generateRawRequest(ctx, reqData, parsed, finalVars, payloads)
+		return r.generateRawRequest(ctx, reqData, parsed, finalVars, payloads, replayBody)
 	}
 
 	reqURL, err := urlutil.ParseAbsoluteURL(reqData, true)
@@ -312,7 +316,13 @@ func (r *requestGenerator) makeSelfContainedRequest(ctx context.Context, input *
 	// default signer vars < variables < cli vars < dynamic values < payload < constants
 
 	// evaluate request
-	data, err := r.renderText(data, values)
+	var replayBody *ReplayBody
+	var err error
+	if isRawRequest {
+		data, replayBody, err = r.renderRawRequest(data, values)
+	} else {
+		data, err = r.renderText(data, values)
+	}
 	if err != nil {
 		return nil, errkit.Wrap(err, "could not evaluate helper expressions")
 	}
@@ -357,12 +367,16 @@ func (r *requestGenerator) makeSelfContainedRequest(ctx context.Context, input *
 		)
 
 		// Evaluate (replace) variable with final values
-		data, err = r.renderText(data, values)
+		if replayBody == nil {
+			data, replayBody, err = r.renderRawRequest(data, values)
+		} else {
+			data, err = r.renderText(data, values)
+		}
 		if err != nil {
 			return nil, errkit.Wrap(err, "could not evaluate helper expressions")
 		}
 
-		return r.generateRawRequest(ctx, data, parsed, values, payloads)
+		return r.generateRawRequest(ctx, data, parsed, values, payloads, replayBody)
 	}
 	if err := expressions.ContainsUnresolvedVariables(data); err != nil && !r.request.SkipVariablesCheck {
 		// early exit: if there are any unresolved variables in `path` after evaluation
@@ -407,7 +421,7 @@ func (r *requestGenerator) generateHttpRequest(ctx context.Context, urlx *urluti
 // generateRawRequest generates Raw Request from request data from template and variables
 // finalVars = contains all variables including generator and protocol specific variables
 // generatorValues = contains variables used in fuzzing or other generator specific values
-func (r *requestGenerator) generateRawRequest(ctx context.Context, rawRequest string, baseURL *urlutil.URL, finalVars, generatorValues map[string]interface{}) (*generatedRequest, error) {
+func (r *requestGenerator) generateRawRequest(ctx context.Context, rawRequest string, baseURL *urlutil.URL, finalVars, generatorValues map[string]interface{}, replayBody *ReplayBody) (*generatedRequest, error) {
 	var rawRequestData *raw.Request
 	var err error
 
@@ -420,6 +434,9 @@ func (r *requestGenerator) generateRawRequest(ctx context.Context, rawRequest st
 	if err != nil {
 		return nil, errkit.Wrap(err, "failed to parse raw request")
 	}
+	if replayBody != nil && r.request.Unsafe {
+		return nil, fmt.Errorf("replayable body is not supported for unsafe raw requests")
+	}
 
 	// Unsafe option uses rawhttp library
 	if r.request.Unsafe {
@@ -474,14 +491,27 @@ func (r *requestGenerator) generateRawRequest(ctx context.Context, rawRequest st
 	if err != nil {
 		return nil, errkit.Wrapf(err, "failed to create request with url %v got %v", rawRequestData.FullURL, err)
 	}
-	req, err := retryablehttp.NewRequestFromURLWithContext(ctx, rawRequestData.Method, urlx, rawRequestData.Data)
+	var requestBody interface{}
+	if replayBody == nil && rawRequestData.Data == "" {
+		requestBody = ""
+	}
+	req, err := retryablehttp.NewRequestFromURLWithContext(ctx, rawRequestData.Method, urlx, requestBody)
 	if err != nil {
 		return nil, err
 	}
+	if replayBody != nil {
+		setReplayBody(req, replayBody)
+	} else if rawRequestData.Data != "" {
+		setReplayableStringBody(req, rawRequestData.Data)
+	}
 
 	// force transfer encoding if conditions are met
-	if len(rawRequestData.Data) > 0 && req.Header.Get("Transfer-Encoding") != "chunked" && !stringsutil.EqualFoldAny(rawRequestData.Method, http.MethodGet, http.MethodHead) {
-		req.ContentLength = int64(len(rawRequestData.Data))
+	if (len(rawRequestData.Data) > 0 || replayBody != nil) && req.Header.Get("Transfer-Encoding") != "chunked" && !stringsutil.EqualFoldAny(rawRequestData.Method, http.MethodGet, http.MethodHead) {
+		if replayBody != nil {
+			req.ContentLength = replayBody.Len()
+		} else {
+			req.ContentLength = int64(len(rawRequestData.Data))
+		}
 	}
 
 	// override the body with a new one that will be used to read the request body in parallel threads
@@ -564,13 +594,15 @@ func (r *requestGenerator) fillRequest(req *retryablehttp.Request, values map[st
 
 	// Check if the user requested a request body
 	if r.request.Body != "" {
-		body, err := r.renderText(r.request.Body, values)
-		if err != nil {
-			return nil, errkit.Wrap(err, "could not evaluate helper expressions")
-		}
+		if replayBody, ok := replayBodyPlaceholder(r.request.Body, values); ok {
+			setReplayBody(req, replayBody)
+		} else {
+			body, err := r.renderText(r.request.Body, values)
+			if err != nil {
+				return nil, errkit.Wrap(err, "could not evaluate helper expressions")
+			}
 
-		if err := req.SetBodyString(body); err != nil {
-			return nil, errors.Wrap(err, "failed to set request body")
+			setReplayableStringBody(req, body)
 		}
 	}
 
@@ -621,3 +653,32 @@ func (r *requestGenerator) renderText(text string, values map[string]interface{}
 
 	return result.Text, nil
 }
+
+func (r *requestGenerator) renderRawRequest(text string, values map[string]interface{}) (string, *ReplayBody, error) {
+	separator := strings.Index(text, "\r\n\r\n")
+	separatorLength := 4
+	lineSeparator := strings.Index(text, "\n\n")
+	if lineSeparator >= 0 && (separator < 0 || lineSeparator < separator) {
+		separator = lineSeparator
+		separatorLength = 2
+	}
+	if separator >= 0 {
+		bodyText := text[separator+separatorLength:]
+		multipart := false
+		for _, line := range strings.Split(text[:separator], "\n") {
+			if strings.Contains(line, "Content-Type") && strings.Contains(line, "multipart/") {
+				multipart = true
+				break
+			}
+		}
+		if !multipart {
+			bodyText = strings.TrimSuffix(bodyText, "\r\n")
+		}
+		if body, ok := replayBodyPlaceholder(bodyText, values); ok {
+			rendered, err := r.renderText(text[:separator+separatorLength], values)
+			return rendered, body, err
+		}
+	}
+	rendered, err := r.renderText(text, values)
+	return rendered, nil, err
+}

pkg/protocols/http/replay_body.go

@@ -0,0 +1,172 @@
+package http
+
+import (
+	"fmt"
+	"io"
+	"math"
+	"strings"
+
+	"github.com/projectdiscovery/retryablehttp-go"
+)
+
+type ReplayPart struct {
+	Pattern string
+	Count   int64
+}
+
+type replaySection struct {
+	pattern string
+	length  int64
+}
+
+type ReplayBody struct {
+	sections []replaySection
+	length   int64
+}
+
+func NewReplayBody(parts []ReplayPart, trimTail int64) (*ReplayBody, error) {
+	if trimTail < 0 {
+		return nil, fmt.Errorf("negative replay body tail trim")
+	}
+	body := &ReplayBody{sections: make([]replaySection, 0, len(parts))}
+	for _, part := range parts {
+		if part.Count < 0 {
+			return nil, fmt.Errorf("negative replay body repeat count")
+		}
+		if len(part.Pattern) == 0 || part.Count == 0 {
+			continue
+		}
+		if part.Count > math.MaxInt64/int64(len(part.Pattern)) {
+			return nil, fmt.Errorf("replay body length overflow")
+		}
+		length := int64(len(part.Pattern)) * part.Count
+		if body.length > math.MaxInt64-length {
+			return nil, fmt.Errorf("replay body length overflow")
+		}
+		body.sections = append(body.sections, replaySection{pattern: part.Pattern, length: length})
+		body.length += length
+	}
+	if trimTail > body.length {
+		return nil, fmt.Errorf("replay body tail trim exceeds length")
+	}
+	body.length -= trimTail
+	if body.length > math.MaxInt64-2 {
+		return nil, fmt.Errorf("replay body length overflow")
+	}
+	remaining := body.length
+	for index := range body.sections {
+		section := &body.sections[index]
+		if remaining >= section.length {
+			remaining -= section.length
+			continue
+		}
+		section.length = remaining
+		body.sections = body.sections[:index+1]
+		break
+	}
+	return body, nil
+}
+
+func (body *ReplayBody) Len() int64 {
+	return body.length
+}
+
+func (body *ReplayBody) NewReader() io.ReadCloser {
+	readers := make([]io.Reader, 0, len(body.sections))
+	for _, section := range body.sections {
+		readers = append(readers, &repeatSectionReader{section: section})
+	}
+	return io.NopCloser(io.MultiReader(readers...))
+}
+
+func (body *ReplayBody) appendLiteral(value string) *ReplayBody {
+	if value == "" {
+		return body
+	}
+	copyBody := &ReplayBody{
+		sections: make([]replaySection, 0, len(body.sections)+1),
+		length:   body.length + int64(len(value)),
+	}
+	copyBody.sections = append(copyBody.sections, body.sections...)
+	copyBody.sections = append(copyBody.sections, replaySection{pattern: value, length: int64(len(value))})
+	return copyBody
+}
+
+type repeatSectionReader struct {
+	section replaySection
+	read    int64
+}
+
+func (reader *repeatSectionReader) Read(dst []byte) (int, error) {
+	if reader.read >= reader.section.length {
+		return 0, io.EOF
+	}
+	remaining := reader.section.length - reader.read
+	if int64(len(dst)) > remaining {
+		dst = dst[:int(remaining)]
+	}
+	if len(dst) == 0 {
+		return 0, nil
+	}
+	pattern := reader.section.pattern
+	offset := reader.read % int64(len(pattern))
+	n := copy(dst, pattern[offset:])
+	if n < len(dst) {
+		n += copy(dst[n:], pattern[:offset])
+	}
+	for n < len(dst) {
+		n += copy(dst[n:], dst[:n])
+	}
+	reader.read += int64(len(dst))
+	return len(dst), nil
+}
+
+type replayableStreamBody struct {
+	reader io.ReadCloser
+	source *ReplayBody
+}
+
+func newReplayableStreamBody(source *ReplayBody) *replayableStreamBody {
+	return &replayableStreamBody{reader: source.NewReader(), source: source}
+}
+
+func (body *replayableStreamBody) Read(dst []byte) (int, error) {
+	n, err := body.reader.Read(dst)
+	if err == io.EOF {
+		if closeErr := body.reader.Close(); closeErr != nil {
+			return n, closeErr
+		}
+		body.reader = body.source.NewReader()
+	}
+	return n, err
+}
+
+func (body *replayableStreamBody) Close() error {
+	return body.reader.Close()
+}
+
+func setReplayBody(request *retryablehttp.Request, source *ReplayBody) {
+	request.Body = newReplayableStreamBody(source)
+	request.GetBody = func() (io.ReadCloser, error) {
+		return newReplayableStreamBody(source), nil
+	}
+	request.ContentLength = source.Len()
+}
+
+func replayBodyPlaceholder(text string, values map[string]interface{}) (*ReplayBody, bool) {
+	var suffix string
+	if strings.HasSuffix(text, "\r\n") {
+		text, suffix = strings.TrimSuffix(text, "\r\n"), "\r\n"
+	} else if strings.HasSuffix(text, "\n") {
+		text, suffix = strings.TrimSuffix(text, "\n"), "\n"
+	}
+	if !strings.HasPrefix(text, "{{") || !strings.HasSuffix(text, "}}") {
+		return nil, false
+	}
+	name := strings.TrimSpace(text[2 : len(text)-2])
+	body, ok := values[name].(*ReplayBody)
+	if !ok {
+		return nil, false
+	}
+	return body.appendLiteral(suffix), true
+}

pkg/protocols/http/replay_body_test.go

@@ -0,0 +1,249 @@
+package http
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"math/rand"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"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/protocols/common/contextargs"
+	"github.com/projectdiscovery/retryablehttp-go"
+)
+
+func TestReplayBodyExactBytesAndIndependentReaders(t *testing.T) {
+	part := "--\nContent-Disposition:name=a[]" + strings.Repeat("[a]", 62) + ";filename=\n\n"
+	for _, probe := range []struct{ sizeMiB, parts int }{{1, 4000}, {8, 1020}, {20, 1020}} {
+		size := probe.sizeMiB * 1048576
+		fillerCount := (size - len(part)*probe.parts) / 3
+		want := "--\n" + strings.Repeat("x:\n", fillerCount) + strings.Repeat(part, probe.parts)[3:]
+		want = want[:len(want)-1]
+		body, err := NewReplayBody([]ReplayPart{
+			{Pattern: "--\n", Count: 1},
+			{Pattern: "x:\n", Count: int64(fillerCount)},
+			{Pattern: part[3:], Count: 1},
+			{Pattern: part, Count: int64(probe.parts - 1)},
+		}, 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if body.Len() != int64(len(want)) {
+			t.Fatalf("%d MiB: got length %d, want %d", probe.sizeMiB, body.Len(), len(want))
+		}
+		for replay := 0; replay < 3; replay++ {
+			reader := body.NewReader()
+			got, err := io.ReadAll(reader)
+			if err != nil || string(got) != want {
+				t.Fatalf("%d MiB replay %d differs: %v", probe.sizeMiB, replay, err)
+			}
+			if err := reader.Close(); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+}
+
+func TestReplayBodyArbitraryPatternsAndBufferSizes(t *testing.T) {
+	random := rand.New(rand.NewSource(109618))
+	patterns := []string{"a", "abc", "x:\n", "\x00\xff", "🦊"}
+	for caseIndex := range 200 {
+		parts := make([]ReplayPart, 0, 4)
+		var original strings.Builder
+		for range random.Intn(5) {
+			pattern := patterns[random.Intn(len(patterns))]
+			count := random.Intn(8)
+			parts = append(parts, ReplayPart{Pattern: pattern, Count: int64(count)})
+			original.WriteString(strings.Repeat(pattern, count))
+		}
+		full := original.String()
+		trim := random.Intn(len(full) + 1)
+		want := full[:len(full)-trim]
+		body, err := NewReplayBody(parts, int64(trim))
+		if err != nil {
+			t.Fatalf("case %d: create body: %v", caseIndex, err)
+		}
+		if body.Len() != int64(len(want)) {
+			t.Fatalf("case %d: length = %d, want %d", caseIndex, body.Len(), len(want))
+		}
+		for _, bufferSize := range []int{1, 2, 3, 7, 64} {
+			reader := body.NewReader()
+			buffer := make([]byte, bufferSize)
+			var got bytes.Buffer
+			for {
+				read, readErr := reader.Read(buffer)
+				got.Write(buffer[:read])
+				if readErr == io.EOF {
+					break
+				}
+				if readErr != nil || read == 0 {
+					t.Fatalf("case %d, buffer %d: read %d bytes: %v", caseIndex, bufferSize, read, readErr)
+				}
+			}
+			if err := reader.Close(); err != nil {
+				t.Fatalf("case %d, buffer %d: close reader: %v", caseIndex, bufferSize, err)
+			}
+			if got.String() != want {
+				t.Fatalf("case %d, buffer %d: replay differs", caseIndex, bufferSize)
+			}
+		}
+	}
+}
+
+func TestReplayBodyRejectsInvalidLengths(t *testing.T) {
+	for _, tc := range []struct {
+		parts []ReplayPart
+		trim  int64
+	}{
+		{parts: []ReplayPart{{Pattern: "x", Count: -1}}},
+		{parts: []ReplayPart{{Pattern: "xx", Count: int64(^uint64(0) >> 1)}}},
+		{parts: []ReplayPart{{Pattern: "x", Count: 1}}, trim: 2},
+		{parts: []ReplayPart{{Pattern: "x", Count: 1}}, trim: -1},
+	} {
+		if _, err := NewReplayBody(tc.parts, tc.trim); err == nil {
+			t.Fatalf("accepted invalid replay body: %+v", tc)
+		}
+	}
+}
+
+func TestReplayBodyRetryAndRedirect(t *testing.T) {
+	value := strings.Repeat("probe", 1<<18)
+	for _, tc := range []struct {
+		name             string
+		firstStatus      int
+		redirectLocation string
+	}{
+		{name: "retry", firstStatus: http.StatusServiceUnavailable},
+		{name: "redirect", firstStatus: http.StatusTemporaryRedirect, redirectLocation: "/final"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var received int
+			transport := replayTransport(func(request *http.Request) (*http.Response, error) {
+				body, err := io.ReadAll(request.Body)
+				if err != nil || string(body) != value {
+					t.Errorf("delivery %d body mismatch: %v", received, err)
+				}
+				received++
+				status := http.StatusOK
+				headers := make(http.Header)
+				if received == 1 {
+					status = tc.firstStatus
+					if tc.redirectLocation != "" {
+						headers.Set("Location", tc.redirectLocation)
+					}
+				}
+				return &http.Response{
+					StatusCode: status,
+					Header:     headers,
+					Body:       io.NopCloser(strings.NewReader("")),
+					Request:    request,
+				}, nil
+			})
+			source, err := NewReplayBody([]ReplayPart{{Pattern: "probe", Count: 1 << 18}}, 0)
+			if err != nil {
+				t.Fatal(err)
+			}
+			request, err := retryablehttp.NewRequest(http.MethodPost, "http://example.test/start", nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			setReplayBody(request, source)
+			if err := reuseRequestBody(request); err != nil {
+				t.Fatal(err)
+			}
+			client := retryablehttp.NewClient(retryablehttp.Options{
+				HttpClient:   &http.Client{Transport: transport, Timeout: time.Second},
+				RetryMax:     1,
+				RetryWaitMin: time.Nanosecond,
+				RetryWaitMax: time.Nanosecond,
+				CheckRetry: func(_ context.Context, response *http.Response, err error) (bool, error) {
+					return err == nil && response.StatusCode == http.StatusServiceUnavailable, err
+				},
+			})
+			response, err := client.Do(request)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() {
+				if err := response.Body.Close(); err != nil {
+					t.Errorf("close replay response: %v", err)
+				}
+			})
+			if response.StatusCode != http.StatusOK || received != 2 {
+				t.Fatalf("got status %d and %d deliveries", response.StatusCode, received)
+			}
+		})
+	}
+}
+
+func TestReplayBodyRawAndStructuredRequests(t *testing.T) {
+	for _, tc := range []struct {
+		raw, selfContained bool
+	}{
+		{raw: false},
+		{raw: true},
+		{raw: true, selfContained: true},
+	} {
+		options := testutils.DefaultOptions
+		testutils.Init(options)
+		request := &Request{ID: "replay-body-test", Name: "replay-body-test", Method: HTTPMethodTypeHolder{MethodType: HTTPPost}, SelfContained: tc.selfContained}
+		if tc.selfContained {
+			request.Raw = []string{"POST http://127.0.0.1/probe HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n{{probe_body}}\r\n"}
+		} else if tc.raw {
+			request.Raw = []string{"POST /probe HTTP/1.1\r\nHost: {{Hostname}}\r\nX-Probe: yes\r\n\r\n{{probe_body}}\r\n"}
+		} else {
+			request.Path = []string{"{{BaseURL}}/probe"}
+			request.Body = "{{probe_body}}"
+		}
+		executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+			ID: "replay-body-test", Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "replay-body-test"},
+		})
+		if err := request.Compile(executerOpts); err != nil {
+			t.Fatal(err)
+		}
+		body, err := NewReplayBody([]ReplayPart{{Pattern: "abc", Count: 3}}, 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		inputURL := "http://example.com"
+		if tc.selfContained {
+			inputURL = ""
+		}
+		input := contextargs.NewWithInput(context.Background(), inputURL)
+		executerOpts.GetTemplateCtx(input.MetaInput).Set("probe_body", body)
+		generator := request.newGenerator(false)
+		data, payloads, ok := generator.nextValue()
+		if !ok {
+			t.Fatal("request generator returned no request")
+		}
+		generated, err := generator.Make(context.Background(), input, data, payloads, map[string]interface{}{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if generated.request.Method != http.MethodPost || generated.request.ContentLength != body.Len() {
+			t.Fatalf("raw=%v self-contained=%v: method=%s length=%d", tc.raw, tc.selfContained, generated.request.Method, generated.request.ContentLength)
+		}
+		if err := reuseRequestBody(generated.request); err != nil {
+			t.Fatal(err)
+		}
+		for replay := 0; replay < 2; replay++ {
+			reader, err := generated.request.GetBody()
+			if err != nil {
+				t.Fatal(err)
+			}
+			got, err := io.ReadAll(reader)
+			if err != nil || string(got) != "abcabcab" {
+				t.Fatalf("raw=%v self-contained=%v replay=%d: %q %v", tc.raw, tc.selfContained, replay, got, err)
+			}
+		}
+		dumped, err := dump(generated, input.MetaInput.Input)
+		if err != nil || !bytes.HasSuffix(dumped, []byte("abcabcab")) {
+			t.Fatalf("raw=%v self-contained=%v dump mismatch: %v", tc.raw, tc.selfContained, err)
+		}
+	}
+}

pkg/protocols/http/replay_path_benchmark_test.go

@@ -0,0 +1,142 @@
+package http
+
+import (
+	"strings"
+	"testing"
+
+	retryablehttp "github.com/projectdiscovery/retryablehttp-go"
+)
+
+var replayBenchmarkDump []byte
+
+func BenchmarkLargeBodyRequestStages(b *testing.B) {
+	body := strings.Repeat("x", 20<<20)
+	newRequest := func(b *testing.B) *retryablehttp.Request {
+		b.Helper()
+		request, err := retryablehttp.NewRequest("POST", "https://example.test/", body)
+		if err != nil {
+			b.Fatal(err)
+		}
+		return request
+	}
+
+	b.Run("construct", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			_ = newRequest(b)
+		}
+	})
+	b.Run("reuse", func(b *testing.B) {
+		request := newRequest(b)
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			if err := reuseRequestBody(request); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("dump", func(b *testing.B) {
+		request := &generatedRequest{request: newRequest(b)}
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			var err error
+			replayBenchmarkDump, err = dump(request, "https://example.test/")
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+	b.Run("wholePath", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			request := newRequest(b)
+			if err := reuseRequestBody(request); err != nil {
+				b.Fatal(err)
+			}
+			generated := &generatedRequest{request: request}
+			for i := 0; i < 2; i++ {
+				var err error
+				replayBenchmarkDump, err = dump(generated, "https://example.test/")
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		}
+	})
+	b.Run("wholePathReplayable", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			request, err := retryablehttp.NewRequest("POST", "https://example.test/", nil)
+			if err != nil {
+				b.Fatal(err)
+			}
+			setReplayableStringBody(request, body)
+			if err := reuseRequestBody(request); err != nil {
+				b.Fatal(err)
+			}
+			generated := &generatedRequest{request: request}
+			for i := 0; i < 2; i++ {
+				replayBenchmarkDump, err = dump(generated, "https://example.test/")
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		}
+	})
+	b.Run("wholePathStreamed", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			source, err := NewReplayBody([]ReplayPart{{Pattern: "x", Count: 20 << 20}}, 0)
+			if err != nil {
+				b.Fatal(err)
+			}
+			request, err := retryablehttp.NewRequest("POST", "https://example.test/", nil)
+			if err != nil {
+				b.Fatal(err)
+			}
+			setReplayBody(request, source)
+			if err := reuseRequestBody(request); err != nil {
+				b.Fatal(err)
+			}
+			generated := &generatedRequest{request: request}
+			for i := 0; i < 2; i++ {
+				replayBenchmarkDump, err = dump(generated, "https://example.test/")
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		}
+	})
+	b.Run("wholePathStreamedWithProjectScope", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		for range b.N {
+			source, err := NewReplayBody([]ReplayPart{{Pattern: "x", Count: 20 << 20}}, 0)
+			if err != nil {
+				b.Fatal(err)
+			}
+			request, err := retryablehttp.NewRequest("POST", "https://example.test/", nil)
+			if err != nil {
+				b.Fatal(err)
+			}
+			setReplayBody(request, source)
+			if err := reuseRequestBody(request); err != nil {
+				b.Fatal(err)
+			}
+			generated := &generatedRequest{request: request}
+			for i := 0; i < 2; i++ {
+				replayBenchmarkDump, err = dump(generated, "https://example.test/")
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+			replayBenchmarkDump = getHTTPProjectCacheScope(replayBenchmarkDump, "https", "example.test")
+		}
+	})
+}

pkg/protocols/http/replayable_body.go

@@ -0,0 +1,34 @@
+package http
+
+import (
+	"io"
+	"strings"
+
+	"github.com/projectdiscovery/retryablehttp-go"
+)
+
+type replayableStringBody struct {
+	reader *strings.Reader
+	value  string
+}
+
+func (body *replayableStringBody) Read(dst []byte) (int, error) {
+	n, err := body.reader.Read(dst)
+	if err == io.EOF {
+		body.reader.Reset(body.value)
+	}
+	return n, err
+}
+
+func (body *replayableStringBody) Close() error {
+	return nil
+}
+
+func setReplayableStringBody(request *retryablehttp.Request, value string) {
+	size := int64(len(value))
+	request.Body = &replayableStringBody{reader: strings.NewReader(value), value: value}
+	request.GetBody = func() (io.ReadCloser, error) {
+		return &replayableStringBody{reader: strings.NewReader(value), value: value}, nil
+	}
+	request.ContentLength = size
+}

pkg/protocols/http/replayable_body_test.go

@@ -0,0 +1,184 @@
+package http
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/projectdiscovery/retryablehttp-go"
+)
+
+type replayTransport func(*http.Request) (*http.Response, error)
+
+func (transport replayTransport) RoundTrip(request *http.Request) (*http.Response, error) {
+	return transport(request)
+}
+
+func TestReplayableStringBodyPreservesRequest(t *testing.T) {
+	for _, tc := range []struct {
+		name, method, body, transferEncoding string
+		contentLength                        int64
+	}{
+		{name: "post", method: http.MethodPost, body: "body", contentLength: 1},
+		{name: "get", method: http.MethodGet, body: "body", contentLength: 3},
+		{name: "head", method: http.MethodHead, body: "body", contentLength: 0},
+		{name: "chunked", method: http.MethodPost, body: "body", transferEncoding: "chunked", contentLength: -1},
+		{name: "empty", method: http.MethodPost, body: "", contentLength: 0},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			legacy, err := retryablehttp.NewRequest(tc.method, "https://example.test/", tc.body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			current, err := retryablehttp.NewRequest(tc.method, "https://example.test/", nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			setReplayableStringBody(current, tc.body)
+			legacy.ContentLength = tc.contentLength
+			current.ContentLength = tc.contentLength
+			if tc.transferEncoding != "" {
+				legacy.Header.Set("Transfer-Encoding", tc.transferEncoding)
+				current.Header.Set("Transfer-Encoding", tc.transferEncoding)
+			}
+			for _, request := range []*retryablehttp.Request{legacy, current} {
+				if err := reuseRequestBody(request); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if legacy.ContentLength != current.ContentLength {
+				t.Fatalf("content length changed: legacy=%d replay=%d", legacy.ContentLength, current.ContentLength)
+			}
+			for _, request := range []*retryablehttp.Request{legacy, current} {
+				got, err := io.ReadAll(request.Body)
+				if err != nil || string(got) != tc.body {
+					t.Fatalf("request body mismatch: %q, %v", got, err)
+				}
+				for i := 0; i < 2; i++ {
+					fresh, err := request.GetBody()
+					if err != nil {
+						t.Fatal(err)
+					}
+					got, err = io.ReadAll(fresh)
+					if err != nil || string(got) != tc.body {
+						t.Fatalf("replay %d mismatch: %q, %v", i, got, err)
+					}
+				}
+			}
+			legacyDump, legacyErr := dump(&generatedRequest{request: legacy}, "https://example.test/")
+			currentDump, currentErr := dump(&generatedRequest{request: current}, "https://example.test/")
+			if legacyErr != nil || currentErr != nil || !bytes.Equal(legacyDump, currentDump) {
+				t.Fatalf("request dump mismatch: legacy=%v replay=%v", legacyErr, currentErr)
+			}
+		})
+	}
+}
+
+func TestReplayableStringBodyConcurrentDump(t *testing.T) {
+	value := strings.Repeat("probe", 1<<20)
+	request, err := retryablehttp.NewRequest(http.MethodPost, "https://example.test/", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	setReplayableStringBody(request, value)
+	if err := reuseRequestBody(request); err != nil {
+		t.Fatal(err)
+	}
+	generated := &generatedRequest{request: request}
+
+	var group sync.WaitGroup
+	errors := make(chan error, 4)
+	for i := 0; i < 4; i++ {
+		group.Add(1)
+		go func() {
+			defer group.Done()
+			got, err := dump(generated, "https://example.test/")
+			if err == nil && !bytes.HasSuffix(got, []byte(value)) {
+				err = io.ErrUnexpectedEOF
+			}
+			errors <- err
+		}()
+	}
+	got, err := io.ReadAll(request.Body)
+	if err != nil || string(got) != value {
+		t.Fatalf("transport body mismatch: %v", err)
+	}
+	group.Wait()
+	close(errors)
+	for err := range errors {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestReplayableStringBodyRetryAndRedirect(t *testing.T) {
+	value := strings.Repeat("probe", 1<<18)
+	for _, tc := range []struct {
+		name             string
+		firstStatus      int
+		redirectLocation string
+	}{
+		{name: "retry", firstStatus: http.StatusServiceUnavailable},
+		{name: "redirect", firstStatus: http.StatusTemporaryRedirect, redirectLocation: "/final"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var received int
+			transport := replayTransport(func(request *http.Request) (*http.Response, error) {
+				body, err := io.ReadAll(request.Body)
+				if err != nil || string(body) != value {
+					t.Errorf("delivery %d body mismatch: %v", received, err)
+				}
+				received++
+				status := http.StatusOK
+				headers := make(http.Header)
+				if received == 1 {
+					status = tc.firstStatus
+					if tc.redirectLocation != "" {
+						headers.Set("Location", tc.redirectLocation)
+					}
+				}
+				return &http.Response{
+					StatusCode: status,
+					Header:     headers,
+					Body:       io.NopCloser(strings.NewReader("")),
+					Request:    request,
+				}, nil
+			})
+			request, err := retryablehttp.NewRequest(http.MethodPost, "http://example.test/start", nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			setReplayableStringBody(request, value)
+			if err := reuseRequestBody(request); err != nil {
+				t.Fatal(err)
+			}
+			client := retryablehttp.NewClient(retryablehttp.Options{
+				HttpClient:   &http.Client{Transport: transport, Timeout: time.Second},
+				RetryMax:     1,
+				RetryWaitMin: time.Nanosecond,
+				RetryWaitMax: time.Nanosecond,
+				CheckRetry: func(_ context.Context, response *http.Response, err error) (bool, error) {
+					return err == nil && response.StatusCode == http.StatusServiceUnavailable, err
+				},
+			})
+			response, err := client.Do(request)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() {
+				if err := response.Body.Close(); err != nil {
+					t.Errorf("close replay response: %v", err)
+				}
+			})
+			if response.StatusCode != http.StatusOK || received != 2 {
+				t.Fatalf("got status %d and %d deliveries", response.StatusCode, received)
+			}
+		})
+	}
+}

pkg/protocols/http/request.go

@@ -664,10 +664,7 @@ executionLoop:
 			hasInteractMatchers := interactsh.HasMatchers(request.CompiledOperators)
 
 			ctx := request.newContext(input)
-			ctxWithTimeout, cancel := context.WithTimeoutCause(ctx, request.options.Options.GetTimeouts().HttpTimeout, ErrHttpEngineRequestDeadline)
-			defer cancel()
-
-			generatedHttpRequest, err := generator.Make(ctxWithTimeout, input, data, payloads, dynamicValue)
+			generatedHttpRequest, err := generator.Make(ctx, input, data, payloads, dynamicValue)
 			if err != nil {
 				if err == types.ErrNoMoreRequests {
 					return false, true, nil, nil
@@ -685,10 +682,12 @@ executionLoop:
 			if err := request.rateLimitTake(ctx, hostBackoffKey(generatedHttpRequest, input)); err != nil {
 				return false, true, nil, err
 			}
-
 			if generatedHttpRequest.customCancelFunction != nil {
 				defer generatedHttpRequest.customCancelFunction()
 			}
+			if generatedHttpRequest.request != nil && generatedHttpRequest.customCancelFunction == nil {
+				generatedHttpRequest.startRequestTimeout = true
+			}
 
 			hasInteractMarkers := interactsh.HasMarkers(data) || len(generatedHttpRequest.interactshURLs) > 0
 			if input.MetaInput.Input == "" {
@@ -1011,6 +1010,18 @@ func looksLikeHTTPToHTTPSMismatch(resp *http.Response) bool {
 // post-send dump reads one while the transport may still be writing another.
 // GET, HEAD, and chunked requests keep their existing content length.
 func reuseRequestBody(req *retryablehttp.Request) error {
+	if body, ok := req.Body.(*replayableStreamBody); ok && req.GetBody != nil {
+		if !stringsutil.EqualFoldAny(req.Method, http.MethodGet, http.MethodHead) && req.Header.Get("Transfer-Encoding") != "chunked" {
+			req.ContentLength = body.source.Len()
+		}
+		return nil
+	}
+	if body, ok := req.Body.(*replayableStringBody); ok && req.GetBody != nil {
+		if !stringsutil.EqualFoldAny(req.Method, http.MethodGet, http.MethodHead) && req.Header.Get("Transfer-Encoding") != "chunked" {
+			req.ContentLength = int64(len(body.value))
+		}
+		return nil
+	}
 	body, err := io.ReadAll(req.Body)
 	if err != nil {
 		return err
@@ -1045,6 +1056,11 @@ func (request *Request) executeRequest(input *contextargs.Context, generatedRequ
 	if request.isUnresponsiveAddress(contextargs.GetCopyIfHostOutdated(input, generatedRequest.URL())) {
 		return fmt.Errorf("hostErrorsCache : host %s is unresponsive", input.MetaInput.Input)
 	}
+	releaseBody, err := admitLargeBody(input.Context(), generatedRequest)
+	if err != nil {
+		return err
+	}
+	defer releaseBody()
 
 	// wrap one more callback for validation and fixing event
 	callback := func(event *output.InternalWrappedEvent) {
@@ -1100,13 +1116,15 @@ func (request *Request) executeRequest(input *contextargs.Context, generatedRequ
 			return dumpError
 		}
 		maxBodylimit := request.getEffectiveReadLimit()
-		if generatedRequest.request != nil && generatedRequest.request.Request != nil && generatedRequest.request.Request.URL != nil {
-			projectCacheKey = getHTTPProjectCacheScope(dumpedRequest, generatedRequest.request.Scheme, generatedRequest.request.Request.URL.Host)
-		} else {
-			projectCacheKey = dumpedRequest
-		}
-		if maxBodylimit > 0 {
-			projectCacheKey = append(projectCacheKey, []byte(fmt.Sprintf(":limit:%d", maxBodylimit))...)
+		if request.options.ProjectFile != nil {
+			if generatedRequest.request != nil && generatedRequest.request.Request != nil && generatedRequest.request.Request.URL != nil {
+				projectCacheKey = getHTTPProjectCacheScope(dumpedRequest, generatedRequest.request.Scheme, generatedRequest.request.Request.URL.Host)
+			} else {
+				projectCacheKey = dumpedRequest
+			}
+			if maxBodylimit > 0 {
+				projectCacheKey = append(projectCacheKey, []byte(fmt.Sprintf(":limit:%d", maxBodylimit))...)
+			}
 		}
 		dumpedRequestString := string(dumpedRequest)
 
@@ -1272,6 +1290,11 @@ func (request *Request) executeRequest(input *contextargs.Context, generatedRequ
 				}
 			}
 
+			if generatedRequest.startRequestTimeout {
+				timedCtx, cancel := context.WithTimeoutCause(generatedRequest.request.Context(), request.options.Options.GetTimeouts().HttpTimeout, ErrHttpEngineRequestDeadline)
+				defer cancel()
+				generatedRequest.request.WithContext(timedCtx)
+			}
 			resp, err = executingClient.Do(generatedRequest.request)
 			sent = true
 
@@ -1401,13 +1424,26 @@ func (request *Request) executeRequest(input *contextargs.Context, generatedRequ
 
 	var curlCommand string
 	if !request.Unsafe && resp != nil && generatedRequest.request != nil && resp.Request != nil && !request.Race {
-		bodyBytes, _ := generatedRequest.request.BodyBytes()
-		// Use a clone to avoid a race condition with the http transport
 		req := resp.Request.Clone(resp.Request.Context())
-		req.Body = io.NopCloser(bytes.NewReader(bodyBytes))
-		command, err := http2curl.GetCurlCommand(req)
-		if err == nil && command != nil {
-			curlCommand = command.String()
+		if generatedRequest.request.GetBody != nil {
+			body, bodyErr := generatedRequest.request.GetBody()
+			if bodyErr != nil {
+				gologger.Warning().Msgf("[%s] Could not replay request body for curl command: %s", request.options.TemplateID, bodyErr)
+			} else {
+				req.Body = body
+				command, commandErr := http2curl.GetCurlCommand(req)
+				if closeErr := body.Close(); closeErr != nil {
+					gologger.Warning().Msgf("[%s] Could not close replayed request body: %s", request.options.TemplateID, closeErr)
+				}
+				if commandErr == nil && command != nil {
+					curlCommand = command.String()
+				}
+			}
+		} else if req.Body == nil {
+			command, commandErr := http2curl.GetCurlCommand(req)
+			if commandErr == nil && command != nil {
+				curlCommand = command.String()
+			}
 		}
 	}
 

pkg/tmplexec/flow/body_admission.go

@@ -0,0 +1,19 @@
+package flow
+
+import (
+	"fmt"
+
+	"github.com/spf13/cast"
+)
+
+func declaredRequestBodySize(metadata map[string]interface{}) (int64, bool, error) {
+	value, exists := metadata["max-request-body-bytes"]
+	if !exists {
+		return 0, false, nil
+	}
+	size, err := cast.ToInt64E(value)
+	if err != nil || size < 0 {
+		return 0, true, fmt.Errorf("invalid max-request-body-bytes metadata: %v", value)
+	}
+	return size, true, nil
+}

pkg/tmplexec/flow/body_admission_test.go

@@ -0,0 +1,23 @@
+package flow
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDeclaredRequestBodySize(t *testing.T) {
+	size, exists, err := declaredRequestBodySize(nil)
+	require.NoError(t, err)
+	require.False(t, exists)
+	require.Zero(t, size)
+
+	size, exists, err = declaredRequestBodySize(map[string]interface{}{"max-request-body-bytes": 20 << 20})
+	require.NoError(t, err)
+	require.True(t, exists)
+	require.EqualValues(t, 20<<20, size)
+
+	_, exists, err = declaredRequestBodySize(map[string]interface{}{"max-request-body-bytes": -1})
+	require.True(t, exists)
+	require.Error(t, err)
+}

pkg/tmplexec/flow/flow_executor.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"fmt"
 	"io"
+	"math"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -12,6 +13,7 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/js/compiler"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/generators"
+	httpbody "github.com/projectdiscovery/nuclei/v3/pkg/protocols/http"
 	"github.com/projectdiscovery/nuclei/v3/pkg/scan"
 	templateTypes "github.com/projectdiscovery/nuclei/v3/pkg/templates/types"
 
@@ -220,6 +222,17 @@ func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 		return ctx.Context().Err()
 	default:
 	}
+	size, declared, err := declaredRequestBodySize(f.options.TemplateInfo.Metadata)
+	if err != nil {
+		return fmt.Errorf("template %s: %w", f.options.TemplateID, err)
+	}
+	if declared {
+		release, err := httpbody.AdmitDeclaredBody(ctx.Context(), size)
+		if err != nil {
+			return err
+		}
+		defer release()
+	}
 
 	f.ctx.Input = ctx.Input
 	// -----Load all types of variables-----
@@ -240,6 +253,7 @@ func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 	defer func() {
 		// remove set builtin
 		_ = runtime.GlobalObject().Delete("set")
+		_ = runtime.GlobalObject().Delete("replayBody")
 		_ = runtime.GlobalObject().Delete("template")
 		for proto := range f.protoFunctions {
 			_ = runtime.GlobalObject().Delete(proto)
@@ -270,6 +284,9 @@ func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 	}); err != nil {
 		return err
 	}
+	if err := registerReplayBody(runtime); err != nil {
+		return err
+	}
 	// also register functions that allow executing protocols from js
 	for proto, fn := range f.protoFunctions {
 		if err := runtime.Set(proto, fn); err != nil {
@@ -288,7 +305,7 @@ func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 	runtime.SetContextValue("executionId", f.options.Options.ExecutionId)
 
 	// pass flow and execute the js vm and handle errors
-	_, err := runtime.RunProgram(f.program)
+	_, err = runtime.RunProgram(f.program)
 	f.reconcileProgress()
 	if err != nil {
 		ctx.LogError(err)
@@ -303,6 +320,44 @@ func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 	return nil
 }
 
+func registerReplayBody(runtime *goja.Runtime) error {
+	return runtime.Set("replayBody", func(call goja.FunctionCall) goja.Value {
+		parts, ok := call.Argument(0).Export().([]interface{})
+		if !ok {
+			panic(runtime.NewTypeError("replayBody expects an array of [pattern, count] pairs"))
+		}
+		trimTail := call.Argument(1).ToFloat()
+		if math.IsNaN(trimTail) || trimTail < 0 || trimTail >= float64(math.MaxInt64) || math.Trunc(trimTail) != trimTail {
+			panic(runtime.NewTypeError("replayBody tail trim must be a nonnegative integer"))
+		}
+		replayParts := make([]httpbody.ReplayPart, 0, len(parts))
+		for _, part := range parts {
+			pair, ok := part.([]interface{})
+			if !ok || len(pair) != 2 {
+				panic(runtime.NewTypeError("replayBody parts must be [pattern, count] pairs"))
+			}
+			pattern, ok := pair[0].(string)
+			if !ok {
+				panic(runtime.NewTypeError("replayBody pattern must be a string"))
+			}
+			count, ok := pair[1].(int64)
+			if !ok {
+				value, floatOK := pair[1].(float64)
+				if !floatOK || math.IsNaN(value) || value < 0 || value >= float64(math.MaxInt64) || math.Trunc(value) != value {
+					panic(runtime.NewTypeError("replayBody count must be a nonnegative integer"))
+				}
+				count = int64(value)
+			}
+			replayParts = append(replayParts, httpbody.ReplayPart{Pattern: pattern, Count: count})
+		}
+		body, bodyErr := httpbody.NewReplayBody(replayParts, int64(trimTail))
+		if bodyErr != nil {
+			panic(runtime.NewTypeError(bodyErr.Error()))
+		}
+		return runtime.ToValue(body)
+	})
+}
+
 // reconcileProgress settles progress, whose total assumed each request runs
 // once: requests the flow never ran count as processed, as protocols do for
 // skipped requests, and extra runs of a request are added to the total

pkg/tmplexec/flow/replay_body_test.go

@@ -0,0 +1,95 @@
+package flow
+
+import (
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/projectdiscovery/goja"
+	httpbody "github.com/projectdiscovery/nuclei/v3/pkg/protocols/http"
+)
+
+func TestReplayBodyBuiltin(t *testing.T) {
+	runtime := goja.New()
+	if err := registerReplayBody(runtime); err != nil {
+		t.Fatal(err)
+	}
+	value, err := runtime.RunString(`replayBody([['abc', 3], ['xyz', 1]], 1)`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, ok := value.Export().(*httpbody.ReplayBody)
+	if !ok {
+		t.Fatalf("unexpected replay body type %T", value.Export())
+	}
+	reader := body.NewReader()
+	got, err := io.ReadAll(reader)
+	if err != nil || string(got) != "abcabcabcxy" {
+		t.Fatalf("replay body = %q, err = %v", got, err)
+	}
+}
+
+func TestReplayBodyMatchesLargeProbe(t *testing.T) {
+	runtime := goja.New()
+	if err := registerReplayBody(runtime); err != nil {
+		t.Fatal(err)
+	}
+	_, err := runtime.RunString(`
+		function repeatStr(s, n) {
+			var r = '';
+			while (n > 0) {
+				if (n & 1) r += s;
+				s += s;
+				n >>= 1;
+			}
+			return r;
+		}
+		function originalProbe(sizeMib, parts) {
+			var size = sizeMib * 1048576;
+			var nested = repeatStr('[a]', 62);
+			var part = '--\nContent-Disposition:name=a[]' + nested + ';filename=\n\n';
+			var base = repeatStr(part, parts);
+			var fillerCount = Math.floor((size - base.length) / 3);
+			var filler = repeatStr('x:\n', fillerCount);
+			var body = '--\n' + filler + base.substring(3);
+			return body.substring(0, body.length - 1);
+		}
+		function replayProbe(sizeMib, parts) {
+			var size = sizeMib * 1048576;
+			var nested = repeatStr('[a]', 62);
+			var part = '--\nContent-Disposition:name=a[]' + nested + ';filename=\n\n';
+			var fillerCount = Math.floor((size - part.length * parts) / 3);
+			return replayBody([['--\n', 1], ['x:\n', fillerCount], [part.substring(3), 1], [part, parts - 1]], 1);
+		}
+	`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	original, _ := goja.AssertFunction(runtime.Get("originalProbe"))
+	replay, _ := goja.AssertFunction(runtime.Get("replayProbe"))
+	for _, test := range []struct {
+		sizeMib int
+		parts   int
+	}{{1, 4000}, {8, 1020}, {20, 1020}} {
+		t.Run(fmt.Sprintf("%dMiB", test.sizeMib), func(t *testing.T) {
+			args := []goja.Value{runtime.ToValue(test.sizeMib), runtime.ToValue(test.parts)}
+			originalValue, err := original(goja.Undefined(), args...)
+			if err != nil {
+				t.Fatal(err)
+			}
+			replayValue, err := replay(goja.Undefined(), args...)
+			if err != nil {
+				t.Fatal(err)
+			}
+			body := replayValue.Export().(*httpbody.ReplayBody)
+			reader := body.NewReader()
+			got, err := io.ReadAll(reader)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(got) != originalValue.String() {
+				t.Fatalf("%d MiB probe differs from original", test.sizeMib)
+			}
+		})
+	}
+}