1 /*
2 * libwebsockets - small server side websockets and web server implementation
3 *
4 * Copyright (C) 2010 - 2020 Andy Green <andy@warmcat.com>
5 *
6 * Permission is hereby granted, free of charge, to any person obtaining a copy
7 * of this software and associated documentation files (the "Software"), to
8 * deal in the Software without restriction, including without limitation the
9 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
10 * sell copies of the Software, and to permit persons to whom the Software is
11 * furnished to do so, subject to the following conditions:
12 *
13 * The above copyright notice and this permission notice shall be included in
14 * all copies or substantial portions of the Software.
15 *
16 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
22 * IN THE SOFTWARE.
23 */
24
25 #ifndef _PRIVATE_LIB_ROLES_MQTT
26 #define _PRIVATE_LIB_ROLES_MQTT 1
27
28 #include <libwebsockets/lws-mqtt.h>
29
30 extern struct lws_role_ops role_ops_mqtt;
31
32 #define lwsi_role_mqtt(wsi) (wsi->role_ops == &role_ops_mqtt)
33
34 #define LWS_MQTT_MAX_CHILDREN 8 /* max child streams on same parent */
35
36 #define LMQCP_LUT_FLAG_RESERVED_FLAGS 0x10
37 #define LMQCP_LUT_FLAG_PACKET_ID_NONE 0x00
38 #define LMQCP_LUT_FLAG_PACKET_ID_HAS 0x20
39 #define LMQCP_LUT_FLAG_PACKET_ID_QOS12 0x40
40 #define LMQCP_LUT_FLAG_PACKET_ID_MASK 0x60
41 #define LMQCP_LUT_FLAG_PAYLOAD 0x80 /* payload req (publish = opt)*/
42
43 #define lws_mqtt_str_is_not_empty(s) ( ((s)) && \
44 ((s))->len && \
45 ((s))->buf && \
46 *((s))->buf )
47
48 #define LWS_MQTT_RESPONSE_TIMEOUT (3 * LWS_US_PER_SEC)
49 #define LWS_MQTT_RETRY_CEILING (60 * LWS_US_PER_SEC)
50
51 typedef enum {
52 LMSPR_COMPLETED = 0,
53 LMSPR_NEED_MORE = 1,
54
55 LMSPR_FAILED_OOM = -1,
56 LMSPR_FAILED_OVERSIZE = -2,
57 LMSPR_FAILED_FORMAT = -3,
58 LMSPR_FAILED_ALREADY_COMPLETED = -4,
59 } lws_mqtt_stateful_primitive_return_t;
60
61 typedef struct {
62 uint32_t value;
63 char budget;
64 char consumed;
65 } lws_mqtt_vbi;
66
67 /* works for vbi, 2-byte and 4-byte fixed length */
68 static inline int
lws_mqtt_mb_first(lws_mqtt_vbi * vbi)69 lws_mqtt_mb_first(lws_mqtt_vbi *vbi) { return !vbi->consumed; }
70
71 int
72 lws_mqtt_vbi_encode(uint32_t value, void *buf);
73
74 /*
75 * Decode is done statefully on an arbitrary amount of input data (which may
76 * be one byte). It's like this so it can continue seamlessly if a buffer ends
77 * partway through the primitive, and the api matches the bulk binary data case.
78 *
79 * VBI decode:
80 *
81 * Initialize the lws_mqtt_vbi state by calling lws_mqtt_vbi_init() on it, then
82 * feed lws_mqtt_vbi_r() bytes to decode.
83 *
84 * Returns <0 for error, LMSPR_COMPLETED if done (vbi->value is valid), or
85 * LMSPR_NEED_MORE if more calls to lws_mqtt_vbi_r() with subsequent bytes
86 * needed.
87 *
88 * *in and *len are updated accordingly.
89 *
90 * 2-byte and 4-byte decode:
91 *
92 * Initialize the lws_mqtt_vbi state by calling lws_mqtt_2byte_init() or
93 * lws_mqtt_4byte_init() on it, then feed lws_mqtt_mb_parse() bytes
94 * to decode.
95 *
96 * Returns <0 for error, LMSPR_COMPLETED if done (vbi->value is valid), or
97 * LMSPR_NEED_MORE if more calls to lws_mqtt_mb_parse() with subsequent
98 * bytes needed.
99 *
100 * *in and *len are updated accordingly.
101 */
102
103 void
104 lws_mqtt_vbi_init(lws_mqtt_vbi *vbi);
105
106 void
107 lws_mqtt_2byte_init(lws_mqtt_vbi *vbi);
108
109 void
110 lws_mqtt_4byte_init(lws_mqtt_vbi *vbi);
111
112 lws_mqtt_stateful_primitive_return_t
113 lws_mqtt_vbi_r(lws_mqtt_vbi *vbi, const uint8_t **in, size_t *len);
114
115 lws_mqtt_stateful_primitive_return_t
116 lws_mqtt_mb_parse(lws_mqtt_vbi *vbi, const uint8_t **in, size_t *len);
117
118 typedef struct lws_mqtt_str_st {
119 uint8_t *buf;
120 uint16_t len;
121
122 uint16_t limit; /* it's cheaper to add the state here than
123 * the pointer to point to it elsewhere */
124 uint16_t pos;
125 char len_valid;
126 char needs_freeing;
127 } lws_mqtt_str_t;
128
129 static inline int
lws_mqtt_str_first(lws_mqtt_str_t * s)130 lws_mqtt_str_first(lws_mqtt_str_t *s) { return !s->buf && !s->pos; }
131
132
133 lws_mqtt_stateful_primitive_return_t
134 lws_mqtt_str_parse(lws_mqtt_str_t *bd, const uint8_t **in, size_t *len);
135
136 typedef enum {
137 LMQCPP_IDLE,
138
139 /* receive packet type part of fixed header took us out of idle... */
140 LMQCPP_CONNECT_PACKET = LMQCP_CTOS_CONNECT << 4,
141 LMQCPP_CONNECT_REMAINING_LEN_VBI,
142 LMQCPP_CONNECT_VH_PNAME,
143 LMQCPP_CONNECT_VH_PVERSION,
144 LMQCPP_CONNECT_VH_FLAGS,
145 LMQCPP_CONNECT_VH_KEEPALIVE,
146 LMQCPP_CONNECT_VH_PROPERTIES_VBI_LEN,
147
148 LMQCPP_CONNACK_PACKET = LMQCP_STOC_CONNACK << 4,
149 LMQCPP_CONNACK_VH_FLAGS,
150 LMQCPP_CONNACK_VH_RETURN_CODE,
151
152 LMQCPP_PUBLISH_PACKET = LMQCP_PUBLISH << 4,
153 LMQCPP_PUBLISH_REMAINING_LEN_VBI,
154 LMQCPP_PUBLISH_VH_TOPIC,
155 LMQCPP_PUBLISH_VH_PKT_ID,
156
157 LMQCPP_PUBACK_PACKET = LMQCP_PUBACK << 4,
158 LMQCPP_PUBACK_VH_PKT_ID,
159 LMQCPP_PUBACK_PROPERTIES_LEN_VBI,
160
161 LMQCPP_SUBACK_PACKET = LMQCP_STOC_SUBACK << 4,
162 LMQCPP_SUBACK_VH_PKT_ID,
163 LMQCPP_SUBACK_PAYLOAD,
164
165 LMQCPP_UNSUBACK_PACKET = LMQCP_STOC_UNSUBACK << 4,
166 LMQCPP_UNSUBACK_VH_PKT_ID,
167
168 LMQCPP_PINGRESP_ZERO = LMQCP_STOC_PINGRESP << 4,
169
170 LMQCPP_PAYLOAD,
171
172 LMQCPP_EAT_PROPERTIES_AND_COMPLETE,
173
174 LMQCPP_PROP_ID_VBI,
175
176 /* all possible property payloads */
177
178 /* 3.3.2.3.2 */
179 LMQCPP_PROP_PAYLOAD_FORMAT_INDICATOR_1BYTE = 0x101,
180
181 LMQCPP_PROP_MSG_EXPIRY_INTERVAL_4BYTE = 0x102,
182
183 LMQCPP_PROP_CONTENT_TYPE_UTF8S = 0x103,
184
185 LMQCPP_PROP_RESPONSE_TOPIC_UTF8S = 0x108,
186
187 LMQCPP_PROP_CORRELATION_BINDATA = 0x109,
188
189 LMQCPP_PROP_SUBSCRIPTION_ID_VBI = 0x10b,
190
191 LMQCPP_PROP_SESSION_EXPIRY_INTERVAL_4BYTE = 0x111,
192
193 LMQCPP_PROP_ASSIGNED_CLIENTID_UTF8S = 0x112,
194
195 LMQCPP_PROP_SERVER_KEEPALIVE_2BYTE = 0x113,
196
197 LMQCPP_PROP_AUTH_METHOD_UTF8S = 0x115,
198
199 LMQCPP_PROP_AUTH_DATA_BINDATA = 0x116,
200
201 LMQCPP_PROP_REQUEST_PROBLEM_INFO_1BYTE = 0x117,
202
203 LMQCPP_PROP_WILL_DELAY_INTERVAL_4BYTE = 0x118,
204
205 LMQCPP_PROP_REQUEST_REPSONSE_INFO_1BYTE = 0x119,
206
207 LMQCPP_PROP_RESPONSE_INFO_UTF8S = 0x11a,
208
209 LMQCPP_PROP_SERVER_REFERENCE_UTF8S = 0x11c,
210
211 LMQCPP_PROP_REASON_STRING_UTF8S = 0x11f,
212
213 LMQCPP_PROP_RECEIVE_MAXIMUM_2BYTE = 0x121,
214
215 LMQCPP_PROP_TOPIC_MAXIMUM_2BYTE = 0x122,
216
217 LMQCPP_PROP_TOPIC_ALIAS_2BYTE = 0x123,
218
219 LMQCPP_PROP_MAXIMUM_QOS_1BYTE = 0x124,
220
221 LMQCPP_PROP_RETAIN_AVAILABLE_1BYTE = 0x125,
222
223 LMQCPP_PROP_USER_PROPERTY_NAME_UTF8S = 0x126,
224 LMQCPP_PROP_USER_PROPERTY_VALUE_UTF8S = 0x226,
225
226 LMQCPP_PROP_MAXIMUM_PACKET_SIZE_4BYTE = 0x127,
227
228 LMQCPP_PROP_WILDCARD_SUBSCRIPTION_AVAILABLE_1BYTE = 0x128,
229
230 LMQCPP_PROP_SUBSCRIPTION_IDENTIFIER_AVAILABLE_1BYTE = 0x129,
231
232 LMQCPP_PROP_SHARED_SUBSCRIPTION_AVAILABLE_1BYTE = 0x12a,
233
234 } lws_mqtt_packet_parse_state_t;
235
236 /*
237 * the states an MQTT connection can be in
238 */
239
240 typedef enum {
241 LGSMQTT_UNKNOWN,
242 LGSMQTT_IDLE,
243 LGSMQTT_TRANSPORT_CONNECTED,
244
245 LGSMQTT_SENT_CONNECT,
246 LGSMQTT_ESTABLISHED,
247
248 LGSMQTT_SENT_SUBSCRIBE,
249 LGSMQTT_SUBSCRIBED,
250
251 } lwsgs_mqtt_states_t;
252
253 typedef struct lws_mqtt_parser_st {
254 /* lws_mqtt_str_t s_content_type; */
255 lws_mqtt_packet_parse_state_t state;
256 lws_mqtt_vbi vbit;
257
258 lws_mqtt_reason_t reason;
259
260 lws_mqtt_str_t s_temp;
261
262 uint8_t fixed_seen[4];
263 uint8_t props_seen[8];
264
265 uint8_t cpkt_flags;
266 uint32_t cpkt_remlen;
267
268 uint32_t props_len;
269 uint32_t consumed;
270 uint32_t prop_id;
271 uint32_t props_consumed;
272 uint32_t payload_consumed;
273
274 uint16_t keepalive;
275 uint16_t cpkt_id;
276 uint32_t n;
277
278 uint8_t temp[32];
279 uint8_t conn_rc;
280 uint8_t payload_format;
281 uint8_t packet_type_flags;
282 uint8_t conn_protocol_version;
283 uint8_t fixed;
284
285 uint8_t flag_pending_send_connack_close:1;
286 uint8_t flag_pending_send_reason_close:1;
287 uint8_t flag_prop_multi:1;
288 uint8_t flag_server:1;
289
290 } lws_mqtt_parser_t;
291
292 typedef struct lws_mqtt_subs {
293 struct lws_mqtt_subs *next;
294
295 uint8_t ref_count; /* number of children referencing */
296
297 /* subscription name + NUL overallocated here */
298 char topic[];
299 } lws_mqtt_subs_t;
300
301 typedef struct lws_mqtts {
302 lws_mqtt_parser_t par;
303 lwsgs_mqtt_states_t estate;
304 struct lws_dll2 active_session_list_head;
305 struct lws_dll2 limbo_session_list_head;
306 } lws_mqtts_t;
307
308 typedef struct lws_mqttc {
309 lws_mqtt_parser_t par;
310 lwsgs_mqtt_states_t estate;
311 lws_mqtt_str_t *id;
312 lws_mqtt_str_t *username;
313 lws_mqtt_str_t *password;
314 struct {
315 lws_mqtt_str_t *topic;
316 lws_mqtt_str_t *message;
317 lws_mqtt_qos_levels_t qos;
318 uint8_t retain;
319 } will;
320 uint16_t keep_alive_secs;
321 uint8_t conn_flags;
322 } lws_mqttc_t;
323
324 struct _lws_mqtt_related {
325 lws_mqttc_t client;
326 lws_sorted_usec_list_t sul_qos1_puback_wait; /* QoS1 puback wait TO */
327 struct lws *wsi; /**< so sul can use lws_container_of */
328 lws_mqtt_subs_t *subs_head; /**< Linked-list of heap-allocated subscription objects */
329 void *rx_cpkt_param;
330 uint16_t pkt_id;
331 uint16_t ack_pkt_id;
332 uint16_t sub_size;
333
334 #if defined(LWS_WITH_CLIENT)
335 uint8_t send_pingreq:1;
336 uint8_t session_resumed:1;
337 #endif
338 uint8_t inside_payload:1;
339 uint8_t inside_subscribe:1;
340 uint8_t inside_unsubscribe:1;
341 uint8_t send_puback:1;
342 uint8_t unacked_publish:1;
343
344 uint8_t done_subscribe:1;
345 };
346
347 /*
348 * New sessions are created by starting CONNECT. If the ClientID sent
349 * by the client matches a different, extant session, then the
350 * existing one is taken over and the new one created for duration of
351 * CONNECT processing is destroyed.
352 *
353 * On the server side, bearing in mind multiple simultaneous,
354 * fragmented CONNECTs may be interleaved ongoing, all state and
355 * parsing temps for a session must live in the session object.
356 */
357
358 struct lws_mqtt_endpoint_st;
359
360 typedef struct lws_mqtts_session_st {
361 struct lws_dll2 session_list;
362
363 } lws_mqtts_session_t;
364
365 #define ctl_pkt_type(x) (x->packet_type_flags >> 4)
366
367
368 void
369 lws_mqttc_state_transition(lws_mqttc_t *ep, lwsgs_mqtt_states_t s);
370
371 int
372 _lws_mqtt_rx_parser(struct lws *wsi, lws_mqtt_parser_t *par,
373 const uint8_t *buf, size_t len);
374
375 int
376 lws_mqtt_client_socket_service(struct lws *wsi, struct lws_pollfd *pollfd,
377 struct lws *wsi_conn);
378
379 int
380 lws_create_client_mqtt_object(const struct lws_client_connect_info *i,
381 struct lws *wsi);
382
383 struct lws *
384 lws_mqtt_client_send_connect(struct lws *wsi);
385
386 int
387 lws_mqtt_fill_fixed_header(uint8_t *p, lws_mqtt_control_packet_t ctrl_pkt_type,
388 uint8_t dup, lws_mqtt_qos_levels_t qos,
389 uint8_t retain);
390
391 struct lws *
392 lws_wsi_mqtt_adopt(struct lws *parent_wsi, struct lws *wsi);
393
394 lws_mqtt_subs_t *
395 lws_mqtt_find_sub(struct _lws_mqtt_related *mqtt, const char *topic);
396
397 #endif /* _PRIVATE_LIB_ROLES_MQTT */
398
399