mqtt.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823
  1. /***************************************************************************
  2. * _ _ ____ _
  3. * Project ___| | | | _ \| |
  4. * / __| | | | |_) | |
  5. * | (__| |_| | _ <| |___
  6. * \___|\___/|_| \_\_____|
  7. *
  8. * Copyright (C) 2020 - 2022, Daniel Stenberg, <daniel@haxx.se>, et al.
  9. * Copyright (C) 2019, 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. #ifndef CURL_DISABLE_MQTT
  27. #include "urldata.h"
  28. #include <curl/curl.h>
  29. #include "transfer.h"
  30. #include "sendf.h"
  31. #include "progress.h"
  32. #include "mqtt.h"
  33. #include "select.h"
  34. #include "strdup.h"
  35. #include "url.h"
  36. #include "escape.h"
  37. #include "warnless.h"
  38. #include "curl_printf.h"
  39. #include "curl_memory.h"
  40. #include "multiif.h"
  41. #include "rand.h"
  42. /* The last #include file should be: */
  43. #include "memdebug.h"
  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_CONNACK_LEN 2
  51. #define MQTT_SUBACK_LEN 3
  52. #define MQTT_CLIENTID_LEN 12 /* "curl0123abcd" */
  53. /*
  54. * Forward declarations.
  55. */
  56. static CURLcode mqtt_do(struct Curl_easy *data, bool *done);
  57. static CURLcode mqtt_done(struct Curl_easy *data,
  58. CURLcode status, bool premature);
  59. static CURLcode mqtt_doing(struct Curl_easy *data, bool *done);
  60. static int mqtt_getsock(struct Curl_easy *data, struct connectdata *conn,
  61. curl_socket_t *sock);
  62. static CURLcode mqtt_setup_conn(struct Curl_easy *data,
  63. struct connectdata *conn);
  64. /*
  65. * MQTT protocol handler.
  66. */
  67. const struct Curl_handler Curl_handler_mqtt = {
  68. "MQTT", /* scheme */
  69. mqtt_setup_conn, /* setup_connection */
  70. mqtt_do, /* do_it */
  71. mqtt_done, /* done */
  72. ZERO_NULL, /* do_more */
  73. ZERO_NULL, /* connect_it */
  74. ZERO_NULL, /* connecting */
  75. mqtt_doing, /* doing */
  76. ZERO_NULL, /* proto_getsock */
  77. mqtt_getsock, /* doing_getsock */
  78. ZERO_NULL, /* domore_getsock */
  79. ZERO_NULL, /* perform_getsock */
  80. ZERO_NULL, /* disconnect */
  81. ZERO_NULL, /* readwrite */
  82. ZERO_NULL, /* connection_check */
  83. ZERO_NULL, /* attach connection */
  84. PORT_MQTT, /* defport */
  85. CURLPROTO_MQTT, /* protocol */
  86. CURLPROTO_MQTT, /* family */
  87. PROTOPT_NONE /* flags */
  88. };
  89. static CURLcode mqtt_setup_conn(struct Curl_easy *data,
  90. struct connectdata *conn)
  91. {
  92. /* allocate the HTTP-specific struct for the Curl_easy, only to survive
  93. during this request */
  94. struct MQTT *mq;
  95. (void)conn;
  96. DEBUGASSERT(data->req.p.mqtt == NULL);
  97. mq = calloc(1, sizeof(struct MQTT));
  98. if(!mq)
  99. return CURLE_OUT_OF_MEMORY;
  100. data->req.p.mqtt = mq;
  101. return CURLE_OK;
  102. }
  103. static CURLcode mqtt_send(struct Curl_easy *data,
  104. char *buf, size_t len)
  105. {
  106. CURLcode result = CURLE_OK;
  107. struct connectdata *conn = data->conn;
  108. curl_socket_t sockfd = conn->sock[FIRSTSOCKET];
  109. struct MQTT *mq = data->req.p.mqtt;
  110. ssize_t n;
  111. result = Curl_write(data, sockfd, buf, len, &n);
  112. if(!result)
  113. Curl_debug(data, CURLINFO_HEADER_OUT, buf, (size_t)n);
  114. if(len != (size_t)n) {
  115. size_t nsend = len - n;
  116. char *sendleftovers = Curl_memdup(&buf[n], nsend);
  117. if(!sendleftovers)
  118. return CURLE_OUT_OF_MEMORY;
  119. mq->sendleftovers = sendleftovers;
  120. mq->nsend = nsend;
  121. }
  122. else {
  123. mq->sendleftovers = NULL;
  124. mq->nsend = 0;
  125. }
  126. return result;
  127. }
  128. /* Generic function called by the multi interface to figure out what socket(s)
  129. to wait for and for what actions during the DOING and PROTOCONNECT
  130. states */
  131. static int mqtt_getsock(struct Curl_easy *data,
  132. struct connectdata *conn,
  133. curl_socket_t *sock)
  134. {
  135. (void)data;
  136. sock[0] = conn->sock[FIRSTSOCKET];
  137. return GETSOCK_READSOCK(FIRSTSOCKET);
  138. }
  139. static int mqtt_encode_len(char *buf, size_t len)
  140. {
  141. unsigned char encoded;
  142. int i;
  143. for(i = 0; (len > 0) && (i<4); i++) {
  144. encoded = len % 0x80;
  145. len /= 0x80;
  146. if(len)
  147. encoded |= 0x80;
  148. buf[i] = encoded;
  149. }
  150. return i;
  151. }
  152. /* add the passwd to the CONNECT packet */
  153. static int add_passwd(const char *passwd, const size_t plen,
  154. char *pkt, const size_t start, int remain_pos)
  155. {
  156. /* magic number that need to be set properly */
  157. const size_t conn_flags_pos = remain_pos + 8;
  158. if(plen > 0xffff)
  159. return 1;
  160. /* set password flag */
  161. pkt[conn_flags_pos] |= 0x40;
  162. /* length of password provided */
  163. pkt[start] = (char)((plen >> 8) & 0xFF);
  164. pkt[start + 1] = (char)(plen & 0xFF);
  165. memcpy(&pkt[start + 2], passwd, plen);
  166. return 0;
  167. }
  168. /* add user to the CONNECT packet */
  169. static int add_user(const char *username, const size_t ulen,
  170. unsigned char *pkt, const size_t start, int remain_pos)
  171. {
  172. /* magic number that need to be set properly */
  173. const size_t conn_flags_pos = remain_pos + 8;
  174. if(ulen > 0xffff)
  175. return 1;
  176. /* set username flag */
  177. pkt[conn_flags_pos] |= 0x80;
  178. /* length of username provided */
  179. pkt[start] = (unsigned char)((ulen >> 8) & 0xFF);
  180. pkt[start + 1] = (unsigned char)(ulen & 0xFF);
  181. memcpy(&pkt[start + 2], username, ulen);
  182. return 0;
  183. }
  184. /* add client ID to the CONNECT packet */
  185. static int add_client_id(const char *client_id, const size_t client_id_len,
  186. char *pkt, const size_t start)
  187. {
  188. if(client_id_len != MQTT_CLIENTID_LEN)
  189. return 1;
  190. pkt[start] = 0x00;
  191. pkt[start + 1] = MQTT_CLIENTID_LEN;
  192. memcpy(&pkt[start + 2], client_id, MQTT_CLIENTID_LEN);
  193. return 0;
  194. }
  195. /* Set initial values of CONNECT packet */
  196. static int init_connpack(char *packet, char *remain, int remain_pos)
  197. {
  198. /* Fixed header starts */
  199. /* packet type */
  200. packet[0] = MQTT_MSG_CONNECT;
  201. /* remaining length field */
  202. memcpy(&packet[1], remain, remain_pos);
  203. /* Fixed header ends */
  204. /* Variable header starts */
  205. /* protocol length */
  206. packet[remain_pos + 1] = 0x00;
  207. packet[remain_pos + 2] = 0x04;
  208. /* protocol name */
  209. packet[remain_pos + 3] = 'M';
  210. packet[remain_pos + 4] = 'Q';
  211. packet[remain_pos + 5] = 'T';
  212. packet[remain_pos + 6] = 'T';
  213. /* protocol level */
  214. packet[remain_pos + 7] = 0x04;
  215. /* CONNECT flag: CleanSession */
  216. packet[remain_pos + 8] = 0x02;
  217. /* keep-alive 0 = disabled */
  218. packet[remain_pos + 9] = 0x00;
  219. packet[remain_pos + 10] = 0x3c;
  220. /*end of variable header*/
  221. return remain_pos + 10;
  222. }
  223. static CURLcode mqtt_connect(struct Curl_easy *data)
  224. {
  225. CURLcode result = CURLE_OK;
  226. int pos = 0;
  227. int rc = 0;
  228. /*remain length*/
  229. int remain_pos = 0;
  230. char remain[4] = {0};
  231. size_t packetlen = 0;
  232. size_t payloadlen = 0;
  233. size_t start_user = 0;
  234. size_t start_pwd = 0;
  235. char client_id[MQTT_CLIENTID_LEN + 1] = "curl";
  236. const size_t clen = strlen("curl");
  237. char *packet = NULL;
  238. /* extracting username from request */
  239. const char *username = data->state.aptr.user ?
  240. data->state.aptr.user : "";
  241. const size_t ulen = strlen(username);
  242. /* extracting password from request */
  243. const char *passwd = data->state.aptr.passwd ?
  244. data->state.aptr.passwd : "";
  245. const size_t plen = strlen(passwd);
  246. payloadlen = ulen + plen + MQTT_CLIENTID_LEN + 2;
  247. /* The plus 2 are for the MSB and LSB describing the length of the string to
  248. * be added on the payload. Refer to spec 1.5.2 and 1.5.4 */
  249. if(ulen)
  250. payloadlen += 2;
  251. if(plen)
  252. payloadlen += 2;
  253. /* getting how much occupy the remain length */
  254. remain_pos = mqtt_encode_len(remain, payloadlen + 10);
  255. /* 10 length of variable header and 1 the first byte of the fixed header */
  256. packetlen = payloadlen + 10 + remain_pos + 1;
  257. /* allocating packet */
  258. if(packetlen > 268435455)
  259. return CURLE_WEIRD_SERVER_REPLY;
  260. packet = malloc(packetlen);
  261. if(!packet)
  262. return CURLE_OUT_OF_MEMORY;
  263. memset(packet, 0, packetlen);
  264. /* set initial values for the CONNECT packet */
  265. pos = init_connpack(packet, remain, remain_pos);
  266. result = Curl_rand_hex(data, (unsigned char *)&client_id[clen],
  267. MQTT_CLIENTID_LEN - clen + 1);
  268. /* add client id */
  269. rc = add_client_id(client_id, strlen(client_id), packet, pos + 1);
  270. if(rc) {
  271. failf(data, "Client ID length mismatched: [%lu]", strlen(client_id));
  272. result = CURLE_WEIRD_SERVER_REPLY;
  273. goto end;
  274. }
  275. infof(data, "Using client id '%s'", client_id);
  276. /* position where starts the user payload */
  277. start_user = pos + 3 + MQTT_CLIENTID_LEN;
  278. /* position where starts the password payload */
  279. start_pwd = start_user + ulen;
  280. /* if user name was provided, add it to the packet */
  281. if(ulen) {
  282. start_pwd += 2;
  283. rc = add_user(username, ulen,
  284. (unsigned char *)packet, start_user, remain_pos);
  285. if(rc) {
  286. failf(data, "Username is too large: [%lu]", ulen);
  287. result = CURLE_WEIRD_SERVER_REPLY;
  288. goto end;
  289. }
  290. }
  291. /* if passwd was provided, add it to the packet */
  292. if(plen) {
  293. rc = add_passwd(passwd, plen, packet, start_pwd, remain_pos);
  294. if(rc) {
  295. failf(data, "Password is too large: [%lu]", plen);
  296. result = CURLE_WEIRD_SERVER_REPLY;
  297. goto end;
  298. }
  299. }
  300. if(!result)
  301. result = mqtt_send(data, packet, packetlen);
  302. end:
  303. if(packet)
  304. free(packet);
  305. Curl_safefree(data->state.aptr.user);
  306. Curl_safefree(data->state.aptr.passwd);
  307. return result;
  308. }
  309. static CURLcode mqtt_disconnect(struct Curl_easy *data)
  310. {
  311. CURLcode result = CURLE_OK;
  312. struct MQTT *mq = data->req.p.mqtt;
  313. result = mqtt_send(data, (char *)"\xe0\x00", 2);
  314. Curl_safefree(mq->sendleftovers);
  315. return result;
  316. }
  317. static CURLcode mqtt_verify_connack(struct Curl_easy *data)
  318. {
  319. CURLcode result;
  320. struct connectdata *conn = data->conn;
  321. curl_socket_t sockfd = conn->sock[FIRSTSOCKET];
  322. unsigned char readbuf[MQTT_CONNACK_LEN];
  323. ssize_t nread;
  324. result = Curl_read(data, sockfd, (char *)readbuf, MQTT_CONNACK_LEN, &nread);
  325. if(result)
  326. goto fail;
  327. Curl_debug(data, CURLINFO_HEADER_IN, (char *)readbuf, (size_t)nread);
  328. /* fixme */
  329. if(nread < MQTT_CONNACK_LEN) {
  330. result = CURLE_WEIRD_SERVER_REPLY;
  331. goto fail;
  332. }
  333. /* verify CONNACK */
  334. if(readbuf[0] != 0x00 || readbuf[1] != 0x00) {
  335. failf(data, "Expected %02x%02x but got %02x%02x",
  336. 0x00, 0x00, readbuf[0], readbuf[1]);
  337. result = CURLE_WEIRD_SERVER_REPLY;
  338. }
  339. fail:
  340. return result;
  341. }
  342. static CURLcode mqtt_get_topic(struct Curl_easy *data,
  343. char **topic, size_t *topiclen)
  344. {
  345. char *path = data->state.up.path;
  346. CURLcode result = CURLE_URL_MALFORMAT;
  347. if(strlen(path) > 1) {
  348. result = Curl_urldecode(path + 1, 0, topic, topiclen, REJECT_NADA);
  349. if(!result && (*topiclen > 0xffff)) {
  350. failf(data, "Too long MQTT topic");
  351. result = CURLE_URL_MALFORMAT;
  352. }
  353. }
  354. else
  355. failf(data, "No MQTT topic found. Forgot to URL encode it?");
  356. return result;
  357. }
  358. static CURLcode mqtt_subscribe(struct Curl_easy *data)
  359. {
  360. CURLcode result = CURLE_OK;
  361. char *topic = NULL;
  362. size_t topiclen;
  363. unsigned char *packet = NULL;
  364. size_t packetlen;
  365. char encodedsize[4];
  366. size_t n;
  367. struct connectdata *conn = data->conn;
  368. result = mqtt_get_topic(data, &topic, &topiclen);
  369. if(result)
  370. goto fail;
  371. conn->proto.mqtt.packetid++;
  372. packetlen = topiclen + 5; /* packetid + topic (has a two byte length field)
  373. + 2 bytes topic length + QoS byte */
  374. n = mqtt_encode_len((char *)encodedsize, packetlen);
  375. packetlen += n + 1; /* add one for the control packet type byte */
  376. packet = malloc(packetlen);
  377. if(!packet) {
  378. result = CURLE_OUT_OF_MEMORY;
  379. goto fail;
  380. }
  381. packet[0] = MQTT_MSG_SUBSCRIBE;
  382. memcpy(&packet[1], encodedsize, n);
  383. packet[1 + n] = (conn->proto.mqtt.packetid >> 8) & 0xff;
  384. packet[2 + n] = conn->proto.mqtt.packetid & 0xff;
  385. packet[3 + n] = (topiclen >> 8) & 0xff;
  386. packet[4 + n ] = topiclen & 0xff;
  387. memcpy(&packet[5 + n], topic, topiclen);
  388. packet[5 + n + topiclen] = 0; /* QoS zero */
  389. result = mqtt_send(data, (char *)packet, packetlen);
  390. fail:
  391. free(topic);
  392. free(packet);
  393. return result;
  394. }
  395. /*
  396. * Called when the first byte was already read.
  397. */
  398. static CURLcode mqtt_verify_suback(struct Curl_easy *data)
  399. {
  400. CURLcode result;
  401. struct connectdata *conn = data->conn;
  402. curl_socket_t sockfd = conn->sock[FIRSTSOCKET];
  403. unsigned char readbuf[MQTT_SUBACK_LEN];
  404. ssize_t nread;
  405. struct mqtt_conn *mqtt = &conn->proto.mqtt;
  406. result = Curl_read(data, sockfd, (char *)readbuf, MQTT_SUBACK_LEN, &nread);
  407. if(result)
  408. goto fail;
  409. Curl_debug(data, CURLINFO_HEADER_IN, (char *)readbuf, (size_t)nread);
  410. /* fixme */
  411. if(nread < MQTT_SUBACK_LEN) {
  412. result = CURLE_WEIRD_SERVER_REPLY;
  413. goto fail;
  414. }
  415. /* verify SUBACK */
  416. if(readbuf[0] != ((mqtt->packetid >> 8) & 0xff) ||
  417. readbuf[1] != (mqtt->packetid & 0xff) ||
  418. readbuf[2] != 0x00)
  419. result = CURLE_WEIRD_SERVER_REPLY;
  420. fail:
  421. return result;
  422. }
  423. static CURLcode mqtt_publish(struct Curl_easy *data)
  424. {
  425. CURLcode result;
  426. char *payload = data->set.postfields;
  427. size_t payloadlen;
  428. char *topic = NULL;
  429. size_t topiclen;
  430. unsigned char *pkt = NULL;
  431. size_t i = 0;
  432. size_t remaininglength;
  433. size_t encodelen;
  434. char encodedbytes[4];
  435. curl_off_t postfieldsize = data->set.postfieldsize;
  436. if(!payload)
  437. return CURLE_BAD_FUNCTION_ARGUMENT;
  438. if(postfieldsize < 0)
  439. payloadlen = strlen(payload);
  440. else
  441. payloadlen = (size_t)postfieldsize;
  442. result = mqtt_get_topic(data, &topic, &topiclen);
  443. if(result)
  444. goto fail;
  445. remaininglength = payloadlen + 2 + topiclen;
  446. encodelen = mqtt_encode_len(encodedbytes, remaininglength);
  447. /* add the control byte and the encoded remaining length */
  448. pkt = malloc(remaininglength + 1 + encodelen);
  449. if(!pkt) {
  450. result = CURLE_OUT_OF_MEMORY;
  451. goto fail;
  452. }
  453. /* assemble packet */
  454. pkt[i++] = MQTT_MSG_PUBLISH;
  455. memcpy(&pkt[i], encodedbytes, encodelen);
  456. i += encodelen;
  457. pkt[i++] = (topiclen >> 8) & 0xff;
  458. pkt[i++] = (topiclen & 0xff);
  459. memcpy(&pkt[i], topic, topiclen);
  460. i += topiclen;
  461. memcpy(&pkt[i], payload, payloadlen);
  462. i += payloadlen;
  463. result = mqtt_send(data, (char *)pkt, i);
  464. fail:
  465. free(pkt);
  466. free(topic);
  467. return result;
  468. }
  469. static size_t mqtt_decode_len(unsigned char *buf,
  470. size_t buflen, size_t *lenbytes)
  471. {
  472. size_t len = 0;
  473. size_t mult = 1;
  474. size_t i;
  475. unsigned char encoded = 128;
  476. for(i = 0; (i < buflen) && (encoded & 128); i++) {
  477. encoded = buf[i];
  478. len += (encoded & 127) * mult;
  479. mult *= 128;
  480. }
  481. if(lenbytes)
  482. *lenbytes = i;
  483. return len;
  484. }
  485. #ifdef CURLDEBUG
  486. static const char *statenames[]={
  487. "MQTT_FIRST",
  488. "MQTT_REMAINING_LENGTH",
  489. "MQTT_CONNACK",
  490. "MQTT_SUBACK",
  491. "MQTT_SUBACK_COMING",
  492. "MQTT_PUBWAIT",
  493. "MQTT_PUB_REMAIN",
  494. "NOT A STATE"
  495. };
  496. #endif
  497. /* The only way to change state */
  498. static void mqstate(struct Curl_easy *data,
  499. enum mqttstate state,
  500. enum mqttstate nextstate) /* used if state == FIRST */
  501. {
  502. struct connectdata *conn = data->conn;
  503. struct mqtt_conn *mqtt = &conn->proto.mqtt;
  504. #ifdef CURLDEBUG
  505. infof(data, "%s (from %s) (next is %s)",
  506. statenames[state],
  507. statenames[mqtt->state],
  508. (state == MQTT_FIRST)? statenames[nextstate] : "");
  509. #endif
  510. mqtt->state = state;
  511. if(state == MQTT_FIRST)
  512. mqtt->nextstate = nextstate;
  513. }
  514. /* for the publish packet */
  515. #define MQTT_HEADER_LEN 5 /* max 5 bytes */
  516. static CURLcode mqtt_read_publish(struct Curl_easy *data, bool *done)
  517. {
  518. CURLcode result = CURLE_OK;
  519. struct connectdata *conn = data->conn;
  520. curl_socket_t sockfd = conn->sock[FIRSTSOCKET];
  521. ssize_t nread;
  522. unsigned char *pkt = (unsigned char *)data->state.buffer;
  523. size_t remlen;
  524. struct mqtt_conn *mqtt = &conn->proto.mqtt;
  525. struct MQTT *mq = data->req.p.mqtt;
  526. unsigned char packet;
  527. switch(mqtt->state) {
  528. MQTT_SUBACK_COMING:
  529. case MQTT_SUBACK_COMING:
  530. result = mqtt_verify_suback(data);
  531. if(result)
  532. break;
  533. mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
  534. break;
  535. case MQTT_SUBACK:
  536. case MQTT_PUBWAIT:
  537. /* we are expecting PUBLISH or SUBACK */
  538. packet = mq->firstbyte & 0xf0;
  539. if(packet == MQTT_MSG_PUBLISH)
  540. mqstate(data, MQTT_PUB_REMAIN, MQTT_NOSTATE);
  541. else if(packet == MQTT_MSG_SUBACK) {
  542. mqstate(data, MQTT_SUBACK_COMING, MQTT_NOSTATE);
  543. goto MQTT_SUBACK_COMING;
  544. }
  545. else if(packet == MQTT_MSG_DISCONNECT) {
  546. infof(data, "Got DISCONNECT");
  547. *done = TRUE;
  548. goto end;
  549. }
  550. else {
  551. result = CURLE_WEIRD_SERVER_REPLY;
  552. goto end;
  553. }
  554. /* -- switched state -- */
  555. remlen = mq->remaining_length;
  556. infof(data, "Remaining length: %zd bytes", remlen);
  557. if(data->set.max_filesize &&
  558. (curl_off_t)remlen > data->set.max_filesize) {
  559. failf(data, "Maximum file size exceeded");
  560. result = CURLE_FILESIZE_EXCEEDED;
  561. goto end;
  562. }
  563. Curl_pgrsSetDownloadSize(data, remlen);
  564. data->req.bytecount = 0;
  565. data->req.size = remlen;
  566. mq->npacket = remlen; /* get this many bytes */
  567. /* FALLTHROUGH */
  568. case MQTT_PUB_REMAIN: {
  569. /* read rest of packet, but no more. Cap to buffer size */
  570. struct SingleRequest *k = &data->req;
  571. size_t rest = mq->npacket;
  572. if(rest > (size_t)data->set.buffer_size)
  573. rest = (size_t)data->set.buffer_size;
  574. result = Curl_read(data, sockfd, (char *)pkt, rest, &nread);
  575. if(result) {
  576. if(CURLE_AGAIN == result) {
  577. infof(data, "EEEE AAAAGAIN");
  578. }
  579. goto end;
  580. }
  581. if(!nread) {
  582. infof(data, "server disconnected");
  583. result = CURLE_PARTIAL_FILE;
  584. goto end;
  585. }
  586. Curl_debug(data, CURLINFO_DATA_IN, (char *)pkt, (size_t)nread);
  587. mq->npacket -= nread;
  588. k->bytecount += nread;
  589. Curl_pgrsSetDownloadCounter(data, k->bytecount);
  590. /* if QoS is set, message contains packet id */
  591. result = Curl_client_write(data, CLIENTWRITE_BODY, (char *)pkt, nread);
  592. if(result)
  593. goto end;
  594. if(!mq->npacket)
  595. /* no more PUBLISH payload, back to subscribe wait state */
  596. mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
  597. break;
  598. }
  599. default:
  600. DEBUGASSERT(NULL); /* illegal state */
  601. result = CURLE_WEIRD_SERVER_REPLY;
  602. goto end;
  603. }
  604. end:
  605. return result;
  606. }
  607. static CURLcode mqtt_do(struct Curl_easy *data, bool *done)
  608. {
  609. CURLcode result = CURLE_OK;
  610. *done = FALSE; /* unconditionally */
  611. result = mqtt_connect(data);
  612. if(result) {
  613. failf(data, "Error %d sending MQTT CONNECT request", result);
  614. return result;
  615. }
  616. mqstate(data, MQTT_FIRST, MQTT_CONNACK);
  617. return CURLE_OK;
  618. }
  619. static CURLcode mqtt_done(struct Curl_easy *data,
  620. CURLcode status, bool premature)
  621. {
  622. struct MQTT *mq = data->req.p.mqtt;
  623. (void)status;
  624. (void)premature;
  625. Curl_safefree(mq->sendleftovers);
  626. return CURLE_OK;
  627. }
  628. static CURLcode mqtt_doing(struct Curl_easy *data, bool *done)
  629. {
  630. CURLcode result = CURLE_OK;
  631. struct connectdata *conn = data->conn;
  632. struct mqtt_conn *mqtt = &conn->proto.mqtt;
  633. struct MQTT *mq = data->req.p.mqtt;
  634. ssize_t nread;
  635. curl_socket_t sockfd = conn->sock[FIRSTSOCKET];
  636. unsigned char *pkt = (unsigned char *)data->state.buffer;
  637. unsigned char byte;
  638. *done = FALSE;
  639. if(mq->nsend) {
  640. /* send the remainder of an outgoing packet */
  641. char *ptr = mq->sendleftovers;
  642. result = mqtt_send(data, mq->sendleftovers, mq->nsend);
  643. free(ptr);
  644. if(result)
  645. return result;
  646. }
  647. infof(data, "mqtt_doing: state [%d]", (int) mqtt->state);
  648. switch(mqtt->state) {
  649. case MQTT_FIRST:
  650. /* Read the initial byte only */
  651. result = Curl_read(data, sockfd, (char *)&mq->firstbyte, 1, &nread);
  652. if(result)
  653. break;
  654. else if(!nread) {
  655. failf(data, "Connection disconnected");
  656. *done = TRUE;
  657. result = CURLE_RECV_ERROR;
  658. break;
  659. }
  660. Curl_debug(data, CURLINFO_HEADER_IN, (char *)&mq->firstbyte, 1);
  661. /* remember the first byte */
  662. mq->npacket = 0;
  663. mqstate(data, MQTT_REMAINING_LENGTH, MQTT_NOSTATE);
  664. /* FALLTHROUGH */
  665. case MQTT_REMAINING_LENGTH:
  666. do {
  667. result = Curl_read(data, sockfd, (char *)&byte, 1, &nread);
  668. if(!nread)
  669. break;
  670. Curl_debug(data, CURLINFO_HEADER_IN, (char *)&byte, 1);
  671. pkt[mq->npacket++] = byte;
  672. } while((byte & 0x80) && (mq->npacket < 4));
  673. if(nread && (byte & 0x80))
  674. /* MQTT supports up to 127 * 128^0 + 127 * 128^1 + 127 * 128^2 +
  675. 127 * 128^3 bytes. server tried to send more */
  676. result = CURLE_WEIRD_SERVER_REPLY;
  677. if(result)
  678. break;
  679. mq->remaining_length = mqtt_decode_len(&pkt[0], mq->npacket, NULL);
  680. mq->npacket = 0;
  681. if(mq->remaining_length) {
  682. mqstate(data, mqtt->nextstate, MQTT_NOSTATE);
  683. break;
  684. }
  685. mqstate(data, MQTT_FIRST, MQTT_FIRST);
  686. if(mq->firstbyte == MQTT_MSG_DISCONNECT) {
  687. infof(data, "Got DISCONNECT");
  688. *done = TRUE;
  689. }
  690. break;
  691. case MQTT_CONNACK:
  692. result = mqtt_verify_connack(data);
  693. if(result)
  694. break;
  695. if(data->state.httpreq == HTTPREQ_POST) {
  696. result = mqtt_publish(data);
  697. if(!result) {
  698. result = mqtt_disconnect(data);
  699. *done = TRUE;
  700. }
  701. mqtt->nextstate = MQTT_FIRST;
  702. }
  703. else {
  704. result = mqtt_subscribe(data);
  705. if(!result) {
  706. mqstate(data, MQTT_FIRST, MQTT_SUBACK);
  707. }
  708. }
  709. break;
  710. case MQTT_SUBACK:
  711. case MQTT_PUBWAIT:
  712. case MQTT_PUB_REMAIN:
  713. result = mqtt_read_publish(data, done);
  714. break;
  715. default:
  716. failf(data, "State not handled yet");
  717. *done = TRUE;
  718. break;
  719. }
  720. if(result == CURLE_AGAIN)
  721. result = CURLE_OK;
  722. return result;
  723. }
  724. #endif /* CURL_DISABLE_MQTT */