box: support MP_ARROW by mp_check()

The missed `mp_validate_arrow()` is added to `msgpack_check_ext_data()`.
Also the xrow body decoding error now contains its cause.

Follow-up #10508

NO_DOC=bugfix
NO_CHANGELOG=No sense in mentioning in CE

Author: Gumix

src/box/msgpack.c

@@ -32,6 +32,7 @@
 #include "msgpuck/msgpuck.h"
 
 #include "mp_extension_types.h"
+#include "mp_arrow.h"
 #include "mp_decimal.h"
 #include "mp_error.h"
 #include "mp_uuid.h"
@@ -143,6 +144,14 @@ msgpack_check_ext_data(int8_t type, const char *data, uint32_t len)
 			return -1;
 		}
 		return 0;
+	case MP_ARROW:
+		if (mp_validate_arrow(data, len) != 0) {
+			/* The error is set by arrow_ipc_decode(). */
+			diag_add(ClientError, ER_INVALID_MSGPACK,
+				 "cannot unpack arrow data");
+			return -1;
+		}
+		return 0;
 	case MP_COMPRESSION:
 	default:
 		return mp_check_ext_data_default(type, data, len);

src/box/xrow.c

@@ -258,23 +258,32 @@ xrow_decode(struct xrow_header *header, const char **pos,
 	/* Restore transaction id from lsn and transaction serial number. */
 	header->tsn = header->lsn - header->tsn;
 
-	/* Nop requests aren't supposed to have a body. */
-	if (*pos < end && header->type != IPROTO_NOP) {
-		const char *body = *pos;
-		if (mp_check(pos, end))
-			goto bad_body;
-		header->bodycnt = 1;
-		header->body[0].iov_base = (void *) body;
-		header->body[0].iov_len = *pos - body;
-	}
-	if (end_is_exact && *pos < end)
-		goto bad_body;
+	if (*pos == end) {
+		/* No body, nothing to validate. */
+		return 0;
+	}
+	if (header->type == IPROTO_NOP) {
+		if (end_is_exact) {
+			diag_set(ClientError, ER_INVALID_MSGPACK,
+				 "junk after packet body");
+			goto dump;
+		}
+		/* Nop requests aren't supposed to have a body. */
+		return 0;
+	}
+
+	const char *body = *pos;
+	int rc = end_is_exact ? mp_check_exact(pos, end) : mp_check(pos, end);
+	if (rc != 0) {
+		diag_add(ClientError, ER_INVALID_MSGPACK, "packet body");
+		goto dump;
+	}
+	header->bodycnt = 1;
+	header->body[0].iov_base = (void *)body;
+	header->body[0].iov_len = *pos - body;
 	return 0;
 bad_header:
 	diag_set(ClientError, ER_INVALID_MSGPACK, "packet header");
-	goto dump;
-bad_body:
-	diag_set(ClientError, ER_INVALID_MSGPACK, "packet body");
 dump:
 	dump_row_hex(start, end);
 	return -1;

src/lib/core/CMakeLists.txt

@@ -29,6 +29,7 @@ set(core_sources
     trigger.cc
     port.c
     arrow_ipc.c
+    mp_arrow.c
     decimal.c
     mp_decimal.c
     cord_buf.c

src/lib/core/mp_arrow.c

@@ -0,0 +1,27 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2024, Tarantool AUTHORS, please see AUTHORS file.
+ */
+
+#include "mp_arrow.h"
+
+#include <assert.h>
+#include <string.h>
+#include "core/arrow_ipc.h"
+
+int
+mp_validate_arrow(const char *data, uint32_t len)
+{
+	struct ArrowArray array;
+	struct ArrowSchema schema;
+	memset(&array, 0, sizeof(array));
+	memset(&schema, 0, sizeof(schema));
+	if (arrow_ipc_decode(&array, &schema, data, data + len) != 0)
+		return 1;
+	assert(array.release != NULL);
+	assert(schema.release != NULL);
+	array.release(&array);
+	schema.release(&schema);
+	return 0;
+}

src/lib/core/mp_arrow.h

@@ -0,0 +1,28 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2024, Tarantool AUTHORS, please see AUTHORS file.
+ */
+
+#pragma once
+
+#include <stdint.h>
+
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+/**
+ * Check that the given buffer contains a valid Arrow record batch.
+ * @param data The buffer containing a record batch in Arrow IPC format, without
+ *             MP_EXT header.
+ * @param len  Length of @a data.
+ * @retval 1   Couldn't decode Arrow data.
+ * @retval 0   Ok.
+ */
+int
+mp_validate_arrow(const char *data, uint32_t len);
+
+#if defined(__cplusplus)
+} /* extern "C" */
+#endif /* defined(__cplusplus) */

test/box-luatest/gh_10508_iproto_insert_arrow_test.lua

@@ -88,6 +88,12 @@ g.test_iproto_insert_arrow_invalid = function(cg)
         --         size: 0}}
         r = _G.iproto_insert_arrow('8210cd020036c70008')
         t.assert_equals(r[box.iproto.key.ERROR_24],
+                        'Invalid MsgPack - packet body')
+        t.assert_equals(r[box.iproto.key.ERROR][0][2][3],
+                        'Invalid MsgPack - invalid extension')
+        t.assert_equals(r[box.iproto.key.ERROR][0][3][3],
+                        'Invalid MsgPack - cannot unpack arrow data')
+        t.assert_equals(r[box.iproto.key.ERROR][0][4][3],
                         'Arrow decode error: unexpected data size')
         -- <MP_MAP> {
         --     IPROTO_SPACE_ID: 512,
@@ -97,6 +103,12 @@ g.test_iproto_insert_arrow_invalid = function(cg)
         --         data: [0xde, 0xad, 0xbe, 0xef]}}
         r = _G.iproto_insert_arrow('8210cd020036c70408deadbeef')
         t.assert_equals(r[box.iproto.key.ERROR_24],
+                        'Invalid MsgPack - packet body')
+        t.assert_equals(r[box.iproto.key.ERROR][0][2][3],
+                        'Invalid MsgPack - invalid extension')
+        t.assert_equals(r[box.iproto.key.ERROR][0][3][3],
+                        'Invalid MsgPack - cannot unpack arrow data')
+        t.assert_equals(r[box.iproto.key.ERROR][0][4][3],
                         'Arrow decode error: ' ..
                         'Expected at least 8 bytes in remainder of stream')
 
@@ -107,6 +119,12 @@ g.test_iproto_insert_arrow_invalid = function(cg)
             000004000000f0ffffff4000000001000000610000000600080004000c0010000400
             080009000c000c000c0000000400000008000a000c00040006000800ffffffff]])
         t.assert_equals(r[box.iproto.key.ERROR_24],
+                        'Invalid MsgPack - packet body')
+        t.assert_equals(r[box.iproto.key.ERROR][0][2][3],
+                        'Invalid MsgPack - invalid extension')
+        t.assert_equals(r[box.iproto.key.ERROR][0][3][3],
+                        'Invalid MsgPack - cannot unpack arrow data')
+        t.assert_equals(r[box.iproto.key.ERROR][0][4][3],
                         'Arrow decode error: ' ..
                         'Expected at least 8 bytes in remainder of stream')
 

test/box-luatest/iproto_request_handlers_overriding_test.lua

@@ -310,7 +310,7 @@ g.test_box_iproto_override_nop_rq_type = function(cg)
                         [box.iproto.key.SQL_TEXT] = 'text'
                     }, {__serialize = 'map'})
         _G.test_cb_err(header, body, box.error.INVALID_MSGPACK,
-                       "Invalid MsgPack %- packet body")
+                       "Invalid MsgPack %- junk after packet body")
         box.iproto.override(box.iproto.type.NOP, nil)
     end)
 end