xref: /curl/lib/mqtt.c (revision fbf5d507)
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