mqtt.c
Go to the documentation of this file.
/* SPDX-FileCopyrightText: 2021-2023 Greenbone AG
*
* SPDX-License-Identifier: GPL-2.0-or-later
*/
#include "mqtt.h"
#include "uuidutils.h" /* gvm_uuid_make */
#include <stdlib.h>
#include <string.h>
#undef G_LOG_DOMAIN
#define G_LOG_DOMAIN "libgvm util"
#define QOS 1
#define TIMEOUT 10000L
typedef struct
{
void *client;
char *client_id;
} mqtt_t;
static const char *global_server_uri = NULL;
static const char *global_username = NULL;
static const char *global_password = NULL;
static mqtt_t *global_mqtt_client = NULL;
static gboolean mqtt_initialized = FALSE;
static void
mqtt_set_initialized_status (gboolean status)
{
mqtt_initialized = status;
}
gboolean
mqtt_is_initialized ()
{
return mqtt_initialized;
}
static void
mqtt_set_global_server_uri (const char *server_uri_in)
{
global_server_uri = server_uri_in;
}
static const char *
mqtt_get_global_server_uri ()
{
return global_server_uri;
}
static void
mqtt_set_global_username (const char *username)
{
global_username = username;
}
static const char *
mqtt_get_global_username ()
{
return global_username;
}
static void
mqtt_set_global_password (const char *password)
{
global_password = password;
}
static const char *
mqtt_get_global_password ()
{
return global_password;
}
static mqtt_t *
mqtt_get_global_client ()
{
return global_mqtt_client;
}
static void
mqtt_set_global_client (mqtt_t *mqtt)
{
global_mqtt_client = mqtt;
}
static int
mqtt_disconnect (mqtt_t *mqtt)
{
int rc;
rc = MQTTClient_disconnect5 (mqtt->client, 200,
MQTTREASONCODE_NORMAL_DISCONNECTION, NULL);
if (rc != MQTTCLIENT_SUCCESS)
{
g_warning ("Failed to disconnect: %s", MQTTClient_strerror (rc));
return -1;
}
return 0;
}
static void
mqtt_client_destroy (mqtt_t *mqtt)
{
if (mqtt && mqtt->client)
{
MQTTClient_destroy (&mqtt->client);
mqtt->client = NULL;
}
}
static void
mqtt_client_data_destroy (mqtt_t **mqtt)
{
g_free ((*mqtt)->client_id);
g_free (*mqtt);
*mqtt = NULL;
}
void
mqtt_reset ()
{
g_debug ("%s: start", __func__);
mqtt_t *mqtt = mqtt_get_global_client ();
if (mqtt == NULL)
return;
mqtt_client_destroy (mqtt);
mqtt_client_data_destroy (&mqtt);
mqtt_set_global_client (NULL);
g_debug ("%s: end", __func__);
return;
}
static MQTTClient
mqtt_create (mqtt_t *mqtt, const char *address)
{
MQTTClient client;
MQTTClient_createOptions create_opts = MQTTClient_createOptions_initializer;
create_opts.MQTTVersion = MQTTVERSION_5;
if (mqtt == NULL || mqtt->client_id == NULL)
return NULL;
int rc = MQTTClient_createWithOptions (&client, address, mqtt->client_id,
MQTTCLIENT_PERSISTENCE_NONE, NULL,
&create_opts);
if (rc != MQTTCLIENT_SUCCESS)
{
g_warning ("%s: Error creating MQTTClient: %s", __func__,
MQTTClient_strerror (rc));
MQTTClient_destroy (&client);
return NULL;
}
return client;
}
static char *
mqtt_set_client_id (mqtt_t *mqtt)
{
if (mqtt == NULL)
return NULL;
char *uuid;
uuid = gvm_uuid_make ();
mqtt->client_id = uuid;
return uuid;
}
static int
mqtt_set_client (mqtt_t *mqtt, MQTTClient client)
{
if (mqtt == NULL)
{
return -1;
}
mqtt->client = client;
return 0;
}
static int
mqtt_connect (mqtt_t *mqtt, const char *server_uri, const char *username,
const char *password)
{
int rc;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;
MQTTProperties connect_properties = MQTTProperties_initializer;
MQTTResponse resp;
if (mqtt == NULL)
return -1;
client = mqtt_create (mqtt, server_uri);
if (!client)
return -2;
conn_opts.keepAliveInterval = 0;
conn_opts.cleanstart = 1;
conn_opts.MQTTVersion = MQTTVERSION_5;
if (username != NULL && password != NULL)
{
conn_opts.username = username;
conn_opts.password = password;
}
resp = MQTTClient_connect5 (client, &conn_opts, &connect_properties, NULL);
rc = resp.reasonCode;
MQTTProperties_free (&connect_properties);
if (rc != MQTTCLIENT_SUCCESS)
{
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_CRITICAL,
"%s: mqtt connection error to %s: %s", __func__, server_uri,
MQTTClient_strerror (rc));
MQTTResponse_free (resp);
return -3;
}
mqtt_set_client (mqtt, client);
MQTTResponse_free (resp);
return 0;
}
int
mqtt_init (const char *server_uri)
{
return mqtt_init_auth (server_uri, NULL, NULL);
}
int
mqtt_init_auth (const char *server_uri, const char *username,
const char *password)
{
mqtt_t *mqtt = NULL;
g_debug ("%s: start", __func__);
mqtt = g_malloc0 (sizeof (mqtt_t));
// Set random uuid as client id
if (mqtt_set_client_id (mqtt) == NULL)
{
g_warning ("%s: Could not set client id.", __func__);
g_free (mqtt);
mqtt = NULL;
return -1;
}
g_debug ("%s: client id set: %s", __func__, mqtt->client_id);
if (mqtt_get_global_server_uri () == NULL)
mqtt_set_global_server_uri (server_uri);
if (mqtt_get_global_username () == NULL)
mqtt_set_global_username (username);
if (mqtt_get_global_password () == NULL)
mqtt_set_global_password (password);
if (mqtt_connect (mqtt, server_uri, username, password))
{
g_warning ("%s: Unable to connect to MQTT broker.", __func__);
g_free (mqtt);
mqtt = NULL;
return -1;
}
mqtt_set_global_client (mqtt);
mqtt_set_initialized_status (TRUE);
g_debug ("%s: end", __func__);
return 0;
}
static void
mqtt_reinit ()
{
const char *server_uri;
const char *username;
const char *password;
server_uri = mqtt_get_global_server_uri ();
if (server_uri == NULL)
{
g_warning ("%s: mqtt_init() has to be called once at program start "
"else the server URI is not set. ",
__func__);
}
username = mqtt_get_global_username ();
password = mqtt_get_global_password ();
mqtt_init_auth (server_uri, username, password);
}
static int
mqtt_client_publish (mqtt_t *mqtt, const char *topic, const char *msg)
{
MQTTClient client;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
MQTTResponse resp;
int rc;
client = mqtt->client;
if (client == NULL)
{
return -1;
}
pubmsg.payload = (char *) msg;
pubmsg.payloadlen = (int) strlen (msg);
pubmsg.qos = QOS;
pubmsg.retained = 0;
resp = MQTTClient_publishMessage5 (client, topic, &pubmsg, &token);
rc = resp.reasonCode;
if (rc != MQTTCLIENT_SUCCESS)
{
g_warning ("Failed to connect: %s", MQTTClient_strerror (rc));
MQTTResponse_free (resp);
return -2;
}
rc = MQTTClient_waitForCompletion (client, token, TIMEOUT);
if (rc != MQTTCLIENT_SUCCESS)
{
g_debug ("Message '%s' with delivery token %d could not be "
"published on topic %s",
msg, token, topic);
}
MQTTResponse_free (resp);
return rc;
}
int
mqtt_publish (const char *topic, const char *msg)
{
mqtt_t *mqtt = NULL;
int rc = 0;
if ((mqtt_get_global_client ()) == NULL)
mqtt_reinit ();
mqtt = mqtt_get_global_client ();
rc = mqtt_client_publish (mqtt, topic, msg);
return rc;
}
int
mqtt_publish_single_message (const char *server_uri_in, const char *topic,
const char *msg)
{
return mqtt_publish_single_message_auth (server_uri_in, NULL, NULL, topic,
msg);
}
int
mqtt_publish_single_message_auth (const char *server_uri_in,
const char *username_in,
const char *passwd_in, const char *topic,
const char *msg)
{
const char *server_uri;
const char *username = NULL;
const char *password = NULL;
mqtt_t *mqtt = NULL;
int ret = 0;
// If server_uri is NULL try to get global
if (server_uri_in == NULL)
{
server_uri = mqtt_get_global_server_uri ();
if (server_uri == NULL)
{
g_warning (
"%s: No server URI provided and no global server URI available.",
__func__);
return -1;
}
}
else
{
server_uri = server_uri_in;
}
if (username_in == NULL || passwd_in == NULL)
{
username = mqtt_get_global_username ();
password = mqtt_get_global_password ();
}
else
{
username = username_in;
password = passwd_in;
}
mqtt = g_malloc0 (sizeof (mqtt_t));
// Set random uuid as client id
if (mqtt_set_client_id (mqtt) == NULL)
{
g_warning ("%s: Could not set client id.", __func__);
g_free (mqtt);
return -2;
}
mqtt_connect (mqtt, server_uri, username, password);
mqtt_client_publish (mqtt, topic, msg);
mqtt_disconnect (mqtt);
mqtt_client_destroy (mqtt);
mqtt_client_data_destroy (&mqtt);
return ret;
}
static int
mqtt_subscribe_r (mqtt_t *mqtt, int qos, const char *topic)
{
if (mqtt == NULL || mqtt->client == NULL)
{
return -1;
}
MQTTSubscribe_options opts = MQTTSubscribe_options_initializer;
MQTTProperties props = MQTTProperties_initializer;
MQTTResponse resp =
MQTTClient_subscribe5 (mqtt->client, topic, qos, &opts, &props);
if (resp.reasonCode != MQTTREASONCODE_GRANTED_QOS_1)
{
MQTTResponse_free (resp);
return -2;
}
MQTTResponse_free (resp);
return 0;
}
int
mqtt_subscribe (const char *topic)
{
if ((mqtt_get_global_client ()) == NULL)
mqtt_reinit ();
return mqtt_subscribe_r (mqtt_get_global_client (), QOS, topic);
}
static int
mqtt_unsubscribe_r (mqtt_t *mqtt, const char *topic)
{
if (mqtt == NULL || mqtt->client == NULL)
{
return -1;
}
if (MQTTClient_unsubscribe (mqtt->client, topic) != MQTTCLIENT_SUCCESS)
{
return -2;
}
return 0;
}
int
mqtt_unsubscribe (const char *topic)
{
return mqtt_unsubscribe_r (mqtt_get_global_client (), topic);
}
static int
mqtt_retrieve_message_r (mqtt_t *mqtt, char **topic, int *topic_len,
char **payload, int *payload_len,
const unsigned int timeout)
{
int rc = -1;
char *tmp = NULL;
MQTTClient_message *message = NULL;
if (mqtt == NULL || mqtt->client == NULL)
{
g_warning ("mqtt is not initialized.");
goto exit;
}
// copy from tmp into topic to make free work as usual and don't force the
// user to double check topic_len and topic
rc = MQTTClient_receive (mqtt->client, &tmp, topic_len, &message, timeout);
if (rc == MQTTCLIENT_SUCCESS || rc == MQTTCLIENT_TOPICNAME_TRUNCATED)
{
if (message)
{
g_debug ("%s: got message %s (%d) on topic %s (%d) \n", __func__,
(char *) message->payload, message->payloadlen, tmp,
*topic_len);
*topic = calloc (1, *topic_len);
if (*topic == NULL)
{
goto exit;
}
rc = 0;
if ((memcpy (*topic, tmp, *topic_len)) == NULL)
{
g_warning ("unable to copy topic");
rc = -1;
goto exit;
}
*payload_len = message->payloadlen;
*payload = calloc (1, message->payloadlen);
if ((memcpy (*payload, (char *) message->payload,
message->payloadlen))
== NULL)
{
g_warning ("unable to copy payload");
rc = -1;
goto exit;
}
}
else
{
rc = 1;
*payload = NULL;
*payload_len = 0;
*topic = NULL;
*topic_len = 0;
}
}
else
{
rc = -1;
}
exit:
if (message != NULL)
MQTTClient_freeMessage (&message);
if (tmp != NULL)
MQTTClient_free (tmp);
return rc;
}
int
mqtt_retrieve_message (char **topic, int *topic_len, char **payload,
int *payload_len, const unsigned int timeout)
{
return mqtt_retrieve_message_r (mqtt_get_global_client (), topic, topic_len,
payload, payload_len, timeout);
}
mqtt_publish_single_message_auth
int mqtt_publish_single_message_auth(const char *server_uri_in, const char *username_in, const char *passwd_in, const char *topic, const char *msg)
Send a single message with credentials.
Definition mqtt.c:550
G_LOG_DOMAIN
#define G_LOG_DOMAIN
Definition mqtt.c:35
QOS
#define QOS
Definition mqtt.c:37
mqtt_connect
static int mqtt_connect(mqtt_t *mqtt, const char *server_uri, const char *username, const char *password)
Make new client and connect to mqtt broker.
Definition mqtt.c:309
mqtt_get_global_server_uri
static const char * mqtt_get_global_server_uri()
Get global server URI.
Definition mqtt.c:91
mqtt_unsubscribe_r
static int mqtt_unsubscribe_r(mqtt_t *mqtt, const char *topic)
unsubscribe a single topic.
Definition mqtt.c:679
mqtt_set_global_password
static void mqtt_set_global_password(const char *password)
Set the global mqtt password.
Definition mqtt.c:122
mqtt_set_global_client
static void mqtt_set_global_client(mqtt_t *mqtt)
Set global client.
Definition mqtt.c:151
global_password
static const char * global_password
Definition mqtt.c:48
mqtt_set_client
static int mqtt_set_client(mqtt_t *mqtt, MQTTClient client)
Set MQTTClient of mqtt_t.
Definition mqtt.c:288
mqtt_disconnect
static int mqtt_disconnect(mqtt_t *mqtt)
Disconnect from the Broker.
Definition mqtt.c:164
TIMEOUT
#define TIMEOUT
Definition mqtt.c:38
mqtt_set_initialized_status
static void mqtt_set_initialized_status(gboolean status)
Set the global init status.
Definition mqtt.c:58
mqtt_init_auth
int mqtt_init_auth(const char *server_uri, const char *username, const char *password)
Init MQTT communication.
Definition mqtt.c:375
mqtt_publish_single_message
int mqtt_publish_single_message(const char *server_uri_in, const char *topic, const char *msg)
Send a single message.
Definition mqtt.c:527
mqtt_get_global_password
static const char * mqtt_get_global_password()
Get global password.
Definition mqtt.c:131
mqtt_is_initialized
gboolean mqtt_is_initialized()
Get the global init status.
Definition mqtt.c:69
mqtt_retrieve_message
int mqtt_retrieve_message(char **topic, int *topic_len, char **payload, int *payload_len, const unsigned int timeout)
wait for a given timeout in ms to retrieve any message of subscribed topics
Definition mqtt.c:837
mqtt_unsubscribe
int mqtt_unsubscribe(const char *topic)
unsubscribe a single topic.
Definition mqtt.c:705
global_server_uri
static const char * global_server_uri
Definition mqtt.c:46
mqtt_publish
int mqtt_publish(const char *topic, const char *msg)
Publish a message on topic using the global client.
Definition mqtt.c:498
mqtt_reset
void mqtt_reset()
Destroy MQTTClient handle and free mqtt_t.
Definition mqtt.c:212
mqtt_reinit
static void mqtt_reinit()
Reinitializes communication after mqtt_reset was used.
Definition mqtt.c:421
mqtt_client_destroy
static void mqtt_client_destroy(mqtt_t *mqtt)
Destroy the MQTTClient client of the mqtt_t.
Definition mqtt.c:186
mqtt_set_global_username
static void mqtt_set_global_username(const char *username)
Set the global mqtt username.
Definition mqtt.c:102
global_mqtt_client
static mqtt_t * global_mqtt_client
Definition mqtt.c:49
mqtt_init
int mqtt_init(const char *server_uri)
Init MQTT communication.
Definition mqtt.c:361
mqtt_get_global_client
static mqtt_t * mqtt_get_global_client()
Definition mqtt.c:142
mqtt_retrieve_message_r
static int mqtt_retrieve_message_r(mqtt_t *mqtt, char **topic, int *topic_len, char **payload, int *payload_len, const unsigned int timeout)
wait for a given timeout in ms to retrieve any message of subscribed topics
Definition mqtt.c:739
mqtt_create
static MQTTClient mqtt_create(mqtt_t *mqtt, const char *address)
Create a new mqtt client.
Definition mqtt.c:238
mqtt_set_client_id
static char * mqtt_set_client_id(mqtt_t *mqtt)
Set a random client ID.
Definition mqtt.c:269
mqtt_subscribe_r
static int mqtt_subscribe_r(mqtt_t *mqtt, int qos, const char *topic)
subscribes to a single topic.
Definition mqtt.c:625
mqtt_set_global_server_uri
static void mqtt_set_global_server_uri(const char *server_uri_in)
Set the global mqtt server URI.
Definition mqtt.c:80
mqtt_subscribe
int mqtt_subscribe(const char *topic)
subscribes to a single topic.
Definition mqtt.c:660
mqtt_initialized
static gboolean mqtt_initialized
Definition mqtt.c:50
mqtt_client_publish
static int mqtt_client_publish(mqtt_t *mqtt, const char *topic, const char *msg)
Use the provided client to publish message on a topic.
Definition mqtt.c:449
mqtt_get_global_username
static const char * mqtt_get_global_username()
Get global username.
Definition mqtt.c:111
global_username
static const char * global_username
Definition mqtt.c:47
mqtt_client_data_destroy
static void mqtt_client_data_destroy(mqtt_t **mqtt)
Destroy the mqtt_t data.
Definition mqtt.c:201
mqtt.h
Protos for MQTT handling.
MQTTClient_destroy
void MQTTClient_destroy(MQTTClient *client)
Definition mqtt_tests.c:15
mqtt_t
Definition mqtt.c:41
mqtt_t::client_id
char * client_id
Definition mqtt.c:43
mqtt_t::client
void * client
Definition mqtt.c:42
gvm_uuid_make
char * gvm_uuid_make(void)
Make a new universal identifier.
Definition uuidutils.c:30
uuidutils.h
UUID creation.