GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-18 21:50:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(mqtt): load the CA in memory

Browse Source
This commit is contained in:
gc87
2022-04-17 22:48:41 +08:00
parent 4b09f83a3d
commit 990f83c968
4 changed files with 77 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client_intf.h
+1 -1
View File
@@ -53,7 +53,7 @@ typedef struct {
int will_retain;
/* TLS options */
char *ca_path;
char *ca_file;
char *ca;
char *cert;
char *key;
char *keypass;
plugins/mqtt/mqtt.c
+2
View File
@@ -346,6 +346,8 @@ static void mqtt_routine_publish(mqtt_routine_t *routine)
}
free(result);
} else {
log_error_node(routine->plugin, "payload or topic is null");
}
}
plugins/mqtt/util.c
+12 -142
View File
@@ -81,9 +81,9 @@ void mqtt_option_uninit(neu_mqtt_option_t *option)
option->ca_path = NULL;
}
if (NULL != option->ca_file) {
free(option->ca_file);
option->ca_file = NULL;
if (NULL != option->ca) {
free(option->ca);
option->ca = NULL;
}
if (NULL != option->cert) {
@@ -97,33 +97,6 @@ void mqtt_option_uninit(neu_mqtt_option_t *option)
}
}
static int b64_file_store(const char *in, const char *file_name)
{
unsigned char *result = NULL;
int out_len = 0;
result = neu_decode64(&out_len, in);
if (NULL == result) {
return -1;
}
FILE *fp = fopen(file_name, "w+");
if (NULL == fp) {
free(result);
return -1;
}
size_t write_len = fwrite(result, sizeof(unsigned char), out_len, fp);
if (0 == write_len) {
free(result);
fclose(fp);
return -1;
}
free(result);
fclose(fp);
return 0;
}
int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
{
if (NULL == config || NULL == option) {
@@ -139,13 +112,9 @@ int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT };
neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR };
neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR };
neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR, .v.val_str = NULL };
neu_json_elem_t cert = { .name = "cert",
.t = NEU_JSON_STR,
.v.val_str = NULL };
neu_json_elem_t key = { .name = "key",
.t = NEU_JSON_STR,
.v.val_str = NULL };
neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR };
neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR };
neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR };
ret = neu_parse_param(config->buf, &error, 10, &id, &format, &ssl, &host,
&port, &username, &password, &ca, &cert, &key);
@@ -153,15 +122,6 @@ int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
return ret;
}
char ca_path[PATH_MAX + 1] = { 0 };
char ca_file[PATH_MAX + 1] = { 0 };
// char cert_file[PATH_MAX + 1] = { 0 };
// char key_file[PATH_MAX + 1] = { 0 };
// MQTT option
option->clientid = NULL; // Use random id
option->MQTT_version = 4; // Version 3.1.1
if (false == ssl.v.val_bool) {
option->connection = strdup("tcp://");
} else {
@@ -169,116 +129,26 @@ int mqtt_option_init(neu_config_t *config, neu_mqtt_option_t *option)
}
if (0 == strcmp(option->connection, "ssl://")) {
if (NULL == ca.v.val_str) {
return -1;
}
if (0 == strlen(ca.v.val_str)) {
free(ca.v.val_str);
return -1;
}
char workdir[PATH_MAX + 1] = { 0 };
char *ret_dir = getcwd(workdir, sizeof(workdir));
if (NULL == ret_dir) {
free(ca.v.val_str);
return -1;
}
ret = snprintf(ca_file, sizeof(ca_file), "%s/%s/%s", workdir,
MQTT_CERT_PATH, MQTT_CA_FILE);
if (ret < 0) {
free(ca.v.val_str);
return -1;
}
ret = snprintf(ca_path, sizeof(ca_path), "%s/%s", workdir,
MQTT_CERT_PATH);
if (ret < 0) {
free(ca.v.val_str);
return -1;
}
ret = b64_file_store(ca.v.val_str, ca_file);
free(ca.v.val_str);
ca.v.val_str = NULL;
if (0 != ret) {
return -1;
}
}
if (0 == strcmp(option->connection, "ssl://")) {
// char workdir[PATH_MAX + 1] = { 0 };
// char *ret_dir = getcwd(workdir, sizeof(workdir));
// if (NULL == ret_dir) {
// return -1;
// }
// if (NULL != cert.v.val_str && 0 < strlen(cert.v.val_str)) {
// ret = snprintf(cert_file, sizeof(cert_file), "%s/%s/%s", workdir,
// MQTT_CERT_PATH, MQTT_CERT_FILE);
// if (ret < 0) {
// free(cert.v.val_str);
// cert.v.val_str = NULL;
// return -1;
// }
// ret = b64_file_store(cert.v.val_str, cert_file);
// free(cert.v.val_str);
// cert.v.val_str = NULL;
// if (0 != ret) {
// return -1;
// }
// }
// if (NULL != key.v.val_str && 0 < strlen(key.v.val_str)) {
// ret = snprintf(key_file, sizeof(key_file), "%s/%s/%s", workdir,
// MQTT_CERT_PATH, MQTT_KEY_FILE);
// if (ret < 0) {
// free(key.v.val_str);
// key.v.val_str = NULL;
// return -1;
// }
// ret = b64_file_store(key.v.val_str, key_file);
// free(key.v.val_str);
// key.v.val_str = NULL;
// if (0 != ret) {
// return -1;
// }
// }
if (NULL != cert.v.val_str && 0 < strlen(cert.v.val_str)) { }
}
option->host = host.v.val_str;
int32_t p = (int) port.v.val_int;
option->port = calloc(10, sizeof(char));
// MQTT option
option->MQTT_version = 4; // Version 3.1.1
option->host = host.v.val_str;
int32_t p = (int) port.v.val_int;
option->port = calloc(10, sizeof(char));
snprintf(option->port, 10, "%d", p);
option->clientid = id.v.val_str;
option->format = format.v.val_int;
option->username = username.v.val_str;
option->password = password.v.val_str;
option->ca_path = strdup(ca_path);
option->ca_file = strdup(ca_file);
option->ca = ca.v.val_str;
option->cert = cert.v.val_str;
option->key = key.v.val_str;
option->keepalive_interval = 30;
option->clean_session = 1;
if (NULL != ca.v.val_str) {
free(ca.v.val_str);
}
// if (NULL != cert.v.val_str) {
// free(cert.v.val_str);
// }
// if (NULL != key.v.val_str) {
// free(key.v.val_str);
// }
return 0;
}
src/connection/mqtt_c_client.c
+62 -88
View File
@@ -87,125 +87,91 @@ static neu_err_code_e client_destroy(mqtt_c_client_t *client);
static neu_err_code_e client_subscribe_send(mqtt_c_client_t * client,
struct subscribe_tuple *tuple);
// static X509 *ssl_ctx_load_cert(const char *cert_file)
// {
// BIO * bio = NULL;
// X509 *cert = NULL;
// bio = BIO_new(BIO_s_file());
// if (NULL == bio) {
// return NULL;
// }
// int ret = BIO_read_filename(bio, cert_file);
// if (0 >= ret) {
// BIO_free_all(bio);
// return NULL;
// }
// cert = PEM_read_bio_X509(bio, NULL, 0, NULL);
// BIO_free_all(bio);
// return cert;
// }
static X509 *ssl_ctx_load_cert(const char *b64)
{
unsigned char *bin = NULL;
int out_len = 0;
bin = neu_decode64(&out_len, b64);
unsigned char *bin = NULL;
int len = 0;
bin = neu_decode64(&len, b64);
if (NULL == bin) {
return NULL;
}
X509 *cert = NULL;
BIO * bio = NULL;
bio = BIO_new_mem_buf((void *) bin, -1);
bio = BIO_new_mem_buf((void *) bin, len);
cert = PEM_read_bio_X509(bio, NULL, 0, NULL);
BIO_free_all(bio);
free(bin);
return cert;
}
// static EVP_PKEY *ssl_ctx_load_key(const char *key_file)
// {
// EVP_PKEY *key = NULL;
// FILE * fp = NULL;
// key = EVP_PKEY_new();
// fp = fopen(key_file, "r");
// if (NULL == fp) {
// EVP_PKEY_free(key);
// return NULL;
// }
// PEM_read_PrivateKey(fp, &key, NULL, NULL);
// fclose(fp);
// return key;
// }
static EVP_PKEY *ssl_ctx_load_key(const char *b64)
{
unsigned char *bin = NULL;
int out_len = 0;
bin = neu_decode64(&out_len, b64);
unsigned char *bin = NULL;
int len = 0;
bin = neu_decode64(&len, b64);
if (NULL == bin) {
return NULL;
}
EVP_PKEY *key = NULL;
BIO * bio = NULL;
bio = BIO_new_mem_buf((void *) bin, -1);
bio = BIO_new_mem_buf((void *) bin, len);
key = PEM_read_bio_PrivateKey(bio, NULL, 0, NULL);
BIO_free_all(bio);
free(bin);
return key;
}
bool ssl_ctx_load_cert_chain(SSL_CTX *context, const char *cert_buffer)
bool ssl_ctx_load_ca(SSL_CTX *ctx, const char *b64)
{
BIO *cbio = BIO_new_mem_buf((void *) cert_buffer, -1);
if (!cbio)
unsigned char *bin = NULL;
int len = 0;
bin = neu_decode64(&len, b64);
if (NULL == bin) {
return false;
}
BIO *cbio = BIO_new_mem_buf((void *) bin, len);
if (!cbio) {
free(bin);
return false;
}
X509_STORE *cts = SSL_CTX_get_cert_store(ctx);
if (!cts) {
free(bin);
return false;
}
X509_INFO *itmp = NULL;
int i = 0; //, count = 0, type = X509_FILETYPE_PEM;
int i = 0;
int count = 0;
STACK_OF(X509_INFO) *inf = PEM_X509_INFO_read_bio(cbio, NULL, NULL, NULL);
if (!inf) {
BIO_free(cbio);
free(bin);
return false;
}
bool first = true;
for (i = 0; i < sk_X509_INFO_num(inf); i++) {
itmp = sk_X509_INFO_value(inf, i);
if (itmp->x509) {
if (first) {
first = false;
X509_STORE_add_cert(cts, itmp->x509);
count++;
}
if (!SSL_CTX_use_certificate(context, itmp->x509)) {
goto Error;
}
if (ERR_peek_error() != 0) {
goto Error;
}
SSL_CTX_clear_chain_certs(context);
} else {
if (!SSL_CTX_add0_chain_cert(context, itmp->x509)) {
goto Error;
}
itmp->x509 = NULL;
}
if (itmp->crl) {
X509_STORE_add_crl(cts, itmp->crl);
count++;
}
}
sk_X509_INFO_pop_free(inf, X509_INFO_free);
BIO_free(cbio);
free(bin);
return true;
Error:
sk_X509_INFO_pop_free(inf, X509_INFO_free);
BIO_free(cbio);
return false;
}
static void ssl_ctx_init(struct reconnect_state *state)
@@ -213,26 +179,23 @@ static void ssl_ctx_init(struct reconnect_state *state)
assert(NULL != state);
const char *ca_file = state->ca_file;
const char *ca_path = state->ca_path;
const char *cert_file = state->cert_file;
const char *key_file = state->key_file;
assert(NULL != ca_file && 0 < strlen(ca_file));
assert(NULL != ca_path && 0 < strlen(ca_path));
SSL_load_error_strings();
ERR_load_BIO_strings();
ERR_load_crypto_strings();
OpenSSL_add_all_algorithms();
SSL_library_init();
state->ssl_ctx = SSL_CTX_new(SSLv23_client_method());
// setup cert
if (NULL != cert_file && 0 < strlen(cert_file)) {
state->cert = ssl_ctx_load_cert(cert_file);
if (NULL != state->cert) {
int rc = SSL_CTX_use_certificate(state->ssl_ctx, state->cert);
if (0 >= rc) {
if (0 >= SSL_CTX_use_certificate(state->ssl_ctx, state->cert)) {
log_error("failed to use certificate");
}
} else {
@@ -240,11 +203,11 @@ static void ssl_ctx_init(struct reconnect_state *state)
}
}
// setup key
if (NULL != key_file && 0 < strlen(key_file)) {
state->key = ssl_ctx_load_key(key_file);
if (NULL != state->key) {
int rc = SSL_CTX_use_PrivateKey(state->ssl_ctx, state->key);
if (0 >= rc) {
if (0 >= SSL_CTX_use_PrivateKey(state->ssl_ctx, state->key)) {
log_error("failed to use privatekey");
}
} else {
@@ -258,7 +221,8 @@ static void ssl_ctx_init(struct reconnect_state *state)
}
}
if (!SSL_CTX_load_verify_locations(state->ssl_ctx, ca_file, ca_path)) {
// setup ca from base64
if (!ssl_ctx_load_ca(state->ssl_ctx, ca_file)) {
log_error("failed to load ca");
}
}
@@ -312,11 +276,6 @@ static BIO *bio_ssl_create(SSL_CTX *ssl_ctx, const char *addr, const char *port)
BIO_set_conn_hostname(bio, addr);
BIO_set_nbio(bio, 1);
BIO_set_conn_port(bio, port);
if (SSL_get_verify_result(ssl) != X509_V_OK) {
log_error("error: x509 certificate verification failed");
}
return bio;
}
@@ -436,6 +395,21 @@ static void subscribe_on_reconnect(mqtt_c_client_t *client)
}
}
enum MQTTErrors inspector_callback(struct mqtt_client *mqtt)
{
struct reconnect_state *state = mqtt->reconnect_state;
if (0 == strcmp(state->connection, "ssl://")) {
SSL *ssl = NULL;
BIO_get_ssl(state->sock_fd, &ssl);
if (X509_V_OK != SSL_get_verify_result(ssl)) {
log_error("x509 certificate verification failed");
mqtt->error = MQTT_ERROR_SOCKET_ERROR;
}
}
return mqtt->error;
}
static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
{
assert(NULL != mqtt);
@@ -443,7 +417,8 @@ static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
struct reconnect_state *state =
*((struct reconnect_state **) p_reconnect_state);
mqtt_c_client_t *client = state->client;
mqtt_c_client_t *client = state->client;
mqtt->inspector_callback = inspector_callback;
bio_update_connection(state);
@@ -562,8 +537,7 @@ static mqtt_c_client_t *client_create(const neu_mqtt_option_t *option,
state->password = strdup(client->option->password);
state->keepalive = client->option->keepalive_interval;
state->clean = client->option->clean_session;
state->ca_file = strdup(client->option->ca_file);
state->ca_path = strdup(client->option->ca_path);
state->ca_file = strdup(client->option->ca);
state->cert_file = strdup(client->option->cert);
state->key_file = strdup(client->option->key);
state->client = client;