check array length

plugins/ekuiper/plugin_ekuiper.c

@@ -30,6 +30,7 @@
 #include "read_write.h"
 
 #define EKUIPER_PLUGIN_URL "tcp://127.0.0.1:7081"
+#define EKUIPER_PLUGIN_RECV_MAX_SIZE (4 * 1024 * 1024)
 
 const neu_plugin_module_t neu_plugin_module;
 
@@ -149,6 +150,8 @@ static inline int start(neu_plugin_t *plugin, const char *url)
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_REM_POST, pipe_rm_cb, plugin);
     nng_socket_set_int(plugin->sock, NNG_OPT_SENDBUF, 2048);
     nng_socket_set_int(plugin->sock, NNG_OPT_RECVBUF, 2048);
+    nng_socket_set_size(plugin->sock, NNG_OPT_RECVMAXSZ,
+                        EKUIPER_PLUGIN_RECV_MAX_SIZE);
 
     if ((rv = nng_listen(plugin->sock, url, NULL, 0)) != 0) {
         plog_error(plugin, "nng_listen: %s", nng_strerror(rv));

src/utils/json.c

@@ -18,6 +18,7 @@
  **/
 
 #include <math.h>
+#include <stdint.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -548,6 +549,13 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
         }
     }
 
+    if (json_is_array(ob) && json_array_size(ob) > UINT16_MAX) {
+        ele->ok = false;
+        zlog_error(neuron, "json decode: array too long (%zu elements)",
+                   json_array_size(ob));
+        return -1;
+    }
+
     ele->ok = true;
 
     switch (ele->t) {