/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 */ |