@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 946 lines 30.6 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #ifndef _GNU_SOURCE
4 #define _GNU_SOURCE
5 #endif
6
7 #include "libnetdata/libnetdata.h"
8 #include "aclk_mqtt_workers.h"
9 #include "mqtt_wss_client.h"
10 #include "mqtt_ng.h"
11 #include "ws_client.h"
12 #include "common_internal.h"
13 #include "../aclk.h"
14 #include "../aclk_util.h"
15
16 #define PIPE_READ_END 0
17 #define PIPE_WRITE_END 1
18 #define POLLFD_SOCKET 0
19 #define POLLFD_PIPE 1
20
21 #define PING_TIMEOUT (60) //Expect a ping response within this time (seconds)
22 time_t ping_timeout = 0;
23
24 #if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110) && (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
25 #include <openssl/conf.h>
26 #endif
27
28 //TODO MQTT_PUBLISH_RETAIN should not be needed anymore
29 #define MQTT_PUBLISH_RETAIN 0x01
30 #define MQTT_CONNECT_CLEAN_SESSION 0x02
31 #define MQTT_CONNECT_WILL_RETAIN 0x20
32
33 char *util_openssl_ret_err(int err)
34 {
35 switch(err){
36 case SSL_ERROR_WANT_READ:
37 return "SSL_ERROR_WANT_READ";
38 case SSL_ERROR_WANT_WRITE:
39 return "SSL_ERROR_WANT_WRITE";
40 case SSL_ERROR_NONE:
41 return "SSL_ERROR_NONE";
42 case SSL_ERROR_ZERO_RETURN:
43 return "SSL_ERROR_ZERO_RETURN";
44 case SSL_ERROR_WANT_CONNECT:
45 return "SSL_ERROR_WANT_CONNECT";
46 case SSL_ERROR_WANT_ACCEPT:
47 return "SSL_ERROR_WANT_ACCEPT";
48 case SSL_ERROR_WANT_X509_LOOKUP:
49 return "SSL_ERROR_WANT_X509_LOOKUP";
50 #ifdef SSL_ERROR_WANT_ASYNC
51 case SSL_ERROR_WANT_ASYNC:
52 return "SSL_ERROR_WANT_ASYNC";
53 #endif
54 #ifdef SSL_ERROR_WANT_ASYNC_JOB
55 case SSL_ERROR_WANT_ASYNC_JOB:
56 return "SSL_ERROR_WANT_ASYNC_JOB";
57 #endif
58 #ifdef SSL_ERROR_WANT_CLIENT_HELLO_CB
59 case SSL_ERROR_WANT_CLIENT_HELLO_CB:
60 return "SSL_ERROR_WANT_CLIENT_HELLO_CB";
61 #endif
62 case SSL_ERROR_SYSCALL:
63 return "SSL_ERROR_SYSCALL";
64 case SSL_ERROR_SSL:
65 return "SSL_ERROR_SSL";
66 default:
67 break;
68 }
69 return "UNKNOWN";
70 }
71
72 struct mqtt_wss_client_struct {
73 ws_client *ws_client;
74
75 // immediate connection (e.g. proxy server)
76 char *host;
77 int port;
78
79 // target of connection (e.g. where we want to connect to)
80 char *target_host;
81 int target_port;
82
83 enum mqtt_wss_proxy_type proxy_type;
84 char *proxy_uname;
85 char *proxy_passwd;
86
87 // nonblock IO related
88 int sockfd;
89 int write_notif_pipe[2];
90 struct pollfd poll_fds[2];
91
92 SSL_CTX *ssl_ctx;
93 SSL *ssl;
94 int ssl_flags;
95
96 struct mqtt_ng_client *mqtt;
97
98 int mqtt_keepalive;
99
100 // signifies that we didn't write all MQTT wanted
101 // us to write during last cycle (e.g. due to buffer
102 // size) and thus we should arm POLLOUT
103 unsigned int mqtt_didnt_finish_write:1;
104
105 unsigned int mqtt_connected:1;
106 unsigned int mqtt_disconnecting:1;
107
108 // Application layer callback pointers
109 void (*msg_callback)(const char *, const void *, size_t, int);
110 void (*puback_callback)(uint16_t packet_id);
111
112 SPINLOCK stat_lock;
113 struct mqtt_wss_stats stats;
114
115 #ifdef MQTT_WSS_DEBUG
116 void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line);
117 #endif
118 };
119
120 static void mws_connack_callback_ng(void *user_ctx, int code)
121 {
122 mqtt_wss_client client = user_ctx;
123 switch(code) {
124 case 0:
125 client->mqtt_connected = 1;
126 break;
127 //TODO manual labor: all the CONNACK error codes with some nice error message
128 default:
129 nd_log(NDLS_DAEMON, NDLP_ERR, "MQTT CONNACK returned error %d", code);
130 break;
131 }
132 }
133
134 static ssize_t mqtt_send_cb(void *user_ctx, const void* buf, size_t len)
135 {
136 mqtt_wss_client client = user_ctx;
137 int ret = ws_client_send(client->ws_client, WS_OP_BINARY_FRAME, buf, len);
138 if (ret >= 0 && (size_t)ret != len)
139 client->mqtt_didnt_finish_write = 1;
140 return ret;
141 }
142
143 mqtt_wss_client mqtt_wss_new(
144 msg_callback_fnc_t msg_callback,
145 void (*puback_callback)(uint16_t packet_id))
146 {
147 SSL_library_init();
148 SSL_load_error_strings();
149
150 mqtt_wss_client client = callocz(1, sizeof(struct mqtt_wss_client_struct));
151
152 spinlock_init(&client->stat_lock);
153
154 client->msg_callback = msg_callback;
155 client->puback_callback = puback_callback;
156
157 client->ws_client = ws_client_new(0, &client->target_host);
158 if (!client->ws_client) {
159 nd_log(NDLS_DAEMON, NDLP_ERR, "Error creating ws_client");
160 goto fail_1;
161 }
162
163 #ifdef __APPLE__
164 if (pipe(client->write_notif_pipe)) {
165 #else
166 if (pipe2(client->write_notif_pipe, O_CLOEXEC /*| O_DIRECT*/)) {
167 #endif
168 nd_log(NDLS_DAEMON, NDLP_ERR, "Couldn't create pipe");
169 goto fail_2;
170 }
171
172 client->poll_fds[POLLFD_PIPE].fd = client->write_notif_pipe[PIPE_READ_END];
173 client->poll_fds[POLLFD_PIPE].events = POLLIN;
174
175 client->poll_fds[POLLFD_SOCKET].events = POLLIN;
176
177 struct mqtt_ng_init settings = {
178 .data_in = client->ws_client->buf_to_mqtt,
179 .data_out_fnc = &mqtt_send_cb,
180 .user_ctx = client,
181 .connack_callback = &mws_connack_callback_ng,
182 .puback_callback = puback_callback,
183 .msg_callback = msg_callback
184 };
185 client->mqtt = mqtt_ng_init(&settings);
186
187 return client;
188
189 fail_2:
190 ws_client_destroy(client->ws_client);
191 fail_1:
192 freez(client);
193 return NULL;
194 }
195
196 void mqtt_wss_set_max_buf_size(mqtt_wss_client client, size_t size)
197 {
198 mqtt_ng_set_max_mem(client->mqtt, size);
199 }
200
201 void mqtt_wss_destroy(mqtt_wss_client client)
202 {
203 mqtt_ng_destroy(client->mqtt);
204
205 close(client->write_notif_pipe[PIPE_WRITE_END]);
206 close(client->write_notif_pipe[PIPE_READ_END]);
207
208 ws_client_destroy(client->ws_client);
209
210 // deleted after client->ws_client
211 // as it "borrows" this pointer and might use it
212 if (client->target_host == client->host)
213 client->target_host = NULL;
214
215 if (client->target_host)
216 freez(client->target_host);
217
218 if (client->host)
219 freez(client->host);
220
221 aclk_sensitive_free(&client->proxy_passwd);
222 freez(client->proxy_uname);
223
224 if (client->ssl)
225 SSL_free(client->ssl);
226
227 if (client->ssl_ctx)
228 SSL_CTX_free(client->ssl_ctx);
229
230 if (client->sockfd > 0)
231 close(client->sockfd);
232
233 freez(client);
234 }
235
236 static int cert_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)
237 {
238 int err = 0;
239
240 SSL* ssl = X509_STORE_CTX_get_ex_data(ctx, SSL_get_ex_data_X509_STORE_CTX_idx());
241 mqtt_wss_client client = SSL_get_ex_data(ssl, 0);
242
243 if (!preverify_ok) {
244 err = X509_STORE_CTX_get_error(ctx);
245 netdata_ssl_log_verify_error(ctx);
246 }
247
248 if (!preverify_ok && (client->ssl_flags & MQTT_WSS_SSL_ALLOW_SELF_SIGNED)) {
249 // MQTT_WSS_SSL_ALLOW_SELF_SIGNED means "this connection accepts a
250 // certificate that wouldn't pass full validation". Cover the errors
251 // that on-prem deployments routinely hit:
252 // - leaf is self-signed (no CA at all)
253 // - cert subject does not match the configured hostname/IP
254 // (DNS aliases, IP-only access, certs without proper SAN)
255 switch (err) {
256 case X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT:
257 preverify_ok = 1;
258 nd_log(NDLS_DAEMON, NDLP_ERR,
259 "Self Signed Certificate Accepted as the connection was "
260 "requested with MQTT_WSS_SSL_ALLOW_SELF_SIGNED");
261 break;
262 case X509_V_ERR_HOSTNAME_MISMATCH:
263 preverify_ok = 1;
264 nd_log(NDLS_DAEMON, NDLP_ERR,
265 "Certificate hostname mismatch accepted as the connection "
266 "was requested with MQTT_WSS_SSL_ALLOW_SELF_SIGNED");
267 break;
268 case X509_V_ERR_IP_ADDRESS_MISMATCH:
269 preverify_ok = 1;
270 nd_log(NDLS_DAEMON, NDLP_ERR,
271 "Certificate IP address mismatch accepted as the connection "
272 "was requested with MQTT_WSS_SSL_ALLOW_SELF_SIGNED");
273 break;
274 default:
275 break;
276 }
277 }
278
279 return preverify_ok;
280 }
281
282 int mqtt_wss_connect(
283 mqtt_wss_client client,
284 char *host,
285 int port,
286 struct mqtt_connect_params *mqtt_params,
287 int ssl_flags,
288 const struct mqtt_wss_proxy *proxy,
289 bool *fallback_ipv4)
290 {
291 if (!mqtt_params) {
292 nd_log(NDLS_DAEMON, NDLP_ERR, "mqtt_params can't be null!");
293 return -1;
294 }
295
296 // reset state in case this is reconnect
297 client->mqtt_didnt_finish_write = 0;
298 client->mqtt_connected = 0;
299 client->mqtt_disconnecting = 0;
300 ws_client_reset(client->ws_client);
301
302 if (client->target_host == client->host)
303 client->target_host = NULL;
304
305 if (client->target_host)
306 freez(client->target_host);
307
308 if (client->host)
309 freez(client->host);
310
311 if (client->proxy_uname) {
312 freez(client->proxy_uname);
313 client->proxy_uname = NULL;
314 }
315
316 if (client->proxy_passwd) {
317 aclk_sensitive_free(&client->proxy_passwd);
318 }
319
320 if (proxy && proxy->type != MQTT_WSS_DIRECT) {
321 client->host = strdupz(proxy->host);
322 client->port = proxy->port;
323 client->target_host = strdupz(host);
324 client->target_port = port;
325 client->proxy_type = proxy->type;
326 if (proxy->username)
327 client->proxy_uname = strdupz(proxy->username);
328 if (proxy->password)
329 client->proxy_passwd = strdupz(proxy->password);
330 } else {
331 client->host = strdupz(host);
332 client->port = port;
333 client->target_host = client->host;
334 client->target_port = port;
335 client->proxy_type = MQTT_WSS_DIRECT;
336 }
337
338 client->ssl_flags = ssl_flags;
339
340 if (client->sockfd > 0)
341 close(client->sockfd);
342
343 char port_str[16];
344 snprintf(port_str, sizeof(port_str) -1, "%d", client->port);
345
346 if (proxy && proxy->type != MQTT_WSS_DIRECT) {
347 const char *proxy_proto = aclk_mqtt_proxy_type_to_scheme(proxy->type);
348 nd_log_daemon(NDLP_INFO, "ACLK: connecting to %s:%d via proxy %s%s:%d%s",
349 client->target_host, client->target_port,
350 proxy_proto, client->host, client->port,
351 client->proxy_uname ? " (with credentials)" : " (without credentials)");
352 }
353 else
354 nd_log_daemon(NDLP_INFO, "ACLK: connecting to %s:%d (no proxy)",
355 client->target_host, client->target_port);
356
357 struct timeval timeout = { .tv_sec = 10, .tv_usec = 0 };
358 int fd = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, client->host, 0, port_str, &timeout, fallback_ipv4);
359 if (fd < 0) {
360 nd_log(NDLS_DAEMON, NDLP_ERR, "Could not connect to remote endpoint \"%s\", port %d.\n", client->host, port);
361 return -3;
362 }
363
364 client->sockfd = fd;
365
366 #ifndef SOCK_CLOEXEC
367 int flags = fcntl(client->sockfd, F_GETFD);
368 if (flags != -1)
369 (void) fcntl(client->sockfd, F_SETFD, flags| FD_CLOEXEC);
370 #endif
371
372 int flag = 1;
373 int result = setsockopt(client->sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int));
374 if (result < 0)
375 nd_log(NDLS_DAEMON, NDLP_ERR, "Could not dissable NAGLE");
376
377 client->poll_fds[POLLFD_SOCKET].fd = client->sockfd;
378
379 if (fcntl(client->sockfd, F_SETFL, fcntl(client->sockfd, F_GETFL, 0) | O_NONBLOCK) == -1) {
380 nd_log(NDLS_DAEMON, NDLP_ERR, "Error setting O_NONBLOCK to TCP socket. \"%s\"", strerror(errno));
381 return -8;
382 }
383
384 if (client->proxy_type != MQTT_WSS_DIRECT) {
385 if (aclk_proxy_negotiation_connect(client->sockfd, client->proxy_type, client->proxy_uname, client->proxy_passwd,
386 client->target_host, client->target_port, 10000))
387 return -4;
388
389 // Credentials are only needed for proxy negotiation; wipe them now.
390 aclk_sensitive_free(&client->proxy_passwd);
391 }
392
393 #if OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
394 #if (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
395 OPENSSL_config(NULL);
396 #endif
397 SSL_load_error_strings();
398 SSL_library_init();
399 #else
400 if (OPENSSL_init_ssl(OPENSSL_INIT_LOAD_CONFIG, NULL) != 1) {
401 nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to initialize SSL");
402 return -1;
403 };
404 #endif
405
406 // free SSL structs from possible previous connections
407 if (client->ssl)
408 SSL_free(client->ssl);
409
410 if (client->ssl_ctx)
411 SSL_CTX_free(client->ssl_ctx);
412
413 client->ssl_ctx = SSL_CTX_new(SSLv23_client_method());
414 if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
415 SSL_CTX_set_default_verify_paths(client->ssl_ctx);
416 SSL_CTX_set_verify(client->ssl_ctx, SSL_VERIFY_PEER | SSL_VERIFY_CLIENT_ONCE, cert_verify_callback);
417 } else
418 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL Certificate checking completely disabled!!!");
419
420 #ifdef MQTT_WSS_DEBUG
421 if(client->ssl_ctx_keylog_cb)
422 SSL_CTX_set_keylog_callback(client->ssl_ctx, client->ssl_ctx_keylog_cb);
423 #endif
424
425 client->ssl = SSL_new(client->ssl_ctx);
426 if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
427 if (!SSL_set_ex_data(client->ssl, 0, client)) {
428 nd_log(NDLS_DAEMON, NDLP_ERR, "Could not SSL_set_ex_data");
429 return -4;
430 }
431 }
432 SSL_set_fd(client->ssl, client->sockfd);
433 SSL_set_connect_state(client->ssl);
434
435 if (!SSL_set_tlsext_host_name(client->ssl, client->target_host)) {
436 nd_log(NDLS_DAEMON, NDLP_ERR, "Error setting TLS SNI host");
437 return -7;
438 }
439
440 if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
441 // target_host may be either a DNS hostname or an IP literal.
442 // X509_VERIFY_PARAM_set1_ip_asc() parses the string as an IP and
443 // matches against the cert's iPAddress SAN; it returns 0 if the
444 // string is not a valid IP. X509_VERIFY_PARAM_set1_host() matches
445 // against the dNSName SAN. Try the IP path first; if the input is
446 // not an IP literal, fall back to hostname matching.
447 X509_VERIFY_PARAM *param = SSL_get0_param(client->ssl);
448 if (!X509_VERIFY_PARAM_set1_ip_asc(param, client->target_host) &&
449 !X509_VERIFY_PARAM_set1_host(param, client->target_host, 0)) {
450 nd_log(NDLS_DAEMON, NDLP_ERR, "Error setting TLS hostname verification host");
451 return -7;
452 }
453 }
454
455 result = SSL_connect(client->ssl);
456 if (result != -1 && result != 1) {
457 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL could not connect");
458 return -5;
459 }
460
461 if (result == -1) {
462 int ec = SSL_get_error(client->ssl, result);
463 if (ec != SSL_ERROR_WANT_READ && ec != SSL_ERROR_WANT_WRITE) {
464 nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to start SSL connection");
465 return -6;
466 }
467 }
468
469 client->mqtt_keepalive = (mqtt_params->keep_alive ? mqtt_params->keep_alive : 400);
470
471 struct mqtt_auth_properties auth;
472 auth.client_id = (char*)mqtt_params->clientid;
473 auth.client_id_free = NULL;
474 auth.username = (char*)mqtt_params->username;
475 auth.username_free = NULL;
476 auth.password = (char*)mqtt_params->password;
477 auth.password_free = NULL;
478
479 struct mqtt_lwt_properties lwt;
480 lwt.will_topic = (char*)mqtt_params->will_topic;
481 lwt.will_topic_free = NULL;
482 lwt.will_message = (void*)mqtt_params->will_msg;
483 lwt.will_message_free = NULL; // TODO expose no copy version to API
484 lwt.will_message_size = mqtt_params->will_msg_len;
485 lwt.will_qos = (int) (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK);
486 lwt.will_retain = (int) mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN;
487
488 int ret = mqtt_ng_connect(client->mqtt, &auth, mqtt_params->will_msg ? &lwt : NULL, client->mqtt_keepalive);
489 if (ret) {
490 nd_log(NDLS_DAEMON, NDLP_ERR, "Error generating MQTT connect");
491 return 1;
492 }
493
494 client->poll_fds[POLLFD_PIPE].events = POLLIN;
495 client->poll_fds[POLLFD_SOCKET].events = POLLIN;
496 // wait till MQTT connection is established
497 while (!client->mqtt_connected) {
498 int rc = mqtt_wss_service(client, 60 * MSEC_PER_SEC);
499 if(rc) {
500 nd_log(NDLS_DAEMON, NDLP_ERR, "Error connecting to MQTT WSS server \"%s\", port %d. Code: %d", host, port, rc);
501 return 2;
502 }
503 }
504
505 return 0;
506 }
507
508 #define MWS_TIMED_OUT 1
509 #define MWS_ERROR 2
510 #define MWS_OK 0
511 static const char *mqtt_wss_error_tos(int ec)
512 {
513 switch(ec) {
514 case MWS_TIMED_OUT:
515 return "Error: Operation was not able to finish in time";
516 case MWS_ERROR:
517 return "Unspecified Error";
518 default:
519 return "Unknown Error Code!";
520 }
521
522 }
523
524 static int mqtt_wss_service_all(mqtt_wss_client client, int timeout_ms)
525 {
526 uint64_t exit_by_us = now_boottime_usec() + (timeout_ms * NSEC_PER_MSEC);
527 client->poll_fds[POLLFD_SOCKET].events |= POLLOUT; // TODO when entering mwtt_wss_service use out buffer size to arm POLLOUT
528 while (rbuf_bytes_available(client->ws_client->buf_write)) {
529 const uint64_t now_us = now_boottime_usec();
530 if (now_us >= exit_by_us)
531 return MWS_TIMED_OUT;
532 if (mqtt_wss_service(client, (exit_by_us - now_us) / USEC_PER_SEC))
533 return MWS_ERROR;
534 }
535 return MWS_OK;
536 }
537
538 void mqtt_wss_disconnect(mqtt_wss_client client, int timeout_ms)
539 {
540 // block application from sending more MQTT messages
541 client->mqtt_disconnecting = 1;
542
543 // send whatever was left at the time of calling this function
544 int ret = mqtt_wss_service_all(client, timeout_ms / 4);
545 if(ret)
546 nd_log(NDLS_DAEMON, NDLP_ERR,
547 "Error while trying to send all remaining data in an attempt "
548 "to gracefully disconnect! EC=%d Desc:\"%s\"",
549 ret,
550 mqtt_wss_error_tos(ret));
551
552 // schedule and send MQTT disconnect
553 mqtt_ng_disconnect(client->mqtt, 0);
554 mqtt_ng_sync(client->mqtt);
555
556 ret = mqtt_wss_service_all(client, timeout_ms / 4);
557 if(ret)
558 nd_log(NDLS_DAEMON, NDLP_ERR,
559 "Error while trying to send MQTT disconnect message in an attempt "
560 "to gracefully disconnect! EC=%d Desc:\"%s\"",
561 ret,
562 mqtt_wss_error_tos(ret));
563
564 // send WebSockets close message
565 uint16_t ws_rc = htobe16(1000);
566 ws_client_send(client->ws_client, WS_OP_CONNECTION_CLOSE, (const char*)&ws_rc, sizeof(ws_rc));
567 ret = mqtt_wss_service_all(client, timeout_ms / 4);
568 if(ret) {
569 // Some MQTT/WSS servers will close socket on receipt of MQTT disconnect and
570 // do not wait for WebSocket to be closed properly
571 nd_log(NDLS_DAEMON, NDLP_WARNING,
572 "Error while trying to send WebSocket disconnect message in an attempt "
573 "to gracefully disconnect! EC=%d Desc:\"%s\".",
574 ret,
575 mqtt_wss_error_tos(ret));
576 }
577
578 // Service WSS connection until remote closes connection (usual)
579 // or timeout happens (unusual) in which case we close
580 mqtt_wss_service_all(client, timeout_ms / 4);
581
582 close(client->sockfd);
583 client->sockfd = -1;
584 }
585
586 static void mqtt_wss_wakeup(mqtt_wss_client client)
587 {
588 if(write(client->write_notif_pipe[PIPE_WRITE_END], " ", 1) <= 0) { ; }
589 }
590
591 #define THROWAWAY_BUF_SIZE 32
592 char throwaway[THROWAWAY_BUF_SIZE];
593 static void util_clear_pipe(int fd)
594 {
595 if(read(fd, throwaway, THROWAWAY_BUF_SIZE) <= 0) { ; }
596 }
597
598 static void set_socket_pollfds(mqtt_wss_client client, int ssl_ret) {
599 if (ssl_ret == SSL_ERROR_WANT_WRITE)
600 client->poll_fds[POLLFD_SOCKET].events |= POLLOUT;
601 if (ssl_ret == SSL_ERROR_WANT_READ)
602 client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
603 }
604
605 static int handle_mqtt_internal(mqtt_wss_client client)
606 {
607 int rc = mqtt_ng_sync(client->mqtt);
608 if (rc) {
609 nd_log(NDLS_DAEMON, NDLP_ERR, "mqtt_ng_sync returned %d != 0", rc);
610 client->mqtt_connected = 0;
611 return 1;
612 }
613 return 0;
614 }
615
616 static int t_till_next_keepalive_ms(mqtt_wss_client client)
617 {
618 time_t last_send_ts = mqtt_ng_last_send_time(client->mqtt);
619 time_t next_mqtt_keep_alive_ts = last_send_ts + client->mqtt_keepalive * 0.75;
620
621 time_t now_ts = now_realtime_sec();
622
623 if(now_ts >= next_mqtt_keep_alive_ts)
624 return 0;
625
626 int timeout_ms = (int)((next_mqtt_keep_alive_ts - now_ts) * MSEC_PER_SEC);
627
628 if(timeout_ms < 1)
629 timeout_ms = 1;
630
631 if(timeout_ms > (int)(45 * MSEC_PER_SEC))
632 timeout_ms = (int)(45 * MSEC_PER_SEC);
633
634 return timeout_ms;
635 }
636
637 int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
638 {
639 char *ptr;
640 size_t size;
641 int ret;
642 int send_keepalive = 0;
643
644 #ifdef MQTT_WSS_CPUSTATS
645 uint64_t t2;
646 uint64_t t1 = now_monotonic_usec();
647 #endif
648
649 // Check user requested TO doesn't interfere with MQTT keep alives
650 if (!ping_timeout) {
651 int till_next_keep_alive = t_till_next_keepalive_ms(client);
652 if (client->mqtt_connected && (timeout_ms < 0 || timeout_ms >= till_next_keep_alive)) {
653 timeout_ms = till_next_keep_alive;
654 send_keepalive = 1;
655 }
656 }
657
658 #ifdef MQTT_WSS_CPUSTATS
659 t2 = now_monotonic_usec();
660 client->stats.time_keepalive += t2 - t1;
661 #endif
662
663 worker_is_idle();
664 if ((ret = poll(client->poll_fds, 2, timeout_ms >= 0 ? timeout_ms : -1)) < 0) {
665 worker_is_busy(WORKER_ACLK_POLL_ERROR);
666
667 if (errno == EINTR) {
668 nd_log(NDLS_DAEMON, NDLP_WARNING, "poll interrupted by EINTR");
669 return MQTT_WSS_OK;
670 }
671 nd_log(NDLS_DAEMON, NDLP_ERR, "poll error \"%s\"", strerror(errno));
672 return MQTT_WSS_ERR_POLL_FAILED;
673 }
674 worker_is_busy(WORKER_ACLK_POLL_OK);
675
676 #ifdef MQTT_WSS_CPUSTATS
677 t1 = now_monotonic_usec();
678 #endif
679
680 if (ret == 0) {
681 time_t now = now_realtime_sec();
682 if (send_keepalive) {
683 // otherwise we shortened the timeout ourselves to take care of
684 // MQTT keep alives
685 mqtt_ng_ping(client->mqtt);
686 ping_timeout = now + PING_TIMEOUT;
687 worker_is_busy(WORKER_ACLK_SENT_PING);
688 } else {
689 if (ping_timeout && ping_timeout < now) {
690 disconnect_req = ACLK_PING_TIMEOUT;
691 ping_timeout = 0;
692 }
693 // if poll timed out and user requested timeout was being used
694 // return here let user do his work and he will call us back soon
695 return MQTT_WSS_OK;
696 }
697 }
698
699 #ifdef MQTT_WSS_CPUSTATS
700 t2 = now_monotonic_usec();
701 client->stats.time_keepalive += t2 - t1;
702 #endif
703
704 client->poll_fds[POLLFD_SOCKET].events = 0;
705
706 if ((ptr = rbuf_get_linear_insert_range(client->ws_client->buf_read, &size))) {
707 worker_is_busy(WORKER_ACLK_RX);
708
709 if((ret = SSL_read(client->ssl, ptr, size)) > 0) {
710 spinlock_lock(&client->stat_lock);
711 client->stats.bytes_rx += ret;
712 spinlock_unlock(&client->stat_lock);
713 rbuf_bump_head(client->ws_client->buf_read, ret);
714 } else {
715 int errnobkp = errno;
716 ret = SSL_get_error(client->ssl, ret);
717 set_socket_pollfds(client, ret);
718
719 if (ret != SSL_ERROR_WANT_READ &&
720 ret != SSL_ERROR_WANT_WRITE) {
721 worker_is_busy(WORKER_ACLK_RX_ERROR);
722 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL_read error: %d %s", ret, util_openssl_ret_err(ret));
723
724 if (ret == SSL_ERROR_ZERO_RETURN) {
725 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL_read connection closed by remote end");
726 return MQTT_WSS_ERR_REMOTE_CLOSED;
727 }
728
729 if (ret == SSL_ERROR_SYSCALL)
730 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL_read SYSCALL errno: %d %s", errnobkp, strerror(errnobkp));
731
732 return MQTT_WSS_ERR_CONN_DROP;
733 }
734 }
735 }
736
737 #ifdef MQTT_WSS_CPUSTATS
738 t1 = now_monotonic_usec();
739 client->stats.time_read_socket += t1 - t2;
740 #endif
741
742 ret = ws_client_process(client->ws_client);
743 switch(ret) {
744 case WS_CLIENT_PROTOCOL_ERROR:
745 return MQTT_WSS_ERR_PROTO_WS;
746
747 case WS_CLIENT_NEED_MORE_BYTES:
748 client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
749 break;
750
751 case WS_CLIENT_CONNECTION_REMOTE_CLOSED:
752 return MQTT_WSS_ERR_REMOTE_CLOSED;
753
754 case WS_CLIENT_CONNECTION_CLOSED:
755 return MQTT_WSS_ERR_CONN_DROP;
756
757 default:
758 return MQTT_WSS_ERR_PROTO_WS;
759 }
760
761 #ifdef MQTT_WSS_CPUSTATS
762 t2 = now_monotonic_usec();
763 client->stats.time_process_websocket += t2 - t1;
764 #endif
765
766 // process MQTT stuff
767 if(client->ws_client->state == WS_ESTABLISHED) {
768 worker_is_busy(WORKER_ACLK_HANDLE_MQTT_INTERNAL);
769 if (handle_mqtt_internal(client))
770 return MQTT_WSS_ERR_PROTO_MQTT;
771 }
772
773 if (client->mqtt_didnt_finish_write) {
774 client->mqtt_didnt_finish_write = 0;
775 client->poll_fds[POLLFD_SOCKET].events |= POLLOUT;
776 }
777
778 #ifdef MQTT_WSS_CPUSTATS
779 t1 = now_monotonic_usec();
780 client->stats.time_process_mqtt += t1 - t2;
781 #endif
782
783 if ((ptr = rbuf_get_linear_read_range(client->ws_client->buf_write, &size))) {
784 worker_is_busy(WORKER_ACLK_TX);
785
786 if ((ret = SSL_write(client->ssl, ptr, size)) > 0) {
787 spinlock_lock(&client->stat_lock);
788 client->stats.bytes_tx += ret;
789 spinlock_unlock(&client->stat_lock);
790 rbuf_bump_tail(client->ws_client->buf_write, ret);
791 } else {
792 int errnobkp = errno;
793 ret = SSL_get_error(client->ssl, ret);
794 set_socket_pollfds(client, ret);
795 if (ret != SSL_ERROR_WANT_READ &&
796 ret != SSL_ERROR_WANT_WRITE) {
797 worker_is_busy(WORKER_ACLK_TX_ERROR);
798 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL_write error: %d %s", ret, util_openssl_ret_err(ret));
799
800 if (ret == SSL_ERROR_ZERO_RETURN) {
801 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL_write connection closed by remote end");
802 return MQTT_WSS_ERR_REMOTE_CLOSED;
803 }
804
805 if (ret == SSL_ERROR_SYSCALL)
806 nd_log(NDLS_DAEMON, NDLP_ERR, "SSL_write SYSCALL errno: %d %s", errnobkp, strerror(errnobkp));
807
808 return MQTT_WSS_ERR_CONN_DROP;
809 }
810 }
811 }
812
813 if(client->poll_fds[POLLFD_PIPE].revents & POLLIN)
814 util_clear_pipe(client->write_notif_pipe[PIPE_READ_END]);
815
816 #ifdef MQTT_WSS_CPUSTATS
817 t2 = now_monotonic_usec();
818 client->stats.time_write_socket += t2 - t1;
819 #endif
820
821 return MQTT_WSS_OK;
822 }
823
824 int mqtt_wss_publish5(mqtt_wss_client client,
825 char *topic,
826 free_fnc_t topic_free,
827 void *msg,
828 free_fnc_t msg_free,
829 size_t msg_len,
830 uint8_t publish_flags,
831 uint16_t *packet_id)
832 {
833 // topic_free is not yet supported: the rollback path inside mqtt_ng_publish
834 // can free topic asymmetrically across failure modes (see contract notes in
835 // mqtt_ng.h and the long comment below). Enforce NULL until that is fixed
836 // so callers don't silently leak a borrowed/allocated topic on failure.
837 internal_fatal(topic_free != NULL, "mqtt_wss_publish5: topic_free must be NULL until rollback ownership is made symmetric");
838
839 const char *fail_reason = NULL;
840 if (client->mqtt_disconnecting)
841 fail_reason = "mqtt_wss is disconnecting can't publish";
842 else if (!client->mqtt_connected)
843 fail_reason = "MQTT is offline. Can't send message.";
844
845 if (fail_reason) {
846 nd_log(NDLS_DAEMON, NDLP_ERR, "%s", fail_reason);
847 if (packet_id)
848 *packet_id = 0;
849 if (msg_free)
850 msg_free(msg);
851 return 1;
852 }
853
854 uint8_t mqtt_flags = (publish_flags & MQTT_WSS_PUB_QOSMASK) << 1;
855 if (publish_flags & MQTT_WSS_PUB_RETAIN)
856 mqtt_flags |= MQTT_PUBLISH_RETAIN;
857
858 // Failure-path ownership contract with mqtt_ng_publish:
859 // - On MQTT_NG_MSGGEN_OK, msg is attached to a buffer fragment and the
860 // transaction buffer will call msg_free after the message is ack'd.
861 // - On any non-OK return, msg is never attached, so ownership stays with
862 // us and we must call msg_free here.
863 //
864 // Single-free invariant: msg is attached to a fragment only at the
865 // final frag_set_external_data() inside mqtt_ng_generate_publish(),
866 // after which the function commits unconditionally -- there is no
867 // `goto fail_rollback` between attachment and commit. Every reachable
868 // fail_rollback site therefore runs with msg unattached, so the
869 // rollback walks no msg-bearing fragment and msg_free() is only ever
870 // called by us. If a future change inserts a failure exit after
871 // attaching msg but before commit, the rollback would also invoke
872 // msg_free and this branch would double-free; preserve the invariant
873 // or move responsibility entirely into mqtt_ng_publish().
874 //
875 // - topic_free is intentionally NOT handled here. mqtt_ng_publish may
876 // attach topic to a fragment via optimized_add() before failing, in
877 // which case the rollback already invokes topic_free; if it fails
878 // earlier, topic_free is never invoked at all. The current callers all
879 // pass topic_free=NULL, so this asymmetry is harmless today.
880 int rc = mqtt_ng_publish(client->mqtt, topic, topic_free, msg, msg_free, msg_len, mqtt_flags, packet_id);
881 if (rc != MQTT_NG_MSGGEN_OK) {
882 if (packet_id)
883 *packet_id = 0;
884 if (msg_free)
885 msg_free(msg);
886 if (rc == MQTT_NG_MSGGEN_MSG_TOO_BIG)
887 return MQTT_WSS_ERR_MSG_TOO_BIG;
888 return rc;
889 }
890
891 mqtt_wss_wakeup(client);
892 return MQTT_WSS_OK;
893 }
894
895 int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level)
896 {
897 (void)max_qos_level; //TODO now hardcoded
898 if (!client->mqtt_connected) {
899 nd_log(NDLS_DAEMON, NDLP_ERR, "MQTT is offline. Can't subscribe.");
900 return 1;
901 }
902
903 if (client->mqtt_disconnecting) {
904 nd_log(NDLS_DAEMON, NDLP_ERR, "mqtt_wss is disconnecting can't subscribe");
905 return 1;
906 }
907
908 struct mqtt_sub sub = {
909 .topic = topic,
910 .topic_free = NULL,
911 .options = /* max_qos_level & 0x3 TODO when QOS > 1 implemented */ 0x01 | (0x01 << 3)
912 };
913 mqtt_ng_subscribe(client->mqtt, &sub, 1);
914
915 mqtt_wss_wakeup(client);
916 return 0;
917 }
918
919 struct mqtt_wss_stats mqtt_wss_get_stats(mqtt_wss_client client)
920 {
921 struct mqtt_wss_stats current;
922 spinlock_lock(&client->stat_lock);
923 current = client->stats;
924 spinlock_unlock(&client->stat_lock);
925 mqtt_ng_get_stats(client->mqtt, &current.mqtt);
926 return current;
927 }
928
929 void mqtt_wss_reset_stats(mqtt_wss_client client)
930 {
931 spinlock_lock(&client->stat_lock);
932 memset(&client->stats, 0, sizeof(client->stats));
933 spinlock_unlock(&client->stat_lock);
934 }
935
936 int mqtt_wss_set_topic_alias(mqtt_wss_client client, const char *topic)
937 {
938 return mqtt_ng_set_topic_alias(client->mqtt, topic);
939 }
940
941 #ifdef MQTT_WSS_DEBUG
942 void mqtt_wss_set_SSL_CTX_keylog_cb(mqtt_wss_client client, void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line))
943 {
944 client->ssl_ctx_keylog_cb = ssl_ctx_keylog_cb;
945 }
946 #endif