check literals

README.md

@@ -256,7 +256,6 @@ 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,7 +407,6 @@ 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,7 +73,6 @@ 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"
@@ -677,7 +676,6 @@ 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,14 +39,12 @@ func createEphemeralObjects(ctx context.Context, base *NucleiEngine, opts *types
 		Catalog:      base.catalog,
 		IssuesClient: base.rc,
 		RateLimiter:  base.rateLimiter,
-
-		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,
+		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,7 +36,6 @@ 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"
 )
@@ -280,7 +279,6 @@ 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,7 +3,6 @@ package core
 import (
 	"fmt"
 	"net/http/cookiejar"
-	"sync"
 	"sync/atomic"
 
 	"github.com/projectdiscovery/gologger"
@@ -33,45 +32,22 @@ 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())
-		// Root steps do not own a waitgroup slot.
-		if err := e.runWorkflowStep(template, newCtx, results, swg, &pending, w, nil); err != nil {
+		if err := e.runWorkflowStep(template, newCtx, results, swg, w); err != nil {
 			gologger.Warning().Msgf(workflowStepExecutionError, template.Template, err)
 		}
 	}
 
-	pending.Wait()
+	swg.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.
-//
-// 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 {
+func (e *Engine) runWorkflowStep(template *workflows.WorkflowTemplate, ctx *scan.ScanContext, results *atomic.Bool, swg *syncutil.AdaptiveWaitGroup, w *workflows.Workflow) error {
 	var firstMatched bool
 	var err error
 	var mainErr error
@@ -155,25 +131,18 @@ 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 pending.Done()
-							release := slotRelease(swg)
-							defer release()
+							defer swg.Done()
 
 							// 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, pending, w, release); err != nil {
+							if err := e.runWorkflowStep(subtemplate, subCtx, results, swg, w); err != nil {
 								gologger.Warning().Msgf(workflowStepExecutionError, subtemplate.Template, err)
 							}
 						}(subtemplate)
@@ -193,27 +162,18 @@ 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, pending, w, release); err != nil {
+				if err := e.runWorkflowStep(template, subCtx, results, swg, w); err != nil {
 					gologger.Warning().Msgf(workflowStepExecutionError, template.Template, err)
 				}
+				swg.Done()
 			}(subtemplate)
 		}
 	}

pkg/core/workflow_execute_test.go

@@ -303,89 +303,6 @@ 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,7 +31,6 @@ 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

@@ -1,34 +0,0 @@
-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,7 +15,6 @@ 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

@@ -1,190 +0,0 @@
-
-/**
- * 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

@@ -1,596 +0,0 @@
-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

@@ -1,395 +0,0 @@
-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

@@ -1,339 +0,0 @@
-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

@@ -1,103 +0,0 @@
-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

@@ -1,125 +0,0 @@
-//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

@@ -1,200 +0,0 @@
-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,9 +82,6 @@ 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,7 +147,6 @@ 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.
@@ -417,7 +416,6 @@ 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,7 +15,6 @@ 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
@@ -226,7 +225,7 @@ func (request *Request) savedRequest(dump string) string {
 	if request.options == nil || request.options.Options == nil {
 		return dump
 	}
-	return string(capDumpBody(convUtil.Bytes(dump), request.options.Options.RequestSaveSize))
+	return string(capDumpBody([]byte(dump), request.options.Options.RequestSaveSize))
 }
 
 // eventResponse returns the full response of event. Events matched after the

pkg/protocols/http/request.go

@@ -331,7 +331,6 @@ 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
 	}
@@ -339,87 +338,72 @@ 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 {
-				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 <-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 {
 					select {
 					case <-spmHandler.Done():
 						spmHandler.Release()
-						return false
-					case spmHandler.ResultChan <- err:
-						request.recordHostResultAndCancelIfUnresponsive(t.updatedInput, err, spmHandler.Cancel)
+						continue
+					case <-ctx.Done():
 						spmHandler.Release()
+						continue
+					case spmHandler.ResultChan <- err:
+						spmHandler.Release()
+						continue
 					}
-					return false
-				}() {
-					return
+				}
+				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()
 				}
 			}
 		}()
@@ -438,7 +422,8 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 
 		select {
 		case <-input.Context().Done():
-			stopWorkers()
+			close(tasks)
+			workersWg.Wait()
 			return input.Context().Err()
 		default:
 		}
@@ -446,7 +431,8 @@ 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 {
-				stopWorkers()
+				close(tasks)
+				workersWg.Wait()
 				return err
 			}
 			// if payload concurrency increased, add more workers
@@ -463,22 +449,15 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 			break
 		}
 
-		// 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)
+		ctx := input.Context()
+		generatedHttpRequest, err := generator.Make(ctx, 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()))
-			stopWorkers()
+			close(tasks)
+			workersWg.Wait()
 			return err
 		}
 		hasInteractMarkers := interactsh.HasMarkers(inputData) || len(generatedHttpRequest.interactshURLs) > 0
@@ -488,32 +467,33 @@ 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()
-			stopWorkers()
+			close(tasks)
+			workersWg.Wait()
 			return nil
 		}
 		select {
 		case <-spmHandler.Done():
-			reservation.done()
-			stopWorkers()
+			close(tasks)
+			workersWg.Wait()
 			spmHandler.Wait()
 			if spmHandler.FoundFirstMatch() {
 				return nil
 			}
 			return multierr.Combine(spmHandler.CombinedResults()...)
 		case <-workerCtx.Done():
-			reservation.done()
-			stopWorkers()
+			close(tasks)
+			workersWg.Wait()
 			if spmHandler.FoundFirstMatch() {
 				return nil
 			}
 			return context.Cause(workerCtx)
-		case tasks <- task{req: generatedHttpRequest, body: reservation, updatedInput: updatedInput, hasInteractMarkers: hasInteractMarkers}:
+		case tasks <- task{req: generatedHttpRequest, updatedInput: updatedInput, hasInteractMarkers: hasInteractMarkers}:
 		}
 		request.options.Progress.IncrementRequests()
 	}
-	stopWorkers()
+	close(tasks)
+	workersWg.Wait()
 	spmHandler.Wait()
 	if spmHandler.FoundFirstMatch() {
 		// ignore any context cancellation and in-transit execution errors
@@ -602,7 +582,7 @@ func (request *Request) executeTurboHTTP(input *contextargs.Context, dynamicValu
 		}
 
 		ctx := input.Context()
-		generatedHttpRequest, reservation, err := request.makeReserved(ctx, generator, input, inputData, payloads, dynamicValues)
+		generatedHttpRequest, err := generator.Make(ctx, input, inputData, payloads, dynamicValues)
 		if err != nil {
 			request.options.Progress.IncrementFailedRequestsBy(int64(generator.Total()))
 			return err
@@ -613,18 +593,15 @@ 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, body *bodyReservation, requestInput *contextargs.Context) {
+		go func(httpRequest *generatedRequest, requestInput *contextargs.Context) {
 			defer spmHandler.Release()
-			defer body.done()
 			if spmHandler.FoundFirstMatch() || request.isUnresponsiveAddress(requestInput) {
 				// skip if first match is found
 				return
@@ -649,7 +626,7 @@ func (request *Request) executeTurboHTTP(input *contextargs.Context, dynamicValu
 				request.recordHostResultAndCancelIfUnresponsive(requestInput, err, spmHandler.Cancel)
 				return
 			}
-		}(generatedHttpRequest, reservation, updatedInput)
+		}(generatedHttpRequest, updatedInput)
 		request.options.Progress.IncrementRequests()
 	}
 	spmHandler.Wait()
@@ -699,17 +676,16 @@ executionLoop:
 			hasInteractMatchers := interactsh.HasMatchers(request.CompiledOperators)
 
 			ctx := input.Context()
-			generatedHttpRequest, reservation, err := request.makeReserved(ctx, generator, input, data, payloads, dynamicValue)
+			ctxWithTimeout, cancel := context.WithTimeoutCause(ctx, request.options.Options.GetTimeouts().HttpTimeout, ErrHttpEngineRequestDeadline)
+			defer cancel()
+
+			generatedHttpRequest, err := generator.Make(ctxWithTimeout, 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
@@ -1084,20 +1060,6 @@ 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,7 +7,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"slices"
 	"strings"
 
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/generators"
@@ -83,9 +82,6 @@ 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
 	}
@@ -93,7 +89,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 = slices.Insert(data, end, acceptEncodingGzip...)
+		data = append(data[:end:end], append([]byte("Accept-Encoding: gzip\r\n"), data[end:]...)...)
 	}
 	return requestDump{data: appendTruncationMarker(data, capture.dropped), truncated: capture.dropped}, nil
 }
@@ -127,10 +123,6 @@ 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 {
@@ -185,19 +177,6 @@ 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,7 +44,6 @@ 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"
 )
 
@@ -107,8 +106,6 @@ 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
@@ -398,7 +395,6 @@ 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,
@@ -710,7 +706,6 @@ 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,6 +23,7 @@ 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"
@@ -34,6 +35,7 @@ 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"
 )
 
@@ -202,6 +204,21 @@ 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
@@ -255,6 +272,23 @@ 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

@@ -0,0 +1,179 @@
+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,8 +370,6 @@ 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
@@ -767,7 +765,6 @@ 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

@@ -1,38 +0,0 @@
-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

@@ -1,68 +0,0 @@
-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()
-}