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/core/workflow_execute.go

@@ -3,6 +3,7 @@ package core
 import (
 	"fmt"
 	"net/http/cookiejar"
+	"sync"
 	"sync/atomic"
 
 	"github.com/projectdiscovery/gologger"
@@ -32,22 +33,45 @@ func (e *Engine) executeWorkflow(ctx *scan.ScanContext, w *workflows.Workflow) b
 		templateThreads++
 	}
 	swg, _ := syncutil.New(syncutil.WithSize(templateThreads))
+	// pending tracks every started step until it returns. swg only limits
+	// concurrency and its slots can be released early, so completion is
+	// tracked here instead.
+	var pending sync.WaitGroup
 
 	for _, template := range w.Workflows {
 		newCtx := scan.NewScanContext(ctx.Context(), ctx.Input.Clone())
-		if err := e.runWorkflowStep(template, newCtx, results, swg, w); err != nil {
+		// Root steps do not own a waitgroup slot.
+		if err := e.runWorkflowStep(template, newCtx, results, swg, &pending, w, nil); err != nil {
 			gologger.Warning().Msgf(workflowStepExecutionError, template.Template, err)
 		}
 	}
 
-	swg.Wait()
+	pending.Wait()
 
 	return results.Load()
 }
 
+// slotRelease returns a func that gives back one swg slot. It is safe to call
+// from several goroutines and several times; only the first call releases.
+func slotRelease(swg *syncutil.AdaptiveWaitGroup) func() {
+	var once sync.Once
+	return func() {
+		once.Do(swg.Done)
+	}
+}
+
 // runWorkflowStep runs a workflow step for the workflow. It executes the workflow
 // in a recursive manner running all subtemplates and matchers.
-func (e *Engine) runWorkflowStep(template *workflows.WorkflowTemplate, ctx *scan.ScanContext, results *atomic.Bool, swg *syncutil.AdaptiveWaitGroup, w *workflows.Workflow) error {
+//
+// releaseSlot, when non-nil, releases the AdaptiveWaitGroup slot held by the
+// goroutine that is running this step. It must be called before this step
+// blocks on swg.AddWithContext for child steps; otherwise every in-flight
+// parent can hold a slot while waiting for another, which deadlocks when the
+// number of matching nested steps reaches TemplateThreads (see #8074).
+//
+// pending is incremented for every child step before it starts, and each child
+// calls Done when it returns, so the root can wait for the whole tree.
+func (e *Engine) runWorkflowStep(template *workflows.WorkflowTemplate, ctx *scan.ScanContext, results *atomic.Bool, swg *syncutil.AdaptiveWaitGroup, pending *sync.WaitGroup, w *workflows.Workflow, releaseSlot func()) error {
 	var firstMatched bool
 	var err error
 	var mainErr error
@@ -131,18 +155,25 @@ func (e *Engine) runWorkflowStep(template *workflows.WorkflowTemplate, ctx *scan
 						continue
 					}
 
+					// Free this step's slot before blocking on child acquires.
+					if releaseSlot != nil {
+						releaseSlot()
+					}
 					for _, subtemplate := range matcher.Subtemplates {
 						if err := swg.AddWithContext(ctx.Context()); err != nil {
 							return
 						}
+						pending.Add(1)
 
 						go func(subtemplate *workflows.WorkflowTemplate) {
-							defer swg.Done()
+							defer pending.Done()
+							release := slotRelease(swg)
+							defer release()
 
 							// create a new context with the same input but with unset callbacks
 							// clone the Input so that other parallel executions won't overwrite the shared variables when subsequent templates are running
 							subCtx := scan.NewScanContext(ctx.Context(), ctx.Input.Clone())
-							if err := e.runWorkflowStep(subtemplate, subCtx, results, swg, w); err != nil {
+							if err := e.runWorkflowStep(subtemplate, subCtx, results, swg, pending, w, release); err != nil {
 								gologger.Warning().Msgf(workflowStepExecutionError, subtemplate.Template, err)
 							}
 						}(subtemplate)
@@ -162,18 +193,27 @@ func (e *Engine) runWorkflowStep(template *workflows.WorkflowTemplate, ctx *scan
 		return mainErr
 	}
 	if len(template.Subtemplates) > 0 && firstMatched {
+		// Free this step's slot before blocking on child acquires so nested
+		// matching steps cannot exhaust TemplateThreads and deadlock (#8074).
+		if releaseSlot != nil {
+			releaseSlot()
+		}
 		for _, subtemplate := range template.Subtemplates {
 			if err := swg.AddWithContext(ctx.Context()); err != nil {
 				return err
 			}
+			pending.Add(1)
 
 			go func(template *workflows.WorkflowTemplate) {
+				defer pending.Done()
+				release := slotRelease(swg)
+				defer release()
+
 				// create a new context with the same input but with unset callbacks
 				subCtx := scan.NewScanContext(ctx.Context(), ctx.Input.Clone())
-				if err := e.runWorkflowStep(template, subCtx, results, swg, w); err != nil {
+				if err := e.runWorkflowStep(template, subCtx, results, swg, pending, w, release); err != nil {
 					gologger.Warning().Msgf(workflowStepExecutionError, template.Template, err)
 				}
-				swg.Done()
 			}(subtemplate)
 		}
 	}

pkg/core/workflow_execute_test.go

@@ -303,6 +303,89 @@ func TestWorkflowsSameStepExecutersRefreshTemplateContext(t *testing.T) {
 	require.Equal(t, "foo", secondWhoami, "could not refresh workflow context into later executers in the same step")
 }
 
+func TestWorkflowNestedSubtemplatesDoNotDeadlockAtConcurrencyLimit(t *testing.T) {
+	// Repro for #8074: with TemplateThreads equal to the number of matching
+	// nested parents, each parent used to hold a waitgroup slot while calling
+	// AddWithContext for its children, so every slot was occupied by a parent
+	// waiting for another slot and the workflow never finished.
+	progressBar, _ := progress.NewStatsTicker(0, false, false, false, 0)
+	matchOutputs := []*output.InternalWrappedEvent{{
+		OperatorsResult: &operators.Result{},
+		Results:         []*output.ResultEvent{{}},
+	}}
+
+	const parents = 3
+	var entered atomic.Int32
+	allHolding := make(chan struct{})
+	releaseParents := make(chan struct{})
+	leafRuns := &atomic.Int32{}
+
+	leaf := &mockExecuter{result: true, executeScanHook: func(ctx *scan.ScanContext) {
+		leafRuns.Add(1)
+	}}
+	parentExec := &mockExecuter{
+		result:  true,
+		outputs: matchOutputs,
+		executeScanHook: func(ctx *scan.ScanContext) {
+			if entered.Add(1) == parents {
+				close(allHolding)
+			}
+			// Stay inside ExecuteWithResults so the parent still owns its
+			// waitgroup slot while every peer also holds one.
+			<-releaseParents
+		},
+	}
+	mkParent := func() *workflows.WorkflowTemplate {
+		return &workflows.WorkflowTemplate{
+			Executers: []*workflows.ProtocolExecuterPair{{
+				Executer: parentExec,
+				Options:  &protocols.ExecutorOptions{Progress: progressBar},
+			}},
+			Subtemplates: []*workflows.WorkflowTemplate{{
+				Executers: []*workflows.ProtocolExecuterPair{{
+					Executer: leaf,
+					Options:  &protocols.ExecutorOptions{Progress: progressBar},
+				}},
+			}},
+		}
+	}
+
+	workflow := &workflows.Workflow{
+		Options: &protocols.ExecutorOptions{Options: &types.Options{TemplateThreads: parents}},
+		Workflows: []*workflows.WorkflowTemplate{{
+			Executers: []*workflows.ProtocolExecuterPair{{
+				Executer: &mockExecuter{result: true, outputs: matchOutputs},
+				Options:  &protocols.ExecutorOptions{Progress: progressBar},
+			}},
+			Subtemplates: []*workflows.WorkflowTemplate{mkParent(), mkParent(), mkParent()},
+		}},
+	}
+
+	engine := &Engine{}
+	input := contextargs.NewWithInput(context.Background(), "https://test.com")
+	ctx := scan.NewScanContext(context.Background(), input)
+
+	done := make(chan bool, 1)
+	go func() {
+		done <- engine.executeWorkflow(ctx, workflow)
+	}()
+
+	select {
+	case <-allHolding:
+	case <-time.After(2 * time.Second):
+		t.Fatal("nested parents did not all acquire TemplateThreads slots")
+	}
+	close(releaseParents)
+
+	select {
+	case matched := <-done:
+		require.True(t, matched, "nested workflow should report a match")
+		require.EqualValues(t, parents, leafRuns.Load(), "each nested parent should run its leaf")
+	case <-time.After(3 * time.Second):
+		t.Fatal("nested workflow deadlocked at TemplateThreads == matching nested parents")
+	}
+}
+
 func TestWorkflowSubtemplatesUnblockCancelledAcquire(t *testing.T) {
 	progressBar, _ := progress.NewStatsTicker(0, false, false, false, 0)
 	var started atomic.Int32

pkg/js/compiler/pool.go

@@ -31,6 +31,7 @@ import (
 	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/libldap"
 	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/libmssql"
 	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/libmysql"
+	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/libmqtt"
 	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/libnet"
 	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/liboracle"
 	_ "github.com/projectdiscovery/nuclei/v3/pkg/js/generated/go/libpop3"

pkg/js/generated/go/libmqtt/mqtt.go

@@ -0,0 +1,34 @@
+package mqtt
+
+import (
+	lib_mqtt "github.com/projectdiscovery/nuclei/v3/pkg/js/libs/mqtt"
+
+	"github.com/projectdiscovery/goja"
+	"github.com/projectdiscovery/nuclei/v3/pkg/js/gojs"
+)
+
+var (
+	module = gojs.NewGojaModule("nuclei/mqtt")
+)
+
+func init() {
+	module.Set(
+		gojs.Objects{
+			// Functions
+			"IsMQTT":             lib_mqtt.IsMQTT,
+			"IsAnonymousAllowed": lib_mqtt.IsAnonymousAllowed,
+
+			// Var and consts
+
+			// Objects / Classes
+			"Client":         gojs.GetClassConstructor[lib_mqtt.Client](&lib_mqtt.Client{}),
+			"ConnectOptions": gojs.GetClassConstructor[lib_mqtt.ConnectOptions](&lib_mqtt.ConnectOptions{}),
+			"ConnectResult":  gojs.GetClassConstructor[lib_mqtt.ConnectResult](&lib_mqtt.ConnectResult{}),
+			"Message":        gojs.GetClassConstructor[lib_mqtt.Message](&lib_mqtt.Message{}),
+		},
+	).Register()
+}
+
+func Enable(runtime *goja.Runtime) {
+	module.Enable(runtime)
+}

pkg/js/generated/ts/index.ts

@@ -15,6 +15,7 @@ export * as krbroast from './krbroast';
 export * as ldap from './ldap';
 export * as mssql from './mssql';
 export * as mysql from './mysql';
+export * as mqtt from './mqtt';
 export * as net from './net';
 export * as oracle from './oracle';
 export * as pop3 from './pop3';

pkg/js/generated/ts/mqtt.ts

@@ -0,0 +1,190 @@
+
+/**
+ * IsMQTT checks whether the target host:port is running an MQTT broker.
+ * @example
+ * ```javascript
+ * const mqtt = require('nuclei/mqtt');
+ * const isMqtt = mqtt.IsMQTT('acme.com', 1883);
+ * log(isMqtt);
+ * ```
+ */
+export function IsMQTT(host: string, port: number): boolean | null {
+    return null;
+}
+
+/**
+ * IsAnonymousAllowed checks whether an MQTT broker permits unauthenticated / anonymous login.
+ * @example
+ * ```javascript
+ * const mqtt = require('nuclei/mqtt');
+ * const allowed = mqtt.IsAnonymousAllowed('acme.com', 1883);
+ * log(allowed);
+ * ```
+ */
+export function IsAnonymousAllowed(host: string, port: number): boolean | null {
+    return null;
+}
+
+/**
+ * Message represents an MQTT published message.
+ * @example
+ * ```javascript
+ * const mqtt = require('nuclei/mqtt');
+ * const client = new mqtt.Client();
+ * client.Connect('broker.hivemq.com', 1883);
+ * const msgs = client.Subscribe('test/topic', 2);
+ * log(msgs[0].Topic, msgs[0].Payload);
+ * ```
+ */
+export interface Message {
+    Topic?: string;
+    Payload?: string;
+    QoS?: number;
+    Retain?: boolean;
+}
+
+/**
+ * ConnectOptions contains connection options for an MQTT broker.
+ * @example
+ * ```javascript
+ * const mqtt = require('nuclei/mqtt');
+ * const opts = new mqtt.ConnectOptions();
+ * opts.Username = 'admin';
+ * opts.Password = 'admin';
+ * opts.CleanSession = true;
+ * ```
+ */
+export interface ConnectOptions {
+    Username?: string;
+    Password?: string;
+    ClientID?: string;
+    CleanSession?: boolean;
+    KeepAlive?: number;
+    TLS?: boolean;
+    InsecureSkipVerify?: boolean;
+    Timeout?: number;
+}
+
+/**
+ * ConnectResult is the response returned by an MQTT connection attempt.
+ * @example
+ * ```javascript
+ * const mqtt = require('nuclei/mqtt');
+ * const client = new mqtt.Client();
+ * const res = client.Connect('acme.com', 1883);
+ * log(res.Success, res.ReturnCode, res.Reason);
+ * ```
+ */
+export interface ConnectResult {
+    Success?: boolean;
+    ReturnCode?: number;
+    Reason?: string;
+    SessionPresent?: boolean;
+}
+
+/**
+ * Client is an MQTT client for testing MQTT broker misconfigurations and authentication.
+ * @example
+ * ```javascript
+ * const mqtt = require('nuclei/mqtt');
+ * const client = new mqtt.Client();
+ * const res = client.Connect('acme.com', 1883);
+ * if (res.Success) {
+ *     const msgs = client.Subscribe('#', 2);
+ *     client.Publish('nuclei/probe', 'test');
+ * }
+ * client.Close();
+ * ```
+ */
+export class Client {
+    constructor() {}
+
+    /**
+     * Connect connects to an MQTT broker using optional credentials and settings.
+     * @example
+     * ```javascript
+     * const mqtt = require('nuclei/mqtt');
+     * const client = new mqtt.Client();
+     * const res = client.Connect('acme.com', 1883);
+     * log(res.Success);
+     * ```
+     */
+    public Connect(host: string, port: number, options?: ConnectOptions): ConnectResult | null {
+        return null;
+    }
+
+    /**
+     * ConnectAuth connects to an MQTT broker with a username and password.
+     * @example
+     * ```javascript
+     * const mqtt = require('nuclei/mqtt');
+     * const client = new mqtt.Client();
+     * const res = client.ConnectAuth('acme.com', 1883, 'admin', 'admin');
+     * log(res.Success);
+     * ```
+     */
+    public ConnectAuth(host: string, port: number, username: string, password: string): ConnectResult | null {
+        return null;
+    }
+
+    /**
+     * Subscribe subscribes to a topic filter and collects received messages up to timeoutSec.
+     * @example
+     * ```javascript
+     * const mqtt = require('nuclei/mqtt');
+     * const client = new mqtt.Client();
+     * client.Connect('acme.com', 1883);
+     * const msgs = client.Subscribe('#', 3);
+     * for (const m of msgs) {
+     *     log(m.Topic, m.Payload);
+     * }
+     * ```
+     */
+    public Subscribe(topic: string, timeoutSec?: number): Message[] | null {
+        return null;
+    }
+
+    /**
+     * Publish publishes a message payload to a topic.
+     * @example
+     * ```javascript
+     * const mqtt = require('nuclei/mqtt');
+     * const client = new mqtt.Client();
+     * client.Connect('acme.com', 1883);
+     * const ok = client.Publish('nuclei/test', 'probe payload');
+     * log(ok);
+     * ```
+     */
+    public Publish(topic: string, payload: string): boolean | null {
+        return null;
+    }
+
+    /**
+     * Ping sends an MQTT PINGREQ and awaits PINGRESP.
+     * @example
+     * ```javascript
+     * const mqtt = require('nuclei/mqtt');
+     * const client = new mqtt.Client();
+     * client.Connect('acme.com', 1883);
+     * const alive = client.Ping();
+     * log(alive);
+     * ```
+     */
+    public Ping(): boolean | null {
+        return null;
+    }
+
+    /**
+     * Close cleanly disconnects and closes the connection.
+     * @example
+     * ```javascript
+     * const mqtt = require('nuclei/mqtt');
+     * const client = new mqtt.Client();
+     * client.Connect('acme.com', 1883);
+     * client.Close();
+     * ```
+     */
+    public Close(): error | null {
+        return null;
+    }
+}

pkg/js/libs/mqtt/mqtt.go

@@ -0,0 +1,596 @@
+package mqtt
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"net"
+	"strconv"
+	"sync"
+	"time"
+
+	libnet "github.com/projectdiscovery/nuclei/v3/pkg/js/libs/net"
+)
+
+var (
+	defaultTimeout = 10 * time.Second
+)
+
+const (
+	maxSubscribedMessages = 1000
+	maxSubscribedBytes    = 10 * 1024 * 1024 // 10MB
+)
+
+type (
+	// Message represents an MQTT published message.
+	// @example
+	// ```javascript
+	// const mqtt = require('nuclei/mqtt');
+	// const client = new mqtt.Client();
+	// client.Connect('broker.hivemq.com', 1883);
+	// const msgs = client.Subscribe('test/topic', 2);
+	// log(msgs[0].Topic, msgs[0].Payload);
+	// ```
+	Message struct {
+		Topic   string `json:"topic"`
+		Payload string `json:"payload"`
+		QoS     int    `json:"qos"`
+		Retain  bool   `json:"retain"`
+	}
+
+	// ConnectOptions contains connection options for an MQTT broker.
+	// @example
+	// ```javascript
+	// const mqtt = require('nuclei/mqtt');
+	// const opts = new mqtt.ConnectOptions();
+	// opts.Username = 'admin';
+	// opts.Password = 'admin';
+	// ```
+	ConnectOptions struct {
+		Username           string `json:"username,omitempty"`
+		Password           string `json:"password,omitempty"`
+		ClientID           string `json:"client_id,omitempty"`
+		CleanSession       *bool  `json:"clean_session,omitempty"`
+		KeepAlive          int    `json:"keep_alive,omitempty"`
+		TLS                bool   `json:"tls,omitempty"`
+		InsecureSkipVerify *bool  `json:"insecure_skip_verify,omitempty"`
+		Timeout            int    `json:"timeout,omitempty"`
+	}
+
+	// ConnectResult is the response returned by an MQTT connection attempt.
+	// @example
+	// ```javascript
+	// const mqtt = require('nuclei/mqtt');
+	// const client = new mqtt.Client();
+	// const res = client.Connect('acme.com', 1883);
+	// log(res.Success, res.ReturnCode, res.Reason);
+	// ```
+	ConnectResult struct {
+		Success        bool   `json:"success"`
+		ReturnCode     int    `json:"return_code"`
+		Reason         string `json:"reason"`
+		SessionPresent bool   `json:"session_present"`
+	}
+
+	// Client is an MQTT client for testing MQTT broker misconfigurations and authentication.
+	// @example
+	// ```javascript
+	// const mqtt = require('nuclei/mqtt');
+	// const client = new mqtt.Client();
+	// const res = client.Connect('acme.com', 1883);
+	// if (res.Success) {
+	//     const msgs = client.Subscribe('#', 2);
+	//     client.Publish('nuclei/probe', 'test');
+	// }
+	// client.Close();
+	// ```
+	Client struct {
+		conn     net.Conn
+		packetID uint16
+		mu       sync.Mutex
+		timeout  time.Duration
+	}
+)
+
+// dial opens a network connection to host:port respecting protocolstate policy and context.
+func dial(ctx context.Context, host string, port int, useTLS bool, insecureSkipVerify bool, timeout time.Duration) (net.Conn, error) {
+	if host == "" || port <= 0 {
+		return nil, fmt.Errorf("invalid host or port")
+	}
+
+	addr := net.JoinHostPort(host, strconv.Itoa(port))
+	conn, err := libnet.Dial(ctx, "tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	_ = conn.SetDeadline(time.Now().Add(timeout))
+
+	if useTLS {
+		tlsConn := tls.Client(conn, &tls.Config{
+			InsecureSkipVerify: insecureSkipVerify,
+			MinVersion:         tls.VersionTLS12,
+			ServerName:         host,
+		})
+		if err := tlsConn.HandshakeContext(ctx); err != nil {
+			_ = conn.Close()
+			return nil, fmt.Errorf("TLS handshake failed: %w", err)
+		}
+		conn = tlsConn
+	}
+
+	_ = conn.SetDeadline(time.Now().Add(timeout))
+	return conn, nil
+}
+
+// IsMQTT checks whether the target host:port is running an MQTT broker.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const isMqtt = mqtt.IsMQTT('acme.com', 1883);
+// log(isMqtt);
+// ```
+func IsMQTT(ctx context.Context, host string, port int) (bool, error) {
+	conn, err := dial(ctx, host, port, false, true, defaultTimeout)
+	if err != nil {
+		return false, err
+	}
+	defer func() { _ = conn.Close() }()
+
+	pkt, err := buildConnectPacket("nuclei-probe", "", "", true, 60)
+	if err != nil {
+		return false, err
+	}
+	_ = conn.SetWriteDeadline(time.Now().Add(defaultTimeout))
+	if _, err := conn.Write(pkt); err != nil {
+		return false, err
+	}
+
+	header, payload, err := readPacket(conn)
+	if err != nil {
+		return false, nil
+	}
+
+	// Any valid CONNACK indicates an active MQTT broker
+	if (header >> 4) == PacketConnack {
+		_, _, err := parseConnack(payload)
+		return err == nil, nil
+	}
+	return false, nil
+}
+
+// IsAnonymousAllowed checks whether an MQTT broker permits unauthenticated / anonymous login.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const allowed = mqtt.IsAnonymousAllowed('acme.com', 1883);
+// log(allowed);
+// ```
+func IsAnonymousAllowed(ctx context.Context, host string, port int) (bool, error) {
+	conn, err := dial(ctx, host, port, false, true, defaultTimeout)
+	if err != nil {
+		return false, err
+	}
+	defer func() { _ = conn.Close() }()
+
+	pkt, err := buildConnectPacket("", "", "", true, 60)
+	if err != nil {
+		return false, err
+	}
+	_ = conn.SetWriteDeadline(time.Now().Add(defaultTimeout))
+	if _, err := conn.Write(pkt); err != nil {
+		return false, err
+	}
+
+	header, payload, err := readPacket(conn)
+	if err != nil {
+		return false, err
+	}
+
+	if (header >> 4) != PacketConnack {
+		return false, fmt.Errorf("expected CONNACK, got packet type %d", header>>4)
+	}
+
+	_, returnCode, err := parseConnack(payload)
+	if err != nil {
+		return false, err
+	}
+
+	return returnCode == ConnAccepted, nil
+}
+
+// Connect connects to an MQTT broker using optional credentials and settings.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const client = new mqtt.Client();
+// const res = client.Connect('acme.com', 1883);
+// log(res.Success);
+// ```
+func (c *Client) Connect(ctx context.Context, host string, port int, options *ConnectOptions) (*ConnectResult, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.conn != nil {
+		_ = c.conn.Close()
+		c.conn = nil
+	}
+
+	timeout := defaultTimeout
+	cleanSession := true
+	keepAlive := uint16(60)
+	clientID := "nuclei"
+	var username, password string
+	useTLS := false
+	insecureSkipVerify := true
+
+	if options != nil {
+		if options.Timeout > 0 {
+			timeout = time.Duration(options.Timeout) * time.Second
+		}
+		if options.KeepAlive > 0 {
+			if options.KeepAlive > 65535 {
+				keepAlive = 65535
+			} else {
+				keepAlive = uint16(options.KeepAlive)
+			}
+		}
+		if options.ClientID != "" {
+			clientID = options.ClientID
+		}
+		if options.CleanSession != nil {
+			cleanSession = *options.CleanSession
+		}
+		if options.InsecureSkipVerify != nil {
+			insecureSkipVerify = *options.InsecureSkipVerify
+		}
+		username = options.Username
+		password = options.Password
+		useTLS = options.TLS
+	}
+
+	c.timeout = timeout
+	conn, err := dial(ctx, host, port, useTLS, insecureSkipVerify, timeout)
+	if err != nil {
+		return &ConnectResult{
+			Success:    false,
+			ReturnCode: -1,
+			Reason:     err.Error(),
+		}, err
+	}
+
+	pkt, err := buildConnectPacket(clientID, username, password, cleanSession, keepAlive)
+	if err != nil {
+		_ = conn.Close()
+		return &ConnectResult{
+			Success:    false,
+			ReturnCode: -1,
+			Reason:     err.Error(),
+		}, err
+	}
+
+	_ = conn.SetWriteDeadline(time.Now().Add(timeout))
+	if _, err := conn.Write(pkt); err != nil {
+		_ = conn.Close()
+		return &ConnectResult{
+			Success:    false,
+			ReturnCode: -1,
+			Reason:     fmt.Sprintf("failed to send CONNECT: %v", err),
+		}, err
+	}
+
+	header, payload, err := readPacket(conn)
+	if err != nil {
+		_ = conn.Close()
+		return &ConnectResult{
+			Success:    false,
+			ReturnCode: -1,
+			Reason:     fmt.Sprintf("failed to read CONNACK: %v", err),
+		}, err
+	}
+
+	if (header >> 4) != PacketConnack {
+		_ = conn.Close()
+		return &ConnectResult{
+			Success:    false,
+			ReturnCode: -1,
+			Reason:     fmt.Sprintf("unexpected packet %d (expected CONNACK)", header>>4),
+		}, fmt.Errorf("unexpected packet %d", header>>4)
+	}
+
+	sessPresent, returnCode, err := parseConnack(payload)
+	if err != nil {
+		_ = conn.Close()
+		return &ConnectResult{
+			Success:    false,
+			ReturnCode: int(returnCode),
+			Reason:     err.Error(),
+		}, err
+	}
+
+	res := &ConnectResult{
+		Success:        returnCode == ConnAccepted,
+		ReturnCode:     int(returnCode),
+		Reason:         ConnackReason(returnCode),
+		SessionPresent: sessPresent,
+	}
+
+	if res.Success {
+		c.conn = conn
+	} else {
+		_ = conn.Close()
+	}
+
+	return res, nil
+}
+
+// ConnectAuth connects to an MQTT broker with a username and password.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const client = new mqtt.Client();
+// const res = client.ConnectAuth('acme.com', 1883, 'admin', 'admin');
+// log(res.Success);
+// ```
+func (c *Client) ConnectAuth(ctx context.Context, host string, port int, username, password string) (*ConnectResult, error) {
+	clean := true
+	return c.Connect(ctx, host, port, &ConnectOptions{
+		Username:     username,
+		Password:     password,
+		CleanSession: &clean,
+	})
+}
+
+// Subscribe subscribes to a topic filter and collects received messages up to timeoutSec.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const client = new mqtt.Client();
+// client.Connect('acme.com', 1883);
+// const msgs = client.Subscribe('#', 3);
+//
+//	for (const m of msgs) {
+//	    log(m.Topic, m.Payload);
+//	}
+//
+// ```
+func (c *Client) Subscribe(ctx context.Context, topic string, timeoutSec int) ([]*Message, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.conn == nil {
+		return nil, fmt.Errorf("not connected to MQTT broker")
+	}
+
+	c.packetID++
+	if c.packetID == 0 {
+		c.packetID = 1
+	}
+	pktID := c.packetID
+
+	subPkt, err := buildSubscribePacket(pktID, topic, 0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build SUBSCRIBE: %w", err)
+	}
+
+	_ = c.conn.SetWriteDeadline(time.Now().Add(c.timeout))
+	if _, err := c.conn.Write(subPkt); err != nil {
+		_ = c.conn.Close()
+		c.conn = nil
+		return nil, fmt.Errorf("failed to send SUBSCRIBE: %w", err)
+	}
+
+	waitDuration := 2 * time.Second
+	if timeoutSec > 0 {
+		if timeoutSec > 60 {
+			timeoutSec = 60
+		}
+		waitDuration = time.Duration(timeoutSec) * time.Second
+	}
+	if deadline, ok := ctx.Deadline(); ok {
+		if remaining := time.Until(deadline); remaining < waitDuration {
+			waitDuration = remaining
+		}
+	}
+
+	readDeadline := time.Now().Add(waitDuration)
+	_ = c.conn.SetDeadline(readDeadline)
+	defer func() {
+		if c.conn != nil {
+			_ = c.conn.SetDeadline(time.Time{})
+		}
+	}()
+
+	var messages []*Message
+	var totalBytes int
+	subackReceived := false
+
+	for {
+		select {
+		case <-ctx.Done():
+			_ = c.conn.Close()
+			c.conn = nil
+			return messages, ctx.Err()
+		default:
+		}
+
+		header, payload, err := readPacket(c.conn)
+		if err != nil {
+			if errors.Is(err, errBrokenStream) {
+				// The window closed in the middle of a packet. The stream cannot be
+				// read in sync again, so the connection is dropped.
+				_ = c.conn.Close()
+				c.conn = nil
+				if !subackReceived {
+					return nil, fmt.Errorf("failed waiting for SUBACK: %w", err)
+				}
+				break
+			}
+			var netErr net.Error
+			if errors.As(err, &netErr) && netErr.Timeout() {
+				// Window finished
+				break
+			}
+			if !subackReceived {
+				_ = c.conn.Close()
+				c.conn = nil
+				return nil, fmt.Errorf("failed waiting for SUBACK: %w", err)
+			}
+			break
+		}
+
+		packetType := header >> 4
+		switch packetType {
+		case PacketSuback:
+			subackPID, grantedQoS, err := parseSuback(payload)
+			if err != nil {
+				_ = c.conn.Close()
+				c.conn = nil
+				return nil, err
+			}
+			if subackPID != pktID {
+				continue // Ignore stale/mismatched SUBACK from an earlier subscription
+			}
+			for _, qos := range grantedQoS {
+				if qos == 0x80 {
+					return nil, fmt.Errorf("subscription rejected by broker for topic %q", topic)
+				}
+			}
+			subackReceived = true
+
+		case PacketPublish:
+			msg, err := parsePublish(header, payload)
+			if err == nil {
+				messages = append(messages, msg)
+				totalBytes += len(msg.Topic) + len(msg.Payload)
+				if len(messages) >= maxSubscribedMessages || totalBytes >= maxSubscribedBytes {
+					return messages, nil
+				}
+			}
+		}
+
+		if time.Now().After(readDeadline) {
+			break
+		}
+	}
+
+	if !subackReceived {
+		_ = c.conn.Close()
+		c.conn = nil
+		return nil, fmt.Errorf("subscription timed out waiting for SUBACK")
+	}
+
+	return messages, nil
+}
+
+// Publish publishes a message payload to a topic.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const client = new mqtt.Client();
+// client.Connect('acme.com', 1883);
+// const ok = client.Publish('nuclei/test', 'probe payload');
+// log(ok);
+// ```
+func (c *Client) Publish(ctx context.Context, topic string, payload string) (bool, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.conn == nil {
+		return false, fmt.Errorf("not connected to MQTT broker")
+	}
+
+	select {
+	case <-ctx.Done():
+		return false, ctx.Err()
+	default:
+	}
+
+	pkt, err := buildPublishPacket(topic, []byte(payload), 0, false, 0)
+	if err != nil {
+		return false, fmt.Errorf("failed to build PUBLISH: %w", err)
+	}
+
+	_ = c.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
+	if _, err := c.conn.Write(pkt); err != nil {
+		_ = c.conn.Close()
+		c.conn = nil
+		return false, fmt.Errorf("failed to send PUBLISH: %w", err)
+	}
+
+	return true, nil
+}
+
+// Ping sends an MQTT PINGREQ and awaits PINGRESP.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const client = new mqtt.Client();
+// client.Connect('acme.com', 1883);
+// const alive = client.Ping();
+// log(alive);
+// ```
+func (c *Client) Ping(ctx context.Context) (bool, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.conn == nil {
+		return false, fmt.Errorf("not connected to MQTT broker")
+	}
+
+	select {
+	case <-ctx.Done():
+		return false, ctx.Err()
+	default:
+	}
+
+	_ = c.conn.SetDeadline(time.Now().Add(5 * time.Second))
+	_ = c.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
+	if _, err := c.conn.Write(buildPingreqPacket()); err != nil {
+		_ = c.conn.Close()
+		c.conn = nil
+		return false, fmt.Errorf("failed to send PINGREQ: %w", err)
+	}
+
+	for {
+		header, _, err := readPacket(c.conn)
+		if err != nil {
+			_ = c.conn.Close()
+			c.conn = nil
+			return false, fmt.Errorf("failed to read PINGRESP: %w", err)
+		}
+
+		packetType := header >> 4
+		if packetType == PacketPingresp {
+			return true, nil
+		}
+		if packetType == PacketPublish {
+			// Broker delivered an incoming publish message while ping was in flight; continue waiting for PINGRESP
+			continue
+		}
+		return false, fmt.Errorf("unexpected ping response packet: %d", packetType)
+	}
+}
+
+// Close cleanly disconnects and closes the connection.
+// @example
+// ```javascript
+// const mqtt = require('nuclei/mqtt');
+// const client = new mqtt.Client();
+// client.Connect('acme.com', 1883);
+// client.Close();
+// ```
+func (c *Client) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.conn != nil {
+		_ = c.conn.SetWriteDeadline(time.Now().Add(3 * time.Second))
+		_, _ = c.conn.Write(buildDisconnectPacket())
+		err := c.conn.Close()
+		c.conn = nil
+		return err
+	}
+	return nil
+}

pkg/js/libs/mqtt/mqtt_test.go

@@ -0,0 +1,395 @@
+package mqtt
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"net"
+	"strings"
+	"testing"
+
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/protocolstate"
+	"github.com/projectdiscovery/nuclei/v3/pkg/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestVarLengthEncoding(t *testing.T) {
+	testCases := []int{0, 1, 64, 127, 128, 16383, 16384, 2097151}
+	for _, tc := range testCases {
+		encoded, err := encodeVarLength(tc)
+		require.NoError(t, err)
+		decoded, err := decodeVarLength(bytes.NewReader(encoded))
+		require.NoError(t, err)
+		require.Equal(t, tc, decoded)
+	}
+
+	// Test negative or overflow
+	_, err := encodeVarLength(-1)
+	require.Error(t, err)
+	_, err = encodeVarLength(MaxRemainingLength + 1)
+	require.Error(t, err)
+}
+
+func TestOversizedStringsAndPackets(t *testing.T) {
+	hugeString := strings.Repeat("A", MaxStringLength+1)
+	_, err := encodeString(hugeString)
+	require.Error(t, err)
+
+	_, err = buildConnectPacket("id", hugeString, "pass", true, 60)
+	require.Error(t, err)
+
+	_, err = buildSubscribePacket(1, hugeString, 0)
+	require.Error(t, err)
+
+	_, err = buildPublishPacket(hugeString, []byte("ok"), 0, false, 0)
+	require.Error(t, err)
+
+	// Oversized packet reading
+	oversizedLen, err := encodeVarLength(MaxPacketSize + 100)
+	require.NoError(t, err)
+	buf := append([]byte{PacketPublish << 4}, oversizedLen...)
+	_, _, err = readPacket(bytes.NewReader(buf))
+	require.ErrorContains(t, err, "exceeds maximum allowed")
+}
+
+func TestPacketConnect(t *testing.T) {
+	pkt, err := buildConnectPacket("test-client", "user1", "pass1", true, 60)
+	require.NoError(t, err)
+	require.NotEmpty(t, pkt)
+	require.Equal(t, byte(PacketConnect<<4), pkt[0])
+
+	r := bytes.NewReader(pkt)
+	header, payload, err := readPacket(r)
+	require.NoError(t, err)
+	require.Equal(t, byte(PacketConnect<<4), header)
+
+	// Verify protocol name "MQTT"
+	protoLen := binary.BigEndian.Uint16(payload[0:2])
+	require.Equal(t, uint16(4), protoLen)
+	require.Equal(t, "MQTT", string(payload[2:6]))
+	require.Equal(t, byte(0x04), payload[6]) // Protocol level 4 (MQTT 3.1.1)
+
+	// CleanSession (0x02) | User (0x80) | Pass (0x40) = 0xC2
+	require.Equal(t, byte(0xC2), payload[7])
+}
+
+func TestPacketPublishAndParse(t *testing.T) {
+	pkt, err := buildPublishPacket("sensors/temperature", []byte("23.5C"), 0, true, 0)
+	require.NoError(t, err)
+	require.NotEmpty(t, pkt)
+
+	r := bytes.NewReader(pkt)
+	header, payload, err := readPacket(r)
+	require.NoError(t, err)
+
+	msg, err := parsePublish(header, payload)
+	require.NoError(t, err)
+	require.Equal(t, "sensors/temperature", msg.Topic)
+	require.Equal(t, "23.5C", msg.Payload)
+	require.Equal(t, 0, msg.QoS)
+	require.True(t, msg.Retain)
+}
+
+func TestPacketSubscribe(t *testing.T) {
+	pkt, err := buildSubscribePacket(42, "devices/#", 0)
+	require.NoError(t, err)
+	require.NotEmpty(t, pkt)
+	require.Equal(t, byte((PacketSubscribe<<4)|0x02), pkt[0])
+
+	r := bytes.NewReader(pkt)
+	header, payload, err := readPacket(r)
+	require.NoError(t, err)
+	require.Equal(t, byte((PacketSubscribe<<4)|0x02), header)
+
+	pid := binary.BigEndian.Uint16(payload[0:2])
+	require.Equal(t, uint16(42), pid)
+
+	subackPayload := []byte{0x00, 0x2A, 0x00} // PID 42, Success QoS 0
+	subackPid, granted, err := parseSuback(subackPayload)
+	require.NoError(t, err)
+	require.Equal(t, uint16(42), subackPid)
+	require.Equal(t, []byte{0x00}, granted)
+}
+
+func TestConnackReasons(t *testing.T) {
+	require.Equal(t, "connection accepted", ConnackReason(ConnAccepted))
+	require.Equal(t, "connection refused: bad user name or password", ConnackReason(ConnRefusedBadAuth))
+	require.Equal(t, "connection refused: not authorized", ConnackReason(ConnRefusedNotAuthorized))
+	require.Contains(t, ConnackReason(99), "connection refused: return code 99")
+}
+
+func TestRequestsHonourNetworkPolicy(t *testing.T) {
+	execID := "mqtt-deny"
+	require.NoError(t, protocolstate.Init(&types.Options{ExecutionId: execID, ExcludeTargets: []string{"203.0.113.50"}}))
+	t.Cleanup(func() { protocolstate.Close(execID) })
+	ctx := context.WithValue(context.Background(), "executionId", execID) //nolint:staticcheck
+
+	_, err := IsMQTT(ctx, "203.0.113.50", 1883)
+	require.Error(t, err)
+
+	client := &Client{}
+	_, err = client.Connect(ctx, "203.0.113.50", 1883, nil)
+	require.Error(t, err)
+}
+
+func startMockMQTTBroker(t *testing.T) (net.Listener, int) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	port := l.Addr().(*net.TCPAddr).Port
+
+	go func() {
+		for {
+			conn, err := l.Accept()
+			if err != nil {
+				return
+			}
+			go handleMockClient(conn)
+		}
+	}()
+
+	return l, port
+}
+
+func handleMockClient(conn net.Conn) {
+	defer func() { _ = conn.Close() }()
+
+	for {
+		header, payload, err := readPacket(conn)
+		if err != nil {
+			return
+		}
+
+		packetType := header >> 4
+		switch packetType {
+		case PacketConnect:
+			// Check if username is present (flags byte at offset 7)
+			if len(payload) >= 8 {
+				flags := payload[7]
+				if (flags & 0x80) != 0 {
+					// Check username
+					offset := 10 // after proto(6) + level(1) + flags(1) + keepalive(2)
+					if len(payload) >= offset+2 {
+						clientLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
+						offset += 2 + clientLen
+						if len(payload) >= offset+2 {
+							userLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
+							user := string(payload[offset+2 : offset+2+userLen])
+							if user == "baduser" {
+								// Send CONNACK Refused Bad Auth
+								_, _ = conn.Write([]byte{0x20, 0x02, 0x00, ConnRefusedBadAuth})
+								return
+							}
+						}
+					}
+				}
+			}
+			// Send CONNACK Accepted
+			_, _ = conn.Write([]byte{0x20, 0x02, 0x00, ConnAccepted})
+
+		case PacketSubscribe:
+			// Send SUBACK
+			pid := binary.BigEndian.Uint16(payload[0:2])
+			suback := []byte{0x90, 0x03, byte(pid >> 8), byte(pid & 0xFF), 0x00}
+			_, _ = conn.Write(suback)
+
+			// Publish test message
+			pubPkt, err := buildPublishPacket("nuclei/alert", []byte("vuln-detected"), 0, false, 0)
+			if err == nil {
+				_, _ = conn.Write(pubPkt)
+			}
+
+		case PacketPublish:
+			// QoS 0, no ack needed
+
+		case PacketPingreq:
+			// Send PINGRESP
+			_, _ = conn.Write([]byte{PacketPingresp << 4, 0x00})
+
+		case PacketDisconnect:
+			return
+		}
+	}
+}
+
+func TestMockBrokerInteractions(t *testing.T) {
+	execID := "mqtt-mock-test"
+	require.NoError(t, protocolstate.Init(&types.Options{ExecutionId: execID}))
+	t.Cleanup(func() { protocolstate.Close(execID) })
+	ctx := context.WithValue(context.Background(), "executionId", execID) //nolint:staticcheck
+
+	l, port := startMockMQTTBroker(t)
+	defer func() { _ = l.Close() }()
+
+	// 1. Test IsMQTT
+	isMQTT, err := IsMQTT(ctx, "127.0.0.1", port)
+	require.NoError(t, err)
+	require.True(t, isMQTT)
+
+	// 2. Test IsAnonymousAllowed
+	anonAllowed, err := IsAnonymousAllowed(ctx, "127.0.0.1", port)
+	require.NoError(t, err)
+	require.True(t, anonAllowed)
+
+	// 3. Test Client Connect Auth Failure
+	badClient := &Client{}
+	badRes, _ := badClient.ConnectAuth(ctx, "127.0.0.1", port, "baduser", "wrongpass")
+	require.False(t, badRes.Success)
+	require.Equal(t, int(ConnRefusedBadAuth), badRes.ReturnCode)
+
+	// 4. Test Client Connect Success with default CleanSession preservation
+	client := &Client{}
+	res, err := client.Connect(ctx, "127.0.0.1", port, &ConnectOptions{KeepAlive: 70000})
+	require.NoError(t, err)
+	require.True(t, res.Success)
+	require.Equal(t, int(ConnAccepted), res.ReturnCode)
+
+	// 5. Test Ping
+	alive, err := client.Ping(ctx)
+	require.NoError(t, err)
+	require.True(t, alive)
+
+	// 6. Test Subscribe and Receive Message
+	msgs, err := client.Subscribe(ctx, "nuclei/alert", 1)
+	require.NoError(t, err)
+	require.NotEmpty(t, msgs)
+	require.Equal(t, "nuclei/alert", msgs[0].Topic)
+	require.Equal(t, "vuln-detected", msgs[0].Payload)
+
+	// 7. Test Publish
+	pubOk, err := client.Publish(ctx, "nuclei/probe", "test payload")
+	require.NoError(t, err)
+	require.True(t, pubOk)
+
+	// 8. Test Close and Double Close
+	require.NoError(t, client.Close())
+	require.NoError(t, client.Close())
+}
+
+func TestPasswordWithoutUsername(t *testing.T) {
+	// MQTT 3.1.1 (MQTT-3.1.2-22): If Password flag is set, User Name flag MUST be set.
+	pkt, err := buildConnectPacket("token-client", "", "my-secret-token", true, 60)
+	require.NoError(t, err)
+
+	r := bytes.NewReader(pkt)
+	_, payload, err := readPacket(r)
+	require.NoError(t, err)
+
+	// CleanSession (0x02) | User (0x80) | Pass (0x40) = 0xC2
+	require.Equal(t, byte(0xC2), payload[7])
+
+	// The client ID comes right after the keep alive at payload[8:10], and the
+	// username length follows it.
+	clientIDLen := int(binary.BigEndian.Uint16(payload[10:12]))
+	off := 12 + clientIDLen
+
+	// Username length should be 0 (empty string)
+	userLen := binary.BigEndian.Uint16(payload[off : off+2])
+	require.Equal(t, uint16(0), userLen)
+	off += 2
+
+	// Password follows the username
+	passLen := binary.BigEndian.Uint16(payload[off : off+2])
+	require.Equal(t, uint16(len("my-secret-token")), passLen)
+	off += 2
+	require.Equal(t, "my-secret-token", string(payload[off:off+int(passLen)]))
+}
+
+// A window that closes in the middle of a PUBLISH must drop the connection.
+// Keeping it would let the next read start inside that packet.
+func TestSubscribePartialPacketClosesConnection(t *testing.T) {
+	execID := "mqtt-partial-subscribe"
+	require.NoError(t, protocolstate.Init(&types.Options{ExecutionId: execID}))
+	t.Cleanup(func() { protocolstate.Close(execID) })
+	ctx := context.WithValue(context.Background(), "executionId", execID) //nolint:staticcheck
+
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer func() { _ = l.Close() }()
+	port := l.Addr().(*net.TCPAddr).Port
+
+	go func() {
+		conn, err := l.Accept()
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+		for {
+			header, payload, err := readPacket(conn)
+			if err != nil {
+				return
+			}
+			switch header >> 4 {
+			case PacketConnect:
+				_, _ = conn.Write([]byte{PacketConnack << 4, 0x02, 0x00, ConnAccepted})
+			case PacketSubscribe:
+				// SUBACK for the packet id, then a PUBLISH that promises 10 bytes
+				// and sends 5. The client's window closes in the middle of it.
+				_, _ = conn.Write([]byte{PacketSuback << 4, 0x03, payload[0], payload[1], 0x00})
+				_, _ = conn.Write([]byte{PacketPublish << 4, 0x0a, 0x00, 0x03, 'a', '/', 'b'})
+			}
+		}
+	}()
+
+	client := &Client{}
+	res, err := client.Connect(ctx, "127.0.0.1", port, nil)
+	require.NoError(t, err)
+	require.True(t, res.Success)
+
+	msgs, err := client.Subscribe(ctx, "a/b", 1)
+	require.NoError(t, err)
+	require.Empty(t, msgs)
+	require.Nil(t, client.conn, "a stream cut in the middle of a packet must not be reused")
+}
+
+func TestPingWithInFlightPublish(t *testing.T) {
+	execID := "mqtt-ping-inflight"
+	require.NoError(t, protocolstate.Init(&types.Options{ExecutionId: execID}))
+	t.Cleanup(func() { protocolstate.Close(execID) })
+	ctx := context.WithValue(context.Background(), "executionId", execID) //nolint:staticcheck
+
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer func() { _ = l.Close() }()
+
+	port := l.Addr().(*net.TCPAddr).Port
+
+	go func() {
+		conn, err := l.Accept()
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+
+		for {
+			header, _, err := readPacket(conn)
+			if err != nil {
+				return
+			}
+			packetType := header >> 4
+			switch packetType {
+			case PacketConnect:
+				_, _ = conn.Write([]byte{0x20, 0x02, 0x00, ConnAccepted})
+			case PacketPingreq:
+				// Send a PUBLISH packet first before PINGRESP
+				pubPkt, _ := buildPublishPacket("sensors/temp", []byte("25.0"), 0, false, 0)
+				_, _ = conn.Write(pubPkt)
+				// Then send PINGRESP
+				_, _ = conn.Write([]byte{PacketPingresp << 4, 0x00})
+			case PacketDisconnect:
+				return
+			}
+		}
+	}()
+
+	client := &Client{}
+	res, err := client.Connect(ctx, "127.0.0.1", port, nil)
+	require.NoError(t, err)
+	require.True(t, res.Success)
+
+	// Ping should skip the in-flight PUBLISH and successfully receive PINGRESP
+	alive, err := client.Ping(ctx)
+	require.NoError(t, err)
+	require.True(t, alive)
+
+	_ = client.Close()
+}

pkg/js/libs/mqtt/packet.go

@@ -0,0 +1,339 @@
+package mqtt
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+)
+
+// Limits to prevent DoS and buffer overflows
+const (
+	MaxStringLength    = 65535
+	MaxRemainingLength = 268435455
+	MaxPacketSize      = 10 * 1024 * 1024 // 10MB limit on incoming packets
+)
+
+// MQTT 3.1.1 Control Packet Types
+const (
+	PacketConnect    byte = 1
+	PacketConnack    byte = 2
+	PacketPublish    byte = 3
+	PacketPuback     byte = 4
+	PacketPubrec     byte = 5
+	PacketPubrel     byte = 6
+	PacketPubcomp    byte = 7
+	PacketSubscribe  byte = 8
+	PacketSuback     byte = 9
+	PacketUnsub      byte = 10
+	PacketUnsuback   byte = 11
+	PacketPingreq    byte = 12
+	PacketPingresp   byte = 13
+	PacketDisconnect byte = 14
+)
+
+// CONNACK Return Codes (MQTT 3.1.1 table 3.1)
+const (
+	ConnAccepted                    byte = 0
+	ConnRefusedUnacceptableProtocol byte = 1
+	ConnRefusedIdentifierRejected   byte = 2
+	ConnRefusedServerUnavailable    byte = 3
+	ConnRefusedBadAuth              byte = 4
+	ConnRefusedNotAuthorized        byte = 5
+)
+
+// ConnackReason returns a human-readable description for CONNACK return codes.
+func ConnackReason(code byte) string {
+	switch code {
+	case ConnAccepted:
+		return "connection accepted"
+	case ConnRefusedUnacceptableProtocol:
+		return "connection refused: unacceptable protocol version"
+	case ConnRefusedIdentifierRejected:
+		return "connection refused: identifier rejected"
+	case ConnRefusedServerUnavailable:
+		return "connection refused: server unavailable"
+	case ConnRefusedBadAuth:
+		return "connection refused: bad user name or password"
+	case ConnRefusedNotAuthorized:
+		return "connection refused: not authorized"
+	default:
+		return fmt.Sprintf("connection refused: return code %d", code)
+	}
+}
+
+// encodeVarLength encodes a length integer as an MQTT variable byte integer.
+func encodeVarLength(length int) ([]byte, error) {
+	if length < 0 || length > MaxRemainingLength {
+		return nil, fmt.Errorf("remaining length %d exceeds MQTT max %d", length, MaxRemainingLength)
+	}
+	var out []byte
+	for {
+		encodedByte := byte(length % 128)
+		length /= 128
+		if length > 0 {
+			encodedByte |= 128
+		}
+		out = append(out, encodedByte)
+		if length == 0 {
+			break
+		}
+	}
+	return out, nil
+}
+
+// decodeVarLength decodes an MQTT variable byte integer from reader.
+func decodeVarLength(r io.Reader) (int, error) {
+	multiplier := 1
+	value := 0
+	buf := make([]byte, 1)
+	for i := 0; i < 4; i++ {
+		if _, err := io.ReadFull(r, buf); err != nil {
+			return 0, err
+		}
+		encodedByte := buf[0]
+		value += int(encodedByte&127) * multiplier
+		if multiplier > 128*128*128 {
+			return 0, errors.New("malformed remaining length")
+		}
+		multiplier *= 128
+		if encodedByte&128 == 0 {
+			return value, nil
+		}
+	}
+	return 0, errors.New("malformed remaining length")
+}
+
+// encodeString encodes a UTF-8 string with a 2-byte big-endian length prefix.
+func encodeString(s string) ([]byte, error) {
+	b := []byte(s)
+	if len(b) > MaxStringLength {
+		return nil, fmt.Errorf("string length %d exceeds MQTT max %d", len(b), MaxStringLength)
+	}
+	out := make([]byte, 2+len(b))
+	binary.BigEndian.PutUint16(out[0:2], uint16(len(b)))
+	copy(out[2:], b)
+	return out, nil
+}
+
+// buildConnectPacket constructs an MQTT 3.1.1 CONNECT packet.
+func buildConnectPacket(clientID, username, password string, cleanSession bool, keepAlive uint16) ([]byte, error) {
+	var varHeader []byte
+	protoName, err := encodeString("MQTT")
+	if err != nil {
+		return nil, err
+	}
+	varHeader = append(varHeader, protoName...)
+	varHeader = append(varHeader, 0x04) // Protocol Level: 4 (MQTT 3.1.1)
+
+	// Connect Flags
+	var flags byte
+	if cleanSession {
+		flags |= 0x02
+	}
+	// Per MQTT 3.1.1 spec (MQTT-3.1.2-22): If the User Name Flag is 0, the Password Flag MUST be 0.
+	// If a password is provided, the User Name Flag MUST be set (encoding an empty username if omitted).
+	hasUsername := username != "" || password != ""
+	if hasUsername {
+		flags |= 0x80
+	}
+	if password != "" {
+		flags |= 0x40
+	}
+	varHeader = append(varHeader, flags)
+
+	// Keep Alive (2 bytes big endian)
+	ka := make([]byte, 2)
+	binary.BigEndian.PutUint16(ka, keepAlive)
+	varHeader = append(varHeader, ka...)
+
+	// Payload
+	var payload []byte
+	cidBytes, err := encodeString(clientID)
+	if err != nil {
+		return nil, fmt.Errorf("invalid clientID: %w", err)
+	}
+	payload = append(payload, cidBytes...)
+
+	if hasUsername {
+		userBytes, err := encodeString(username)
+		if err != nil {
+			return nil, fmt.Errorf("invalid username: %w", err)
+		}
+		payload = append(payload, userBytes...)
+	}
+	if password != "" {
+		passBytes, err := encodeString(password)
+		if err != nil {
+			return nil, fmt.Errorf("invalid password: %w", err)
+		}
+		payload = append(payload, passBytes...)
+	}
+
+	remainingLength := len(varHeader) + len(payload)
+	lenBytes, err := encodeVarLength(remainingLength)
+	if err != nil {
+		return nil, err
+	}
+
+	var pkt []byte
+	pkt = append(pkt, PacketConnect<<4)
+	pkt = append(pkt, lenBytes...)
+	pkt = append(pkt, varHeader...)
+	pkt = append(pkt, payload...)
+	return pkt, nil
+}
+
+// buildSubscribePacket constructs an MQTT 3.1.1 SUBSCRIBE packet for a single topic.
+func buildSubscribePacket(packetID uint16, topic string, qos byte) ([]byte, error) {
+	var varHeader [2]byte
+	binary.BigEndian.PutUint16(varHeader[:], packetID)
+
+	topicBytes, err := encodeString(topic)
+	if err != nil {
+		return nil, fmt.Errorf("invalid topic: %w", err)
+	}
+
+	var payload []byte
+	payload = append(payload, topicBytes...)
+	payload = append(payload, qos&0x03)
+
+	remainingLength := len(varHeader) + len(payload)
+	lenBytes, err := encodeVarLength(remainingLength)
+	if err != nil {
+		return nil, err
+	}
+
+	var pkt []byte
+	// SUBSCRIBE fixed header flags must be 0010 (0x02)
+	pkt = append(pkt, (PacketSubscribe<<4)|0x02)
+	pkt = append(pkt, lenBytes...)
+	pkt = append(pkt, varHeader[:]...)
+	pkt = append(pkt, payload...)
+	return pkt, nil
+}
+
+// buildPublishPacket constructs an MQTT 3.1.1 PUBLISH packet.
+func buildPublishPacket(topic string, payload []byte, qos byte, retain bool, packetID uint16) ([]byte, error) {
+	headerByte := (PacketPublish << 4)
+	if retain {
+		headerByte |= 0x01
+	}
+	headerByte |= (qos & 0x03) << 1
+
+	topicBytes, err := encodeString(topic)
+	if err != nil {
+		return nil, fmt.Errorf("invalid topic: %w", err)
+	}
+
+	var varHeader []byte
+	varHeader = append(varHeader, topicBytes...)
+	if qos > 0 {
+		pid := make([]byte, 2)
+		binary.BigEndian.PutUint16(pid, packetID)
+		varHeader = append(varHeader, pid...)
+	}
+
+	remainingLength := len(varHeader) + len(payload)
+	lenBytes, err := encodeVarLength(remainingLength)
+	if err != nil {
+		return nil, err
+	}
+
+	var pkt []byte
+	pkt = append(pkt, headerByte)
+	pkt = append(pkt, lenBytes...)
+	pkt = append(pkt, varHeader...)
+	pkt = append(pkt, payload...)
+	return pkt, nil
+}
+
+// buildPingreqPacket constructs an MQTT PINGREQ packet.
+func buildPingreqPacket() []byte {
+	return []byte{PacketPingreq << 4, 0x00}
+}
+
+// buildDisconnectPacket constructs an MQTT DISCONNECT packet.
+func buildDisconnectPacket() []byte {
+	return []byte{PacketDisconnect << 4, 0x00}
+}
+
+// errBrokenStream means a read failed after part of a packet was consumed, or
+// a packet was refused, so the next read would start in the middle of a packet.
+// The connection must not be used again.
+var errBrokenStream = errors.New("MQTT stream is out of sync")
+
+// readPacket reads a full MQTT packet from reader, rejecting packets over MaxPacketSize.
+// Failures before the first byte is read are returned as they are. Failures after
+// it wrap errBrokenStream.
+func readPacket(r io.Reader) (headerByte byte, payload []byte, err error) {
+	var header [1]byte
+	if _, err := io.ReadFull(r, header[:]); err != nil {
+		return 0, nil, err
+	}
+	length, err := decodeVarLength(r)
+	if err != nil {
+		return 0, nil, fmt.Errorf("%w: %w", errBrokenStream, err)
+	}
+	if length > MaxPacketSize {
+		return 0, nil, fmt.Errorf("%w: MQTT packet length %d exceeds maximum allowed %d", errBrokenStream, length, MaxPacketSize)
+	}
+	buf := make([]byte, length)
+	if length > 0 {
+		if _, err := io.ReadFull(r, buf); err != nil {
+			return 0, nil, fmt.Errorf("%w: %w", errBrokenStream, err)
+		}
+	}
+	return header[0], buf, nil
+}
+
+// parseConnack decodes a CONNACK variable header.
+func parseConnack(payload []byte) (sessionPresent bool, returnCode byte, err error) {
+	if len(payload) < 2 {
+		return false, 0, errors.New("short CONNACK response")
+	}
+	sessionPresent = (payload[0] & 0x01) != 0
+	returnCode = payload[1]
+	return sessionPresent, returnCode, nil
+}
+
+// parseSuback decodes a SUBACK packet payload.
+func parseSuback(payload []byte) (packetID uint16, grantedQoS []byte, err error) {
+	if len(payload) < 3 {
+		return 0, nil, errors.New("short SUBACK response")
+	}
+	packetID = binary.BigEndian.Uint16(payload[0:2])
+	grantedQoS = payload[2:]
+	return packetID, grantedQoS, nil
+}
+
+// parsePublish decodes a PUBLISH packet.
+func parsePublish(headerByte byte, body []byte) (*Message, error) {
+	qos := (headerByte >> 1) & 0x03
+	retain := (headerByte & 0x01) != 0
+
+	if len(body) < 2 {
+		return nil, errors.New("short PUBLISH packet")
+	}
+	topicLen := int(binary.BigEndian.Uint16(body[0:2]))
+	if len(body) < 2+topicLen {
+		return nil, errors.New("short PUBLISH topic")
+	}
+	topic := string(body[2 : 2+topicLen])
+	offset := 2 + topicLen
+
+	if qos > 0 {
+		if len(body) < offset+2 {
+			return nil, errors.New("short PUBLISH packet ID")
+		}
+		offset += 2
+	}
+
+	payload := string(body[offset:])
+	return &Message{
+		Topic:   topic,
+		Payload: payload,
+		QoS:     int(qos),
+		Retain:  retain,
+	}, nil
+}

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/protocols/ssl/ssl.go

@@ -23,7 +23,6 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/generators"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/helpers/eventcreator"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/helpers/responsehighlighter"
-	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/protocolstate"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/render"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/utils/vardump"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/network/networkclientpool"
@@ -35,7 +34,6 @@ import (
 	"github.com/projectdiscovery/tlsx/pkg/tlsx/clients"
 	"github.com/projectdiscovery/tlsx/pkg/tlsx/openssl"
 	"github.com/projectdiscovery/utils/errkit"
-	iputil "github.com/projectdiscovery/utils/ip"
 	stringsutil "github.com/projectdiscovery/utils/strings"
 )
 
@@ -204,21 +202,6 @@ func (request *Request) GetID() string {
 	return request.ID
 }
 
-// allowedAddress returns the first address of host that the network policy allows.
-func (request *Request) allowedAddress(host string) (string, error) {
-	dnsData, err := request.dialer.GetDNSData(host)
-	if err != nil {
-		return "", err
-	}
-	addresses := append(append([]string{}, dnsData.A...), dnsData.AAAA...)
-	for _, address := range addresses {
-		if protocolstate.IsHostAllowed(request.options.Options.ExecutionId, address) {
-			return address, nil
-		}
-	}
-	return "", protocolstate.ErrHostDenied.Msgf(host)
-}
-
 // ExecuteWithResults executes the protocol requests and returns results instead of writing them.
 func (request *Request) ExecuteWithResults(input *contextargs.Context, dynamicValues, previous output.InternalEvent, callback protocols.OutputEventCallback) error {
 	hostPort := input.MetaInput.Input
@@ -272,23 +255,6 @@ func (request *Request) ExecuteWithResults(input *contextargs.Context, dynamicVa
 	} else {
 		hostIp = host
 	}
-	if !iputil.IsIP(hostIp) {
-		// The openssl scan mode runs the openssl binary, which resolves a host
-		// name again by itself, so it gets an address the policy allows.
-		address, err := request.allowedAddress(hostIp)
-		if err != nil {
-			requestOptions.Output.Request(requestOptions.TemplateID, input.MetaInput.Input, request.Type().String(), err)
-			requestOptions.Progress.IncrementFailedRequestsBy(1)
-			return errkit.Wrap(err, "could not resolve host")
-		}
-		hostIp = address
-	} else if !protocolstate.IsHostAllowed(request.options.Options.ExecutionId, hostIp) {
-		// A custom dialer can allow an address that the execution policy denies.
-		err := protocolstate.ErrHostDenied.Msgf(hostIp)
-		requestOptions.Output.Request(requestOptions.TemplateID, input.MetaInput.Input, request.Type().String(), err)
-		requestOptions.Progress.IncrementFailedRequestsBy(1)
-		return err
-	}
 
 	if err := request.options.HostBackoffWait(input.Context(), addressToDial); err != nil {
 		return err

pkg/protocols/ssl/ssl_openssl_test.go

@@ -1,179 +0,0 @@
-package ssl
-
-import (
-	"context"
-	"crypto/ecdsa"
-	"crypto/elliptic"
-	"crypto/rand"
-	"crypto/tls"
-	"crypto/x509"
-	"crypto/x509/pkix"
-	"math/big"
-	"net"
-	"slices"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/projectdiscovery/fastdialer/fastdialer"
-	"github.com/stretchr/testify/require"
-
-	"github.com/projectdiscovery/nuclei/v3/internal/tests/testutils"
-	"github.com/projectdiscovery/nuclei/v3/pkg/output"
-	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
-	"github.com/projectdiscovery/tlsx/pkg/tlsx/openssl"
-)
-
-func selfSignedCertificate(t *testing.T, name string) tls.Certificate {
-	t.Helper()
-	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
-	require.NoError(t, err)
-	template := &x509.Certificate{
-		SerialNumber: big.NewInt(1),
-		Subject:      pkix.Name{CommonName: name},
-		NotBefore:    time.Now().Add(-time.Hour),
-		NotAfter:     time.Now().Add(time.Hour),
-	}
-	der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
-	require.NoError(t, err)
-	return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
-}
-
-// serveTLS counts the TLS handshakes that reach address.
-func serveTLS(t *testing.T, address, name string, handshakes *atomic.Int32) string {
-	t.Helper()
-	listener, err := tls.Listen("tcp", address, &tls.Config{Certificates: []tls.Certificate{selfSignedCertificate(t, name)}})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = listener.Close() })
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				return
-			}
-			go func() {
-				defer func() { _ = conn.Close() }()
-				_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
-				if conn.(*tls.Conn).Handshake() == nil {
-					handshakes.Add(1)
-				}
-			}()
-		}
-	}()
-	return listener.Addr().String()
-}
-
-func TestSSLOpenSSLObeysNetworkPolicy(t *testing.T) {
-	if !openssl.IsAvailable() {
-		t.Skip("openssl binary is not installed")
-	}
-	addrs, err := net.LookupHost("localhost")
-	if err != nil || len(addrs) < 2 {
-		t.Skip("localhost does not resolve to both loopback addresses")
-	}
-
-	// The policy allows 127.0.0.1 and denies ::1, so localhost passes the
-	// dial check. The openssl binary must not connect to ::1.
-	var allowedHandshakes, deniedHandshakes atomic.Int32
-	allowed := serveTLS(t, "127.0.0.1:0", "allowed", &allowedHandshakes)
-	_, port, err := net.SplitHostPort(allowed)
-	require.NoError(t, err)
-	serveTLS(t, net.JoinHostPort("::1", port), "denied", &deniedHandshakes)
-
-	options := testutils.DefaultOptions.Copy()
-	options.SetExecutionID(t.Name())
-	options.ExcludeTargets = []string{"::1"}
-	testutils.Init(options)
-	t.Cleanup(func() { testutils.Cleanup(options) })
-	executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{ID: "ssl-openssl-policy"})
-	executerOpts.IsMultiProtocol = true
-
-	request := &Request{ID: "ssl-openssl-policy", Address: "{{Host}}:{{Port}}", ScanMode: "openssl"}
-	require.NoError(t, request.Compile(executerOpts))
-	// The scan resolves names with the fastdialer, not the system resolver. On
-	// Windows the hosts file has no localhost entry, so the name does not resolve.
-	if !resolvesToBothLoopbacks(request) {
-		t.Skip("the scan dialer does not resolve localhost to both loopback addresses")
-	}
-
-	var event output.InternalEvent
-	input := contextargs.NewWithInput(context.Background(), net.JoinHostPort("localhost", port))
-	err = request.ExecuteWithResults(input, nil, nil, func(e *output.InternalWrappedEvent) {
-		event = e.InternalEvent
-	})
-	require.NoError(t, err)
-	require.Equal(t, "allowed", event["subject_cn"])
-	require.Zero(t, deniedHandshakes.Load(), "openssl connected to a denied address")
-}
-
-func resolvesToBothLoopbacks(request *Request) bool {
-	data, err := request.dialer.GetDNSData("localhost")
-	if err != nil {
-		return false
-	}
-	return slices.Contains(data.A, "127.0.0.1") && slices.Contains(data.AAAA, "::1")
-}
-
-func TestSSLOpenSSLLiteralIPObeysExecutionPolicy(t *testing.T) {
-	if !openssl.IsAvailable() {
-		t.Skip("openssl binary is not installed")
-	}
-
-	// The input pins ::1 as its custom ip, and a custom dialer with no deny
-	// list would connect to it. The execution policy denies it, so the openssl
-	// binary must not connect to it.
-	var deniedHandshakes atomic.Int32
-	denied := serveTLS(t, "[::1]:0", "denied", &deniedHandshakes)
-	_, port, err := net.SplitHostPort(denied)
-	require.NoError(t, err)
-
-	options := testutils.DefaultOptions.Copy()
-	options.SetExecutionID(t.Name())
-	options.ExcludeTargets = []string{"::1"}
-	testutils.Init(options)
-	t.Cleanup(func() { testutils.Cleanup(options) })
-	executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{ID: "ssl-openssl-literal-ip"})
-	executerOpts.IsMultiProtocol = true
-	customDialer, err := fastdialer.NewDialer(fastdialer.DefaultOptions)
-	require.NoError(t, err)
-	t.Cleanup(customDialer.Close)
-	executerOpts.CustomFastdialer = customDialer
-
-	request := &Request{ID: "ssl-openssl-literal-ip", Address: "{{Host}}:{{Port}}", ScanMode: "openssl"}
-	require.NoError(t, request.Compile(executerOpts))
-
-	input := contextargs.NewWithInput(context.Background(), net.JoinHostPort("localhost", port))
-	input.MetaInput.CustomIP = "::1"
-	err = request.ExecuteWithResults(input, nil, nil, func(*output.InternalWrappedEvent) {})
-	require.Error(t, err)
-	require.Zero(t, deniedHandshakes.Load(), "openssl connected to an address the execution policy denies")
-}
-
-func TestSSLOpenSSLCustomIPObeysNetworkPolicy(t *testing.T) {
-	if !openssl.IsAvailable() {
-		t.Skip("openssl binary is not installed")
-	}
-
-	// The input pins the denied address ::1 as its custom ip, so the request
-	// skips the name lookup. The openssl binary must still not connect to it.
-	var deniedHandshakes atomic.Int32
-	denied := serveTLS(t, "[::1]:0", "denied", &deniedHandshakes)
-	_, port, err := net.SplitHostPort(denied)
-	require.NoError(t, err)
-
-	options := testutils.DefaultOptions.Copy()
-	options.SetExecutionID(t.Name())
-	options.ExcludeTargets = []string{"::1"}
-	testutils.Init(options)
-	t.Cleanup(func() { testutils.Cleanup(options) })
-	executerOpts := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{ID: "ssl-openssl-custom-ip"})
-	executerOpts.IsMultiProtocol = true
-
-	request := &Request{ID: "ssl-openssl-custom-ip", Address: "{{Host}}:{{Port}}", ScanMode: "openssl"}
-	require.NoError(t, request.Compile(executerOpts))
-
-	input := contextargs.NewWithInput(context.Background(), net.JoinHostPort("localhost", port))
-	input.MetaInput.CustomIP = "::1"
-	_ = request.ExecuteWithResults(input, nil, nil, func(*output.InternalWrappedEvent) {})
-	require.Zero(t, deniedHandshakes.Load(), "openssl connected to a denied custom ip")
-}

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()
+}