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