X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fmqtt.c;h=dbef915ae4d681ba4f634e3042df52defdc7c609;hb=edd9af8a874ebc9f2a7f02846807229a648917db;hp=1b71d423d458109247e095cd8b5ce97de30b0ce4;hpb=9b39918bdadabb12ec040f0f191c0d21cb6ddd53;p=collectd.git diff --git a/src/mqtt.c b/src/mqtt.c index 1b71d423..dbef915a 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -24,6 +24,7 @@ * Authors: * Marc Falzon * Florian octo Forster + * Jan-Piet Mens **/ // Reference: http://mosquitto.org/api/files/mosquitto-h.html @@ -35,8 +36,6 @@ #include "utils_cache.h" #include "utils_complain.h" -#include - #include #define MQTT_MAX_TOPIC_SIZE 1024 @@ -48,6 +47,9 @@ #ifndef MQTT_KEEPALIVE # define MQTT_KEEPALIVE 60 #endif +#ifndef SSL_VERIFY_PEER +# define SSL_VERIFY_PEER 1 +#endif /* @@ -67,6 +69,11 @@ struct mqtt_client_conf char *username; char *password; int qos; + char *cacertificatefile; + char *certificatefile; + char *certificatekeyfile; + char *tlsprotocol; + char *ciphersuite; /* For publishing */ char *topic_prefix; @@ -176,9 +183,10 @@ static void on_message ( char *payload; int status; - if ((msg->payloadlen <= 0) - || (((uint8_t *) msg->payload)[msg->payloadlen - 1] != 0)) + if (msg->payloadlen <= 0) { + DEBUG ("mqtt plugin: message has empty payload"); return; + } topic = strdup (msg->topic); name = strip_prefix (topic); @@ -207,7 +215,16 @@ static void on_message ( } vl.values_len = ds->ds_num; - payload = strdup ((void *) msg->payload); + payload = malloc (msg->payloadlen+1); + if (payload == NULL) + { + ERROR ("mqtt plugin: malloc for payload buffer failed."); + sfree (vl.values); + return; + } + memmove (payload, msg->payload, msg->payloadlen); + payload[msg->payloadlen] = 0; + DEBUG ("mqtt plugin: payload = \"%s\"", payload); status = parse_values (payload, &vl, ds); if (status != 0) @@ -277,6 +294,35 @@ static int mqtt_connect (mqtt_client_conf_t *conf) return (-1); } +#if LIBMOSQUITTO_MAJOR != 0 + if (conf->cacertificatefile) { + status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL, + conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL); + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status)); + mosquitto_destroy (conf->mosq); + conf->mosq = NULL; + return (-1); + } + + status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite); + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status)); + mosquitto_destroy (conf->mosq); + conf->mosq = NULL; + return (-1); + } + + status = mosquitto_tls_insecure_set(conf->mosq, false); + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status)); + mosquitto_destroy (conf->mosq); + conf->mosq = NULL; + return (-1); + } + } +#endif + if (conf->username && conf->password) { status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password); @@ -494,6 +540,10 @@ static int mqtt_write (const data_set_t *ds, const value_list_t *vl, * StoreRates true * Retain false * QoS 0 + * CACert "ca.pem" Enables TLS if set + * CertificateFile "client-cert.pem" optional + * CertificateKeyFile "client-key.pem" optional + * TLSProtocol "tlsv1.2" optional * */ static int mqtt_config_publisher (oconfig_item_t *ci) @@ -507,7 +557,7 @@ static int mqtt_config_publisher (oconfig_item_t *ci) conf = calloc (1, sizeof (*conf)); if (conf == NULL) { - ERROR ("mqtt plugin: malloc failed."); + ERROR ("mqtt plugin: calloc failed."); return (-1); } conf->publish = 1; @@ -570,6 +620,16 @@ static int mqtt_config_publisher (oconfig_item_t *ci) cf_util_get_boolean (child, &conf->store_rates); else if (strcasecmp ("Retain", child->key) == 0) cf_util_get_boolean (child, &conf->retain); + else if (strcasecmp ("CACert", child->key) == 0) + cf_util_get_string (child, &conf->cacertificatefile); + else if (strcasecmp ("CertificateFile", child->key) == 0) + cf_util_get_string (child, &conf->certificatefile); + else if (strcasecmp ("CertificateKeyFile", child->key) == 0) + cf_util_get_string (child, &conf->certificatekeyfile); + else if (strcasecmp ("TLSProtocol", child->key) == 0) + cf_util_get_string (child, &conf->tlsprotocol); + else if (strcasecmp ("CipherSuite", child->key) == 0) + cf_util_get_string (child, &conf->ciphersuite); else ERROR ("mqtt plugin: Unknown config option: %s", child->key); } @@ -590,7 +650,7 @@ static int mqtt_config_publisher (oconfig_item_t *ci) * User "guest" * Password "secret" * Topic "collectd/#" - * + * */ static int mqtt_config_subscriber (oconfig_item_t *ci) { @@ -602,7 +662,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) conf = calloc (1, sizeof (*conf)); if (conf == NULL) { - ERROR ("mqtt plugin: malloc failed."); + ERROR ("mqtt plugin: calloc failed."); return (-1); } conf->publish = 0; @@ -638,11 +698,11 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) cf_util_get_string (child, &conf->host); else if (strcasecmp ("Port", child->key) == 0) { - int tmp = cf_util_get_port_number (child); - if (tmp < 0) + status = cf_util_get_port_number (child); + if (status < 0) ERROR ("mqtt plugin: Invalid port number."); else - conf->port = tmp; + conf->port = status; } else if (strcasecmp ("ClientId", child->key) == 0) cf_util_get_string (child, &conf->client_id); @@ -652,12 +712,12 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) cf_util_get_string (child, &conf->password); else if (strcasecmp ("QoS", child->key) == 0) { - int tmp = -1; - status = cf_util_get_int (child, &tmp); - if ((status != 0) || (tmp < 0) || (tmp > 2)) + int qos = -1; + status = cf_util_get_int (child, &qos); + if ((status != 0) || (qos < 0) || (qos > 2)) ERROR ("mqtt plugin: Not a valid QoS setting."); else - conf->qos = tmp; + conf->qos = qos; } else if (strcasecmp ("Topic", child->key) == 0) cf_util_get_string (child, &conf->topic); @@ -667,7 +727,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) ERROR ("mqtt plugin: Unknown config option: %s", child->key); } - tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num); + tmp = realloc (subscribers, sizeof (*subscribers) * (subscribers_num + 1) ); if (tmp == NULL) { ERROR ("mqtt plugin: realloc failed.");