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