How to use MQTT+TLS in nRF9160?

[Yusuke]

I'm working on a project using nRF9160 DK.

How can you use MQTT+TLS in nRF9160?

Where should ca certificate file be stored?

I modified the code below in mqtt_simple project. What code should I modify additionaly?

<prj.conf>
CONFIG_MQTT_LIB_TLS=y

<main.c>

/**@brief Initialize the MQTT client structure
 */
static void client_init(struct mqtt_client *client)
{
	mqtt_client_init(client);

	broker_init();

        /* Add from here */
        static struct mqtt_utf8 password;
	static struct mqtt_utf8 user_name;

	password.utf8 = (u8_t *)MQTT_PASSWORD;
	password.size = strlen(MQTT_PASSWORD);
	user_name.utf8 = (u8_t *)MQTT_USERNAME;
	user_name.size = strlen(MQTT_USERNAME);
        /* to here */

	/* MQTT client configuration */
	client->broker = &broker;
	client->evt_cb = mqtt_evt_handler;
	client->client_id.utf8 = (u8_t *)CONFIG_MQTT_CLIENT_ID;
	client->client_id.size = strlen(CONFIG_MQTT_CLIENT_ID);
	client->password = &password; // Add here
	client->user_name = &user_name; // Add here
	client->protocol_version = MQTT_VERSION_3_1_0; // originally MQTT_VERSION_3_1_1

	/* MQTT buffers configuration */
	client->rx_buf = rx_buffer;
	client->rx_buf_size = sizeof(rx_buffer);
	client->tx_buf = tx_buffer;
	client->tx_buf_size = sizeof(tx_buffer);

	/* MQTT transport configuration */ 
    /* MODIFIED HERE */
    #if defined(CONFIG_MQTT_LIB_TLS)
       client->transport.type = MQTT_TRANSPORT_SECURE;
       client->transport.tls.config.peer_verify = 0;
       client->transport.tls.config.cipher_count = 0;
       client->transport.tls.config.cipher_list = NULL;
       client->transport.tls.config.sec_tag_count = 0;
       client->transport.tls.config.seg_tag_list = NULL;
       client->transport.tls.config.hostname = NULL;
    #else
       client->transport.type = MQTT_TRANSPORT_NON_SECURE;
    #endif
}

[Kenta]

Hi !

prj.conf

# General config
CONFIG_TEST_RANDOM_GENERATOR=y
CONFIG_SPM=n
CONFIG_STDOUT_CONSOLE=y
CONFIG_NEWLIB_LIBC=y

# Networking
CONFIG_NETWORKING=y
CONFIG_NET_SOCKETS=y
CONFIG_NET_SOCKETS_OFFLOAD=y
CONFIG_NET_SOCKETS_POSIX_NAMES=y

# LTE link control
#CONFIG_POWER_OPTIMIZATION_ENABLE=n
CONFIG_LTE_LINK_CONTROL=y
CONFIG_LTE_AUTO_INIT_AND_CONNECT=n

# nRF Cloud
CONFIG_NRF_CLOUD_PROVISION_CERTIFICATES=y
#CONFIG_NRF_CLOUD=y

# Modem info
#CONFIG_MODEM_INFO=y

# BSD library
CONFIG_BSD_LIBRARY=y

# AT Host
CONFIG_UART_INTERRUPT_DRIVEN=y
CONFIG_AT_HOST_LIBRARY=y

# MQTT
#CONFIG_MQTT_LIB=y
CONFIG_MQTT_SOCKET_LIB=y
CONFIG_MQTT_LIB_TLS=y
CONFIG_MQTT_MAX_PACKET_LENGTH=2048


# Appliaction
#CONFIG_MQTT_PUB_TOPIC="my/publish/topic"
#CONFIG_MQTT_SUB_TOPIC="my/subscribe/topic"

CONFIG_MQTT_PUB_TOPIC="myTopic/publish"
CONFIG_MQTT_SUB_TOPIC="myTopic/subscribe"
CONFIG_MQTT_CLIENT_ID="nRF9160-DK"
CONFIG_MQTT_BROKER_HOSTNAME="a544w27l82h92-ats.iot.us-east-1.amazonaws.com"
CONFIG_MQTT_BROKER_PORT= 8883

CONFIG_NRF_CLOUD_SEC_TAG=1234



# Main thread
CONFIG_MAIN_THREAD_PRIORITY=7
#CONFIG_MAIN_STACK_SIZE=4096
#CONFIG_HEAP_MEM_POOL_SIZE=1024

CONFIG_HEAP_MEM_POOL_SIZE=16384
CONFIG_MAIN_STACK_SIZE=8192
CONFIG_GPS_SIM_THREAD_STACK_SIZE=1024
CONFIG_SYSTEM_WORKQUEUE_STACK_SIZE=1500
CONFIG_HW_STACK_PROTECTION=y

main.c

/*
 * Copyright (c) 2018 Nordic Semiconductor ASA
 *
 * SPDX-License-Identifier: LicenseRef-BSD-5-Clause-Nordic
 */

#include <zephyr.h>
#include <stdio.h>
#include <uart.h>
#include <string.h>

#include <net/mqtt.h>
#include <net/socket.h>
#include <lte_lc.h>

#if defined(CONFIG_BSD_LIBRARY)
#include "nrf_inbuilt_key.h"
#endif

#include "certificates.h"

/*
 * Copyright (c) 2018 Nordic Semiconductor ASA
 *
 * SPDX-License-Identifier: BSD-5-Clause-Nordic
 */

//#define CONFIG_NRF_CLOUD_SEC_TAG 1
#define NRF_CLOUD_HOSTNAME CONFIG_MQTT_BROKER_HOSTNAME
//#define NRF_CLOUD_SEC_TAG  CONFIG_NRF_CLOUD_SEC_TAG

static struct nct {
    struct mqtt_sec_config tls_config;
    struct mqtt_client client;
    struct sockaddr_storage broker;
    struct mqtt_utf8 dc_tx_endp;
    struct mqtt_utf8 dc_rx_endp;
    u32_t message_id;
} nct;


/* Buffers for MQTT client. */
static u8_t rx_buffer[CONFIG_MQTT_MESSAGE_BUFFER_SIZE];
static u8_t tx_buffer[CONFIG_MQTT_MESSAGE_BUFFER_SIZE];
static u8_t payload_buf[CONFIG_MQTT_PAYLOAD_BUFFER_SIZE];

/* The mqtt client struct */
static struct mqtt_client client;

/* MQTT Broker details. */
static struct sockaddr_storage broker;

/* Connected flag */
static bool connected;

/* File descriptor */
static struct pollfd fds;

#if defined(CONFIG_BSD_LIBRARY)

/**@brief Recoverable BSD library error. */
void bsd_recoverable_error_handler(uint32_t err)
{
	printk("bsdlib recoverable error: %u\n", err);
}

/**@brief Irrecoverable BSD library error. */
void bsd_irrecoverable_error_handler(uint32_t err)
{
	printk("bsdlib irrecoverable error: %u\n", err);

	__ASSERT_NO_MSG(false);
}

#endif /* defined(CONFIG_BSD_LIBRARY) */

/**@brief Function to print strings without null-termination
 */
static void data_print(u8_t *prefix, u8_t *data, size_t len)
{
	char buf[len + 1];

	memcpy(buf, data, len);
	buf[len] = 0;
	printk("%s%s\n", prefix, buf);
}

static void data_sub_print(u8_t *prefix, u8_t *data, size_t len)
{
    printk("data_sub_print\n");
    printk("%s%s\n", prefix, data);
    printk("data receive end.\n");
    //printk("%s", data);
    //for(int i=0; i<len; i++){
    //    printk("%d", data[i]);
    //}
    //printk("\n");
    
}

/**@brief Function to publish data on the configured topic
 */
static int data_publish(struct mqtt_client *c, enum mqtt_qos qos,
	u8_t *data, size_t len)
{
	struct mqtt_publish_param param;

	param.message.topic.qos = qos;
	param.message.topic.topic.utf8 = CONFIG_MQTT_PUB_TOPIC;
	param.message.topic.topic.size = strlen(CONFIG_MQTT_PUB_TOPIC);
	param.message.payload.data = data;
	param.message.payload.len = len;
	param.message_id = sys_rand32_get();
	param.dup_flag = 0;
	param.retain_flag = 0;

	data_print("Publishing: ", data, len);
	printk("to topic: %s len: %u\n",
		CONFIG_MQTT_PUB_TOPIC,
		(unsigned int)strlen(CONFIG_MQTT_PUB_TOPIC));

	return mqtt_publish(c, &param);
}

/**@brief Function to subscribe to the configured topic
 */
static int subscribe(void)
{
	struct mqtt_topic subscribe_topic = {
		.topic = {
			.utf8 = CONFIG_MQTT_SUB_TOPIC,
			.size = strlen(CONFIG_MQTT_SUB_TOPIC)
		},
		.qos = MQTT_QOS_1_AT_LEAST_ONCE
	};

	const struct mqtt_subscription_list subscription_list = {
		.list = &subscribe_topic,
		.list_count = 1,
		.message_id = 1234
	};

	printk("Subscribing to: %s len %u\n", CONFIG_MQTT_SUB_TOPIC,
		(unsigned int)strlen(CONFIG_MQTT_SUB_TOPIC));

	return mqtt_subscribe(&client, &subscription_list);
}

/**@brief Function to read the published payload.
 */
static int publish_get_payload(struct mqtt_client *c, size_t length)
{
	u8_t *buf = payload_buf;
	u8_t *end = buf + length;

	if (length > sizeof(payload_buf)) {
		return -EMSGSIZE;
	}

	while (buf < end) {
		//int ret = mqtt_read_publish_payload(c, buf, end - buf);
                int ret = 0;
		if (ret < 0) {
			int err;

			if (ret != -EAGAIN) {
				return ret;
			}

			printk("mqtt_read_publish_payload: EAGAIN\n");

			err = poll(&fds, 1, K_SECONDS(CONFIG_MQTT_KEEPALIVE));
			if (err > 0 && (fds.revents & POLLIN) == POLLIN) {
				continue;
			} else {
				return -EIO;
			}
		}

		if (ret == 0) {
			return -EIO;
		}

		buf += ret;
	}

	return 0;
}

//static int subscribe_get_payload(struct mqtt_binstr)
//{
//
//}

/**@brief MQTT client event handler
 */
void mqtt_evt_handler(struct mqtt_client *const c,
		      const struct mqtt_evt *evt)
{
	int err;

	switch (evt->type) {
	case MQTT_EVT_CONNACK:
		if (evt->result != 0) {
			printk("MQTT connect failed %d\n", evt->result);
			break;
		}

		connected = true;
		printk("[%s:%d] MQTT client connected!\n", __func__, __LINE__);
		subscribe();
		break;

	case MQTT_EVT_DISCONNECT:
		printk("[%s:%d] MQTT client disconnected %d\n", __func__,
		       __LINE__, evt->result);

		connected = false;
		break;

	case MQTT_EVT_PUBLISH: {
		const struct mqtt_publish_param *p = &evt->param.publish;

		printk("[%s:%d] MQTT PUBLISH result=%d len=%d\n", __func__,
		       __LINE__, evt->result, p->message.payload.len);
		
//                err = publish_get_payload(c, p->message.payload.len);
//		if (err >= 0) {
//			data_print("Received: ", payload_buf,
//				p->message.payload.len);
//			/* Echo back received data */
//			//data_publish(&client, MQTT_QOS_1_AT_LEAST_ONCE,
//			//	payload_buf, p->message.payload.len);
//		} else {
//			printk("mqtt_read_publish_payload: Failed! %d\n", err);
//			printk("Disconnecting MQTT client...\n");
//
//			err = mqtt_disconnect(c);
//			if (err) {
//				printk("Could not disconnect: %d\n", err);
//			}
//		}
	} break;

	case MQTT_EVT_PUBACK:
		if (evt->result != 0) {
			printk("MQTT PUBACK error %d\n", evt->result);
			break;
		}

		printk("[%s:%d] PUBACK packet id: %u\n", __func__, __LINE__,
				evt->param.puback.message_id);
		break;

	case MQTT_EVT_SUBACK:
		if (evt->result != 0) {
			printk("MQTT SUBACK error %d\n", evt->result);
			break;
		}

		printk("[%s:%d] SUBACK packet id: %u\n", __func__, __LINE__,
				evt->param.suback.message_id);
        
                const struct mqtt_suback_param* suback = &evt->param.suback;
        //suback->return_codes->data
                data_sub_print("Subscribe: ", suback->return_codes.data, suback->return_codes.len);
		break;

	default:
		printk("[%s:%d] default: %d\n", __func__, __LINE__,
				evt->type);
		break;
	}
}

/**@brief Resolves the configured hostname and
 * initializes the MQTT broker structure
 */
static void broker_init(void)
{
	int err;
	struct addrinfo *result;
	struct addrinfo *addr;
	struct addrinfo hints = {
		.ai_family = AF_INET,
		.ai_socktype = SOCK_STREAM
	};

	err = getaddrinfo(CONFIG_MQTT_BROKER_HOSTNAME, NULL, &hints, &result);
	if (err) {
		printk("ERROR: getaddrinfo failed %d\n", err);

		return;
	}

	addr = result;
	err = -ENOENT;

	/* Look for address of the broker. */
	while (addr != NULL) {
		/* IPv4 Address. */
		if (addr->ai_addrlen == sizeof(struct sockaddr_in)) {
			struct sockaddr_in *broker4 =
				((struct sockaddr_in *)&broker);

			broker4->sin_addr.s_addr =
				((struct sockaddr_in *)addr->ai_addr)
				->sin_addr.s_addr;
			broker4->sin_family = AF_INET;
			broker4->sin_port = htons(CONFIG_MQTT_BROKER_PORT);
			printk("IPv4 Address found 0x%08x\n",
				broker4->sin_addr.s_addr);
			break;
		} else {
			printk("ai_addrlen = %u should be %u or %u\n",
				(unsigned int)addr->ai_addrlen,
				(unsigned int)sizeof(struct sockaddr_in),
				(unsigned int)sizeof(struct sockaddr_in6));
		}

		addr = addr->ai_next;
		break;
	}

	/* Free the address. */
	freeaddrinfo(result);
}

/**@brief Initialize the MQTT client structure
 */
static void client_init(struct mqtt_client *client)
{
	mqtt_client_init(client);

	broker_init();

	/* MQTT client configuration */
	client->broker = &broker;
	client->evt_cb = mqtt_evt_handler;
	client->client_id.utf8 = (u8_t *)CONFIG_MQTT_CLIENT_ID;
	client->client_id.size = strlen(CONFIG_MQTT_CLIENT_ID);
	client->password = NULL;
	client->user_name = NULL;
	client->protocol_version = MQTT_VERSION_3_1_1;

	/* MQTT buffers configuration */
	client->rx_buf = rx_buffer;
	client->rx_buf_size = sizeof(rx_buffer);
	client->tx_buf = tx_buffer;
	client->tx_buf_size = sizeof(tx_buffer);

	/* MQTT transport configuration */
	//client->transport.type = MQTT_TRANSPORT_NON_SECURE;
#if defined(CONFIG_MQTT_LIB_TLS)

    //nct.client.transport.type = MQTT_TRANSPORT_SECURE;
    //struct mqtt_sec_config *tls_config = &nct.client.transport.tls.config;
    //memcpy(tls_config, &nct.tls_config, sizeof(struct mqtt_sec_config));

    client->transport.type = MQTT_TRANSPORT_SECURE;
    
//    static sec_tag_t sec_tag_list[] = { CONFIG_MY_TAG };
//    struct mqtt_sec_config* tls_config = &client->transport.tls.config;
//
//    tls_config->peer_verify = 2;
//    tls_config->cipher_list = NULL;
//    tls_config->sec_tag_list = sec_tag_list;
//    tls_config->sec_tag_count = ARRAY_SIZE(sec_tag_list);
//    tls_config->hostname = CONFIG_MQTT_BROKER_HOSTNAME;
    
    struct mqtt_sec_config *tls_config = &nct.client.transport.tls.config;
    memcpy(tls_config, &nct.tls_config, sizeof(struct mqtt_sec_config));

#else
    client->transport.type = MQTT_TRANSPORT_NON_SECURE;
#endif
    
}

/**@brief Initialize the file descriptor structure used by poll.
 */
static int fds_init(struct mqtt_client *c)
{
	if (c->transport.type == MQTT_TRANSPORT_NON_SECURE) {
		fds.fd = c->transport.tcp.sock;
	} else {
#if defined(CONFIG_MQTT_LIB_TLS)
		fds.fd = c->transport.tls.sock;
#else
		return -ENOTSUP;
#endif
	}

	fds.events = POLLIN;

	return 0;
}

/**@brief Configures modem to provide LTE link. Blocks until link is
 * successfully established.
 */
static void modem_configure(void)
{
#if defined(CONFIG_LTE_LINK_CONTROL)
	if (IS_ENABLED(CONFIG_LTE_AUTO_INIT_AND_CONNECT)) {
		/* Do nothing, modem is already turned on
		 * and connected.
		 */
	} else {
		int err;

		printk("LTE Link Connecting ...\n");
		err = lte_lc_init_and_connect();
		__ASSERT(err == 0, "LTE link could not be established.");
		printk("LTE Link Connected!\n");
	}
#endif
}


/* Provisions root CA certificate using nrf_inbuilt_key API */
static int nct_provision(void)
{
    static sec_tag_t sec_tag_list[] = {CONFIG_NRF_CLOUD_SEC_TAG};

    nct.tls_config.peer_verify = 2;
    nct.tls_config.cipher_count = 0;
    nct.tls_config.cipher_list = NULL;
    nct.tls_config.sec_tag_count = ARRAY_SIZE(sec_tag_list);
    nct.tls_config.sec_tag_list = sec_tag_list;
    nct.tls_config.hostname = NRF_CLOUD_HOSTNAME;

#if defined(CONFIG_NRF_CLOUD_PROVISION_CERTIFICATES)
#if defined(CONFIG_BSD_LIBRARY)
    {
        int err;

        /* Delete certificates */
        nrf_sec_tag_t sec_tag = CONFIG_NRF_CLOUD_SEC_TAG;

        for (nrf_key_mgnt_cred_type_t type = 0; type < 5; type++) {
            err = nrf_inbuilt_key_delete(sec_tag, type);
            printk("nrf_inbuilt_key_delete(%lu, %d) => result=%d",
                sec_tag, type, err);
        }

        /* Provision CA Certificate. */
        err = nrf_inbuilt_key_write(CONFIG_NRF_CLOUD_SEC_TAG,
                    NRF_KEY_MGMT_CRED_TYPE_CA_CHAIN,
                    NRF_CLOUD_CA_CERTIFICATE,
                    strlen(NRF_CLOUD_CA_CERTIFICATE));
        if (err) {
            printk("NRF_CLOUD_CA_CERTIFICATE err: %d", err);
            return err;
        }

        /* Provision Private Certificate. */
        err = nrf_inbuilt_key_write(
            CONFIG_NRF_CLOUD_SEC_TAG,
            NRF_KEY_MGMT_CRED_TYPE_PRIVATE_CERT,
            NRF_CLOUD_CLIENT_PRIVATE_KEY,
            strlen(NRF_CLOUD_CLIENT_PRIVATE_KEY));
        if (err) {
            printk("NRF_CLOUD_CLIENT_PRIVATE_KEY err: %d", err);
            return err;
        }

        /* Provision Public Certificate. */
        err = nrf_inbuilt_key_write(
            CONFIG_NRF_CLOUD_SEC_TAG,
            NRF_KEY_MGMT_CRED_TYPE_PUBLIC_CERT,
            NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE,
            strlen(NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE));
        if (err) {
            printk("NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE err: %d",
                err);
            return err;
        }
    }
#else
    {
        int err;

        err = tls_credential_add(CONFIG_NRF_CLOUD_SEC_TAG,
            TLS_CREDENTIAL_CA_CERTIFICATE,
            NRF_CLOUD_CA_CERTIFICATE,
            sizeof(NRF_CLOUD_CA_CERTIFICATE));
        if (err < 0) {
            LOG_ERR("Failed to register ca certificate: %d",
                err);
            return err;
        }
        err = tls_credential_add(CONFIG_NRF_CLOUD_SEC_TAG,
            TLS_CREDENTIAL_PRIVATE_KEY,
            NRF_CLOUD_CLIENT_PRIVATE_KEY,
            sizeof(NRF_CLOUD_CLIENT_PRIVATE_KEY));
        if (err < 0) {
            LOG_ERR("Failed to register private key: %d",
                err);
            return err;
        }
        err = tls_credential_add(CONFIG_NRF_CLOUD_SEC_TAG,
            TLS_CREDENTIAL_SERVER_CERTIFICATE,
            NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE,
            sizeof(NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE));
        if (err < 0) {
            LOG_ERR("Failed to register public certificate: %d",
                err);
            return err;
        }

    }
#endif /* defined(CONFIG_BSD_LIBRARY) */
#endif /* defined(CONFIG_NRF_CLOUD_PROVISION_CERTIFICATES) */

    return 0;
}


void main(void)
{
	int err;

	printk("The MQTT simple sample started\n");
        k_sleep(1000);

    int err_provision = nct_provision();
    if (err_provision != 0) {
        printk("ERROR: nct_provision failure %d\n", err_provision);
        return;
    }
    printk("err_provision = %d\n", err_provision);

	modem_configure();

	client_init(&client);

	err = mqtt_connect(&client);
	if (err != 0) {
		printk("ERROR: mqtt_connect %d\n", err);
		return;
	}
        printk("connect MQTT Broker.\r\n");

	err = fds_init(&client);
	if (err != 0) {
		printk("ERROR: fds_init %d\n", err);
		return;
	}

	while (1) {
		err = poll(&fds, 1, K_SECONDS(CONFIG_MQTT_KEEPALIVE));
		if (err < 0) {
			printk("ERROR: poll %d\n", errno);
			break;
		}

		err = mqtt_live(&client);
		if (err != 0) {
			printk("ERROR: mqtt_live %d\n", err);
			break;
		}

		if ((fds.revents & POLLIN) == POLLIN) {
			err = mqtt_input(&client);
			if (err != 0) {
				printk("ERROR: mqtt_input %d\n", err);
				break;
			}
		}

		if ((fds.revents & POLLERR) == POLLERR) {
			printk("POLLERR\n");
			break;
		}

		if ((fds.revents & POLLNVAL) == POLLNVAL) {
			printk("POLLNVAL\n");
			break;
		}

//                subscribe();
//                k_sleep(K_MSEC(10000));
	}

	printk("Disconnecting MQTT client...\n");

	err = mqtt_disconnect(&client);
	if (err) {
		printk("Could not disconnect MQTT client. Error: %d\n", err);
	}
}

But because it does not work, we are considering other ways already....