Fix: avoids double interruption (envoyproxy#126)

Co-authored-by: Stefan Schlesinger <sts@ono.at>
Co-authored-by: José Carlos Chávez <jcchavezs@gmail.com>

e2e/docker-compose.yml

@@ -4,7 +4,7 @@ services:
   envoy:
     depends_on:
       - httpbin
-    image: envoyproxy/envoy:v1.23-latest
+    image: ${ENVOY_IMAGE:-envoyproxy/envoy:v1.23-latest}
     command:
       - -c
       - /conf/envoy-config.yaml

e2e/e2e-example.sh

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 ENVOY_HOST=${ENVOY_HOST:-"localhost:8080"}
 HTTPBIN_HOST=${HTTPBIN_HOST:-"localhost:8081"}
+TIMEOUT_SECS=${TIMEOUT_SECS:-5}
 
 [[ "${DEBUG}" == "true" ]] && set -x
 
@@ -47,7 +48,7 @@ function check_status() {
     local url=${1}
     local status=${2}
     local args=("${@:3}" --write-out '%{http_code}' --silent --output /dev/null)
-    status_code=$(curl "${args[@]}" "${url}")
+    status_code=$(curl --max-time ${TIMEOUT_SECS} "${args[@]}" "${url}")
     if [[ "${status_code}" -ne ${status} ]] ; then
       echo "[Fail] Unexpected response with code ${status_code} from ${url}"
       exit 1
@@ -64,7 +65,7 @@ function check_body() {
     local url=${1}
     local empty=${2}
     local args=("${@:3}" --silent)
-    response_body=$(curl "${args[@]}" "${url}")
+    response_body=$(curl --max-time ${TIMEOUT_SECS} "${args[@]}" "${url}")
     if [[ "${empty}" == "true" ]] && [[ -n "${response_body}" ]]; then
       echo -e "[Fail] Unexpected response with a body. Body dump:\n${response_body}"
       exit 1
@@ -94,6 +95,10 @@ wait_for_service "${envoy_url_echo}?arg=arg_1" 20
 ((step+=1))
 echo "[${step}/${total_steps}] (onRequestheaders) Testing true positive custom rule"
 check_status "${envoy_url_filtered}" 403
+# This test ensures the response body is empty on interruption. Specifically this makes
+# sure no body is returned although actionContinue is passed in phase 3 & 4.
+# See https://github.com/corazawaf/coraza-proxy-wasm/pull/126
+check_body "${envoy_url_filtered}" true
 
 # Testing body true negative
 ((step+=1))

example/docker-compose.yml

@@ -3,6 +3,9 @@ services:
     image: mccutchen/go-httpbin:v2.5.0
     environment:
       - MAX_BODY_SIZE=15728640 # 15 MiB
+    ports:
+      - 8081:8080
+
   chown:
     image: alpine:3.16
     command:
@@ -11,11 +14,12 @@ services:
       - chown -R 101:101 /home/envoy/logs
     volumes:
       - logs:/home/envoy/logs:rw
+
   envoy:
     depends_on:
       - chown
       - httpbin
-    image: envoyproxy/envoy:v1.23-latest
+    image: ${ENVOY_IMAGE:-envoyproxy/envoy:v1.23-latest}
     command:
       - -c
       - /conf/envoy-config.yaml

ftw/docker-compose.yml

@@ -13,7 +13,7 @@ services:
     depends_on:
       - chown
       - httpbin
-    image: envoyproxy/envoy:v1.23-latest
+    image: ${ENVOY_IMAGE:-envoyproxy/envoy:v1.23-latest}
     command:
       - -c
       - ${ENVOY_CONFIG:-/conf/envoy-config.yaml}

magefiles/magefile.go

@@ -237,6 +237,7 @@ func Ftw() error {
 	}()
 	env := map[string]string{
 		"FTW_CLOUDMODE": os.Getenv("FTW_CLOUDMODE"),
+		"ENVOY_IMAGE":   os.Getenv("ENVOY_IMAGE"),
 	}
 	if os.Getenv("ENVOY_NOWASM") == "true" {
 		env["ENVOY_CONFIG"] = "/conf/envoy-config-nowasm.yaml"
@@ -250,7 +251,7 @@ func Ftw() error {
 
 // RunExample spins up the test environment, access at http://localhost:8080. Requires docker-compose.
 func RunExample() error {
-	return sh.RunV("docker-compose", "--file", "example/docker-compose.yml", "up", "-d", "envoy-logs")
+	return sh.RunWithV(map[string]string{"ENVOY_IMAGE": os.Getenv("ENVOY_IMAGE")}, "docker-compose", "--file", "example/docker-compose.yml", "up", "-d", "envoy-logs")
 }
 
 // TeardownExample tears down the test environment. Requires docker-compose.

wasmplugin/plugin.go

@@ -102,11 +102,13 @@ type httpContext struct {
 	requestBodySize       int
 	responseBodySize      int
 	metrics               *wafMetrics
+	interruptionHandled   bool
 }
 
 // Override types.DefaultHttpContext.
 func (ctx *httpContext) OnHttpRequestHeaders(numHeaders int, endOfStream bool) types.Action {
 	defer logTime("OnHttpRequestHeaders", currentTime())
+
 	ctx.metrics.CountTX()
 	tx := ctx.tx
 
@@ -174,6 +176,12 @@ func (ctx *httpContext) OnHttpRequestHeaders(numHeaders int, endOfStream bool) t
 
 func (ctx *httpContext) OnHttpRequestBody(bodySize int, endOfStream bool) types.Action {
 	defer logTime("OnHttpRequestBody", currentTime())
+
+	if ctx.interruptionHandled {
+		proxywasm.LogErrorf("interruption already handled")
+		return types.ActionPause
+	}
+
 	tx := ctx.tx
 
 	if tx.IsRuleEngineOff() {
@@ -222,6 +230,22 @@ func (ctx *httpContext) OnHttpRequestBody(bodySize int, endOfStream bool) types.
 
 func (ctx *httpContext) OnHttpResponseHeaders(numHeaders int, endOfStream bool) types.Action {
 	defer logTime("OnHttpResponseHeaders", currentTime())
+
+	if ctx.interruptionHandled {
+		// Handling the interruption (see handleInterruption) generates a HttpResponse with the required status code.
+		// If handleInterruption is raised during OnHttpRequestHeaders or OnHttpRequestBody, the crafted response is sent
+		// downstream via the filter chain, therefore OnHttpResponseHeaders is called.
+		// We expect a response that is ending the stream, with exactly one header (:status) and no body.
+		// See https://github.com/corazawaf/coraza-proxy-wasm/pull/126
+		if numHeaders == 1 && endOfStream {
+			proxywasm.LogDebugf("interruption already handled, sending downstream the local response")
+			return types.ActionContinue
+		} else {
+			proxywasm.LogErrorf("interruption already handled, unexpected local response")
+			return types.ActionPause
+		}
+	}
+
 	tx := ctx.tx
 
 	if tx.IsRuleEngineOff() {
@@ -272,6 +296,13 @@ func (ctx *httpContext) OnHttpResponseHeaders(numHeaders int, endOfStream bool)
 
 func (ctx *httpContext) OnHttpResponseBody(bodySize int, endOfStream bool) types.Action {
 	defer logTime("OnHttpResponseBody", currentTime())
+
+	if ctx.interruptionHandled {
+		// Sending the crafted HttpResponse with empty body, we don't expect to trigger OnHttpResponseBody
+		proxywasm.LogErrorf("interruption already handled")
+		return types.ActionPause
+	}
+
 	tx := ctx.tx
 
 	if tx.IsRuleEngineOff() {
@@ -357,19 +388,28 @@ func (ctx *httpContext) OnHttpStreamDone() {
 	logMemStats()
 }
 
+const noGRPCStream int32 = -1
+
 func (ctx *httpContext) handleInterruption(phase string, interruption *ctypes.Interruption) types.Action {
+	if ctx.interruptionHandled {
+		// handleInterruption should never be called more the once
+		panic("interruption already handled")
+	}
+
 	ctx.metrics.CountTXInterruption(phase, interruption.RuleID)
 
-	proxywasm.LogInfof("%d interrupted, action %q", ctx.contextID, interruption.Action)
+	proxywasm.LogInfof("%d interrupted, action %q, phase %q", ctx.contextID, interruption.Action, phase)
 	statusCode := interruption.Status
 	if statusCode == 0 {
 		statusCode = 403
 	}
-
-	if err := proxywasm.SendHttpResponse(uint32(statusCode), nil, nil, -1); err != nil {
+	if err := proxywasm.SendHttpResponse(uint32(statusCode), nil, nil, noGRPCStream); err != nil {
 		panic(err)
 	}
 
+	ctx.interruptionHandled = true
+
+	// SendHttpResponse must be followed by ActionPause in order to stop malicious content
 	return types.ActionPause
 }