1、背景
由于项目有需求在一个现有的产品上增加MQTT通信的功能，且出于安全考虑，MQTT要走TLS，采用单向认证的方式。
2、方案选择
由于是在现有的产品上新增功能，那么为了减少总的成本，故选择只动应用软件的来实现需求。
MQTT的功能直接选择PahoMqtt这个第三方库来实现，因为以前用过，比较熟悉。由于只想动应用软件，那么只能选择他的embedded-c分支，这样才可以直接集成代码，而不需要编译成so放到固件里，同时也减少程序体积的增加。
embedded-c分支不支持TLS，那么就要想办法自己给embedded-c实现TLS的功能，经过考虑，现有的产品里有openssl库，故使用openssl来给embedded-c添加TLS的支持。
最终方案为：paho.mqtt.embedded-c实现MQTT的连接、断开和收\发，使用openssl来给embedded-c添加TLS的支持。
3、实现代码
step1：先去github把源码下载下来
GitHub - eclipse/paho.mqtt.embedded-c: Paho MQTT C client library for embedded systems. Paho is an Eclipse IoT project (https://iot.eclipse.org/)
step2：把对应的代码文件集成到项目里。
这里只需要MQTTPacket/src目录下的所有代码、MQTTClient-C/src/linux目录下的所有代码、MQTTClient-C/src目录下的MQTTClient.h和MQTTClient.c这两个文件就行了。具体内容如下图：
到这一步，paho.mqtt.embedded-c基本的就集成完了，对应的调用例子可以参考MQTTClient/samples/linux/main.cpp和MQTTClient-C/samples/linux/stdoutsub.c这两个文件，这里就不多赘述了。
step3：使用openssl给embedded-c增加TLS单向认证的支持
1、修改MQTTLinux.h文件
主要是修改点有两点：1、struct Network的内容，增加ssl的支持；2、新增两条NetworkConnect函数，一条用来走TCP，一条用来走TLS。（这里我选择保留最原始的NetworkConnect函数，方便以后debug的时候对比用。）
/*******************************************************************************
* Copyright (c) 2014 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Allan Stockdill-Mander - initial API and implementation and/or initial documentation
*******************************************************************************/
#if !defined(__MQTT_LINUX_)
#define __MQTT_LINUX_
#if defined(WIN32_DLL) || defined(WIN64_DLL)
#define DLLImport __declspec(dllimport)
#define DLLExport __declspec(dllexport)
#elif defined(LINUX_SO)
#define DLLImport extern
#define DLLExport __attribute__ ((visibility ("default")))
#else
#define DLLImport
#define DLLExport
#endif
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/param.h>
#include <sys/time.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
#include <openssl/evp.h>
typedef struct Timer
{
struct timeval end_time;
} Timer;
void TimerInit(Timer*);
char TimerIsExpired(Timer*);
void TimerCountdownMS(Timer*, unsigned int);
void TimerCountdown(Timer*, unsigned int);
int TimerLeftMS(Timer*);
typedef struct Network
{
/* socket的描述符 */
int my_socket;
/* 读取MQTT消息的函数 */
int (*mqttread) (struct Network*, unsigned char*, int, int);
/* 发送MQTT消息的函数 */
int (*mqttwrite) (struct Network*, unsigned char*, int, int);
/* 使用SSL时的描述符 */
SSL *ssl;
/* 是否使用SSL 0：否 1：是 */
int useSSL;
} Network;
int linux_read(Network*, unsigned char*, int, int);
int linux_write(Network*, unsigned char*, int, int);
/* 初始化Network这个结构体 */
DLLExport void NetworkInit(Network*);
/* 连接Network（普通socket） */
DLLExport int NetworkConnect(Network*, char*, int);
/* 连接Network（带SSL的socket） */
DLLExport int NetworkConnectBySSL(Network*, const char*, const char*, const char*);
/* 连接Network（不带SSL的socket） */
DLLExport int NetworkConnectNotSSL(Network*, const char*, const char*);
/* 断开Network的连接 */
DLLExport void NetworkDisconnect(Network*);
/* 打印证书的内容 */
DLLExport void ShowCerts(SSL *);
#endif
2、修改MQTTLinux.c文件
这里就把定义的那些给实现出来。
用Network这个结构体存放是否需要走ssl的信息，同时把SSL用的描述符也放在里面，方便后续传递。
调用不同的NetworkConnect的时候，在Network里面存储不同的信息
/*******************************************************************************
* Copyright (c) 2014, 2017 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Allan Stockdill-Mander - initial API and implementation and/or initial documentation
* Ian Craggs - return codes from linux_read
*******************************************************************************/
#include "MQTTLinux.h"
void TimerInit(Timer* timer)
{
timer->end_time = (struct timeval){0, 0};
}
char TimerIsExpired(Timer* timer)
{
struct timeval now, res;
gettimeofday(&now, NULL);
timersub(&timer->end_time, &now, &res);
return res.tv_sec < 0 || (res.tv_sec == 0 && res.tv_usec <= 0);
}
void TimerCountdownMS(Timer* timer, unsigned int timeout)
{
struct timeval now;
gettimeofday(&now, NULL);
struct timeval interval = {timeout / 1000, (timeout % 1000) * 1000};
timeradd(&now, &interval, &timer->end_time);
}
void TimerCountdown(Timer* timer, unsigned int timeout)
{
struct timeval now;
gettimeofday(&now, NULL);
struct timeval interval = {timeout, 0};
timeradd(&now, &interval, &timer->end_time);
}
int TimerLeftMS(Timer* timer)
{
struct timeval now, res;
gettimeofday(&now, NULL);
timersub(&timer->end_time, &now, &res);
//printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000);
return (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000;
}
/**
* 从MQTT读数据
*
* @param n : Network对象的指针
* @param buffer ： 存放数据的指针
* @param len ： 指针长度
* @param timeout_ms : 超时时间
* @return 读出的长度
*/
int linux_read(Network* n, unsigned char* buffer, int len, int timeout_ms)
{
struct timeval interval = {timeout_ms / 1000, (timeout_ms % 1000) * 1000};
if (interval.tv_sec < 0 || (interval.tv_sec == 0 && interval.tv_usec <= 0))
{
interval.tv_sec = 0;
interval.tv_usec = 100;
}
setsockopt(n->my_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval));
setsockopt(n->ssl, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval));
int bytes = 0;
while (bytes < len)
{
int rc = -1;
/* 是不是使用SSL决定了从哪个描述符读数据 */
if(n->useSSL==1){
rc = SSL_read(n->ssl, buffer+bytes, (len - bytes));
}else{
rc = read(n->my_socket, buffer+bytes, (len - bytes));
}
if (rc == -1)
{
if (errno != EAGAIN && errno != EWOULDBLOCK)
bytes = -1;
break;
}
else if (rc == 0)
{
bytes = 0;
break;
}
else
bytes += rc;
}
return bytes;
}
/**
* 写数据到mqtt
*
* @param n : Network对象的指针
* @param buffer ： 存放数据的指针
* @param len ： 数据长度
* @param timeout_ms : 超时时间
* @return 写出的长度
*/
int linux_write(Network* n, unsigned char* buffer, int len, int timeout_ms)
{
struct timeval tv;
tv.tv_sec = 0; /* 30 Secs Timeout */
tv.tv_usec = timeout_ms * 1000; // Not init'ing this can cause strange errors
setsockopt(n->my_socket, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv,sizeof(struct timeval));
int rc = -1;
/* 是不是使用SSL决定了从哪个描述符写数据 */
if(n->useSSL == 1){
rc = SSL_write(n->ssl, buffer, len);
}else{
rc = write(n->my_socket, buffer, len);
}
return rc;
}
/**
* 初始化Network对象
*
* @param n : Network对象的指针
*/
void NetworkInit(Network* n)
{
n->my_socket = 0;
n->mqttread = linux_read;
n->mqttwrite = linux_write;
n->useSSL = 0;
}
/**
* 原版的不带SSL的连接函数（这里改动了一点就是给useSSL赋值了）
*
* @param n : Network对象的指针
* @param addr ： MQTT服务器的IP地址
* @param port ： MQTT服务器的IP地址
* @return 0：成功 其他：失败
*/
int NetworkConnect(Network* n, char* addr, int port)
{
int type = SOCK_STREAM;
struct sockaddr_in address;
int rc = -1;
sa_family_t family = AF_INET;
struct addrinfo *result = NULL;
struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL};
if ((rc = getaddrinfo(addr, NULL, &hints, &result)) == 0)
{
struct addrinfo* res = result;
/* prefer ip4 addresses */
while (res)
{
if (res->ai_family == AF_INET)
{
result = res;
break;
}
res = res->ai_next;
}
if (result->ai_family == AF_INET)
{
address.sin_port = htons(port);
address.sin_family = family = AF_INET;
address.sin_addr = ((struct sockaddr_in*)(result->ai_addr))->sin_addr;
}
else
rc = -1;
freeaddrinfo(result);
}
if (rc == 0)
{
n->my_socket = socket(family, type, 0);
n->useSSL = 0;
if (n->my_socket != -1){
rc = connect(n->my_socket, (struct sockaddr*)&address, sizeof(address));
}
}
return rc;
}
/**
* 打印证书内容
*
* @param ssl ： SSL的描述符
*/
void ShowCerts(SSL * ssl)
{
X509 *cert;
char *line;
cert = SSL_get_peer_certificate(ssl);
if (cert != NULL) {
printf("数字证书信息:\n");
line = X509_NAME_oneline(X509_get_subject_name(cert), 0, 0);
printf("证书: %s\n", line);
free(line);
line = X509_NAME_oneline(X509_get_issuer_name(cert), 0, 0);
printf("颁发者: %s\n", line);
free(line);
X509_free(cert);
} else {
printf("无证书信息！\n");
}
}
/**
* 连接Network（带SSL的socket）
*
* @param n : Network对象的指针
* @param addr ： MQTT服务器的IP地址
* @param port ： MQTT服务器的IP地址
* @param crtFilePath ： 证书路径
* @return 0：成功 其他：失败
*/
int NetworkConnectBySSL(Network* n, const char* addr, const char* port, const char* crtFilePath)
{
SSL_CTX* ssl_context;
SSL *ssl;
/*SSL初始化*/
SSL_library_init();
OpenSSL_add_all_algorithms();
SSL_load_error_strings();
/* 创建SSL的上下文 */
ssl_context = SSL_CTX_new(TLSv1_2_client_method());
/*设置只验证服务器的证书*/
SSL_CTX_set_verify(ssl_context, SSL_VERIFY_PEER, NULL);
/* 设置用来进行校验的证书 */
if (SSL_CTX_use_certificate_file(ssl_context, crtFilePath, SSL_FILETYPE_PEM) != 1) {
SSL_CTX_free(ssl_context);
printf("Failed to load client certificate from %s", crtFilePath);
}
SSL_CTX_set_default_verify_paths(ssl_context);
/* 创建Socket并连接到服务器 */
int socketFd = -1;
if ((socketFd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
printf("create socket failed! %s", strerror(errno));
return -1;
}
/* 解析链接地址和消息 */
struct addrinfo hints = {};
struct addrinfo* serverInfo;
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
int result = getaddrinfo(addr, port, &hints, &serverInfo);
if (result != 0) {
printf("Failed to get addr info= %s addr=%s port=%s", gai_strerror(result),addr,port);
return -1;
}
/* 连接服务端socket */
if (connect(socketFd, serverInfo->ai_addr, serverInfo->ai_addrlen) != 0) {
printf("Connect socket failed! %s", strerror(errno));
return -1;
}
/* 让socket走SSL */
ssl = SSL_new(ssl_context);
SSL_set_fd(ssl, socketFd);
if (SSL_connect(ssl) == -1) {
ERR_print_errors_fp(stderr);
return -1;
} else {
printf("Connected with %s encryption\n", SSL_get_cipher(ssl));
/* 打印一下证书的内容 */
ShowCerts(ssl);
/* 检查一下证书 */
/* 把对应的描述符什么的存起来 */
n->my_socket = socketFd;
n->ssl = ssl;
n->useSSL = 1;
}
return 0;
}
/**
* 连接Network（不带SSL的socket）
*
* @param n : Network对象的指针
* @param addr ： MQTT服务器的IP地址
* @param port ： MQTT服务器的IP地址
* @return 0：成功 其他：失败
*/
int NetworkConnectNotSSL(Network* n, const char* addr, const char* port)
{
n->useSSL = 0;
/* 创建Socket并连接到服务器 */
int socketFd = -1;
if ((socketFd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
printf("create socket failed! %s", strerror(errno));
return -1;
}
/* 解析链接地址和消息 */
struct addrinfo hints = {};
struct addrinfo* serverInfo;
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
int result = getaddrinfo(addr, port, &hints, &serverInfo);
if (result != 0) {
printf("Failed to get addr info= %s addr=%s port=%s", gai_strerror(result),addr,port);
return -1;
}
/* 连接服务端socket */
if (connect(socketFd, serverInfo->ai_addr, serverInfo->ai_addrlen) != 0) {
printf("Connect socket failed! %s", strerror(errno));
return -1;
}else{
n->my_socket = socketFd;
return 0;
}
}
/**
* 断开连接
*
* @param n : Network对象的指针
*/
void NetworkDisconnect(Network* n)
{
close(n->my_socket);
if(n->useSSL == 1){
SSL_shutdown(n->ssl);
}
}
到这里我们就已经给embedded-c添加完TLS的支持，并使用的是单向校验
step4：发起MQTT连接连接
这里就直接贴调用代码好了
/**
* 连接MQTT
*
* @return 0：成功 其他：失败
*/
int connectMqtt() {
/* MQTT的发送缓冲区和接收缓冲区 */
uint8_t *sendbuf;
uint8_t *recvbuf;
/* MQTT的发送缓冲区和接收缓冲区的大小 */
const int MAX_SENDBUF_SIZE = 1024 * 30;
const int MAX_RECVBUF_SIZE = 1024 * 30;
/* MQTT的收发超时时间，单位ms */
const int MAX_MQTT_TIMEOUT_TIME = 1000;
/* 定义实现网络操作的工具 */
Network mqttNetworkUtil;
/* MQTT客户端操作类 */
MQTTClient mqttClient;
int rc = -1;
/* 初始化一下网络操作工具 */
NetworkInit(&mqttNetworkUtil);
/* 使用SSL的方式连接MQTT服务器的Socket */
NetworkConnectBySSL(&mqttNetworkUtil, G3_Configuration::getInstance().getNdsMqttConnectParams().host.c_str(),
std::to_string(G3_Configuration::getInstance().getNdsMqttConnectParams().port).c_str(), G3_Configuration::getInstance().getNdsMqttConnectParams().crtFilePath.c_str());
/* 初始化MQTT客户端 */
MQTTClientInit(&mqttClient, &mqttNetworkUtil, MAX_MQTT_TIMEOUT_TIME, sendbuf, MAX_SENDBUF_SIZE, recvbuf,
MAX_RECVBUF_SIZE);
/* 设置一下接收MQTT消息的回调函数，当MQTT收到消息后会自动回调 */
mqttClient.defaultMessageHandler = defaultMessageHandler;
/* 设置一下MQTT的一些连接参数 */
MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
data.willFlag = 1; // 启用遗嘱
data.will.topicName.cstring = G3_Configuration::getInstance().getNdsMqttConnectParams().willTopic.c_str(); // 设置遗嘱的topic名
data.will.qos = QOS2; // 遗嘱的QOS
data.will.message.cstring = G3_Configuration::getInstance().getNdsMqttConnectParams().clientId.c_str(); // 遗嘱的内容（这里使用clientId作为遗嘱的内容）
data.MQTTVersion = 3;
data.clientID.cstring = G3_Configuration::getInstance().getNdsMqttConnectParams().clientId.c_str();
data.username.cstring = G3_Configuration::getInstance().getNdsMqttConnectParams().userName.c_str();
data.password.cstring = G3_Configuration::getInstance().getNdsMqttConnectParams().password.c_str();
data.keepAliveInterval = 60;
data.cleansession = 1;
printf("Connecting to %s %d\n", G3_Configuration::getInstance().getNdsMqttConnectParams().host.c_str(),
G3_Configuration::getInstance().getNdsMqttConnectParams().port);
/* 发起MQTT的连接消息，前面只是socket，这个里发送MQTT协议中的连接消息，这里成功了才算MQTT连接成功 */
rc = MQTTConnect(&mqttClient, &data);
printf("MQTTConnect %d, Connect aliyun IoT Cloud Success!\n", rc);
/* 如果连接成功，就订阅一下用来接收消息的订阅字 */
if (rc == 0) {
printf("Subscribing to %s\n", G3_Configuration::getInstance().getNdsMqttConnectParams().subTopic.c_str());
rc = MQTTSubscribe(&mqttClient, G3_Configuration::getInstance().getNdsMqttConnectParams().subTopic.c_str(),
QOS2, defaultMessageHandler);
printf("MQTTSubscribe %d\n", rc);
}
return rc;
}
/**
* 接收到MQTT消息的时候的回调函数
*
* @param md ： MQTT消息的封装
*/
void messageArrived(MessageData *md) {
MQTTMessage *message = md->message;
callback->onGetDataFromMQTT(curMqttClientType, md->topicName->lenstring.data, md->topicName->lenstring.len,
(char *) message->payload, (int) message->payloadlen);
// printf("messageArrived %.*s\t", md->topicName->lenstring.len, md->topicName->lenstring.data);
// printf("messageArrived %.*s\n", (int) message->payloadlen, (char *) message->payload);
}
/**
* 发布消息
*
* @param mqttClient ： MQTT客户端的对象
* @param msg ： 封装好的消息
*
* @return 0：成功 其他:失败
*/
int publishMessage(MQTTClient &mqttClient,MQTTMessage &msg) {
return MQTTPublish(&mqttClient, G3_Configuration::getInstance().getNdsMqttConnectParams().pubTopic.c_str(), &msg);
}到这里连接也做完了，直接调用connectMqtt()就可以连接到MQTT了，需要发送调用publishMessage（）就可以发送了。如果MQTT那边有订阅的消息过来，那么会自动回messageArrived（）。这里涉及到一个结构体MQTTMessage，这里简单放下它怎么封装的
MQTTMessage msg = {
QOS2,//服务质量等级
0, //消息是否被保留。如果设置为1，则该消息将被代理服务器保留，这样任何新订阅该主题的客户端都可以立即接收到它。为0的话就算只有现在已经订阅了的能收到
0, //是否重复的消息。当设置为1时，它表示这是一个重复的消息。为0则为新消息
0, //消息ID，理论上每条都应该唯一，这里直接0好了
strData.c_str(), //负载数据，就是这条消息实际的内容
strData.size(), //负载数据的长度
};4、其他
1、如果想使用非TLS的MQTT连接，直接把connectMqtt()里面的NetworkConnectBySSL()改成NetworkConnectNotSSL()就好了；
2、如果想使用TLS但是不校验任何证书，那么就修改MQTTLinux.c里面的NetworkConnectBySSL()中SSL_CTX_set_verify（）的第二个参数就好了，把SSL_VERIFY_PEER改成SSL_VERIFY_NONE。这样就即走TLS，但是又不验证服务器的证书。
3、如果想使用双向认证，作为客户端我们这边什么都不用变，因为我们已经开启了认证服务器的证书，服务器那边需要开启认证客户端的证书就行了。
4、connectMqtt()里面填写的clientId和username和password也很重要，因为MQTT服务器可以开启用户认证模式，这样lientId和username和password有一个错误的话都会导致无法连接到服务器。
