@cryptotaxi247 / netdata-1 / commits / 3994ea0ad

Support IPV6 when establishing MQTT connection to the cloud (#18217)

* Remove ws_client_get_nonce, use uuid instead Use connect_to_this_ip46 function Replace mw_malloc, mw_calloc, mw_free, mw_strdup, mw_realloc Cleanup * Remove unused pub_lock Switch stat lock to spinlock * Set correct host

Stelios Fragkakis committed Jul 30, 2024 at 12:47 UTC 3994ea0adaa20152e0f94fa39bc8d4a39f5fc732

6 files changed +139 -169

src/aclk/helpers/mqtt_wss_pal.h

-6

@@ -10,10 +10,4 @@

10 #undef OPENSSL_VERSION_110

11 #undef OPENSSL_VERSION_111

13 -#define mw_malloc(...) mallocz(__VA_ARGS__)

14 -#define mw_calloc(...) callocz(__VA_ARGS__)

15 -#define mw_free(...) freez(__VA_ARGS__)

16 -#define mw_strdup(...) strdupz(__VA_ARGS__)

17 -#define mw_realloc(...) reallocz(__VA_ARGS__)

18 -

13 #endif /* MQTT_WSS_PAL_H */

src/aclk/https_client.c

+1 -1

@@ -696,7 +696,7 @@ int https_request(https_req_t *request, https_req_response_t *response) {

696 goto exit_CTX;

697 }

698

699 - if (!SSL_set_tlsext_host_name(ctx->ssl, connect_host)) {

699 + if (!SSL_set_tlsext_host_name(ctx->ssl, request->host)) {

700 netdata_log_error("Error setting TLS SNI host");

701 goto exit_CTX;

702 }

src/aclk/mqtt_websockets/mqtt_ng.c

+28 -28

@@ -423,7 +423,7 @@ static void buffer_frag_free_data(struct buffer_fragment *frag)

423 if ( frag->flags & BUFFER_FRAG_DATA_EXTERNAL && frag->data != NULL) {

424 switch (ptr2memory_mode(frag->free_fnc)) {

425 case MEMCPY:

426 - mw_free(frag->data);

426 + freez(frag->data);

427 break;

428 case EXTERNAL_FREE_AFTER_USE:

429 frag->free_fnc(frag->data);

@@ -563,7 +563,7 @@ static int transaction_buffer_grow(struct transaction_buffer *buf, mqtt_wss_log_

563 if (buf->hdr_buffer.size > max)

564 buf->hdr_buffer.size = max;

565

566 - void *ret = mw_realloc(buf->hdr_buffer.data, buf->hdr_buffer.size);

566 + void *ret = reallocz(buf->hdr_buffer.data, buf->hdr_buffer.size);

567 if (ret == NULL) {

568 mws_warn(log_ctx, "Buffer growth failed (realloc)");

569 return 1;

@@ -581,7 +581,7 @@ inline static int transaction_buffer_init(struct transaction_buffer *to_init, si

581 pthread_mutex_init(&to_init->mutex, NULL);

582

583 to_init->hdr_buffer.size = size;

584 - to_init->hdr_buffer.data = mw_malloc(size);

584 + to_init->hdr_buffer.data = mallocz(size);

585 if (to_init->hdr_buffer.data == NULL)

586 return 1;

587

@@ -594,7 +594,7 @@ static void transaction_buffer_destroy(struct transaction_buffer *to_init)

594 {

595 buffer_purge(&to_init->hdr_buffer);

596 pthread_mutex_destroy(&to_init->mutex);

597 - mw_free(to_init->hdr_buffer.data);

597 + freez(to_init->hdr_buffer.data);

598 }

599

600 // Creates transaction

@@ -628,7 +628,7 @@ void transaction_buffer_transaction_rollback(struct transaction_buffer *buf, str

628 #define RX_ALIASES_INITIALIZE() c_rhash_new(UINT16_MAX >> 8)

629 struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings)

630 {

631 - struct mqtt_ng_client *client = mw_calloc(1, sizeof(struct mqtt_ng_client));

631 + struct mqtt_ng_client *client = callocz(1, sizeof(struct mqtt_ng_client));

632 if (client == NULL)

633 return NULL;

634

@@ -672,7 +672,7 @@ err_free_rx_alias:

672 err_free_trx_buf:

673 transaction_buffer_destroy(&client->main_buffer);

674 err_free_client:

675 - mw_free(client);

675 + freez(client);

676 return NULL;

677 }

678

@@ -688,7 +688,7 @@ static void mqtt_ng_destroy_rx_alias_hash(c_rhash hash)

688 void *to_free;

689 while(!c_rhash_iter_uint64_keys(hash, &i, &stored_key)) {

690 c_rhash_get_ptr_by_uint64(hash, stored_key, &to_free);

691 - mw_free(to_free);

691 + freez(to_free);

692 }

693 c_rhash_destroy(hash);

694 }

@@ -700,7 +700,7 @@ static void mqtt_ng_destroy_tx_alias_hash(c_rhash hash)

700 void *to_free;

701 while(!c_rhash_iter_str_keys(hash, &i, &stored_key)) {

702 c_rhash_get_ptr_by_str(hash, stored_key, &to_free);

703 - mw_free(to_free);

703 + freez(to_free);

704 }

705 c_rhash_destroy(hash);

706 }

@@ -714,7 +714,7 @@ void mqtt_ng_destroy(struct mqtt_ng_client *client)

714 pthread_rwlock_destroy(&client->tx_topic_aliases.rwlock);

715 mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);

716

717 - mw_free(client);

717 + freez(client);

718 }

719

720 int frag_set_external_data(mqtt_wss_log_ctx_t log, struct buffer_fragment *frag, void *data, size_t data_len, free_fnc_t data_free_fnc)

@@ -730,7 +730,7 @@ int frag_set_external_data(mqtt_wss_log_ctx_t log, struct buffer_fragment *frag,

730

731 switch (ptr2memory_mode(data_free_fnc)) {

732 case MEMCPY:

733 - frag->data = mw_malloc(data_len);

733 + frag->data = mallocz(data_len);

734 if (frag->data == NULL) {

735 mws_error(log, UNIT_LOG_PREFIX "OOM while malloc @_optimized_add");

736 return 1;

@@ -1408,12 +1408,12 @@ static void mqtt_properties_parser_ctx_reset(struct mqtt_properties_parser_ctx *

1408 struct mqtt_property *f = ctx->head;

1409 ctx->head = ctx->head->next;

1410 if (f->type == MQTT_TYPE_STR || f->type == MQTT_TYPE_STR_PAIR)

1411 - mw_free(f->data.strings[0]);

1411 + freez(f->data.strings[0]);

1412 if (f->type == MQTT_TYPE_STR_PAIR)

1413 - mw_free(f->data.strings[1]);

1413 + freez(f->data.strings[1]);

1414 if (f->type == MQTT_TYPE_BIN)

1415 - mw_free(f->data.bindata);

1416 - mw_free(f);

1415 + freez(f->data.bindata);

1416 + freez(f);

1417 }

1418 ctx->tail = NULL;

1419 ctx->properties_length = 0;

@@ -1498,7 +1498,7 @@ static int parse_properties_array(struct mqtt_properties_parser_ctx *ctx, rbuf_t

1498 return rc;

1499 case PROPERTY_CREATE:

1500 BUF_READ_CHECK_AT_LEAST(data, 1);

1501 - struct mqtt_property *prop = mw_calloc(1, sizeof(struct mqtt_property));

1501 + struct mqtt_property *prop = callocz(1, sizeof(struct mqtt_property));

1502 if (ctx->head == NULL) {

1503 ctx->head = prop;

1504 ctx->tail = prop;

@@ -1558,7 +1558,7 @@ static int parse_properties_array(struct mqtt_properties_parser_ctx *ctx, rbuf_t

1558 break;

1559 case PROPERTY_TYPE_STR:

1560 BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len);

1561 - ctx->tail->data.strings[ctx->str_idx] = mw_malloc(ctx->tail->bindata_len + 1);

1561 + ctx->tail->data.strings[ctx->str_idx] = mallocz(ctx->tail->bindata_len + 1);

1562 rbuf_pop(data, ctx->tail->data.strings[ctx->str_idx], ctx->tail->bindata_len);

1563 ctx->tail->data.strings[ctx->str_idx][ctx->tail->bindata_len] = 0;

1564 ctx->str_idx++;

@@ -1571,7 +1571,7 @@ static int parse_properties_array(struct mqtt_properties_parser_ctx *ctx, rbuf_t

1571 break;

1572 case PROPERTY_TYPE_BIN:

1573 BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len);

1574 - ctx->tail->data.bindata = mw_malloc(ctx->tail->bindata_len);

1574 + ctx->tail->data.bindata = mallocz(ctx->tail->bindata_len);

1575 rbuf_pop(data, ctx->tail->data.bindata, ctx->tail->bindata_len);

1576 ctx->bytes_consumed += ctx->tail->bindata_len;

1577 ctx->state = PROPERTY_NEXT;

@@ -1721,7 +1721,7 @@ static int parse_suback_varhdr(struct mqtt_ng_client *client)

1721 return rc;

1722 parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;

1723 suback->reason_code_count = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;

1724 - suback->reason_codes = mw_calloc(suback->reason_code_count, sizeof(*suback->reason_codes));

1724 + suback->reason_codes = callocz(suback->reason_code_count, sizeof(*suback->reason_codes));

1725 suback->reason_codes_pending = suback->reason_code_count;

1726 parser->varhdr_state = MQTT_PARSE_REASONCODES;

1727 /* FALLTHROUGH */

@@ -1760,7 +1760,7 @@ static int parse_publish_varhdr(struct mqtt_ng_client *client)

1760 parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;

1761 break;

1762 }

1763 - publish->topic = mw_calloc(1, publish->topic_len + 1 /* add 0x00 */);

1763 + publish->topic = callocz(1, publish->topic_len + 1 /* add 0x00 */);

1764 if (publish->topic == NULL)

1765 return MQTT_NG_CLIENT_OOM;

1766 parser->varhdr_state = MQTT_PARSE_VARHDR_TOPICNAME;

@@ -1796,7 +1796,7 @@ static int parse_publish_varhdr(struct mqtt_ng_client *client)

1796 /* FALLTHROUGH */

1797 case MQTT_PARSE_PAYLOAD:

1798 if (parser->mqtt_fixed_hdr_remaining_length < parser->mqtt_parsed_len) {

1799 - mw_free(publish->topic);

1799 + freez(publish->topic);

1800 publish->topic = NULL;

1801 ERROR("Error parsing PUBLISH message");

1802 return MQTT_NG_CLIENT_PROTOCOL_ERROR;

@@ -1808,9 +1808,9 @@ static int parse_publish_varhdr(struct mqtt_ng_client *client)

1808 }

1809 BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->data_len);

1810

1811 - publish->data = mw_malloc(publish->data_len);

1811 + publish->data = mallocz(publish->data_len);

1812 if (publish->data == NULL) {

1813 - mw_free(publish->topic);

1813 + freez(publish->topic);

1814 publish->topic = NULL;

1815 return MQTT_NG_CLIENT_OOM;

1816 }

@@ -1867,7 +1867,7 @@ static int parse_data(struct mqtt_ng_client *client)

1867 case MQTT_CPT_SUBACK:

1868 rc = parse_suback_varhdr(client);

1869 if (rc != MQTT_NG_CLIENT_NEED_MORE_BYTES && rc != MQTT_NG_CLIENT_OK_CALL_AGAIN) {

1870 - mw_free(parser->mqtt_packet.suback.reason_codes);

1870 + freez(parser->mqtt_packet.suback.reason_codes);

1871 }

1872 if (rc == MQTT_NG_CLIENT_PARSE_DONE) {

1873 parser->state = MQTT_PARSE_MQTT_PACKET_DONE;

@@ -2096,8 +2096,8 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

2096 #endif

2097 pub = &client->parser.mqtt_packet.publish;

2098 if (pub->qos > 1) {

2099 - mw_free(pub->topic);

2100 - mw_free(pub->data);

2099 + freez(pub->topic);

2100 + freez(pub->data);

2101 return MQTT_NG_CLIENT_NOT_IMPL_YET;

2102 }

2103 if ( pub->qos == 1 && (rc = mqtt_ng_puback(client, pub->packet_id, 0)) ) {

@@ -2127,8 +2127,8 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

2127 // in case we have property topic alias and we have topic we take over the string

2128 // and add pointer to it into topic alias list

2129 if (prop == NULL)

2130 - mw_free(pub->topic);

2131 - mw_free(pub->data);

2130 + freez(pub->topic);

2131 + freez(pub->data);

2132 return MQTT_NG_CLIENT_WANT_WRITE;

2133 case MQTT_CPT_DISCONNECT:

2134 INFO ("Got MQTT DISCONNECT control packet from server. Reason code: %d", (int)client->parser.mqtt_packet.disconnect.reason_code);

@@ -2225,7 +2225,7 @@ int mqtt_ng_set_topic_alias(struct mqtt_ng_client *client, const char *topic)

2225 return idx;

2226 }

2227

2228 - alias = mw_malloc(sizeof(struct topic_alias_data));

2228 + alias = mallocz(sizeof(struct topic_alias_data));

2229 idx = ++client->tx_topic_aliases.idx_assigned;

2230 alias->idx = idx;

2231 __atomic_store_n(&alias->usage_count, 0, __ATOMIC_SEQ_CST);

src/aclk/mqtt_websockets/mqtt_wss_client.c

+68 -80

@@ -1,5 +1,4 @@

1 // SPDX-License-Identifier: GPL-3.0-only

3 #ifndef _GNU_SOURCE

4 #define _GNU_SOURCE

@@ -19,9 +18,6 @@

19 #include <sys/socket.h>

20 #include <netinet/in.h>

22 -#include <arpa/inet.h>

23 -#include <netinet/tcp.h> //TCP_NODELAY

24 -#include <netdb.h>

22 #include <openssl/err.h>

23 #include <openssl/ssl.h>

@@ -107,8 +103,6 @@ struct mqtt_wss_client_struct {

103

104 int mqtt_keepalive;

105

110 - pthread_mutex_t pub_lock;

111 -

106 // signifies that we didn't write all MQTT wanted

107 // us to write during last cycle (e.g. due to buffer

108 // size) and thus we should arm POLLOUT

@@ -121,7 +115,7 @@ struct mqtt_wss_client_struct {

115 void (*msg_callback)(const char *, const void *, size_t, int);

116 void (*puback_callback)(uint16_t packet_id);

117

124 - pthread_mutex_t stat_lock;

118 + SPINLOCK stat_lock;

119 struct mqtt_wss_stats stats;

120

121 #ifdef MQTT_WSS_DEBUG

@@ -173,14 +167,13 @@ mqtt_wss_client mqtt_wss_new(const char *log_prefix,

167 SSL_library_init();

168 SSL_load_error_strings();

169

176 - mqtt_wss_client client = mw_calloc(1, sizeof(struct mqtt_wss_client_struct));

170 + mqtt_wss_client client = callocz(1, sizeof(struct mqtt_wss_client_struct));

171 if (!client) {

172 mws_error(log, "OOM alocating mqtt_wss_client");

173 goto fail;

174 }

175

182 - pthread_mutex_init(&client->pub_lock, NULL);

183 - pthread_mutex_init(&client->stat_lock, NULL);

176 + spinlock_init(&client->stat_lock);

177

178 client->msg_callback = msg_callback;

179 client->puback_callback = puback_callback;

@@ -229,7 +222,7 @@ fail_3:

222 fail_2:

223 ws_client_destroy(client->ws_client);

224 fail_1:

232 - mw_free(client);

225 + freez(client);

226 fail:

227 mqtt_wss_log_ctx_destroy(log);

228 return NULL;

@@ -253,12 +246,15 @@ void mqtt_wss_destroy(mqtt_wss_client client)

246 // as it "borrows" this pointer and might use it

247 if (client->target_host == client->host)

248 client->target_host = NULL;

249 +

250 if (client->target_host)

257 - mw_free(client->target_host);

251 + freez(client->target_host);

252 +

253 if (client->host)

259 - mw_free(client->host);

260 - mw_free(client->proxy_passwd);

261 - mw_free(client->proxy_uname);

254 + freez(client->host);

255 +

256 + freez(client->proxy_passwd);

257 + freez(client->proxy_uname);

258

259 if (client->ssl)

260 SSL_free(client->ssl);

@@ -269,11 +265,8 @@ void mqtt_wss_destroy(mqtt_wss_client client)

265 if (client->sockfd > 0)

266 close(client->sockfd);

267

272 - pthread_mutex_destroy(&client->pub_lock);

273 - pthread_mutex_destroy(&client->stat_lock);

274 -

268 mqtt_wss_log_ctx_destroy(client->log);

276 - mw_free(client);

269 + freez(client);

270 }

271

272 static int cert_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)

@@ -298,7 +291,7 @@ static int cert_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)

291 mws_error(client->log, "verify error:num=%d:%s:depth=%d:%s", err,

292 X509_verify_cert_error_string(err), depth, err_str);

293

301 - mw_free(err_str);

294 + freez(err_str);

295 }

296

297 if (!preverify_ok && err == X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT &&

@@ -362,14 +355,14 @@ static int http_parse_reply(mqtt_wss_client client, rbuf_t buf)

355 }

356

357 if (http_code != 200) {

365 - ptr = mw_malloc(idx + 1);

358 + ptr = mallocz(idx + 1);

359 if (!ptr)

360 return 6;

361 rbuf_pop(buf, ptr, idx);

362 ptr[idx] = 0;

363

364 mws_error(client->log, "http_proxy returned error code %d \"%s\"", http_code, ptr);

372 - mw_free(ptr);

365 + freez(ptr);

366 return 7;

367 }/* else

368 rbuf_bump_tail(buf, idx);*/

@@ -450,7 +443,7 @@ static int http_proxy_connect(mqtt_wss_client client)

443

444 if (client->proxy_uname) {

445 size_t creds_plain_len = strlen(client->proxy_uname) + strlen(client->proxy_passwd) + 2;

453 - char *creds_plain = mw_malloc(creds_plain_len);

446 + char *creds_plain = mallocz(creds_plain_len);

447 if (!creds_plain) {

448 mws_error(client->log, "OOM creds_plain");

449 rc = 6;

@@ -460,9 +453,9 @@ static int http_proxy_connect(mqtt_wss_client client)

453 // OpenSSL encoder puts newline every 64 output bytes

454 // we remove those but during encoding we need that space in the buffer

455 creds_base64_len += (1+(creds_base64_len/64)) * strlen("\n");

463 - char *creds_base64 = mw_malloc(creds_base64_len + 1);

456 + char *creds_base64 = mallocz(creds_base64_len + 1);

457 if (!creds_base64) {

465 - mw_free(creds_plain);

458 + freez(creds_plain);

459 mws_error(client->log, "OOM creds_base64");

460 rc = 6;

461 goto cleanup;

@@ -475,12 +468,12 @@ static int http_proxy_connect(mqtt_wss_client client)

468

469 int b64_len;

470 base64_encode_helper((unsigned char*)creds_base64, &b64_len, (unsigned char*)creds_plain, strlen(creds_plain));

478 - mw_free(creds_plain);

471 + freez(creds_plain);

472

473 r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);

474 snprintf(r_buf_ptr, r_buf_linear_insert_capacity,"Proxy-Authorization: Basic %s" HTTP_ENDLINE, creds_base64);

475 write(client->sockfd, r_buf_ptr, strlen(r_buf_ptr));

483 - mw_free(creds_base64);

476 + freez(creds_base64);

477 }

478 write(client->sockfd, HTTP_ENDLINE, strlen(HTTP_ENDLINE));

479

@@ -523,15 +516,14 @@ cleanup:

516 return rc;

517 }

518

526 -int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_connect_params *mqtt_params, int ssl_flags, struct mqtt_wss_proxy *proxy)

519 +int mqtt_wss_connect(

520 + mqtt_wss_client client,

521 + char *host,

522 + int port,

523 + struct mqtt_connect_params *mqtt_params,

524 + int ssl_flags,

525 + struct mqtt_wss_proxy *proxy)

526 {

528 - struct sockaddr_in addr;

529 - memset(&addr, 0, sizeof(addr));

530 - addr.sin_family = AF_INET;

531 -

532 - struct hostent *he;

533 - struct in_addr **addr_list;

534 -

527 if (!mqtt_params) {

528 mws_error(client->log, "mqtt_params can't be null!");

529 return -1;

@@ -545,23 +537,35 @@ int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_c

537

538 if (client->target_host == client->host)

539 client->target_host = NULL;

540 +

541 if (client->target_host)

549 - mw_free(client->target_host);

542 + freez(client->target_host);

543 +

544 if (client->host)

551 - mw_free(client->host);

545 + freez(client->host);

546 +

547 + if (client->proxy_uname) {

548 + freez(client->proxy_uname);

549 + client->proxy_uname = NULL;

550 + }

551 +

552 + if (client->proxy_passwd) {

553 + freez(client->proxy_passwd);

554 + client->proxy_passwd = NULL;

555 + }

556

557 if (proxy && proxy->type != MQTT_WSS_DIRECT) {

554 - client->host = mw_strdup(proxy->host);

558 + client->host = strdupz(proxy->host);

559 client->port = proxy->port;

556 - client->target_host = mw_strdup(host);

560 + client->target_host = strdupz(host);

561 client->target_port = port;

562 client->proxy_type = proxy->type;

563 if (proxy->username)

560 - client->proxy_uname = mw_strdup(proxy->username);

564 + client->proxy_uname = strdupz(proxy->username);

565 if (proxy->password)

562 - client->proxy_passwd = mw_strdup(proxy->password);

566 + client->proxy_passwd = strdupz(proxy->password);

567 } else {

564 - client->host = mw_strdup(host);

568 + client->host = strdupz(host);

569 client->port = port;

570 client->target_host = client->host;

571 client->target_port = port;

@@ -569,30 +573,19 @@ int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_c

573

574 client->ssl_flags = ssl_flags;

575

572 - //TODO gethostbyname -> getaddinfo

573 - // hstrerror -> gai_strerror

574 - if ((he = gethostbyname(client->host)) == NULL) {

575 - mws_error(client->log, "gethostbyname() error \"%s\"", hstrerror(h_errno));

576 - return -1;

577 - }

578 -

579 - addr_list = (struct in_addr **)he->h_addr_list;

580 - if(!addr_list[0]) {

581 - mws_error(client->log, "No IP addr resolved");

582 - return -1;

583 - }

584 - mws_debug(client->log, "Resolved IP: %s", inet_ntoa(*addr_list[0]));

585 - addr.sin_addr = *addr_list[0];

586 - addr.sin_port = htons(client->port);

587 -

576 if (client->sockfd > 0)

577 close(client->sockfd);

590 - client->sockfd = socket(AF_INET, SOCK_STREAM | DEFAULT_SOCKET_FLAGS, 0);

591 - if (client->sockfd < 0) {

592 - mws_error(client->log, "Couldn't create socket()");

593 - return -1;

578 +

579 + char port_str[16];

580 + snprintf(port_str, sizeof(port_str) -1, "%d", client->port);

581 + int fd = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, client->host, 0, port_str, NULL);

582 + if (fd < 0) {

583 + mws_error(client->log, "Could not connect to remote endpoint \"%s\", port %d.\n", client->host, port);

584 + return -3;

585 }

586

587 + client->sockfd = fd;

588 +

589 #ifndef SOCK_CLOEXEC

590 int flags = fcntl(client->sockfd, F_GETFD);

591 if (flags != -1)

@@ -600,19 +593,10 @@ int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_c

593 #endif

594

595 int flag = 1;

603 - int result = setsockopt(client->sockfd,

604 - IPPROTO_TCP,

605 - TCP_NODELAY,

606 - &flag,

607 - sizeof(int));

596 + int result = setsockopt(client->sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int));

597 if (result < 0)

598 mws_error(client->log, "Could not dissable NAGLE");

599

611 - if (connect(client->sockfd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {

612 - mws_error(client->log, "Could not connect to remote endpoint \"%s\", port %d.\n", client->host, client->port);

613 - return -3;

614 - }

615 -

600 client->poll_fds[POLLFD_SOCKET].fd = client->sockfd;

601

602 if (fcntl(client->sockfd, F_SETFL, fcntl(client->sockfd, F_GETFL, 0) | O_NONBLOCK) == -1) {

@@ -640,6 +624,7 @@ int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_c

624 // free SSL structs from possible previous connections

625 if (client->ssl)

626 SSL_free(client->ssl);

627 +

628 if (client->ssl_ctx)

629 SSL_CTX_free(client->ssl_ctx);

630

@@ -675,6 +660,7 @@ int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_c

660 mws_error(client->log, "SSL could not connect");

661 return -5;

662 }

663 +

664 if (result == -1) {

665 int ec = SSL_get_error(client->ssl, result);

666 if (ec != SSL_ERROR_WANT_READ && ec != SSL_ERROR_WANT_WRITE) {

@@ -693,14 +679,16 @@ int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_c

679 auth.username_free = NULL;

680 auth.password = (char*)mqtt_params->password;

681 auth.password_free = NULL;

682 +

683 struct mqtt_lwt_properties lwt;

684 lwt.will_topic = (char*)mqtt_params->will_topic;

685 lwt.will_topic_free = NULL;

686 lwt.will_message = (void*)mqtt_params->will_msg;

687 lwt.will_message_free = NULL; // TODO expose no copy version to API

688 lwt.will_message_size = mqtt_params->will_msg_len;

702 - lwt.will_qos = (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK);

703 - lwt.will_retain = mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN;

689 + lwt.will_qos = (int) (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK);

690 + lwt.will_retain = (int) mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN;

691 +

692 int ret = mqtt_ng_connect(client->mqtt, &auth, mqtt_params->will_msg ? &lwt : NULL, 1, client->mqtt_keepalive);

693 if (ret) {

694 mws_error(client->log, "Error generating MQTT connect");

@@ -955,9 +943,9 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)

943 #ifdef DEBUG_ULTRA_VERBOSE

944 mws_debug(client->log, "SSL_Read: Read %d.", ret);

945 #endif

958 - pthread_mutex_lock(&client->stat_lock);

946 + spinlock_lock(&client->stat_lock);

947 client->stats.bytes_rx += ret;

960 - pthread_mutex_unlock(&client->stat_lock);

948 + spinlock_unlock(&client->stat_lock);

949 rbuf_bump_head(client->ws_client->buf_read, ret);

950 } else {

951 int errnobkp = errno;

@@ -1023,9 +1011,9 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)

1011 #ifdef DEBUG_ULTRA_VERBOSE

1012 mws_debug(client->log, "SSL_Write: Written %d of avail %d.", ret, size);

1013 #endif

1026 - pthread_mutex_lock(&client->stat_lock);

1014 + spinlock_lock(&client->stat_lock);

1015 client->stats.bytes_tx += ret;

1028 - pthread_mutex_unlock(&client->stat_lock);

1016 + spinlock_unlock(&client->stat_lock);

1017 rbuf_bump_tail(client->ws_client->buf_write, ret);

1018 } else {

1019 int errnobkp = errno;

@@ -1115,10 +1103,10 @@ int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level)

1103 struct mqtt_wss_stats mqtt_wss_get_stats(mqtt_wss_client client)

1104 {

1105 struct mqtt_wss_stats current;

1118 - pthread_mutex_lock(&client->stat_lock);

1106 + spinlock_lock(&client->stat_lock);

1107 current = client->stats;

1108 memset(&client->stats, 0, sizeof(client->stats));

1121 - pthread_mutex_unlock(&client->stat_lock);

1109 + spinlock_unlock(&client->stat_lock);

1110 mqtt_ng_get_stats(client->mqtt, &current.mqtt);

1111 return current;

1112 }

src/aclk/mqtt_websockets/mqtt_wss_log.c

+6 -6

@@ -25,13 +25,13 @@ struct mqtt_wss_log_ctx {

25 #endif

26 mqtt_wss_log_ctx_t mqtt_wss_log_ctx_create(const char *ctx_prefix, mqtt_wss_log_callback_t log_callback)

27 {

28 - mqtt_wss_log_ctx_t ctx = mw_calloc(1, sizeof(struct mqtt_wss_log_ctx));

28 + mqtt_wss_log_ctx_t ctx = callocz(1, sizeof(struct mqtt_wss_log_ctx));

29 if(!ctx)

30 return NULL;

32 if(log_callback) {

33 ctx->extern_log_fnc = log_callback;

34 - ctx->buffer = mw_calloc(1, LOG_BUFFER_SIZE);

34 + ctx->buffer = callocz(1, LOG_BUFFER_SIZE);

35 if(!ctx->buffer)

36 goto cleanup;

@@ -60,15 +60,15 @@ mqtt_wss_log_ctx_t mqtt_wss_log_ctx_create(const char *ctx_prefix, mqtt_wss_log_

60 return ctx;

62 cleanup:

63 - mw_free(ctx);

63 + freez(ctx);

64 return NULL;

65 }

67 void mqtt_wss_log_ctx_destroy(mqtt_wss_log_ctx_t ctx)

68 {

69 - mw_free(ctx->ctx_prefix);

70 - mw_free(ctx->buffer);

71 - mw_free(ctx);

69 + freez(ctx->ctx_prefix);

70 + freez(ctx->buffer);

71 + freez(ctx);

72 }

74 static inline char severity_to_c(int severity)

src/aclk/mqtt_websockets/ws_client.c

+36 -48

@@ -53,7 +53,7 @@ ws_client *ws_client_new(size_t buf_size, char **host, mqtt_wss_log_ctx_t log)

53 if(!host)

54 return NULL;

56 - client = mw_calloc(1, sizeof(ws_client));

56 + client = callocz(1, sizeof(ws_client));

57 if (!client)

58 return NULL;

@@ -87,7 +87,7 @@ cleanup_2:

87 cleanup_1:

88 rbuf_free(client->buf_read);

89 cleanup:

90 - mw_free(client);

90 + freez(client);

91 return NULL;

92 }

@@ -99,7 +99,7 @@ void ws_client_free_headers(ws_client *client)

99 while (ptr) {

100 tmp = ptr;

101 ptr = ptr->next;

102 - mw_free(tmp);

102 + freez(tmp);

103 }

104

105 client->hs.headers = NULL;

@@ -110,25 +110,28 @@ void ws_client_free_headers(ws_client *client)

110 void ws_client_destroy(ws_client *client)

111 {

112 ws_client_free_headers(client);

113 - mw_free(client->hs.nonce_reply);

114 - mw_free(client->hs.http_reply_msg);

113 + freez(client->hs.nonce_reply);

114 + freez(client->hs.http_reply_msg);

115 close(client->entropy_fd);

116 rbuf_free(client->buf_read);

117 rbuf_free(client->buf_write);

118 rbuf_free(client->buf_to_mqtt);

119 - mw_free(client);

119 + freez(client);

120 }

121

122 void ws_client_reset(ws_client *client)

123 {

124 ws_client_free_headers(client);

125 - mw_free(client->hs.nonce_reply);

125 + freez(client->hs.nonce_reply);

126 client->hs.nonce_reply = NULL;

127 - mw_free(client->hs.http_reply_msg);

127 +

128 + freez(client->hs.http_reply_msg);

129 client->hs.http_reply_msg = NULL;

130 +

131 rbuf_flush(client->buf_read);

132 rbuf_flush(client->buf_write);

133 rbuf_flush(client->buf_to_mqtt);

134 +

135 client->state = WS_RAW;

136 client->hs.hdr_state = WS_HDR_HTTP;

137 client->rx.parse_state = WS_FIRST_2BYTES;

@@ -158,31 +161,11 @@ int ws_client_want_write(ws_client *client)

161 return rbuf_bytes_available(client->buf_write);

162 }

163

161 -#define RAND_SRC "/dev/urandom"

162 -static int ws_client_get_nonce(ws_client *client, char *dest, unsigned int size)

163 -{

164 - // we do not need crypto secure random here

165 - // it's just used for protocol negotiation

166 - int rd;

167 - int f = open(RAND_SRC, O_RDONLY | O_CLOEXEC);

168 - if (f < 0) {

169 - ERROR("Error opening \"%s\". Err: \"%s\"", RAND_SRC, strerror(errno));

170 - return -2;

171 - }

172 -

173 - if ((rd = read(f, dest, size)) > 0) {

174 - close(f);

175 - return rd;

176 - }

177 - close(f);

178 - return -1;

179 -}

180 -

164 #define WEBSOCKET_NONCE_SIZE 16

165 #define TEMP_BUF_SIZE 4096

166 int ws_client_start_handshake(ws_client *client)

167 {

185 - char nonce[WEBSOCKET_NONCE_SIZE];

168 + nd_uuid_t nonce;

169 char nonce_b64[256];

170 char second[TEMP_BUF_SIZE];

171 unsigned int md_len;

@@ -190,16 +173,15 @@ int ws_client_start_handshake(ws_client *client)

173 EVP_MD_CTX *md_ctx;

174 const EVP_MD *md;

175

193 - if(!*client->host) {

176 + if(!client->host || !*client->host) {

177 ERROR("Hostname has not been set. We should not be able to come here!");

178 return 1;

179 }

180

198 - ws_client_get_nonce(client, nonce, WEBSOCKET_NONCE_SIZE);

181 + uuid_generate_random(nonce);

182 EVP_EncodeBlock((unsigned char *)nonce_b64, (const unsigned char *)nonce, WEBSOCKET_NONCE_SIZE);

200 - snprintf(second, TEMP_BUF_SIZE, websocket_upgrage_hdr,

201 - *client->host,

202 - nonce_b64);

183 + snprintf(second, TEMP_BUF_SIZE, websocket_upgrage_hdr, *client->host, nonce_b64);

184 +

185 if(rbuf_bytes_free(client->buf_write) < strlen(second)) {

186 ERROR("Write buffer capacity too low.");

187 return 1;

@@ -236,10 +218,10 @@ int ws_client_start_handshake(ws_client *client)

218 EVP_DigestUpdate(md_ctx, second, strlen(second));

219 EVP_DigestFinal_ex(md_ctx, digest, &md_len);

220

239 - EVP_EncodeBlock((unsigned char *)nonce_b64, digest, md_len);

221 + EVP_EncodeBlock((unsigned char *)nonce_b64, digest, (int) md_len);

222

241 - mw_free(client->hs.nonce_reply);

242 - client->hs.nonce_reply = mw_strdup(nonce_b64);

223 + freez(client->hs.nonce_reply);

224 + client->hs.nonce_reply = strdupz(nonce_b64);

225

226 OPENSSL_free(digest);

227

@@ -263,7 +245,7 @@ int ws_client_start_handshake(ws_client *client)

245 if (rbuf_bytes_available(client->buf_read) < x) \

246 return WS_CLIENT_NEED_MORE_BYTES;

247

266 -#define MAX_HTTP_LINE_LENGTH 1024*4

248 +#define MAX_HTTP_LINE_LENGTH (1024 * 4)

249 #define HTTP_SC_LENGTH 4 // "XXX " http status code as C string

250 #define WS_CLIENT_HTTP_HDR "HTTP/1.1 "

251 #define WS_CONN_ACCEPT "sec-websocket-accept"

@@ -278,11 +260,11 @@ int ws_client_start_handshake(ws_client *client)

260 #error "Buffer too small"

261 #endif

262

281 -#define HTTP_HDR_LINE_CHECK_LIMIT(x) if ((x) >= MAX_HTTP_LINE_LENGTH) \

282 -{ \

283 - ERROR("HTTP line received is too long. Maximum is %d", MAX_HTTP_LINE_LENGTH); \

284 - return WS_CLIENT_PROTOCOL_ERROR; \

285 -}

263 +#define HTTP_HDR_LINE_CHECK_LIMIT(x) \

264 + if ((x) >= MAX_HTTP_LINE_LENGTH) { \

265 + ERROR("HTTP line received is too long. Maximum is %d", MAX_HTTP_LINE_LENGTH); \

266 + return WS_CLIENT_PROTOCOL_ERROR; \

267 + }

268

269 int ws_client_parse_handshake_resp(ws_client *client)

270 {

@@ -290,6 +272,7 @@ int ws_client_parse_handshake_resp(ws_client *client)

272 int idx_crlf, idx_sep;

273 char *ptr;

274 size_t bytes;

275 +

276 switch (client->hs.hdr_state) {

277 case WS_HDR_HTTP:

278 BUF_READ_CHECK_AT_LEAST(strlen(WS_CLIENT_HTTP_HDR))

@@ -297,6 +280,7 @@ int ws_client_parse_handshake_resp(ws_client *client)

280 rbuf_bump_tail(client->buf_read, strlen(WS_CLIENT_HTTP_HDR));

281 client->hs.hdr_state = WS_HDR_RC;

282 break;

283 +

284 case WS_HDR_RC:

285 BUF_READ_CHECK_AT_LEAST(HTTP_SC_LENGTH); // "XXX " http return code

286 rbuf_pop(client->buf_read, buf, HTTP_SC_LENGTH);

@@ -312,6 +296,7 @@ int ws_client_parse_handshake_resp(ws_client *client)

296 }

297 client->hs.hdr_state = WS_HDR_ENDLINE;

298 break;

299 +

300 case WS_HDR_ENDLINE:

301 ptr = rbuf_find_bytes(client->buf_read, WS_HTTP_NEWLINE, strlen(WS_HTTP_NEWLINE), &idx_crlf);

302 if (!ptr) {

@@ -321,12 +306,13 @@ int ws_client_parse_handshake_resp(ws_client *client)

306 }

307 HTTP_HDR_LINE_CHECK_LIMIT(idx_crlf);

308

324 - client->hs.http_reply_msg = mw_malloc(idx_crlf+1);

309 + client->hs.http_reply_msg = mallocz(idx_crlf+1);

310 rbuf_pop(client->buf_read, client->hs.http_reply_msg, idx_crlf);

311 client->hs.http_reply_msg[idx_crlf] = 0;

312 rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));

313 client->hs.hdr_state = WS_HDR_PARSE_HEADERS;

314 break;

315 +

316 case WS_HDR_PARSE_HEADERS:

317 ptr = rbuf_find_bytes(client->buf_read, WS_HTTP_NEWLINE, strlen(WS_HTTP_NEWLINE), &idx_crlf);

318 if (!ptr) {

@@ -357,7 +343,7 @@ int ws_client_parse_handshake_resp(ws_client *client)

343 return WS_CLIENT_PROTOCOL_ERROR;

344 }

345

360 - struct http_header *hdr = mw_calloc(1, sizeof(struct http_header) + idx_crlf); //idx_crlf includes ": " that will be used as 2 \0 bytes

346 + struct http_header *hdr = callocz(1, sizeof(struct http_header) + idx_crlf); //idx_crlf includes ": " that will be used as 2 \0 bytes

347 hdr->key = ((char*)hdr) + sizeof(struct http_header);

348 hdr->value = hdr->key + idx_sep + 1;

349

@@ -384,6 +370,7 @@ int ws_client_parse_handshake_resp(ws_client *client)

370 }

371

372 break;

373 +

374 case WS_HDR_PARSE_DONE:

375 if (!client->hs.nonce_matched) {

376 ERROR("Missing " WS_CONN_ACCEPT " header");

@@ -398,6 +385,7 @@ int ws_client_parse_handshake_resp(ws_client *client)

385 client->hs.hdr_state = WS_HDR_ALL_DONE;

386 INFO("Websocket Connection Accepted By Server");

387 return WS_CLIENT_PARSING_DONE;

388 +

389 case WS_HDR_ALL_DONE:

390 FATAL("This is error we should never come here!");

391 return WS_CLIENT_PROTOCOL_ERROR;

@@ -642,7 +630,7 @@ int ws_client_process_rx_ws(ws_client *client)

630 break;

631 case WS_PAYLOAD_CONNECTION_CLOSE_MSG:

632 if (!client->rx.specific_data.op_close.reason)

645 - client->rx.specific_data.op_close.reason = mw_malloc(client->rx.payload_length + 1);

633 + client->rx.specific_data.op_close.reason = mallocz(client->rx.payload_length + 1);

634

635 while (client->rx.payload_processed < client->rx.payload_length) {

636 if (!rbuf_bytes_available(client->buf_read))

@@ -655,7 +643,7 @@ int ws_client_process_rx_ws(ws_client *client)

643 INFO("WebSocket server closed the connection with EC=%d and reason \"%s\"",

644 client->rx.specific_data.op_close.ec,

645 client->rx.specific_data.op_close.reason);

658 - mw_free(client->rx.specific_data.op_close.reason);

646 + freez(client->rx.specific_data.op_close.reason);

647 client->rx.specific_data.op_close.reason = NULL;

648 client->rx.parse_state = WS_PACKET_DONE;

649 break;

@@ -672,7 +660,7 @@ int ws_client_process_rx_ws(ws_client *client)

660 return WS_CLIENT_INTERNAL_ERROR;

661 }

662 BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);

675 - client->rx.specific_data.ping_msg = mw_malloc(client->rx.payload_length);

663 + client->rx.specific_data.ping_msg = mallocz(client->rx.payload_length);

664 rbuf_pop(client->buf_read, client->rx.specific_data.ping_msg, client->rx.payload_length);

665 // TODO schedule this instead of sending right away

666 // then attempt to send as soon as buffer space clears up