feat: support getting request body (#90)

Author: spacewander

README.md

@@ -99,6 +99,30 @@ end
 assert(wasm.on_http_request_headers(ctx))
 ```
 
+### on_http_request_body
+
+`syntax: ok, err = proxy_wasm.on_http_request_body(plugin_ctx, body, end_of_body)`
+
+Run the HTTP request body filter in the plugin of the given plugin ctx.
+
+```lua
+local plugin, err = proxy_wasm.load("plugin","t/testdata/plugin_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
+-- get_body is a utility method to get the whole request body
+local body = request.get_body()
+-- if the body is not the whole request body, for example, it comes from
+-- lua-resty-upload, remember to set end_of_body to false
+assert(wasm.on_http_request_body(ctx, body, true))
+```
+
 ### on_http_response_headers
 
 `syntax: ok, err = proxy_wasm.on_http_response_headers(plugin_ctx)`

lib/resty/proxy-wasm.lua

@@ -47,7 +47,8 @@ void ngx_http_wasm_unload_plugin(void *plugin);
 void *ngx_http_wasm_on_configure(void *plugin, const char *conf, size_t size);
 void ngx_http_wasm_delete_plugin_ctx(void *hwp_ctx);
 
-ngx_int_t ngx_http_wasm_on_http(void *hwp_ctx, void *r, int type);
+ngx_int_t ngx_http_wasm_on_http(void *hwp_ctx, void *r, int type,
+                                const u_char *body, size_t size, int end_of_body);
 ngx_str_t *ngx_http_wasm_fetch_local_body(void *r);
 
 ngx_int_t ngx_http_wasm_call_max_headers_count(void *r);
@@ -65,7 +66,7 @@ local ngx_table_size = ffi.sizeof("proxy_wasm_table_elt_t")
 
 local _M = {}
 local HTTP_REQUEST_HEADERS = 1
---local HTTP_REQUEST_BODY = 2
+local HTTP_REQUEST_BODY = 2
 local HTTP_RESPONSE_HEADERS = 4
 --local HTTP_RESPONSE_BODY = 8
 
@@ -258,7 +259,7 @@ local function handle_http_callback(plugin_ctx, r, res)
 end
 
 
-function _M.on_http_request_headers(plugin_ctx)
+local function on_http_request(plugin_ctx, ty, body, end_of_body)
     if type(plugin_ctx) ~= "cdata" then
         return nil, "bad plugin ctx"
     end
@@ -268,9 +269,23 @@ function _M.on_http_request_headers(plugin_ctx)
         return nil, "bad request"
     end
 
-    local rc = C.ngx_http_wasm_on_http(plugin_ctx, r, HTTP_REQUEST_HEADERS)
+    local err
+    if ty == HTTP_REQUEST_HEADERS then
+        err = "failed to run proxy_on_http_request_headers"
+    else
+        err = "failed to run proxy_on_http_request_body"
+    end
+
+    local body_size
+    if not body then
+        body_size = 0
+    else
+        body_size = #body
+    end
+
+    local rc = C.ngx_http_wasm_on_http(plugin_ctx, r, ty, body, body_size, end_of_body)
     if rc < 0 then
-        return nil, "failed to run proxy_on_http_request_headers"
+        return nil, err
     end
 
     while true do
@@ -304,6 +319,16 @@ function _M.on_http_request_headers(plugin_ctx)
 end
 
 
+function _M.on_http_request_headers(plugin_ctx)
+    return on_http_request(plugin_ctx, HTTP_REQUEST_HEADERS, nil, 1)
+end
+
+
+function _M.on_http_request_body(plugin_ctx, body, end_of_body)
+    return on_http_request(plugin_ctx, HTTP_REQUEST_BODY, body, end_of_body and 1 or 0)
+end
+
+
 function _M.on_http_response_headers(plugin_ctx)
     if type(plugin_ctx) ~= "cdata" then
         return nil, "bad plugin ctx"
@@ -314,7 +339,7 @@ function _M.on_http_response_headers(plugin_ctx)
         return nil, "bad request"
     end
 
-    local rc = C.ngx_http_wasm_on_http(plugin_ctx, r, HTTP_RESPONSE_HEADERS)
+    local rc = C.ngx_http_wasm_on_http(plugin_ctx, r, HTTP_RESPONSE_HEADERS, nil, 0, 1)
     if rc < 0 then
         return nil, "failed to run proxy_on_http_response_headers"
     end

proxy_wasm_abi.md

@@ -115,11 +115,22 @@ might be retrieved using `proxy_get_buffer`.
 * returns:
   - `i32 (proxy_action_t) next_action`
 
-Called when HTTP request headers are received from the client. TODO: Headers can be retrieved using
-`proxy_get_map` and/or `proxy_get_map_value`.
+Called when HTTP request headers are received from the client.
 
 TODO: pass a correct `num_headers` but not 0.
 
+### `proxy_on_request_body`
+
+* params:
+  - `i32 (uint32_t) context_id`
+  - `i32 (size_t) body_size`
+  - `i32 (bool) end_of_stream`
+* returns:
+  - `i32 (proxy_action_t) next_action`
+
+Called for HTTP request body received from the client. Request body can be retrieved
+using `proxy_get_buffer_bytes`.
+
 ### `proxy_on_response_headers`
 
 * params:
@@ -129,8 +140,7 @@ TODO: pass a correct `num_headers` but not 0.
 * returns:
   - `i32 (proxy_action_t) next_action`
 
-Called when HTTP response headers are received from the upstream. TODO: Headers can be retrieved using
-`proxy_get_map` and/or `proxy_get_map_value`.
+Called when HTTP response headers are received from the upstream.
 
 TODO: pass a correct `num_headers` but not 0.
 

src/http/ngx_http_wasm_api.c

@@ -495,9 +495,9 @@ proxy_get_buffer_bytes(int32_t type, int32_t start, int32_t size,
                        int32_t addr, int32_t size_addr)
 {
     ngx_log_t            *log;
-    const u_char         *data;
-    int32_t               len;
-    const ngx_str_t      *conf;
+    const u_char         *data = NULL;
+    int32_t               len = 0;
+    const ngx_str_t      *buffer;
 
     ngx_http_request_t      *r;
     ngx_http_wasm_ctx_t     *ctx;
@@ -506,9 +506,19 @@ proxy_get_buffer_bytes(int32_t type, int32_t start, int32_t size,
 
     switch (type) {
     case PROXY_BUFFER_TYPE_PLUGIN_CONFIGURATION:
-        conf = ngx_http_wasm_get_conf();
-        data = conf->data;
-        len = conf->len;
+        buffer = ngx_http_wasm_get_conf();
+        if (buffer != NULL) {
+            data = buffer->data;
+            len = buffer->len;
+        }
+        break;
+
+    case PROXY_BUFFER_TYPE_HTTP_REQUEST_BODY:
+        buffer = ngx_http_wasm_get_body();
+        if (buffer != NULL) {
+            data = buffer->data;
+            len = buffer->len;
+        }
         break;
 
     case PROXY_BUFFER_TYPE_HTTP_CALL_RESPONSE_BODY:

src/http/ngx_http_wasm_module.c

@@ -46,6 +46,8 @@ static ngx_str_t proxy_on_done = ngx_string("proxy_on_done");
 static ngx_str_t proxy_on_delete = ngx_string("proxy_on_delete");
 static ngx_str_t proxy_on_request_headers =
     ngx_string("proxy_on_request_headers");
+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_http_call_response =
@@ -647,7 +649,8 @@ ngx_http_wasm_handle_rc_before_proxy(ngx_http_wasm_main_conf_t *wmcf,
 
 ngx_int_t
 ngx_http_wasm_on_http(ngx_http_wasm_plugin_ctx_t *hwp_ctx, ngx_http_request_t *r,
-                      ngx_http_wasm_phase_t type)
+                      ngx_http_wasm_phase_t type, const u_char *body, size_t size,
+                      int end_of_body)
 {
     ngx_int_t                        rc;
     ngx_log_t                       *log;
@@ -665,6 +668,12 @@ ngx_http_wasm_on_http(ngx_http_wasm_plugin_ctx_t *hwp_ctx, ngx_http_request_t *r
 
     wmcf = ngx_http_get_module_main_conf(r, ngx_http_wasm_module);
     hwp_ctx->state->r = r;
+
+    if (body != NULL) {
+        hwp_ctx->state->body.data = (u_char *) body;
+        hwp_ctx->state->body.len = size;
+    }
+
     ngx_http_wasm_set_state(hwp_ctx->state);
 
     http_ctx = ngx_http_wasm_fetch_http_ctx(hwp_ctx, r);
@@ -677,19 +686,29 @@ ngx_http_wasm_on_http(ngx_http_wasm_plugin_ctx_t *hwp_ctx, ngx_http_request_t *r
 
     if (type == HTTP_RESPONSE_HEADERS) {
         cb_name = &proxy_on_response_headers;
+    } else if (type == HTTP_REQUEST_BODY) {
+        cb_name = &proxy_on_request_body;
     } else {
         cb_name = &proxy_on_request_headers;
     }
 
-    if (hwp_ctx->hw_plugin->abi_version == PROXY_WASM_ABI_VER_010) {
-        rc = ngx_wasm_vm.call(hwp_ctx->hw_plugin->plugin,
-                              cb_name,
-                              true, NGX_WASM_PARAM_I32_I32, http_ctx->id, 0);
+    if (type == HTTP_REQUEST_HEADERS || type == HTTP_RESPONSE_HEADERS) {
+        if (hwp_ctx->hw_plugin->abi_version == PROXY_WASM_ABI_VER_010) {
+            rc = ngx_wasm_vm.call(hwp_ctx->hw_plugin->plugin,
+                                  cb_name,
+                                  true, NGX_WASM_PARAM_I32_I32, http_ctx->id, 0);
+        } else {
+            rc = ngx_wasm_vm.call(hwp_ctx->hw_plugin->plugin,
+                                  cb_name,
+                                  true, NGX_WASM_PARAM_I32_I32_I32, http_ctx->id,
+                                  0, 1);
+        }
+
     } else {
         rc = ngx_wasm_vm.call(hwp_ctx->hw_plugin->plugin,
                               cb_name,
                               true, NGX_WASM_PARAM_I32_I32_I32, http_ctx->id,
-                              0, 1);
+                              size, end_of_body);
     }
 
     ngx_http_wasm_set_state(NULL);

src/http/ngx_http_wasm_state.c

@@ -23,6 +23,12 @@ static ngx_http_wasm_state_t    *cur_state = NULL;
 void
 ngx_http_wasm_set_state(ngx_http_wasm_state_t *state)
 {
+    if (state == NULL) {
+        /* clear state data */
+        cur_state->body.data = NULL;
+        cur_state->body.len = 0;
+    }
+
     cur_state = state;
 }
 
@@ -69,3 +75,14 @@ ngx_http_wasm_get_plugin_name(void)
 
     return cur_state->plugin_name;
 }
+
+
+const ngx_str_t *
+ngx_http_wasm_get_body(void)
+{
+    if (cur_state == NULL) {
+        return NULL;
+    }
+
+    return &cur_state->body;
+}

src/http/ngx_http_wasm_state.h

@@ -24,6 +24,7 @@
 
 typedef struct {
     ngx_str_t                 conf;
+    ngx_str_t                 body;
     ngx_http_request_t       *r;
     ngx_str_t                *plugin_name;
 } ngx_http_wasm_state_t;
@@ -34,6 +35,7 @@ const ngx_str_t *ngx_http_wasm_get_conf(void);
 ngx_http_request_t *ngx_http_wasm_get_req(void);
 ngx_log_t *ngx_http_wasm_get_log(void);
 ngx_str_t *ngx_http_wasm_get_plugin_name(void);
+const ngx_str_t *ngx_http_wasm_get_body(void);
 
 
 #endif // NGX_HTTP_WASM_STATE_H