Merge extensions into body for webhook interceptors

With CEL interceptor writing to the `extensions` field, it is impossible to
chain added fields from a CEL interceptor to a webhook interceptor today. This
is because we do not pass the extensions field to the webhook interceptor yet.

This commit attempts to fix this by merging any extensions to the body before
sending it over to the Webhook interceptor. This is temporary until we fully
move webhook interceptors to the new pluggable interface. One implication of
this change is that the body may now contain an `extensions` field separate
from the top level extensions field if one uses the Webhook interceptor.

In addition, the CEL environment also did not have access to extensions which I
fixed.

Fixes #857

Signed-off-by: Dibyo Mukherjee <dibyo@google.com>

Add extensions support to CEL

docs/eventlisteners.md

@@ -34,6 +34,7 @@ using [Event Interceptors](#Interceptors).
     - [Bitbucket Interceptors](#bitbucket-interceptors)
     - [CEL Interceptors](#cel-interceptors)
       - [Overlays](#overlays)
+    - [Chaining Interceptors](#chaining-interceptors)
   - [EventListener Response](#eventlistener-response)
   - [How does the EventListener work?](#how-does-the-eventlistener-work)
   - [Examples](#examples)
@@ -675,6 +676,41 @@ spec:
     value: $(extensions.short_sha)
 ```
 
+### Chaining Interceptors
+
+This section documents the current behavior for passing data between interceptors. This will change as we fully implement #271.
+
+**CEL Interceptor:** Overlays from the CEL interceptor do not modify the event body. Instead they add fields to the top level `extensions` field.
+
+**Webhook Interceptors:** Webhook Interceptors *can* modify the event body currently. However, this will change in a future release when they will have to write to the extensions field like the CEL interceptor. Since the webhook interceptor does not have access to the top level `extensions` field, the EventListener will add the `extensions` field to the body before sending it to the webhook interceptor. As an example, consider the following interceptor chain:
+
+  ```yaml
+    interceptors:
+      - cel:
+          overlays:
+            - key: "truncated_sha"
+              expression: "body.sha.truncate(5)"
+      - webhook:
+        objectRef:
+          kind: Service
+          name: some-interceptor
+          apiVersion: v1
+      - cel:
+          filter: "body.extensions.truncated_sha == \"abcde\"" # Can also be extensions.truncated_sha == \"abcde\"
+  ```
+
+In the above snipped, the first CEL interceptor adds the `truncated_sha` field. To ensure the following webhook interceptor can use this field, the EventListener will add it to the body. So, the body received by the webhook interceptor will look as follows:
+
+```
+{
+  "sha": "abcdefghi", // Original field
+  "extensions": {
+     "truncated_sha": "abcde" 
+  }
+}
+```
+Assuming the webhook interceptor does not then modify the body, the last CEL interceptor (as well as any bindings) will have access to truncated_sha both via the body as well as via extensions i.e both `$(body.extensions.truncated_sha)` as well as `$(extensions.truncated_sha)`
+
 ## EventListener Response
 
 The EventListener responds with 201 Created status code when at least one of the trigger is executed successfully. Otherwise, it returns 202 Accepted status code.

pkg/interceptors/cel/cel.go

@@ -106,11 +106,12 @@ func makeCelEnv(ns string, k kubernetes.Interface) (*cel.Env, error) {
 		cel.Declarations(
 			decls.NewVar("body", mapStrDyn),
 			decls.NewVar("header", mapStrDyn),
+			decls.NewVar("extensions", mapStrDyn),
 			decls.NewVar("requestURL", decls.String),
 		))
 }
 
-func makeEvalContext(body []byte, h http.Header, url string) (map[string]interface{}, error) {
+func makeEvalContext(body []byte, h http.Header, url string, extensions map[string]interface{}) (map[string]interface{}, error) {
 	var jsonMap map[string]interface{}
 	err := json.Unmarshal(body, &jsonMap)
 	if err != nil {
@@ -120,6 +121,7 @@ func makeEvalContext(body []byte, h http.Header, url string) (map[string]interfa
 		"body":       jsonMap,
 		"header":     h,
 		"requestURL": url,
+		"extensions": extensions,
 	}, nil
 }
 
@@ -154,7 +156,7 @@ func (w *Interceptor) Process(ctx context.Context, r *triggersv1.InterceptorRequ
 		payload = r.Body
 	}
 
-	evalContext, err := makeEvalContext(payload, r.Header, r.Context.EventURL)
+	evalContext, err := makeEvalContext(payload, r.Header, r.Context.EventURL, r.Extensions)
 	if err != nil {
 		return &triggersv1.InterceptorResponse{
 			Continue: false,

pkg/interceptors/cel/cel_test.go

@@ -49,6 +49,7 @@ func TestInterceptor_Process(t *testing.T) {
 		name           string
 		CEL            *triggersv1.CELInterceptor
 		body           []byte
+		extensions     map[string]interface{}
 		wantExtensions map[string]interface{}
 	}{{
 		name: "simple body check with matching body",
@@ -229,6 +230,20 @@ func TestInterceptor_Process(t *testing.T) {
 				"other": "thing",
 			},
 		},
+	}, {
+		name: "filters and overlays can access passed in extensions",
+		CEL: &triggersv1.CELInterceptor{
+			Filter: `extensions.foo == "bar"`,
+			Overlays: []triggersv1.CELOverlay{
+				{Key: "one", Expression: "extensions.foo"},
+			},
+		},
+		extensions: map[string]interface{}{
+			"foo": "bar",
+		},
+		wantExtensions: map[string]interface{}{
+			"one": "bar",
+		},
 	},
 	}
 	for _, tt := range tests {
@@ -247,7 +262,7 @@ func TestInterceptor_Process(t *testing.T) {
 					"X-Test":         []string{"test-value"},
 					"X-Secret-Token": []string{"secrettoken"},
 				},
-				Extensions: nil,
+				Extensions: tt.extensions,
 				InterceptorParams: map[string]interface{}{
 					"filter":   tt.CEL.Filter,
 					"overlays": tt.CEL.Overlays,
@@ -710,7 +725,7 @@ func TestMakeEvalContextWithError(t *testing.T) {
 	req := httptest.NewRequest(http.MethodPost, "/", nil)
 	payload := []byte(`{"tes`)
 
-	_, err := makeEvalContext(payload, req.Header, req.URL.String())
+	_, err := makeEvalContext(payload, req.Header, req.URL.String(), map[string]interface{}{})
 
 	if !matchError(t, "failed to parse the body as JSON: unexpected end of JSON input", err) {
 		t.Fatalf("failed to match the error: %s", err)

pkg/sink/sink.go

@@ -36,6 +36,7 @@ import (
 	"github.com/tektoncd/triggers/pkg/interceptors/webhook"
 	"github.com/tektoncd/triggers/pkg/resources"
 	"github.com/tektoncd/triggers/pkg/template"
+	"github.com/tidwall/sjson"
 	"go.uber.org/zap"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -300,12 +301,17 @@ func (r Sink) ExecuteInterceptors(t triggersv1.Trigger, in *http.Request, event
 			// Clear interceptorParams for the next interceptor in chain
 			request.InterceptorParams = map[string]interface{}{}
 		} else {
-			// Old style interceptor (everything but CEL at the moment)
+			// Old style interceptors (Webhook)
+			// Merge any extensions into body to enable chaining behavior
+			body, err := extendBodyWithExtensions(request.Body, request.Extensions)
+			if err != nil {
+				return nil, nil, nil, fmt.Errorf("could not merge extensions with body: %w", err)
+			}
 			req := &http.Request{
 				Method: http.MethodPost,
 				Header: request.Header,
 				URL:    in.URL,
-				Body:   ioutil.NopCloser(bytes.NewBuffer(request.Body)),
+				Body:   ioutil.NopCloser(bytes.NewBuffer(body)),
 			}
 
 			res, err := interceptor.ExecuteTrigger(req)
@@ -355,3 +361,19 @@ func (r Sink) CreateResources(triggerNS, sa string, res []json.RawMessage, trigg
 	}
 	return nil
 }
+
+// extendBodyWithExtensions merges the extensions into the given body.
+func extendBodyWithExtensions(body []byte, extensions map[string]interface{}) ([]byte, error) {
+	for k, v := range extensions {
+		vb, err := json.Marshal(v)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal value to JSON: %w", err)
+		}
+		body, err = sjson.SetRawBytes(body, fmt.Sprintf("extensions.%s", k), vb)
+		if err != nil {
+			return nil, fmt.Errorf("failed to sjson extensions to body: %w", err)
+		}
+	}
+
+	return body, nil
+}

pkg/sink/sink_test.go

@@ -1053,6 +1053,118 @@ func TestExecuteInterceptor_NotContinue(t *testing.T) {
 	}
 }
 
+type echoInterceptor struct {
+	body map[string]interface{}
+}
+
+func (f *echoInterceptor) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	var data map[string]interface{}
+	if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = w.Write([]byte(err.Error()))
+		return
+	}
+	defer r.Body.Close()
+	f.body = data
+
+	if err := json.NewEncoder(w).Encode(data); err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = w.Write([]byte(err.Error()))
+	}
+}
+
+func TestExecuteInterceptor_ExtensionChaining(t *testing.T) {
+	echoServer := &echoInterceptor{}
+	srv := httptest.NewServer(echoServer)
+	defer srv.Close()
+	client := srv.Client()
+	// Redirect all requests to the fake server.
+	u, _ := url.Parse(srv.URL)
+	client.Transport = &http.Transport{
+		Proxy: http.ProxyURL(u),
+	}
+
+	logger := zaptest.NewLogger(t)
+
+	r := Sink{
+		HTTPClient: srv.Client(),
+		Logger:     logger.Sugar(),
+	}
+
+	webhook := &triggersv1.EventInterceptor{
+		Webhook: &triggersv1.WebhookInterceptor{
+			ObjectRef: &corev1.ObjectReference{
+				APIVersion: "v1",
+				Kind:       "Service",
+				Name:       "foo",
+			},
+		},
+	}
+	sha := "abcdefghi" // Fake "sha" to send via body
+	preWebhookCEL := &triggersv1.EventInterceptor{
+		CEL: &triggersv1.CELInterceptor{
+			Overlays: []triggersv1.CELOverlay{{
+				Key:        "truncated_sha",
+				Expression: "body.sha.truncate(5)",
+			}},
+		},
+	}
+	postWebhookCEL := &triggersv1.EventInterceptor{
+		CEL: &triggersv1.CELInterceptor{
+			Filter: "body.extensions.truncated_sha == \"abcde\" && extensions.truncated_sha == \"abcde\"",
+		},
+	}
+	trigger := triggersv1.Trigger{
+		Spec: triggersv1.TriggerSpec{
+			Interceptors: []*triggersv1.EventInterceptor{preWebhookCEL, webhook, postWebhookCEL}},
+	}
+
+	req, err := http.NewRequest("POST", "/", nil)
+	if err != nil {
+		t.Fatalf("http.NewRequest: %v", err)
+	}
+	body := fmt.Sprintf(`{"sha": "%s"}`, sha)
+	resp, _, iresp, err := r.ExecuteInterceptors(trigger, req, []byte(body), logger.Sugar(), eventID)
+	if err != nil {
+		t.Fatalf("executeInterceptors: %v", err)
+	}
+
+	wantBody := map[string]interface{}{
+		"sha": sha,
+		"extensions": map[string]interface{}{
+			"truncated_sha": "abcde",
+		},
+	}
+	var gotBody map[string]interface{}
+	if err := json.Unmarshal(resp, &gotBody); err != nil {
+		t.Fatalf("json.Unmarshal response body : %v\n Interceptor response is: %+v", err, iresp)
+	}
+
+	if diff := cmp.Diff(wantBody, gotBody); diff != "" {
+		t.Errorf("Body: -want +got: %s", diff)
+	}
+
+	// Check Interceptor got the body with extensions
+	if diff := cmp.Diff(wantBody, echoServer.body); diff != "" {
+		t.Errorf("Echo Interceptor did not get correct body: -want +got: %s", diff)
+	}
+
+	// Check that we forward the extension correctly to the last interceptor
+	if !iresp.Continue {
+		t.Errorf("Response.continue expected true but got false. Response: %v", iresp)
+	}
+
+	// Check we maintain the extensions outside the body as well
+	wantExtensions := map[string]interface{}{
+		"truncated_sha": "abcde",
+	}
+
+	if diff := cmp.Diff(iresp.Extensions, wantExtensions); diff != "" {
+		t.Errorf("Extensions: -want +got: %s", diff)
+	}
+
+}
+
 const userWithPermissions = "user-with-permissions"
 const userWithoutPermissions = "user-with-no-permissions"
 const userWithForbiddenAccess = "user-forbidden"
@@ -1449,3 +1561,67 @@ func makeCapturingLogger(t *testing.T) (*observer.ObservedLogs, *zap.SugaredLogg
 	l := zaptest.NewLogger(t, zaptest.WrapOptions(zap.WrapCore(func(zapcore.Core) zapcore.Core { return core }))).Sugar()
 	return logs, l
 }
+
+func TestExtendBodyWithExtensions(t *testing.T) {
+	tests := []struct {
+		name       string
+		body       []byte
+		extensions map[string]interface{}
+		want       map[string]interface{}
+	}{{
+		name: "merges all extensions to an extension field",
+		body: json.RawMessage(`{"sha": "abcdef"}`),
+		extensions: map[string]interface{}{
+			"added_field": "val1",
+			"nested": map[string]interface{}{
+				"field": "nestedVal",
+			},
+		},
+		want: map[string]interface{}{
+			"sha": "abcdef",
+			"extensions": map[string]interface{}{
+				"added_field": "val1",
+				"nested": map[string]interface{}{
+					"field": "nestedVal",
+				},
+			},
+		},
+	}, {
+		name: "body contains an extension already",
+		body: json.RawMessage(`{"sha": "abcdef", "extensions": {"foo": "bar"}}`),
+		extensions: map[string]interface{}{
+			"added_field": "val1",
+		},
+		want: map[string]interface{}{
+			"sha": "abcdef",
+			"extensions": map[string]interface{}{
+				"foo":         "bar",
+				"added_field": "val1",
+			},
+		},
+	}, {
+		name:       "no extensions",
+		body:       json.RawMessage(`{"sha": "abcdef"}`),
+		extensions: map[string]interface{}{},
+		want: map[string]interface{}{
+			"sha": "abcdef",
+		},
+	}}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := extendBodyWithExtensions(tc.body, tc.extensions)
+			if err != nil {
+				t.Fatalf("extendBodyWithExtensions() unexpected error: %v", err)
+			}
+			gotMap := map[string]interface{}{}
+			if err := json.Unmarshal(got, &gotMap); err != nil {
+				t.Fatalf("extendBodyWithExtensions() failed to unmarshal result: %v", err)
+			}
+			if diff := cmp.Diff(tc.want, gotMap); diff != "" {
+				t.Fatalf("extendBodyWithExtensions() diff -want/+got: %s", diff)
+			}
+		})
+	}
+
+}