fix missing results in flow template + feature: internal matchers using internal: true (#4582)

* log warnings + use scanContext in flow

* refactor flow to use scanContext + log all events

* feat: internal matcher

* fix integration test

* bug fix extractor: merge dynamic values, fix missing extractors in file

* flow: fix 'No Results Found' if last statement output is false

* fix unit test

.gitignore

@@ -25,3 +25,6 @@ pkg/protocols/headless/engine/.cache
 /bindgen
 /jsdocgen
 /scrapefuncs
+/integration_tests/.cache/
+/integration_tests/.nuclei-config/
+/*.yaml

cmd/integration-test/flow.go

@@ -67,7 +67,7 @@ func (t *iterateValuesFlow) Execute(filePath string) error {
 	if err != nil {
 		return err
 	}
-	return expectResultsCount(results, 1)
+	return expectResultsCount(results, 2)
 }
 
 type dnsNsProbe struct{}
@@ -77,7 +77,7 @@ func (t *dnsNsProbe) Execute(filePath string) error {
 	if err != nil {
 		return err
 	}
-	return expectResultsCount(results, 1)
+	return expectResultsCount(results, 2)
 }
 
 func getBase64(input string) string {

integration_tests/flow/conditional-flow-negative.yaml

@@ -15,6 +15,7 @@ dns:
       - type: word
         words:
           - "ghost.io"
+        internal: true
 
 http:
   - method: GET

integration_tests/flow/conditional-flow.yaml

@@ -15,6 +15,7 @@ dns:
       - type: word
         words:
           - "ghost.io"
+        internal: true
 
 http:
   - method: GET

integration_tests/flow/dns-ns-probe.yaml

@@ -22,6 +22,7 @@ dns:
       - type: word
         words:
           - "IN\tNS"
+        internal: true
     extractors:
       - type: regex
         internal: true

integration_tests/flow/flow-hide-matcher.yaml

@@ -1,10 +1,10 @@
 id: flow-hide-matcher
 
 info:
-  name: Test HTTP Template
+  name: Test Flow Hide Matcher
   author: pdteam
   severity: info
-  description: In flow matcher output of previous step is hidden and only last event matcher output is shown
+  description: In Template any matcher can be marked as internal which hides it from the output.
 
 flow: http(1) && http(2)
 
@@ -17,6 +17,7 @@ http:
       - type: word
         words:
           - ok
+        internal: true
 
   - method: GET
     path:

integration_tests/flow/iterate-values-flow.yaml

@@ -21,9 +21,9 @@ http:
     extractors:
       - type: regex
         name: emails
-        internal: true
         regex:
           - '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
+        internal: true
 
   - method: GET
     path:
@@ -33,3 +33,9 @@ http:
       - type: word
         words:
           - "Welcome"
+    
+    extractors:
+      - type: dsl
+        name: email
+        dsl:
+          - email

pkg/operators/matchers/matchers.go

@@ -120,6 +120,14 @@ type Matcher struct {
 	//   - false
 	//   - true
 	MatchAll bool `yaml:"match-all,omitempty" json:"match-all,omitempty" jsonschema:"title=match all values,description=match all matcher values ignoring condition"`
+	// description: |
+	//  Internal when true hides the matcher from output. Default is false.
+	// It is meant to be used in multiprotocol / flow templates to create internal matcher condition without printing it in output.
+	// or other similar use cases.
+	// values:
+	//   - false
+	//   - true
+	Internal bool `yaml:"internal,omitempty" json:"internal,omitempty" jsonschema:"title=hide matcher from output,description=hide matcher from output"`
 
 	// cached data for the compiled matcher
 	condition     ConditionType // todo: this field should be the one used for overridden marshal ops

pkg/operators/matchers/validate.go

@@ -11,7 +11,7 @@ import (
 	"gopkg.in/yaml.v3"
 )
 
-var commonExpectedFields = []string{"Type", "Condition", "Name", "MatchAll", "Negative"}
+var commonExpectedFields = []string{"Type", "Condition", "Name", "MatchAll", "Negative", "Internal"}
 
 // Validate perform initial validation on the matcher structure
 func (matcher *Matcher) Validate() error {

pkg/operators/operators.go

@@ -90,6 +90,8 @@ type Result struct {
 
 	// Optional lineCounts for file protocol
 	LineCount string
+	// Operators is reference to operators that generated this result (Read-Only)
+	Operators *Operators
 }
 
 func (result *Result) HasMatch(name string) bool {
@@ -194,7 +196,11 @@ func (r *Result) Merge(result *Result) {
 		}
 	}
 	for k, v := range result.DynamicValues {
+		if _, ok := r.DynamicValues[k]; !ok {
 			r.DynamicValues[k] = v
+		} else {
+			r.DynamicValues[k] = sliceutil.Dedupe(append(r.DynamicValues[k], v...))
+		}
 	}
 	for k, v := range result.PayloadValues {
 		r.PayloadValues[k] = v
@@ -217,6 +223,7 @@ func (operators *Operators) Execute(data map[string]interface{}, match MatchFunc
 		Extracts:      make(map[string][]string),
 		DynamicValues: make(map[string][]string),
 		outputUnique:  make(map[string]struct{}),
+		Operators:     operators,
 	}
 
 	// state variable to check if all extractors are internal

pkg/protocols/file/operators.go

@@ -48,6 +48,10 @@ func (request *Request) Extract(data map[string]interface{}, extractor *extracto
 		return extractor.ExtractRegex(itemStr)
 	case extractors.KValExtractor:
 		return extractor.ExtractKval(data)
+	case extractors.JSONExtractor:
+		return extractor.ExtractJSON(itemStr)
+	case extractors.XPathExtractor:
+		return extractor.ExtractXPath(itemStr)
 	case extractors.DSLExtractor:
 		return extractor.ExtractDSL(data)
 	}

pkg/scan/scan_context.go

@@ -2,7 +2,9 @@ package scan
 
 import (
 	"context"
+	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/projectdiscovery/nuclei/v3/pkg/output"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
@@ -10,46 +12,52 @@ import (
 
 type ScanContext struct {
 	context.Context
+	// exported / configurable fields
 	Input *contextargs.Context
-	errors []error
-	events []*output.InternalWrappedEvent
 
+	// callbacks or hooks
 	OnError  func(error)
 	OnResult func(e *output.InternalWrappedEvent)
+
+	// unexported state fields
+	errors   []error
+	warnings []string
+	events   []*output.InternalWrappedEvent
+
+	// might not be required but better to sync
+	m sync.Mutex
 }
 
+// NewScanContext creates a new scan context using input
 func NewScanContext(input *contextargs.Context) *ScanContext {
 	return &ScanContext{Input: input}
 }
 
+// GenerateResult returns final results slice from all events
 func (s *ScanContext) GenerateResult() []*output.ResultEvent {
+	s.m.Lock()
+	defer s.m.Unlock()
 	return aggregateResults(s.events)
 }
 
-func aggregateResults(events []*output.InternalWrappedEvent) []*output.ResultEvent {
+// LogEvent logs events to all events and triggeres any callbacks
-	var results []*output.ResultEvent
-	for _, e := range events {
-		results = append(results, e.Results...)
-	}
-	return results
-}
-
-func joinErrors(errors []error) string {
-	var errorMessages []string
-	for _, e := range errors {
-		errorMessages = append(errorMessages, e.Error())
-	}
-	return strings.Join(errorMessages, "; ")
-}
-
 func (s *ScanContext) LogEvent(e *output.InternalWrappedEvent) {
+	s.m.Lock()
+	defer s.m.Unlock()
+	if e == nil {
+		// do not log nil events
+		return
+	}
 	if s.OnResult != nil {
 		s.OnResult(e)
 	}
 	s.events = append(s.events, e)
 }
 
+// LogError logs error to all events and triggeres any callbacks
 func (s *ScanContext) LogError(err error) {
+	s.m.Lock()
+	defer s.m.Unlock()
 	if err == nil {
 		return
 	}
@@ -68,3 +76,37 @@ func (s *ScanContext) LogError(err error) {
 		e.InternalEvent["error"] = errorMessage
 	}
 }
+
+// LogWarning logs warning to all events
+func (s *ScanContext) LogWarning(format string, args ...any) {
+	s.m.Lock()
+	defer s.m.Unlock()
+	val := fmt.Sprintf(format, args...)
+	s.warnings = append(s.warnings, val)
+
+	for _, e := range s.events {
+		if e.InternalEvent != nil {
+			e.InternalEvent["warning"] = strings.Join(s.warnings, "; ")
+		}
+	}
+}
+
+// aggregateResults aggregates results from multiple events
+func aggregateResults(events []*output.InternalWrappedEvent) []*output.ResultEvent {
+	var results []*output.ResultEvent
+	for _, e := range events {
+		results = append(results, e.Results...)
+	}
+	return results
+}
+
+// joinErrors joins multiple errors and returns a single error string
+func joinErrors(errors []error) string {
+	var errorMessages []string
+	for _, e := range errors {
+		if e != nil {
+			errorMessages = append(errorMessages, e.Error())
+		}
+	}
+	return strings.Join(errorMessages, "; ")
+}

pkg/tmplexec/exec.go

@@ -47,7 +47,7 @@ func NewTemplateExecuter(requests []protocols.Request, options *protocols.Execut
 		// we use a dummy input here because goal of flow executor at this point is to just check
 		// syntax and other things are correct before proceeding to actual execution
 		// during execution new instance of flow will be created as it is tightly coupled with lot of executor options
-		e.engine = flow.NewFlowExecutor(requests, contextargs.NewWithInput("dummy"), options, e.results)
+		e.engine = flow.NewFlowExecutor(requests, scan.NewScanContext(contextargs.NewWithInput("dummy")), options, e.results)
 	} else {
 		// Review:
 		// multiproto engine is only used if there is more than one protocol in template
@@ -117,6 +117,22 @@ func (e *TemplateExecuter) Execute(ctx *scan.ScanContext) (bool, error) {
 			// something went wrong
 			return
 		}
+		// check for internal true matcher event
+		if event.HasOperatorResult() && event.OperatorsResult.Matched && event.OperatorsResult.Operators != nil {
+			// note all matchers should have internal:true if it is a combination then print it
+			allInternalMatchers := true
+			for _, matcher := range event.OperatorsResult.Operators.Matchers {
+				if allInternalMatchers && !matcher.Internal {
+					allInternalMatchers = false
+					break
+				}
+			}
+			if allInternalMatchers {
+				// this is a internal event and no meant to be printed
+				return
+			}
+		}
+
 		// If no results were found, and also interactsh is not being used
 		// in that case we can skip it, otherwise we've to show failure in
 		// case of matcher-status flag.
@@ -139,8 +155,9 @@ func (e *TemplateExecuter) Execute(ctx *scan.ScanContext) (bool, error) {
 	// so in compile step earlier we compile it to validate javascript syntax and other things
 	// and while executing we create new instance of flow executor everytime
 	if e.options.Flow != "" {
-		flowexec := flow.NewFlowExecutor(e.requests, ctx.Input, e.options, results)
+		flowexec := flow.NewFlowExecutor(e.requests, ctx, e.options, results)
 		if err := flowexec.Compile(); err != nil {
+			ctx.LogError(err)
 			return false, err
 		}
 		err = flowexec.ExecuteWithResults(ctx)

pkg/tmplexec/flow/flow_executor.go

@@ -9,9 +9,7 @@ import (
 
 	"github.com/dop251/goja"
 	"github.com/projectdiscovery/gologger"
-	"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/protocols/common/generators"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/protocolstate"
 	"github.com/projectdiscovery/nuclei/v3/pkg/scan"
@@ -38,13 +36,12 @@ type ProtoOptions struct {
 
 // FlowExecutor is a flow executor for executing a flow
 type FlowExecutor struct {
-	input   *contextargs.Context
+	ctx     *scan.ScanContext // scan context (includes target etc)
 	options *protocols.ExecutorOptions
 
 	// javascript runtime reference and compiled program
 	jsVM    *goja.Runtime
 	program *goja.Program // compiled js program
-	lastEvent *output.InternalWrappedEvent // contains last event that was emitted
 
 	// protocol requests and their callback functions
 	allProtocols   map[string][]protocols.Request
@@ -56,7 +53,9 @@ type FlowExecutor struct {
 }
 
 // NewFlowExecutor creates a new flow executor from a list of requests
-func NewFlowExecutor(requests []protocols.Request, input *contextargs.Context, options *protocols.ExecutorOptions, results *atomic.Bool) *FlowExecutor {
+// Note: Unlike other engine for every target x template flow needs to be compiled and executed everytime
+// unlike other engines where we compile once and execute multiple times
+func NewFlowExecutor(requests []protocols.Request, ctx *scan.ScanContext, options *protocols.ExecutorOptions, results *atomic.Bool) *FlowExecutor {
 	allprotos := make(map[string][]protocols.Request)
 	for _, req := range requests {
 		switch req.Type() {
@@ -81,7 +80,8 @@ func NewFlowExecutor(requests []protocols.Request, input *contextargs.Context, o
 		case templateTypes.JavascriptProtocol:
 			allprotos[templateTypes.JavascriptProtocol.String()] = append(allprotos[templateTypes.JavascriptProtocol.String()], req)
 		default:
-			gologger.Error().Msgf("invalid request type %s", req.Type().String())
+			ctx.LogError(fmt.Errorf("invalid request type %s", req.Type().String()))
+			return nil
 		}
 	}
 	f := &FlowExecutor{
@@ -94,7 +94,7 @@ func NewFlowExecutor(requests []protocols.Request, input *contextargs.Context, o
 		protoFunctions: map[string]func(call goja.FunctionCall) goja.Value{},
 		results:        results,
 		jsVM:           protocolstate.NewJSRuntime(),
-		input:          input,
+		ctx:            ctx,
 	}
 	return f
 }
@@ -105,7 +105,7 @@ func (f *FlowExecutor) Compile() error {
 		f.results = new(atomic.Bool)
 	}
 	// load all variables and evaluate with existing data
-	variableMap := f.options.Variables.Evaluate(f.options.GetTemplateCtx(f.input.MetaInput).GetAll())
+	variableMap := f.options.Variables.Evaluate(f.options.GetTemplateCtx(f.ctx.Input.MetaInput).GetAll())
 	// cli options
 	optionVars := generators.BuildPayloadFromOptions(f.options.Options)
 	// constants
@@ -118,11 +118,11 @@ func (f *FlowExecutor) Compile() error {
 			if value, err := f.ReadDataFromFile(str); err == nil {
 				allVars[k] = value
 			} else {
-				gologger.Warning().Msgf("could not load file '%s' for variable '%s': %s", str, k, err)
+				f.ctx.LogWarning("could not load file '%s' for variable '%s': %s", str, k, err)
 			}
 		}
 	}
-	f.options.GetTemplateCtx(f.input.MetaInput).Merge(allVars) // merge all variables into template context
+	f.options.GetTemplateCtx(f.ctx.Input.MetaInput).Merge(allVars) // merge all variables into template context
 
 	// ---- define callback functions/objects----
 	f.protoFunctions = map[string]func(call goja.FunctionCall) goja.Value{}
@@ -165,24 +165,24 @@ func (f *FlowExecutor) Compile() error {
 func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 	defer func() {
 		if e := recover(); e != nil {
+			f.ctx.LogError(fmt.Errorf("panic occurred while executing target %v with flow: %v", ctx.Input.MetaInput.Input, e))
 			gologger.Error().Label(f.options.TemplateID).Msgf("panic occurred while executing target %v with flow: %v", ctx.Input.MetaInput.Input, e)
-			panic(e)
 		}
 	}()
 
-	f.input = ctx.Input
+	f.ctx.Input = ctx.Input
 	// -----Load all types of variables-----
 	// add all input args to template context
-	if f.input != nil && f.input.HasArgs() {
+	if f.ctx.Input != nil && f.ctx.Input.HasArgs() {
-		f.input.ForEach(func(key string, value interface{}) {
+		f.ctx.Input.ForEach(func(key string, value interface{}) {
-			f.options.GetTemplateCtx(f.input.MetaInput).Set(key, value)
+			f.options.GetTemplateCtx(f.ctx.Input.MetaInput).Set(key, value)
 		})
 	}
 	if ctx.OnResult == nil {
 		return fmt.Errorf("output callback cannot be nil")
 	}
 	// pass flow and execute the js vm and handle errors
-	value, err := f.jsVM.RunProgram(f.program)
+	_, err := f.jsVM.RunProgram(f.program)
 	if err != nil {
 		ctx.LogError(err)
 		return errorutil.NewWithErr(err).Msgf("failed to execute flow\n%v\n", f.options.Flow)
@@ -192,13 +192,7 @@ func (f *FlowExecutor) ExecuteWithResults(ctx *scan.ScanContext) error {
 		ctx.LogError(runtimeErr)
 		return errorutil.NewWithErr(runtimeErr).Msgf("got following errors while executing flow")
 	}
-	// this is where final result is generated/created
+
-	ctx.LogEvent(f.lastEvent)
-	if value.Export() != nil {
-		f.results.Store(value.ToBoolean())
-	} else {
-		f.results.Store(true)
-	}
 	return nil
 }
 

pkg/tmplexec/flow/flow_internal.go

@@ -1,6 +1,7 @@
 package flow
 
 import (
+	"fmt"
 	"reflect"
 	"sync/atomic"
 
@@ -21,11 +22,11 @@ import (
 func (f *FlowExecutor) requestExecutor(reqMap mapsutil.Map[string, protocols.Request], opts *ProtoOptions) bool {
 	defer func() {
 		// evaluate all variables after execution of each protocol
-		variableMap := f.options.Variables.Evaluate(f.options.GetTemplateCtx(f.input.MetaInput).GetAll())
+		variableMap := f.options.Variables.Evaluate(f.options.GetTemplateCtx(f.ctx.Input.MetaInput).GetAll())
-		f.options.GetTemplateCtx(f.input.MetaInput).Merge(variableMap) // merge all variables into template context
+		f.options.GetTemplateCtx(f.ctx.Input.MetaInput).Merge(variableMap) // merge all variables into template context
 
 		// to avoid polling update template variables everytime we execute a protocol
-		var m map[string]interface{} = f.options.GetTemplateCtx(f.input.MetaInput).GetAll()
+		var m map[string]interface{} = f.options.GetTemplateCtx(f.ctx.Input.MetaInput).GetAll()
 		_ = f.jsVM.Set("template", m)
 	}()
 	matcherStatus := &atomic.Bool{} // due to interactsh matcher polling logic this needs to be atomic bool
@@ -34,7 +35,7 @@ func (f *FlowExecutor) requestExecutor(reqMap mapsutil.Map[string, protocols.Req
 		// execution logic for http()/dns() etc
 		for index := range f.allProtocols[opts.protoName] {
 			req := f.allProtocols[opts.protoName][index]
-			err := req.ExecuteWithResults(f.input, output.InternalEvent(f.options.GetTemplateCtx(f.input.MetaInput).GetAll()), nil, f.getProtoRequestCallback(req, matcherStatus, opts))
+			err := req.ExecuteWithResults(f.ctx.Input, output.InternalEvent(f.options.GetTemplateCtx(f.ctx.Input.MetaInput).GetAll()), nil, f.protocolResultCallback(req, matcherStatus, opts))
 			if err != nil {
 				// save all errors in a map with id as key
 				// its less likely that there will be race condition but just in case
@@ -44,7 +45,7 @@ func (f *FlowExecutor) requestExecutor(reqMap mapsutil.Map[string, protocols.Req
 				}
 				err = f.allErrs.Set(opts.protoName+":"+id, err)
 				if err != nil {
-					gologger.Error().Msgf("failed to store flow runtime errors got %v", err)
+					f.ctx.LogError(fmt.Errorf("failed to store flow runtime errors got %v", err))
 				}
 				return matcherStatus.Load()
 			}
@@ -56,36 +57,38 @@ func (f *FlowExecutor) requestExecutor(reqMap mapsutil.Map[string, protocols.Req
 	for _, id := range opts.reqIDS {
 		req, ok := reqMap[id]
 		if !ok {
-			gologger.Error().Msgf("[%v] invalid request id '%s' provided", f.options.TemplateID, id)
+			f.ctx.LogError(fmt.Errorf("[%v] invalid request id '%s' provided", f.options.TemplateID, id))
 			// compile error
 			if err := f.allErrs.Set(opts.protoName+":"+id, ErrInvalidRequestID.Msgf(f.options.TemplateID, id)); err != nil {
-				gologger.Error().Msgf("failed to store flow runtime errors got %v", err)
+				f.ctx.LogError(fmt.Errorf("failed to store flow runtime errors got %v", err))
 			}
 			return matcherStatus.Load()
 		}
-		err := req.ExecuteWithResults(f.input, output.InternalEvent(f.options.GetTemplateCtx(f.input.MetaInput).GetAll()), nil, f.getProtoRequestCallback(req, matcherStatus, opts))
+		err := req.ExecuteWithResults(f.ctx.Input, output.InternalEvent(f.options.GetTemplateCtx(f.ctx.Input.MetaInput).GetAll()), nil, f.protocolResultCallback(req, matcherStatus, opts))
 		if err != nil {
 			index := id
 			err = f.allErrs.Set(opts.protoName+":"+index, err)
 			if err != nil {
-				gologger.Error().Msgf("failed to store flow runtime errors got %v", err)
+				f.ctx.LogError(fmt.Errorf("failed to store flow runtime errors got %v", err))
 			}
 		}
 	}
 	return matcherStatus.Load()
 }
 
-// getProtoRequestCallback returns a callback that is executed
+// protocolResultCallback returns a callback that is executed
 // after execution of each protocol request
-func (f *FlowExecutor) getProtoRequestCallback(req protocols.Request, matcherStatus *atomic.Bool, opts *ProtoOptions) func(result *output.InternalWrappedEvent) {
+func (f *FlowExecutor) protocolResultCallback(req protocols.Request, matcherStatus *atomic.Bool, opts *ProtoOptions) func(result *output.InternalWrappedEvent) {
 	return func(result *output.InternalWrappedEvent) {
 		if result != nil {
-			f.results.CompareAndSwap(false, true)
+			// Note: flow specific implicit behaviours should be handled here
-			f.lastEvent = result
+			// before logging the event
+			f.ctx.LogEvent(result)
 			// export dynamic values from operators (i.e internal:true)
 			// add add it to template context
 			// this is a conflicting behaviour with iterate-all
 			if result.HasOperatorResult() {
+				f.results.CompareAndSwap(false, true)
 				// this is to handle case where there is any operator result (matcher or extractor)
 				matcherStatus.CompareAndSwap(false, result.OperatorsResult.Matched)
 				if !result.OperatorsResult.Matched && !hasMatchers(req.GetCompiledOperators()) {
@@ -95,7 +98,7 @@ func (f *FlowExecutor) getProtoRequestCallback(req protocols.Request, matcherSta
 				}
 				if len(result.OperatorsResult.DynamicValues) > 0 {
 					for k, v := range result.OperatorsResult.DynamicValues {
-						f.options.GetTemplateCtx(f.input.MetaInput).Set(k, v)
+						f.options.GetTemplateCtx(f.ctx.Input.MetaInput).Set(k, v)
 					}
 				}
 			} else if !result.HasOperatorResult() && !hasOperators(req.GetCompiledOperators()) {
@@ -130,7 +133,7 @@ func (f *FlowExecutor) registerBuiltInFunctions() error {
 		default:
 			gologger.DefaultLogger.Print().Msgf("[%v] %v", aurora.BrightCyan("JS"), value)
 		}
-		return goja.Null()
+		return call.Argument(0) // return the same value
 	}); err != nil {
 		return err
 	}
@@ -138,7 +141,7 @@ func (f *FlowExecutor) registerBuiltInFunctions() error {
 	if err := f.jsVM.Set("set", func(call goja.FunctionCall) goja.Value {
 		varName := call.Argument(0).Export()
 		varValue := call.Argument(1).Export()
-		f.options.GetTemplateCtx(f.input.MetaInput).Set(types.ToString(varName), varValue)
+		f.options.GetTemplateCtx(f.ctx.Input.MetaInput).Set(types.ToString(varName), varValue)
 		return goja.Null()
 	}); err != nil {
 		return err
@@ -179,7 +182,7 @@ func (f *FlowExecutor) registerBuiltInFunctions() error {
 		return err
 	}
 
-	var m = f.options.GetTemplateCtx(f.input.MetaInput).GetAll()
+	var m = f.options.GetTemplateCtx(f.ctx.Input.MetaInput).GetAll()
 	if m == nil {
 		m = map[string]interface{}{}
 	}