summaryrefslogtreecommitdiffstats
path: root/mqtt_websockets/src
diff options
context:
space:
mode:
Diffstat (limited to 'mqtt_websockets/src')
-rw-r--r--mqtt_websockets/src/common_public.c7
-rw-r--r--mqtt_websockets/src/include/common_internal.h37
-rw-r--r--mqtt_websockets/src/include/common_public.h33
-rw-r--r--mqtt_websockets/src/include/endian_compat.h41
-rw-r--r--mqtt_websockets/src/include/mqtt_constants.h101
-rw-r--r--mqtt_websockets/src/include/mqtt_ng.h97
-rw-r--r--mqtt_websockets/src/include/mqtt_wss_client.h172
-rw-r--r--mqtt_websockets/src/include/mqtt_wss_log.h37
-rw-r--r--mqtt_websockets/src/include/ws_client.h130
-rw-r--r--mqtt_websockets/src/mqtt_ng.c2233
-rw-r--r--mqtt_websockets/src/mqtt_wss_client.c1138
-rw-r--r--mqtt_websockets/src/mqtt_wss_log.c127
-rw-r--r--mqtt_websockets/src/test.c100
-rw-r--r--mqtt_websockets/src/ws_client.c744
14 files changed, 4997 insertions, 0 deletions
diff --git a/mqtt_websockets/src/common_public.c b/mqtt_websockets/src/common_public.c
new file mode 100644
index 00000000..7f74fa51
--- /dev/null
+++ b/mqtt_websockets/src/common_public.c
@@ -0,0 +1,7 @@
+#include "common_public.h"
+
+// this dummy exists to have a special pointer with special meaning
+// other than NULL
+void _caller_responsibility(void *ptr) {
+ (void)(ptr);
+}
diff --git a/mqtt_websockets/src/include/common_internal.h b/mqtt_websockets/src/include/common_internal.h
new file mode 100644
index 00000000..2e656355
--- /dev/null
+++ b/mqtt_websockets/src/include/common_internal.h
@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef COMMON_INTERNAL_H
+#define COMMON_INTERNAL_H
+
+#include "endian_compat.h"
+
+#ifdef MQTT_WSS_CUSTOM_ALLOC
+#include "mqtt_wss_pal.h"
+#else
+#define mw_malloc(...) malloc(__VA_ARGS__)
+#define mw_calloc(...) calloc(__VA_ARGS__)
+#define mw_free(...) free(__VA_ARGS__)
+#define mw_strdup(...) strdup(__VA_ARGS__)
+#define mw_realloc(...) realloc(__VA_ARGS__)
+#endif
+
+#ifndef MQTT_WSS_FRAG_MEMALIGN
+#define MQTT_WSS_FRAG_MEMALIGN (8)
+#endif
+
+#define OPENSSL_VERSION_095 0x00905100L
+#define OPENSSL_VERSION_097 0x00907000L
+#define OPENSSL_VERSION_110 0x10100000L
+#define OPENSSL_VERSION_111 0x10101000L
+
+#endif /* COMMON_INTERNAL_H */
diff --git a/mqtt_websockets/src/include/common_public.h b/mqtt_websockets/src/include/common_public.h
new file mode 100644
index 00000000..a855737f
--- /dev/null
+++ b/mqtt_websockets/src/include/common_public.h
@@ -0,0 +1,33 @@
+#ifndef MQTT_WEBSOCKETS_COMMON_PUBLIC_H
+#define MQTT_WEBSOCKETS_COMMON_PUBLIC_H
+
+#include <stddef.h>
+
+/* free_fnc_t in general (in whatever function or struct it is used)
+ * decides how the related data will be handled.
+ * - If NULL the data are copied internally (causing malloc and later free)
+ * - If pointer provided the free function pointed will be called when data are no longer needed
+ * to free associated memory. This is effectively transfering ownership of that pointer to the library.
+ * This also allows caller to provide custom free function other than system one.
+ * - If == CALLER_RESPONSIBILITY the library will not copy the data pointed to and will not call free
+ * at the end. This is usefull to avoid copying memory (and associated malloc/free) when data are for
+ * example static. In this case caller has to guarantee the memory pointed to will be valid for entire duration
+ * it is needed. For example by freeing the data after PUBACK is received or by data being static.
+ */
+typedef void (*free_fnc_t)(void *ptr);
+void _caller_responsibility(void *ptr);
+#define CALLER_RESPONSIBILITY ((free_fnc_t)&_caller_responsibility)
+
+struct mqtt_ng_stats {
+ size_t tx_bytes_queued;
+ int tx_messages_queued;
+ int tx_messages_sent;
+ int rx_messages_rcvd;
+ size_t tx_buffer_used;
+ size_t tx_buffer_free;
+ size_t tx_buffer_size;
+ // part of transaction buffer that containes mesages we can free alredy during the garbage colleciton step
+ size_t tx_buffer_reclaimable;
+};
+
+#endif /* MQTT_WEBSOCKETS_COMMON_PUBLIC_H */
diff --git a/mqtt_websockets/src/include/endian_compat.h b/mqtt_websockets/src/include/endian_compat.h
new file mode 100644
index 00000000..076ccbe8
--- /dev/null
+++ b/mqtt_websockets/src/include/endian_compat.h
@@ -0,0 +1,41 @@
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef MQTT_WSS_ENDIAN_COMPAT_H
+#define MQTT_WSS_ENDIAN_COMPAT_H
+
+#ifdef __APPLE__
+ #include <libkern/OSByteOrder.h>
+
+ #define htobe16(x) OSSwapHostToBigInt16(x)
+ #define htole16(x) OSSwapHostToLittleInt16(x)
+ #define be16toh(x) OSSwapBigToHostInt16(x)
+ #define le16toh(x) OSSwapLittleToHostInt16(x)
+
+ #define htobe32(x) OSSwapHostToBigInt32(x)
+ #define htole32(x) OSSwapHostToLittleInt32(x)
+ #define be32toh(x) OSSwapBigToHostInt32(x)
+ #define le32toh(x) OSSwapLittleToHostInt32(x)
+
+ #define htobe64(x) OSSwapHostToBigInt64(x)
+ #define htole64(x) OSSwapHostToLittleInt64(x)
+ #define be64toh(x) OSSwapBigToHostInt64(x)
+ #define le64toh(x) OSSwapLittleToHostInt64(x)
+#else
+#ifdef __FreeBSD__
+ #include <sys/endian.h>
+#else
+ #include <endian.h>
+#endif
+#endif
+
+#endif /* MQTT_WSS_ENDIAN_COMPAT_H */
diff --git a/mqtt_websockets/src/include/mqtt_constants.h b/mqtt_websockets/src/include/mqtt_constants.h
new file mode 100644
index 00000000..1db49897
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_constants.h
@@ -0,0 +1,101 @@
+#ifndef MQTT_CONSTANTS_H
+#define MQTT_CONSTANTS_H
+
+#define MQTT_MAX_QOS 0x02
+
+#define MQTT_VERSION_5_0 0x5
+
+/* [MQTT-1.5.5] most significant bit
+ of MQTT Variable Byte Integer signifies
+ there are more bytes following */
+#define MQTT_VBI_CONTINUATION_FLAG 0x80
+#define MQTT_VBI_DATA_MASK 0x7F
+#define MQTT_VBI_MAXBYTES 4
+
+/* MQTT control packet types as defined in
+ 2.1.2 MQTT Control Packet type */
+#define MQTT_CPT_CONNECT 0x1
+#define MQTT_CPT_CONNACK 0x2
+#define MQTT_CPT_PUBLISH 0x3
+#define MQTT_CPT_PUBACK 0x4
+#define MQTT_CPT_PUBREC 0x5
+#define MQTT_CPT_PUBREL 0x6
+#define MQTT_CPT_PUBCOMP 0x7
+#define MQTT_CPT_SUBSCRIBE 0x8
+#define MQTT_CPT_SUBACK 0x9
+#define MQTT_CPT_UNSUBSCRIBE 0xA
+#define MQTT_CPT_UNSUBACK 0xB
+#define MQTT_CPT_PINGREQ 0xC
+#define MQTT_CPT_PINGRESP 0xD
+#define MQTT_CPT_DISCONNECT 0xE
+#define MQTT_CPT_AUTH 0xF
+
+// MQTT CONNECT FLAGS (spec:3.1.2.3)
+#define MQTT_CONNECT_FLAG_USERNAME 0x80
+#define MQTT_CONNECT_FLAG_PASSWORD 0x40
+#define MQTT_CONNECT_FLAG_LWT_RETAIN 0x20
+#define MQTT_CONNECT_FLAG_LWT 0x04
+#define MQTT_CONNECT_FLAG_CLEAN_START 0x02
+
+#define MQTT_CONNECT_FLAG_QOS_MASK 0x18
+#define MQTT_CONNECT_FLAG_QOS_BITSHIFT 3
+
+#define MQTT_MAX_CLIENT_ID 23 /* [MQTT-3.1.3-5] */
+
+// MQTT Property identifiers [MQTT-2.2.2.2]
+#define MQTT_PROP_PAYLOAD_FMT_INDICATOR 0x01
+#define MQTT_PROP_PAYLOAD_FMT_INDICATOR_NAME "Payload Format Indicator"
+#define MQTT_PROP_MSG_EXPIRY_INTERVAL 0x02
+#define MQTT_PROP_MSG_EXPIRY_INTERVAL_NAME "Message Expiry Interval"
+#define MQTT_PROP_CONTENT_TYPE 0x03
+#define MQTT_PROP_CONTENT_TYPE_NAME "Content Type"
+#define MQTT_PROP_RESPONSE_TOPIC 0x08
+#define MQTT_PROP_RESPONSE_TOPIC_NAME "Response Topic"
+#define MQTT_PROP_CORRELATION_DATA 0x09
+#define MQTT_PROP_CORRELATION_DATA_NAME "Correlation Data"
+#define MQTT_PROP_SUB_IDENTIFIER 0x0B
+#define MQTT_PROP_SUB_IDENTIFIER_NAME "Subscription Identifier"
+#define MQTT_PROP_SESSION_EXPIRY_INTERVAL 0x11
+#define MQTT_PROP_SESSION_EXPIRY_INTERVAL_NAME "Session Expiry Interval"
+#define MQTT_PROP_ASSIGNED_CLIENT_ID 0x12
+#define MQTT_PROP_ASSIGNED_CLIENT_ID_NAME "Assigned Client Identifier"
+#define MQTT_PROP_SERVER_KEEP_ALIVE 0x13
+#define MQTT_PROP_SERVER_KEEP_ALIVE_NAME "Server Keep Alive"
+#define MQTT_PROP_AUTH_METHOD 0x15
+#define MQTT_PROP_AUTH_METHOD_NAME "Authentication Method"
+#define MQTT_PROP_AUTH_DATA 0x16
+#define MQTT_PROP_AUTH_DATA_NAME "Authentication Data"
+#define MQTT_PROP_REQ_PROBLEM_INFO 0x17
+#define MQTT_PROP_REQ_PROBLEM_INFO_NAME "Request Problem Information"
+#define MQTT_PROP_WILL_DELAY_INTERVAL 0x18
+#define MQTT_PROP_WIIL_DELAY_INTERVAL_NAME "Will Delay Interval"
+#define MQTT_PROP_REQ_RESP_INFORMATION 0x19
+#define MQTT_PROP_REQ_RESP_INFORMATION_NAME "Request Response Information"
+#define MQTT_PROP_RESP_INFORMATION 0x1A
+#define MQTT_PROP_RESP_INFORMATION_NAME "Response Information"
+#define MQTT_PROP_SERVER_REF 0x1C
+#define MQTT_PROP_SERVER_REF_NAME "Server Reference"
+#define MQTT_PROP_REASON_STR 0x1F
+#define MQTT_PROP_REASON_STR_NAME "Reason String"
+#define MQTT_PROP_RECEIVE_MAX 0x21
+#define MQTT_PROP_RECEIVE_MAX_NAME "Receive Maximum"
+#define MQTT_PROP_TOPIC_ALIAS_MAX 0x22
+#define MQTT_PROP_TOPIC_ALIAS_MAX_NAME "Topic Alias Maximum"
+#define MQTT_PROP_TOPIC_ALIAS 0x23
+#define MQTT_PROP_TOPIC_ALIAS_NAME "Topic Alias"
+#define MQTT_PROP_MAX_QOS 0x24
+#define MQTT_PROP_MAX_QOS_NAME "Maximum QoS"
+#define MQTT_PROP_RETAIN_AVAIL 0x25
+#define MQTT_PROP_RETAIN_AVAIL_NAME "Retain Available"
+#define MQTT_PROP_USR 0x26
+#define MQTT_PROP_USR_NAME "User Property"
+#define MQTT_PROP_MAX_PKT_SIZE 0x27
+#define MQTT_PROP_MAX_PKT_SIZE_NAME "Maximum Packet Size"
+#define MQTT_PROP_WILDCARD_SUB_AVAIL 0x28
+#define MQTT_PROP_WILDCARD_SUB_AVAIL_NAME "Wildcard Subscription Available"
+#define MQTT_PROP_SUB_ID_AVAIL 0x29
+#define MQTT_PROP_SUB_ID_AVAIL_NAME "Subscription Identifier Available"
+#define MQTT_PROP_SHARED_SUB_AVAIL 0x2A
+#define MQTT_PROP_SHARED_SUB_AVAIL_NAME "Shared Subscription Available"
+
+#endif /* MQTT_CONSTANTS_H */
diff --git a/mqtt_websockets/src/include/mqtt_ng.h b/mqtt_websockets/src/include/mqtt_ng.h
new file mode 100644
index 00000000..09668d09
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_ng.h
@@ -0,0 +1,97 @@
+#include <stdint.h>
+#include <sys/types.h>
+#include <time.h>
+
+#include "ringbuffer.h"
+#include "common_public.h"
+
+#define MQTT_NG_MSGGEN_OK 0
+// MQTT_NG_MSGGEN_USER_ERROR means parameters given to this function
+// do not make sense or are out of MQTT specs
+#define MQTT_NG_MSGGEN_USER_ERROR 1
+#define MQTT_NG_MSGGEN_BUFFER_OOM 2
+#define MQTT_NG_MSGGEN_MSG_TOO_BIG 3
+
+struct mqtt_ng_client;
+
+/* Converts integer to MQTT Variable Byte Integer as per 1.5.5 of MQTT 5 specs
+ * @param input value to be converted
+ * @param output pointer to memory where output will be written to. Must allow up to 4 bytes to be written.
+ * @return number of bytes written to output or <= 0 if error in which case contents of output are undefined
+ */
+int uint32_to_mqtt_vbi(uint32_t input, char *output);
+
+struct mqtt_lwt_properties {
+ char *will_topic;
+ free_fnc_t will_topic_free;
+
+ void *will_message;
+ free_fnc_t will_message_free;
+ size_t will_message_size;
+
+ int will_qos;
+ int will_retain;
+};
+
+struct mqtt_auth_properties {
+ char *client_id;
+ free_fnc_t client_id_free;
+ char *username;
+ free_fnc_t username_free;
+ char *password;
+ free_fnc_t password_free;
+};
+
+int mqtt_ng_connect(struct mqtt_ng_client *client,
+ struct mqtt_auth_properties *auth,
+ struct mqtt_lwt_properties *lwt,
+ uint8_t clean_start,
+ uint16_t keep_alive);
+
+int mqtt_ng_publish(struct mqtt_ng_client *client,
+ char *topic,
+ free_fnc_t topic_free,
+ void *msg,
+ free_fnc_t msg_free,
+ size_t msg_len,
+ uint8_t publish_flags,
+ uint16_t *packet_id);
+
+struct mqtt_sub {
+ char *topic;
+ free_fnc_t topic_free;
+ uint8_t options;
+};
+
+int mqtt_ng_subscribe(struct mqtt_ng_client *client, struct mqtt_sub *subscriptions, size_t subscription_count);
+
+int mqtt_ng_ping(struct mqtt_ng_client *client);
+
+typedef ssize_t (*mqtt_ng_send_fnc_t)(void *user_ctx, const void* buf, size_t len);
+
+struct mqtt_ng_init {
+ mqtt_wss_log_ctx_t log;
+ rbuf_t data_in;
+ mqtt_ng_send_fnc_t data_out_fnc;
+ void *user_ctx;
+
+ void (*puback_callback)(uint16_t packet_id);
+ void (*connack_callback)(void* user_ctx, int connack_reply);
+ void (*msg_callback)(const char *topic, const void *msg, size_t msglen, int qos);
+};
+
+struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings);
+
+void mqtt_ng_destroy(struct mqtt_ng_client *client);
+
+int mqtt_ng_disconnect(struct mqtt_ng_client *client, uint8_t reason_code);
+
+int mqtt_ng_sync(struct mqtt_ng_client *client);
+
+time_t mqtt_ng_last_send_time(struct mqtt_ng_client *client);
+
+void mqtt_ng_set_max_mem(struct mqtt_ng_client *client, size_t bytes);
+
+void mqtt_ng_get_stats(struct mqtt_ng_client *client, struct mqtt_ng_stats *stats);
+
+int mqtt_ng_set_topic_alias(struct mqtt_ng_client *client, const char *topic);
diff --git a/mqtt_websockets/src/include/mqtt_wss_client.h b/mqtt_websockets/src/include/mqtt_wss_client.h
new file mode 100644
index 00000000..e325961b
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_wss_client.h
@@ -0,0 +1,172 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef MQTT_WSS_CLIENT_H
+#define MQTT_WSS_CLIENT_H
+
+#include <stdint.h>
+#include <stddef.h> //size_t
+
+#include "mqtt_wss_log.h"
+#include "common_public.h"
+
+// All OK call me at your earliest convinience
+#define MQTT_WSS_OK 0
+/* All OK, poll timeout you requested when calling mqtt_wss_service expired - you might want to know if timeout
+ * happened or we got some data or handle same as MQTT_WSS_OK
+ */
+#define MQTT_WSS_OK_TO 1
+// Connection was closed by remote
+#define MQTT_WSS_ERR_CONN_DROP -1
+// Error in MQTT protocol (e.g. malformed packet)
+#define MQTT_WSS_ERR_PROTO_MQTT -2
+// Error in WebSocket protocol (e.g. malformed packet)
+#define MQTT_WSS_ERR_PROTO_WS -3
+
+#define MQTT_WSS_ERR_TX_BUF_TOO_SMALL -4
+#define MQTT_WSS_ERR_RX_BUF_TOO_SMALL -5
+
+#define MQTT_WSS_ERR_TOO_BIG_FOR_SERVER -6
+// if client was initialized with MQTT 3 but MQTT 5 feature
+// was requested by user of library
+#define MQTT_WSS_ERR_CANT_DO -8
+
+typedef struct mqtt_wss_client_struct *mqtt_wss_client;
+
+typedef void (*msg_callback_fnc_t)(const char *topic, const void *msg, size_t msglen, int qos);
+/* Creates new instance of MQTT over WSS. Doesn't start connection.
+ * @param log_prefix this is prefix to be used when logging to discern between multiple
+ * mqtt_wss instances. Can be NULL.
+ * @param log_callback is function pointer to fnc to be called when mqtt_wss wants
+ * to log. This allows plugging this library into your own logging system/solution.
+ * If NULL STDOUT/STDERR will be used.
+ * @param msg_callback is function pointer to function which will be called
+ * when application level message arrives from broker (for subscribed topics).
+ * Can be NULL if you are not interested about incoming messages.
+ * @param puback_callback is function pointer to function to be called when QOS1 Publish
+ * is acknowledged by server
+ */
+mqtt_wss_client mqtt_wss_new(const char *log_prefix,
+ mqtt_wss_log_callback_t log_callback,
+ msg_callback_fnc_t msg_callback,
+ void (*puback_callback)(uint16_t packet_id));
+
+void mqtt_wss_set_max_buf_size(mqtt_wss_client client, size_t size);
+
+void mqtt_wss_destroy(mqtt_wss_client client);
+
+struct mqtt_connect_params;
+struct mqtt_wss_proxy;
+
+#define MQTT_WSS_SSL_CERT_CHECK_FULL 0x00
+#define MQTT_WSS_SSL_ALLOW_SELF_SIGNED 0x01
+#define MQTT_WSS_SSL_DONT_CHECK_CERTS 0x08
+
+/* Will block until the MQTT over WSS connection is established or return error
+ * @param client mqtt_wss_client which should connect
+ * @param host to connect to (where MQTT over WSS server is listening)
+ * @param port to connect to (where MQTT over WSS server is listening)
+ * @param mqtt_params pointer to mqtt_connect_params structure which contains MQTT credentials and settings
+ * @param ssl_flags parameters for OpenSSL, 0=MQTT_WSS_SSL_CERT_CHECK_FULL
+ */
+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);
+int mqtt_wss_service(mqtt_wss_client client, int timeout_ms);
+void mqtt_wss_disconnect(mqtt_wss_client client, int timeout_ms);
+
+// we redefine this instead of using MQTT-C flags as in future
+// we want to support different MQTT implementations if needed
+enum mqtt_wss_publish_flags {
+ MQTT_WSS_PUB_QOS0 = 0x0,
+ MQTT_WSS_PUB_QOS1 = 0x1,
+ MQTT_WSS_PUB_QOS2 = 0x2,
+ MQTT_WSS_PUB_QOSMASK = 0x3,
+ MQTT_WSS_PUB_RETAIN = 0x4
+};
+
+struct mqtt_connect_params {
+ const char *clientid;
+ const char *username;
+ const char *password;
+ const char *will_topic;
+ const void *will_msg;
+ enum mqtt_wss_publish_flags will_flags;
+ size_t will_msg_len;
+ int keep_alive;
+ int drop_on_publish_fail;
+};
+
+enum mqtt_wss_proxy_type {
+ MQTT_WSS_DIRECT = 0,
+ MQTT_WSS_PROXY_HTTP
+};
+
+struct mqtt_wss_proxy {
+ enum mqtt_wss_proxy_type type;
+ const char *host;
+ int port;
+ const char *username;
+ const char *password;
+};
+
+/* TODO!!! update the description
+ * Publishes MQTT message and gets message id
+ * @param client mqtt_wss_client which should transfer the message
+ * @param topic MQTT topic to publish message to (0 terminated C string)
+ * @param msg Message to be published (no need for 0 termination)
+ * @param msg_len Length of the message to be published
+ * @param publish_flags see enum mqtt_wss_publish_flags e.g. (MQTT_WSS_PUB_QOS1 | MQTT_WSS_PUB_RETAIN)
+ * @param packet_id is 16 bit unsigned int representing ID that can be used to pair with PUBACK callback
+ * for usages where application layer wants to know which messages are delivered when
+ * @return Returns 0 on success
+ */
+int mqtt_wss_publish5(mqtt_wss_client client,
+ char *topic,
+ free_fnc_t topic_free,
+ void *msg,
+ free_fnc_t msg_free,
+ size_t msg_len,
+ uint8_t publish_flags,
+ uint16_t *packet_id);
+
+int mqtt_wss_set_topic_alias(mqtt_wss_client client, const char *topic);
+
+/* Subscribes to MQTT topic
+ * @param client mqtt_wss_client which should do the subscription
+ * @param topic MQTT topic to subscribe to
+ * @param max_qos_level maximum QOS level that broker can send to us on this subscription
+ * @return Returns 0 on success
+ */
+int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level);
+
+
+struct mqtt_wss_stats {
+ uint64_t bytes_tx;
+ uint64_t bytes_rx;
+#ifdef MQTT_WSS_CPUSTATS
+ uint64_t time_keepalive;
+ uint64_t time_read_socket;
+ uint64_t time_write_socket;
+ uint64_t time_process_websocket;
+ uint64_t time_process_mqtt;
+#endif
+ struct mqtt_ng_stats mqtt;
+};
+
+struct mqtt_wss_stats mqtt_wss_get_stats(mqtt_wss_client client);
+
+#ifdef MQTT_WSS_DEBUG
+#include <openssl/ssl.h>
+void mqtt_wss_set_SSL_CTX_keylog_cb(mqtt_wss_client client, void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line));
+#endif
+
+#endif /* MQTT_WSS_CLIENT_H */
diff --git a/mqtt_websockets/src/include/mqtt_wss_log.h b/mqtt_websockets/src/include/mqtt_wss_log.h
new file mode 100644
index 00000000..a33c460c
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_wss_log.h
@@ -0,0 +1,37 @@
+#ifndef MQTT_WSS_LOG_H
+#define MQTT_WSS_LOG_H
+
+typedef enum mqtt_wss_log_type {
+ MQTT_WSS_LOG_DEBUG = 0x01,
+ MQTT_WSS_LOG_INFO = 0x02,
+ MQTT_WSS_LOG_WARN = 0x03,
+ MQTT_WSS_LOG_ERROR = 0x81,
+ MQTT_WSS_LOG_FATAL = 0x88
+} mqtt_wss_log_type_t;
+
+typedef void (*mqtt_wss_log_callback_t)(mqtt_wss_log_type_t, const char*);
+
+typedef struct mqtt_wss_log_ctx *mqtt_wss_log_ctx_t;
+
+/** Creates logging context with optional prefix and optional callback
+ * @param ctx_prefix String to be prefixed to every log message.
+ * This is useful if multiple clients are instantiated to be able to
+ * know which one this message belongs to. Can be `NULL` for no prefix.
+ * @param log_callback Callback to be called instead of logging to
+ * `STDOUT` or `STDERR` (if debug enabled otherwise silent). Callback has to be
+ * pointer to function of `void function(mqtt_wss_log_type_t, const char*)` type.
+ * If `NULL` default will be used (silent or STDERR/STDOUT).
+ * @return mqtt_wss_log_ctx_t or `NULL` on error */
+mqtt_wss_log_ctx_t mqtt_wss_log_ctx_create(const char *ctx_prefix, mqtt_wss_log_callback_t log_callback);
+
+/** Destroys logging context and cleans up the memory
+ * @param ctx Context to destroy */
+void mqtt_wss_log_ctx_destroy(mqtt_wss_log_ctx_t ctx);
+
+void mws_fatal(mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_error(mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_warn (mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_info (mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_debug(mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+
+#endif /* MQTT_WSS_LOG_H */
diff --git a/mqtt_websockets/src/include/ws_client.h b/mqtt_websockets/src/include/ws_client.h
new file mode 100644
index 00000000..de4fac40
--- /dev/null
+++ b/mqtt_websockets/src/include/ws_client.h
@@ -0,0 +1,130 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef WS_CLIENT_H
+#define WS_CLIENT_H
+
+#include "ringbuffer.h"
+#include "mqtt_wss_log.h"
+
+#include <stdint.h>
+
+#define WS_CLIENT_NEED_MORE_BYTES 0x10
+#define WS_CLIENT_PARSING_DONE 0x11
+#define WS_CLIENT_CONNECTION_CLOSED 0x12
+#define WS_CLIENT_PROTOCOL_ERROR -0x10
+#define WS_CLIENT_BUFFER_FULL -0x11
+#define WS_CLIENT_INTERNAL_ERROR -0x12
+
+enum websocket_client_conn_state {
+ WS_RAW = 0,
+ WS_HANDSHAKE,
+ WS_ESTABLISHED,
+ WS_ERROR, // connection has to be restarted if this is reached
+ WS_CONN_CLOSED_GRACEFUL
+};
+
+enum websocket_client_hdr_parse_state {
+ WS_HDR_HTTP = 0, // need to check HTTP/1.1
+ WS_HDR_RC, // need to read HTTP code
+ WS_HDR_ENDLINE, // need to read rest of the first line
+ WS_HDR_PARSE_HEADERS, // rest of the header until CRLF CRLF
+ WS_HDR_PARSE_DONE,
+ WS_HDR_ALL_DONE
+};
+
+enum websocket_client_rx_ws_parse_state {
+ WS_FIRST_2BYTES = 0,
+ WS_PAYLOAD_EXTENDED_16,
+ WS_PAYLOAD_EXTENDED_64,
+ WS_PAYLOAD_DATA, // BINARY payload to be passed to MQTT
+ WS_PAYLOAD_CONNECTION_CLOSE,
+ WS_PAYLOAD_CONNECTION_CLOSE_EC,
+ WS_PAYLOAD_CONNECTION_CLOSE_MSG,
+ WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD,
+ WS_PAYLOAD_PING_REQ_PAYLOAD, // PING payload to be sent back as PONG
+ WS_PACKET_DONE
+};
+
+enum websocket_opcode {
+ WS_OP_CONTINUATION_FRAME = 0x0,
+ WS_OP_TEXT_FRAME = 0x1,
+ WS_OP_BINARY_FRAME = 0x2,
+ WS_OP_CONNECTION_CLOSE = 0x8,
+ WS_OP_PING = 0x9,
+ WS_OP_PONG = 0xA
+};
+
+struct ws_op_close_payload {
+ uint16_t ec;
+ char *reason;
+};
+
+struct http_header {
+ char *key;
+ char *value;
+ struct http_header *next;
+};
+
+typedef struct websocket_client {
+ enum websocket_client_conn_state state;
+
+ struct ws_handshake {
+ enum websocket_client_hdr_parse_state hdr_state;
+ char *nonce_reply;
+ int nonce_matched;
+ int http_code;
+ char *http_reply_msg;
+ struct http_header *headers;
+ struct http_header *headers_tail;
+ int hdr_count;
+ } hs;
+
+ struct ws_rx {
+ enum websocket_client_rx_ws_parse_state parse_state;
+ enum websocket_opcode opcode;
+ uint64_t payload_length;
+ uint64_t payload_processed;
+ union {
+ struct ws_op_close_payload op_close;
+ char *ping_msg;
+ } specific_data;
+ } rx;
+
+ rbuf_t buf_read; // from SSL
+ rbuf_t buf_write; // to SSL and then to socket
+ // TODO if ringbuffer gets multiple tail support
+ // we can work without buf_to_mqtt and thus reduce
+ // memory usage and remove one more memcpy buf_read->buf_to_mqtt
+ rbuf_t buf_to_mqtt; // RAW data for MQTT lib
+
+ int entropy_fd;
+
+ // careful host is borrowed, don't free
+ char **host;
+ mqtt_wss_log_ctx_t log;
+} ws_client;
+
+ws_client *ws_client_new(size_t buf_size, char **host, mqtt_wss_log_ctx_t log);
+void ws_client_destroy(ws_client *client);
+void ws_client_reset(ws_client *client);
+
+int ws_client_start_handshake(ws_client *client);
+
+int ws_client_want_write(ws_client *client);
+
+int ws_client_process(ws_client *client);
+
+int ws_client_send(ws_client *client, enum websocket_opcode frame_type, const char *data, size_t size);
+
+#endif /* WS_CLIENT_H */
diff --git a/mqtt_websockets/src/mqtt_ng.c b/mqtt_websockets/src/mqtt_ng.c
new file mode 100644
index 00000000..81cffccf
--- /dev/null
+++ b/mqtt_websockets/src/mqtt_ng.c
@@ -0,0 +1,2233 @@
+#define _GNU_SOURCE
+
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <pthread.h>
+#include <inttypes.h>
+
+#include "c_rhash.h"
+
+#include "common_internal.h"
+#include "mqtt_constants.h"
+#include "mqtt_wss_log.h"
+#include "mqtt_ng.h"
+
+#define UNIT_LOG_PREFIX "mqtt_client: "
+#define FATAL(fmt, ...) mws_fatal(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define ERROR(fmt, ...) mws_error(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define WARN(fmt, ...) mws_warn (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define INFO(fmt, ...) mws_info (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define DEBUG(fmt, ...) mws_debug(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+
+#define SMALL_STRING_DONT_FRAGMENT_LIMIT 128
+
+#define MIN(a,b) (((a)<(b))?(a):(b))
+
+#define LOCK_HDR_BUFFER(buffer) pthread_mutex_lock(&((buffer)->mutex))
+#define UNLOCK_HDR_BUFFER(buffer) pthread_mutex_unlock(&((buffer)->mutex))
+
+#define BUFFER_FRAG_GARBAGE_COLLECT 0x01
+// some packets can be marked for garbage collection
+// immediately when they are sent (e.g. sent PUBACK on QoS1)
+#define BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND 0x02
+// as buffer fragment can point to both
+// external data and data in the same buffer
+// we mark the former case with BUFFER_FRAG_DATA_EXTERNAL
+#define BUFFER_FRAG_DATA_EXTERNAL 0x04
+// as single MQTT Packet can be stored into multiple
+// buffer fragments (depending on copy requirements)
+// this marks this fragment to be the first/last
+#define BUFFER_FRAG_MQTT_PACKET_HEAD 0x10
+#define BUFFER_FRAG_MQTT_PACKET_TAIL 0x20
+
+typedef uint16_t buffer_frag_flag_t;
+struct buffer_fragment {
+ size_t len;
+ size_t sent;
+ buffer_frag_flag_t flags;
+ void (*free_fnc)(void *ptr);
+ char *data;
+
+ uint16_t packet_id;
+
+ struct buffer_fragment *next;
+};
+
+typedef struct buffer_fragment *mqtt_msg_data;
+
+// buffer used for MQTT headers only
+// not for actual data sent
+struct header_buffer {
+ size_t size;
+ char *data;
+ char *tail;
+ struct buffer_fragment *tail_frag;
+};
+
+struct transaction_buffer {
+ struct header_buffer hdr_buffer;
+ // used while building new message
+ // to be able to revert state easily
+ // in case of error mid processing
+ struct header_buffer state_backup;
+ pthread_mutex_t mutex;
+ struct buffer_fragment *sending_frag;
+};
+
+enum mqtt_client_state {
+ RAW = 0,
+ CONNECT_PENDING,
+ CONNECTING,
+ CONNECTED,
+ ERROR,
+ DISCONNECTED
+};
+
+enum parser_state {
+ MQTT_PARSE_FIXED_HEADER_PACKET_TYPE = 0,
+ MQTT_PARSE_FIXED_HEADER_LEN,
+ MQTT_PARSE_VARIABLE_HEADER,
+ MQTT_PARSE_MQTT_PACKET_DONE
+};
+
+enum varhdr_parser_state {
+ MQTT_PARSE_VARHDR_INITIAL = 0,
+ MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE,
+ MQTT_PARSE_VARHDR_PROPS,
+ MQTT_PARSE_VARHDR_TOPICNAME,
+ MQTT_PARSE_VARHDR_POST_TOPICNAME,
+ MQTT_PARSE_VARHDR_PACKET_ID,
+ MQTT_PARSE_REASONCODES,
+ MQTT_PARSE_PAYLOAD
+};
+
+struct mqtt_vbi_parser_ctx {
+ char data[MQTT_VBI_MAXBYTES];
+ uint8_t bytes;
+ uint32_t result;
+};
+
+enum mqtt_datatype {
+ MQTT_TYPE_UNKNOWN = 0,
+ MQTT_TYPE_UINT_8,
+ MQTT_TYPE_UINT_16,
+ MQTT_TYPE_UINT_32,
+ MQTT_TYPE_VBI,
+ MQTT_TYPE_STR,
+ MQTT_TYPE_STR_PAIR,
+ MQTT_TYPE_BIN
+};
+
+struct mqtt_property {
+ uint8_t id;
+ enum mqtt_datatype type;
+ union {
+ char *strings[2];
+ void *bindata;
+ uint8_t uint8;
+ uint16_t uint16;
+ uint32_t uint32;
+ } data;
+ size_t bindata_len;
+ struct mqtt_property *next;
+};
+
+enum mqtt_properties_parser_state {
+ PROPERTIES_LENGTH = 0,
+ PROPERTY_CREATE,
+ PROPERTY_ID,
+ PROPERTY_TYPE_UINT8,
+ PROPERTY_TYPE_UINT16,
+ PROPERTY_TYPE_UINT32,
+ PROPERTY_TYPE_STR_BIN_LEN,
+ PROPERTY_TYPE_STR,
+ PROPERTY_TYPE_BIN,
+ PROPERTY_TYPE_VBI,
+ PROPERTY_NEXT
+};
+
+struct mqtt_properties_parser_ctx {
+ enum mqtt_properties_parser_state state;
+ struct mqtt_property *head;
+ struct mqtt_property *tail;
+ uint32_t properties_length;
+ uint32_t vbi_length;
+ struct mqtt_vbi_parser_ctx vbi_parser_ctx;
+ size_t bytes_consumed;
+ int str_idx;
+};
+
+struct mqtt_connack {
+ uint8_t flags;
+ uint8_t reason_code;
+};
+struct mqtt_puback {
+ uint16_t packet_id;
+ uint8_t reason_code;
+};
+
+struct mqtt_suback {
+ uint16_t packet_id;
+ uint8_t *reason_codes;
+ uint8_t reason_code_count;
+ uint8_t reason_codes_pending;
+};
+
+struct mqtt_publish {
+ uint16_t topic_len;
+ char *topic;
+ uint16_t packet_id;
+ size_t data_len;
+ char *data;
+ uint8_t qos;
+};
+
+struct mqtt_disconnect {
+ uint8_t reason_code;
+};
+
+struct mqtt_ng_parser {
+ rbuf_t received_data;
+
+ uint8_t mqtt_control_packet_type;
+ uint32_t mqtt_fixed_hdr_remaining_length;
+ size_t mqtt_parsed_len;
+
+ struct mqtt_vbi_parser_ctx vbi_parser;
+ struct mqtt_properties_parser_ctx properties_parser;
+
+ enum parser_state state;
+ enum varhdr_parser_state varhdr_state;
+
+ struct mqtt_property *varhdr_properties;
+
+ union {
+ struct mqtt_connack connack;
+ struct mqtt_puback puback;
+ struct mqtt_suback suback;
+ struct mqtt_publish publish;
+ struct mqtt_disconnect disconnect;
+ } mqtt_packet;
+};
+
+struct topic_alias_data {
+ uint16_t idx;
+ uint32_t usage_count;
+};
+
+struct topic_aliases_data {
+ c_rhash stoi_dict;
+ uint32_t idx_max;
+ uint32_t idx_assigned;
+ pthread_rwlock_t rwlock;
+};
+
+struct mqtt_ng_client {
+ struct transaction_buffer main_buffer;
+
+ enum mqtt_client_state client_state;
+
+ mqtt_msg_data connect_msg;
+
+ mqtt_wss_log_ctx_t log;
+
+ mqtt_ng_send_fnc_t send_fnc_ptr;
+ void *user_ctx;
+
+ // time when last fragment of MQTT message was sent
+ time_t time_of_last_send;
+
+ struct mqtt_ng_parser parser;
+
+ size_t max_mem_bytes;
+
+ void (*puback_callback)(uint16_t packet_id);
+ void (*connack_callback)(void* user_ctx, int connack_reply);
+ void (*msg_callback)(const char *topic, const void *msg, size_t msglen, int qos);
+
+ unsigned int ping_pending:1;
+
+ struct mqtt_ng_stats stats;
+ pthread_mutex_t stats_mutex;
+
+ struct topic_aliases_data tx_topic_aliases;
+ c_rhash rx_aliases;
+
+ size_t max_msg_size;
+};
+
+char pingreq[] = { MQTT_CPT_PINGREQ << 4, 0x00 };
+
+struct buffer_fragment ping_frag = {
+ .data = pingreq,
+ .flags = BUFFER_FRAG_MQTT_PACKET_HEAD | BUFFER_FRAG_MQTT_PACKET_TAIL,
+ .free_fnc = NULL,
+ .len = sizeof(pingreq),
+ .next = NULL,
+ .sent = 0,
+ .packet_id = 0
+};
+
+int uint32_to_mqtt_vbi(uint32_t input, char *output) {
+ int i = 1;
+ *output = 0;
+
+ /* MQTT 5 specs allows max 4 bytes of output
+ making it 0xFF, 0xFF, 0xFF, 0x7F
+ representing number 268435455 decimal
+ see 1.5.5. Variable Byte Integer */
+ if(input >= 256 * 1024 * 1024)
+ return 0;
+
+ if(!input) {
+ *output = 0;
+ return 1;
+ }
+
+ while(input) {
+ output[i-1] = input & MQTT_VBI_DATA_MASK;
+ input >>= 7;
+ if (input)
+ output[i-1] |= MQTT_VBI_CONTINUATION_FLAG;
+ i++;
+ }
+ return i - 1;
+}
+
+int mqtt_vbi_to_uint32(char *input, uint32_t *output) {
+ // dont want to operate directly on output
+ // as I want it to be possible for input and output
+ // pointer to be the same
+ uint32_t result = 0;
+ uint32_t multiplier = 1;
+
+ do {
+ result += (uint32_t)(*input & MQTT_VBI_DATA_MASK) * multiplier;
+ if (multiplier > 128*128*128)
+ return 1;
+ multiplier <<= 7;
+ } while (*input++ & MQTT_VBI_CONTINUATION_FLAG);
+ *output = result;
+ return 0;
+}
+
+#ifdef TESTS
+#include <stdio.h>
+#define MQTT_VBI_MAXLEN 4
+// we add extra byte to check we dont write out of bounds
+// in case where 4 bytes are supposed to be written
+static const char _mqtt_vbi_0[MQTT_VBI_MAXLEN + 1] = { 0x00, 0x00, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_127[MQTT_VBI_MAXLEN + 1] = { 0x7F, 0x00, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_128[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x01, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_16383[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0x7F, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_16384[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x01, 0x00, 0x00 };
+static const char _mqtt_vbi_2097151[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0xFF, 0x7F, 0x00, 0x00 };
+static const char _mqtt_vbi_2097152[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x80, 0x01, 0x00 };
+static const char _mqtt_vbi_268435455[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0xFF, 0xFF, 0x7F, 0x00 };
+static const char _mqtt_vbi_999999999[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x80, 0x80, 0x01 };
+
+#define MQTT_VBI_TESTCASE(case, expected_len) \
+ { \
+ memset(buf, 0, MQTT_VBI_MAXLEN + 1); \
+ int len; \
+ if ((len=uint32_to_mqtt_vbi(case, buf)) != expected_len) { \
+ fprintf(stderr, "uint32_to_mqtt_vbi(case:%d, line:%d): Incorrect length returned. Expected %d, Got %d\n", case, __LINE__, expected_len, len); \
+ return 1; \
+ } \
+ if (memcmp(buf, _mqtt_vbi_ ## case, MQTT_VBI_MAXLEN + 1 )) { \
+ fprintf(stderr, "uint32_to_mqtt_vbi(case:%d, line:%d): Wrong output\n", case, __LINE__); \
+ return 1; \
+ } }
+
+
+int test_uint32_mqtt_vbi() {
+ char buf[MQTT_VBI_MAXLEN + 1];
+
+ MQTT_VBI_TESTCASE(0, 1)
+ MQTT_VBI_TESTCASE(127, 1)
+ MQTT_VBI_TESTCASE(128, 2)
+ MQTT_VBI_TESTCASE(16383, 2)
+ MQTT_VBI_TESTCASE(16384, 3)
+ MQTT_VBI_TESTCASE(2097151, 3)
+ MQTT_VBI_TESTCASE(2097152, 4)
+ MQTT_VBI_TESTCASE(268435455, 4)
+
+ memset(buf, 0, MQTT_VBI_MAXLEN + 1);
+ int len;
+ if ((len=uint32_to_mqtt_vbi(268435456, buf)) != 0) {
+ fprintf(stderr, "uint32_to_mqtt_vbi(case:268435456, line:%d): Incorrect length returned. Expected 0, Got %d\n", __LINE__, len);
+ return 1;
+ }
+
+ return 0;
+}
+
+#define MQTT_VBI2UINT_TESTCASE(case, expected_error) \
+ { \
+ uint32_t result; \
+ int ret = mqtt_vbi_to_uint32(_mqtt_vbi_ ## case, &result); \
+ if (ret && !(expected_error)) { \
+ fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Unexpectedly Errored\n", (case), __LINE__); \
+ return 1; \
+ } \
+ if (!ret && (expected_error)) { \
+ fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Should return error but didnt\n", (case), __LINE__); \
+ return 1; \
+ } \
+ if (!ret && result != (case)) { \
+ fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Returned wrong result %d\n", (case), __LINE__, result); \
+ return 1; \
+ }}
+
+
+int test_mqtt_vbi_to_uint32() {
+ MQTT_VBI2UINT_TESTCASE(0, 0)
+ MQTT_VBI2UINT_TESTCASE(127, 0)
+ MQTT_VBI2UINT_TESTCASE(128, 0)
+ MQTT_VBI2UINT_TESTCASE(16383, 0)
+ MQTT_VBI2UINT_TESTCASE(16384, 0)
+ MQTT_VBI2UINT_TESTCASE(2097151, 0)
+ MQTT_VBI2UINT_TESTCASE(2097152, 0)
+ MQTT_VBI2UINT_TESTCASE(268435455, 0)
+ MQTT_VBI2UINT_TESTCASE(999999999, 1)
+ return 0;
+}
+#endif /* TESTS */
+
+// this helps with switch statements
+// as they have to use integer type (not pointer)
+enum memory_mode {
+ MEMCPY,
+ EXTERNAL_FREE_AFTER_USE,
+ CALLER_RESPONSIBLE
+};
+
+static inline enum memory_mode ptr2memory_mode(void * ptr) {
+ if (ptr == NULL)
+ return MEMCPY;
+ if (ptr == CALLER_RESPONSIBILITY)
+ return CALLER_RESPONSIBLE;
+ return EXTERNAL_FREE_AFTER_USE;
+}
+
+#define frag_is_marked_for_gc(frag) ((frag->flags & BUFFER_FRAG_GARBAGE_COLLECT) || ((frag->flags & BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND) && frag->sent == frag->len))
+#define FRAG_SIZE_IN_BUFFER(frag) (sizeof(struct buffer_fragment) + ((frag->flags & BUFFER_FRAG_DATA_EXTERNAL) ? 0 : frag->len))
+
+static void buffer_frag_free_data(struct buffer_fragment *frag)
+{
+ if ( frag->flags & BUFFER_FRAG_DATA_EXTERNAL && frag->data != NULL) {
+ switch (ptr2memory_mode(frag->free_fnc)) {
+ case MEMCPY:
+ mw_free(frag->data);
+ break;
+ case EXTERNAL_FREE_AFTER_USE:
+ frag->free_fnc(frag->data);
+ break;
+ case CALLER_RESPONSIBLE:
+ break;
+ }
+ frag->data = NULL;
+ }
+}
+
+#define HEADER_BUFFER_SIZE 1024*1024
+#define GROWTH_FACTOR 1.25
+
+#define BUFFER_BYTES_USED(buf) ((size_t)((buf)->tail - (buf)->data))
+#define BUFFER_BYTES_AVAILABLE(buf) ((buf)->size - BUFFER_BYTES_USED(buf))
+#define BUFFER_FIRST_FRAG(buf) ((struct buffer_fragment *)((buf)->tail_frag ? (buf)->data : NULL))
+static void buffer_purge(struct header_buffer *buf) {
+ struct buffer_fragment *frag = BUFFER_FIRST_FRAG(buf);
+ while (frag) {
+ buffer_frag_free_data(frag);
+ frag = frag->next;
+ }
+ buf->tail = buf->data;
+ buf->tail_frag = NULL;
+}
+
+#define FRAG_PADDING(addr) ((MQTT_WSS_FRAG_MEMALIGN - ((uintptr_t)addr % MQTT_WSS_FRAG_MEMALIGN)) % MQTT_WSS_FRAG_MEMALIGN)
+static struct buffer_fragment *buffer_new_frag(struct header_buffer *buf, buffer_frag_flag_t flags)
+{
+ uint8_t padding = FRAG_PADDING(buf->tail);
+
+ if (BUFFER_BYTES_AVAILABLE(buf) < sizeof(struct buffer_fragment) + padding)
+ return NULL;
+
+ struct buffer_fragment *frag = (struct buffer_fragment *)(buf->tail + padding);
+
+ memset(frag, 0, sizeof(*frag));
+ buf->tail += sizeof(*frag) + padding;
+
+ if (/*!((frag)->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) &&*/ buf->tail_frag)
+ buf->tail_frag->next = frag;
+
+ buf->tail_frag = frag;
+
+ frag->data = buf->tail;
+
+ frag->flags = flags;
+
+ return frag;
+}
+
+static void buffer_rebuild(struct header_buffer *buf)
+{
+ struct buffer_fragment *frag = (struct buffer_fragment*)buf->data;
+ do {
+ buf->tail = (char*)frag + sizeof(struct buffer_fragment);
+ buf->tail_frag = frag;
+ if (!(frag->flags & BUFFER_FRAG_DATA_EXTERNAL)) {
+ buf->tail_frag->data = buf->tail;
+ buf->tail += frag->len;
+ }
+ if (frag->next != NULL)
+ frag->next = (struct buffer_fragment*)(buf->tail + FRAG_PADDING(buf->tail));
+ frag = frag->next;
+ } while(frag);
+}
+
+static void buffer_garbage_collect(struct header_buffer *buf, mqtt_wss_log_ctx_t log_ctx)
+{
+#if !defined(MQTT_DEBUG_VERBOSE) && !defined(ADDITIONAL_CHECKS)
+ (void) log_ctx;
+#endif
+#ifdef MQTT_DEBUG_VERBOSE
+ mws_debug(log_ctx, "Buffer Garbage Collection!");
+#endif
+
+ struct buffer_fragment *frag = BUFFER_FIRST_FRAG(buf);
+ while (frag) {
+ if (!frag_is_marked_for_gc(frag))
+ break;
+
+ buffer_frag_free_data(frag);
+
+ frag = frag->next;
+ }
+
+ if (frag == BUFFER_FIRST_FRAG(buf)) {
+#ifdef MQTT_DEBUG_VERBOSE
+ mws_debug(log_ctx, "Buffer Garbage Collection! No Space Reclaimed!");
+#endif
+ return;
+ }
+
+ if (!frag) {
+ buf->tail_frag = NULL;
+ buf->tail = buf->data;
+ return;
+ }
+
+#ifdef ADDITIONAL_CHECKS
+ if (!(frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD)) {
+ mws_error(log_ctx, "Expected to find end of buffer (NULL) or next packet head!");
+ return;
+ }
+#endif
+
+ memmove(buf->data, frag, buf->tail - (char*)frag);
+ buffer_rebuild(buf);
+}
+
+static void transaction_buffer_garbage_collect(struct transaction_buffer *buf, mqtt_wss_log_ctx_t log_ctx)
+{
+#ifdef MQTT_DEBUG_VERBOSE
+ mws_debug(log_ctx, "Transaction Buffer Garbage Collection! %s", buf->sending_frag == NULL ? "NULL" : "in flight message");
+#endif
+
+ // Invalidate the cached sending fragment
+ // as we will move data around
+ if (buf->sending_frag != &ping_frag)
+ buf->sending_frag = NULL;
+
+ buffer_garbage_collect(&buf->hdr_buffer, log_ctx);
+}
+
+static int transaction_buffer_grow(struct transaction_buffer *buf, mqtt_wss_log_ctx_t log_ctx, float rate, size_t max)
+{
+ if (buf->hdr_buffer.size >= max)
+ return 0;
+
+ // Invalidate the cached sending fragment
+ // as we will move data around
+ if (buf->sending_frag != &ping_frag)
+ buf->sending_frag = NULL;
+
+ buf->hdr_buffer.size *= rate;
+ if (buf->hdr_buffer.size > max)
+ buf->hdr_buffer.size = max;
+
+ void *ret = mw_realloc(buf->hdr_buffer.data, buf->hdr_buffer.size);
+ if (ret == NULL) {
+ mws_warn(log_ctx, "Buffer growth failed (realloc)");
+ return 1;
+ }
+
+ mws_debug(log_ctx, "Message metadata buffer was grown");
+
+ buf->hdr_buffer.data = ret;
+ buffer_rebuild(&buf->hdr_buffer);
+ return 0;
+}
+
+inline static int transaction_buffer_init(struct transaction_buffer *to_init, size_t size)
+{
+ pthread_mutex_init(&to_init->mutex, NULL);
+
+ to_init->hdr_buffer.size = size;
+ to_init->hdr_buffer.data = mw_malloc(size);
+ if (to_init->hdr_buffer.data == NULL)
+ return 1;
+
+ to_init->hdr_buffer.tail = to_init->hdr_buffer.data;
+ to_init->hdr_buffer.tail_frag = NULL;
+ return 0;
+}
+
+static void transaction_buffer_destroy(struct transaction_buffer *to_init)
+{
+ buffer_purge(&to_init->hdr_buffer);
+ pthread_mutex_destroy(&to_init->mutex);
+ mw_free(to_init->hdr_buffer.data);
+}
+
+// Creates transaction
+// saves state of buffer before any operation was done
+// allowing for rollback if things go wrong
+#define transaction_buffer_transaction_start(buf) \
+ { LOCK_HDR_BUFFER(buf); \
+ memcpy(&(buf)->state_backup, &(buf)->hdr_buffer, sizeof((buf)->hdr_buffer)); }
+
+#define transaction_buffer_transaction_commit(buf) UNLOCK_HDR_BUFFER(buf);
+
+void transaction_buffer_transaction_rollback(struct transaction_buffer *buf, struct buffer_fragment *frag)
+{
+ memcpy(&buf->hdr_buffer, &buf->state_backup, sizeof(buf->hdr_buffer));
+ if (buf->hdr_buffer.tail_frag != NULL)
+ buf->hdr_buffer.tail_frag->next = NULL;
+
+ while(frag) {
+ buffer_frag_free_data(frag);
+ // we are not actually freeing the structure itself
+ // just the data it manages
+ // structure itself is in permanent buffer
+ // which is locked by HDR_BUFFER lock
+ frag = frag->next;
+ }
+
+ UNLOCK_HDR_BUFFER(buf);
+}
+
+#define TX_ALIASES_INITIALIZE() c_rhash_new(0)
+#define RX_ALIASES_INITIALIZE() c_rhash_new(UINT16_MAX >> 8)
+struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings)
+{
+ struct mqtt_ng_client *client = mw_calloc(1, sizeof(struct mqtt_ng_client));
+ if (client == NULL)
+ return NULL;
+
+ if (transaction_buffer_init(&client->main_buffer, HEADER_BUFFER_SIZE))
+ goto err_free_client;
+
+ client->rx_aliases = RX_ALIASES_INITIALIZE();
+ if (client->rx_aliases == NULL)
+ goto err_free_trx_buf;
+
+ if (pthread_mutex_init(&client->stats_mutex, NULL))
+ goto err_free_rx_alias;
+
+ client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
+ if (client->tx_topic_aliases.stoi_dict == NULL)
+ goto err_free_stats_mutex;
+ client->tx_topic_aliases.idx_max = UINT16_MAX;
+
+ if (pthread_rwlock_init(&client->tx_topic_aliases.rwlock, NULL))
+ goto err_free_tx_alias;
+
+ // TODO just embed the struct into mqtt_ng_client
+ client->parser.received_data = settings->data_in;
+ client->send_fnc_ptr = settings->data_out_fnc;
+ client->user_ctx = settings->user_ctx;
+
+ client->log = settings->log;
+
+ client->puback_callback = settings->puback_callback;
+ client->connack_callback = settings->connack_callback;
+ client->msg_callback = settings->msg_callback;
+
+ return client;
+
+err_free_tx_alias:
+ c_rhash_destroy(client->tx_topic_aliases.stoi_dict);
+err_free_stats_mutex:
+ pthread_mutex_destroy(&client->stats_mutex);
+err_free_rx_alias:
+ c_rhash_destroy(client->rx_aliases);
+err_free_trx_buf:
+ transaction_buffer_destroy(&client->main_buffer);
+err_free_client:
+ mw_free(client);
+ return NULL;
+}
+
+static inline uint8_t get_control_packet_type(uint8_t first_hdr_byte)
+{
+ return first_hdr_byte >> 4;
+}
+
+static void mqtt_ng_destroy_rx_alias_hash(c_rhash hash)
+{
+ c_rhash_iter_t i = C_RHASH_ITER_T_INITIALIZER;
+ uint64_t stored_key;
+ void *to_free;
+ while(!c_rhash_iter_uint64_keys(hash, &i, &stored_key)) {
+ c_rhash_get_ptr_by_uint64(hash, stored_key, &to_free);
+ mw_free(to_free);
+ }
+ c_rhash_destroy(hash);
+}
+
+static void mqtt_ng_destroy_tx_alias_hash(c_rhash hash)
+{
+ c_rhash_iter_t i = C_RHASH_ITER_T_INITIALIZER;
+ const char *stored_key;
+ void *to_free;
+ while(!c_rhash_iter_str_keys(hash, &i, &stored_key)) {
+ c_rhash_get_ptr_by_str(hash, stored_key, &to_free);
+ mw_free(to_free);
+ }
+ c_rhash_destroy(hash);
+}
+
+void mqtt_ng_destroy(struct mqtt_ng_client *client)
+{
+ transaction_buffer_destroy(&client->main_buffer);
+ pthread_mutex_destroy(&client->stats_mutex);
+
+ mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
+ pthread_rwlock_destroy(&client->tx_topic_aliases.rwlock);
+ mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
+
+ mw_free(client);
+}
+
+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)
+{
+ if (frag->len) {
+ // TODO?: This could potentially be done in future if we set rule
+ // external data always follows in buffer data
+ // could help reduce fragmentation in some messages but
+ // currently not worth it considering time is tight
+ mws_fatal(log, UNIT_LOG_PREFIX "INTERNAL ERROR: Cannot set external data to fragment already containing in buffer data!");
+ return 1;
+ }
+
+ switch (ptr2memory_mode(data_free_fnc)) {
+ case MEMCPY:
+ frag->data = mw_malloc(data_len);
+ if (frag->data == NULL) {
+ mws_error(log, UNIT_LOG_PREFIX "OOM while malloc @_optimized_add");
+ return 1;
+ }
+ memcpy(frag->data, data, data_len);
+ break;
+ case EXTERNAL_FREE_AFTER_USE:
+ case CALLER_RESPONSIBLE:
+ frag->data = data;
+ break;
+ }
+ frag->free_fnc = data_free_fnc;
+ frag->len = data_len;
+
+ frag->flags |= BUFFER_FRAG_DATA_EXTERNAL;
+ return 0;
+ }
+
+// this is fixed part of variable header for connect packet
+// mqtt-v5.0-cs1, 3.1.2.1, 2.1.2.2
+static const char mqtt_protocol_name_frag[] =
+ { 0x00, 0x04, 'M', 'Q', 'T', 'T', MQTT_VERSION_5_0 };
+
+#define MQTT_UTF8_STRING_SIZE(string) (2 + strlen(string))
+
+// see 1.5.5
+#define MQTT_VARSIZE_INT_BYTES(value) ( value > 2097152 ? 4 : ( value > 16384 ? 3 : ( value > 128 ? 2 : 1 ) ) )
+
+static size_t mqtt_ng_connect_size(struct mqtt_auth_properties *auth,
+ struct mqtt_lwt_properties *lwt)
+{
+ // First get the size of payload + variable header
+ size_t size =
+ + sizeof(mqtt_protocol_name_frag) /* Proto Name and Version */
+ + 1 /* Connect Flags */
+ + 2 /* Keep Alive */
+ + 4 /* 3.1.2.11.1 Property Length - for now fixed to only Topic Alias Maximum, TODO TODO*/;
+
+ // CONNECT payload. 3.1.3
+ if (auth->client_id)
+ size += MQTT_UTF8_STRING_SIZE(auth->client_id);
+
+ if (lwt) {
+ // 3.1.3.2 will properties TODO TODO
+ size += 1;
+
+ // 3.1.3.3
+ if (lwt->will_topic)
+ size += MQTT_UTF8_STRING_SIZE(lwt->will_topic);
+
+ // 3.1.3.4 will payload
+ if (lwt->will_message) {
+ size += 2 + lwt->will_message_size;
+ }
+ }
+
+ // 3.1.3.5
+ if (auth->username)
+ size += MQTT_UTF8_STRING_SIZE(auth->username);
+
+ // 3.1.3.6
+ if (auth->password)
+ size += MQTT_UTF8_STRING_SIZE(auth->password);
+
+ return size;
+}
+
+#define BUFFER_TRANSACTION_NEW_FRAG(buf, flags, frag, on_fail) \
+ { if(frag==NULL) { \
+ frag = buffer_new_frag(buf, (flags)); } \
+ if(frag==NULL) { on_fail; }}
+
+#define CHECK_BYTES_AVAILABLE(buf, needed, fail) \
+ { if (BUFFER_BYTES_AVAILABLE(buf) < (size_t)needed) { \
+ fail; } }
+
+#define DATA_ADVANCE(buf, bytes, frag) { size_t b = (bytes); (buf)->tail += b; (frag)->len += b; }
+
+// TODO maybe just user client->buf.tail?
+#define WRITE_POS(frag) (&(frag->data[frag->len]))
+
+// [MQTT-1.5.2] Two Byte Integer
+#define PACK_2B_INT(buffer, integer, frag) { *(uint16_t *)WRITE_POS(frag) = htobe16((integer)); \
+ DATA_ADVANCE(buffer, sizeof(uint16_t), frag); }
+
+static int _optimized_add(struct header_buffer *buf, mqtt_wss_log_ctx_t log_ctx, void *data, size_t data_len, free_fnc_t data_free_fnc, struct buffer_fragment **frag)
+{
+ if (data_len > SMALL_STRING_DONT_FRAGMENT_LIMIT) {
+ buffer_frag_flag_t flags = BUFFER_FRAG_DATA_EXTERNAL;
+ if ((*frag)->flags & BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND)
+ flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
+ if( (*frag = buffer_new_frag(buf, flags)) == NULL ) {
+ mws_error(log_ctx, "Out of buffer space while generating the message");
+ return 1;
+ }
+ if (frag_set_external_data(log_ctx, *frag, data, data_len, data_free_fnc)) {
+ mws_error(log_ctx, "Error adding external data to newly created fragment");
+ return 1;
+ }
+ // we dont want to write to this fragment anymore
+ *frag = NULL;
+ } else if (data_len) {
+ // if the data are small dont bother creating new fragments
+ // store in buffer directly
+ CHECK_BYTES_AVAILABLE(buf, data_len, return 1);
+ memcpy(buf->tail, data, data_len);
+ DATA_ADVANCE(buf, data_len, *frag);
+ }
+ return 0;
+}
+
+#define TRY_GENERATE_MESSAGE(generator_function, client, ...) \
+ int rc = generator_function(&client->main_buffer, client->log, ##__VA_ARGS__); \
+ if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) { \
+ LOCK_HDR_BUFFER(&client->main_buffer); \
+ transaction_buffer_garbage_collect((&client->main_buffer), client->log); \
+ UNLOCK_HDR_BUFFER(&client->main_buffer); \
+ rc = generator_function(&client->main_buffer, client->log, ##__VA_ARGS__); \
+ if (rc == MQTT_NG_MSGGEN_BUFFER_OOM && client->max_mem_bytes) { \
+ LOCK_HDR_BUFFER(&client->main_buffer); \
+ transaction_buffer_grow((&client->main_buffer), client->log, GROWTH_FACTOR, client->max_mem_bytes); \
+ UNLOCK_HDR_BUFFER(&client->main_buffer); \
+ rc = generator_function(&client->main_buffer, client->log, ##__VA_ARGS__); \
+ } \
+ if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) \
+ mws_error(client->log, "%s failed to generate message due to insufficient buffer space (line %d)", __FUNCTION__, __LINE__); \
+ } \
+ if (rc == MQTT_NG_MSGGEN_OK) { \
+ pthread_mutex_lock(&client->stats_mutex); \
+ client->stats.tx_messages_queued++; \
+ pthread_mutex_unlock(&client->stats_mutex); \
+ } \
+ return rc;
+
+mqtt_msg_data mqtt_ng_generate_connect(struct transaction_buffer *trx_buf,
+ mqtt_wss_log_ctx_t log_ctx,
+ struct mqtt_auth_properties *auth,
+ struct mqtt_lwt_properties *lwt,
+ uint8_t clean_start,
+ uint16_t keep_alive)
+{
+ // Sanity Checks First (are given parameters correct and up to MQTT spec)
+ if (!auth->client_id) {
+ mws_error(log_ctx, "ClientID must be set. [MQTT-3.1.3-3]");
+ return NULL;
+ }
+
+ size_t len = strlen(auth->client_id);
+ if (!len) {
+ // [MQTT-3.1.3-6] server MAY allow empty client_id and treat it
+ // as specific client_id (not same as client_id not given)
+ // however server MUST allow ClientIDs between 1-23 bytes [MQTT-3.1.3-5]
+ // so we will warn client server might not like this and he is using it
+ // at his own risk!
+ mws_warn(log_ctx, "client_id provided is empty string. This might not be allowed by server [MQTT-3.1.3-6]");
+ }
+ if(len > MQTT_MAX_CLIENT_ID) {
+ // [MQTT-3.1.3-5] server MUST allow client_id length 1-32
+ // server MAY allow longer client_id, if user provides longer client_id
+ // warn them he is doing so at his own risk!
+ mws_warn(log_ctx, "client_id provided is longer than 23 bytes, server might not allow that [MQTT-3.1.3-5]");
+ }
+
+ if (lwt) {
+ if (lwt->will_message && lwt->will_message_size > 65535) {
+ mws_error(log_ctx, "Will message cannot be longer than 65535 bytes due to MQTT protocol limitations [MQTT-3.1.3-4] and [MQTT-1.5.6]");
+ return NULL;
+ }
+
+ if (!lwt->will_topic) { //TODO topic given with strlen==0 ? check specs
+ mws_error(log_ctx, "If will message is given will topic must also be given [MQTT-3.1.3.3]");
+ return NULL;
+ }
+
+ if (lwt->will_qos > MQTT_MAX_QOS) {
+ // refer to [MQTT-3-1.2-12]
+ mws_error(log_ctx, "QOS for LWT message is bigger than max");
+ return NULL;
+ }
+ }
+
+ // >> START THE RODEO <<
+ transaction_buffer_transaction_start(trx_buf);
+
+ // Calculate the resulting message size sans fixed MQTT header
+ size_t size = mqtt_ng_connect_size(auth, lwt);
+
+ // Start generating the message
+ struct buffer_fragment *frag = NULL;
+ mqtt_msg_data ret = NULL;
+
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback );
+ ret = frag;
+
+ // MQTT Fixed Header
+ size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + sizeof(mqtt_protocol_name_frag) + 1 /* CONNECT FLAGS */ + 2 /* keepalive */ + 1 /* Properties TODO now fixed 0*/;
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+ *WRITE_POS(frag) = MQTT_CPT_CONNECT << 4;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+ memcpy(WRITE_POS(frag), mqtt_protocol_name_frag, sizeof(mqtt_protocol_name_frag));
+ DATA_ADVANCE(&trx_buf->hdr_buffer, sizeof(mqtt_protocol_name_frag), frag);
+
+ // [MQTT-3.1.2.3] Connect flags
+ char *connect_flags = WRITE_POS(frag);
+ *connect_flags = 0;
+ if (auth->username)
+ *connect_flags |= MQTT_CONNECT_FLAG_USERNAME;
+ if (auth->password)
+ *connect_flags |= MQTT_CONNECT_FLAG_PASSWORD;
+ if (lwt) {
+ *connect_flags |= MQTT_CONNECT_FLAG_LWT;
+ *connect_flags |= lwt->will_qos << MQTT_CONNECT_FLAG_QOS_BITSHIFT;
+ if (lwt->will_retain)
+ *connect_flags |= MQTT_CONNECT_FLAG_LWT_RETAIN;
+ }
+ if (clean_start)
+ *connect_flags |= MQTT_CONNECT_FLAG_CLEAN_START;
+
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+ PACK_2B_INT(&trx_buf->hdr_buffer, keep_alive, frag);
+
+ // TODO Property Length [MQTT-3.1.3.2.1] temporary fixed to 3 (one property topic alias max)
+ DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(3, WRITE_POS(frag)), frag);
+ *WRITE_POS(frag) = MQTT_PROP_TOPIC_ALIAS_MAX;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+ PACK_2B_INT(&trx_buf->hdr_buffer, 65535, frag);
+
+ // [MQTT-3.1.3.1] Client identifier
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+ PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->client_id), frag);
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, auth->client_id, strlen(auth->client_id), auth->client_id_free, &frag))
+ goto fail_rollback;
+
+ if (lwt != NULL) {
+ // Will Properties [MQTT-3.1.3.2]
+ // TODO for now fixed 0
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 1, goto fail_rollback);
+ *WRITE_POS(frag) = 0;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+ // Will Topic [MQTT-3.1.3.3]
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+ PACK_2B_INT(&trx_buf->hdr_buffer, strlen(lwt->will_topic), frag);
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, lwt->will_topic, strlen(lwt->will_topic), lwt->will_topic_free, &frag))
+ goto fail_rollback;
+
+ // Will Payload [MQTT-3.1.3.4]
+ if (lwt->will_message_size) {
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+ PACK_2B_INT(&trx_buf->hdr_buffer, lwt->will_message_size, frag);
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, lwt->will_message, lwt->will_message_size, lwt->will_topic_free, &frag))
+ goto fail_rollback;
+ }
+ }
+
+ // [MQTT-3.1.3.5]
+ if (auth->username) {
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+ PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->username), frag);
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, auth->username, strlen(auth->username), auth->username_free, &frag))
+ goto fail_rollback;
+ }
+
+ // [MQTT-3.1.3.6]
+ if (auth->password) {
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+ PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->password), frag);
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, auth->password, strlen(auth->password), auth->password_free, &frag))
+ goto fail_rollback;
+ }
+ trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+ transaction_buffer_transaction_commit(trx_buf);
+ return ret;
+fail_rollback:
+ transaction_buffer_transaction_rollback(trx_buf, ret);
+ return NULL;
+}
+
+int mqtt_ng_connect(struct mqtt_ng_client *client,
+ struct mqtt_auth_properties *auth,
+ struct mqtt_lwt_properties *lwt,
+ uint8_t clean_start,
+ uint16_t keep_alive)
+{
+ client->client_state = RAW;
+ client->parser.state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
+
+ LOCK_HDR_BUFFER(&client->main_buffer);
+ client->main_buffer.sending_frag = NULL;
+ if (clean_start)
+ buffer_purge(&client->main_buffer.hdr_buffer);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+
+ pthread_rwlock_wrlock(&client->tx_topic_aliases.rwlock);
+ // according to MQTT spec topic aliases should not be persisted
+ // even if clean session is true
+ mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
+ client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
+ if (client->tx_topic_aliases.stoi_dict == NULL) {
+ pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+ return 1;
+ }
+ client->tx_topic_aliases.idx_assigned = 0;
+ pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+
+ mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
+ client->rx_aliases = RX_ALIASES_INITIALIZE();
+ if (client->rx_aliases == NULL)
+ return 1;
+
+ client->connect_msg = mqtt_ng_generate_connect(&client->main_buffer, client->log, auth, lwt, clean_start, keep_alive);
+ if (client->connect_msg == NULL)
+ return 1;
+
+ pthread_mutex_lock(&client->stats_mutex);
+ if (clean_start)
+ client->stats.tx_messages_queued = 1;
+ else
+ client->stats.tx_messages_queued++;
+
+ client->stats.tx_messages_sent = 0;
+ client->stats.rx_messages_rcvd = 0;
+ pthread_mutex_unlock(&client->stats_mutex);
+
+ client->client_state = CONNECT_PENDING;
+ return 0;
+}
+
+uint16_t get_unused_packet_id() {
+ static uint16_t packet_id = 0;
+ packet_id++;
+ return packet_id ? packet_id : ++packet_id;
+}
+
+static inline size_t mqtt_ng_publish_size(const char *topic,
+ size_t msg_len,
+ uint16_t topic_id)
+{
+ size_t retval = 2 /* Topic Name Length */
+ + (topic == NULL ? 0 : strlen(topic))
+ + 2 /* Packet identifier */
+ + 1 /* Properties Length TODO for now fixed to 1 property */
+ + msg_len;
+
+ if (topic_id)
+ retval += 3;
+
+ return retval;
+}
+
+int mqtt_ng_generate_publish(struct transaction_buffer *trx_buf,
+ mqtt_wss_log_ctx_t log_ctx,
+ char *topic,
+ free_fnc_t topic_free,
+ void *msg,
+ free_fnc_t msg_free,
+ size_t msg_len,
+ uint8_t publish_flags,
+ uint16_t *packet_id,
+ uint16_t topic_alias)
+{
+ // >> START THE RODEO <<
+ transaction_buffer_transaction_start(trx_buf);
+
+ // Calculate the resulting message size sans fixed MQTT header
+ size_t size = mqtt_ng_publish_size(topic, msg_len, topic_alias);
+
+ // Start generating the message
+ struct buffer_fragment *frag = NULL;
+ mqtt_msg_data mqtt_msg = NULL;
+
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback );
+ // in case of QOS 0 we can garbage collect immediatelly after sending
+ uint8_t qos = (publish_flags >> 1) & 0x03;
+ if (!qos)
+ frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
+ mqtt_msg = frag;
+
+ // MQTT Fixed Header
+ size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + size - msg_len;
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+ *WRITE_POS(frag) = (MQTT_CPT_PUBLISH << 4) | (publish_flags & 0xF);
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+ // MQTT Variable Header
+ // [MQTT-3.3.2.1]
+ PACK_2B_INT(&trx_buf->hdr_buffer, topic == NULL ? 0 : strlen(topic), frag);
+ if (topic != NULL) {
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, topic, strlen(topic), topic_free, &frag))
+ goto fail_rollback;
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ }
+
+ // [MQTT-3.3.2.2]
+ mqtt_msg->packet_id = get_unused_packet_id();
+ *packet_id = mqtt_msg->packet_id;
+ PACK_2B_INT(&trx_buf->hdr_buffer, mqtt_msg->packet_id, frag);
+
+ // [MQTT-3.3.2.3.1] TODO Property Length for now fixed 0
+ *WRITE_POS(frag) = topic_alias ? 3 : 0;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+ if(topic_alias) {
+ *WRITE_POS(frag) = MQTT_PROP_TOPIC_ALIAS;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+ PACK_2B_INT(&trx_buf->hdr_buffer, topic_alias, frag);
+ }
+
+ if( (frag = buffer_new_frag(&trx_buf->hdr_buffer, BUFFER_FRAG_DATA_EXTERNAL)) == NULL )
+ goto fail_rollback;
+
+ if (frag_set_external_data(log_ctx, frag, msg, msg_len, msg_free))
+ goto fail_rollback;
+
+ trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+ if (!qos)
+ trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
+ transaction_buffer_transaction_commit(trx_buf);
+ return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+ transaction_buffer_transaction_rollback(trx_buf, mqtt_msg);
+ return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+#define PUBLISH_SP_SIZE 64
+int mqtt_ng_publish(struct mqtt_ng_client *client,
+ char *topic,
+ free_fnc_t topic_free,
+ void *msg,
+ free_fnc_t msg_free,
+ size_t msg_len,
+ uint8_t publish_flags,
+ uint16_t *packet_id)
+{
+ struct topic_alias_data *alias = NULL;
+ pthread_rwlock_rdlock(&client->tx_topic_aliases.rwlock);
+ c_rhash_get_ptr_by_str(client->tx_topic_aliases.stoi_dict, topic, (void**)&alias);
+ pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+
+ uint16_t topic_id = 0;
+
+ if (alias != NULL) {
+ topic_id = alias->idx;
+ uint32_t cnt = __atomic_fetch_add(&alias->usage_count, 1, __ATOMIC_SEQ_CST);
+ if (cnt) {
+ topic = NULL;
+ topic_free = NULL;
+ }
+ }
+
+ if (client->max_msg_size && PUBLISH_SP_SIZE + mqtt_ng_publish_size(topic, msg_len, topic_id) > client->max_msg_size) {
+ mws_error(client->log, "Message too big for server: %zu", msg_len);
+ return MQTT_NG_MSGGEN_MSG_TOO_BIG;
+ }
+
+ TRY_GENERATE_MESSAGE(mqtt_ng_generate_publish, client, topic, topic_free, msg, msg_free, msg_len, publish_flags, packet_id, topic_id);
+}
+
+static inline size_t mqtt_ng_subscribe_size(struct mqtt_sub *subs, size_t sub_count)
+{
+ size_t len = 2 /* Packet Identifier */ + 1 /* Properties Length TODO for now fixed 0 */;
+ len += sub_count * (2 /* topic filter string length */ + 1 /* [MQTT-3.8.3.1] Subscription Options Byte */);
+
+ for (size_t i = 0; i < sub_count; i++) {
+ len += strlen(subs[i].topic);
+ }
+ return len;
+}
+
+int mqtt_ng_generate_subscribe(struct transaction_buffer *trx_buf, mqtt_wss_log_ctx_t log_ctx, struct mqtt_sub *subs, size_t sub_count)
+{
+ // >> START THE RODEO <<
+ transaction_buffer_transaction_start(trx_buf);
+
+ // Calculate the resulting message size sans fixed MQTT header
+ size_t size = mqtt_ng_subscribe_size(subs, sub_count);
+
+ // Start generating the message
+ struct buffer_fragment *frag = NULL;
+ mqtt_msg_data ret = NULL;
+
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback);
+ ret = frag;
+
+ // MQTT Fixed Header
+ size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + 3 /*Packet ID + Property Length*/;
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+ *WRITE_POS(frag) = (MQTT_CPT_SUBSCRIBE << 4) | 0x2 /* [MQTT-3.8.1-1] */;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+ // MQTT Variable Header
+ // [MQTT-3.8.2] PacketID
+ ret->packet_id = get_unused_packet_id();
+ PACK_2B_INT(&trx_buf->hdr_buffer, ret->packet_id, frag);
+
+ // [MQTT-3.8.2.1.1] Property Length // TODO for now fixed 0
+ *WRITE_POS(frag) = 0;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+ for (size_t i = 0; i < sub_count; i++) {
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ PACK_2B_INT(&trx_buf->hdr_buffer, strlen(subs[i].topic), frag);
+ if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, subs[i].topic, strlen(subs[i].topic), subs[i].topic_free, &frag))
+ goto fail_rollback;
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+ *WRITE_POS(frag) = subs[i].options;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ }
+
+ trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+ transaction_buffer_transaction_commit(trx_buf);
+ return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+ transaction_buffer_transaction_rollback(trx_buf, ret);
+ return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+int mqtt_ng_subscribe(struct mqtt_ng_client *client, struct mqtt_sub *subs, size_t sub_count)
+{
+ TRY_GENERATE_MESSAGE(mqtt_ng_generate_subscribe, client, subs, sub_count);
+}
+
+int mqtt_ng_generate_disconnect(struct transaction_buffer *trx_buf, mqtt_wss_log_ctx_t log_ctx, uint8_t reason_code)
+{
+ (void) log_ctx;
+ // >> START THE RODEO <<
+ transaction_buffer_transaction_start(trx_buf);
+
+ // Calculate the resulting message size sans fixed MQTT header
+ size_t size = reason_code ? 1 : 0;
+
+ // Start generating the message
+ struct buffer_fragment *frag = NULL;
+ mqtt_msg_data ret = NULL;
+
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback);
+ ret = frag;
+
+ // MQTT Fixed Header
+ size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + (reason_code ? 1 : 0);
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+ *WRITE_POS(frag) = MQTT_CPT_DISCONNECT << 4;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+ if (reason_code) {
+ // MQTT Variable Header
+ // [MQTT-3.14.2.1] PacketID
+ *WRITE_POS(frag) = reason_code;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ }
+
+ trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+ transaction_buffer_transaction_commit(trx_buf);
+ return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+ transaction_buffer_transaction_rollback(trx_buf, ret);
+ return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+int mqtt_ng_disconnect(struct mqtt_ng_client *client, uint8_t reason_code)
+{
+ TRY_GENERATE_MESSAGE(mqtt_ng_generate_disconnect, client, reason_code);
+}
+
+static int mqtt_generate_puback(struct transaction_buffer *trx_buf, mqtt_wss_log_ctx_t log_ctx, uint16_t packet_id, uint8_t reason_code)
+{
+ (void) log_ctx;
+ // >> START THE RODEO <<
+ transaction_buffer_transaction_start(trx_buf);
+
+ // Calculate the resulting message size sans fixed MQTT header
+ size_t size = 2 /* Packet ID */ + (reason_code ? 1 : 0) /* reason code */;
+
+ // Start generating the message
+ struct buffer_fragment *frag = NULL;
+
+ BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD | BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND, frag, goto fail_rollback);
+
+ // MQTT Fixed Header
+ size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + size;
+ CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+ *WRITE_POS(frag) = MQTT_CPT_PUBACK << 4;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+ // MQTT Variable Header
+ PACK_2B_INT(&trx_buf->hdr_buffer, packet_id, frag);
+
+ if (reason_code) {
+ // MQTT Variable Header
+ // [MQTT-3.14.2.1] PacketID
+ *WRITE_POS(frag) = reason_code;
+ DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+ }
+
+ trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+ transaction_buffer_transaction_commit(trx_buf);
+ return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+ transaction_buffer_transaction_rollback(trx_buf, frag);
+ return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+static int mqtt_ng_puback(struct mqtt_ng_client *client, uint16_t packet_id, uint8_t reason_code)
+{
+ TRY_GENERATE_MESSAGE(mqtt_generate_puback, client, packet_id, reason_code);
+}
+
+int mqtt_ng_ping(struct mqtt_ng_client *client)
+{
+ client->ping_pending = 1;
+ return MQTT_NG_MSGGEN_OK;
+}
+
+#define MQTT_NG_CLIENT_NEED_MORE_BYTES 0x10
+#define MQTT_NG_CLIENT_MQTT_PACKET_DONE 0x11
+#define MQTT_NG_CLIENT_PARSE_DONE 0x12
+#define MQTT_NG_CLIENT_WANT_WRITE 0x13
+#define MQTT_NG_CLIENT_OK_CALL_AGAIN 0
+#define MQTT_NG_CLIENT_PROTOCOL_ERROR -1
+#define MQTT_NG_CLIENT_SERVER_RETURNED_ERROR -2
+#define MQTT_NG_CLIENT_NOT_IMPL_YET -3
+#define MQTT_NG_CLIENT_OOM -4
+#define MQTT_NG_CLIENT_INTERNAL_ERROR -5
+
+#define BUF_READ_CHECK_AT_LEAST(buf, x) \
+ if (rbuf_bytes_available(buf) < (x)) \
+ return MQTT_NG_CLIENT_NEED_MORE_BYTES;
+
+#define vbi_parser_reset_ctx(ctx) memset(ctx, 0, sizeof(struct mqtt_vbi_parser_ctx))
+
+static int vbi_parser_parse(struct mqtt_vbi_parser_ctx *ctx, rbuf_t data, mqtt_wss_log_ctx_t log)
+{
+ if (ctx->bytes > MQTT_VBI_MAXBYTES - 1) {
+ mws_error(log, "MQTT Variable Byte Integer can't be longer than %d bytes", MQTT_VBI_MAXBYTES);
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+ if (!ctx->bytes || ctx->data[ctx->bytes-1] & MQTT_VBI_CONTINUATION_FLAG) {
+ BUF_READ_CHECK_AT_LEAST(data, 1);
+ ctx->bytes++;
+ rbuf_pop(data, &ctx->data[ctx->bytes-1], 1);
+ if ( ctx->data[ctx->bytes-1] & MQTT_VBI_CONTINUATION_FLAG )
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+ }
+
+ if (mqtt_vbi_to_uint32(ctx->data, &ctx->result)) {
+ mws_error(log, "MQTT Variable Byte Integer failed to be parsed.");
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+
+ return MQTT_NG_CLIENT_PARSE_DONE;
+}
+
+static void mqtt_properties_parser_ctx_reset(struct mqtt_properties_parser_ctx *ctx)
+{
+ ctx->state = PROPERTIES_LENGTH;
+ while (ctx->head) {
+ struct mqtt_property *f = ctx->head;
+ ctx->head = ctx->head->next;
+ if (f->type == MQTT_TYPE_STR || f->type == MQTT_TYPE_STR_PAIR)
+ mw_free(f->data.strings[0]);
+ if (f->type == MQTT_TYPE_STR_PAIR)
+ mw_free(f->data.strings[1]);
+ if (f->type == MQTT_TYPE_BIN)
+ mw_free(f->data.bindata);
+ mw_free(f);
+ }
+ ctx->tail = NULL;
+ ctx->properties_length = 0;
+ ctx->bytes_consumed = 0;
+ vbi_parser_reset_ctx(&ctx->vbi_parser_ctx);
+}
+
+struct mqtt_property_type {
+ uint8_t id;
+ enum mqtt_datatype datatype;
+ const char* name;
+};
+
+const struct mqtt_property_type mqtt_property_types[] = {
+ { .id = MQTT_PROP_TOPIC_ALIAS, .name = MQTT_PROP_TOPIC_ALIAS_NAME, .datatype = MQTT_TYPE_UINT_16 },
+
+ { .id = MQTT_PROP_PAYLOAD_FMT_INDICATOR, .name = MQTT_PROP_PAYLOAD_FMT_INDICATOR_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_MSG_EXPIRY_INTERVAL, .name = MQTT_PROP_MSG_EXPIRY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32 },
+ { .id = MQTT_PROP_CONTENT_TYPE, .name = MQTT_PROP_CONTENT_TYPE_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_RESPONSE_TOPIC, .name = MQTT_PROP_RESPONSE_TOPIC_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_CORRELATION_DATA, .name = MQTT_PROP_CORRELATION_DATA_NAME, .datatype = MQTT_TYPE_BIN },
+ { .id = MQTT_PROP_SUB_IDENTIFIER, .name = MQTT_PROP_SUB_IDENTIFIER_NAME, .datatype = MQTT_TYPE_VBI },
+ { .id = MQTT_PROP_SESSION_EXPIRY_INTERVAL, .name = MQTT_PROP_SESSION_EXPIRY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32 },
+ { .id = MQTT_PROP_ASSIGNED_CLIENT_ID, .name = MQTT_PROP_ASSIGNED_CLIENT_ID_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_SERVER_KEEP_ALIVE, .name = MQTT_PROP_SERVER_KEEP_ALIVE_NAME, .datatype = MQTT_TYPE_UINT_16 },
+ { .id = MQTT_PROP_AUTH_METHOD, .name = MQTT_PROP_AUTH_METHOD_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_AUTH_DATA, .name = MQTT_PROP_AUTH_DATA_NAME, .datatype = MQTT_TYPE_BIN },
+ { .id = MQTT_PROP_REQ_PROBLEM_INFO, .name = MQTT_PROP_REQ_PROBLEM_INFO_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_WILL_DELAY_INTERVAL, .name = MQTT_PROP_WIIL_DELAY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32 },
+ { .id = MQTT_PROP_REQ_RESP_INFORMATION, .name = MQTT_PROP_REQ_RESP_INFORMATION_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_RESP_INFORMATION, .name = MQTT_PROP_RESP_INFORMATION_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_SERVER_REF, .name = MQTT_PROP_SERVER_REF_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_REASON_STR, .name = MQTT_PROP_REASON_STR_NAME, .datatype = MQTT_TYPE_STR },
+ { .id = MQTT_PROP_RECEIVE_MAX, .name = MQTT_PROP_RECEIVE_MAX_NAME, .datatype = MQTT_TYPE_UINT_16 },
+ { .id = MQTT_PROP_TOPIC_ALIAS_MAX, .name = MQTT_PROP_TOPIC_ALIAS_MAX_NAME, .datatype = MQTT_TYPE_UINT_16 },
+ // MQTT_PROP_TOPIC_ALIAS is first as it is most often used
+ { .id = MQTT_PROP_MAX_QOS, .name = MQTT_PROP_MAX_QOS_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_RETAIN_AVAIL, .name = MQTT_PROP_RETAIN_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_USR, .name = MQTT_PROP_USR_NAME, .datatype = MQTT_TYPE_STR_PAIR },
+ { .id = MQTT_PROP_MAX_PKT_SIZE, .name = MQTT_PROP_MAX_PKT_SIZE_NAME, .datatype = MQTT_TYPE_UINT_32 },
+ { .id = MQTT_PROP_WILDCARD_SUB_AVAIL, .name = MQTT_PROP_WILDCARD_SUB_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_SUB_ID_AVAIL, .name = MQTT_PROP_SUB_ID_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = MQTT_PROP_SHARED_SUB_AVAIL, .name = MQTT_PROP_SHARED_SUB_AVAIL_NAME, .datatype = MQTT_TYPE_UINT_8 },
+ { .id = 0, .name = NULL, .datatype = MQTT_TYPE_UNKNOWN }
+};
+
+static int get_property_type_by_id(uint8_t property_id) {
+ for (int i = 0; mqtt_property_types[i].datatype != MQTT_TYPE_UNKNOWN; i++) {
+ if (mqtt_property_types[i].id == property_id)
+ return mqtt_property_types[i].datatype;
+ }
+ return MQTT_TYPE_UNKNOWN;
+}
+
+struct mqtt_property *get_property_by_id(struct mqtt_property *props, uint8_t property_id)
+{
+ while (props) {
+ if (props->id == property_id) {
+ return props;
+ }
+ props = props->next;
+ }
+ return NULL;
+}
+
+// Parses [MQTT-2.2.2]
+static int parse_properties_array(struct mqtt_properties_parser_ctx *ctx, rbuf_t data, mqtt_wss_log_ctx_t log)
+{
+ int rc;
+ switch (ctx->state) {
+ case PROPERTIES_LENGTH:
+ rc = vbi_parser_parse(&ctx->vbi_parser_ctx, data, log);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ ctx->properties_length = ctx->vbi_parser_ctx.result;
+ ctx->bytes_consumed += ctx->vbi_parser_ctx.bytes;
+ ctx->vbi_length = ctx->vbi_parser_ctx.bytes;
+ if (!ctx->properties_length)
+ return MQTT_NG_CLIENT_PARSE_DONE;
+ ctx->state = PROPERTY_CREATE;
+ break;
+ }
+ return rc;
+ case PROPERTY_CREATE:
+ BUF_READ_CHECK_AT_LEAST(data, 1);
+ struct mqtt_property *prop = mw_calloc(1, sizeof(struct mqtt_property));
+ if (ctx->head == NULL) {
+ ctx->head = prop;
+ ctx->tail = prop;
+ } else {
+ ctx->tail->next = prop;
+ ctx->tail = ctx->tail->next;
+ }
+ ctx->state = PROPERTY_ID;
+ /* FALLTHROUGH */
+ case PROPERTY_ID:
+ rbuf_pop(data, (char*)&ctx->tail->id, 1);
+ ctx->bytes_consumed += 1;
+ ctx->tail->type = get_property_type_by_id(ctx->tail->id);
+ switch (ctx->tail->type) {
+ case MQTT_TYPE_UINT_16:
+ ctx->state = PROPERTY_TYPE_UINT16;
+ break;
+ case MQTT_TYPE_UINT_32:
+ ctx->state = PROPERTY_TYPE_UINT32;
+ break;
+ case MQTT_TYPE_UINT_8:
+ ctx->state = PROPERTY_TYPE_UINT8;
+ break;
+ case MQTT_TYPE_VBI:
+ ctx->state = PROPERTY_TYPE_VBI;
+ vbi_parser_reset_ctx(&ctx->vbi_parser_ctx);
+ break;
+ case MQTT_TYPE_STR:
+ case MQTT_TYPE_STR_PAIR:
+ ctx->str_idx = 0;
+ /* FALLTHROUGH */
+ case MQTT_TYPE_BIN:
+ ctx->state = PROPERTY_TYPE_STR_BIN_LEN;
+ break;
+ default:
+ mws_error(log, "Unsupported property type %d for property id %d.", (int)ctx->tail->type, (int)ctx->tail->id);
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+ break;
+ case PROPERTY_TYPE_STR_BIN_LEN:
+ BUF_READ_CHECK_AT_LEAST(data, sizeof(uint16_t));
+ rbuf_pop(data, (char*)&ctx->tail->bindata_len, sizeof(uint16_t));
+ ctx->tail->bindata_len = be16toh(ctx->tail->bindata_len);
+ ctx->bytes_consumed += 2;
+ switch (ctx->tail->type) {
+ case MQTT_TYPE_BIN:
+ ctx->state = PROPERTY_TYPE_BIN;
+ break;
+ case MQTT_TYPE_STR:
+ case MQTT_TYPE_STR_PAIR:
+ ctx->state = PROPERTY_TYPE_STR;
+ break;
+ default:
+ mws_error(log, "Unexpected datatype in PROPERTY_TYPE_STR_BIN_LEN %d", (int)ctx->tail->type);
+ return MQTT_NG_CLIENT_INTERNAL_ERROR;
+ }
+ break;
+ case PROPERTY_TYPE_STR:
+ BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len);
+ ctx->tail->data.strings[ctx->str_idx] = mw_malloc(ctx->tail->bindata_len + 1);
+ rbuf_pop(data, ctx->tail->data.strings[ctx->str_idx], ctx->tail->bindata_len);
+ ctx->tail->data.strings[ctx->str_idx][ctx->tail->bindata_len] = 0;
+ ctx->str_idx++;
+ ctx->bytes_consumed += ctx->tail->bindata_len;
+ if (ctx->tail->type == MQTT_TYPE_STR_PAIR && ctx->str_idx < 2) {
+ ctx->state = PROPERTY_TYPE_STR_BIN_LEN;
+ break;
+ }
+ ctx->state = PROPERTY_NEXT;
+ break;
+ case PROPERTY_TYPE_BIN:
+ BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len);
+ ctx->tail->data.bindata = mw_malloc(ctx->tail->bindata_len);
+ rbuf_pop(data, ctx->tail->data.bindata, ctx->tail->bindata_len);
+ ctx->bytes_consumed += ctx->tail->bindata_len;
+ ctx->state = PROPERTY_NEXT;
+ break;
+ case PROPERTY_TYPE_VBI:
+ rc = vbi_parser_parse(&ctx->vbi_parser_ctx, data, log);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ ctx->tail->data.uint32 = ctx->vbi_parser_ctx.result;
+ ctx->bytes_consumed += ctx->vbi_parser_ctx.bytes;
+ ctx->state = PROPERTY_NEXT;
+ break;
+ }
+ return rc;
+ case PROPERTY_TYPE_UINT8:
+ BUF_READ_CHECK_AT_LEAST(data, sizeof(uint8_t));
+ rbuf_pop(data, (char*)&ctx->tail->data.uint8, sizeof(uint8_t));
+ ctx->bytes_consumed += sizeof(uint8_t);
+ ctx->state = PROPERTY_NEXT;
+ break;
+ case PROPERTY_TYPE_UINT32:
+ BUF_READ_CHECK_AT_LEAST(data, sizeof(uint32_t));
+ rbuf_pop(data, (char*)&ctx->tail->data.uint32, sizeof(uint32_t));
+ ctx->tail->data.uint32 = be32toh(ctx->tail->data.uint32);
+ ctx->bytes_consumed += sizeof(uint32_t);
+ ctx->state = PROPERTY_NEXT;
+ break;
+ case PROPERTY_TYPE_UINT16:
+ BUF_READ_CHECK_AT_LEAST(data, sizeof(uint16_t));
+ rbuf_pop(data, (char*)&ctx->tail->data.uint16, sizeof(uint16_t));
+ ctx->tail->data.uint16 = be16toh(ctx->tail->data.uint16);
+ ctx->bytes_consumed += sizeof(uint16_t);
+ ctx->state = PROPERTY_NEXT;
+ /* FALLTHROUGH */
+ case PROPERTY_NEXT:
+ if (ctx->properties_length > ctx->bytes_consumed - ctx->vbi_length) {
+ ctx->state = PROPERTY_CREATE;
+ break;
+ } else
+ return MQTT_NG_CLIENT_PARSE_DONE;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_connack_varhdr(struct mqtt_ng_client *client)
+{
+ struct mqtt_ng_parser *parser = &client->parser;
+ switch (parser->varhdr_state) {
+ case MQTT_PARSE_VARHDR_INITIAL:
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+ rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.flags, 1);
+ rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.reason_code, 1);
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+ mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+ break;
+ case MQTT_PARSE_VARHDR_PROPS:
+ return parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+ default:
+ ERROR("invalid state for connack varhdr parser");
+ return MQTT_NG_CLIENT_INTERNAL_ERROR;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_disconnect_varhdr(struct mqtt_ng_client *client)
+{
+ struct mqtt_ng_parser *parser = &client->parser;
+ switch (parser->varhdr_state) {
+ case MQTT_PARSE_VARHDR_INITIAL:
+ if (!parser->mqtt_fixed_hdr_remaining_length) {
+ // [MQTT-3.14.2.1] if reason code omitted act same as == 0
+ parser->mqtt_packet.disconnect.reason_code = 0;
+ return MQTT_NG_CLIENT_PARSE_DONE;
+ }
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 1);
+ rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.reason_code, 1);
+ if (parser->mqtt_fixed_hdr_remaining_length == 1)
+ return MQTT_NG_CLIENT_PARSE_DONE;
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+ mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+ break;
+ case MQTT_PARSE_VARHDR_PROPS:
+ return parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+ default:
+ ERROR("invalid state for connack varhdr parser");
+ return MQTT_NG_CLIENT_INTERNAL_ERROR;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_puback_varhdr(struct mqtt_ng_client *client)
+{
+ struct mqtt_ng_parser *parser = &client->parser;
+ switch (parser->varhdr_state) {
+ case MQTT_PARSE_VARHDR_INITIAL:
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+ rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.puback.packet_id, 2);
+ parser->mqtt_packet.puback.packet_id = be16toh(parser->mqtt_packet.puback.packet_id);
+ if (parser->mqtt_fixed_hdr_remaining_length < 3) {
+ // [MQTT-3.4.2.1] if length is not big enough for reason code
+ // it is omitted and handled same as if it was present and == 0
+ // initially missed this detail and was wondering WTF is going on (sigh)
+ parser->mqtt_packet.puback.reason_code = 0;
+ return MQTT_NG_CLIENT_PARSE_DONE;
+ }
+ parser->varhdr_state = MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE:
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 1);
+ rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.puback.reason_code, 1);
+ // LOL so in CONNACK you have to have 0 byte to
+ // signify empty properties list
+ // but in PUBACK it can be omitted if remaining length doesn't allow it (sigh)
+ if (parser->mqtt_fixed_hdr_remaining_length < 4)
+ return MQTT_NG_CLIENT_PARSE_DONE;
+
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+ mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_PROPS:
+ return parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+ default:
+ ERROR("invalid state for puback varhdr parser");
+ return MQTT_NG_CLIENT_INTERNAL_ERROR;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_suback_varhdr(struct mqtt_ng_client *client)
+{
+ int rc;
+ size_t avail;
+ struct mqtt_ng_parser *parser = &client->parser;
+ struct mqtt_suback *suback = &client->parser.mqtt_packet.suback;
+ switch (parser->varhdr_state) {
+ case MQTT_PARSE_VARHDR_INITIAL:
+ suback->reason_codes = NULL;
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+ rbuf_pop(parser->received_data, (char*)&suback->packet_id, 2);
+ suback->packet_id = be16toh(suback->packet_id);
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+ parser->mqtt_parsed_len = 2;
+ mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_PROPS:
+ rc = parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+ if (rc != MQTT_NG_CLIENT_PARSE_DONE)
+ return rc;
+ parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;
+ suback->reason_code_count = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;
+ suback->reason_codes = mw_calloc(suback->reason_code_count, sizeof(*suback->reason_codes));
+ suback->reason_codes_pending = suback->reason_code_count;
+ parser->varhdr_state = MQTT_PARSE_REASONCODES;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_REASONCODES:
+ avail = rbuf_bytes_available(parser->received_data);
+ if (avail < 1)
+ return MQTT_NG_CLIENT_NEED_MORE_BYTES;
+
+ suback->reason_codes_pending -= rbuf_pop(parser->received_data, (char*)suback->reason_codes, MIN(suback->reason_codes_pending, avail));
+
+ if (!suback->reason_codes_pending)
+ return MQTT_NG_CLIENT_PARSE_DONE;
+
+ return MQTT_NG_CLIENT_NEED_MORE_BYTES;
+ default:
+ ERROR("invalid state for suback varhdr parser");
+ return MQTT_NG_CLIENT_INTERNAL_ERROR;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_publish_varhdr(struct mqtt_ng_client *client)
+{
+ int rc;
+ struct mqtt_ng_parser *parser = &client->parser;
+ struct mqtt_publish *publish = &client->parser.mqtt_packet.publish;
+ switch (parser->varhdr_state) {
+ case MQTT_PARSE_VARHDR_INITIAL:
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+ publish->topic = NULL;
+ publish->qos = ((parser->mqtt_control_packet_type >> 1) & 0x03);
+ rbuf_pop(parser->received_data, (char*)&publish->topic_len, 2);
+ publish->topic_len = be16toh(publish->topic_len);
+ parser->mqtt_parsed_len = 2;
+ if (!publish->topic_len) {
+ parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;
+ break;
+ }
+ publish->topic = mw_calloc(1, publish->topic_len + 1 /* add 0x00 */);
+ if (publish->topic == NULL)
+ return MQTT_NG_CLIENT_OOM;
+ parser->varhdr_state = MQTT_PARSE_VARHDR_TOPICNAME;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_TOPICNAME:
+ // TODO check empty topic can be valid? In which case we have to skip this step
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->topic_len);
+ rbuf_pop(parser->received_data, publish->topic, publish->topic_len);
+ parser->mqtt_parsed_len += publish->topic_len;
+ parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_POST_TOPICNAME:
+ mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+ if (!publish->qos) { // PacketID present only for QOS > 0 [MQTT-3.3.2.2]
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+ break;
+ }
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PACKET_ID;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_PACKET_ID:
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+ rbuf_pop(parser->received_data, (char*)&publish->packet_id, 2);
+ publish->packet_id = be16toh(publish->packet_id);
+ parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+ parser->mqtt_parsed_len += 2;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_VARHDR_PROPS:
+ rc = parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+ if (rc != MQTT_NG_CLIENT_PARSE_DONE)
+ return rc;
+ parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;
+ parser->varhdr_state = MQTT_PARSE_PAYLOAD;
+ /* FALLTHROUGH */
+ case MQTT_PARSE_PAYLOAD:
+ if (parser->mqtt_fixed_hdr_remaining_length < parser->mqtt_parsed_len) {
+ mw_free(publish->topic);
+ publish->topic = NULL;
+ ERROR("Error parsing PUBLISH message");
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+ publish->data_len = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;
+ if (!publish->data_len) {
+ publish->data = NULL;
+ return MQTT_NG_CLIENT_PARSE_DONE; // 0 length payload is OK [MQTT-3.3.3]
+ }
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->data_len);
+
+ publish->data = mw_malloc(publish->data_len);
+ if (publish->data == NULL) {
+ mw_free(publish->topic);
+ publish->topic = NULL;
+ return MQTT_NG_CLIENT_OOM;
+ }
+
+ rbuf_pop(parser->received_data, publish->data, publish->data_len);
+ parser->mqtt_parsed_len += publish->data_len;
+
+ return MQTT_NG_CLIENT_PARSE_DONE;
+ default:
+ ERROR("invalid state for publish varhdr parser");
+ return MQTT_NG_CLIENT_INTERNAL_ERROR;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+// TODO move to separate file, dont send whole client pointer just to be able
+// to access LOG context send parser only which should include log
+static int parse_data(struct mqtt_ng_client *client)
+{
+ int rc;
+ struct mqtt_ng_parser *parser = &client->parser;
+ switch(parser->state) {
+ case MQTT_PARSE_FIXED_HEADER_PACKET_TYPE:
+ BUF_READ_CHECK_AT_LEAST(parser->received_data, 1);
+ rbuf_pop(parser->received_data, (char*)&parser->mqtt_control_packet_type, 1);
+ vbi_parser_reset_ctx(&parser->vbi_parser);
+ parser->state = MQTT_PARSE_FIXED_HEADER_LEN;
+ break;
+ case MQTT_PARSE_FIXED_HEADER_LEN:
+ rc = vbi_parser_parse(&parser->vbi_parser, parser->received_data, client->log);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ parser->mqtt_fixed_hdr_remaining_length = parser->vbi_parser.result;
+ parser->state = MQTT_PARSE_VARIABLE_HEADER;
+ parser->varhdr_state = MQTT_PARSE_VARHDR_INITIAL;
+ break;
+ }
+ return rc;
+ case MQTT_PARSE_VARIABLE_HEADER:
+ switch (get_control_packet_type(parser->mqtt_control_packet_type)) {
+ case MQTT_CPT_CONNACK:
+ rc = parse_connack_varhdr(client);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ break;
+ }
+ return rc;
+ case MQTT_CPT_PUBACK:
+ rc = parse_puback_varhdr(client);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ break;
+ }
+ return rc;
+ case MQTT_CPT_SUBACK:
+ rc = parse_suback_varhdr(client);
+ if (rc != MQTT_NG_CLIENT_NEED_MORE_BYTES && rc != MQTT_NG_CLIENT_OK_CALL_AGAIN) {
+ mw_free(parser->mqtt_packet.suback.reason_codes);
+ }
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ break;
+ }
+ return rc;
+ case MQTT_CPT_PUBLISH:
+ rc = parse_publish_varhdr(client);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ break;
+ }
+ return rc;
+ case MQTT_CPT_PINGRESP:
+ if (parser->mqtt_fixed_hdr_remaining_length) {
+ ERROR ("PINGRESP has to be 0 Remaining Length."); // [MQTT-3.13.1]
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ break;
+ case MQTT_CPT_DISCONNECT:
+ rc = parse_disconnect_varhdr(client);
+ if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ break;
+ }
+ return rc;
+ default:
+ ERROR("Parsing Control Packet Type %" PRIu8 " not implemented yet.", get_control_packet_type(parser->mqtt_control_packet_type));
+ rbuf_bump_tail(parser->received_data, parser->mqtt_fixed_hdr_remaining_length);
+ parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+ return MQTT_NG_CLIENT_NOT_IMPL_YET;
+ }
+ // we could also return MQTT_NG_CLIENT_OK_CALL_AGAIN
+ // and be called again later
+ /* FALLTHROUGH */
+ case MQTT_PARSE_MQTT_PACKET_DONE:
+ parser->state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
+ return MQTT_NG_CLIENT_MQTT_PACKET_DONE;
+ }
+ return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+// set next MQTT fragment to send
+// return 1 if nothing to send
+// return -1 on error
+// return 0 if there is fragment set
+static int mqtt_ng_next_to_send(struct mqtt_ng_client *client) {
+ if (client->client_state == CONNECT_PENDING) {
+ client->main_buffer.sending_frag = client->connect_msg;
+ client->client_state = CONNECTING;
+ return 0;
+ }
+ if (client->client_state != CONNECTED)
+ return -1;
+
+ struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
+ while (frag) {
+ if ( frag->sent != frag->len )
+ break;
+ frag = frag->next;
+ }
+
+ if ( client->ping_pending && (!frag || (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD && frag->sent == 0)) ) {
+ client->ping_pending = 0;
+ ping_frag.sent = 0;
+ client->main_buffer.sending_frag = &ping_frag;
+ return 0;
+ }
+
+ client->main_buffer.sending_frag = frag;
+ return frag == NULL ? 1 : 0;
+}
+
+// send current fragment
+// return 0 if whole remaining length could be sent as a whole
+// return -1 if send buffer was filled and
+// nothing could be written anymore
+// return 1 if last fragment of a message was fully sent
+static int send_fragment(struct mqtt_ng_client *client) {
+ struct buffer_fragment *frag = client->main_buffer.sending_frag;
+
+ // for readability
+ char *ptr = frag->data + frag->sent;
+ size_t bytes = frag->len - frag->sent;
+
+ size_t processed = 0;
+
+ if (bytes)
+ processed = client->send_fnc_ptr(client->user_ctx, ptr, bytes);
+ else
+ WARN("This fragment was fully sent already. This should not happen!");
+
+ frag->sent += processed;
+ if (frag->sent != frag->len)
+ return -1;
+
+ if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL) {
+ client->time_of_last_send = time(NULL);
+ pthread_mutex_lock(&client->stats_mutex);
+ if (client->main_buffer.sending_frag != &ping_frag)
+ client->stats.tx_messages_queued--;
+ client->stats.tx_messages_sent++;
+ pthread_mutex_unlock(&client->stats_mutex);
+ client->main_buffer.sending_frag = NULL;
+ return 1;
+ }
+
+ client->main_buffer.sending_frag = frag->next;
+
+ return 0;
+}
+
+// attempt sending all fragments of current single MQTT packet
+static int send_all_message_fragments(struct mqtt_ng_client *client) {
+ int rc;
+ while ( !(rc = send_fragment(client)) );
+ return rc;
+}
+
+static void try_send_all(struct mqtt_ng_client *client) {
+ do {
+ if (client->main_buffer.sending_frag == NULL && mqtt_ng_next_to_send(client))
+ return;
+ } while(send_all_message_fragments(client) >= 0);
+}
+
+static inline void mark_message_for_gc(struct buffer_fragment *frag)
+{
+ while (frag) {
+ frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT;
+ buffer_frag_free_data(frag);
+ if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL)
+ return;
+ frag = frag->next;
+ }
+}
+
+static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)
+{
+ LOCK_HDR_BUFFER(&client->main_buffer);
+ struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
+ while (frag) {
+ if ( (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->packet_id == packet_id) {
+ if (!frag->sent) {
+ ERROR("Received packet_id (%" PRIu16 ") belongs to MQTT packet which was not yet sent!", packet_id);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+ return 1;
+ }
+ mark_message_for_gc(frag);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+ return 0;
+ }
+ frag = frag->next;
+ }
+ ERROR("Received packet_id (%" PRIu16 ") is unknown!", packet_id);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+ return 1;
+}
+
+int handle_incoming_traffic(struct mqtt_ng_client *client)
+{
+ int rc;
+ struct mqtt_publish *pub;
+ while( (rc = parse_data(client)) == MQTT_NG_CLIENT_OK_CALL_AGAIN );
+ if ( rc == MQTT_NG_CLIENT_MQTT_PACKET_DONE ) {
+ struct mqtt_property *prop;
+#ifdef MQTT_DEBUG_VERBOSE
+ DEBUG("MQTT Packet Parsed Successfully!");
+#endif
+ pthread_mutex_lock(&client->stats_mutex);
+ client->stats.rx_messages_rcvd++;
+ pthread_mutex_unlock(&client->stats_mutex);
+
+ switch (get_control_packet_type(client->parser.mqtt_control_packet_type)) {
+ case MQTT_CPT_CONNACK:
+#ifdef MQTT_DEBUG_VERBOSE
+ DEBUG("Received CONNACK");
+#endif
+ LOCK_HDR_BUFFER(&client->main_buffer);
+ mark_message_for_gc(client->connect_msg);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+ client->connect_msg = NULL;
+ if (client->client_state != CONNECTING) {
+ ERROR("Received unexpected CONNACK");
+ client->client_state = ERROR;
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+ if ((prop = get_property_by_id(client->parser.properties_parser.head, MQTT_PROP_MAX_PKT_SIZE)) != NULL) {
+ INFO("MQTT server limits message size to %" PRIu32, prop->data.uint32);
+ client->max_msg_size = prop->data.uint32;
+ }
+ if (client->connack_callback)
+ client->connack_callback(client->user_ctx, client->parser.mqtt_packet.connack.reason_code);
+ if (!client->parser.mqtt_packet.connack.reason_code) {
+ INFO("MQTT Connection Accepted By Server");
+ client->client_state = CONNECTED;
+ break;
+ }
+ client->client_state = ERROR;
+ return MQTT_NG_CLIENT_SERVER_RETURNED_ERROR;
+ case MQTT_CPT_PUBACK:
+#ifdef MQTT_DEBUG_VERBOSE
+ DEBUG("Received PUBACK %" PRIu16, client->parser.mqtt_packet.puback.packet_id);
+#endif
+ if (mark_packet_acked(client, client->parser.mqtt_packet.puback.packet_id))
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ if (client->puback_callback)
+ client->puback_callback(client->parser.mqtt_packet.puback.packet_id);
+ break;
+ case MQTT_CPT_PINGRESP:
+#ifdef MQTT_DEBUG_VERBOSE
+ DEBUG("Received PINGRESP");
+#endif
+ break;
+ case MQTT_CPT_SUBACK:
+#ifdef MQTT_DEBUG_VERBOSE
+ DEBUG("Received SUBACK %" PRIu16, client->parser.mqtt_packet.suback.packet_id);
+#endif
+ if (mark_packet_acked(client, client->parser.mqtt_packet.suback.packet_id))
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ break;
+ case MQTT_CPT_PUBLISH:
+#ifdef MQTT_DEBUG_VERBOSE
+ DEBUG("Recevied PUBLISH");
+#endif
+ pub = &client->parser.mqtt_packet.publish;
+ if (pub->qos > 1) {
+ mw_free(pub->topic);
+ mw_free(pub->data);
+ return MQTT_NG_CLIENT_NOT_IMPL_YET;
+ }
+ if ( pub->qos == 1 && (rc = mqtt_ng_puback(client, pub->packet_id, 0)) ) {
+ client->client_state = ERROR;
+ ERROR("Error generating PUBACK reply for PUBLISH");
+ return rc;
+ }
+ if ( (prop = get_property_by_id(client->parser.properties_parser.head, MQTT_PROP_TOPIC_ALIAS)) != NULL ) {
+ // Topic Alias property was sent from server
+ void *topic_ptr;
+ if (!c_rhash_get_ptr_by_uint64(client->rx_aliases, prop->data.uint8, &topic_ptr)) {
+ if (pub->topic != NULL) {
+ ERROR("We do not yet support topic alias reassignment");
+ return MQTT_NG_CLIENT_NOT_IMPL_YET;
+ }
+ pub->topic = topic_ptr;
+ } else {
+ if (pub->topic == NULL) {
+ ERROR("Topic alias with id %d unknown and topic not set by server!", prop->data.uint8);
+ return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+ }
+ c_rhash_insert_uint64_ptr(client->rx_aliases, prop->data.uint8, pub->topic);
+ }
+ }
+ if (client->msg_callback)
+ client->msg_callback(pub->topic, pub->data, pub->data_len, pub->qos);
+ // in case we have property topic alias and we have topic we take over the string
+ // and add pointer to it into topic alias list
+ if (prop == NULL)
+ mw_free(pub->topic);
+ mw_free(pub->data);
+ return MQTT_NG_CLIENT_WANT_WRITE;
+ case MQTT_CPT_DISCONNECT:
+ INFO ("Got MQTT DISCONNECT control packet from server. Reason code: %d", (int)client->parser.mqtt_packet.disconnect.reason_code);
+ client->client_state = DISCONNECTED;
+ break;
+ }
+ }
+
+ return rc;
+}
+
+int mqtt_ng_sync(struct mqtt_ng_client *client)
+{
+ if (client->client_state == RAW || client->client_state == DISCONNECTED)
+ return 0;
+
+ if (client->client_state == ERROR)
+ return 1;
+
+ LOCK_HDR_BUFFER(&client->main_buffer);
+ try_send_all(client);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+
+ int rc;
+
+ while ((rc = handle_incoming_traffic(client)) != MQTT_NG_CLIENT_NEED_MORE_BYTES) {
+ if (rc < 0)
+ break;
+ if (rc == MQTT_NG_CLIENT_WANT_WRITE) {
+ LOCK_HDR_BUFFER(&client->main_buffer);
+ try_send_all(client);
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+ }
+ }
+
+ if (rc < 0)
+ return rc;
+
+ return 0;
+}
+
+time_t mqtt_ng_last_send_time(struct mqtt_ng_client *client)
+{
+ return client->time_of_last_send;
+}
+
+void mqtt_ng_set_max_mem(struct mqtt_ng_client *client, size_t bytes)
+{
+ client->max_mem_bytes = bytes;
+}
+
+void mqtt_ng_get_stats(struct mqtt_ng_client *client, struct mqtt_ng_stats *stats)
+{
+ pthread_mutex_lock(&client->stats_mutex);
+ memcpy(stats, &client->stats, sizeof(struct mqtt_ng_stats));
+ pthread_mutex_unlock(&client->stats_mutex);
+
+ stats->tx_bytes_queued = 0;
+ stats->tx_buffer_reclaimable = 0;
+
+ LOCK_HDR_BUFFER(&client->main_buffer);
+ stats->tx_buffer_used = BUFFER_BYTES_USED(&client->main_buffer.hdr_buffer);
+ stats->tx_buffer_free = BUFFER_BYTES_AVAILABLE(&client->main_buffer.hdr_buffer);
+ stats->tx_buffer_size = client->main_buffer.hdr_buffer.size;
+ struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
+ while (frag) {
+ stats->tx_bytes_queued += frag->len - frag->sent;
+ if (frag_is_marked_for_gc(frag))
+ stats->tx_buffer_reclaimable += FRAG_SIZE_IN_BUFFER(frag);
+
+ frag = frag->next;
+ }
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
+}
+
+int mqtt_ng_set_topic_alias(struct mqtt_ng_client *client, const char *topic)
+{
+ uint16_t idx;
+ pthread_rwlock_wrlock(&client->tx_topic_aliases.rwlock);
+
+ if (client->tx_topic_aliases.idx_assigned >= client->tx_topic_aliases.idx_max) {
+ pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+ mws_error(client->log, "Tx topic alias indexes were exhausted (current version of the library doesn't support reassigning yet. Feel free to contribute.");
+ return 0; //0 is not a valid topic alias
+ }
+
+ struct topic_alias_data *alias;
+ if (!c_rhash_get_ptr_by_str(client->tx_topic_aliases.stoi_dict, topic, (void**)&alias)) {
+ // this is not a problem for library but might be helpful to warn user
+ // as it might indicate bug in their program (but also might be expected)
+ idx = alias->idx;
+ pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+ mws_debug(client->log, "%s topic \"%s\" already has alias set. Ignoring.", __FUNCTION__, topic);
+ return idx;
+ }
+
+ alias = mw_malloc(sizeof(struct topic_alias_data));
+ idx = ++client->tx_topic_aliases.idx_assigned;
+ alias->idx = idx;
+ __atomic_store_n(&alias->usage_count, 0, __ATOMIC_SEQ_CST);
+
+ c_rhash_insert_str_ptr(client->tx_topic_aliases.stoi_dict, topic, (void*)alias);
+
+ pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+ return idx;
+}
diff --git a/mqtt_websockets/src/mqtt_wss_client.c b/mqtt_websockets/src/mqtt_wss_client.c
new file mode 100644
index 00000000..01e2ffce
--- /dev/null
+++ b/mqtt_websockets/src/mqtt_wss_client.c
@@ -0,0 +1,1138 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+#define _GNU_SOURCE
+
+#include "mqtt_wss_client.h"
+#include "mqtt_ng.h"
+#include "ws_client.h"
+#include "common_internal.h"
+
+#include <stdlib.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <poll.h>
+#include <string.h>
+#include <time.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netinet/tcp.h> //TCP_NODELAY
+#include <netdb.h>
+
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+
+#define PIPE_READ_END 0
+#define PIPE_WRITE_END 1
+#define POLLFD_SOCKET 0
+#define POLLFD_PIPE 1
+
+#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110) && (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
+#include <openssl/conf.h>
+#endif
+
+//TODO MQTT_PUBLISH_RETAIN should not be needed anymore
+#define MQTT_PUBLISH_RETAIN 0x01
+#define MQTT_CONNECT_CLEAN_SESSION 0x02
+#define MQTT_CONNECT_WILL_RETAIN 0x20
+
+char *util_openssl_ret_err(int err)
+{
+ switch(err){
+ case SSL_ERROR_WANT_READ:
+ return "SSL_ERROR_WANT_READ";
+ case SSL_ERROR_WANT_WRITE:
+ return "SSL_ERROR_WANT_WRITE";
+ case SSL_ERROR_NONE:
+ return "SSL_ERROR_NONE";
+ case SSL_ERROR_ZERO_RETURN:
+ return "SSL_ERROR_ZERO_RETURN";
+ case SSL_ERROR_WANT_CONNECT:
+ return "SSL_ERROR_WANT_CONNECT";
+ case SSL_ERROR_WANT_ACCEPT:
+ return "SSL_ERROR_WANT_ACCEPT";
+ case SSL_ERROR_WANT_X509_LOOKUP:
+ return "SSL_ERROR_WANT_X509_LOOKUP";
+#ifdef SSL_ERROR_WANT_ASYNC
+ case SSL_ERROR_WANT_ASYNC:
+ return "SSL_ERROR_WANT_ASYNC";
+#endif
+#ifdef SSL_ERROR_WANT_ASYNC_JOB
+ case SSL_ERROR_WANT_ASYNC_JOB:
+ return "SSL_ERROR_WANT_ASYNC_JOB";
+#endif
+#ifdef SSL_ERROR_WANT_CLIENT_HELLO_CB
+ case SSL_ERROR_WANT_CLIENT_HELLO_CB:
+ return "SSL_ERROR_WANT_CLIENT_HELLO_CB";
+#endif
+ case SSL_ERROR_SYSCALL:
+ return "SSL_ERROR_SYSCALL";
+ case SSL_ERROR_SSL:
+ return "SSL_ERROR_SSL";
+ }
+ return "UNKNOWN";
+}
+
+struct mqtt_wss_client_struct {
+ ws_client *ws_client;
+
+ mqtt_wss_log_ctx_t log;
+
+// immediate connection (e.g. proxy server)
+ char *host;
+ int port;
+
+// target of connection (e.g. where we want to connect to)
+ char *target_host;
+ int target_port;
+
+ enum mqtt_wss_proxy_type proxy_type;
+ char *proxy_uname;
+ char *proxy_passwd;
+
+// nonblock IO related
+ int sockfd;
+ int write_notif_pipe[2];
+ struct pollfd poll_fds[2];
+
+ SSL_CTX *ssl_ctx;
+ SSL *ssl;
+ int ssl_flags;
+
+ struct mqtt_ng_client *mqtt;
+
+ int mqtt_keepalive;
+
+ pthread_mutex_t pub_lock;
+
+// signifies that we didn't write all MQTT wanted
+// us to write during last cycle (e.g. due to buffer
+// size) and thus we should arm POLLOUT
+ unsigned int mqtt_didnt_finish_write:1;
+
+ unsigned int mqtt_connected:1;
+ unsigned int mqtt_disconnecting:1;
+
+// Application layer callback pointers
+ void (*msg_callback)(const char *, const void *, size_t, int);
+ void (*puback_callback)(uint16_t packet_id);
+
+ pthread_mutex_t stat_lock;
+ struct mqtt_wss_stats stats;
+
+#ifdef MQTT_WSS_DEBUG
+ void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line);
+#endif
+};
+
+static void mws_connack_callback_ng(void *user_ctx, int code)
+{
+ mqtt_wss_client client = user_ctx;
+ switch(code) {
+ case 0:
+ client->mqtt_connected = 1;
+ return;
+//TODO manual labor: all the CONNACK error codes with some nice error message
+ default:
+ mws_error(client->log, "MQTT CONNACK returned error %d", code);
+ return;
+ }
+}
+
+static ssize_t mqtt_send_cb(void *user_ctx, const void* buf, size_t len)
+{
+ mqtt_wss_client mqtt_wss_client = user_ctx;
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(mqtt_wss_client->log, "mqtt_pal_sendall(len=%d)", len);
+#endif
+ int ret = ws_client_send(mqtt_wss_client->ws_client, WS_OP_BINARY_FRAME, buf, len);
+ if (ret >= 0 && (size_t)ret != len) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(mqtt_wss_client->log, "Not complete message sent (Msg=%d,Sent=%d). Need to arm POLLOUT!", len, ret);
+#endif
+ mqtt_wss_client->mqtt_didnt_finish_write = 1;
+ }
+ return ret;
+}
+
+mqtt_wss_client mqtt_wss_new(const char *log_prefix,
+ mqtt_wss_log_callback_t log_callback,
+ msg_callback_fnc_t msg_callback,
+ void (*puback_callback)(uint16_t packet_id))
+{
+ mqtt_wss_log_ctx_t log;
+
+ log = mqtt_wss_log_ctx_create(log_prefix, log_callback);
+ if(!log)
+ return NULL;
+
+ SSL_library_init();
+ SSL_load_error_strings();
+
+ mqtt_wss_client client = mw_calloc(1, sizeof(struct mqtt_wss_client_struct));
+ if (!client) {
+ mws_error(log, "OOM alocating mqtt_wss_client");
+ goto fail;
+ }
+
+ pthread_mutex_init(&client->pub_lock, NULL);
+ pthread_mutex_init(&client->stat_lock, NULL);
+
+ client->msg_callback = msg_callback;
+ client->puback_callback = puback_callback;
+
+ client->ws_client = ws_client_new(0, &client->target_host, log);
+ if (!client->ws_client) {
+ mws_error(log, "Error creating ws_client");
+ goto fail_1;
+ }
+
+ client->log = log;
+
+#ifdef __APPLE__
+ if (pipe(client->write_notif_pipe)) {
+#else
+ if (pipe2(client->write_notif_pipe, O_CLOEXEC /*| O_DIRECT*/)) {
+#endif
+ mws_error(log, "Couldn't create pipe");
+ goto fail_2;
+ }
+
+ client->poll_fds[POLLFD_PIPE].fd = client->write_notif_pipe[PIPE_READ_END];
+ client->poll_fds[POLLFD_PIPE].events = POLLIN;
+
+ client->poll_fds[POLLFD_SOCKET].events = POLLIN;
+
+ struct mqtt_ng_init settings = {
+ .log = log,
+ .data_in = client->ws_client->buf_to_mqtt,
+ .data_out_fnc = &mqtt_send_cb,
+ .user_ctx = client,
+ .connack_callback = &mws_connack_callback_ng,
+ .puback_callback = puback_callback,
+ .msg_callback = msg_callback
+ };
+ if ( (client->mqtt = mqtt_ng_init(&settings)) == NULL ) {
+ mws_error(log, "Error initializing internal MQTT client");
+ goto fail_3;
+ }
+
+ return client;
+
+fail_3:
+ close(client->write_notif_pipe[PIPE_WRITE_END]);
+ close(client->write_notif_pipe[PIPE_READ_END]);
+fail_2:
+ ws_client_destroy(client->ws_client);
+fail_1:
+ mw_free(client);
+fail:
+ mqtt_wss_log_ctx_destroy(log);
+ return NULL;
+}
+
+void mqtt_wss_set_max_buf_size(mqtt_wss_client client, size_t size)
+{
+ mqtt_ng_set_max_mem(client->mqtt, size);
+}
+
+void mqtt_wss_destroy(mqtt_wss_client client)
+{
+ mqtt_ng_destroy(client->mqtt);
+
+ close(client->write_notif_pipe[PIPE_WRITE_END]);
+ close(client->write_notif_pipe[PIPE_READ_END]);
+
+ ws_client_destroy(client->ws_client);
+
+ // deleted after client->ws_client
+ // as it "borrows" this pointer and might use it
+ if (client->target_host == client->host)
+ client->target_host = NULL;
+ if (client->target_host)
+ mw_free(client->target_host);
+ if (client->host)
+ mw_free(client->host);
+ mw_free(client->proxy_passwd);
+ mw_free(client->proxy_uname);
+
+ if (client->ssl)
+ SSL_free(client->ssl);
+
+ if (client->ssl_ctx)
+ SSL_CTX_free(client->ssl_ctx);
+
+ if (client->sockfd > 0)
+ close(client->sockfd);
+
+ pthread_mutex_destroy(&client->pub_lock);
+ pthread_mutex_destroy(&client->stat_lock);
+
+ mqtt_wss_log_ctx_destroy(client->log);
+ mw_free(client);
+}
+
+static int cert_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)
+{
+ SSL *ssl;
+ X509 *err_cert;
+ mqtt_wss_client client;
+ int err = 0, depth;
+ char *err_str;
+
+ ssl = X509_STORE_CTX_get_ex_data(ctx, SSL_get_ex_data_X509_STORE_CTX_idx());
+ client = SSL_get_ex_data(ssl, 0);
+
+ // TODO handle depth as per https://www.openssl.org/docs/man1.0.2/man3/SSL_CTX_set_verify.html
+
+ if (!preverify_ok) {
+ err = X509_STORE_CTX_get_error(ctx);
+ depth = X509_STORE_CTX_get_error_depth(ctx);
+ err_cert = X509_STORE_CTX_get_current_cert(ctx);
+ err_str = X509_NAME_oneline(X509_get_subject_name(err_cert), NULL, 0);
+
+ mws_error(client->log, "verify error:num=%d:%s:depth=%d:%s", err,
+ X509_verify_cert_error_string(err), depth, err_str);
+
+ mw_free(err_str);
+ }
+
+ if (!preverify_ok && err == X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT &&
+ client->ssl_flags & MQTT_WSS_SSL_ALLOW_SELF_SIGNED)
+ {
+ preverify_ok = 1;
+ mws_error(client->log, "Self Signed Certificate Accepted as the connection was "
+ "requested with MQTT_WSS_SSL_ALLOW_SELF_SIGNED");
+ }
+
+ return preverify_ok;
+}
+
+#define PROXY_CONNECT "CONNECT"
+#define PROXY_HTTP "HTTP/1.1"
+#define HTTP_ENDLINE "\x0D\x0A"
+#define HTTP_HDR_TERMINATOR "\x0D\x0A\x0D\x0A"
+#define HTTP_CODE_LEN 4
+#define HTTP_REASON_MAX_LEN 512
+static int http_parse_reply(mqtt_wss_client client, rbuf_t buf)
+{
+ char *ptr;
+ char http_code_s[4];
+ int http_code;
+ int idx;
+
+ if (rbuf_memcmp_n(buf, PROXY_HTTP, strlen(PROXY_HTTP))) {
+ mws_error(client->log, "http_proxy expected reply with \"" PROXY_HTTP "\"");
+ return 1;
+ }
+
+ rbuf_bump_tail(buf, strlen(PROXY_HTTP));
+
+ if (!rbuf_pop(buf, http_code_s, 1) || http_code_s[0] != 0x20) {
+ mws_error(client->log, "http_proxy missing space after \"" PROXY_HTTP "\"");
+ return 2;
+ }
+
+ if (!rbuf_pop(buf, http_code_s, HTTP_CODE_LEN)) {
+ mws_error(client->log, "http_proxy missing HTTP code");
+ return 3;
+ }
+
+ for (int i = 0; i < HTTP_CODE_LEN - 1; i++)
+ if (http_code_s[i] > 0x39 || http_code_s[i] < 0x30) {
+ mws_error(client->log, "http_proxy HTTP code non numeric");
+ return 4;
+ }
+
+ http_code_s[HTTP_CODE_LEN - 1] = 0;
+ http_code = atoi(http_code_s);
+
+ // TODO check if we ever have more headers here
+ rbuf_find_bytes(buf, HTTP_ENDLINE, strlen(HTTP_ENDLINE), &idx);
+ if (idx >= HTTP_REASON_MAX_LEN) {
+ mws_error(client->log, "http_proxy returned reason that is too long");
+ return 5;
+ }
+
+ if (http_code != 200) {
+ ptr = mw_malloc(idx + 1);
+ if (!ptr)
+ return 6;
+ rbuf_pop(buf, ptr, idx);
+ ptr[idx] = 0;
+
+ mws_error(client->log, "http_proxy returned error code %d \"%s\"", http_code, ptr);
+ mw_free(ptr);
+ return 7;
+ }/* else
+ rbuf_bump_tail(buf, idx);*/
+
+ rbuf_find_bytes(buf, HTTP_HDR_TERMINATOR, strlen(HTTP_HDR_TERMINATOR), &idx);
+ if (idx)
+ rbuf_bump_tail(buf, idx);
+
+ rbuf_bump_tail(buf, strlen(HTTP_HDR_TERMINATOR));
+
+ if (rbuf_bytes_available(buf)) {
+ mws_error(client->log, "http_proxy unexpected trailing bytes after end of HTTP hdr");
+ return 8;
+ }
+
+ mws_debug(client->log, "http_proxy CONNECT succeeded");
+ return 0;
+}
+
+#if defined(OPENSSL_VERSION_NUMBER) && OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
+static EVP_ENCODE_CTX *EVP_ENCODE_CTX_new(void)
+{
+ EVP_ENCODE_CTX *ctx = OPENSSL_malloc(sizeof(*ctx));
+
+ if (ctx != NULL) {
+ memset(ctx, 0, sizeof(*ctx));
+ }
+ return ctx;
+}
+static void EVP_ENCODE_CTX_free(EVP_ENCODE_CTX *ctx)
+{
+ OPENSSL_free(ctx);
+ return;
+}
+#endif
+
+inline static int base64_encode_helper(unsigned char *out, int *outl, const unsigned char *in, int in_len)
+{
+ int len;
+ unsigned char *str = out;
+ EVP_ENCODE_CTX *ctx = EVP_ENCODE_CTX_new();
+ EVP_EncodeInit(ctx);
+ EVP_EncodeUpdate(ctx, str, outl, in, in_len);
+ str += *outl;
+ EVP_EncodeFinal(ctx, str, &len);
+ *outl += len;
+
+ str = out;
+ while(*str) {
+ if (*str != 0x0D && *str != 0x0A)
+ *out++ = *str++;
+ else
+ str++;
+ }
+ *out = 0;
+
+ EVP_ENCODE_CTX_free(ctx);
+ return 0;
+}
+
+static int http_proxy_connect(mqtt_wss_client client)
+{
+ int rc;
+ struct pollfd poll_fd;
+ rbuf_t r_buf = rbuf_create(4096);
+ if (!r_buf)
+ return 1;
+ char *r_buf_ptr;
+ size_t r_buf_linear_insert_capacity;
+
+ poll_fd.fd = client->sockfd;
+ poll_fd.events = POLLIN;
+
+ r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);
+ snprintf(r_buf_ptr, r_buf_linear_insert_capacity,"%s %s:%d %s" HTTP_ENDLINE, PROXY_CONNECT, client->target_host, client->target_port, PROXY_HTTP);
+ write(client->sockfd, r_buf_ptr, strlen(r_buf_ptr));
+
+ if (client->proxy_uname) {
+ size_t creds_plain_len = strlen(client->proxy_uname) + strlen(client->proxy_passwd) + 2;
+ char *creds_plain = mw_malloc(creds_plain_len);
+ if (!creds_plain) {
+ mws_error(client->log, "OOM creds_plain");
+ rc = 6;
+ goto cleanup;
+ }
+ int creds_base64_len = (((4 * creds_plain_len / 3) + 3) & ~3);
+ // OpenSSL encoder puts newline every 64 output bytes
+ // we remove those but during encoding we need that space in the buffer
+ creds_base64_len += (1+(creds_base64_len/64)) * strlen("\n");
+ char *creds_base64 = mw_malloc(creds_base64_len + 1);
+ if (!creds_base64) {
+ mw_free(creds_plain);
+ mws_error(client->log, "OOM creds_base64");
+ rc = 6;
+ goto cleanup;
+ }
+ char *ptr = creds_plain;
+ strcpy(ptr, client->proxy_uname);
+ ptr += strlen(client->proxy_uname);
+ *ptr++ = ':';
+ strcpy(ptr, client->proxy_passwd);
+
+ int b64_len;
+ base64_encode_helper((unsigned char*)creds_base64, &b64_len, (unsigned char*)creds_plain, strlen(creds_plain));
+ mw_free(creds_plain);
+
+ r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);
+ snprintf(r_buf_ptr, r_buf_linear_insert_capacity,"Proxy-Authorization: Basic %s" HTTP_ENDLINE, creds_base64);
+ write(client->sockfd, r_buf_ptr, strlen(r_buf_ptr));
+ mw_free(creds_base64);
+ }
+ write(client->sockfd, HTTP_ENDLINE, strlen(HTTP_ENDLINE));
+
+ // read until you find CRLF, CRLF (HTTP HDR end)
+ // or ring buffer is full
+ // or timeout
+ while ((rc = poll(&poll_fd, 1, 1000)) >= 0) {
+ if (!rc) {
+ mws_error(client->log, "http_proxy timeout waiting reply from proxy server");
+ rc = 2;
+ goto cleanup;
+ }
+ r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);
+ if (!r_buf_ptr) {
+ mws_error(client->log, "http_proxy read ring buffer full");
+ rc = 3;
+ goto cleanup;
+ }
+ if ((rc = read(client->sockfd, r_buf_ptr, r_buf_linear_insert_capacity)) < 0) {
+ if (errno == EWOULDBLOCK || errno == EAGAIN) {
+ continue;
+ }
+ mws_error(client->log, "http_proxy error reading from socket \"%s\"", strerror(errno));
+ rc = 4;
+ goto cleanup;
+ }
+ rbuf_bump_head(r_buf, rc);
+ if (rbuf_find_bytes(r_buf, HTTP_HDR_TERMINATOR, strlen(HTTP_HDR_TERMINATOR), &rc)) {
+ rc = 0;
+ if (http_parse_reply(client, r_buf))
+ rc = 5;
+
+ goto cleanup;
+ }
+ }
+ mws_error(client->log, "proxy negotiation poll error \"%s\"", strerror(errno));
+ rc = 5;
+cleanup:
+ rbuf_free(r_buf);
+ return rc;
+}
+
+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)
+{
+ struct sockaddr_in addr;
+ memset(&addr, 0, sizeof(addr));
+ addr.sin_family = AF_INET;
+
+ struct hostent *he;
+ struct in_addr **addr_list;
+
+ if (!mqtt_params) {
+ mws_error(client->log, "mqtt_params can't be null!");
+ return -1;
+ }
+
+ // reset state in case this is reconnect
+ client->mqtt_didnt_finish_write = 0;
+ client->mqtt_connected = 0;
+ client->mqtt_disconnecting = 0;
+ ws_client_reset(client->ws_client);
+
+ if (client->target_host == client->host)
+ client->target_host = NULL;
+ if (client->target_host)
+ mw_free(client->target_host);
+ if (client->host)
+ mw_free(client->host);
+
+ if (proxy && proxy->type != MQTT_WSS_DIRECT) {
+ client->host = mw_strdup(proxy->host);
+ client->port = proxy->port;
+ client->target_host = mw_strdup(host);
+ client->target_port = port;
+ client->proxy_type = proxy->type;
+ if (proxy->username)
+ client->proxy_uname = mw_strdup(proxy->username);
+ if (proxy->password)
+ client->proxy_passwd = mw_strdup(proxy->password);
+ } else {
+ client->host = mw_strdup(host);
+ client->port = port;
+ client->target_host = client->host;
+ client->target_port = port;
+ }
+
+ client->ssl_flags = ssl_flags;
+
+ //TODO gethostbyname -> getaddinfo
+ // hstrerror -> gai_strerror
+ if ((he = gethostbyname(client->host)) == NULL) {
+ mws_error(client->log, "gethostbyname() error \"%s\"", hstrerror(h_errno));
+ return -1;
+ }
+
+ addr_list = (struct in_addr **)he->h_addr_list;
+ if(!addr_list[0]) {
+ mws_error(client->log, "No IP addr resolved");
+ return -1;
+ }
+ mws_debug(client->log, "Resolved IP: %s", inet_ntoa(*addr_list[0]));
+ addr.sin_addr = *addr_list[0];
+ addr.sin_port = htons(client->port);
+
+ if (client->sockfd > 0)
+ close(client->sockfd);
+ client->sockfd = socket(AF_INET, SOCK_STREAM, 0);
+ if (client->sockfd < 0) {
+ mws_error(client->log, "Couldn't create socket()");
+ return -1;
+ }
+
+ int flag = 1;
+ int result = setsockopt(client->sockfd,
+ IPPROTO_TCP,
+ TCP_NODELAY,
+ &flag,
+ sizeof(int));
+ if (result < 0)
+ mws_error(client->log, "Could not dissable NAGLE");
+
+ if (connect(client->sockfd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
+ mws_error(client->log, "Could not connect to remote endpoint \"%s\", port %d.\n", client->host, client->port);
+ return -3;
+ }
+
+ client->poll_fds[POLLFD_SOCKET].fd = client->sockfd;
+
+ if (fcntl(client->sockfd, F_SETFL, fcntl(client->sockfd, F_GETFL, 0) | O_NONBLOCK) == -1) {
+ mws_error(client->log, "Error setting O_NONBLOCK to TCP socket. \"%s\"", strerror(errno));
+ return -8;
+ }
+
+ if (client->proxy_type != MQTT_WSS_DIRECT)
+ if (http_proxy_connect(client))
+ return -4;
+
+#if OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
+#if (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
+ OPENSSL_config(NULL);
+#endif
+ SSL_load_error_strings();
+ SSL_library_init();
+#else
+ if (OPENSSL_init_ssl(OPENSSL_INIT_LOAD_CONFIG, NULL) != 1) {
+ mws_error(client->log, "Failed to initialize SSL");
+ return -1;
+ };
+#endif
+
+ // free SSL structs from possible previous connections
+ if (client->ssl)
+ SSL_free(client->ssl);
+ if (client->ssl_ctx)
+ SSL_CTX_free(client->ssl_ctx);
+
+ client->ssl_ctx = SSL_CTX_new(SSLv23_client_method());
+ if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
+ SSL_CTX_set_default_verify_paths(client->ssl_ctx);
+ SSL_CTX_set_verify(client->ssl_ctx, SSL_VERIFY_PEER | SSL_VERIFY_CLIENT_ONCE, cert_verify_callback);
+ } else
+ mws_error(client->log, "SSL Certificate checking completely disabled!!!");
+
+#ifdef MQTT_WSS_DEBUG
+ if(client->ssl_ctx_keylog_cb)
+ SSL_CTX_set_keylog_callback(client->ssl_ctx, client->ssl_ctx_keylog_cb);
+#endif
+
+ client->ssl = SSL_new(client->ssl_ctx);
+ if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
+ if (!SSL_set_ex_data(client->ssl, 0, client)) {
+ mws_error(client->log, "Could not SSL_set_ex_data");
+ return -4;
+ }
+ }
+ SSL_set_fd(client->ssl, client->sockfd);
+ SSL_set_connect_state(client->ssl);
+
+ if (!SSL_set_tlsext_host_name(client->ssl, client->target_host)) {
+ mws_error(client->log, "Error setting TLS SNI host");
+ return -7;
+ }
+
+ result = SSL_connect(client->ssl);
+ if (result != -1 && result != 1) {
+ mws_error(client->log, "SSL could not connect");
+ return -5;
+ }
+ if (result == -1) {
+ int ec = SSL_get_error(client->ssl, result);
+ if (ec != SSL_ERROR_WANT_READ && ec != SSL_ERROR_WANT_WRITE) {
+ mws_error(client->log, "Failed to start SSL connection");
+ return -6;
+ }
+ }
+
+ uint8_t mqtt_flags = (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK) << 3;
+ if (mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN)
+ mqtt_flags |= MQTT_CONNECT_WILL_RETAIN;
+ mqtt_flags |= MQTT_CONNECT_CLEAN_SESSION;
+
+ client->mqtt_keepalive = (mqtt_params->keep_alive ? mqtt_params->keep_alive : 400);
+
+ mws_info(client->log, "Going to connect using internal MQTT 5 implementation");
+ struct mqtt_auth_properties auth;
+ auth.client_id = (char*)mqtt_params->clientid;
+ auth.client_id_free = NULL;
+ auth.username = (char*)mqtt_params->username;
+ auth.username_free = NULL;
+ auth.password = (char*)mqtt_params->password;
+ auth.password_free = NULL;
+ struct mqtt_lwt_properties lwt;
+ lwt.will_topic = (char*)mqtt_params->will_topic;
+ lwt.will_topic_free = NULL;
+ lwt.will_message = (void*)mqtt_params->will_msg;
+ lwt.will_message_free = NULL; // TODO expose no copy version to API
+ lwt.will_message_size = mqtt_params->will_msg_len;
+ lwt.will_qos = (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK);
+ lwt.will_retain = mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN;
+ int ret = mqtt_ng_connect(client->mqtt, &auth, mqtt_params->will_msg ? &lwt : NULL, 1, client->mqtt_keepalive);
+ if (ret) {
+ mws_error(client->log, "Error generating MQTT connect");
+ return 1;
+ }
+
+ client->poll_fds[POLLFD_PIPE].events = POLLIN;
+ client->poll_fds[POLLFD_SOCKET].events = POLLIN;
+ // wait till MQTT connection is established
+ while (!client->mqtt_connected) {
+ if(mqtt_wss_service(client, -1)) {
+ mws_error(client->log, "Error connecting to MQTT WSS server \"%s\", port %d.", host, port);
+ return 2;
+ }
+ }
+
+ return 0;
+}
+
+#define NSEC_PER_USEC 1000ULL
+#define USEC_PER_SEC 1000000ULL
+#define NSEC_PER_MSEC 1000000ULL
+#define NSEC_PER_SEC 1000000000ULL
+
+static inline uint64_t boottime_usec(mqtt_wss_client client) {
+ struct timespec ts;
+#if defined(__APPLE__) || defined(__FreeBSD__)
+ if (clock_gettime(CLOCK_MONOTONIC, &ts) == -1) {
+#else
+ if (clock_gettime(CLOCK_BOOTTIME, &ts) == -1) {
+#endif
+ mws_error(client->log, "clock_gettimte failed");
+ return 0;
+ }
+ return (uint64_t)ts.tv_sec * USEC_PER_SEC + (ts.tv_nsec % NSEC_PER_SEC) / NSEC_PER_USEC;
+}
+
+#define MWS_TIMED_OUT 1
+#define MWS_ERROR 2
+#define MWS_OK 0
+static inline const char *mqtt_wss_error_tos(int ec)
+{
+ switch(ec) {
+ case MWS_TIMED_OUT:
+ return "Error: Operation was not able to finish in time";
+ case MWS_ERROR:
+ return "Unspecified Error";
+ default:
+ return "Unknown Error Code!";
+ }
+
+}
+
+static inline int mqtt_wss_service_all(mqtt_wss_client client, int timeout_ms)
+{
+ uint64_t exit_by = boottime_usec(client) + (timeout_ms * NSEC_PER_MSEC);
+ uint64_t now;
+ client->poll_fds[POLLFD_SOCKET].events |= POLLOUT; // TODO when entering mwtt_wss_service use out buffer size to arm POLLOUT
+ while (rbuf_bytes_available(client->ws_client->buf_write)) {
+ now = boottime_usec(client);
+ if (now >= exit_by)
+ return MWS_TIMED_OUT;
+ if (mqtt_wss_service(client, exit_by - now))
+ return MWS_ERROR;
+ }
+ return MWS_OK;
+}
+
+void mqtt_wss_disconnect(mqtt_wss_client client, int timeout_ms)
+{
+ int ret;
+
+ // block application from sending more MQTT messages
+ client->mqtt_disconnecting = 1;
+
+ // send whatever was left at the time of calling this function
+ ret = mqtt_wss_service_all(client, timeout_ms / 4);
+ if(ret)
+ mws_error(client->log,
+ "Error while trying to send all remaining data in an attempt "
+ "to gracefully disconnect! EC=%d Desc:\"%s\"",
+ ret,
+ mqtt_wss_error_tos(ret));
+
+ // schedule and send MQTT disconnect
+ mqtt_ng_disconnect(client->mqtt, 0);
+ mqtt_ng_sync(client->mqtt);
+
+ ret = mqtt_wss_service_all(client, timeout_ms / 4);
+ if(ret)
+ mws_error(client->log,
+ "Error while trying to send MQTT disconnect message in an attempt "
+ "to gracefully disconnect! EC=%d Desc:\"%s\"",
+ ret,
+ mqtt_wss_error_tos(ret));
+
+ // send WebSockets close message
+ uint16_t ws_rc = htobe16(1000);
+ ws_client_send(client->ws_client, WS_OP_CONNECTION_CLOSE, (const char*)&ws_rc, sizeof(ws_rc));
+ ret = mqtt_wss_service_all(client, timeout_ms / 4);
+ if(ret) {
+ // Some MQTT/WSS servers will close socket on receipt of MQTT disconnect and
+ // do not wait for WebSocket to be closed properly
+ mws_warn(client->log,
+ "Error while trying to send WebSocket disconnect message in an attempt "
+ "to gracefully disconnect! EC=%d Desc:\"%s\".",
+ ret,
+ mqtt_wss_error_tos(ret));
+ }
+
+ // Service WSS connection until remote closes connection (usual)
+ // or timeout happens (unusual) in which case we close
+ mqtt_wss_service_all(client, timeout_ms / 4);
+
+ close(client->sockfd);
+ client->sockfd = -1;
+}
+
+static inline void mqtt_wss_wakeup(mqtt_wss_client client)
+{
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "mqtt_wss_wakup - forcing wake up of main loop");
+#endif
+ write(client->write_notif_pipe[PIPE_WRITE_END], " ", 1);
+}
+
+#define THROWAWAY_BUF_SIZE 32
+char throwaway[THROWAWAY_BUF_SIZE];
+static inline void util_clear_pipe(int fd)
+{
+ (void)read(fd, throwaway, THROWAWAY_BUF_SIZE);
+}
+
+static inline void set_socket_pollfds(mqtt_wss_client client, int ssl_ret) {
+ if (ssl_ret == SSL_ERROR_WANT_WRITE)
+ client->poll_fds[POLLFD_SOCKET].events |= POLLOUT;
+ if (ssl_ret == SSL_ERROR_WANT_READ)
+ client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
+}
+
+static int handle_mqtt_internal(mqtt_wss_client client)
+{
+ int rc = mqtt_ng_sync(client->mqtt);
+ if (rc) {
+ mws_error(client->log, "mqtt_ng_sync returned %d != 0", rc);
+ client->mqtt_connected = 0;
+ return 1;
+ }
+ return 0;
+}
+
+#define SEC_TO_MSEC 1000
+static inline long long int t_till_next_keepalive_ms(mqtt_wss_client client)
+{
+ time_t last_send = mqtt_ng_last_send_time(client->mqtt);
+ long long int next_mqtt_keep_alive = (last_send * SEC_TO_MSEC)
+ + (client->mqtt_keepalive * (SEC_TO_MSEC * 0.75 /* SEND IN ADVANCE */));
+ return(next_mqtt_keep_alive - (time(NULL) * SEC_TO_MSEC));
+}
+
+#ifdef MQTT_WSS_CPUSTATS
+static inline uint64_t mqtt_wss_now_usec(mqtt_wss_client client) {
+ struct timespec ts;
+ if(clock_gettime(CLOCK_MONOTONIC, &ts) == -1) {
+ mws_error(client->log, "clock_gettime(CLOCK_MONOTONIC, &timespec) failed.");
+ return 0;
+ }
+ return (uint64_t)ts.tv_sec * USEC_PER_SEC + (ts.tv_nsec % NSEC_PER_SEC) / NSEC_PER_USEC;
+}
+#endif
+
+int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
+{
+ char *ptr;
+ size_t size;
+ int ret;
+ int send_keepalive = 0;
+
+#ifdef MQTT_WSS_CPUSTATS
+ uint64_t t1,t2;
+ t1 = mqtt_wss_now_usec(client);
+#endif
+
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, ">>>>> mqtt_wss_service <<<<<");
+ mws_debug(client->log, "Waiting for events: %s%s%s",
+ (client->poll_fds[POLLFD_SOCKET].events & POLLIN) ? "SOCKET_POLLIN " : "",
+ (client->poll_fds[POLLFD_SOCKET].events & POLLOUT) ? "SOCKET_POLLOUT " : "",
+ (client->poll_fds[POLLFD_PIPE].events & POLLIN) ? "PIPE_POLLIN" : "" );
+#endif
+
+ // Check user requested TO doesn't interfere with MQTT keep alives
+ long long int till_next_keep_alive = t_till_next_keepalive_ms(client);
+ if (client->mqtt_connected && (timeout_ms < 0 || timeout_ms >= till_next_keep_alive)) {
+ #ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "Shortening Timeout requested %d to %lld to ensure keep-alive can be sent", timeout_ms, till_next_keep_alive);
+ #endif
+ timeout_ms = till_next_keep_alive;
+ send_keepalive = 1;
+ }
+
+#ifdef MQTT_WSS_CPUSTATS
+ t2 = mqtt_wss_now_usec(client);
+ client->stats.time_keepalive += t2 - t1;
+#endif
+
+ if ((ret = poll(client->poll_fds, 2, timeout_ms >= 0 ? timeout_ms : -1)) < 0) {
+ if (errno == EINTR) {
+ mws_warn(client->log, "poll interrupted by EINTR");
+ return 0;
+ }
+ mws_error(client->log, "poll error \"%s\"", strerror(errno));
+ return -2;
+ }
+
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "Poll events happened: %s%s%s%s",
+ (client->poll_fds[POLLFD_SOCKET].revents & POLLIN) ? "SOCKET_POLLIN " : "",
+ (client->poll_fds[POLLFD_SOCKET].revents & POLLOUT) ? "SOCKET_POLLOUT " : "",
+ (client->poll_fds[POLLFD_PIPE].revents & POLLIN) ? "PIPE_POLLIN " : "",
+ (!ret) ? "POLL_TIMEOUT" : "");
+#endif
+
+#ifdef MQTT_WSS_CPUSTATS
+ t1 = mqtt_wss_now_usec(client);
+#endif
+
+ if (ret == 0) {
+ if (send_keepalive) {
+ // otherwise we shortened the timeout ourselves to take care of
+ // MQTT keep alives
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "Forcing MQTT Ping/keep-alive");
+#endif
+ mqtt_ng_ping(client->mqtt);
+ } else {
+ // if poll timed out and user requested timeout was being used
+ // return here let user do his work and he will call us back soon
+ return 0;
+ }
+ }
+
+#ifdef MQTT_WSS_CPUSTATS
+ t2 = mqtt_wss_now_usec(client);
+ client->stats.time_keepalive += t2 - t1;
+#endif
+
+ client->poll_fds[POLLFD_SOCKET].events = 0;
+
+ if ((ptr = rbuf_get_linear_insert_range(client->ws_client->buf_read, &size))) {
+ if((ret = SSL_read(client->ssl, ptr, size)) > 0) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "SSL_Read: Read %d.", ret);
+#endif
+ pthread_mutex_lock(&client->stat_lock);
+ client->stats.bytes_rx += ret;
+ pthread_mutex_unlock(&client->stat_lock);
+ rbuf_bump_head(client->ws_client->buf_read, ret);
+ } else {
+ int errnobkp = errno;
+ ret = SSL_get_error(client->ssl, ret);
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "Read Err: %s", util_openssl_ret_err(ret));
+#endif
+ set_socket_pollfds(client, ret);
+ if (ret != SSL_ERROR_WANT_READ &&
+ ret != SSL_ERROR_WANT_WRITE) {
+ mws_error(client->log, "SSL_read error: %d %s", ret, util_openssl_ret_err(ret));
+ if (ret == SSL_ERROR_SYSCALL)
+ mws_error(client->log, "SSL_read SYSCALL errno: %d %s", errnobkp, strerror(errnobkp));
+ return MQTT_WSS_ERR_CONN_DROP;
+ }
+ }
+ }
+
+#ifdef MQTT_WSS_CPUSTATS
+ t1 = mqtt_wss_now_usec(client);
+ client->stats.time_read_socket += t1 - t2;
+#endif
+
+ ret = ws_client_process(client->ws_client);
+ switch(ret) {
+ case WS_CLIENT_PROTOCOL_ERROR:
+ return MQTT_WSS_ERR_PROTO_WS;
+ case WS_CLIENT_NEED_MORE_BYTES:
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "WSCLIENT WANT READ");
+#endif
+ client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
+ break;
+ case WS_CLIENT_CONNECTION_CLOSED:
+ return MQTT_WSS_ERR_CONN_DROP;
+ }
+
+#ifdef MQTT_WSS_CPUSTATS
+ t2 = mqtt_wss_now_usec(client);
+ client->stats.time_process_websocket += t2 - t1;
+#endif
+
+ // process MQTT stuff
+ if(client->ws_client->state == WS_ESTABLISHED)
+ if (handle_mqtt_internal(client))
+ return MQTT_WSS_ERR_PROTO_MQTT;
+
+ if (client->mqtt_didnt_finish_write) {
+ client->mqtt_didnt_finish_write = 0;
+ client->poll_fds[POLLFD_SOCKET].events |= POLLOUT;
+ }
+
+#ifdef MQTT_WSS_CPUSTATS
+ t1 = mqtt_wss_now_usec(client);
+ client->stats.time_process_mqtt += t1 - t2;
+#endif
+
+ if ((ptr = rbuf_get_linear_read_range(client->ws_client->buf_write, &size))) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "Have data to write to SSL");
+#endif
+ if ((ret = SSL_write(client->ssl, ptr, size)) > 0) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "SSL_Write: Written %d of avail %d.", ret, size);
+#endif
+ pthread_mutex_lock(&client->stat_lock);
+ client->stats.bytes_tx += ret;
+ pthread_mutex_unlock(&client->stat_lock);
+ rbuf_bump_tail(client->ws_client->buf_write, ret);
+ } else {
+ int errnobkp = errno;
+ ret = SSL_get_error(client->ssl, ret);
+#ifdef DEBUG_ULTRA_VERBOSE
+ mws_debug(client->log, "Write Err: %s", util_openssl_ret_err(ret));
+#endif
+ set_socket_pollfds(client, ret);
+ if (ret != SSL_ERROR_WANT_READ &&
+ ret != SSL_ERROR_WANT_WRITE) {
+ mws_error(client->log, "SSL_write error: %d %s", ret, util_openssl_ret_err(ret));
+ if (ret == SSL_ERROR_SYSCALL)
+ mws_error(client->log, "SSL_write SYSCALL errno: %d %s", errnobkp, strerror(errnobkp));
+ return MQTT_WSS_ERR_CONN_DROP;
+ }
+ }
+ }
+
+ if(client->poll_fds[POLLFD_PIPE].revents & POLLIN)
+ util_clear_pipe(client->write_notif_pipe[PIPE_READ_END]);
+
+#ifdef MQTT_WSS_CPUSTATS
+ t2 = mqtt_wss_now_usec(client);
+ client->stats.time_write_socket += t2 - t1;
+#endif
+
+ return MQTT_WSS_OK;
+}
+
+int mqtt_wss_publish5(mqtt_wss_client client,
+ char *topic,
+ free_fnc_t topic_free,
+ void *msg,
+ free_fnc_t msg_free,
+ size_t msg_len,
+ uint8_t publish_flags,
+ uint16_t *packet_id)
+{
+ if (client->mqtt_disconnecting) {
+ mws_error(client->log, "mqtt_wss is disconnecting can't publish");
+ return 1;
+ }
+
+ if (!client->mqtt_connected) {
+ mws_error(client->log, "MQTT is offline. Can't send message.");
+ return 1;
+ }
+ uint8_t mqtt_flags = 0;
+
+ mqtt_flags = (publish_flags & MQTT_WSS_PUB_QOSMASK) << 1;
+ if (publish_flags & MQTT_WSS_PUB_RETAIN)
+ mqtt_flags |= MQTT_PUBLISH_RETAIN;
+
+ int rc = mqtt_ng_publish(client->mqtt, topic, topic_free, msg, msg_free, msg_len, mqtt_flags, packet_id);
+ if (rc == MQTT_NG_MSGGEN_MSG_TOO_BIG)
+ return MQTT_WSS_ERR_TOO_BIG_FOR_SERVER;
+
+ mqtt_wss_wakeup(client);
+
+ return rc;
+}
+
+int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level)
+{
+ (void)max_qos_level; //TODO now hardcoded
+ if (!client->mqtt_connected) {
+ mws_error(client->log, "MQTT is offline. Can't subscribe.");
+ return 1;
+ }
+
+ if (client->mqtt_disconnecting) {
+ mws_error(client->log, "mqtt_wss is disconnecting can't subscribe");
+ return 1;
+ }
+
+ struct mqtt_sub sub = {
+ .topic = topic,
+ .topic_free = NULL,
+ .options = /* max_qos_level & 0x3 TODO when QOS > 1 implemented */ 0x01 | (0x01 << 3)
+ };
+ mqtt_ng_subscribe(client->mqtt, &sub, 1);
+
+ mqtt_wss_wakeup(client);
+ return 0;
+}
+
+struct mqtt_wss_stats mqtt_wss_get_stats(mqtt_wss_client client)
+{
+ struct mqtt_wss_stats current;
+ pthread_mutex_lock(&client->stat_lock);
+ current = client->stats;
+ memset(&client->stats, 0, sizeof(client->stats));
+ pthread_mutex_unlock(&client->stat_lock);
+ mqtt_ng_get_stats(client->mqtt, &current.mqtt);
+ return current;
+}
+
+int mqtt_wss_set_topic_alias(mqtt_wss_client client, const char *topic)
+{
+ return mqtt_ng_set_topic_alias(client->mqtt, topic);
+}
+
+#ifdef MQTT_WSS_DEBUG
+void mqtt_wss_set_SSL_CTX_keylog_cb(mqtt_wss_client client, void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line))
+{
+ client->ssl_ctx_keylog_cb = ssl_ctx_keylog_cb;
+}
+#endif
diff --git a/mqtt_websockets/src/mqtt_wss_log.c b/mqtt_websockets/src/mqtt_wss_log.c
new file mode 100644
index 00000000..2c8cf32e
--- /dev/null
+++ b/mqtt_websockets/src/mqtt_wss_log.c
@@ -0,0 +1,127 @@
+#include "mqtt_wss_log.h"
+#include <stdlib.h>
+#include <stdarg.h>
+#include <string.h>
+#include <stdio.h>
+#include "common_internal.h"
+
+struct mqtt_wss_log_ctx {
+ mqtt_wss_log_callback_t extern_log_fnc;
+ char *ctx_prefix;
+ char *buffer;
+ char *buffer_w_ptr;
+ size_t buffer_bytes_avail;
+};
+
+#define LOG_BUFFER_SIZE 1024 * 4
+#define LOG_CTX_PREFIX_SEV_STR " : "
+#define LOG_CTX_PREFIX_LIMIT 15
+#define LOG_CTX_PREFIX_LIMIT_STR (LOG_CTX_PREFIX_LIMIT - (2 + strlen(LOG_CTX_PREFIX_SEV_STR))) // with [] characters and affixed ' ' it is total 15 chars
+#if (LOG_CTX_PREFIX_LIMIT * 10) > LOG_BUFFER_SIZE
+#error "LOG_BUFFER_SIZE too small"
+#endif
+mqtt_wss_log_ctx_t mqtt_wss_log_ctx_create(const char *ctx_prefix, mqtt_wss_log_callback_t log_callback)
+{
+ mqtt_wss_log_ctx_t ctx = mw_calloc(1, sizeof(struct mqtt_wss_log_ctx));
+ if(!ctx)
+ return NULL;
+
+ if(log_callback) {
+ ctx->extern_log_fnc = log_callback;
+ ctx->buffer = mw_calloc(1, LOG_BUFFER_SIZE);
+ if(!ctx->buffer)
+ goto cleanup;
+
+ ctx->buffer_w_ptr = ctx->buffer;
+ if(ctx_prefix) {
+ *(ctx->buffer_w_ptr++) = '[';
+ strncpy(ctx->buffer_w_ptr, ctx_prefix, LOG_CTX_PREFIX_LIMIT_STR);
+ ctx->buffer_w_ptr += strnlen(ctx_prefix, LOG_CTX_PREFIX_LIMIT_STR);
+ *(ctx->buffer_w_ptr++) = ']';
+ }
+ strcpy(ctx->buffer_w_ptr, LOG_CTX_PREFIX_SEV_STR);
+ ctx->buffer_w_ptr += strlen(LOG_CTX_PREFIX_SEV_STR);
+ // no term '\0' -> calloc is used
+
+ ctx->buffer_bytes_avail = LOG_BUFFER_SIZE - strlen(ctx->buffer);
+
+ return ctx;
+ }
+
+ if(ctx_prefix) {
+ ctx->ctx_prefix = strndup(ctx_prefix, LOG_CTX_PREFIX_LIMIT_STR);
+ if(!ctx->ctx_prefix)
+ goto cleanup;
+ }
+
+ return ctx;
+
+cleanup:
+ mw_free(ctx);
+ return NULL;
+}
+
+void mqtt_wss_log_ctx_destroy(mqtt_wss_log_ctx_t ctx)
+{
+ mw_free(ctx->ctx_prefix);
+ mw_free(ctx->buffer);
+ mw_free(ctx);
+}
+
+static inline char severity_to_c(int severity)
+{
+ switch (severity) {
+ case MQTT_WSS_LOG_FATAL:
+ return 'F';
+ case MQTT_WSS_LOG_ERROR:
+ return 'E';
+ case MQTT_WSS_LOG_WARN:
+ return 'W';
+ case MQTT_WSS_LOG_INFO:
+ return 'I';
+ case MQTT_WSS_LOG_DEBUG:
+ return 'D';
+ default:
+ return '?';
+ }
+}
+
+void mws_log(int severity, mqtt_wss_log_ctx_t ctx, const char *fmt, va_list args)
+{
+ size_t size;
+
+ if(ctx->extern_log_fnc) {
+ size = vsnprintf(ctx->buffer_w_ptr, ctx->buffer_bytes_avail, fmt, args);
+ *(ctx->buffer_w_ptr - 3) = severity_to_c(severity);
+
+ ctx->extern_log_fnc(severity, ctx->buffer);
+
+ if(size >= ctx->buffer_bytes_avail)
+ mws_error(ctx, "Last message of this type was truncated! Consider what you log or increase LOG_BUFFER_SIZE if really needed.");
+
+ return;
+ }
+
+ if(ctx->ctx_prefix)
+ printf("[%s] ", ctx->ctx_prefix);
+
+ printf("%c: ", severity_to_c(severity));
+
+ vprintf(fmt, args);
+ putchar('\n');
+}
+
+#define DEFINE_MWS_SEV_FNC(severity_fncname, severity) \
+void mws_ ## severity_fncname(mqtt_wss_log_ctx_t ctx, const char *fmt, ...) \
+{ \
+ va_list args; \
+ va_start(args, fmt); \
+ mws_log(severity, ctx, fmt, args); \
+ va_end(args); \
+}
+
+DEFINE_MWS_SEV_FNC(fatal, MQTT_WSS_LOG_FATAL)
+DEFINE_MWS_SEV_FNC(error, MQTT_WSS_LOG_ERROR)
+DEFINE_MWS_SEV_FNC(warn, MQTT_WSS_LOG_WARN )
+DEFINE_MWS_SEV_FNC(info, MQTT_WSS_LOG_INFO )
+DEFINE_MWS_SEV_FNC(debug, MQTT_WSS_LOG_DEBUG)
diff --git a/mqtt_websockets/src/test.c b/mqtt_websockets/src/test.c
new file mode 100644
index 00000000..aee23545
--- /dev/null
+++ b/mqtt_websockets/src/test.c
@@ -0,0 +1,100 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#include <unistd.h>
+#include <stdio.h>
+#include <string.h>
+#include <stdlib.h>
+
+#include "mqtt_wss_client.h"
+
+int test_exit = 0;
+int port = 0;
+
+void mqtt_wss_log_cb(mqtt_wss_log_type_t log_type, const char* str)
+{
+ (void)log_type;
+ puts(str);
+}
+
+#define TEST_MSGLEN_MAX 512
+void msg_callback(const char *topic, const void *msg, size_t msglen, int qos)
+{
+ char cmsg[TEST_MSGLEN_MAX];
+ size_t len = (msglen < TEST_MSGLEN_MAX - 1) ? msglen : (TEST_MSGLEN_MAX - 1);
+ memcpy(cmsg,
+ msg,
+ len);
+ cmsg[len] = 0;
+
+ if (!strcmp(cmsg, "shutdown"))
+ test_exit = 1;
+
+ printf("Got Message From Broker Topic \"%s\" QOS %d MSG: \"%s\"\n", topic, qos, cmsg);
+}
+
+#define TESTMSG "Hello World!"
+int client_handle(mqtt_wss_client client)
+{
+ struct mqtt_connect_params params = {
+ .clientid = "test",
+ .username = "anon",
+ .password = "anon",
+ .keep_alive = 10
+ };
+
+/* struct mqtt_wss_proxy proxy = {
+ .host = "127.0.0.1",
+ .port = 3128,
+ .type = MQTT_WSS_PROXY_HTTP
+ };*/
+
+ while (mqtt_wss_connect(client, "127.0.0.1", port, &params, MQTT_WSS_SSL_ALLOW_SELF_SIGNED, NULL /*&proxy*/)) {
+ printf("Connect failed\n");
+ sleep(1);
+ printf("Attempting Reconnect\n");
+ }
+ printf("Connection succeeded\n");
+
+ mqtt_wss_subscribe(client, "test", 1);
+
+ while (!test_exit) {
+ if(mqtt_wss_service(client, -1) < 0)
+ return 1;
+ }
+ return 0;
+}
+
+int main(int argc, char **argv)
+{
+ if (argc >= 2)
+ port = atoi(argv[1]);
+ if (!port)
+ port = 9002;
+ printf("Using port %d\n", port);
+
+ mqtt_wss_client client = mqtt_wss_new("main", mqtt_wss_log_cb, msg_callback, NULL);
+ if (!client) {
+ printf("Couldn't initialize mqtt_wss\n");
+ return 1;
+ }
+ while (!test_exit) {
+ printf("client_handle = %d\n", client_handle(client));
+ }
+ if (test_exit) {
+ mqtt_wss_disconnect(client, 2000);
+ }
+
+ mqtt_wss_destroy(client);
+ return 0;
+}
diff --git a/mqtt_websockets/src/ws_client.c b/mqtt_websockets/src/ws_client.c
new file mode 100644
index 00000000..47f633e7
--- /dev/null
+++ b/mqtt_websockets/src/ws_client.c
@@ -0,0 +1,744 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#include <fcntl.h>
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+#include <ctype.h>
+
+#include <openssl/evp.h>
+
+#include "ws_client.h"
+#include "common_internal.h"
+
+#ifdef MQTT_WEBSOCKETS_DEBUG
+#include "../c-rbuf/src/ringbuffer_internal.h"
+#endif
+
+#define UNIT_LOG_PREFIX "ws_client: "
+#define FATAL(fmt, ...) mws_fatal(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define ERROR(fmt, ...) mws_error(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define WARN(fmt, ...) mws_warn (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define INFO(fmt, ...) mws_info (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define DEBUG(fmt, ...) mws_debug(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+
+const char *websocket_upgrage_hdr = "GET /mqtt HTTP/1.1\x0D\x0A"
+ "Host: %s\x0D\x0A"
+ "Upgrade: websocket\x0D\x0A"
+ "Connection: Upgrade\x0D\x0A"
+ "Sec-WebSocket-Key: %s\x0D\x0A"
+ "Origin: http://example.com\x0D\x0A"
+ "Sec-WebSocket-Protocol: mqtt\x0D\x0A"
+ "Sec-WebSocket-Version: 13\x0D\x0A\x0D\x0A";
+
+const char *mqtt_protoid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+#define DEFAULT_RINGBUFFER_SIZE (1024*128)
+#define ENTROPY_SOURCE "/dev/urandom"
+ws_client *ws_client_new(size_t buf_size, char **host, mqtt_wss_log_ctx_t log)
+{
+ ws_client *client;
+
+ if(!host)
+ return NULL;
+
+ client = mw_calloc(1, sizeof(ws_client));
+ if (!client)
+ return NULL;
+
+ client->host = host;
+ client->log = log;
+
+ client->buf_read = rbuf_create(buf_size ? buf_size : DEFAULT_RINGBUFFER_SIZE);
+ if (!client->buf_read)
+ goto cleanup;
+
+ client->buf_write = rbuf_create(buf_size ? buf_size : DEFAULT_RINGBUFFER_SIZE);
+ if (!client->buf_write)
+ goto cleanup_1;
+
+ client->buf_to_mqtt = rbuf_create(buf_size ? buf_size : DEFAULT_RINGBUFFER_SIZE);
+ if (!client->buf_to_mqtt)
+ goto cleanup_2;
+
+ client->entropy_fd = open(ENTROPY_SOURCE, O_RDONLY);
+ if (client->entropy_fd < 1) {
+ ERROR("Error opening entropy source \"" ENTROPY_SOURCE "\". Reason: \"%s\"", strerror(errno));
+ goto cleanup_3;
+ }
+
+ return client;
+
+cleanup_3:
+ rbuf_free(client->buf_to_mqtt);
+cleanup_2:
+ rbuf_free(client->buf_write);
+cleanup_1:
+ rbuf_free(client->buf_read);
+cleanup:
+ mw_free(client);
+ return NULL;
+}
+
+void ws_client_free_headers(ws_client *client)
+{
+ struct http_header *ptr = client->hs.headers;
+ struct http_header *tmp;
+
+ while (ptr) {
+ tmp = ptr;
+ ptr = ptr->next;
+ mw_free(tmp);
+ }
+
+ client->hs.headers = NULL;
+ client->hs.headers_tail = NULL;
+ client->hs.hdr_count = 0;
+}
+
+void ws_client_destroy(ws_client *client)
+{
+ ws_client_free_headers(client);
+ mw_free(client->hs.nonce_reply);
+ mw_free(client->hs.http_reply_msg);
+ close(client->entropy_fd);
+ rbuf_free(client->buf_read);
+ rbuf_free(client->buf_write);
+ rbuf_free(client->buf_to_mqtt);
+ mw_free(client);
+}
+
+void ws_client_reset(ws_client *client)
+{
+ ws_client_free_headers(client);
+ mw_free(client->hs.nonce_reply);
+ client->hs.nonce_reply = NULL;
+ mw_free(client->hs.http_reply_msg);
+ client->hs.http_reply_msg = NULL;
+ rbuf_flush(client->buf_read);
+ rbuf_flush(client->buf_write);
+ rbuf_flush(client->buf_to_mqtt);
+ client->state = WS_RAW;
+ client->hs.hdr_state = WS_HDR_HTTP;
+ client->rx.parse_state = WS_FIRST_2BYTES;
+}
+
+#define MAX_HTTP_HDR_COUNT 128
+int ws_client_add_http_header(ws_client *client, struct http_header *hdr)
+{
+ if (client->hs.hdr_count > MAX_HTTP_HDR_COUNT) {
+ ERROR("Too many HTTP response header fields");
+ return -1;
+ }
+
+ if (client->hs.headers)
+ client->hs.headers_tail->next = hdr;
+ else
+ client->hs.headers = hdr;
+
+ client->hs.headers_tail = hdr;
+ client->hs.hdr_count++;
+
+ return 0;
+}
+
+int ws_client_want_write(ws_client *client)
+{
+ return rbuf_bytes_available(client->buf_write);
+}
+
+#define RAND_SRC "/dev/urandom"
+static int ws_client_get_nonce(ws_client *client, char *dest, unsigned int size)
+{
+ // we do not need crypto secure random here
+ // it's just used for protocol negotiation
+ int rd;
+ int f = open(RAND_SRC, O_RDONLY);
+ if (f < 0) {
+ ERROR("Error opening \"%s\". Err: \"%s\"", RAND_SRC, strerror(errno));
+ return -2;
+ }
+
+ if ((rd = read(f, dest, size)) > 0) {
+ close(f);
+ return rd;
+ }
+ close(f);
+ return -1;
+}
+
+#define WEBSOCKET_NONCE_SIZE 16
+#define TEMP_BUF_SIZE 4096
+int ws_client_start_handshake(ws_client *client)
+{
+ char nonce[WEBSOCKET_NONCE_SIZE];
+ char nonce_b64[256];
+ char second[TEMP_BUF_SIZE];
+ unsigned int md_len;
+ unsigned char *digest;
+ EVP_MD_CTX *md_ctx;
+ const EVP_MD *md;
+
+ if(!*client->host) {
+ ERROR("Hostname has not been set. We should not be able to come here!");
+ return 1;
+ }
+
+ ws_client_get_nonce(client, nonce, WEBSOCKET_NONCE_SIZE);
+ EVP_EncodeBlock((unsigned char *)nonce_b64, (const unsigned char *)nonce, WEBSOCKET_NONCE_SIZE);
+ snprintf(second, TEMP_BUF_SIZE, websocket_upgrage_hdr,
+ *client->host,
+ nonce_b64);
+ if(rbuf_bytes_free(client->buf_write) < strlen(second)) {
+ ERROR("Write buffer capacity too low.");
+ return 1;
+ }
+
+ rbuf_push(client->buf_write, second, strlen(second));
+ client->state = WS_HANDSHAKE;
+
+ //Calculating expected Sec-WebSocket-Accept reply
+ snprintf(second, TEMP_BUF_SIZE, "%s%s", nonce_b64, mqtt_protoid);
+
+#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110)
+ md_ctx = EVP_MD_CTX_create();
+#else
+ md_ctx = EVP_MD_CTX_new();
+#endif
+ if (md_ctx == NULL) {
+ ERROR("Cant create EVP_MD Context");
+ return 1;
+ }
+
+ md = EVP_get_digestbyname("sha1");
+ if (!md) {
+ ERROR("Unknown message digest");
+ return 1;
+ }
+
+ if ((digest = (unsigned char *)OPENSSL_malloc(EVP_MD_size(EVP_sha256()))) == NULL) {
+ ERROR("Cant alloc digest");
+ return 1;
+ }
+
+ EVP_DigestInit_ex(md_ctx, md, NULL);
+ EVP_DigestUpdate(md_ctx, second, strlen(second));
+ EVP_DigestFinal_ex(md_ctx, digest, &md_len);
+
+ EVP_EncodeBlock((unsigned char *)nonce_b64, digest, md_len);
+
+ mw_free(client->hs.nonce_reply);
+ client->hs.nonce_reply = mw_strdup(nonce_b64);
+
+ OPENSSL_free(digest);
+
+#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110)
+ EVP_MD_CTX_destroy(md_ctx);
+#else
+ EVP_MD_CTX_free(md_ctx);
+#endif
+
+ return 0;
+}
+
+#define BUF_READ_MEMCMP_CONST(const, err) \
+ if (rbuf_memcmp_n(client->buf_read, const, strlen(const))) { \
+ ERROR(err); \
+ rbuf_flush(client->buf_read); \
+ return WS_CLIENT_PROTOCOL_ERROR; \
+ }
+
+#define BUF_READ_CHECK_AT_LEAST(x) \
+ if (rbuf_bytes_available(client->buf_read) < x) \
+ return WS_CLIENT_NEED_MORE_BYTES;
+
+#define MAX_HTTP_LINE_LENGTH 1024*4
+#define HTTP_SC_LENGTH 4 // "XXX " http status code as C string
+#define WS_CLIENT_HTTP_HDR "HTTP/1.1 "
+#define WS_CONN_ACCEPT "sec-websocket-accept"
+#define HTTP_HDR_SEPARATOR ": "
+#define WS_NONCE_STRLEN_B64 28
+#define WS_HTTP_NEWLINE "\r\n"
+#define HTTP_HEADER_NAME_MAX_LEN 256
+#if HTTP_HEADER_NAME_MAX_LEN > MAX_HTTP_LINE_LENGTH
+#error "Buffer too small"
+#endif
+#if WS_NONCE_STRLEN_B64 > MAX_HTTP_LINE_LENGTH
+#error "Buffer too small"
+#endif
+
+#define HTTP_HDR_LINE_CHECK_LIMIT(x) if ((x) >= MAX_HTTP_LINE_LENGTH) \
+{ \
+ ERROR("HTTP line received is too long. Maximum is %d", MAX_HTTP_LINE_LENGTH); \
+ return WS_CLIENT_PROTOCOL_ERROR; \
+}
+
+int ws_client_parse_handshake_resp(ws_client *client)
+{
+ char buf[HTTP_SC_LENGTH];
+ int idx_crlf, idx_sep;
+ char *ptr;
+ size_t bytes;
+ switch (client->hs.hdr_state) {
+ case WS_HDR_HTTP:
+ BUF_READ_CHECK_AT_LEAST(strlen(WS_CLIENT_HTTP_HDR))
+ BUF_READ_MEMCMP_CONST(WS_CLIENT_HTTP_HDR, "Expected \"HTTP1.1\" header");
+ rbuf_bump_tail(client->buf_read, strlen(WS_CLIENT_HTTP_HDR));
+ client->hs.hdr_state = WS_HDR_RC;
+ break;
+ case WS_HDR_RC:
+ BUF_READ_CHECK_AT_LEAST(HTTP_SC_LENGTH); // "XXX " http return code
+ rbuf_pop(client->buf_read, buf, HTTP_SC_LENGTH);
+ if (buf[HTTP_SC_LENGTH - 1] != 0x20) {
+ ERROR("HTTP status code received is not terminated by space (0x20)");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ buf[HTTP_SC_LENGTH - 1] = 0;
+ client->hs.http_code = atoi(buf);
+ if (client->hs.http_code < 100 || client->hs.http_code >= 600) {
+ ERROR("HTTP status code received not in valid range 100-600");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ client->hs.hdr_state = WS_HDR_ENDLINE;
+ break;
+ case WS_HDR_ENDLINE:
+ ptr = rbuf_find_bytes(client->buf_read, WS_HTTP_NEWLINE, strlen(WS_HTTP_NEWLINE), &idx_crlf);
+ if (!ptr) {
+ bytes = rbuf_bytes_available(client->buf_read);
+ HTTP_HDR_LINE_CHECK_LIMIT(bytes);
+ return WS_CLIENT_NEED_MORE_BYTES;
+ }
+ HTTP_HDR_LINE_CHECK_LIMIT(idx_crlf);
+
+ client->hs.http_reply_msg = mw_malloc(idx_crlf+1);
+ rbuf_pop(client->buf_read, client->hs.http_reply_msg, idx_crlf);
+ client->hs.http_reply_msg[idx_crlf] = 0;
+ rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));
+ client->hs.hdr_state = WS_HDR_PARSE_HEADERS;
+ break;
+ case WS_HDR_PARSE_HEADERS:
+ ptr = rbuf_find_bytes(client->buf_read, WS_HTTP_NEWLINE, strlen(WS_HTTP_NEWLINE), &idx_crlf);
+ if (!ptr) {
+ bytes = rbuf_bytes_available(client->buf_read);
+ HTTP_HDR_LINE_CHECK_LIMIT(bytes);
+ return WS_CLIENT_NEED_MORE_BYTES;
+ }
+ HTTP_HDR_LINE_CHECK_LIMIT(idx_crlf);
+
+ if (!idx_crlf) { // empty line, header end
+ rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));
+ client->hs.hdr_state = WS_HDR_PARSE_DONE;
+ return 0;
+ }
+
+ ptr = rbuf_find_bytes(client->buf_read, HTTP_HDR_SEPARATOR, strlen(HTTP_HDR_SEPARATOR), &idx_sep);
+ if (!ptr || idx_sep > idx_crlf) {
+ ERROR("Expected HTTP hdr field key/value separator \": \" before endline in non empty HTTP header line");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ if (idx_crlf == idx_sep + (int)strlen(HTTP_HDR_SEPARATOR)) {
+ ERROR("HTTP Header value cannot be empty");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+
+ if (idx_sep > HTTP_HEADER_NAME_MAX_LEN) {
+ ERROR("HTTP header too long (%d)", idx_sep);
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+
+ struct http_header *hdr = mw_calloc(1, sizeof(struct http_header) + idx_crlf); //idx_crlf includes ": " that will be used as 2 \0 bytes
+ hdr->key = ((char*)hdr) + sizeof(struct http_header);
+ hdr->value = hdr->key + idx_sep + 1;
+
+ bytes = rbuf_pop(client->buf_read, hdr->key, idx_sep);
+ rbuf_bump_tail(client->buf_read, strlen(HTTP_HDR_SEPARATOR));
+
+ bytes = rbuf_pop(client->buf_read, hdr->value, idx_crlf - idx_sep - strlen(HTTP_HDR_SEPARATOR));
+ rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));
+
+ for (int i = 0; hdr->key[i]; i++)
+ hdr->key[i] = tolower(hdr->key[i]);
+
+// DEBUG("HTTP header \"%s\" received. Value \"%s\"", hdr->key, hdr->value);
+
+ if (ws_client_add_http_header(client, hdr))
+ return WS_CLIENT_PROTOCOL_ERROR;
+
+ if (!strcmp(hdr->key, WS_CONN_ACCEPT)) {
+ if (strcmp(client->hs.nonce_reply, hdr->value)) {
+ ERROR("Received NONCE \"%s\" does not match expected nonce of \"%s\"", hdr->value, client->hs.nonce_reply);
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ client->hs.nonce_matched = 1;
+ }
+
+ break;
+ case WS_HDR_PARSE_DONE:
+ if (!client->hs.nonce_matched) {
+ ERROR("Missing " WS_CONN_ACCEPT " header");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ if (client->hs.http_code != 101) {
+ ERROR("HTTP return code not 101. Received %d with msg \"%s\".", client->hs.http_code, client->hs.http_reply_msg);
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+
+ client->state = WS_ESTABLISHED;
+ client->hs.hdr_state = WS_HDR_ALL_DONE;
+ INFO("Websocket Connection Accepted By Server");
+ return WS_CLIENT_PARSING_DONE;
+ case WS_HDR_ALL_DONE:
+ FATAL("This is error we should never come here!");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ return 0;
+}
+
+#define BYTE_MSB 0x80
+#define WS_FINAL_FRAG BYTE_MSB
+#define WS_PAYLOAD_MASKED BYTE_MSB
+
+static inline size_t get_ws_hdr_size(size_t payload_size)
+{
+ size_t hdr_len = 2 + 4 /*mask*/;
+ if(payload_size > 125)
+ hdr_len += 2;
+ if(payload_size > 65535)
+ hdr_len += 6;
+ return hdr_len;
+}
+
+#define MAX_POSSIBLE_HDR_LEN 14
+int ws_client_send(ws_client *client, enum websocket_opcode frame_type, const char *data, size_t size)
+{
+ // TODO maybe? implement fragmenting, it is not necessary though
+ // as both tested MQTT brokers have no reuirement of one MQTT envelope
+ // be equal to one WebSockets envelope. Therefore there is no need to send
+ // one big MQTT message as single fragmented WebSocket envelope
+ char hdr[MAX_POSSIBLE_HDR_LEN];
+ char *ptr = hdr;
+ char *mask;
+ int size_written = 0;
+ size_t j = 0;
+
+ size_t w_buff_free = rbuf_bytes_free(client->buf_write);
+ size_t hdr_len = get_ws_hdr_size(size);
+
+ if (w_buff_free < hdr_len * 2) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ DEBUG("Write buffer full. Can't write requested %d size.", size);
+#endif
+ return 0;
+ }
+
+ if (w_buff_free < (hdr_len + size)) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ DEBUG("Can't write whole MQTT packet of %d bytes into the buffer. Will do partial send of %d.", size, w_buff_free - hdr_len);
+#endif
+ size = w_buff_free - hdr_len;
+ hdr_len = get_ws_hdr_size(size);
+ // the actual needed header size might decrease if we cut number of bytes
+ // if decrease of size crosses 65535 or 125 boundary
+ // but I can live with that at least for now
+ // worst case is we have 6 more bytes we could have written
+ // no bigus dealus
+ }
+
+ *ptr++ = frame_type | WS_FINAL_FRAG;
+
+ //generate length
+ *ptr = WS_PAYLOAD_MASKED;
+ if (size > 65535) {
+ *ptr++ |= 0x7f;
+ uint64_t be = htobe64(size);
+ memcpy(ptr, (void *)&be, sizeof(be));
+ ptr += sizeof(be);
+ } else if (size > 125) {
+ *ptr++ |= 0x7e;
+ uint16_t be = htobe16(size);
+ memcpy(ptr, (void *)&be, sizeof(be));
+ ptr += sizeof(be);
+ } else
+ *ptr++ |= size;
+
+ mask = ptr;
+ if (read(client->entropy_fd, mask, sizeof(uint32_t)) < (ssize_t)sizeof(uint32_t)) {
+ ERROR("Unable to get mask from \"" ENTROPY_SOURCE "\"");
+ return -2;
+ }
+
+ rbuf_push(client->buf_write, hdr, hdr_len);
+
+ if (!size)
+ return 0;
+
+ // copy and mask data in the write ringbuffer
+ while (size - size_written) {
+ size_t writable_bytes;
+ char *w_ptr = rbuf_get_linear_insert_range(client->buf_write, &writable_bytes);
+ if(!writable_bytes)
+ break;
+
+ writable_bytes = (writable_bytes > size) ? (size - size_written) : writable_bytes;
+
+ memcpy(w_ptr, &data[size_written], writable_bytes);
+ rbuf_bump_head(client->buf_write, writable_bytes);
+
+ for (size_t i = 0; i < writable_bytes; i++, j++)
+ w_ptr[i] ^= mask[j % 4];
+ size_written += writable_bytes;
+ }
+ return size_written;
+}
+
+static int check_opcode(ws_client *client,enum websocket_opcode oc)
+{
+ switch(oc) {
+ case WS_OP_BINARY_FRAME:
+ case WS_OP_CONNECTION_CLOSE:
+ case WS_OP_PING:
+ return 0;
+ case WS_OP_CONTINUATION_FRAME:
+ FATAL("WS_OP_CONTINUATION_FRAME NOT IMPLEMENTED YET!!!!");
+ return 0;
+ case WS_OP_TEXT_FRAME:
+ FATAL("WS_OP_TEXT_FRAME NOT IMPLEMENTED YET!!!!");
+ return 0;
+ case WS_OP_PONG:
+ FATAL("WS_OP_PONG NOT IMPLEMENTED YET!!!!");
+ return 0;
+ default:
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+}
+
+static inline void ws_client_rx_post_hdr_state(ws_client *client)
+{
+ switch(client->rx.opcode) {
+ case WS_OP_BINARY_FRAME:
+ client->rx.parse_state = WS_PAYLOAD_DATA;
+ return;
+ case WS_OP_CONNECTION_CLOSE:
+ client->rx.parse_state = WS_PAYLOAD_CONNECTION_CLOSE;
+ return;
+ case WS_OP_PING:
+ client->rx.parse_state = WS_PAYLOAD_PING_REQ_PAYLOAD;
+ return;
+ default:
+ client->rx.parse_state = WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD;
+ return;
+ }
+}
+
+#define LONGEST_POSSIBLE_HDR_PART 8
+int ws_client_process_rx_ws(ws_client *client)
+{
+ char buf[LONGEST_POSSIBLE_HDR_PART];
+ size_t size;
+ switch (client->rx.parse_state) {
+ case WS_FIRST_2BYTES:
+ BUF_READ_CHECK_AT_LEAST(2);
+ rbuf_pop(client->buf_read, buf, 2);
+ client->rx.opcode = buf[0] & (char)~BYTE_MSB;
+
+ if (!(buf[0] & (char)~WS_FINAL_FRAG)) {
+ ERROR("Not supporting fragmented messages yet!");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+
+ if (check_opcode(client, client->rx.opcode) == WS_CLIENT_PROTOCOL_ERROR)
+ return WS_CLIENT_PROTOCOL_ERROR;
+
+ if (buf[1] & (char)WS_PAYLOAD_MASKED) {
+ ERROR("Mask is not allowed in Server->Client Websocket direction.");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+
+ switch (buf[1]) {
+ case 127:
+ client->rx.parse_state = WS_PAYLOAD_EXTENDED_64;
+ break;
+ case 126:
+ client->rx.parse_state = WS_PAYLOAD_EXTENDED_16;
+ break;
+ default:
+ client->rx.payload_length = buf[1];
+ ws_client_rx_post_hdr_state(client);
+ }
+ break;
+ case WS_PAYLOAD_EXTENDED_16:
+ BUF_READ_CHECK_AT_LEAST(2);
+ rbuf_pop(client->buf_read, buf, 2);
+ client->rx.payload_length = be16toh(*((uint16_t *)buf));
+ ws_client_rx_post_hdr_state(client);
+ break;
+ case WS_PAYLOAD_EXTENDED_64:
+ BUF_READ_CHECK_AT_LEAST(LONGEST_POSSIBLE_HDR_PART);
+ rbuf_pop(client->buf_read, buf, LONGEST_POSSIBLE_HDR_PART);
+ client->rx.payload_length = be64toh(*((uint64_t *)buf));
+ ws_client_rx_post_hdr_state(client);
+ break;
+ case WS_PAYLOAD_DATA:
+ // TODO not pretty?
+ while (client->rx.payload_processed < client->rx.payload_length) {
+ size_t remaining = client->rx.payload_length - client->rx.payload_processed;
+ if (!rbuf_bytes_available(client->buf_read))
+ return WS_CLIENT_NEED_MORE_BYTES;
+ char *insert = rbuf_get_linear_insert_range(client->buf_to_mqtt, &size);
+ if (!insert) {
+#ifdef DEBUG_ULTRA_VERBOSE
+ DEBUG("BUFFER TOO FULL. Avail %d req %d", (int)size, (int)remaining);
+#endif
+ return WS_CLIENT_BUFFER_FULL;
+ }
+ size = (size > remaining) ? remaining : size;
+ size = rbuf_pop(client->buf_read, insert, size);
+ rbuf_bump_head(client->buf_to_mqtt, size);
+ client->rx.payload_processed += size;
+ }
+ client->rx.parse_state = WS_PACKET_DONE;
+ break;
+ case WS_PAYLOAD_CONNECTION_CLOSE:
+ // for WS_OP_CONNECTION_CLOSE allowed is
+ // a) empty payload
+ // b) 2byte reason code
+ // c) 2byte reason code followed by message
+ if (client->rx.payload_length == 1) {
+ ERROR("WebScoket CONNECTION_CLOSE can't have payload of size 1");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ if (!client->rx.payload_length) {
+ INFO("WebSocket server closed the connection without giving reason.");
+ client->rx.parse_state = WS_PACKET_DONE;
+ break;
+ }
+ client->rx.parse_state = WS_PAYLOAD_CONNECTION_CLOSE_EC;
+ break;
+ case WS_PAYLOAD_CONNECTION_CLOSE_EC:
+ BUF_READ_CHECK_AT_LEAST(sizeof(uint16_t));
+
+ rbuf_pop(client->buf_read, buf, sizeof(uint16_t));
+ client->rx.specific_data.op_close.ec = be16toh(*((uint16_t *)buf));
+ client->rx.payload_processed += sizeof(uint16_t);
+
+ if(client->rx.payload_processed == client->rx.payload_length) {
+ INFO("WebSocket server closed the connection with EC=%d. Without message.",
+ client->rx.specific_data.op_close.ec);
+ client->rx.parse_state = WS_PACKET_DONE;
+ break;
+ }
+ client->rx.parse_state = WS_PAYLOAD_CONNECTION_CLOSE_MSG;
+ break;
+ case WS_PAYLOAD_CONNECTION_CLOSE_MSG:
+ if (!client->rx.specific_data.op_close.reason)
+ client->rx.specific_data.op_close.reason = mw_malloc(client->rx.payload_length + 1);
+
+ while (client->rx.payload_processed < client->rx.payload_length) {
+ if (!rbuf_bytes_available(client->buf_read))
+ return WS_CLIENT_NEED_MORE_BYTES;
+ client->rx.payload_processed += rbuf_pop(client->buf_read,
+ &client->rx.specific_data.op_close.reason[client->rx.payload_processed - sizeof(uint16_t)],
+ client->rx.payload_length - client->rx.payload_processed);
+ }
+ client->rx.specific_data.op_close.reason[client->rx.payload_length] = 0;
+ INFO("WebSocket server closed the connection with EC=%d and reason \"%s\"",
+ client->rx.specific_data.op_close.ec,
+ client->rx.specific_data.op_close.reason);
+ mw_free(client->rx.specific_data.op_close.reason);
+ client->rx.specific_data.op_close.reason = NULL;
+ client->rx.parse_state = WS_PACKET_DONE;
+ break;
+ case WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD:
+ BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);
+ WARN("Skipping Websocket Packet of unsupported/unknown type");
+ if (client->rx.payload_length)
+ rbuf_bump_tail(client->buf_read, client->rx.payload_length);
+ client->rx.parse_state = WS_PACKET_DONE;
+ return WS_CLIENT_PARSING_DONE;
+ case WS_PAYLOAD_PING_REQ_PAYLOAD:
+ if (client->rx.payload_length > rbuf_get_capacity(client->buf_read) / 2) {
+ ERROR("Ping arrived with payload which is too big!");
+ return WS_CLIENT_INTERNAL_ERROR;
+ }
+ BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);
+ client->rx.specific_data.ping_msg = mw_malloc(client->rx.payload_length);
+ rbuf_pop(client->buf_read, client->rx.specific_data.ping_msg, client->rx.payload_length);
+ // TODO schedule this instead of sending right away
+ // then attempt to send as soon as buffer space clears up
+ size = ws_client_send(client, WS_OP_PONG, client->rx.specific_data.ping_msg, client->rx.payload_length);
+ if (size != client->rx.payload_length) {
+ ERROR("Unable to send the PONG as one packet back. Closing connection.");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ }
+ client->rx.parse_state = WS_PACKET_DONE;
+ return WS_CLIENT_PARSING_DONE;
+ case WS_PACKET_DONE:
+ client->rx.parse_state = WS_FIRST_2BYTES;
+ client->rx.payload_processed = 0;
+ if (client->rx.opcode == WS_OP_CONNECTION_CLOSE)
+ return WS_CLIENT_CONNECTION_CLOSED;
+ return WS_CLIENT_PARSING_DONE;
+ default:
+ FATAL("Unknown parse state");
+ return WS_CLIENT_INTERNAL_ERROR;
+ }
+ return 0;
+}
+
+int ws_client_process(ws_client *client)
+{
+ int ret;
+ switch(client->state) {
+ case WS_RAW:
+ if (ws_client_start_handshake(client))
+ return WS_CLIENT_INTERNAL_ERROR;
+ return WS_CLIENT_NEED_MORE_BYTES;
+ case WS_HANDSHAKE:
+ do {
+ ret = ws_client_parse_handshake_resp(client);
+ if (ret == WS_CLIENT_PROTOCOL_ERROR)
+ client->state = WS_ERROR;
+ if (ret == WS_CLIENT_PARSING_DONE && client->state == WS_ESTABLISHED)
+ ret = WS_CLIENT_NEED_MORE_BYTES;
+ } while (!ret);
+ break;
+ case WS_ESTABLISHED:
+ do {
+ ret = ws_client_process_rx_ws(client);
+ switch(ret) {
+ case WS_CLIENT_PROTOCOL_ERROR:
+ client->state = WS_ERROR;
+ break;
+ case WS_CLIENT_CONNECTION_CLOSED:
+ client->state = WS_CONN_CLOSED_GRACEFUL;
+ break;
+ }
+ // if ret == 0 we can continue parsing
+ // if ret == WS_CLIENT_PARSING_DONE we processed
+ // one websocket packet and attempt processing
+ // next one if data available in the buffer
+ } while (!ret || ret == WS_CLIENT_PARSING_DONE);
+ break;
+ case WS_ERROR:
+ ERROR("ws_client is in error state. Restart the connection!");
+ return WS_CLIENT_PROTOCOL_ERROR;
+ case WS_CONN_CLOSED_GRACEFUL:
+ ERROR("Connection has been gracefully closed. Calling this is useless (and probably bug) until you reconnect again.");
+ return WS_CLIENT_CONNECTION_CLOSED;
+ default:
+ FATAL("Unknown connection state! Probably memory corruption.");
+ return WS_CLIENT_INTERNAL_ERROR;
+ }
+ return ret;
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-17 18:15:17 +0000