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