/* * Copyright (c) 2018 Nordic Semiconductor ASA * * SPDX-License-Identifier: LicenseRef-BSD-5-Clause-Nordic */ #include #include #include #include #include #include #include #include "certificates.h" #if defined(CONFIG_PROVISION_CERTIFICATES) #if defined(CONFIG_BSD_LIBRARY) #include "nrf_inbuilt_key.h" #endif #endif #if defined CONFIG_MQTT_LIB_TLS static sec_tag_t sec_tag_list[] = { CONFIG_SEC_TAG }; #endif #define AZ_Pass \ "azureiothubpass" #define AZ_User_Name \ "azureiothubusername" #define Test_Pass "XXXXXXXXX" #define Test_User_Name "XXXXXXXXXX" /* 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; /*Client username/pass*/ struct mqtt_utf8 pass, user_name; /* 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); } /**@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, ¶m); } /**@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); 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; } /**@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); //this is executed...why? 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); 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); char ipv4_addr[NET_IPV4_ADDR_LEN]; 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); inet_ntop(AF_INET, &broker4->sin_addr.s_addr, ipv4_addr, sizeof(ipv4_addr)); printk("IPv4 Address found %s\n", ipv4_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); pass.size = strlen(AZ_Pass); pass.utf8 = (u8_t *)AZ_Pass; user_name.size = strlen(AZ_User_Name); user_name.utf8 = (u8_t *)AZ_User_Name; 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 = &pass; client->user_name = &user_name; 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 */ #if defined(CONFIG_MQTT_LIB_TLS) client->transport.type = MQTT_TRANSPORT_SECURE; client->transport.tls.config.peer_verify = 2; //formerly 0 client->transport.tls.config.cipher_count = 0; client->transport.tls.config.cipher_list = NULL; client->transport.tls.config.sec_tag_count = ARRAY_SIZE(sec_tag_list); // formerly 0 client->transport.tls.config.sec_tag_list = sec_tag_list; //formerly NULL client->transport.tls.config.hostname = CONFIG_MQTT_BROKER_HOSTNAME; #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 } static int provision_certificates(void) { { #if defined(CONFIG_PROVISION_CERTIFICATES) #if defined(CONFIG_BSD_LIBRARY) int err; /* Delete Certificates */ nrf_sec_tag_t sec_tag = (nrf_sec_tag_t)sec_tag_list[0]; for (nrf_key_mgnt_cred_type_t type = 0; type < 5; type++) { printk("Deleting certs sec_tag: %d\n", sec_tag); err = nrf_inbuilt_key_delete(sec_tag, type); printk("nrf_inbuilt_key_delete(%u, %d) => result%d\n", sec_tag, type, err); } #if defined(AZ_ROOT_CERTIFICATE) /*provision CA certificates*/ printk("Write ca certs sec_tag: %d\n", sec_tag); err = nrf_inbuilt_key_write(sec_tag, NRF_KEY_MGMT_CRED_TYPE_CA_CHAIN, AZ_ROOT_CERTIFICATE, strlen(AZ_ROOT_CERTIFICATE)); if (err) { printk("AZ_ROOT_CERTIFICATE err: %d", err); return err; } #endif //end az_root_certificate #if defined(CLIENT_PRIVATE_KEY) /* provision private certificate */ printk("Write private cert sec_tag: %d\n", sec_tag); err = nrf_inbuilt_key_write(sec_tag, NRF_KEY_MGMT_CRED_TYPE_PRIVATE_CERT, CLIENT_PRIVATE_KEY, strlen(CLIENT_PRIVATE_KEY)); if (err) { printk("CLIENT_PRIVATE_KEY err: %d\n", err); return err; } #endif //client_private_key #if defined(CLIENT_PUBLIC_CERTIFICATE) /* provision public certificate */ printk("Write public cert sec_tag: %d\n", sec_tag); err = nrf_inbuilt_key_write(sec_tag, NRF_KEY_MGMT_CRED_TYPE_PUBLIC_CERT, CLIENT_PUBLIC_CERTIFICATE, strlen(CLIENT_PUBLIC_CERTIFICATE)); if (err) { printk("CLIENT_PUBLIC_CERTIFICATE err: %d\n", err); return err; } } #endif //client_public_certificate #else { int err; err = tls_credential_add(CONFIG_SEC_TAG, TLS_CREDENTIAL_CA_CERTIFICATE, NRF_CLOUD_CA_CERTIFICATE, sizeof(NRF_CLOUD_CA_CERTIFICATE)); if (err < 0) { printk("Failed to register ca certificate: %d\n", err); return err; } err = tls_credential_add(CONFIG_SEC_TAG, TLS_CREDENTIAL_PRIVATE_KEY, NRF_CLOUD_CLIENT_PRIVATE_KEY, sizeof(NRF_CLOUD_CLIENT_PRIVATE_KEY)); if (err < 0) { printk("Failed to register private key: %d\n", err); return err; } err = tls_credential_add( CONFIG_SEC_TAG, TLS_CREDENTIAL_SERVER_CERTIFICATE, NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE, sizeof(NRF_CLOUD_CLIENT_PUBLIC_CERTIFICATE)); if (err < 0) { printk("Failed to register public certificate: %d\n", err); return err; } #endif //defined(config_bsd_library) #endif //defined(config_provision_certificates) } return 0; } void main(void) { int err; printk("The MQTT simple sample started\n"); provision_certificates(); modem_configure(); client_init(&client); err = mqtt_connect(&client); if (err != 0) { printk("ERROR: mqtt_connect %d\n", err); return; } err = fds_init(&client); if (err != 0) { printk("ERROR: fds_init %d\n", err); return; } printk("NOW ENTERING ENDLESS POLLING LOOP\n"); 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; } } printk("Disconnecting MQTT client...\n"); err = mqtt_disconnect(&client); if (err) { printk("Could not disconnect MQTT client. Error: %d\n", err); } }