Adds annotation to disable payload validation in eventlistener

This adds an annotation support tekton.dev/payload-validation
for eventlistener. If it is added to el with value as false then
the payload from events will not be validated and will be directly
passed to interceptors.
By default the payload validation is enabled. Only if annotation
is defined and its value is false then it will be disabled
for that particular el.

Signed-off-by: Shivam Mukhade [email protected]

cmd/eventlistenersink/main.go

@@ -149,6 +149,7 @@ func main() {
 		HTTPClient:             http.DefaultClient,
 		EventListenerName:      sinkArgs.ElName,
 		EventListenerNamespace: sinkArgs.ElNamespace,
+		PayloadValidation:      sinkArgs.PayloadValidation,
 		Logger:                 logger,
 		Recorder:               recorder,
 		Auth:                   sink.DefaultAuthOverride{},

pkg/apis/triggers/v1alpha1/event_listener_validation.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/tektoncd/triggers/pkg/apis/triggers"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
@@ -43,6 +44,11 @@ func (e *EventListener) Validate(ctx context.Context) *apis.FieldError {
 		// Since `el-` is added as the prefix of EventListener services, the name of EventListener must be no more than 60 characters long.
 		errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("eventListener name '%s' must be no more than 60 characters long", e.ObjectMeta.Name), "metadata.name"))
 	}
+
+	if len(e.ObjectMeta.Annotations) != 0 {
+		errs = errs.Also(triggers.ValidateAnnotations(e.ObjectMeta.Annotations))
+	}
+
 	if apis.IsInDelete(ctx) {
 		return nil
 	}

pkg/apis/triggers/v1alpha1/event_listener_validation_test.go

@@ -22,6 +22,7 @@ import (
 
 	pipelinev1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	"github.com/tektoncd/triggers/pkg/apis/triggers"
 	"github.com/tektoncd/triggers/pkg/apis/triggers/v1alpha1"
 	"github.com/tektoncd/triggers/test"
 	corev1 "k8s.io/api/core/v1"
@@ -102,6 +103,20 @@ func Test_EventListenerValidate(t *testing.T) {
 				}},
 			},
 		},
+	}, {
+		name: "Valid EventListener with Annotation",
+		el: &v1alpha1.EventListener{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "name",
+				Namespace:   "namespace",
+				Annotations: map[string]string{triggers.PayloadValidationAnnotation: "true"},
+			},
+			Spec: v1alpha1.EventListenerSpec{
+				Triggers: []v1alpha1.EventListenerTrigger{{
+					TriggerRef: "tt",
+				}},
+			},
+		},
 	}, {
 		name: "Valid EventListener with TriggerBinding",
 		el: &v1alpha1.EventListener{
@@ -557,6 +572,20 @@ func TestEventListenerValidate_error(t *testing.T) {
 				}},
 			},
 		},
+	}, {
+		name: "Invalid Annotation value",
+		el: &v1alpha1.EventListener{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "name",
+				Namespace:   "namespace",
+				Annotations: map[string]string{triggers.PayloadValidationAnnotation: "xyz"},
+			},
+			Spec: v1alpha1.EventListenerSpec{
+				Triggers: []v1alpha1.EventListenerTrigger{{
+					TriggerRef: "tt",
+				}},
+			},
+		},
 	}, {
 		name: "TriggerBinding with no ref or spec",
 		el: &v1alpha1.EventListener{

pkg/apis/triggers/v1beta1/event_listener_validation.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/tektoncd/triggers/pkg/apis/triggers"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
@@ -43,6 +44,11 @@ func (e *EventListener) Validate(ctx context.Context) *apis.FieldError {
 		// Since `el-` is added as the prefix of EventListener services, the name of EventListener must be no more than 60 characters long.
 		errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("eventListener name '%s' must be no more than 60 characters long", e.ObjectMeta.Name), "metadata.name"))
 	}
+
+	if len(e.GetObjectMeta().GetAnnotations()) != 0 {
+		errs = errs.Also(triggers.ValidateAnnotations(e.GetObjectMeta().GetAnnotations()))
+	}
+
 	if apis.IsInDelete(ctx) {
 		return nil
 	}

pkg/apis/triggers/v1beta1/event_listener_validation_test.go

@@ -22,6 +22,7 @@ import (
 
 	pipelinev1alpha1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
 	pipelinev1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	"github.com/tektoncd/triggers/pkg/apis/triggers"
 	triggersv1beta1 "github.com/tektoncd/triggers/pkg/apis/triggers/v1beta1"
 	"github.com/tektoncd/triggers/test"
 	corev1 "k8s.io/api/core/v1"
@@ -102,6 +103,20 @@ func Test_EventListenerValidate(t *testing.T) {
 				}},
 			},
 		},
+	}, {
+		name: "Valid EventListener with Annotation",
+		el: &triggersv1beta1.EventListener{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "name",
+				Namespace:   "namespace",
+				Annotations: map[string]string{triggers.PayloadValidationAnnotation: "true"},
+			},
+			Spec: triggersv1beta1.EventListenerSpec{
+				Triggers: []triggersv1beta1.EventListenerTrigger{{
+					TriggerRef: "tt",
+				}},
+			},
+		},
 	}, {
 		name: "Valid EventListener with TriggerBinding",
 		el: &triggersv1beta1.EventListener{
@@ -557,6 +572,20 @@ func TestEventListenerValidate_error(t *testing.T) {
 				}},
 			},
 		},
+	}, {
+		name: "Invalid Annotation value",
+		el: &triggersv1beta1.EventListener{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "name",
+				Namespace:   "namespace",
+				Annotations: map[string]string{triggers.PayloadValidationAnnotation: "xyz"},
+			},
+			Spec: triggersv1beta1.EventListenerSpec{
+				Triggers: []triggersv1beta1.EventListenerTrigger{{
+					TriggerRef: "tt",
+				}},
+			},
+		},
 	}, {
 		name: "TriggerBinding with no ref or spec",
 		el: &triggersv1beta1.EventListener{

pkg/apis/triggers/validation.go

@@ -0,0 +1,39 @@
+/*
+Copyright 2021 The Tekton Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package triggers
+
+import (
+	"fmt"
+
+	"knative.dev/pkg/apis"
+)
+
+const (
+	PayloadValidationAnnotation = "tekton.dev/payload-validation"
+)
+
+func ValidateAnnotations(annotations map[string]string) *apis.FieldError {
+	var errs *apis.FieldError
+
+	if value, ok := annotations[PayloadValidationAnnotation]; ok {
+		if value != "true" && value != "false" {
+			errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("%s annotation must have value 'true' or 'false'", PayloadValidationAnnotation), "metadata.annotations"))
+		}
+	}
+
+	return errs
+}

pkg/apis/triggers/validation_test.go

@@ -0,0 +1,37 @@
+/*
+Copyright 2021 The Tekton Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package triggers
+
+import (
+	"testing"
+)
+
+func Test_PayloadValidationAnnotation_Valid(t *testing.T) {
+	annotations := map[string]string{PayloadValidationAnnotation: "false"}
+	err := ValidateAnnotations(annotations)
+	if err != nil {
+		t.Errorf("Unexpected Error: %v", err)
+	}
+}
+
+func Test_PayloadValidationAnnotation_InvalidValue(t *testing.T) {
+	annotations := map[string]string{PayloadValidationAnnotation: "abc"}
+	err := ValidateAnnotations(annotations)
+	if err == nil {
+		t.Errorf("Expected Error but got nil")
+	}
+}

pkg/reconciler/eventlistener/eventlistener.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/tektoncd/triggers/pkg/apis/triggers"
 	"github.com/tektoncd/triggers/pkg/apis/triggers/contexts"
 	"github.com/tektoncd/triggers/pkg/apis/triggers/v1beta1"
 	triggersclientset "github.com/tektoncd/triggers/pkg/client/clientset/versioned"
@@ -883,6 +884,13 @@ func getContainer(el *v1beta1.EventListener, c Config, pod *duckv1.WithPod) core
 		isMultiNS = true
 	}
 
+	payloadValidation := true
+	if value, ok := el.GetAnnotations()[triggers.PayloadValidationAnnotation]; ok {
+		if value == "false" {
+			payloadValidation = false
+		}
+	}
+
 	return corev1.Container{
 		Name:  "event-listener",
 		Image: *c.Image,
@@ -900,6 +908,7 @@ func getContainer(el *v1beta1.EventListener, c Config, pod *duckv1.WithPod) core
 			"--idletimeout=" + strconv.FormatInt(*c.IdleTimeOut, 10),
 			"--timeouthandler=" + strconv.FormatInt(*c.TimeOutHandler, 10),
 			"--is-multi-ns=" + strconv.FormatBool(isMultiNS),
+			"--payload-validation=" + strconv.FormatBool(payloadValidation),
 		},
 		Env: env,
 	}

pkg/sink/initialization.go

@@ -57,6 +57,8 @@ var (
 		"The filename for the TLS certificate.")
 	tlsKeyFlag = flag.String("tls-key", "",
 		"The filename for the TLS key.")
+	payloadValidation = flag.Bool("payload-validation", true,
+		"Whether to disable payload validation or not.")
 )
 
 // Args define the arguments for Sink.
@@ -81,6 +83,8 @@ type Args struct {
 	Key string
 	// Cert defines the filename for tls Cert.
 	Cert string
+	// PayloadValidation defines whether to validate payload or not
+	PayloadValidation bool
 }
 
 // Clients define the set of client dependencies Sink requires.
@@ -104,16 +108,17 @@ func GetArgs() (Args, error) {
 	}
 
 	return Args{
-		ElName:           *nameFlag,
-		ElNamespace:      *namespaceFlag,
-		Port:             *portFlag,
-		IsMultiNS:        *isMultiNSFlag,
-		ELReadTimeOut:    time.Duration(*elReadTimeOut),
-		ELWriteTimeOut:   time.Duration(*elWriteTimeOut),
-		ELIdleTimeOut:    time.Duration(*elIdleTimeOut),
-		ELTimeOutHandler: time.Duration(*elTimeOutHandler),
-		Cert:             *tlsCertFlag,
-		Key:              *tlsKeyFlag,
+		ElName:            *nameFlag,
+		ElNamespace:       *namespaceFlag,
+		Port:              *portFlag,
+		IsMultiNS:         *isMultiNSFlag,
+		PayloadValidation: *payloadValidation,
+		ELReadTimeOut:     time.Duration(*elReadTimeOut),
+		ELWriteTimeOut:    time.Duration(*elWriteTimeOut),
+		ELIdleTimeOut:     time.Duration(*elIdleTimeOut),
+		ELTimeOutHandler:  time.Duration(*elTimeOutHandler),
+		Cert:              *tlsCertFlag,
+		Key:               *tlsKeyFlag,
 	}, nil
 }
 

pkg/sink/initialization_test.go

@@ -52,6 +52,9 @@ func Test_GetArgs(t *testing.T) {
 	if sinkArgs.IsMultiNS != true {
 		t.Errorf("Error EL Type want type, got %t", sinkArgs.IsMultiNS)
 	}
+	if sinkArgs.PayloadValidation != true {
+		t.Errorf("Error EL PayloadValidation want true, got %t", sinkArgs.PayloadValidation)
+	}
 }
 
 func Test_GetArgs_error(t *testing.T) {

pkg/sink/sink.go

@@ -57,7 +57,7 @@ type Sink struct {
 	Logger                 *zap.SugaredLogger
 	Recorder               *Recorder
 	Auth                   AuthOverride
-
+	PayloadValidation      bool
 	// WGProcessTriggers keeps track of triggers currently being processed
 	// Currently only used in tests to wait for all triggers to finish processing
 	WGProcessTriggers *sync.WaitGroup

pkg/sink/sink_test.go

@@ -156,6 +156,7 @@ func getSinkAssets(t *testing.T, resources test.Resources, elName string, webhoo
 		ClusterTriggerBindingLister: clustertriggerbindinginformer.Get(ctx).Lister(),
 		TriggerTemplateLister:       triggertemplateinformer.Get(ctx).Lister(),
 		ClusterInterceptorLister:    interceptorinformer.Get(ctx).Lister(),
+		PayloadValidation:           true,
 	}
 	return r, dynamicClient
 }

pkg/sink/validate_payload.go

@@ -34,22 +34,24 @@ func (r Sink) IsValidPayload(eventHandler http.Handler) http.Handler {
 			response.WriteHeader(http.StatusInternalServerError)
 			return
 		}
-		var event map[string]interface{}
-		if err := json.Unmarshal([]byte(payload), &event); err != nil {
-			errMsg := fmt.Sprintf("Invalid event body format format: %s", err)
-			r.recordCountMetrics(failTag)
-			r.Logger.Error(errMsg)
-			response.WriteHeader(http.StatusBadRequest)
-			response.Header().Set("Content-Type", "application/json")
-			body := Response{
-				EventListener: r.EventListenerName,
-				Namespace:     r.EventListenerNamespace,
-				ErrorMessage:  errMsg,
-			}
-			if err := json.NewEncoder(response).Encode(body); err != nil {
-				r.Logger.Errorf("failed to write back sink response: %v", err)
+		if r.PayloadValidation {
+			var event map[string]interface{}
+			if err := json.Unmarshal([]byte(payload), &event); err != nil {
+				errMsg := fmt.Sprintf("Invalid event body format format: %s", err)
+				r.recordCountMetrics(failTag)
+				r.Logger.Error(errMsg)
+				response.WriteHeader(http.StatusBadRequest)
+				response.Header().Set("Content-Type", "application/json")
+				body := Response{
+					EventListener: r.EventListenerName,
+					Namespace:     r.EventListenerNamespace,
+					ErrorMessage:  errMsg,
+				}
+				if err := json.NewEncoder(response).Encode(body); err != nil {
+					r.Logger.Errorf("failed to write back sink response: %v", err)
+				}
+				return
 			}
-			return
 		}
 		eventHandler.ServeHTTP(response, request)
 	})