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 #485 from gc87/prepare-spb

Browse Source 
Prepare work for sparkplugb plugin
This commit is contained in:
fengzeroz
2022-05-16 11:10:44 +08:00
committed by GitHub
parent 0e0a03199e 241aeb8654
commit db31cee551
6 changed files with 75 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/common.resource
+1 -1
View File
@@ -75,7 +75,7 @@ ${MQTT_CONFIG_HOST} 127.0.0.1
${MQTT_CONFIG_PORT} ${1883}
${MQTT_CONFIG} {"client-id": "${MQTT_CLIENTID}", "upload-topic": "", "format": 1,
... "ssl": false, "host": "${MQTT_CONFIG_HOST}", "port": ${MQTT_CONFIG_PORT},
... "username": "", "password": "", "ca":"", "cert":"", "key":""}
... "username": "", "password": "", "ca":"", "cert":"", "key":"", "keypass":""}
${MODBUS_TCP_CONFIG} {"host": "127.0.0.1", "port": 60502, "connection_mode": 0}
ft/persistence_test.robot
+1 -1
View File
@@ -93,7 +93,7 @@ Get the setted adapters information after stopping and restarting Neuron, it sho
Node Setting ${modbus_tcp2_id} {"host": "127.0.0.1", "port": 502, "connection_mode": 0}
${mqtt_id} = Get Node ID ${NODE_MQTT} mqtt-1
Node Setting ${mqtt_id} {"client-id":"upload123", "upload-topic":"", "format": 0, "ssl":false,"host":"127.0.0.1","port":1883,"username":"admin","password":"0000","ca":"", "cert":"", "key":""}
Node Setting ${mqtt_id} {"client-id":"upload123", "upload-topic":"", "format": 0, "ssl":false,"host":"127.0.0.1","port":1883,"username":"admin","password":"0000","ca":"", "cert":"", "key":"", "keypass":""}
Neuron Stop Running ${process}
${process} = Neuron Start Running
plugins/mqtt/mqtt.json
+14
View File
@@ -128,5 +128,19 @@
"valid": {
"length": 81960
}
},
"keypass": {
"name": "keypass",
"description": "key password",
"attribute": "optional",
"type": "file",
"condition": {
"field": "ssl",
"value": true
},
"default": "",
"valid": {
"length": 256
}
}
}
plugins/mqtt/util.c
+12 -2
View File
@@ -80,6 +80,11 @@ void mqtt_option_uninit(neu_mqtt_option_t *option)
free(option->key);
option->key = NULL;
}
if (NULL != option->keypass) {
free(option->keypass);
option->keypass = NULL;
}
}
int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
@@ -101,9 +106,11 @@ int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR };
neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR };
neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR };
neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR };
ret = neu_parse_param(config->buf, &error, 11, &id, &upload, &format, &ssl,
&host, &port, &username, &password, &ca, &cert, &key);
ret = neu_parse_param(config->buf, &error, 12, &id, &upload, &format, &ssl,
&host, &port, &username, &password, &ca, &cert, &key,
&keypass);
if (0 != ret) {
return ret;
}
@@ -137,5 +144,8 @@ int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
option->key = key.v.val_str;
option->keepalive_interval = 30;
option->clean_session = 1;
option->will_topic = NULL;
option->will_payload = NULL;
option->keypass = keypass.v.val_str;
return 0;
}
src/connection/mqtt_c_client.c
+45 -9
View File
@@ -44,10 +44,14 @@ struct reconnect_state {
char * password;
int keepalive;
int clean;
char * will_topic;
char * will_payload;
int will_payload_len;
char * ca_file;
char * ca_path;
char * cert_file;
char * key_file;
char * keypass;
mqtt_c_client_t *client;
uint8_t send_buf[SEND_BUF_SIZE];
uint8_t recv_buf[RECV_BUF_SIZE];
@@ -87,7 +91,7 @@ static neu_err_code_e client_destroy(mqtt_c_client_t *client);
static neu_err_code_e client_subscribe_send(mqtt_c_client_t * client,
struct subscribe_tuple *tuple);
static X509 *ssl_ctx_load_cert(const char *b64)
static X509 *ssl_ctx_load_cert(const char *b64, const char *password)
{
unsigned char *bin = NULL;
int len = 0;
@@ -99,13 +103,13 @@ static X509 *ssl_ctx_load_cert(const char *b64)
X509 *cert = NULL;
BIO * bio = NULL;
bio = BIO_new_mem_buf((void *) bin, len);
cert = PEM_read_bio_X509(bio, NULL, 0, NULL);
cert = PEM_read_bio_X509(bio, NULL, 0, (void *) password);
BIO_free_all(bio);
free(bin);
return cert;
}
static EVP_PKEY *ssl_ctx_load_key(const char *b64)
static EVP_PKEY *ssl_ctx_load_key(const char *b64, const char *password)
{
unsigned char *bin = NULL;
int len = 0;
@@ -117,7 +121,7 @@ static EVP_PKEY *ssl_ctx_load_key(const char *b64)
EVP_PKEY *key = NULL;
BIO * bio = NULL;
bio = BIO_new_mem_buf((void *) bin, len);
key = PEM_read_bio_PrivateKey(bio, NULL, 0, NULL);
key = PEM_read_bio_PrivateKey(bio, NULL, 0, (void *) password);
BIO_free_all(bio);
free(bin);
return key;
@@ -193,7 +197,7 @@ static void ssl_ctx_init(struct reconnect_state *state)
// setup cert
if (NULL != cert_file && 0 < strlen(cert_file)) {
state->cert = ssl_ctx_load_cert(cert_file);
state->cert = ssl_ctx_load_cert(cert_file, NULL);
if (NULL != state->cert) {
if (0 >= SSL_CTX_use_certificate(state->ssl_ctx, state->cert)) {
log_error("failed to use certificate");
@@ -205,7 +209,8 @@ static void ssl_ctx_init(struct reconnect_state *state)
// setup key
if (NULL != key_file && 0 < strlen(key_file)) {
state->key = ssl_ctx_load_key(key_file);
const char *keypass = state->keypass;
state->key = ssl_ctx_load_key(key_file, keypass);
if (NULL != state->key) {
if (0 >= SSL_CTX_use_PrivateKey(state->ssl_ctx, state->key)) {
log_error("failed to use privatekey");
@@ -378,6 +383,18 @@ static void reconnect_state_uninit(struct reconnect_state *state)
pthread_mutex_destroy(&state->sock_state_mutex);
pthread_mutex_destroy(&state->running_mutex);
pthread_mutex_destroy(&state->working_mutex);
if (NULL != state->will_topic) {
free(state->will_topic);
}
if (NULL != state->will_payload) {
free(state->will_payload);
}
if (NULL != state->keypass) {
free(state->keypass);
}
}
static void subscribe_on_reconnect(mqtt_c_client_t *client)
@@ -429,9 +446,13 @@ static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
connect_flags = MQTT_CONNECT_CLEAN_SESSION;
}
enum MQTTErrors rc =
mqtt_connect(mqtt, client_id, NULL, NULL, 0, state->username,
state->password, connect_flags, state->keepalive);
const char *will_topic = state->will_topic;
const char *will_payload = state->will_payload;
size_t payload_len = state->will_payload_len;
enum MQTTErrors rc = mqtt_connect(
mqtt, client_id, will_topic, will_payload, payload_len, state->username,
state->password, connect_flags, state->keepalive);
if (MQTT_OK != rc) {
return;
@@ -551,6 +572,21 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option,
pthread_mutex_init(&state->running_mutex, NULL);
pthread_mutex_init(&state->working_mutex, NULL);
if (NULL != client->option->will_topic) {
state->will_topic = strdup(option->will_topic);
if (NULL != client->option->will_payload) {
state->will_payload = strdup(client->option->will_payload);
state->will_payload_len = strlen(state->will_payload);
}
}
if (NULL != client->option->keypass &&
0 < strlen(client->option->keypass)) {
state->keypass = strdup(client->option->keypass);
} else {
state->keypass = NULL;
}
client->mqtt.publish_response_callback_state = state;
return client;
}
tests/mqtt_test.cc
+2 -1
View File
@@ -13,7 +13,8 @@ TEST(MQTTTest, mqtt_option_init_by_config)
"\"upload123\", \"upload-topic\":\"\", \"format\": "
"0,\"ssl\":false,\"host\":\"192.168.10."
"116\",\"port\":1883,\"username\":\"\",\"password\":"
"\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\"}}";
"\"\",\"ca\":\"\", \"cert\":\"\", \"key\":\"\", "
"\"keypass\":\"\"}}";
int rc = mqtt_option_init(&config, &option);
EXPECT_EQ(0, rc);