GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-22 17:50:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #558 from gc87/mqtt-leak-fix

Browse Source 
fix(mqtt): double free and context leak
This commit is contained in:
fengzeroz
2022-06-17 22:15:46 +08:00
committed by GitHub
parent 341bdfd61f 41dec9b209
commit 4343c2ef74
10 changed files with 101 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/json/neu_json_fn.h
-3
View File
@@ -37,9 +37,6 @@ int neu_json_encode_with_mqtt(void *param, neu_json_encode_fn fn,
void *mqtt_param, neu_json_encode_fn mqtt_fn,
char **result);
int neu_json_encode_setting_with_mqtt(uint32_t node_id, char *setting,
void *mqtt_param, char **result);
#define NEU_JSON_RESPONSE_ERROR(err, func) \
{ \
neu_json_error_resp_t error_code = { 0 }; \
include/neuron/json/neu_json_mqtt.h
-1
View File
@@ -33,7 +33,6 @@ extern "C" {
typedef struct {
char *uuid;
char *command;
} neu_json_mqtt_t;
int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result);
plugins/mqtt/command/command.c
+26 -20
View File
@@ -22,11 +22,12 @@
#include "read_write.h"
#include "utils/log.h"
#include "../mqtt.h"
#include "command.h"
// Read tags with group config
static int read_response(mqtt_response_t *response, neu_plugin_t *plugin,
neu_json_mqtt_t *mqtt, char *json_str)
static int read_request(mqtt_response_t *response, neu_plugin_t *plugin,
neu_json_mqtt_t *mqtt, char *json_str)
{
UNUSED(response);
@@ -41,8 +42,8 @@ static int read_response(mqtt_response_t *response, neu_plugin_t *plugin,
}
// Write tags with group config
static int write_response(mqtt_response_t *response, neu_plugin_t *plugin,
neu_json_mqtt_t *mqtt, char *json_str)
static int write_request(mqtt_response_t *response, neu_plugin_t *plugin,
neu_json_mqtt_t *mqtt, char *json_str)
{
UNUSED(response);
@@ -81,45 +82,50 @@ void command_response_handle(mqtt_response_t *response)
// neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt,
// neu_json_encode_mqtt_resp, &ret_str);
zlog_error(neuron, "JSON parsing mqtt failed");
plog_error(plugin, "json parsing mqtt failed");
free(json_str);
return;
}
switch (response->type) {
case TOPIC_TYPE_READ: {
rc = read_response(response, plugin, mqtt, json_str);
rc = read_request(response, plugin, mqtt, json_str);
if (0 != rc) {
neu_json_decode_mqtt_req_free(mqtt);
}
break;
}
case TOPIC_TYPE_WRITE: {
rc = write_response(response, plugin, mqtt, json_str);
rc = write_request(response, plugin, mqtt, json_str);
if (0 != rc) {
neu_json_decode_mqtt_req_free(mqtt);
}
break;
}
default:
assert(1 == 0);
plog_error(plugin, "invalid topic type");
break;
}
if (NULL != json_str) {
free(json_str);
}
neu_json_decode_mqtt_req_free(mqtt);
free(json_str);
}
char *command_read_once_response(neu_reqresp_head_t * head,
char *command_read_once_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_read_group_t *data, int format)
{
return command_rw_read_once_response(head, data, format);
return NULL;
return command_rw_read_once_response(plugin, head, data, format);
}
char *command_read_periodic_response(neu_reqresp_trans_data_t *data, int format)
char *command_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data, int format)
{
return command_rw_read_periodic_response(data, format);
return command_rw_read_periodic_response(plugin, data, format);
}
char *command_write_response(neu_reqresp_head_t *head, neu_resp_error_t *data)
char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data)
{
return command_rw_write_response(head, data);
return command_rw_write_response(plugin, head, data);
}
plugins/mqtt/command/command.h
+5 -3
View File
@@ -46,13 +46,15 @@ typedef struct {
void command_response_handle(mqtt_response_t *response);
char *command_read_once_response(neu_reqresp_head_t * head,
char *command_read_once_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_read_group_t *data, int format);
char *command_read_periodic_response(neu_reqresp_trans_data_t *data,
char *command_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data,
int format);
char *command_write_response(neu_reqresp_head_t *head, neu_resp_error_t *data);
char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data);
#ifdef __cplusplus
}
plugins/mqtt/command/read_write.c
+20 -7
View File
@@ -22,6 +22,7 @@
#include <stdlib.h>
#include <sys/time.h>
#include "../mqtt.h"
#include "read_write.h"
#include "utils/log.h"
@@ -36,8 +37,8 @@ static uint64_t current_time()
int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req)
{
zlog_info(neuron, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
req->group, req->node);
plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
req->node);
neu_reqresp_head_t header = { 0 };
header.ctx = mqtt;
@@ -145,7 +146,8 @@ static void clean_read_response_json(neu_json_read_resp_t *json)
free(json->tags);
}
char *command_rw_read_once_response(neu_reqresp_head_t * head,
char *command_rw_read_once_response(neu_plugin_t * plugin,
neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format)
{
UNUSED(format);
@@ -154,16 +156,23 @@ char *command_rw_read_once_response(neu_reqresp_head_t * head,
uint16_t len = data->n_tag;
char * json_str = NULL;
neu_json_read_resp_t json = { 0 };
plog_info(plugin, "read resp uuid: %s",
((neu_json_mqtt_t *) head->ctx)->uuid);
wrap_read_response_json(tags, len, &json);
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, head->ctx,
neu_json_encode_mqtt_resp, &json_str);
clean_read_response_json(&json);
neu_json_decode_mqtt_req_free(head->ctx);
return json_str;
}
char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
char *command_rw_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data,
int format)
{
UNUSED(plugin);
neu_resp_tag_value_t * tags = data->tags;
uint16_t len = data->n_tag;
char * json_str = NULL;
@@ -193,7 +202,7 @@ char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_write_req_t *req)
{
zlog_info(neuron, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
req->group, req->node);
neu_reqresp_head_t header = { 0 };
@@ -236,12 +245,16 @@ int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
return 0;
}
char *command_rw_write_response(neu_reqresp_head_t *head,
neu_resp_error_t * data)
char *command_rw_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data)
{
neu_json_error_resp_t error = { .error = data->error };
char * json_str = NULL;
plog_info(plugin, "write resp uuid: %s",
((neu_json_mqtt_t *) head->ctx)->uuid);
neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, head->ctx,
neu_json_encode_mqtt_resp, &json_str);
neu_json_decode_mqtt_req_free(head->ctx);
return json_str;
}
plugins/mqtt/command/read_write.h
+6 -4
View File
@@ -31,17 +31,19 @@ extern "C" {
int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_read_req_t *req);
char *command_rw_read_once_response(neu_reqresp_head_t * head,
char *command_rw_read_once_response(neu_plugin_t * plugin,
neu_reqresp_head_t * head,
neu_resp_read_group_t *data, int format);
char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data,
char *command_rw_read_periodic_response(neu_plugin_t * plugin,
neu_reqresp_trans_data_t *data,
int format);
int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
neu_json_write_req_t *req);
char *command_rw_write_response(neu_reqresp_head_t *head,
neu_resp_error_t * data);
char *command_rw_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data);
#ifdef __cplusplus
}
plugins/mqtt/mqtt.c
+36 -29
View File
@@ -18,15 +18,25 @@
**/
#include <assert.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/time.h>
#include "command/command.h"
#include "mqtt.h"
#include "neuron.h"
#include "util.h"
#define INTERVAL 100000U
#define TIMEOUT 3000U
#define TOPIC_READ_REQ "neuron/%s/read/req"
#define TOPIC_WRITE_REQ "neuron/%s/write/req"
#define TOPIC_READ_RES "neuron/%s/read/resp"
#define TOPIC_WRITE_RES "neuron/%s/write/resp"
#define TOPIC_UPLOAD_RES "neuron/%s/upload"
#define QOS0 0
#define QOS1 1
#define QOS2 2
const neu_plugin_module_t neu_plugin_module;
struct topic_pair {
@@ -37,20 +47,15 @@ struct topic_pair {
int type;
};
typedef struct mqtt_routine {
struct mqtt_routine {
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
UT_array * topics;
} mqtt_routine_t;
struct neu_plugin {
neu_plugin_common_t common;
bool running;
mqtt_routine_t * routine;
char * config;
};
static neu_plugin_t *plugin_log = NULL;
static char *topics_format(char *format, char *name)
{
char temp[256] = { '\0' };
@@ -73,8 +78,8 @@ static void topics_add(UT_array *topics, char *request, int qos_request,
pair->qos_response = qos_response;
pair->type = type;
utarray_push_back(topics, &pair);
nlog_info("add topic-req:%s, topic-res:%s, count:%d", pair->topic_request,
pair->topic_response, utarray_len(topics));
plog_info(plugin_log, "add topic-req:%s, topic-res:%s, count:%d",
pair->topic_request, pair->topic_response, utarray_len(topics));
}
static void topics_cleanup(UT_array *topics)
@@ -84,8 +89,8 @@ static void topics_cleanup(UT_array *topics)
NULL != pair;
pair = (struct topic_pair **) utarray_next(topics, pair)) {
nlog_info("remove topic-req:%s, topic-res:%s", (*pair)->topic_request,
(*pair)->topic_response);
plog_info(plugin_log, "remove topic-req:%s, topic-res:%s",
(*pair)->topic_request, (*pair)->topic_response);
if (NULL != (*pair)->topic_request) {
free((*pair)->topic_request);
}
@@ -257,6 +262,7 @@ static neu_plugin_t *mqtt_plugin_open(void)
{
neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
plugin_log = plugin; // for plog
return plugin;
}
@@ -267,6 +273,7 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
const char *name = neu_plugin_module.module_name;
plog_info(plugin, "success to free plugin:%s", name);
plugin_log = NULL;
free(plugin);
return NEU_ERR_SUCCESS;
}
@@ -382,13 +389,13 @@ static neu_err_code_e write_response(neu_plugin_t * plugin,
return NEU_ERR_MQTT_FAILURE;
}
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = command_write_response(head, write_data);
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = command_write_response(plugin, head, write_data);
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
@@ -414,10 +421,10 @@ static neu_err_code_e read_response(neu_plugin_t * plugin,
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_READ;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str =
command_read_once_response(head, read_data, routine->option.format);
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
char * json_str = command_read_once_response(plugin, head, read_data,
routine->option.format);
const char *topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error =
neu_mqtt_client_publish(routine->client, topic, qos,
(unsigned char *) json_str, strlen(json_str));
@@ -443,8 +450,8 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
mqtt_routine_t * routine = plugin->routine;
int type = TOPIC_TYPE_UPLOAD;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str =
command_read_periodic_response(trans_data, routine->option.format);
char *json_str = command_read_periodic_response(plugin, trans_data,
routine->option.format);
if (NULL == json_str) {
return NEU_ERR_MQTT_FAILURE;
}
plugins/mqtt/mqtt.h
+8 -13
View File
@@ -25,21 +25,16 @@ extern "C" {
#endif
#include <neuron.h>
#include <stdlib.h>
#define UNUSED(x) (void) (x)
#define INTERVAL 100000U
#define TIMEOUT 3000U
typedef struct mqtt_routine mqtt_routine_t;
#define TOPIC_READ_REQ "neuron/%s/read/req"
#define TOPIC_WRITE_REQ "neuron/%s/write/req"
#define TOPIC_READ_RES "neuron/%s/read/resp"
#define TOPIC_WRITE_RES "neuron/%s/write/resp"
#define TOPIC_UPLOAD_RES "neuron/%s/upload"
#define QOS0 0
#define QOS1 1
#define QOS2 2
struct neu_plugin {
neu_plugin_common_t common;
bool running;
mqtt_routine_t * routine;
char * config;
};
#ifdef __cplusplus
}
src/parser/neu_json_fn.c
-70
View File
@@ -67,76 +67,6 @@ int neu_json_encode_with_mqtt(void *param, neu_json_encode_fn fn,
return neu_json_encode(object, result);
}
int neu_json_encode_setting_with_mqtt(uint32_t node_id, char *setting,
void *mqtt_param, char **result)
{
neu_json_mqtt_t *mqtt = mqtt_param;
char * params = strstr(setting, "\"params\"");
if (NULL == params) {
return -1;
}
int i = sizeof("\"params\"");
int unmatch = 0;
while (params[i]) {
switch (params[i]) {
case '{':
++unmatch; // fall through
case ':':
case ' ':
case '\t':
case '\n':
case '\r':
++i;
break;
default:
return -1;
}
if (unmatch) {
break;
}
}
while (params[i]) {
if ('{' == params[i]) {
++unmatch;
} else if ('}' == params[i]) {
--unmatch;
}
++i;
if (0 == unmatch) {
break;
}
}
if (unmatch) {
return -1;
}
// At this point, the string started at `params` with length `i` is the
// the settings that should present in the response.
// Note that this is not an elegant solution due to the limitation of
// the current implementation.
const char *fmt = "{\"node_id\": %u, \"uuid\": \"%s\", "
"\"command\":\"%s\", "
"%.*s, \"error\": 0}";
// plus 1 for '\0'
int size = 1 +
snprintf(NULL, 0, fmt, node_id, mqtt->uuid, mqtt->command, i, params);
*result = calloc(size, sizeof(char));
if (NULL == *result) {
return -1;
}
snprintf(*result, size, fmt, node_id, mqtt->uuid, mqtt->command, i, params);
return 0;
}
int neu_parse_param(char *buf, char **err_param, int n, neu_json_elem_t *ele,
...)
{
src/parser/neu_json_mqtt.c
-17
View File
@@ -39,11 +39,6 @@ int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result)
.t = NEU_JSON_STR,
} };
neu_json_elem_t elems_command[] = { {
.name = "command",
.t = NEU_JSON_STR,
} };
ret = neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elems_uuid), elems_uuid);
if (ret == 0) {
req->uuid = elems_uuid[0].v.val_str;
@@ -51,14 +46,6 @@ int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result)
goto decode_fail;
}
ret =
neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elems_command), elems_command);
if (ret == 0) {
req->command = elems_command[0].v.val_str;
} else {
req->command = strdup("");
}
*result = req;
return 0;
@@ -79,10 +66,6 @@ void neu_json_decode_mqtt_req_free(neu_json_mqtt_t *req)
free(req->uuid);
}
if (NULL != req->command) {
free(req->command);
}
free(req);
}