Coverage Report

Created: 2024-05-04 12:45

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