Make request headers and query params available to the HTTP Request Executor

Refs #1354
Signed-off-by: thisarawelmilla <thisara@wso2.com>

Author: Thisara-Welmilla

backend/internal/flow/core/utils.go

@@ -16,7 +16,19 @@ import (
 // TODO: Extend to support {{user(key)}}, {{env(key)}}, etc.
 var placeholderPattern = regexp.MustCompile(`{{\s*ctx\(\s*(\w+)\s*\)\s*}}`)
 
-// ResolvePlaceholder resolves a single placeholder string using the "{{ctx(key)}}" syntax.
+// requestPlaceholderPattern matches {{request(<selector>)}} with optional whitespace, capturing the
+// raw selector so it can be parsed into an optional source, a type (header/query) and a name.
+var requestPlaceholderPattern = regexp.MustCompile(`{{\s*request\(\s*([^)]*?)\s*\)\s*}}`)
+
+// Request selector source tokens used in {{request(...)}} placeholders. requestSourceInit refers
+// to the flow-initiation request (the default) and requestSourceFlow to the current flow step request.
+const (
+	requestSourceInit = "init"
+	requestSourceFlow = "flow"
+)
+
+// ResolvePlaceholder resolves a single placeholder string using the "{{ctx(key)}}" and
+// "{{request(...)}}" syntaxes.
 // If no placeholder is found, the original value is returned.
 // If a placeholder is found but the key doesn't exist in any data source, the placeholder is kept as-is.
 func ResolvePlaceholder(ctx *providers.NodeContext, value string, execResp *providers.ExecutorResponse,
@@ -27,7 +39,10 @@ func ResolvePlaceholder(ctx *providers.NodeContext, value string, execResp *prov
 
 	var contextUserRef *providers.EntityReference
 
-	return placeholderPattern.ReplaceAllStringFunc(value, func(match string) string {
+	// Resolve {{ctx(...)}} first, then {{request(...)}}. Because ReplaceAllStringFunc does not
+	// re-scan its own output, resolving request placeholders last keeps any {{ctx(...)}}-looking
+	// text that a request header or query value happens to contain from being resolved as context.
+	value = placeholderPattern.ReplaceAllStringFunc(value, func(match string) string {
 		submatches := placeholderPattern.FindStringSubmatch(match)
 		if len(submatches) < 2 {
 			return match
@@ -77,6 +92,97 @@ func ResolvePlaceholder(ctx *providers.NodeContext, value string, execResp *prov
 		// If not found, keep the placeholder as-is
 		return match
 	})
+
+	return resolveRequestPlaceholders(ctx, value)
+}
+
+// resolveRequestPlaceholders resolves {{request(...)}} placeholders using the HTTP request data
+// carried on the node context. The selector is "[source.]type.name" where source is "init"
+// (flow-initiation request, the default) or "flow" (current flow step request), and type is
+// "header" or "query". Header lookups are case-insensitive per RFC 7230; query lookups are
+// case-sensitive. When multiple values exist, the first is returned. Unresolvable placeholders
+// (missing request, unknown source/type, or absent name) are kept as-is.
+func resolveRequestPlaceholders(ctx *providers.NodeContext, value string) string {
+	return requestPlaceholderPattern.ReplaceAllStringFunc(value, func(match string) string {
+		submatches := requestPlaceholderPattern.FindStringSubmatch(match)
+		if len(submatches) < 2 {
+			return match
+		}
+
+		source, kind, name, ok := parseRequestSelector(submatches[1])
+		if !ok {
+			return match
+		}
+
+		var req *providers.InitiatorRequest
+		switch source {
+		case requestSourceInit:
+			req = ctx.GetInitiatorRequest()
+		case requestSourceFlow:
+			req = ctx.GetCurrentRequest()
+		}
+		if req == nil {
+			return match
+		}
+
+		var resolved string
+		var found bool
+		switch kind {
+		case "header":
+			resolved, found = firstHeaderValue(req.Headers, name)
+		case "query":
+			resolved, found = firstValue(req.QueryParams[name])
+		}
+		if !found {
+			return match
+		}
+		return resolved
+	})
+}
+
+// parseRequestSelector splits a request selector into its source, type and name components.
+// It accepts "type.name" (source defaults to "init") or "source.type.name". source must be
+// "init" or "flow" and type must be "header" or "query"; name is the remainder and may contain dots.
+func parseRequestSelector(selector string) (source, kind, name string, ok bool) {
+	source = requestSourceInit
+
+	head, rest, hasRest := strings.Cut(selector, ".")
+	if !hasRest {
+		return "", "", "", false
+	}
+	if head == requestSourceInit || head == requestSourceFlow {
+		source = head
+		head, rest, hasRest = strings.Cut(rest, ".")
+		if !hasRest {
+			return "", "", "", false
+		}
+	}
+
+	if head != "header" && head != "query" {
+		return "", "", "", false
+	}
+	if rest == "" {
+		return "", "", "", false
+	}
+	return source, head, rest, true
+}
+
+// firstHeaderValue returns the first value for a header, matching the name case-insensitively.
+func firstHeaderValue(headers map[string][]string, name string) (string, bool) {
+	for key, values := range headers {
+		if strings.EqualFold(key, name) {
+			return firstValue(values)
+		}
+	}
+	return "", false
+}
+
+// firstValue returns the first element of a string slice.
+func firstValue(values []string) (string, bool) {
+	if len(values) == 0 {
+		return "", false
+	}
+	return values[0], true
 }
 
 // fetchContextUserRef attempts to resolve the authenticated user's entity reference using the authn provider.

backend/internal/flow/core/utils_test.go

@@ -382,3 +382,101 @@ func (s *UtilsTestSuite) TestResolvePlaceholderSpecialCharactersInValue() {
 		})
 	}
 }
+
+func (s *UtilsTestSuite) TestResolveRequestPlaceholderFromInitiatorRequest() {
+	ctx := &providers.NodeContext{}
+	ctx.SetInitiatorRequest(&providers.InitiatorRequest{
+		Headers:     map[string][]string{"User-Agent": {"curl/8.0"}},
+		QueryParams: map[string][]string{"utm_source": {"newsletter"}},
+	})
+
+	tests := []struct {
+		name     string
+		input    string
+		expected string
+	}{
+		{"Header default source", "{{request(header.User-Agent)}}", "curl/8.0"},
+		{"Header explicit init source", "{{request(init.header.User-Agent)}}", "curl/8.0"},
+		{"Header case-insensitive lookup", "{{request(header.user-agent)}}", "curl/8.0"},
+		{"Query default source", "{{request(query.utm_source)}}", "newsletter"},
+		{"Query explicit init source", "{{request(init.query.utm_source)}}", "newsletter"},
+		{"Embedded in larger value", "ua={{request(header.User-Agent)}}", "ua=curl/8.0"},
+	}
+
+	for _, tt := range tests {
+		s.Run(tt.name, func() {
+			result := ResolvePlaceholder(ctx, tt.input, nil, nil, nil)
+			s.Equal(tt.expected, result)
+		})
+	}
+}
+
+func (s *UtilsTestSuite) TestResolveRequestPlaceholderFromCurrentRequest() {
+	ctx := &providers.NodeContext{}
+	ctx.SetInitiatorRequest(&providers.InitiatorRequest{
+		Headers: map[string][]string{"User-Agent": {"init-agent"}},
+	})
+	ctx.SetCurrentRequest(&providers.InitiatorRequest{
+		Headers:     map[string][]string{"User-Agent": {"flow-agent"}},
+		QueryParams: map[string][]string{"step": {"otp"}},
+	})
+
+	s.Equal("flow-agent", ResolvePlaceholder(ctx, "{{request(flow.header.User-Agent)}}", nil, nil, nil))
+	s.Equal("otp", ResolvePlaceholder(ctx, "{{request(flow.query.step)}}", nil, nil, nil))
+	// Default source stays init even when a current request is present.
+	s.Equal("init-agent", ResolvePlaceholder(ctx, "{{request(header.User-Agent)}}", nil, nil, nil))
+}
+
+func (s *UtilsTestSuite) TestResolveRequestPlaceholderQueryIsCaseSensitive() {
+	ctx := &providers.NodeContext{}
+	ctx.SetInitiatorRequest(&providers.InitiatorRequest{
+		QueryParams: map[string][]string{"utm_source": {"newsletter"}},
+	})
+
+	s.Equal("{{request(query.UTM_SOURCE)}}",
+		ResolvePlaceholder(ctx, "{{request(query.UTM_SOURCE)}}", nil, nil, nil),
+		"query lookups must be case-sensitive")
+}
+
+func (s *UtilsTestSuite) TestResolveRequestPlaceholderFirstValueWins() {
+	ctx := &providers.NodeContext{}
+	ctx.SetInitiatorRequest(&providers.InitiatorRequest{
+		Headers: map[string][]string{"X-Forwarded-For": {"1.1.1.1", "2.2.2.2"}},
+	})
+
+	s.Equal("1.1.1.1", ResolvePlaceholder(ctx, "{{request(header.X-Forwarded-For)}}", nil, nil, nil))
+}
+
+func (s *UtilsTestSuite) TestResolveRequestPlaceholderUnresolvable() {
+	ctxWithReq := &providers.NodeContext{}
+	ctxWithReq.SetInitiatorRequest(&providers.InitiatorRequest{
+		Headers: map[string][]string{"User-Agent": {"curl/8.0"}},
+	})
+
+	ctxWithoutReq := &providers.NodeContext{}
+
+	tests := []struct {
+		name     string
+		ctx      *providers.NodeContext
+		input    string
+		expected string
+	}{
+		{"Nil initiator request", ctxWithoutReq, "{{request(header.User-Agent)}}",
+			"{{request(header.User-Agent)}}"},
+		{"Nil current request", ctxWithReq, "{{request(flow.header.User-Agent)}}",
+			"{{request(flow.header.User-Agent)}}"},
+		{"Unknown header", ctxWithReq, "{{request(header.X-Missing)}}", "{{request(header.X-Missing)}}"},
+		{"Unknown source", ctxWithReq, "{{request(other.header.User-Agent)}}",
+			"{{request(other.header.User-Agent)}}"},
+		{"Unknown type", ctxWithReq, "{{request(cookie.session)}}", "{{request(cookie.session)}}"},
+		{"Missing name", ctxWithReq, "{{request(header.)}}", "{{request(header.)}}"},
+		{"Missing type separator", ctxWithReq, "{{request(header)}}", "{{request(header)}}"},
+	}
+
+	for _, tt := range tests {
+		s.Run(tt.name, func() {
+			result := ResolvePlaceholder(tt.ctx, tt.input, nil, nil, nil)
+			s.Equal(tt.expected, result)
+		})
+	}
+}

backend/internal/flow/executor/http_request_executor_test.go

@@ -122,6 +122,56 @@ func (suite *HTTPRequestExecutorTestSuite) TestResolvePlaceholdersInConfig() {
 	assert.Equal(suite.T(), "test@example.com", receivedBody["email"])
 }
 
+func (suite *HTTPRequestExecutorTestSuite) TestResolveRequestPlaceholdersInConfig() {
+	var receivedURL string
+	var receivedHeaders http.Header
+	var receivedBody map[string]interface{}
+
+	suite.mockServer = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		receivedURL = r.URL.RequestURI()
+		receivedHeaders = r.Header
+		err := json.NewDecoder(r.Body).Decode(&receivedBody)
+		if err != nil {
+			receivedBody = nil
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+
+	ctx := &providers.NodeContext{
+		ExecutionID: "test-flow",
+		NodeProperties: map[string]interface{}{
+			"url":    suite.mockServer.URL + "/collect?src={{request(query.utm_source)}}",
+			"method": "POST",
+			"headers": map[string]interface{}{
+				"X-Forwarded-Agent": "{{request(header.User-Agent)}}",
+				"X-Step-Agent":      "{{request(flow.header.User-Agent)}}",
+			},
+			"body": map[string]interface{}{
+				"host":   "{{request(header.Host)}}",
+				"source": "{{request(init.query.utm_source)}}",
+			},
+		},
+	}
+	ctx.SetInitiatorRequest(&providers.InitiatorRequest{
+		Headers:     map[string][]string{"User-Agent": {"init-agent"}, "Host": {"id.example.com"}},
+		QueryParams: map[string][]string{"utm_source": {"newsletter"}},
+	})
+	ctx.SetCurrentRequest(&providers.InitiatorRequest{
+		Headers: map[string][]string{"User-Agent": {"step-agent"}},
+	})
+
+	execResp, err := suite.executor.Execute(ctx)
+
+	assert.NoError(suite.T(), err)
+	assert.Equal(suite.T(), providers.ExecComplete, execResp.Status)
+
+	assert.Equal(suite.T(), "/collect?src=newsletter", receivedURL)
+	assert.Equal(suite.T(), "init-agent", receivedHeaders.Get("X-Forwarded-Agent"))
+	assert.Equal(suite.T(), "step-agent", receivedHeaders.Get("X-Step-Agent"))
+	assert.Equal(suite.T(), "id.example.com", receivedBody["host"])
+	assert.Equal(suite.T(), "newsletter", receivedBody["source"])
+}
+
 func (suite *HTTPRequestExecutorTestSuite) TestResolvePlaceholderUserIDSpecialHandling() {
 	var receivedBody map[string]interface{}
 

backend/internal/flow/flowexec/engine.go

@@ -196,6 +196,7 @@ func (fe *flowEngine) executeNodePackage(ctx *EngineContext,
 		ExecutionHistory:  ctx.ExecutionHistory,
 	}
 	nodeCtx.SetInitiatorRequest(ctx.GetInitiatorRequest())
+	nodeCtx.SetCurrentRequest(ctx.GetCurrentRequest())
 	if nodeCtx.NodeInputs == nil {
 		nodeCtx.NodeInputs = make([]providers.Input, 0)
 	}

backend/internal/flow/flowexec/handler.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/thunder-id/thunderid/internal/flow/session"
 	tidcommon "github.com/thunder-id/thunderid/pkg/thunderidengine/common"
+	"github.com/thunder-id/thunderid/pkg/thunderidengine/providers"
 
 	serverconst "github.com/thunder-id/thunderid/internal/system/constants"
 	"github.com/thunder-id/thunderid/internal/system/error/apierror"
@@ -60,6 +61,13 @@ func (h *flowExecutionHandler) HandleFlowExecutionRequest(w http.ResponseWriter,
 	// them available to the flow service, which selects the handle once the flow is known.
 	ctx := session.WithInbound(r.Context(), h.ssoTransport.Read(r))
 
+	// Carry the current step's request headers and query params so flow executors can publish them
+	// via {{request(flow.*)}} placeholders. Credential-bearing headers are stripped at this boundary.
+	ctx = withCurrentRequest(ctx, &providers.InitiatorRequest{
+		Headers:     sysutils.FilterSensitiveHeaders(r.Header),
+		QueryParams: r.URL.Query(),
+	})
+
 	var flowStep *FlowStep
 	var flowErr *tidcommon.ServiceError
 	if flowID != "" {

backend/internal/flow/flowexec/handler_test.go

@@ -191,6 +191,40 @@ func (s *HandlerTestSuite) TestHandleFlowExecutionRequest_PropagatesInboundSSOCo
 	s.Equal("inbound-handle", gotInbound.HandleFor("flow-1"))
 }
 
+// TestHandleFlowExecutionRequest_CapturesCurrentRequest verifies the current step's request headers
+// and query params are captured onto the service context, with credential headers stripped.
+func (s *HandlerTestSuite) TestHandleFlowExecutionRequest_CapturesCurrentRequest() {
+	t := s.T()
+	mockSvc := NewFlowExecServiceInterfaceMock(t)
+
+	var gotRequest *providers.InitiatorRequest
+	mockSvc.EXPECT().Execute(mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Run(func(ctx context.Context, _ string, _ string, _ string, _ bool, _ string,
+			_ map[string]string, _ string, _ string, _ string) {
+			gotRequest = currentRequestFrom(ctx)
+		}).
+		Return(&FlowStep{ExecutionID: "exec-1", Status: providers.FlowStatusIncomplete},
+			(*tidcommon.ServiceError)(nil))
+
+	h := newFlowExecutionHandler(mockSvc, session.NewCookieTransport(false), 0)
+	req := httptest.NewRequest(http.MethodPost, "/flow/execute?utm_source=newsletter",
+		bytes.NewBufferString(testFlowExecRequestBody))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("User-Agent", "curl/8.0")
+	req.Header.Set("Authorization", "Bearer secret")
+	w := httptest.NewRecorder()
+
+	h.HandleFlowExecutionRequest(w, req)
+
+	s.Equal(http.StatusOK, w.Code)
+	s.Require().NotNil(gotRequest, "current request must be captured onto the service context")
+	s.Equal("curl/8.0", http.Header(gotRequest.Headers).Get("User-Agent"))
+	s.Empty(http.Header(gotRequest.Headers).Get("Authorization"), "credential headers must be stripped")
+	s.Require().Contains(gotRequest.QueryParams, "utm_source")
+	s.Equal("newsletter", gotRequest.QueryParams["utm_source"][0])
+}
+
 // TestHandleFlowExecutionRequest_WritesSSOHandleCookie verifies a minted handle is emitted as the
 // per-flow cookie with the configured TTL and secure/http-only transport settings.
 func (s *HandlerTestSuite) TestHandleFlowExecutionRequest_WritesSSOHandleCookie() {

backend/internal/flow/flowexec/model.go

@@ -41,6 +41,9 @@ type frame struct {
 type EngineContext struct {
 	Context          context.Context
 	initiatorRequest *providers.InitiatorRequest
+	// currentRequest carries the HTTP request driving the current flow step. It is transient:
+	// supplied on each execution from the request context and never persisted with the flow context.
+	currentRequest *providers.InitiatorRequest
 
 	ExecutionID    string
 	FlowType       providers.FlowType
@@ -103,6 +106,16 @@ func (ec *EngineContext) SetInitiatorRequest(req *providers.InitiatorRequest) {
 	ec.initiatorRequest = req
 }
 
+// GetCurrentRequest returns the HTTP request that drives the current flow step.
+func (ec *EngineContext) GetCurrentRequest() *providers.InitiatorRequest {
+	return ec.currentRequest
+}
+
+// SetCurrentRequest sets the HTTP request that drives the current flow step.
+func (ec *EngineContext) SetCurrentRequest(req *providers.InitiatorRequest) {
+	ec.currentRequest = req
+}
+
 // mergeRuntimeData merges the given data into RuntimeData.
 func (ec *EngineContext) mergeRuntimeData(data map[string]string) {
 	if ec.RuntimeData == nil {