luna - curl/lib/mqtt.c

   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#include "curl_setup.h"
  26#include "urldata.h"
  27
  28#ifndef CURL_DISABLE_MQTT
  29
  30#include "transfer.h"
  31#include "sendf.h"
  32#include "curl_trc.h"
  33#include "progress.h"
  34#include "mqtt.h"
  35#include "select.h"
  36#include "url.h"
  37#include "escape.h"
  38#include "rand.h"
  39#include "cfilters.h"
  40#include "connect.h"
  41
  42/* first byte is command.
  43   second byte is for flags. */
  44#define MQTT_MSG_CONNECT    0x10
  45/* #define MQTT_MSG_CONNACK    0x20 */
  46#define MQTT_MSG_PUBLISH    0x30
  47#define MQTT_MSG_SUBSCRIBE  0x82
  48#define MQTT_MSG_SUBACK     0x90
  49#define MQTT_MSG_DISCONNECT 0xe0
  50/* #define MQTT_MSG_PINGREQ    0xC0 */
  51#define MQTT_MSG_PINGRESP   0xD0
  52
  53#define MQTT_CONNACK_LEN  2
  54#define MQTT_SUBACK_LEN   3
  55#define MQTT_CLIENTID_LEN 12 /* "curl0123abcd" */
  56
  57/* meta key for storing protocol meta at easy handle */
  58#define CURL_META_MQTT_EASY   "meta:proto:mqtt:easy"
  59/* meta key for storing protocol meta at connection */
  60#define CURL_META_MQTT_CONN   "meta:proto:mqtt:conn"
  61
  62enum mqttstate {
  63  MQTT_FIRST,             /* 0 */
  64  MQTT_REMAINING_LENGTH,  /* 1 */
  65  MQTT_CONNACK,           /* 2 */
  66  MQTT_SUBACK,            /* 3 */
  67  MQTT_SUBACK_COMING,     /* 4 - the SUBACK remainder */
  68  MQTT_PUBWAIT,    /* 5 - wait for publish */
  69  MQTT_PUB_REMAIN,  /* 6 - wait for the remainder of the publish */
  70
  71  MQTT_NOSTATE /* 7 - never used an actual state */
  72};
  73
  74struct mqtt_conn {
  75  enum mqttstate state;
  76  enum mqttstate nextstate; /* switch to this after remaining length is
  77                               done */
  78  unsigned int packetid;
  79};
  80
  81/* protocol-specific transfer-related data */
  82struct MQTT {
  83  struct dynbuf sendbuf;
  84  /* when receiving */
  85  struct dynbuf recvbuf;
  86  size_t npacket; /* byte counter */
  87  size_t remaining_length;
  88  unsigned char pkt_hd[4]; /* for decoding the arriving packet length */
  89  struct curltime lastTime; /* last time we sent or received data */
  90  unsigned char firstbyte;
  91  BIT(pingsent); /* 1 while we wait for ping response */
  92};
  93
  94static void mqtt_easy_dtor(void *key, size_t klen, void *entry)
  95{
  96  struct MQTT *mq = entry;
  97  (void)key;
  98  (void)klen;
  99  curlx_dyn_free(&mq->sendbuf);
 100  curlx_dyn_free(&mq->recvbuf);
 101  curlx_free(mq);
 102}
 103
 104static void mqtt_conn_dtor(void *key, size_t klen, void *entry)
 105{
 106  (void)key;
 107  (void)klen;
 108  curlx_free(entry);
 109}
 110
 111static CURLcode mqtt_setup_conn(struct Curl_easy *data,
 112                                struct connectdata *conn)
 113{
 114  /* setup MQTT specific meta data at easy handle and connection */
 115  struct mqtt_conn *mqtt;
 116  struct MQTT *mq;
 117
 118  mqtt = curlx_calloc(1, sizeof(*mqtt));
 119  if(!mqtt ||
 120     Curl_conn_meta_set(conn, CURL_META_MQTT_CONN, mqtt, mqtt_conn_dtor))
 121    return CURLE_OUT_OF_MEMORY;
 122
 123  mq = curlx_calloc(1, sizeof(struct MQTT));
 124  if(!mq)
 125    return CURLE_OUT_OF_MEMORY;
 126  curlx_dyn_init(&mq->recvbuf, DYN_MQTT_RECV);
 127  curlx_dyn_init(&mq->sendbuf, DYN_MQTT_SEND);
 128  if(Curl_meta_set(data, CURL_META_MQTT_EASY, mq, mqtt_easy_dtor))
 129    return CURLE_OUT_OF_MEMORY;
 130  return CURLE_OK;
 131}
 132
 133static CURLcode mqtt_send(struct Curl_easy *data,
 134                          const char *buf, size_t len)
 135{
 136  size_t n;
 137  CURLcode result;
 138  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 139
 140  if(!mq)
 141    return CURLE_FAILED_INIT;
 142
 143  result = Curl_xfer_send(data, buf, len, FALSE, &n);
 144  if(result)
 145    return result;
 146  mq->lastTime = *Curl_pgrs_now(data);
 147  Curl_debug(data, CURLINFO_HEADER_OUT, buf, n);
 148  if(len != n) {
 149    size_t nsend = len - n;
 150    if(curlx_dyn_len(&mq->sendbuf)) {
 151      DEBUGASSERT(curlx_dyn_len(&mq->sendbuf) >= nsend);
 152      result = curlx_dyn_tail(&mq->sendbuf, nsend); /* keep this much */
 153    }
 154    else {
 155      result = curlx_dyn_addn(&mq->sendbuf, &buf[n], nsend);
 156    }
 157  }
 158  else
 159    curlx_dyn_reset(&mq->sendbuf);
 160  return result;
 161}
 162
 163/* Generic function called by the multi interface to figure out what socket(s)
 164   to wait for and for what actions during the DOING and PROTOCONNECT
 165   states */
 166static CURLcode mqtt_pollset(struct Curl_easy *data,
 167                             struct easy_pollset *ps)
 168{
 169  return Curl_pollset_add_in(data, ps, data->conn->sock[FIRSTSOCKET]);
 170}
 171
 172static int mqtt_encode_len(char *buf, size_t len)
 173{
 174  int i;
 175
 176  for(i = 0; (len > 0) && (i < 4); i++) {
 177    unsigned char encoded;
 178    encoded = len % 0x80;
 179    len /= 0x80;
 180    if(len)
 181      encoded |= 0x80;
 182    buf[i] = (char)encoded;
 183  }
 184
 185  return i;
 186}
 187
 188/* add the passwd to the CONNECT packet */
 189static int add_passwd(const char *passwd, const size_t plen,
 190                      char *pkt, const size_t start, int remain_pos)
 191{
 192  /* magic number that need to be set properly */
 193  const size_t conn_flags_pos = remain_pos + 8;
 194  if(plen > 0xffff)
 195    return 1;
 196
 197  /* set password flag */
 198  pkt[conn_flags_pos] |= 0x40;
 199
 200  /* length of password provided */
 201  pkt[start] = (char)((plen >> 8) & 0xFF);
 202  pkt[start + 1] = (char)(plen & 0xFF);
 203  memcpy(&pkt[start + 2], passwd, plen);
 204  return 0;
 205}
 206
 207/* add user to the CONNECT packet */
 208static int add_user(const char *username, const size_t ulen,
 209                    unsigned char *pkt, const size_t start, int remain_pos)
 210{
 211  /* magic number that need to be set properly */
 212  const size_t conn_flags_pos = remain_pos + 8;
 213  if(ulen > 0xffff)
 214    return 1;
 215
 216  /* set username flag */
 217  pkt[conn_flags_pos] |= 0x80;
 218  /* length of username provided */
 219  pkt[start] = (unsigned char)((ulen >> 8) & 0xFF);
 220  pkt[start + 1] = (unsigned char)(ulen & 0xFF);
 221  memcpy(&pkt[start + 2], username, ulen);
 222  return 0;
 223}
 224
 225/* add client ID to the CONNECT packet */
 226static int add_client_id(const char *client_id, const size_t client_id_len,
 227                         char *pkt, const size_t start)
 228{
 229  if(client_id_len != MQTT_CLIENTID_LEN)
 230    return 1;
 231  pkt[start] = 0x00;
 232  pkt[start + 1] = MQTT_CLIENTID_LEN;
 233  memcpy(&pkt[start + 2], client_id, MQTT_CLIENTID_LEN);
 234  return 0;
 235}
 236
 237/* Set initial values of CONNECT packet */
 238static int init_connpack(char *packet, char *remain, int remain_pos)
 239{
 240  /* Fixed header starts */
 241  /* packet type */
 242  packet[0] = MQTT_MSG_CONNECT;
 243  /* remaining length field */
 244  memcpy(&packet[1], remain, remain_pos);
 245  /* Fixed header ends */
 246
 247  /* Variable header starts */
 248  /* protocol length */
 249  packet[remain_pos + 1] = 0x00;
 250  packet[remain_pos + 2] = 0x04;
 251  /* protocol name */
 252  packet[remain_pos + 3] = 'M';
 253  packet[remain_pos + 4] = 'Q';
 254  packet[remain_pos + 5] = 'T';
 255  packet[remain_pos + 6] = 'T';
 256  /* protocol level */
 257  packet[remain_pos + 7] = 0x04;
 258  /* CONNECT flag: CleanSession */
 259  packet[remain_pos + 8] = 0x02;
 260  /* keep-alive 0 = disabled */
 261  packet[remain_pos + 9] = 0x00;
 262  packet[remain_pos + 10] = 0x3c;
 263  /* end of variable header */
 264  return remain_pos + 10;
 265}
 266
 267static CURLcode mqtt_connect(struct Curl_easy *data)
 268{
 269  CURLcode result = CURLE_OK;
 270  int pos = 0;
 271  int rc = 0;
 272  /* remain length */
 273  int remain_pos = 0;
 274  char remain[4] = { 0 };
 275  size_t packetlen = 0;
 276  size_t start_user = 0;
 277  size_t start_pwd = 0;
 278  char client_id[MQTT_CLIENTID_LEN + 1] = "curl";
 279  const size_t clen = strlen("curl");
 280  char *packet = NULL;
 281
 282  /* extracting username from request */
 283  struct Curl_creds *creds = data->state.creds;
 284  const size_t ulen = creds ? strlen(creds->user) : 0;
 285  const size_t plen = creds ? strlen(creds->passwd) : 0;
 286  const size_t payloadlen = ulen + plen + MQTT_CLIENTID_LEN + 2 +
 287  /* The plus 2s below are for the MSB and LSB describing the length of the
 288     string to be added on the payload. Refer to spec 1.5.2 and 1.5.4 */
 289    (ulen ? 2 : 0) +
 290    (plen ? 2 : 0);
 291
 292  /* getting how much occupy the remain length */
 293  remain_pos = mqtt_encode_len(remain, payloadlen + 10);
 294
 295  /* 10 length of variable header and 1 the first byte of the fixed header */
 296  packetlen = payloadlen + 10 + remain_pos + 1;
 297
 298  /* allocating packet */
 299  if(packetlen > 0xFFFFFFF)
 300    return CURLE_WEIRD_SERVER_REPLY;
 301  packet = curlx_calloc(1, packetlen);
 302  if(!packet)
 303    return CURLE_OUT_OF_MEMORY;
 304
 305  /* set initial values for the CONNECT packet */
 306  pos = init_connpack(packet, remain, remain_pos);
 307
 308  result = Curl_rand_alnum(data, (unsigned char *)&client_id[clen],
 309                           MQTT_CLIENTID_LEN - clen + 1);
 310  /* add client id */
 311  rc = add_client_id(client_id, strlen(client_id), packet, pos + 1);
 312  if(rc) {
 313    failf(data, "Client ID length mismatched: [%zu]", strlen(client_id));
 314    result = CURLE_WEIRD_SERVER_REPLY;
 315    goto end;
 316  }
 317  infof(data, "Using client id '%s'", client_id);
 318
 319  /* position where the user payload starts */
 320  start_user = pos + 3 + MQTT_CLIENTID_LEN;
 321  /* position where the password payload starts */
 322  start_pwd = start_user + ulen;
 323  /* if username was provided, add it to the packet */
 324  if(ulen) {
 325    start_pwd += 2;
 326
 327    rc = add_user(creds->user, ulen,
 328                  (unsigned char *)packet, start_user, remain_pos);
 329    if(rc) {
 330      failf(data, "Username too long: [%zu]", ulen);
 331      result = CURLE_WEIRD_SERVER_REPLY;
 332      goto end;
 333    }
 334  }
 335
 336  /* if passwd was provided, add it to the packet */
 337  if(plen) {
 338    rc = add_passwd(creds->passwd, plen, packet, start_pwd, remain_pos);
 339    if(rc) {
 340      failf(data, "Password too long: [%zu]", plen);
 341      result = CURLE_WEIRD_SERVER_REPLY;
 342      goto end;
 343    }
 344  }
 345
 346  if(!result)
 347    result = mqtt_send(data, packet, packetlen);
 348
 349end:
 350  if(packet)
 351    curlx_free(packet);
 352  Curl_creds_unlink(&data->state.creds);
 353  return result;
 354}
 355
 356static CURLcode mqtt_disconnect(struct Curl_easy *data)
 357{
 358  return mqtt_send(data, "\xe0\x00", 2);
 359}
 360
 361static CURLcode mqtt_recv_atleast(struct Curl_easy *data, size_t nbytes)
 362{
 363  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 364  size_t rlen;
 365  CURLcode result;
 366
 367  if(!mq)
 368    return CURLE_FAILED_INIT;
 369  rlen = curlx_dyn_len(&mq->recvbuf);
 370
 371  if(rlen < nbytes) {
 372    unsigned char readbuf[1024];
 373    size_t nread;
 374
 375    DEBUGASSERT(nbytes - rlen < sizeof(readbuf));
 376    result = Curl_xfer_recv(data, (char *)readbuf, nbytes - rlen, &nread);
 377    if(result)
 378      return result;
 379    if(!nread) /* EOF */
 380       return CURLE_RECV_ERROR;
 381    if(curlx_dyn_addn(&mq->recvbuf, readbuf, nread))
 382      return CURLE_OUT_OF_MEMORY;
 383    rlen = curlx_dyn_len(&mq->recvbuf);
 384  }
 385  return (rlen >= nbytes) ? CURLE_OK : CURLE_AGAIN;
 386}
 387
 388static void mqtt_recv_consume(struct Curl_easy *data, size_t nbytes)
 389{
 390  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 391  DEBUGASSERT(mq);
 392  if(mq) {
 393    size_t rlen = curlx_dyn_len(&mq->recvbuf);
 394    if(rlen <= nbytes)
 395      curlx_dyn_reset(&mq->recvbuf);
 396    else
 397      curlx_dyn_tail(&mq->recvbuf, rlen - nbytes);
 398  }
 399}
 400
 401static CURLcode mqtt_verify_connack(struct Curl_easy *data)
 402{
 403  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 404  CURLcode result;
 405  const char *ptr;
 406
 407  DEBUGASSERT(mq);
 408  if(!mq)
 409    return CURLE_FAILED_INIT;
 410  if(mq->remaining_length != 2) {
 411    failf(data, "CONNACK expected Remaining Length 2, got %zu",
 412          mq->remaining_length);
 413    return CURLE_WEIRD_SERVER_REPLY;
 414  }
 415
 416  result = mqtt_recv_atleast(data, MQTT_CONNACK_LEN);
 417  if(result)
 418    return result;
 419
 420  /* verify CONNACK */
 421  DEBUGASSERT(curlx_dyn_len(&mq->recvbuf) >= MQTT_CONNACK_LEN);
 422  ptr = curlx_dyn_ptr(&mq->recvbuf);
 423  Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_CONNACK_LEN);
 424
 425  if(ptr[0] != 0x00 || ptr[1] != 0x00) {
 426    failf(data, "Expected %02x%02x but got %02x%02x",
 427          0x00, 0x00, ptr[0], ptr[1]);
 428    curlx_dyn_reset(&mq->recvbuf);
 429    return CURLE_WEIRD_SERVER_REPLY;
 430  }
 431  mqtt_recv_consume(data, MQTT_CONNACK_LEN);
 432  return CURLE_OK;
 433}
 434
 435static CURLcode mqtt_get_topic(struct Curl_easy *data,
 436                               char **topic, size_t *topiclen)
 437{
 438  const char *path = data->state.up.path;
 439  CURLcode result = CURLE_URL_MALFORMAT;
 440  if(strlen(path) > 1) {
 441    result = Curl_urldecode(path + 1, 0, topic, topiclen, REJECT_NADA);
 442    if(!result && (*topiclen > 0xffff)) {
 443      failf(data, "Too long MQTT topic");
 444      result = CURLE_URL_MALFORMAT;
 445    }
 446  }
 447  else
 448    failf(data, "No MQTT topic found. Forgot to URL encode it?");
 449
 450  return result;
 451}
 452
 453static CURLcode mqtt_subscribe(struct Curl_easy *data)
 454{
 455  CURLcode result = CURLE_OK;
 456  char *topic = NULL;
 457  size_t topiclen;
 458  unsigned char *packet = NULL;
 459  size_t packetlen;
 460  char encodedsize[4];
 461  size_t n;
 462  struct connectdata *conn = data->conn;
 463  struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
 464
 465  if(!mqtt)
 466    return CURLE_FAILED_INIT;
 467
 468  result = mqtt_get_topic(data, &topic, &topiclen);
 469  if(result)
 470    goto fail;
 471
 472  mqtt->packetid++;
 473
 474  packetlen = topiclen + 5; /* packetid + topic (has a two byte length field)
 475                               + 2 bytes topic length + QoS byte */
 476  n = mqtt_encode_len((char *)encodedsize, packetlen);
 477  packetlen += n + 1; /* add one for the control packet type byte */
 478
 479  packet = curlx_malloc(packetlen);
 480  if(!packet) {
 481    result = CURLE_OUT_OF_MEMORY;
 482    goto fail;
 483  }
 484
 485  packet[0] = MQTT_MSG_SUBSCRIBE;
 486  memcpy(&packet[1], encodedsize, n);
 487  packet[1 + n] = (mqtt->packetid >> 8) & 0xff;
 488  packet[2 + n] = mqtt->packetid & 0xff;
 489  packet[3 + n] = (topiclen >> 8) & 0xff;
 490  packet[4 + n] = topiclen & 0xff;
 491  memcpy(&packet[5 + n], topic, topiclen);
 492  packet[5 + n + topiclen] = 0; /* QoS zero */
 493
 494  result = mqtt_send(data, (const char *)packet, packetlen);
 495
 496fail:
 497  curlx_free(topic);
 498  curlx_free(packet);
 499  return result;
 500}
 501
 502/*
 503 * Called when the first byte was already read.
 504 */
 505static CURLcode mqtt_verify_suback(struct Curl_easy *data)
 506{
 507  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 508  struct connectdata *conn = data->conn;
 509  struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
 510  CURLcode result;
 511  const char *ptr;
 512
 513  if(!mqtt || !mq)
 514    return CURLE_FAILED_INIT;
 515
 516  if(mq->remaining_length != 3) {
 517    failf(data, "SUBACK expected Remaining Length 3, got %zu",
 518          mq->remaining_length);
 519    return CURLE_WEIRD_SERVER_REPLY;
 520  }
 521
 522  result = mqtt_recv_atleast(data, MQTT_SUBACK_LEN);
 523  if(result)
 524    goto fail;
 525
 526  /* verify SUBACK */
 527  DEBUGASSERT(curlx_dyn_len(&mq->recvbuf) >= MQTT_SUBACK_LEN);
 528  ptr = curlx_dyn_ptr(&mq->recvbuf);
 529  Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_SUBACK_LEN);
 530
 531  if(((unsigned char)ptr[0]) != ((mqtt->packetid >> 8) & 0xff) ||
 532     ((unsigned char)ptr[1]) != (mqtt->packetid & 0xff) ||
 533     ptr[2] != 0x00) {
 534    curlx_dyn_reset(&mq->recvbuf);
 535    result = CURLE_WEIRD_SERVER_REPLY;
 536    goto fail;
 537  }
 538  mqtt_recv_consume(data, MQTT_SUBACK_LEN);
 539fail:
 540  return result;
 541}
 542
 543#define MAX_MQTT_MESSAGE_SIZE 0xFFFFFFF
 544
 545static CURLcode mqtt_publish(struct Curl_easy *data)
 546{
 547  CURLcode result;
 548  char *payload = data->set.postfields;
 549  size_t payloadlen;
 550  char *topic = NULL;
 551  size_t topiclen;
 552  unsigned char *pkt = NULL;
 553  size_t i = 0;
 554  size_t remaininglength;
 555  size_t encodelen;
 556  char encodedbytes[4];
 557  curl_off_t postfieldsize = data->set.postfieldsize;
 558
 559  if(!payload) {
 560    DEBUGF(infof(data, "mqtt_publish without payload, return bad arg"));
 561    return CURLE_BAD_FUNCTION_ARGUMENT;
 562  }
 563  if(!curlx_sotouz_fits(postfieldsize, &payloadlen)) {
 564    if(postfieldsize > 0) /* off_t does not fit into size_t */
 565      return CURLE_BAD_FUNCTION_ARGUMENT;
 566    payloadlen = strlen(payload);
 567  }
 568
 569  result = mqtt_get_topic(data, &topic, &topiclen);
 570  if(result)
 571    goto fail;
 572
 573  remaininglength = payloadlen + 2 + topiclen;
 574  encodelen = mqtt_encode_len(encodedbytes, remaininglength);
 575  if(remaininglength > (MAX_MQTT_MESSAGE_SIZE - encodelen - 1)) {
 576    result = CURLE_TOO_LARGE;
 577    goto fail;
 578  }
 579
 580  /* add the control byte and the encoded remaining length */
 581  pkt = curlx_malloc(remaininglength + 1 + encodelen);
 582  if(!pkt) {
 583    result = CURLE_OUT_OF_MEMORY;
 584    goto fail;
 585  }
 586
 587  /* assemble packet */
 588  pkt[i++] = MQTT_MSG_PUBLISH;
 589  memcpy(&pkt[i], encodedbytes, encodelen);
 590  i += encodelen;
 591  pkt[i++] = (topiclen >> 8) & 0xff;
 592  pkt[i++] = (topiclen & 0xff);
 593  memcpy(&pkt[i], topic, topiclen);
 594  i += topiclen;
 595  memcpy(&pkt[i], payload, payloadlen);
 596  i += payloadlen;
 597  result = mqtt_send(data, (const char *)pkt, i);
 598
 599fail:
 600  curlx_free(pkt);
 601  curlx_free(topic);
 602  return result;
 603}
 604
 605/* return 0 on success, non-zero on error */
 606static int mqtt_decode_len(size_t *lenp, const unsigned char *buf,
 607                           size_t buflen)
 608{
 609  size_t len = 0;
 610  size_t mult = 1;
 611  size_t i;
 612  unsigned char encoded = 128;
 613
 614  for(i = 0; (i < buflen) && (encoded & 128); i++) {
 615    if(i == 4)
 616      return 1; /* bad size */
 617    encoded = buf[i];
 618    len += (encoded & 127) * mult;
 619    mult *= 128;
 620  }
 621
 622  *lenp = len;
 623  return 0;
 624}
 625
 626#if defined(DEBUGBUILD) && defined(CURLVERBOSE)
 627static const char *statenames[] = {
 628  "MQTT_FIRST",
 629  "MQTT_REMAINING_LENGTH",
 630  "MQTT_CONNACK",
 631  "MQTT_SUBACK",
 632  "MQTT_SUBACK_COMING",
 633  "MQTT_PUBWAIT",
 634  "MQTT_PUB_REMAIN",
 635
 636  "NOT A STATE"
 637};
 638#endif
 639
 640/* The only way to change state */
 641static void mqstate(struct Curl_easy *data,
 642                    enum mqttstate state,
 643                    enum mqttstate nextstate) /* used if state == FIRST */
 644{
 645  struct connectdata *conn = data->conn;
 646  struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
 647  DEBUGASSERT(mqtt);
 648  if(!mqtt)
 649    return;
 650#ifdef DEBUGBUILD
 651  infof(data, "%s (from %s) (next is %s)",
 652        statenames[state],
 653        statenames[mqtt->state],
 654        (state == MQTT_FIRST) ? statenames[nextstate] : "");
 655#endif
 656  mqtt->state = state;
 657  if(state == MQTT_FIRST)
 658    mqtt->nextstate = nextstate;
 659}
 660
 661static CURLcode mqtt_read_publish(struct Curl_easy *data, bool *done)
 662{
 663  CURLcode result = CURLE_OK;
 664  struct connectdata *conn = data->conn;
 665  size_t nread;
 666  size_t remlen;
 667  struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
 668  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 669  unsigned char packet;
 670
 671  DEBUGASSERT(mqtt);
 672  if(!mqtt || !mq)
 673    return CURLE_FAILED_INIT;
 674
 675  switch(mqtt->state) {
 676MQTT_SUBACK_COMING:
 677  case MQTT_SUBACK_COMING:
 678    result = mqtt_verify_suback(data);
 679    if(result)
 680      break;
 681
 682    mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
 683    break;
 684
 685  case MQTT_SUBACK:
 686  case MQTT_PUBWAIT:
 687    /* we are expecting PUBLISH or SUBACK */
 688    packet = mq->firstbyte & 0xf0;
 689    if(packet == MQTT_MSG_PUBLISH)
 690      mqstate(data, MQTT_PUB_REMAIN, MQTT_NOSTATE);
 691    else if(packet == MQTT_MSG_SUBACK) {
 692      mqstate(data, MQTT_SUBACK_COMING, MQTT_NOSTATE);
 693      goto MQTT_SUBACK_COMING;
 694    }
 695    else if(packet == MQTT_MSG_DISCONNECT) {
 696      infof(data, "Got DISCONNECT");
 697      *done = TRUE;
 698      goto end;
 699    }
 700    else {
 701      result = CURLE_WEIRD_SERVER_REPLY;
 702      goto end;
 703    }
 704
 705    /* -- switched state -- */
 706    remlen = mq->remaining_length;
 707    infof(data, "Remaining length: %zu bytes", remlen);
 708    if(data->set.max_filesize &&
 709       (curl_off_t)remlen > data->set.max_filesize) {
 710      failf(data, "Maximum file size exceeded");
 711      result = CURLE_FILESIZE_EXCEEDED;
 712      goto end;
 713    }
 714    Curl_pgrsSetDownloadSize(data, remlen);
 715    data->req.bytecount = 0;
 716    data->req.size = remlen;
 717    mq->npacket = remlen; /* get this many bytes */
 718    FALLTHROUGH();
 719  case MQTT_PUB_REMAIN: {
 720    /* read rest of packet, but no more. Cap to buffer size */
 721    char buffer[4 * 1024];
 722    size_t rest = mq->npacket;
 723    if(rest > sizeof(buffer))
 724      rest = sizeof(buffer);
 725    result = Curl_xfer_recv(data, buffer, rest, &nread);
 726    if(result) {
 727      if(result == CURLE_AGAIN) {
 728        infof(data, "EEEE AAAAGAIN");
 729      }
 730      goto end;
 731    }
 732    if(!nread) {
 733      infof(data, "server disconnected");
 734      result = CURLE_PARTIAL_FILE;
 735      goto end;
 736    }
 737
 738    /* we received something */
 739    mq->lastTime = *Curl_pgrs_now(data);
 740
 741    /* if QoS is set, message contains packet id */
 742    result = Curl_client_write(data, CLIENTWRITE_BODY, buffer, nread);
 743    if(result)
 744      goto end;
 745
 746    mq->npacket -= nread;
 747    if(!mq->npacket)
 748      /* no more PUBLISH payload, back to subscribe wait state */
 749      mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
 750    break;
 751  }
 752  default:
 753    DEBUGASSERT(NULL); /* illegal state */
 754    result = CURLE_WEIRD_SERVER_REPLY;
 755    goto end;
 756  }
 757end:
 758  return result;
 759}
 760
 761static CURLcode mqtt_do(struct Curl_easy *data, bool *done)
 762{
 763  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 764  CURLcode result = CURLE_OK;
 765  *done = FALSE; /* unconditionally */
 766
 767  if(!mq)
 768    return CURLE_FAILED_INIT;
 769  mq->lastTime = *Curl_pgrs_now(data);
 770  mq->pingsent = FALSE;
 771
 772  result = mqtt_connect(data);
 773  if(result) {
 774    failf(data, "Error %d sending MQTT CONNECT request", result);
 775    return result;
 776  }
 777  mqstate(data, MQTT_FIRST, MQTT_CONNACK);
 778  return CURLE_OK;
 779}
 780
 781static CURLcode mqtt_done(struct Curl_easy *data,
 782                          CURLcode status, bool premature)
 783{
 784  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 785  (void)status;
 786  (void)premature;
 787  if(mq) {
 788    curlx_dyn_free(&mq->sendbuf);
 789    curlx_dyn_free(&mq->recvbuf);
 790  }
 791  return CURLE_OK;
 792}
 793
 794/* we ping regularly to avoid being disconnected by the server */
 795static CURLcode mqtt_ping(struct Curl_easy *data)
 796{
 797  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 798  CURLcode result = CURLE_OK;
 799  struct connectdata *conn = data->conn;
 800  struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
 801
 802  if(!mqtt || !mq)
 803    return CURLE_FAILED_INIT;
 804
 805  if(mqtt->state == MQTT_FIRST &&
 806     !mq->pingsent &&
 807     data->set.upkeep_interval_ms > 0) {
 808    struct curltime t = *Curl_pgrs_now(data);
 809    timediff_t diff = curlx_ptimediff_ms(&t, &mq->lastTime);
 810
 811    if(diff > data->set.upkeep_interval_ms) {
 812      /* 0xC0 is PINGREQ, and 0x00 is remaining length */
 813      unsigned char packet[2] = { 0xC0, 0x00 };
 814      size_t packetlen = sizeof(packet);
 815
 816      result = mqtt_send(data, (char *)packet, packetlen);
 817      if(!result) {
 818        mq->pingsent = TRUE;
 819      }
 820      infof(data, "mqtt_ping: sent ping request.");
 821    }
 822  }
 823  return result;
 824}
 825
 826static CURLcode mqtt_doing(struct Curl_easy *data, bool *done)
 827{
 828  struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
 829  CURLcode result = CURLE_OK;
 830  size_t nread;
 831  unsigned char recvbyte;
 832  struct mqtt_conn *mqtt = Curl_conn_meta_get(data->conn, CURL_META_MQTT_CONN);
 833
 834  if(!mqtt || !mq)
 835    return CURLE_FAILED_INIT;
 836
 837  *done = FALSE;
 838
 839  if(curlx_dyn_len(&mq->sendbuf)) {
 840    /* send the remainder of an outgoing packet */
 841    result = mqtt_send(data, curlx_dyn_ptr(&mq->sendbuf),
 842                       curlx_dyn_len(&mq->sendbuf));
 843    if(result)
 844      return result;
 845  }
 846
 847  result = mqtt_ping(data);
 848  if(result)
 849    return result;
 850
 851  infof(data, "mqtt_doing: state [%d]", (int)mqtt->state);
 852  switch(mqtt->state) {
 853  case MQTT_FIRST:
 854    /* Read the initial byte only */
 855    result = Curl_xfer_recv(data, (char *)&mq->firstbyte, 1, &nread);
 856    if(result)
 857      break;
 858    else if(!nread) {
 859      failf(data, "Connection disconnected");
 860      *done = TRUE;
 861      result = CURLE_RECV_ERROR;
 862      break;
 863    }
 864    Curl_debug(data, CURLINFO_HEADER_IN, (const char *)&mq->firstbyte, 1);
 865
 866    /* we received something */
 867    mq->lastTime = *Curl_pgrs_now(data);
 868
 869    /* remember the first byte */
 870    mq->npacket = 0;
 871    mqstate(data, MQTT_REMAINING_LENGTH, MQTT_NOSTATE);
 872    FALLTHROUGH();
 873  case MQTT_REMAINING_LENGTH:
 874    do {
 875      result = Curl_xfer_recv(data, (char *)&recvbyte, 1, &nread);
 876      if(result || !nread)
 877        break;
 878      Curl_debug(data, CURLINFO_HEADER_IN, (const char *)&recvbyte, 1);
 879      mq->pkt_hd[mq->npacket++] = recvbyte;
 880    } while((recvbyte & 0x80) && (mq->npacket < 4));
 881    if(!result && nread && (recvbyte & 0x80))
 882      /* MQTT supports up to 127 * 128^0 + 127 * 128^1 + 127 * 128^2 +
 883         127 * 128^3 bytes. server tried to send more */
 884      result = CURLE_WEIRD_SERVER_REPLY;
 885    if(result)
 886      break;
 887    if(mqtt_decode_len(&mq->remaining_length, mq->pkt_hd, mq->npacket)) {
 888      result = CURLE_WEIRD_SERVER_REPLY;
 889      break;
 890    }
 891    mq->npacket = 0;
 892    /* PINGRESP and DISCONNECT must have remaining_length == 0 and
 893     * reserved bits (low nibble) must be zero per MQTT 3.1.1
 894     * sections 2.2.2, 3.13.1 and 3.14.1. Reject before state
 895     * dispatch to prevent nextstate confusion. */
 896    {
 897      const unsigned char type = mq->firstbyte & 0xF0;
 898      const unsigned char reserved = mq->firstbyte & 0x0F;
 899      if((type == MQTT_MSG_DISCONNECT || type == MQTT_MSG_PINGRESP) &&
 900         (mq->remaining_length || reserved)) {
 901        failf(data,
 902              "Broker sent malformed %s "
 903              "(remaining_length=%zu, header byte=0x%02x)",
 904              type == MQTT_MSG_DISCONNECT ? "DISCONNECT" : "PINGRESP",
 905              mq->remaining_length, mq->firstbyte);
 906        result = CURLE_WEIRD_SERVER_REPLY;
 907        break;
 908      }
 909    }
 910    if(mq->remaining_length) {
 911      mqstate(data, mqtt->nextstate, MQTT_NOSTATE);
 912      break;
 913    }
 914    mqstate(data, MQTT_FIRST, MQTT_FIRST);
 915
 916    if(mq->firstbyte == MQTT_MSG_DISCONNECT) {
 917      infof(data, "Got DISCONNECT");
 918      *done = TRUE;
 919    }
 920
 921    /* ping response */
 922    if(mq->firstbyte == MQTT_MSG_PINGRESP) {
 923      infof(data, "Received ping response.");
 924      mq->pingsent = FALSE;
 925      mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
 926    }
 927    break;
 928  case MQTT_CONNACK:
 929    result = mqtt_verify_connack(data);
 930    if(result)
 931      break;
 932
 933    if(data->state.httpreq == HTTPREQ_POST) {
 934      result = mqtt_publish(data);
 935      if(!result) {
 936        result = mqtt_disconnect(data);
 937        *done = TRUE;
 938      }
 939      mqtt->nextstate = MQTT_FIRST;
 940    }
 941    else {
 942      result = mqtt_subscribe(data);
 943      if(!result) {
 944        mqstate(data, MQTT_FIRST, MQTT_SUBACK);
 945      }
 946    }
 947    break;
 948
 949  case MQTT_SUBACK:
 950  case MQTT_PUBWAIT:
 951  case MQTT_PUB_REMAIN:
 952    result = mqtt_read_publish(data, done);
 953    break;
 954
 955  default:
 956    failf(data, "State not handled yet");
 957    *done = TRUE;
 958    break;
 959  }
 960
 961  if(result == CURLE_AGAIN)
 962    result = CURLE_OK;
 963  return result;
 964}
 965
 966#ifdef USE_SSL
 967
 968static CURLcode mqtts_connecting(struct Curl_easy *data, bool *done)
 969{
 970  struct connectdata *conn = data->conn;
 971  CURLcode result;
 972
 973  result = Curl_conn_connect(data, FIRSTSOCKET, TRUE, done);
 974  if(result)
 975    connclose(conn, "Failed TLS connection");
 976  return result;
 977}
 978
 979/*
 980 * MQTTS protocol.
 981 */
 982const struct Curl_protocol Curl_protocol_mqtts = {
 983  mqtt_setup_conn,                    /* setup_connection */
 984  mqtt_do,                            /* do_it */
 985  mqtt_done,                          /* done */
 986  ZERO_NULL,                          /* do_more */
 987  ZERO_NULL,                          /* connect_it */
 988  mqtts_connecting,                   /* connecting */
 989  mqtt_doing,                         /* doing */
 990  ZERO_NULL,                          /* proto_pollset */
 991  mqtt_pollset,                       /* doing_pollset */
 992  ZERO_NULL,                          /* domore_pollset */
 993  ZERO_NULL,                          /* perform_pollset */
 994  ZERO_NULL,                          /* disconnect */
 995  ZERO_NULL,                          /* write_resp */
 996  ZERO_NULL,                          /* write_resp_hd */
 997  ZERO_NULL,                          /* connection_is_dead */
 998  ZERO_NULL,                          /* attach connection */
 999  ZERO_NULL,                          /* follow */
1000};
1001
1002#endif
1003
1004/*
1005 * MQTT protocol.
1006 */
1007const struct Curl_protocol Curl_protocol_mqtt = {
1008  mqtt_setup_conn,                    /* setup_connection */
1009  mqtt_do,                            /* do_it */
1010  mqtt_done,                          /* done */
1011  ZERO_NULL,                          /* do_more */
1012  ZERO_NULL,                          /* connect_it */
1013  ZERO_NULL,                          /* connecting */
1014  mqtt_doing,                         /* doing */
1015  ZERO_NULL,                          /* proto_pollset */
1016  mqtt_pollset,                       /* doing_pollset */
1017  ZERO_NULL,                          /* domore_pollset */
1018  ZERO_NULL,                          /* perform_pollset */
1019  ZERO_NULL,                          /* disconnect */
1020  ZERO_NULL,                          /* write_resp */
1021  ZERO_NULL,                          /* write_resp_hd */
1022  ZERO_NULL,                          /* connection_is_dead */
1023  ZERO_NULL,                          /* attach connection */
1024  ZERO_NULL,                          /* follow */
1025};
1026
1027#endif /* CURL_DISABLE_MQTT */