[PATCH v2 5/6] net.box: add helpers to decode msgpack headers

[Alexander Turenko <alexander.turenko@tarantool.org>]

Needed for #3276.

@TarantoolBot document
Title: net.box: helpers to decode msgpack headers

They allow to skip iproto packet and msgpack array headers and pass raw
msgpack data to some other function, say, merger.

Contracts:

```
net_box.check_iproto_data(buf.rpos, buf.wpos - buf.rpos)
    -> new_rpos
    -> nil, err_msg
msgpack.check_array(buf.rpos, buf.wpos - buf.rpos, [, arr_len])
    -> new_rpos, arr_len
    -> nil, err_msg
```

Below the example with msgpack.decode() as the function that need raw
msgpack data. It is just to illustrate the approach, there is no sense
to skip iproto/array headers manually in Lua and then decode the rest in
Lua. But it worth when the raw msgpack data is subject to process in a C
module.

```lua
local function single_select(space, ...)
    return box.space[space]:select(...)
end

local function batch_select(spaces, ...)
    local res = {}
    for _, space in ipairs(spaces) do
        table.insert(res, box.space[space]:select(...))
    end
    return res
end

_G.single_select = single_select
_G.batch_select = batch_select

local res

local buf = buffer.ibuf()
conn.space.s:select(nil, {buffer = buf})
-- check and skip iproto_data header
buf.rpos = assert(net_box.check_iproto_data(buf.rpos, buf.wpos - buf.rpos))
-- check that we really got data from :select() as result
res, buf.rpos = msgpack.decode(buf.rpos, buf.wpos - buf.rpos)
-- check that the buffer ends
assert(buf.rpos == buf.wpos)

buf:recycle()
conn:call('single_select', {'s'}, {buffer = buf})
-- check and skip the iproto_data header
buf.rpos = assert(net_box.check_iproto_data(buf.rpos, buf.wpos - buf.rpos))
-- check and skip the array around return values
buf.rpos = assert(msgpack.check_array(buf.rpos, buf.wpos - buf.rpos, 1))
-- check that we really got data from :select() as result
res, buf.rpos = msgpack.decode(buf.rpos, buf.wpos - buf.rpos)
-- check that the buffer ends
assert(buf.rpos == buf.wpos)

buf:recycle()
local spaces = {'s', 't'}
conn:call('batch_select', {spaces}, {buffer = buf})
-- check and skip the iproto_data header
buf.rpos = assert(net_box.check_iproto_data(buf.rpos, buf.wpos - buf.rpos))
-- check and skip the array around return values
buf.rpos = assert(msgpack.check_array(buf.rpos, buf.wpos - buf.rpos, 1))
-- check and skip the array header before the first select result
buf.rpos = assert(msgpack.check_array(buf.rpos, buf.wpos - buf.rpos, #spaces))
-- check that we really got data from s:select() as result
res, buf.rpos = msgpack.decode(buf.rpos, buf.wpos - buf.rpos)
-- t:select() data
res, buf.rpos = msgpack.decode(buf.rpos, buf.wpos - buf.rpos)
-- check that the buffer ends
assert(buf.rpos == buf.wpos)
```
---
 src/box/lua/net_box.c         |  49 +++++++++++
 src/box/lua/net_box.lua       |   1 +
 src/lua/msgpack.c             |  66 ++++++++++++++
 test/app-tap/msgpack.test.lua | 157 +++++++++++++++++++++++++++++++++-
 test/box/net.box.result       |  58 +++++++++++++
 test/box/net.box.test.lua     |  26 ++++++
 6 files changed, 356 insertions(+), 1 deletion(-)

diff --git a/src/box/lua/net_box.c b/src/box/lua/net_box.c
index c7063d9c8..d71f33768 100644
--- a/src/box/lua/net_box.c
+++ b/src/box/lua/net_box.c
@@ -51,6 +51,9 @@
 
 #define cfg luaL_msgpack_default
 
+static uint32_t CTID_CHAR_PTR;
+static uint32_t CTID_CONST_CHAR_PTR;
+
 static inline size_t
 netbox_prepare_request(lua_State *L, struct mpstream *stream, uint32_t r_type)
 {
@@ -745,9 +748,54 @@ netbox_decode_execute(struct lua_State *L)
 	return 2;
 }
 
+/**
+ * net_box.check_iproto_data(buf.rpos, buf.wpos - buf.rpos)
+ *     -> new_rpos
+ *     -> nil, err_msg
+ */
+int
+netbox_check_iproto_data(struct lua_State *L)
+{
+	uint32_t ctypeid;
+	const char *data = *(const char **) luaL_checkcdata(L, 1, &ctypeid);
+	if (ctypeid != CTID_CHAR_PTR && ctypeid != CTID_CONST_CHAR_PTR)
+		return luaL_error(L,
+			"net_box.check_iproto_data: 'char *' or "
+			"'const char *' expected");
+
+	if (!lua_isnumber(L, 2))
+		return luaL_error(L, "net_box.check_iproto_data: number "
+				  "expected as 2nd argument");
+	const char *end = data + lua_tointeger(L, 2);
+
+	int ok = data < end &&
+		mp_typeof(*data) == MP_MAP &&
+		mp_check_map(data, end) <= 0 &&
+		mp_decode_map(&data) == 1 &&
+		data < end &&
+		mp_typeof(*data) == MP_UINT &&
+		mp_check_uint(data, end) <= 0 &&
+		mp_decode_uint(&data) == IPROTO_DATA;
+
+	if (!ok) {
+		lua_pushnil(L);
+		lua_pushstring(L,
+			"net_box.check_iproto_data: wrong iproto data packet");
+		return 2;
+	}
+
+	*(const char **) luaL_pushcdata(L, ctypeid) = data;
+	return 1;
+}
+
 int
 luaopen_net_box(struct lua_State *L)
 {
+	CTID_CHAR_PTR = luaL_ctypeid(L, "char *");
+	assert(CTID_CHAR_PTR != 0);
+	CTID_CONST_CHAR_PTR = luaL_ctypeid(L, "const char *");
+	assert(CTID_CONST_CHAR_PTR != 0);
+
 	static const luaL_Reg net_box_lib[] = {
 		{ "encode_ping",    netbox_encode_ping },
 		{ "encode_call_16", netbox_encode_call_16 },
@@ -765,6 +813,7 @@ luaopen_net_box(struct lua_State *L)
 		{ "communicate",    netbox_communicate },
 		{ "decode_select",  netbox_decode_select },
 		{ "decode_execute", netbox_decode_execute },
+		{ "check_iproto_data", netbox_check_iproto_data },
 		{ NULL, NULL}
 	};
 	/* luaL_register_module polutes _G */
diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua
index 2bf772aa8..0a38efa5a 100644
--- a/src/box/lua/net_box.lua
+++ b/src/box/lua/net_box.lua
@@ -1424,6 +1424,7 @@ local this_module = {
     new = connect, -- Tarantool < 1.7.1 compatibility,
     wrap = wrap,
     establish_connection = establish_connection,
+    check_iproto_data = internal.check_iproto_data,
 }
 
 function this_module.timeout(timeout, ...)
diff --git a/src/lua/msgpack.c b/src/lua/msgpack.c
index b47006038..fca440660 100644
--- a/src/lua/msgpack.c
+++ b/src/lua/msgpack.c
@@ -51,6 +51,7 @@ luamp_error(void *error_ctx)
 }
 
 static uint32_t CTID_CHAR_PTR;
+static uint32_t CTID_CONST_CHAR_PTR;
 static uint32_t CTID_STRUCT_IBUF;
 
 struct luaL_serializer *luaL_msgpack_default = NULL;
@@ -418,6 +419,68 @@ lua_ibuf_msgpack_decode(lua_State *L)
 	return 2;
 }
 
+/**
+ * msgpack.check_array(buf.rpos, buf.wpos - buf.rpos, [, arr_len])
+ *     -> new_rpos, arr_len
+ *     -> nil, err_msg
+ */
+static int
+lua_check_array(lua_State *L)
+{
+	uint32_t ctypeid;
+	const char *data = *(const char **) luaL_checkcdata(L, 1, &ctypeid);
+	if (ctypeid != CTID_CHAR_PTR && ctypeid != CTID_CONST_CHAR_PTR)
+		return luaL_error(L, "msgpack.check_array: 'char *' or "
+				  "'const char *' expected");
+
+	if (!lua_isnumber(L, 2))
+		return luaL_error(L, "msgpack.check_array: number expected as "
+				  "2nd argument");
+	const char *end = data + lua_tointeger(L, 2);
+
+	if (!lua_isnoneornil(L, 3) && !lua_isnumber(L, 3))
+		return luaL_error(L, "msgpack.check_array: number or nil "
+				  "expected as 3rd argument");
+
+	static const char *end_of_buffer_msg = "msgpack.check_array: "
+		"unexpected end of buffer";
+
+	if (data >= end) {
+		lua_pushnil(L);
+		lua_pushstring(L, end_of_buffer_msg);
+		return 2;
+	}
+
+	if (mp_typeof(*data) != MP_ARRAY) {
+		lua_pushnil(L);
+		lua_pushstring(L, "msgpack.check_array: wrong array header");
+		return 2;
+	}
+
+	if (mp_check_array(data, end) > 0) {
+		lua_pushnil(L);
+		lua_pushstring(L, end_of_buffer_msg);
+		return 2;
+	}
+
+	uint32_t len = mp_decode_array(&data);
+
+	if (!lua_isnoneornil(L, 3)) {
+		uint32_t exp_len = (uint32_t) lua_tointeger(L, 3);
+		if (len != exp_len) {
+			lua_pushnil(L);
+			lua_pushfstring(L, "msgpack.check_array: expected "
+					"array of length %d, got length %d",
+					len, exp_len);
+			return 2;
+		}
+	}
+
+	*(const char **) luaL_pushcdata(L, ctypeid) = data;
+	lua_pushinteger(L, len);
+	return 2;
+}
+
 static int
 lua_msgpack_new(lua_State *L);
 
@@ -426,6 +489,7 @@ static const luaL_Reg msgpacklib[] = {
 	{ "decode", lua_msgpack_decode },
 	{ "decode_unchecked", lua_msgpack_decode_unchecked },
 	{ "ibuf_decode", lua_ibuf_msgpack_decode },
+	{ "check_array", lua_check_array },
 	{ "new", lua_msgpack_new },
 	{ NULL, NULL }
 };
@@ -447,6 +511,8 @@ luaopen_msgpack(lua_State *L)
 	assert(CTID_STRUCT_IBUF != 0);
 	CTID_CHAR_PTR = luaL_ctypeid(L, "char *");
 	assert(CTID_CHAR_PTR != 0);
+	CTID_CONST_CHAR_PTR = luaL_ctypeid(L, "const char *");
+	assert(CTID_CONST_CHAR_PTR != 0);
 	luaL_msgpack_default = luaL_newserializer(L, "msgpack", msgpacklib);
 	return 1;
 }
diff --git a/test/app-tap/msgpack.test.lua b/test/app-tap/msgpack.test.lua
index 0e1692ad9..d481d2da9 100755
--- a/test/app-tap/msgpack.test.lua
+++ b/test/app-tap/msgpack.test.lua
@@ -49,9 +49,163 @@ local function test_misc(test, s)
     test:ok(not st and e:match("null"), "null ibuf")
 end
 
+local function test_check_array(test, s)
+    local ffi = require('ffi')
+
+    local good_cases = {
+        {
+            'fixarray',
+            data = '\x94\x01\x02\x03\x04',
+            exp_len = 4,
+            exp_rewind = 1,
+        },
+        {
+            'array 16',
+            data = '\xdc\x00\x04\x01\x02\x03\x04',
+            exp_len = 4,
+            exp_rewind = 3,
+        },
+        {
+            'array 32',
+            data = '\xdd\x00\x00\x00\x04\x01\x02\x03\x04',
+            exp_len = 4,
+            exp_rewind = 5,
+        },
+    }
+
+    local bad_cases = {
+        {
+            'fixmap',
+            data = '\x80',
+            exp_err = 'msgpack.check_array: wrong array header',
+        },
+        {
+            'truncated array 16',
+            data = '\xdc\x00',
+            exp_err = 'msgpack.check_array: unexpected end of buffer',
+        },
+        {
+            'truncated array 32',
+            data = '\xdd\x00\x00\x00',
+            exp_err = 'msgpack.check_array: unexpected end of buffer',
+        },
+        {
+            'zero size buffer',
+            data = '\x90',
+            size = 0,
+            exp_err = 'msgpack.check_array: unexpected end of buffer',
+        },
+        {
+            'negative size buffer',
+            data = '\x90',
+            size = -1,
+            exp_err = 'msgpack.check_array: unexpected end of buffer',
+        },
+    }
+
+    local wrong_1_arg_not_cdata_err = 'expected cdata as 1 argument'
+    local wrong_1_arg_err = "msgpack.check_array: 'char *' or " ..
+        "'const char *' expected"
+    local wrong_2_arg_err = 'msgpack.check_array: number expected as 2nd ' ..
+        'argument'
+    local wrong_3_arg_err = 'msgpack.check_array: number or nil expected as ' ..
+        '3rd argument'
+
+    local bad_api_cases = {
+        {
+            '1st argument: nil',
+            args = {nil, 1},
+            exp_err = wrong_1_arg_not_cdata_err,
+        },
+        {
+            '1st argument: not cdata',
+            args = {1, 1},
+            exp_err = wrong_1_arg_not_cdata_err,
+        },
+        {
+            '1st argument: wrong cdata type',
+            args = {box.tuple.new(), 1},
+            exp_err = wrong_1_arg_err,
+        },
+        {
+            '2nd argument: nil',
+            args = {ffi.cast('char *', '\x90'), nil},
+            exp_err = wrong_2_arg_err,
+        },
+        {
+            '2nd argument: wrong type',
+            args = {ffi.cast('char *', '\x90'), 'eee'},
+            exp_err = wrong_2_arg_err,
+        },
+        {
+            '3rd argument: wrong type',
+            args = {ffi.cast('char *', '\x90'), 1, 'eee'},
+            exp_err = wrong_3_arg_err,
+        },
+    }
+
+    -- Add good cases with wrong expected length to the bad cases.
+    for _, case in ipairs(good_cases) do
+        table.insert(bad_cases, {
+            case[1],
+            data = case.data,
+            exp_len = case.exp_len + 1,
+            exp_err = 'msgpack.check_array: expected array of length'
+        })
+    end
+
+    test:plan(4 * #good_cases + 2 * #bad_cases + #bad_api_cases)
+
+    -- Good cases: don't pass 2nd argument.
+    for _, ctype in ipairs({'char *', 'const char *'}) do
+        for _, case in ipairs(good_cases) do
+            local buf = ffi.cast(ctype, case.data)
+            local size = case.size or #case.data
+            local new_buf, len = s.check_array(buf, size)
+            local rewind = new_buf - buf
+            local description = ('good; %s; %s; %s'):format(case[1], ctype,
+                'do not pass 2nd argument')
+            test:is_deeply({len, rewind}, {case.exp_len, case.exp_rewind},
+                description)
+        end
+    end
+
+    -- Good cases: pass right 2nd argument.
+    for _, ctype in ipairs({'char *', 'const char *'}) do
+        for _, case in ipairs(good_cases) do
+            local buf = ffi.cast(ctype, case.data)
+            local size = case.size or #case.data
+            local new_buf, len = s.check_array(buf, size, case.exp_len)
+            local rewind = new_buf - buf
+            local description = ('good; %s; %s; %s'):format(case[1], ctype,
+                'pass right 2nd argument')
+            test:is_deeply({len, rewind}, {case.exp_len, case.exp_rewind},
+                description)
+        end
+    end
+
+    -- Bad cases.
+    for _, ctype in ipairs({'char *', 'const char *'}) do
+        for _, case in ipairs(bad_cases) do
+            local buf = ffi.cast(ctype, case.data)
+            local size = case.size or #case.data
+            local n, err = s.check_array(buf, size, case.exp_len)
+            local description = ('bad; %s; %s'):format(case[1], ctype)
+            test:ok(n == nil and err:startswith(case.exp_err), description)
+        end
+    end
+
+    -- Bad API usage cases.
+    for _, case in ipairs(bad_api_cases) do
+        local ok, err = pcall(s.check_array, unpack(case.args))
+        local description = 'bad API usage; ' .. case[1]
+        test:is_deeply({ok, err}, {false, case.exp_err},  description)
+    end
+end
+
 tap.test("msgpack", function(test)
     local serializer = require('msgpack')
-    test:plan(10)
+    test:plan(11)
     test:test("unsigned", common.test_unsigned, serializer)
     test:test("signed", common.test_signed, serializer)
     test:test("double", common.test_double, serializer)
@@ -62,4 +216,5 @@ tap.test("msgpack", function(test)
     test:test("ucdata", common.test_ucdata, serializer)
     test:test("offsets", test_offsets, serializer)
     test:test("misc", test_misc, serializer)
+    test:test("check_array", test_check_array, serializer)
 end)
diff --git a/test/box/net.box.result b/test/box/net.box.result
index 2b5a84646..98ba9598e 100644
--- a/test/box/net.box.result
+++ b/test/box/net.box.result
@@ -3433,6 +3433,64 @@ c
 c:close()
 ---
 ...
+ffi = require('ffi')
+---
+...
+-- Case: valid iproto_data packet; char *.
+data = '\x81\x30\x90'
+---
+...
+rpos = ffi.cast('char *', data)
+---
+...
+net.check_iproto_data(rpos, #data) - rpos -- 2
+---
+- 2
+...
+-- Case: valid iproto_data packet; const char *.
+rpos = ffi.cast('const char *', data)
+---
+...
+net.check_iproto_data(rpos, #data) - rpos -- 2
+---
+- 2
+...
+-- Case: invalid iproto_data packet.
+data = '\x91\x01'
+---
+...
+rpos = ffi.cast('char *', data)
+---
+...
+net.check_iproto_data(rpos, #data) -- error
+---
+- null
+- 'net_box.check_iproto_data: wrong iproto data packet'
+...
+-- Case: truncated msgpack.
+data = '\x81'
+---
+...
+rpos = ffi.cast('char *', data)
+---
+...
+net.check_iproto_data(rpos, #data) -- error
+---
+- null
+- 'net_box.check_iproto_data: wrong iproto data packet'
+...
+-- Case: zero size buffer.
+data = ''
+---
+...
+rpos = ffi.cast('char *', data)
+---
+...
+net.check_iproto_data(rpos, #data) -- error
+---
+- null
+- 'net_box.check_iproto_data: wrong iproto data packet'
+...
 box.schema.func.drop('do_long')
 ---
 ...
diff --git a/test/box/net.box.test.lua b/test/box/net.box.test.lua
index 96d822820..f89cf7f4d 100644
--- a/test/box/net.box.test.lua
+++ b/test/box/net.box.test.lua
@@ -1388,6 +1388,32 @@ c = net.connect('8.8.8.8:123456', {wait_connected = false})
 c
 c:close()
 
+ffi = require('ffi')
+
+-- Case: valid iproto_data packet; char *.
+data = '\x81\x30\x90'
+rpos = ffi.cast('char *', data)
+net.check_iproto_data(rpos, #data) - rpos -- 2
+
+-- Case: valid iproto_data packet; const char *.
+rpos = ffi.cast('const char *', data)
+net.check_iproto_data(rpos, #data) - rpos -- 2
+
+-- Case: invalid iproto_data packet.
+data = '\x91\x01'
+rpos = ffi.cast('char *', data)
+net.check_iproto_data(rpos, #data) -- error
+
+-- Case: truncated msgpack.
+data = '\x81'
+rpos = ffi.cast('char *', data)
+net.check_iproto_data(rpos, #data) -- error
+
+-- Case: zero size buffer.
+data = ''
+rpos = ffi.cast('char *', data)
+net.check_iproto_data(rpos, #data) -- error
+
 box.schema.func.drop('do_long')
 box.schema.user.revoke('guest', 'write', 'space', '_schema')
 box.schema.user.revoke('guest', 'read,write', 'space', '_space')
-- 
2.20.1