feat: get response body (#95)

Author: spacewander

README.md

@@ -144,6 +144,28 @@ end
 assert(wasm.on_http_response_headers(ctx))
 ```
 
+### on_http_response_body
+
+`syntax: ok, err = proxy_wasm.on_http_response_body(plugin_ctx)`
+
+Run the HTTP response body filter in the plugin of the given plugin ctx.
+This method need to be called in `body_filter_by_lua` phase and may be run
+multiple times.
+
+```lua
+local plugin, err = proxy_wasm.load("plugin","t/testdata/http_lifecycle/main.go.wasm")
+if not plugin then
+    ngx.log(ngx.ERR, "failed to load wasm ", err)
+    return
+end
+local ctx, err = wasm.on_configure(plugin, '{"body":512}')
+if not ctx then
+    ngx.log(ngx.ERR, "failed to create plugin ctx ", err)
+    return
+end
+assert(wasm.on_http_response_body(ctx))
+```
+
 ## proxy-wasm ABI
 
 Implemented proxy-wasm ABI can be found in [proxy_wasm_abi](./proxy_wasm_abi.md).

lib/resty/proxy-wasm.lua

@@ -68,7 +68,7 @@ local _M = {}
 local HTTP_REQUEST_HEADERS = 1
 local HTTP_REQUEST_BODY = 2
 local HTTP_RESPONSE_HEADERS = 4
---local HTTP_RESPONSE_BODY = 8
+local HTTP_RESPONSE_BODY = 8
 
 local RC_NEED_HTTP_CALL = 1
 
@@ -348,4 +348,26 @@ function _M.on_http_response_headers(plugin_ctx)
 end
 
 
+function _M.on_http_response_body(plugin_ctx)
+    if type(plugin_ctx) ~= "cdata" then
+        return nil, "bad plugin ctx"
+    end
+
+    local r = get_request()
+    if not r then
+        return nil, "bad request"
+    end
+
+    -- TODO: rewrite this by exporting FFI interfaces in OpenResty to save a copy
+    local body = ngx.arg[1]
+    local eof = ngx.arg[2]
+    local rc = C.ngx_http_wasm_on_http(plugin_ctx, r, HTTP_RESPONSE_BODY, body, #body, eof)
+    if rc < 0 then
+        return nil, "failed to run proxy_on_http_response_body"
+    end
+
+    return true
+end
+
+
 return _M

src/http/ngx_http_wasm_api.c

@@ -487,6 +487,7 @@ proxy_get_buffer_bytes(int32_t type, int32_t start, int32_t size,
         break;
 
     case PROXY_BUFFER_TYPE_HTTP_REQUEST_BODY:
+    case PROXY_BUFFER_TYPE_HTTP_RESPONSE_BODY:
         buffer = ngx_http_wasm_get_body();
         if (buffer != NULL) {
             data = buffer->data;

src/http/ngx_http_wasm_module.c

@@ -50,6 +50,8 @@ static ngx_str_t proxy_on_request_body =
     ngx_string("proxy_on_request_body");
 static ngx_str_t proxy_on_response_headers =
     ngx_string("proxy_on_response_headers");
+static ngx_str_t proxy_on_response_body =
+    ngx_string("proxy_on_response_body");
 static ngx_str_t proxy_on_http_call_response =
     ngx_string("proxy_on_http_call_response");
 
@@ -678,12 +680,14 @@ ngx_http_wasm_on_http(ngx_http_wasm_plugin_ctx_t *hwp_ctx, ngx_http_request_t *r
 
     ctx = ngx_http_wasm_get_module_ctx(r);
 
-    if (type == HTTP_RESPONSE_HEADERS) {
-        cb_name = &proxy_on_response_headers;
+    if (type == HTTP_REQUEST_HEADERS) {
+        cb_name = &proxy_on_request_headers;
     } else if (type == HTTP_REQUEST_BODY) {
         cb_name = &proxy_on_request_body;
+    } else if (type == HTTP_RESPONSE_HEADERS) {
+        cb_name = &proxy_on_response_headers;
     } else {
-        cb_name = &proxy_on_request_headers;
+        cb_name = &proxy_on_response_body;
     }
 
     if (type == HTTP_REQUEST_HEADERS || type == HTTP_RESPONSE_HEADERS) {

t/http_resp_body.t

@@ -0,0 +1,178 @@
+# Copyright 2022 Shenzhen ZhiLiu Technology Co., Ltd.
+#
+# 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.
+#
+use t::WASM 'no_plan';
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: sanity
+--- config
+location /t {
+    content_by_lua_block {
+        local wasm = require("resty.proxy-wasm")
+        local plugin = assert(wasm.load("plugin", "t/testdata/http_body/main.go.wasm"))
+        ngx.ctx.ctx = assert(wasm.on_configure(plugin, '{}'))
+
+        ngx.print("1234")
+        ngx.print("12345")
+    }
+    body_filter_by_lua_block {
+        local wasm = require("resty.proxy-wasm")
+        ngx.log(ngx.WARN, ngx.arg[1], " ", ngx.arg[2])
+        local ctx = ngx.ctx.ctx
+        assert(wasm.on_http_response_body(ctx))
+    }
+}
+--- grep_error_log eval
+qr/body size \d+, end of stream \w+/
+--- grep_error_log_out
+body size 4, end of stream false
+body size 5, end of stream false
+body size 0, end of stream true
+
+
+
+=== TEST 2: flush
+--- config
+location /t {
+    content_by_lua_block {
+        local wasm = require("resty.proxy-wasm")
+        local plugin = assert(wasm.load("plugin", "t/testdata/http_body/main.go.wasm"))
+        ngx.ctx.ctx = assert(wasm.on_configure(plugin, '{}'))
+
+        ngx.print("1234")
+        ngx.flush()
+        ngx.print("12345")
+    }
+    body_filter_by_lua_block {
+        local wasm = require("resty.proxy-wasm")
+        local ctx = ngx.ctx.ctx
+        assert(wasm.on_http_response_body(ctx))
+    }
+}
+--- grep_error_log eval
+qr/body size \d+, end of stream \w+/
+--- grep_error_log_out
+body size 4, end of stream false
+body size 0, end of stream false
+body size 5, end of stream false
+body size 0, end of stream true
+
+
+
+=== TEST 3: no body
+--- config
+location /t {
+    content_by_lua_block {
+        local wasm = require("resty.proxy-wasm")
+        local plugin = assert(wasm.load("plugin", "t/testdata/http_body/main.go.wasm"))
+        ngx.ctx.ctx = assert(wasm.on_configure(plugin, '{"action":"offset"}'))
+    }
+    body_filter_by_lua_block {
+        local wasm = require("resty.proxy-wasm")
+        local ctx = ngx.ctx.ctx
+        assert(wasm.on_http_response_body(ctx))
+    }
+}
+--- grep_error_log eval
+qr/body size \d+, end of stream \w+/
+--- grep_error_log_out
+body size 0, end of stream true
+--- error_log
+get body err: error status returned by host: not found
+
+
+
+=== TEST 4: get body
+--- config
+location /t {
+    return 200 "helloworld";
+    body_filter_by_lua_block {
+        if not ngx.arg[2] then
+            return
+        end
+
+        local json = require("cjson")
+        local wasm = require("resty.proxy-wasm")
+        local plugin = assert(wasm.load("plugin", "t/testdata/http_body/main.go.wasm"))
+        local ctx = assert(wasm.on_configure(plugin, '{"action":"offset"}'))
+        local conf = {action = "offset"}
+        for _, e in ipairs({
+            {0, 2},
+            {1, 1},
+            {1, 2},
+            {0, 10},
+            {1, 9},
+            {8, 2},
+            {9, 1},
+            {1, 10},
+        }) do
+            conf.start = e[1]
+            conf.size = e[2]
+            local ctx = assert(wasm.on_configure(plugin, json.encode(conf)))
+            ngx.arg[1] = "helloworld"
+            assert(wasm.on_http_response_body(ctx))
+        end
+    }
+}
+--- grep_error_log eval
+qr/get body \[\w+\]/
+--- grep_error_log_out
+get body [he]
+get body [e]
+get body [el]
+get body [helloworld]
+get body [elloworld]
+get body [ld]
+get body [d]
+get body [elloworld]
+
+
+
+=== TEST 5: get body, bad cases
+--- config
+location /t {
+    return 200 "helloworld";
+    body_filter_by_lua_block {
+        if not ngx.arg[2] then
+            return
+        end
+
+        local json = require("cjson")
+        local wasm = require("resty.proxy-wasm")
+        local plugin = assert(wasm.load("plugin", "t/testdata/http_body/main.go.wasm"))
+        local ctx = assert(wasm.on_configure(plugin, '{"action":"offset"}'))
+        local conf = {action = "offset"}
+        for _, e in ipairs({
+            {-1, 2},
+            {1, 0},
+            {1, -1},
+            {10, 2},
+        }) do
+            conf.start = e[1]
+            conf.size = e[2]
+            local ctx = assert(wasm.on_configure(plugin, json.encode(conf)))
+            ngx.arg[1] = "helloworld"
+            assert(wasm.on_http_response_body(ctx))
+        end
+    }
+}
+--- error_log
+[error]
+--- grep_error_log eval
+qr/error status returned by host: bad argument/
+--- grep_error_log_out eval
+"error status returned by host: bad argument\n" x 4

t/testdata/http_body/main.go

@@ -93,3 +93,41 @@ func (ctx *httpContext) OnHttpRequestBody(bodySize int, endOfStream bool) types.
 
 	return types.ActionContinue
 }
+
+func (ctx *httpContext) OnHttpResponseBody(bodySize int, endOfStream bool) types.Action {
+	data, err := proxywasm.GetPluginConfiguration()
+	if err != nil {
+		proxywasm.LogErrorf("error reading plugin configuration: %v", err)
+		return types.ActionContinue
+	}
+
+	proxywasm.LogWarnf("body size %d, end of stream %v", bodySize, endOfStream)
+
+	var action string
+	var conf *fastjson.Value
+	var p fastjson.Parser
+	conf, err = p.ParseBytes(data)
+	if err != nil {
+		proxywasm.LogErrorf("error decoding plugin configuration: %v", err)
+		return types.ActionContinue
+	}
+
+	action = string(conf.GetStringBytes("action"))
+
+	switch action {
+	case "offset":
+		if !endOfStream {
+			return types.ActionContinue
+		}
+		start := conf.GetInt("start")
+		size := conf.GetInt("size")
+		body, err := proxywasm.GetHttpRequestBody(start, size)
+		if err != nil {
+			proxywasm.LogErrorf("get body err: %v", err)
+			return types.ActionContinue
+		}
+		proxywasm.LogWarnf("get body [%s]", string(body))
+	}
+
+	return types.ActionContinue
+}