From 990f83c968df93ae18f1bd5e4ea34458eebfefeb Mon Sep 17 00:00:00 2001 From: gc87 Date: Sun, 17 Apr 2022 22:48:41 +0800 Subject: [PATCH] refactor(mqtt): load the CA in memory --- include/neuron/connection/mqtt_client_intf.h | 2 +- plugins/mqtt/mqtt.c | 2 + plugins/mqtt/util.c | 154 ++----------------- src/connection/mqtt_c_client.c | 150 ++++++++---------- 4 files changed, 77 insertions(+), 231 deletions(-) diff --git a/include/neuron/connection/mqtt_client_intf.h b/include/neuron/connection/mqtt_client_intf.h index c7194f8c..ea565f9b 100644 --- a/include/neuron/connection/mqtt_client_intf.h +++ b/include/neuron/connection/mqtt_client_intf.h @@ -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; diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index fc73a0b2..a636bb56 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -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"); } } diff --git a/plugins/mqtt/util.c b/plugins/mqtt/util.c index 882c6b6e..90de792e 100644 --- a/plugins/mqtt/util.c +++ b/plugins/mqtt/util.c @@ -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; } diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index 99fde134..4eb55f64 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -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;