1 /***************************************************************************
2 * _ _ ____ _
3 * Project ___| | | | _ \| |
4 * / __| | | | |_) | |
5 * | (__| |_| | _ <| |___
6 * \___|\___/|_| \_\_____|
7 *
8 * Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
9 * Copyright (C) Björn Stenberg, <bjorn@haxx.se>
10 *
11 * This software is licensed as described in the file COPYING, which
12 * you should have received as part of this distribution. The terms
13 * are also available at https://curl.se/docs/copyright.html.
14 *
15 * You may opt to use, copy, modify, merge, publish, distribute and/or sell
16 * copies of the Software, and permit persons to whom the Software is
17 * furnished to do so, under the terms of the COPYING file.
18 *
19 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
20 * KIND, either express or implied.
21 *
22 * SPDX-License-Identifier: curl
23 *
24 ***************************************************************************/
25
26 #include "curl_setup.h"
27
28 #ifndef CURL_DISABLE_MQTT
29
30 #include "urldata.h"
31 #include <curl/curl.h>
32 #include "transfer.h"
33 #include "sendf.h"
34 #include "progress.h"
35 #include "mqtt.h"
36 #include "select.h"
37 #include "strdup.h"
38 #include "url.h"
39 #include "escape.h"
40 #include "warnless.h"
41 #include "curl_printf.h"
42 #include "curl_memory.h"
43 #include "multiif.h"
44 #include "rand.h"
45
46 /* The last #include file should be: */
47 #include "memdebug.h"
48
49 #define MQTT_MSG_CONNECT 0x10
50 #define MQTT_MSG_CONNACK 0x20
51 #define MQTT_MSG_PUBLISH 0x30
52 #define MQTT_MSG_SUBSCRIBE 0x82
53 #define MQTT_MSG_SUBACK 0x90
54 #define MQTT_MSG_DISCONNECT 0xe0
55
56 #define MQTT_CONNACK_LEN 2
57 #define MQTT_SUBACK_LEN 3
58 #define MQTT_CLIENTID_LEN 12 /* "curl0123abcd" */
59
60 /*
61 * Forward declarations.
62 */
63
64 static CURLcode mqtt_do(struct Curl_easy *data, bool *done);
65 static CURLcode mqtt_done(struct Curl_easy *data,
66 CURLcode status, bool premature);
67 static CURLcode mqtt_doing(struct Curl_easy *data, bool *done);
68 static int mqtt_getsock(struct Curl_easy *data, struct connectdata *conn,
69 curl_socket_t *sock);
70 static CURLcode mqtt_setup_conn(struct Curl_easy *data,
71 struct connectdata *conn);
72
73 /*
74 * MQTT protocol handler.
75 */
76
77 const struct Curl_handler Curl_handler_mqtt = {
78 "mqtt", /* scheme */
79 mqtt_setup_conn, /* setup_connection */
80 mqtt_do, /* do_it */
81 mqtt_done, /* done */
82 ZERO_NULL, /* do_more */
83 ZERO_NULL, /* connect_it */
84 ZERO_NULL, /* connecting */
85 mqtt_doing, /* doing */
86 ZERO_NULL, /* proto_getsock */
87 mqtt_getsock, /* doing_getsock */
88 ZERO_NULL, /* domore_getsock */
89 ZERO_NULL, /* perform_getsock */
90 ZERO_NULL, /* disconnect */
91 ZERO_NULL, /* write_resp */
92 ZERO_NULL, /* write_resp_hd */
93 ZERO_NULL, /* connection_check */
94 ZERO_NULL, /* attach connection */
95 PORT_MQTT, /* defport */
96 CURLPROTO_MQTT, /* protocol */
97 CURLPROTO_MQTT, /* family */
98 PROTOPT_NONE /* flags */
99 };
100
mqtt_setup_conn(struct Curl_easy * data,struct connectdata * conn)101 static CURLcode mqtt_setup_conn(struct Curl_easy *data,
102 struct connectdata *conn)
103 {
104 /* allocate the HTTP-specific struct for the Curl_easy, only to survive
105 during this request */
106 struct MQTT *mq;
107 (void)conn;
108 DEBUGASSERT(data->req.p.mqtt == NULL);
109
110 mq = calloc(1, sizeof(struct MQTT));
111 if(!mq)
112 return CURLE_OUT_OF_MEMORY;
113 Curl_dyn_init(&mq->recvbuf, DYN_MQTT_RECV);
114 data->req.p.mqtt = mq;
115 return CURLE_OK;
116 }
117
mqtt_send(struct Curl_easy * data,char * buf,size_t len)118 static CURLcode mqtt_send(struct Curl_easy *data,
119 char *buf, size_t len)
120 {
121 CURLcode result = CURLE_OK;
122 struct MQTT *mq = data->req.p.mqtt;
123 size_t n;
124 result = Curl_xfer_send(data, buf, len, FALSE, &n);
125 if(result)
126 return result;
127 Curl_debug(data, CURLINFO_HEADER_OUT, buf, (size_t)n);
128 if(len != n) {
129 size_t nsend = len - n;
130 char *sendleftovers = Curl_memdup(&buf[n], nsend);
131 if(!sendleftovers)
132 return CURLE_OUT_OF_MEMORY;
133 mq->sendleftovers = sendleftovers;
134 mq->nsend = nsend;
135 }
136 else {
137 mq->sendleftovers = NULL;
138 mq->nsend = 0;
139 }
140 return result;
141 }
142
143 /* Generic function called by the multi interface to figure out what socket(s)
144 to wait for and for what actions during the DOING and PROTOCONNECT
145 states */
mqtt_getsock(struct Curl_easy * data,struct connectdata * conn,curl_socket_t * sock)146 static int mqtt_getsock(struct Curl_easy *data,
147 struct connectdata *conn,
148 curl_socket_t *sock)
149 {
150 (void)data;
151 sock[0] = conn->sock[FIRSTSOCKET];
152 return GETSOCK_READSOCK(FIRSTSOCKET);
153 }
154
mqtt_encode_len(char * buf,size_t len)155 static int mqtt_encode_len(char *buf, size_t len)
156 {
157 int i;
158
159 for(i = 0; (len > 0) && (i < 4); i++) {
160 unsigned char encoded;
161 encoded = len % 0x80;
162 len /= 0x80;
163 if(len)
164 encoded |= 0x80;
165 buf[i] = (char)encoded;
166 }
167
168 return i;
169 }
170
171 /* add the passwd to the CONNECT packet */
add_passwd(const char * passwd,const size_t plen,char * pkt,const size_t start,int remain_pos)172 static int add_passwd(const char *passwd, const size_t plen,
173 char *pkt, const size_t start, int remain_pos)
174 {
175 /* magic number that need to be set properly */
176 const size_t conn_flags_pos = remain_pos + 8;
177 if(plen > 0xffff)
178 return 1;
179
180 /* set password flag */
181 pkt[conn_flags_pos] |= 0x40;
182
183 /* length of password provided */
184 pkt[start] = (char)((plen >> 8) & 0xFF);
185 pkt[start + 1] = (char)(plen & 0xFF);
186 memcpy(&pkt[start + 2], passwd, plen);
187 return 0;
188 }
189
190 /* add user to the CONNECT packet */
add_user(const char * username,const size_t ulen,unsigned char * pkt,const size_t start,int remain_pos)191 static int add_user(const char *username, const size_t ulen,
192 unsigned char *pkt, const size_t start, int remain_pos)
193 {
194 /* magic number that need to be set properly */
195 const size_t conn_flags_pos = remain_pos + 8;
196 if(ulen > 0xffff)
197 return 1;
198
199 /* set username flag */
200 pkt[conn_flags_pos] |= 0x80;
201 /* length of username provided */
202 pkt[start] = (unsigned char)((ulen >> 8) & 0xFF);
203 pkt[start + 1] = (unsigned char)(ulen & 0xFF);
204 memcpy(&pkt[start + 2], username, ulen);
205 return 0;
206 }
207
208 /* add client ID to the CONNECT packet */
add_client_id(const char * client_id,const size_t client_id_len,char * pkt,const size_t start)209 static int add_client_id(const char *client_id, const size_t client_id_len,
210 char *pkt, const size_t start)
211 {
212 if(client_id_len != MQTT_CLIENTID_LEN)
213 return 1;
214 pkt[start] = 0x00;
215 pkt[start + 1] = MQTT_CLIENTID_LEN;
216 memcpy(&pkt[start + 2], client_id, MQTT_CLIENTID_LEN);
217 return 0;
218 }
219
220 /* Set initial values of CONNECT packet */
init_connpack(char * packet,char * remain,int remain_pos)221 static int init_connpack(char *packet, char *remain, int remain_pos)
222 {
223 /* Fixed header starts */
224 /* packet type */
225 packet[0] = MQTT_MSG_CONNECT;
226 /* remaining length field */
227 memcpy(&packet[1], remain, remain_pos);
228 /* Fixed header ends */
229
230 /* Variable header starts */
231 /* protocol length */
232 packet[remain_pos + 1] = 0x00;
233 packet[remain_pos + 2] = 0x04;
234 /* protocol name */
235 packet[remain_pos + 3] = 'M';
236 packet[remain_pos + 4] = 'Q';
237 packet[remain_pos + 5] = 'T';
238 packet[remain_pos + 6] = 'T';
239 /* protocol level */
240 packet[remain_pos + 7] = 0x04;
241 /* CONNECT flag: CleanSession */
242 packet[remain_pos + 8] = 0x02;
243 /* keep-alive 0 = disabled */
244 packet[remain_pos + 9] = 0x00;
245 packet[remain_pos + 10] = 0x3c;
246 /* end of variable header */
247 return remain_pos + 10;
248 }
249
mqtt_connect(struct Curl_easy * data)250 static CURLcode mqtt_connect(struct Curl_easy *data)
251 {
252 CURLcode result = CURLE_OK;
253 int pos = 0;
254 int rc = 0;
255 /* remain length */
256 int remain_pos = 0;
257 char remain[4] = {0};
258 size_t packetlen = 0;
259 size_t payloadlen = 0;
260 size_t start_user = 0;
261 size_t start_pwd = 0;
262 char client_id[MQTT_CLIENTID_LEN + 1] = "curl";
263 const size_t clen = strlen("curl");
264 char *packet = NULL;
265
266 /* extracting username from request */
267 const char *username = data->state.aptr.user ?
268 data->state.aptr.user : "";
269 const size_t ulen = strlen(username);
270 /* extracting password from request */
271 const char *passwd = data->state.aptr.passwd ?
272 data->state.aptr.passwd : "";
273 const size_t plen = strlen(passwd);
274
275 payloadlen = ulen + plen + MQTT_CLIENTID_LEN + 2;
276 /* The plus 2 are for the MSB and LSB describing the length of the string to
277 * be added on the payload. Refer to spec 1.5.2 and 1.5.4 */
278 if(ulen)
279 payloadlen += 2;
280 if(plen)
281 payloadlen += 2;
282
283 /* getting how much occupy the remain length */
284 remain_pos = mqtt_encode_len(remain, payloadlen + 10);
285
286 /* 10 length of variable header and 1 the first byte of the fixed header */
287 packetlen = payloadlen + 10 + remain_pos + 1;
288
289 /* allocating packet */
290 if(packetlen > 268435455)
291 return CURLE_WEIRD_SERVER_REPLY;
292 packet = malloc(packetlen);
293 if(!packet)
294 return CURLE_OUT_OF_MEMORY;
295 memset(packet, 0, packetlen);
296
297 /* set initial values for the CONNECT packet */
298 pos = init_connpack(packet, remain, remain_pos);
299
300 result = Curl_rand_alnum(data, (unsigned char *)&client_id[clen],
301 MQTT_CLIENTID_LEN - clen + 1);
302 /* add client id */
303 rc = add_client_id(client_id, strlen(client_id), packet, pos + 1);
304 if(rc) {
305 failf(data, "Client ID length mismatched: [%zu]", strlen(client_id));
306 result = CURLE_WEIRD_SERVER_REPLY;
307 goto end;
308 }
309 infof(data, "Using client id '%s'", client_id);
310
311 /* position where starts the user payload */
312 start_user = pos + 3 + MQTT_CLIENTID_LEN;
313 /* position where starts the password payload */
314 start_pwd = start_user + ulen;
315 /* if username was provided, add it to the packet */
316 if(ulen) {
317 start_pwd += 2;
318
319 rc = add_user(username, ulen,
320 (unsigned char *)packet, start_user, remain_pos);
321 if(rc) {
322 failf(data, "Username is too large: [%zu]", ulen);
323 result = CURLE_WEIRD_SERVER_REPLY;
324 goto end;
325 }
326 }
327
328 /* if passwd was provided, add it to the packet */
329 if(plen) {
330 rc = add_passwd(passwd, plen, packet, start_pwd, remain_pos);
331 if(rc) {
332 failf(data, "Password is too large: [%zu]", plen);
333 result = CURLE_WEIRD_SERVER_REPLY;
334 goto end;
335 }
336 }
337
338 if(!result)
339 result = mqtt_send(data, packet, packetlen);
340
341 end:
342 if(packet)
343 free(packet);
344 Curl_safefree(data->state.aptr.user);
345 Curl_safefree(data->state.aptr.passwd);
346 return result;
347 }
348
mqtt_disconnect(struct Curl_easy * data)349 static CURLcode mqtt_disconnect(struct Curl_easy *data)
350 {
351 CURLcode result = CURLE_OK;
352 struct MQTT *mq = data->req.p.mqtt;
353 result = mqtt_send(data, (char *)"\xe0\x00", 2);
354 Curl_safefree(mq->sendleftovers);
355 Curl_dyn_free(&mq->recvbuf);
356 return result;
357 }
358
mqtt_recv_atleast(struct Curl_easy * data,size_t nbytes)359 static CURLcode mqtt_recv_atleast(struct Curl_easy *data, size_t nbytes)
360 {
361 struct MQTT *mq = data->req.p.mqtt;
362 size_t rlen = Curl_dyn_len(&mq->recvbuf);
363 CURLcode result;
364
365 if(rlen < nbytes) {
366 unsigned char readbuf[1024];
367 ssize_t nread;
368
369 DEBUGASSERT(nbytes - rlen < sizeof(readbuf));
370 result = Curl_xfer_recv(data, (char *)readbuf, nbytes - rlen, &nread);
371 if(result)
372 return result;
373 DEBUGASSERT(nread >= 0);
374 if(Curl_dyn_addn(&mq->recvbuf, readbuf, (size_t)nread))
375 return CURLE_OUT_OF_MEMORY;
376 rlen = Curl_dyn_len(&mq->recvbuf);
377 }
378 return (rlen >= nbytes) ? CURLE_OK : CURLE_AGAIN;
379 }
380
mqtt_recv_consume(struct Curl_easy * data,size_t nbytes)381 static void mqtt_recv_consume(struct Curl_easy *data, size_t nbytes)
382 {
383 struct MQTT *mq = data->req.p.mqtt;
384 size_t rlen = Curl_dyn_len(&mq->recvbuf);
385 if(rlen <= nbytes)
386 Curl_dyn_reset(&mq->recvbuf);
387 else
388 Curl_dyn_tail(&mq->recvbuf, rlen - nbytes);
389 }
390
mqtt_verify_connack(struct Curl_easy * data)391 static CURLcode mqtt_verify_connack(struct Curl_easy *data)
392 {
393 struct MQTT *mq = data->req.p.mqtt;
394 CURLcode result;
395 char *ptr;
396
397 result = mqtt_recv_atleast(data, MQTT_CONNACK_LEN);
398 if(result)
399 goto fail;
400
401 /* verify CONNACK */
402 DEBUGASSERT(Curl_dyn_len(&mq->recvbuf) >= MQTT_CONNACK_LEN);
403 ptr = Curl_dyn_ptr(&mq->recvbuf);
404 Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_CONNACK_LEN);
405
406 if(ptr[0] != 0x00 || ptr[1] != 0x00) {
407 failf(data, "Expected %02x%02x but got %02x%02x",
408 0x00, 0x00, ptr[0], ptr[1]);
409 Curl_dyn_reset(&mq->recvbuf);
410 result = CURLE_WEIRD_SERVER_REPLY;
411 goto fail;
412 }
413 mqtt_recv_consume(data, MQTT_CONNACK_LEN);
414 fail:
415 return result;
416 }
417
mqtt_get_topic(struct Curl_easy * data,char ** topic,size_t * topiclen)418 static CURLcode mqtt_get_topic(struct Curl_easy *data,
419 char **topic, size_t *topiclen)
420 {
421 char *path = data->state.up.path;
422 CURLcode result = CURLE_URL_MALFORMAT;
423 if(strlen(path) > 1) {
424 result = Curl_urldecode(path + 1, 0, topic, topiclen, REJECT_NADA);
425 if(!result && (*topiclen > 0xffff)) {
426 failf(data, "Too long MQTT topic");
427 result = CURLE_URL_MALFORMAT;
428 }
429 }
430 else
431 failf(data, "No MQTT topic found. Forgot to URL encode it?");
432
433 return result;
434 }
435
mqtt_subscribe(struct Curl_easy * data)436 static CURLcode mqtt_subscribe(struct Curl_easy *data)
437 {
438 CURLcode result = CURLE_OK;
439 char *topic = NULL;
440 size_t topiclen;
441 unsigned char *packet = NULL;
442 size_t packetlen;
443 char encodedsize[4];
444 size_t n;
445 struct connectdata *conn = data->conn;
446
447 result = mqtt_get_topic(data, &topic, &topiclen);
448 if(result)
449 goto fail;
450
451 conn->proto.mqtt.packetid++;
452
453 packetlen = topiclen + 5; /* packetid + topic (has a two byte length field)
454 + 2 bytes topic length + QoS byte */
455 n = mqtt_encode_len((char *)encodedsize, packetlen);
456 packetlen += n + 1; /* add one for the control packet type byte */
457
458 packet = malloc(packetlen);
459 if(!packet) {
460 result = CURLE_OUT_OF_MEMORY;
461 goto fail;
462 }
463
464 packet[0] = MQTT_MSG_SUBSCRIBE;
465 memcpy(&packet[1], encodedsize, n);
466 packet[1 + n] = (conn->proto.mqtt.packetid >> 8) & 0xff;
467 packet[2 + n] = conn->proto.mqtt.packetid & 0xff;
468 packet[3 + n] = (topiclen >> 8) & 0xff;
469 packet[4 + n ] = topiclen & 0xff;
470 memcpy(&packet[5 + n], topic, topiclen);
471 packet[5 + n + topiclen] = 0; /* QoS zero */
472
473 result = mqtt_send(data, (char *)packet, packetlen);
474
475 fail:
476 free(topic);
477 free(packet);
478 return result;
479 }
480
481 /*
482 * Called when the first byte was already read.
483 */
mqtt_verify_suback(struct Curl_easy * data)484 static CURLcode mqtt_verify_suback(struct Curl_easy *data)
485 {
486 struct MQTT *mq = data->req.p.mqtt;
487 struct connectdata *conn = data->conn;
488 struct mqtt_conn *mqtt = &conn->proto.mqtt;
489 CURLcode result;
490 char *ptr;
491
492 result = mqtt_recv_atleast(data, MQTT_SUBACK_LEN);
493 if(result)
494 goto fail;
495
496 /* verify SUBACK */
497 DEBUGASSERT(Curl_dyn_len(&mq->recvbuf) >= MQTT_SUBACK_LEN);
498 ptr = Curl_dyn_ptr(&mq->recvbuf);
499 Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_SUBACK_LEN);
500
501 if(((unsigned char)ptr[0]) != ((mqtt->packetid >> 8) & 0xff) ||
502 ((unsigned char)ptr[1]) != (mqtt->packetid & 0xff) ||
503 ptr[2] != 0x00) {
504 Curl_dyn_reset(&mq->recvbuf);
505 result = CURLE_WEIRD_SERVER_REPLY;
506 goto fail;
507 }
508 mqtt_recv_consume(data, MQTT_SUBACK_LEN);
509 fail:
510 return result;
511 }
512
mqtt_publish(struct Curl_easy * data)513 static CURLcode mqtt_publish(struct Curl_easy *data)
514 {
515 CURLcode result;
516 char *payload = data->set.postfields;
517 size_t payloadlen;
518 char *topic = NULL;
519 size_t topiclen;
520 unsigned char *pkt = NULL;
521 size_t i = 0;
522 size_t remaininglength;
523 size_t encodelen;
524 char encodedbytes[4];
525 curl_off_t postfieldsize = data->set.postfieldsize;
526
527 if(!payload) {
528 DEBUGF(infof(data, "mqtt_publish without payload, return bad arg"));
529 return CURLE_BAD_FUNCTION_ARGUMENT;
530 }
531 if(postfieldsize < 0)
532 payloadlen = strlen(payload);
533 else
534 payloadlen = (size_t)postfieldsize;
535
536 result = mqtt_get_topic(data, &topic, &topiclen);
537 if(result)
538 goto fail;
539
540 remaininglength = payloadlen + 2 + topiclen;
541 encodelen = mqtt_encode_len(encodedbytes, remaininglength);
542
543 /* add the control byte and the encoded remaining length */
544 pkt = malloc(remaininglength + 1 + encodelen);
545 if(!pkt) {
546 result = CURLE_OUT_OF_MEMORY;
547 goto fail;
548 }
549
550 /* assemble packet */
551 pkt[i++] = MQTT_MSG_PUBLISH;
552 memcpy(&pkt[i], encodedbytes, encodelen);
553 i += encodelen;
554 pkt[i++] = (topiclen >> 8) & 0xff;
555 pkt[i++] = (topiclen & 0xff);
556 memcpy(&pkt[i], topic, topiclen);
557 i += topiclen;
558 memcpy(&pkt[i], payload, payloadlen);
559 i += payloadlen;
560 result = mqtt_send(data, (char *)pkt, i);
561
562 fail:
563 free(pkt);
564 free(topic);
565 return result;
566 }
567
mqtt_decode_len(unsigned char * buf,size_t buflen,size_t * lenbytes)568 static size_t mqtt_decode_len(unsigned char *buf,
569 size_t buflen, size_t *lenbytes)
570 {
571 size_t len = 0;
572 size_t mult = 1;
573 size_t i;
574 unsigned char encoded = 128;
575
576 for(i = 0; (i < buflen) && (encoded & 128); i++) {
577 encoded = buf[i];
578 len += (encoded & 127) * mult;
579 mult *= 128;
580 }
581
582 if(lenbytes)
583 *lenbytes = i;
584
585 return len;
586 }
587
588 #ifdef DEBUGBUILD
589 static const char *statenames[]={
590 "MQTT_FIRST",
591 "MQTT_REMAINING_LENGTH",
592 "MQTT_CONNACK",
593 "MQTT_SUBACK",
594 "MQTT_SUBACK_COMING",
595 "MQTT_PUBWAIT",
596 "MQTT_PUB_REMAIN",
597
598 "NOT A STATE"
599 };
600 #endif
601
602 /* The only way to change state */
mqstate(struct Curl_easy * data,enum mqttstate state,enum mqttstate nextstate)603 static void mqstate(struct Curl_easy *data,
604 enum mqttstate state,
605 enum mqttstate nextstate) /* used if state == FIRST */
606 {
607 struct connectdata *conn = data->conn;
608 struct mqtt_conn *mqtt = &conn->proto.mqtt;
609 #ifdef DEBUGBUILD
610 infof(data, "%s (from %s) (next is %s)",
611 statenames[state],
612 statenames[mqtt->state],
613 (state == MQTT_FIRST) ? statenames[nextstate] : "");
614 #endif
615 mqtt->state = state;
616 if(state == MQTT_FIRST)
617 mqtt->nextstate = nextstate;
618 }
619
620
mqtt_read_publish(struct Curl_easy * data,bool * done)621 static CURLcode mqtt_read_publish(struct Curl_easy *data, bool *done)
622 {
623 CURLcode result = CURLE_OK;
624 struct connectdata *conn = data->conn;
625 ssize_t nread;
626 size_t remlen;
627 struct mqtt_conn *mqtt = &conn->proto.mqtt;
628 struct MQTT *mq = data->req.p.mqtt;
629 unsigned char packet;
630
631 switch(mqtt->state) {
632 MQTT_SUBACK_COMING:
633 case MQTT_SUBACK_COMING:
634 result = mqtt_verify_suback(data);
635 if(result)
636 break;
637
638 mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
639 break;
640
641 case MQTT_SUBACK:
642 case MQTT_PUBWAIT:
643 /* we are expecting PUBLISH or SUBACK */
644 packet = mq->firstbyte & 0xf0;
645 if(packet == MQTT_MSG_PUBLISH)
646 mqstate(data, MQTT_PUB_REMAIN, MQTT_NOSTATE);
647 else if(packet == MQTT_MSG_SUBACK) {
648 mqstate(data, MQTT_SUBACK_COMING, MQTT_NOSTATE);
649 goto MQTT_SUBACK_COMING;
650 }
651 else if(packet == MQTT_MSG_DISCONNECT) {
652 infof(data, "Got DISCONNECT");
653 *done = TRUE;
654 goto end;
655 }
656 else {
657 result = CURLE_WEIRD_SERVER_REPLY;
658 goto end;
659 }
660
661 /* -- switched state -- */
662 remlen = mq->remaining_length;
663 infof(data, "Remaining length: %zu bytes", remlen);
664 if(data->set.max_filesize &&
665 (curl_off_t)remlen > data->set.max_filesize) {
666 failf(data, "Maximum file size exceeded");
667 result = CURLE_FILESIZE_EXCEEDED;
668 goto end;
669 }
670 Curl_pgrsSetDownloadSize(data, remlen);
671 data->req.bytecount = 0;
672 data->req.size = remlen;
673 mq->npacket = remlen; /* get this many bytes */
674 FALLTHROUGH();
675 case MQTT_PUB_REMAIN: {
676 /* read rest of packet, but no more. Cap to buffer size */
677 char buffer[4*1024];
678 size_t rest = mq->npacket;
679 if(rest > sizeof(buffer))
680 rest = sizeof(buffer);
681 result = Curl_xfer_recv(data, buffer, rest, &nread);
682 if(result) {
683 if(CURLE_AGAIN == result) {
684 infof(data, "EEEE AAAAGAIN");
685 }
686 goto end;
687 }
688 if(!nread) {
689 infof(data, "server disconnected");
690 result = CURLE_PARTIAL_FILE;
691 goto end;
692 }
693
694 /* if QoS is set, message contains packet id */
695 result = Curl_client_write(data, CLIENTWRITE_BODY, buffer, nread);
696 if(result)
697 goto end;
698
699 mq->npacket -= nread;
700 if(!mq->npacket)
701 /* no more PUBLISH payload, back to subscribe wait state */
702 mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
703 break;
704 }
705 default:
706 DEBUGASSERT(NULL); /* illegal state */
707 result = CURLE_WEIRD_SERVER_REPLY;
708 goto end;
709 }
710 end:
711 return result;
712 }
713
mqtt_do(struct Curl_easy * data,bool * done)714 static CURLcode mqtt_do(struct Curl_easy *data, bool *done)
715 {
716 CURLcode result = CURLE_OK;
717 *done = FALSE; /* unconditionally */
718
719 result = mqtt_connect(data);
720 if(result) {
721 failf(data, "Error %d sending MQTT CONNECT request", result);
722 return result;
723 }
724 mqstate(data, MQTT_FIRST, MQTT_CONNACK);
725 return CURLE_OK;
726 }
727
mqtt_done(struct Curl_easy * data,CURLcode status,bool premature)728 static CURLcode mqtt_done(struct Curl_easy *data,
729 CURLcode status, bool premature)
730 {
731 struct MQTT *mq = data->req.p.mqtt;
732 (void)status;
733 (void)premature;
734 Curl_safefree(mq->sendleftovers);
735 Curl_dyn_free(&mq->recvbuf);
736 return CURLE_OK;
737 }
738
mqtt_doing(struct Curl_easy * data,bool * done)739 static CURLcode mqtt_doing(struct Curl_easy *data, bool *done)
740 {
741 CURLcode result = CURLE_OK;
742 struct connectdata *conn = data->conn;
743 struct mqtt_conn *mqtt = &conn->proto.mqtt;
744 struct MQTT *mq = data->req.p.mqtt;
745 ssize_t nread;
746 unsigned char recvbyte;
747
748 *done = FALSE;
749
750 if(mq->nsend) {
751 /* send the remainder of an outgoing packet */
752 char *ptr = mq->sendleftovers;
753 result = mqtt_send(data, mq->sendleftovers, mq->nsend);
754 free(ptr);
755 if(result)
756 return result;
757 }
758
759 infof(data, "mqtt_doing: state [%d]", (int) mqtt->state);
760 switch(mqtt->state) {
761 case MQTT_FIRST:
762 /* Read the initial byte only */
763 result = Curl_xfer_recv(data, (char *)&mq->firstbyte, 1, &nread);
764 if(result)
765 break;
766 else if(!nread) {
767 failf(data, "Connection disconnected");
768 *done = TRUE;
769 result = CURLE_RECV_ERROR;
770 break;
771 }
772 Curl_debug(data, CURLINFO_HEADER_IN, (char *)&mq->firstbyte, 1);
773 /* remember the first byte */
774 mq->npacket = 0;
775 mqstate(data, MQTT_REMAINING_LENGTH, MQTT_NOSTATE);
776 FALLTHROUGH();
777 case MQTT_REMAINING_LENGTH:
778 do {
779 result = Curl_xfer_recv(data, (char *)&recvbyte, 1, &nread);
780 if(result || !nread)
781 break;
782 Curl_debug(data, CURLINFO_HEADER_IN, (char *)&recvbyte, 1);
783 mq->pkt_hd[mq->npacket++] = recvbyte;
784 } while((recvbyte & 0x80) && (mq->npacket < 4));
785 if(!result && nread && (recvbyte & 0x80))
786 /* MQTT supports up to 127 * 128^0 + 127 * 128^1 + 127 * 128^2 +
787 127 * 128^3 bytes. server tried to send more */
788 result = CURLE_WEIRD_SERVER_REPLY;
789 if(result)
790 break;
791 mq->remaining_length = mqtt_decode_len(mq->pkt_hd, mq->npacket, NULL);
792 mq->npacket = 0;
793 if(mq->remaining_length) {
794 mqstate(data, mqtt->nextstate, MQTT_NOSTATE);
795 break;
796 }
797 mqstate(data, MQTT_FIRST, MQTT_FIRST);
798
799 if(mq->firstbyte == MQTT_MSG_DISCONNECT) {
800 infof(data, "Got DISCONNECT");
801 *done = TRUE;
802 }
803 break;
804 case MQTT_CONNACK:
805 result = mqtt_verify_connack(data);
806 if(result)
807 break;
808
809 if(data->state.httpreq == HTTPREQ_POST) {
810 result = mqtt_publish(data);
811 if(!result) {
812 result = mqtt_disconnect(data);
813 *done = TRUE;
814 }
815 mqtt->nextstate = MQTT_FIRST;
816 }
817 else {
818 result = mqtt_subscribe(data);
819 if(!result) {
820 mqstate(data, MQTT_FIRST, MQTT_SUBACK);
821 }
822 }
823 break;
824
825 case MQTT_SUBACK:
826 case MQTT_PUBWAIT:
827 case MQTT_PUB_REMAIN:
828 result = mqtt_read_publish(data, done);
829 break;
830
831 default:
832 failf(data, "State not handled yet");
833 *done = TRUE;
834 break;
835 }
836
837 if(result == CURLE_AGAIN)
838 result = CURLE_OK;
839 return result;
840 }
841
842 #endif /* CURL_DISABLE_MQTT */
843
