fix for headers responses data (+directions) + haproxy needed changes (pointers)

Author: e-n-0

contrib/envoyproxy/go-control-plane/envoy.go

@@ -185,17 +185,21 @@ func (s *appsecEnvoyExternalProcessorServer) processMessage(ctx context.Context,
 	switch v := req.Request.(type) {
 	case *envoyextproc.ProcessingRequest_RequestHeaders:
 		var (
-			action message_processor.Action
-			err    error
+			action   message_processor.Action
+			err      error
+			reqState *message_processor.RequestState
 		)
-		*currentRequest, action, err = s.messageProcessor.OnRequestHeaders(ctx, &requestHeadersEnvoy{v, s.config.Integration})
+		reqState, action, err = s.messageProcessor.OnRequestHeaders(ctx, &requestHeadersEnvoy{v, s.config.Integration})
+		if err == nil {
+			*currentRequest = *reqState
+		}
 		return action, err
 
 	case *envoyextproc.ProcessingRequest_RequestBody:
 		if !currentRequest.Ongoing {
 			return message_processor.Action{}, status.Errorf(codes.InvalidArgument, "Received request body without request headers")
 		}
-		return s.messageProcessor.OnRequestBody(&requestBodyEnvoy{v}, *currentRequest)
+		return s.messageProcessor.OnRequestBody(&requestBodyEnvoy{v}, currentRequest)
 
 	case *envoyextproc.ProcessingRequest_ResponseHeaders:
 		if !currentRequest.Ongoing {
@@ -204,13 +208,13 @@ func (s *appsecEnvoyExternalProcessorServer) processMessage(ctx context.Context,
 				" and can happen when a malformed request is sent to Envoy where the header Host is missing. See link to issue https://github.com/envoyproxy/envoy/issues/38022")
 			return message_processor.Action{}, status.Errorf(codes.InvalidArgument, "Error processing response headers from ext_proc: can't process the response")
 		}
-		return s.messageProcessor.OnResponseHeaders(&responseHeadersEnvoy{v}, *currentRequest)
+		return s.messageProcessor.OnResponseHeaders(&responseHeadersEnvoy{v}, currentRequest)
 
 	case *envoyextproc.ProcessingRequest_ResponseBody:
 		if !currentRequest.Ongoing {
 			return message_processor.Action{}, status.Errorf(codes.InvalidArgument, "Received response body without request context")
 		}
-		return s.messageProcessor.OnResponseBody(&responseBodyEnvoy{v}, *currentRequest)
+		return s.messageProcessor.OnResponseBody(&responseBodyEnvoy{v}, currentRequest)
 
 	case *envoyextproc.ProcessingRequest_RequestTrailers:
 		instr.Logger().Debug("external_processing: received unexpected message of type RequestTrailers, ignoring it. " +

contrib/envoyproxy/go-control-plane/envoy_utils.go

@@ -49,26 +49,34 @@ func buildImmediateResponse(data *message_processor.BlockResponseData) *envoyext
 	}
 }
 
-// buildHeadersResponse creates an Envoy HeadersResponse from provided data answering a RequestHeaders message
+// buildHeadersResponse creates an Envoy HeadersResponse from provided data answering a RequestHeaders or ResponseHeaders message
 func buildHeadersResponse(data *message_processor.HeadersResponseData) *envoyextproc.ProcessingResponse {
 	var modeOverride *envoyextprocfilter.ProcessingMode
 	if data.RequestBody {
 		modeOverride = &envoyextprocfilter.ProcessingMode{RequestBodyMode: envoyextprocfilter.ProcessingMode_STREAMED}
 	}
 
-	return &envoyextproc.ProcessingResponse{
-		Response: &envoyextproc.ProcessingResponse_RequestHeaders{
-			RequestHeaders: &envoyextproc.HeadersResponse{
-				Response: &envoyextproc.CommonResponse{
-					Status: envoyextproc.CommonResponse_CONTINUE,
-					HeaderMutation: &envoyextproc.HeaderMutation{
-						SetHeaders: convertHeadersToEnvoy(data.HeaderMutation),
-					},
-				},
+	processingResponse := &envoyextproc.ProcessingResponse{ModeOverride: modeOverride}
+	headersResponse := &envoyextproc.HeadersResponse{
+		Response: &envoyextproc.CommonResponse{
+			Status: envoyextproc.CommonResponse_CONTINUE,
+			HeaderMutation: &envoyextproc.HeaderMutation{
+				SetHeaders: convertHeadersToEnvoy(data.HeaderMutation),
 			},
 		},
-		ModeOverride: modeOverride,
 	}
+
+	if data.Direction == message_processor.DirectionRequest {
+		processingResponse.Response = &envoyextproc.ProcessingResponse_RequestHeaders{
+			RequestHeaders: headersResponse,
+		}
+	} else {
+		processingResponse.Response = &envoyextproc.ProcessingResponse_ResponseHeaders{
+			ResponseHeaders: headersResponse,
+		}
+	}
+
+	return processingResponse
 }
 
 // convertHeadersToEnvoy converts standard HTTP headers to Envoy HeaderValueOption format

contrib/envoyproxy/go-control-plane/message_processor/action.go

@@ -20,10 +20,14 @@ func newContinueAction() Action {
 
 // newContinueActionWithResponseData creates an ActionTypeContinue with header mutations and requestBody flag.
 // The requestBody flag indicates whether the body should be requested from the proxy to the external processing service.
-func newContinueActionWithResponseData(mutations http.Header, requestBody bool) Action {
+func newContinueActionWithResponseData(mutations http.Header, requestBody bool, direction Direction) Action {
 	return Action{
-		Type:     ActionTypeContinue,
-		Response: &HeadersResponseData{HeaderMutation: mutations, RequestBody: requestBody},
+		Type: ActionTypeContinue,
+		Response: &HeadersResponseData{
+			HeaderMutation: mutations,
+			RequestBody:    requestBody,
+			Direction:      direction,
+		},
 	}
 }
 
@@ -46,10 +50,19 @@ func newFinishAction() Action {
 	}
 }
 
+// Direction indicates the direction of the message being processed.
+type Direction int
+
+const (
+	DirectionRequest  Direction = iota // DirectionRequest indicates a request message.
+	DirectionResponse                  // DirectionResponse indicates a response message.
+)
+
 // HeadersResponseData is the data for a headers response.
 type HeadersResponseData struct {
 	HeaderMutation http.Header
 	RequestBody    bool
+	Direction      Direction
 }
 
 // BlockResponseData is the data for a blocking response.

contrib/envoyproxy/go-control-plane/message_processor/message_processor.go

@@ -35,18 +35,17 @@ func NewMessageProcessor(config MessageProcessorConfig, instr *instrumentation.I
 }
 
 // OnRequestHeaders handles incoming request headers
-func (mp *MessageProcessor) OnRequestHeaders(ctx context.Context, req RequestHeaders) (RequestState, Action, error) {
+func (mp *MessageProcessor) OnRequestHeaders(ctx context.Context, req RequestHeaders) (*RequestState, Action, error) {
 	httpReq, err := req.NewRequest(ctx)
 	if err != nil {
-		return RequestState{}, Action{}, err
+		return nil, Action{}, err
 	}
 
 	componentName := req.Component(ctx)
 	framework := req.Framework()
 	reqState, blocked := newRequestState(httpReq, mp.instr, mp.config.BodyParsingSizeLimit, componentName, framework)
-	if reqState.span == nil {
-		reqState.Close()
-		return RequestState{}, Action{}, fmt.Errorf("error getting span from context")
+	if reqState == nil {
+		return nil, Action{}, fmt.Errorf("error getting span from context")
 	}
 
 	if !mp.config.BlockingUnavailable && blocked {
@@ -57,7 +56,7 @@ func (mp *MessageProcessor) OnRequestHeaders(ctx context.Context, req RequestHea
 	headerMutation, err := reqState.PropagationHeaders()
 	if err != nil {
 		reqState.Close()
-		return RequestState{}, Action{}, err
+		return nil, Action{}, err
 	}
 
 	// Determine if we instruct the proxy to send the body
@@ -68,11 +67,11 @@ func (mp *MessageProcessor) OnRequestHeaders(ctx context.Context, req RequestHea
 		// Todo: Set telemetry body size (using content-length)
 	}
 
-	return reqState, newContinueActionWithResponseData(headerMutation, requestBody), nil
+	return reqState, newContinueActionWithResponseData(headerMutation, requestBody, DirectionRequest), nil
 }
 
 // OnRequestBody handles incoming request body chunks
-func (mp *MessageProcessor) OnRequestBody(req RequestBody, reqState RequestState) (Action, error) {
+func (mp *MessageProcessor) OnRequestBody(req RequestBody, reqState *RequestState) (Action, error) {
 	if mp.config.BodyParsingSizeLimit <= 0 || !reqState.AwaitingRequestBody {
 		mp.instr.Logger().Error("message_processor: the body parsing has been wrongly configured. " +
 			"Please refer to the official documentation for guidance on the proper settings or contact support.")
@@ -89,7 +88,7 @@ func (mp *MessageProcessor) OnRequestBody(req RequestBody, reqState RequestState
 }
 
 // OnResponseHeaders handles incoming response headers
-func (mp *MessageProcessor) OnResponseHeaders(res ResponseHeaders, reqState RequestState) (Action, error) {
+func (mp *MessageProcessor) OnResponseHeaders(res ResponseHeaders, reqState *RequestState) (Action, error) {
 	mp.instr.Logger().Debug("message_processor: received response headers")
 	reqState.AwaitingRequestBody = false
 
@@ -128,11 +127,11 @@ func (mp *MessageProcessor) OnResponseHeaders(res ResponseHeaders, reqState Requ
 
 	// Todo: Set telemetry body size (using content-length)
 
-	return newContinueActionWithResponseData(nil, true), nil
+	return newContinueActionWithResponseData(nil, true, DirectionResponse), nil
 }
 
 // OnResponseBody handles incoming response body chunks
-func (mp *MessageProcessor) OnResponseBody(res ResponseBody, reqState RequestState) (Action, error) {
+func (mp *MessageProcessor) OnResponseBody(res ResponseBody, reqState *RequestState) (Action, error) {
 	mp.instr.Logger().Debug("message_processor: received response body: %v - EOS: %v\n", len(res.Body()), res.EndOfStream())
 
 	if mp.config.BodyParsingSizeLimit <= 0 || !reqState.AwaitingResponseBody {

contrib/envoyproxy/go-control-plane/message_processor/request_state.go

@@ -22,7 +22,7 @@ var _ io.Closer = (*RequestState)(nil)
 // RequestState manages the state of a single request through its lifecycle
 type RequestState struct {
 	ctx         context.Context
-	span        *tracer.Span
+	Span        *tracer.Span
 	afterHandle func()
 
 	instr *instrumentation.Instrumentation
@@ -43,7 +43,7 @@ type RequestState struct {
 }
 
 // newRequestState creates a new request state
-func newRequestState(request *http.Request, instr *instrumentation.Instrumentation, bodyLimit int, componentName string, framework string) (RequestState, bool) {
+func newRequestState(request *http.Request, instr *instrumentation.Instrumentation, bodyLimit int, componentName string, framework string) (*RequestState, bool) {
 	fakeResponseWriter := newFakeResponseWriter()
 	wrappedResponseWriter, spanRequest, afterHandle, blocked := httptrace.BeforeHandle(&httptrace.ServeConfig{
 		Framework: framework,
@@ -56,7 +56,7 @@ func newRequestState(request *http.Request, instr *instrumentation.Instrumentati
 
 	span, ok := tracer.SpanFromContext(spanRequest.Context())
 	if !ok {
-		return RequestState{}, false
+		return nil, false
 	}
 
 	var requestBuffer *bodyBuffer
@@ -69,9 +69,9 @@ func newRequestState(request *http.Request, instr *instrumentation.Instrumentati
 		responseBuffer = newBodyBuffer(bodyLimit)
 	}
 
-	return RequestState{
+	return &RequestState{
 		ctx:                   spanRequest.Context(),
-		span:                  span,
+		Span:                  span,
 		afterHandle:           afterHandle,
 		instr:                 instr,
 		fakeResponseWriter:    fakeResponseWriter,
@@ -85,7 +85,7 @@ func newRequestState(request *http.Request, instr *instrumentation.Instrumentati
 // PropagationHeaders creates header mutations for trace propagation
 func (rs *RequestState) PropagationHeaders() (http.Header, error) {
 	newHeaders := make(http.Header)
-	if err := tracer.Inject(rs.span.Context(), tracer.HTTPHeadersCarrier(newHeaders)); err != nil {
+	if err := tracer.Inject(rs.Span.Context(), tracer.HTTPHeadersCarrier(newHeaders)); err != nil {
 		return nil, err
 	}