1 /***************************************************************************
2 * _ _ ____ _
3 * Project ___| | | | _ \| |
4 * / __| | | | |_) | |
5 * | (__| |_| | _ <| |___
6 * \___|\___/|_| \_\_____|
7 *
8 * Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
9 *
10 * This software is licensed as described in the file COPYING, which
11 * you should have received as part of this distribution. The terms
12 * are also available at https://curl.se/docs/copyright.html.
13 *
14 * You may opt to use, copy, modify, merge, publish, distribute and/or sell
15 * copies of the Software, and permit persons to whom the Software is
16 * furnished to do so, under the terms of the COPYING file.
17 *
18 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
19 * KIND, either express or implied.
20 *
21 * SPDX-License-Identifier: curl
22 *
23 ***************************************************************************/
24 #include "server_setup.h"
25 #include <stdlib.h>
26 #include <string.h>
27 #include "util.h"
28
29 /* Function
30 *
31 * Accepts a TCP connection on a custom port (IPv4 or IPv6). Speaks MQTT.
32 *
33 * Read commands from FILE (set with --config). The commands control how to
34 * act and is reset to defaults each client TCP connect.
35 *
36 * Config file keywords:
37 *
38 * TODO
39 */
40
41 /* based on sockfilt.c */
42
43 #include <signal.h>
44 #ifdef HAVE_NETINET_IN_H
45 #include <netinet/in.h>
46 #endif
47 #ifdef HAVE_NETINET_IN6_H
48 #include <netinet/in6.h>
49 #endif
50 #ifdef HAVE_ARPA_INET_H
51 #include <arpa/inet.h>
52 #endif
53 #ifdef HAVE_NETDB_H
54 #include <netdb.h>
55 #endif
56
57 #include "curlx.h" /* from the private lib dir */
58 #include "getpart.h"
59 #include "inet_pton.h"
60 #include "server_sockaddr.h"
61 #include "warnless.h"
62
63 #include "tool_binmode.h"
64
65 /* include memdebug.h last */
66 #include "memdebug.h"
67
68 #ifdef USE_WINSOCK
69 #undef EINTR
70 #define EINTR 4 /* errno.h value */
71 #undef EAGAIN
72 #define EAGAIN 11 /* errno.h value */
73 #undef ENOMEM
74 #define ENOMEM 12 /* errno.h value */
75 #undef EINVAL
76 #define EINVAL 22 /* errno.h value */
77 #endif
78
79 #define DEFAULT_PORT 1883 /* MQTT default port */
80
81 #ifndef DEFAULT_LOGFILE
82 #define DEFAULT_LOGFILE "log/mqttd.log"
83 #endif
84
85 #ifndef DEFAULT_CONFIG
86 #define DEFAULT_CONFIG "mqttd.config"
87 #endif
88
89 #define MQTT_MSG_CONNECT 0x10
90 #define MQTT_MSG_CONNACK 0x20
91 #define MQTT_MSG_PUBLISH 0x30
92 #define MQTT_MSG_PUBACK 0x40
93 #define MQTT_MSG_SUBSCRIBE 0x82
94 #define MQTT_MSG_SUBACK 0x90
95 #define MQTT_MSG_DISCONNECT 0xe0
96
97 #define MQTT_CONNACK_LEN 4
98 #define MQTT_SUBACK_LEN 5
99 #define MQTT_CLIENTID_LEN 12 /* "curl0123abcd" */
100
101 struct configurable {
102 unsigned char version; /* initial version byte in the request must match
103 this */
104 bool publish_before_suback;
105 bool short_publish;
106 bool excessive_remaining;
107 unsigned char error_connack;
108 int testnum;
109 };
110
111 #define REQUEST_DUMP "server.input"
112 #define CONFIG_VERSION 5
113
114 static struct configurable config;
115
116 const char *serverlogfile = DEFAULT_LOGFILE;
117 static const char *configfile = DEFAULT_CONFIG;
118 static const char *logdir = "log";
119 static char loglockfile[256];
120
121 #ifdef USE_IPV6
122 static bool use_ipv6 = FALSE;
123 #endif
124 static const char *ipv_inuse = "IPv4";
125 static unsigned short port = DEFAULT_PORT;
126
resetdefaults(void)127 static void resetdefaults(void)
128 {
129 logmsg("Reset to defaults");
130 config.version = CONFIG_VERSION;
131 config.publish_before_suback = FALSE;
132 config.short_publish = FALSE;
133 config.excessive_remaining = FALSE;
134 config.error_connack = 0;
135 config.testnum = 0;
136 }
137
byteval(char * value)138 static unsigned char byteval(char *value)
139 {
140 unsigned long num = strtoul(value, NULL, 10);
141 return num & 0xff;
142 }
143
getconfig(void)144 static void getconfig(void)
145 {
146 FILE *fp = fopen(configfile, FOPEN_READTEXT);
147 resetdefaults();
148 if(fp) {
149 char buffer[512];
150 logmsg("parse config file");
151 while(fgets(buffer, sizeof(buffer), fp)) {
152 char key[32];
153 char value[32];
154 if(2 == sscanf(buffer, "%31s %31s", key, value)) {
155 if(!strcmp(key, "version")) {
156 config.version = byteval(value);
157 logmsg("version [%d] set", config.version);
158 }
159 else if(!strcmp(key, "PUBLISH-before-SUBACK")) {
160 logmsg("PUBLISH-before-SUBACK set");
161 config.publish_before_suback = TRUE;
162 }
163 else if(!strcmp(key, "short-PUBLISH")) {
164 logmsg("short-PUBLISH set");
165 config.short_publish = TRUE;
166 }
167 else if(!strcmp(key, "error-CONNACK")) {
168 config.error_connack = byteval(value);
169 logmsg("error-CONNACK = %d", config.error_connack);
170 }
171 else if(!strcmp(key, "Testnum")) {
172 config.testnum = atoi(value);
173 logmsg("testnum = %d", config.testnum);
174 }
175 else if(!strcmp(key, "excessive-remaining")) {
176 logmsg("excessive-remaining set");
177 config.excessive_remaining = TRUE;
178 }
179 }
180 }
181 fclose(fp);
182 }
183 else {
184 logmsg("No config file '%s' to read", configfile);
185 }
186 }
187
loghex(unsigned char * buffer,ssize_t len)188 static void loghex(unsigned char *buffer, ssize_t len)
189 {
190 char data[12000];
191 ssize_t i;
192 unsigned char *ptr = buffer;
193 char *optr = data;
194 ssize_t width = 0;
195 int left = sizeof(data);
196
197 for(i = 0; i < len && (left >= 0); i++) {
198 msnprintf(optr, left, "%02x", ptr[i]);
199 width += 2;
200 optr += 2;
201 left -= 2;
202 }
203 if(width)
204 logmsg("'%s'", data);
205 }
206
207 typedef enum {
208 FROM_CLIENT,
209 FROM_SERVER
210 } mqttdir;
211
logprotocol(mqttdir dir,const char * prefix,size_t remlen,FILE * output,unsigned char * buffer,ssize_t len)212 static void logprotocol(mqttdir dir,
213 const char *prefix, size_t remlen,
214 FILE *output,
215 unsigned char *buffer, ssize_t len)
216 {
217 char data[12000] = "";
218 ssize_t i;
219 unsigned char *ptr = buffer;
220 char *optr = data;
221 int left = sizeof(data);
222
223 for(i = 0; i < len && (left >= 0); i++) {
224 msnprintf(optr, left, "%02x", ptr[i]);
225 optr += 2;
226 left -= 2;
227 }
228 fprintf(output, "%s %s %zx %s\n",
229 dir == FROM_CLIENT ? "client" : "server",
230 prefix, remlen, data);
231 }
232
233
234 /* return 0 on success */
connack(FILE * dump,curl_socket_t fd)235 static int connack(FILE *dump, curl_socket_t fd)
236 {
237 unsigned char packet[]={
238 MQTT_MSG_CONNACK, 0x02,
239 0x00, 0x00
240 };
241 ssize_t rc;
242
243 packet[3] = config.error_connack;
244
245 rc = swrite(fd, (char *)packet, sizeof(packet));
246 if(rc > 0) {
247 logmsg("WROTE %zd bytes [CONNACK]", rc);
248 loghex(packet, rc);
249 logprotocol(FROM_SERVER, "CONNACK", 2, dump, packet, sizeof(packet));
250 }
251 if(rc == sizeof(packet)) {
252 return 0;
253 }
254 return 1;
255 }
256
257 /* return 0 on success */
suback(FILE * dump,curl_socket_t fd,unsigned short packetid)258 static int suback(FILE *dump, curl_socket_t fd, unsigned short packetid)
259 {
260 unsigned char packet[]={
261 MQTT_MSG_SUBACK, 0x03,
262 0, 0, /* filled in below */
263 0x00
264 };
265 ssize_t rc;
266 packet[2] = (unsigned char)(packetid >> 8);
267 packet[3] = (unsigned char)(packetid & 0xff);
268
269 rc = swrite(fd, (char *)packet, sizeof(packet));
270 if(rc == sizeof(packet)) {
271 logmsg("WROTE %zd bytes [SUBACK]", rc);
272 loghex(packet, rc);
273 logprotocol(FROM_SERVER, "SUBACK", 3, dump, packet, rc);
274 return 0;
275 }
276 return 1;
277 }
278
279 #ifdef QOS
280 /* return 0 on success */
puback(FILE * dump,curl_socket_t fd,unsigned short packetid)281 static int puback(FILE *dump, curl_socket_t fd, unsigned short packetid)
282 {
283 unsigned char packet[]={
284 MQTT_MSG_PUBACK, 0x00,
285 0, 0 /* filled in below */
286 };
287 ssize_t rc;
288 packet[2] = (unsigned char)(packetid >> 8);
289 packet[3] = (unsigned char)(packetid & 0xff);
290
291 rc = swrite(fd, (char *)packet, sizeof(packet));
292 if(rc == sizeof(packet)) {
293 logmsg("WROTE %zd bytes [PUBACK]", rc);
294 loghex(packet, rc);
295 logprotocol(FROM_SERVER, dump, packet, rc);
296 return 0;
297 }
298 logmsg("Failed sending [PUBACK]");
299 return 1;
300 }
301 #endif
302
303 /* return 0 on success */
disconnect(FILE * dump,curl_socket_t fd)304 static int disconnect(FILE *dump, curl_socket_t fd)
305 {
306 unsigned char packet[]={
307 MQTT_MSG_DISCONNECT, 0x00,
308 };
309 ssize_t rc = swrite(fd, (char *)packet, sizeof(packet));
310 if(rc == sizeof(packet)) {
311 logmsg("WROTE %zd bytes [DISCONNECT]", rc);
312 loghex(packet, rc);
313 logprotocol(FROM_SERVER, "DISCONNECT", 0, dump, packet, rc);
314 return 0;
315 }
316 logmsg("Failed sending [DISCONNECT]");
317 return 1;
318 }
319
320
321
322 /*
323 do
324
325 encodedByte = X MOD 128
326
327 X = X DIV 128
328
329 // if there are more data to encode, set the top bit of this byte
330
331 if ( X > 0 )
332
333 encodedByte = encodedByte OR 128
334
335 endif
336
337 'output' encodedByte
338
339 while ( X > 0 )
340
341 */
342
343 /* return number of bytes used */
encode_length(size_t packetlen,unsigned char * remlength)344 static size_t encode_length(size_t packetlen,
345 unsigned char *remlength) /* 4 bytes */
346 {
347 size_t bytes = 0;
348 unsigned char encode;
349
350 do {
351 encode = packetlen % 0x80;
352 packetlen /= 0x80;
353 if(packetlen)
354 encode |= 0x80;
355
356 remlength[bytes++] = encode;
357
358 if(bytes > 3) {
359 logmsg("too large packet!");
360 return 0;
361 }
362 } while(packetlen);
363
364 return bytes;
365 }
366
367
decode_length(unsigned char * buf,size_t buflen,size_t * lenbytes)368 static size_t decode_length(unsigned char *buf,
369 size_t buflen, size_t *lenbytes)
370 {
371 size_t len = 0;
372 size_t mult = 1;
373 size_t i;
374 unsigned char encoded = 0x80;
375
376 for(i = 0; (i < buflen) && (encoded & 0x80); i++) {
377 encoded = buf[i];
378 len += (encoded & 0x7f) * mult;
379 mult *= 0x80;
380 }
381
382 if(lenbytes)
383 *lenbytes = i;
384
385 return len;
386 }
387
388
389 /* return 0 on success */
publish(FILE * dump,curl_socket_t fd,unsigned short packetid,char * topic,char * payload,size_t payloadlen)390 static int publish(FILE *dump,
391 curl_socket_t fd, unsigned short packetid,
392 char *topic, char *payload, size_t payloadlen)
393 {
394 size_t topiclen = strlen(topic);
395 unsigned char *packet;
396 size_t payloadindex;
397 size_t remaininglength = topiclen + 2 + payloadlen;
398 size_t packetlen;
399 size_t sendamount;
400 ssize_t rc;
401 unsigned char rembuffer[4];
402 size_t encodedlen;
403
404 if(config.excessive_remaining) {
405 /* manually set illegal remaining length */
406 rembuffer[0] = 0xff;
407 rembuffer[1] = 0xff;
408 rembuffer[2] = 0xff;
409 rembuffer[3] = 0x80; /* maximum allowed here by spec is 0x7f */
410 encodedlen = 4;
411 }
412 else
413 encodedlen = encode_length(remaininglength, rembuffer);
414
415 /* one packet type byte (possibly two more for packetid) */
416 packetlen = remaininglength + encodedlen + 1;
417 packet = malloc(packetlen);
418 if(!packet)
419 return 1;
420
421 packet[0] = MQTT_MSG_PUBLISH; /* TODO: set QoS? */
422 memcpy(&packet[1], rembuffer, encodedlen);
423
424 (void)packetid;
425 /* packet_id if QoS is set */
426
427 packet[1 + encodedlen] = (unsigned char)(topiclen >> 8);
428 packet[2 + encodedlen] = (unsigned char)(topiclen & 0xff);
429 memcpy(&packet[3 + encodedlen], topic, topiclen);
430
431 payloadindex = 3 + topiclen + encodedlen;
432 memcpy(&packet[payloadindex], payload, payloadlen);
433
434 sendamount = packetlen;
435 if(config.short_publish)
436 sendamount -= 2;
437
438 rc = swrite(fd, (char *)packet, sendamount);
439 if(rc > 0) {
440 logmsg("WROTE %zd bytes [PUBLISH]", rc);
441 loghex(packet, rc);
442 logprotocol(FROM_SERVER, "PUBLISH", remaininglength, dump, packet, rc);
443 }
444 free(packet);
445 if((size_t)rc == packetlen)
446 return 0;
447 return 1;
448 }
449
450 #define MAX_TOPIC_LENGTH 65535
451 #define MAX_CLIENT_ID_LENGTH 32
452
453 static char topic[MAX_TOPIC_LENGTH + 1];
454
fixedheader(curl_socket_t fd,unsigned char * bytep,size_t * remaining_lengthp,size_t * remaining_length_bytesp)455 static int fixedheader(curl_socket_t fd,
456 unsigned char *bytep,
457 size_t *remaining_lengthp,
458 size_t *remaining_length_bytesp)
459 {
460 /* get the fixed header */
461 unsigned char buffer[10];
462
463 /* get the first two bytes */
464 ssize_t rc = sread(fd, (char *)buffer, 2);
465 size_t i;
466 if(rc < 2) {
467 logmsg("READ %zd bytes [SHORT!]", rc);
468 return 1; /* fail */
469 }
470 logmsg("READ %zd bytes", rc);
471 loghex(buffer, rc);
472 *bytep = buffer[0];
473
474 /* if the length byte has the top bit set, get the next one too */
475 i = 1;
476 while(buffer[i] & 0x80) {
477 i++;
478 rc = sread(fd, (char *)&buffer[i], 1);
479 if(rc != 1) {
480 logmsg("Remaining Length broken");
481 return 1;
482 }
483 }
484 *remaining_lengthp = decode_length(&buffer[1], i, remaining_length_bytesp);
485 logmsg("Remaining Length: %zu [%zu bytes]", *remaining_lengthp,
486 *remaining_length_bytesp);
487 return 0;
488 }
489
mqttit(curl_socket_t fd)490 static curl_socket_t mqttit(curl_socket_t fd)
491 {
492 size_t buff_size = 10*1024;
493 unsigned char *buffer = NULL;
494 ssize_t rc;
495 unsigned char byte;
496 unsigned short packet_id;
497 size_t payload_len;
498 size_t client_id_length;
499 size_t topic_len;
500 size_t remaining_length = 0;
501 size_t bytes = 0; /* remaining length field size in bytes */
502 char client_id[MAX_CLIENT_ID_LENGTH];
503 long testno;
504 FILE *stream = NULL;
505 FILE *dump;
506 char dumpfile[256];
507
508 static const char protocol[7] = {
509 0x00, 0x04, /* protocol length */
510 'M','Q','T','T', /* protocol name */
511 0x04 /* protocol level */
512 };
513 msnprintf(dumpfile, sizeof(dumpfile), "%s/%s", logdir, REQUEST_DUMP);
514 dump = fopen(dumpfile, "ab");
515 if(!dump)
516 goto end;
517
518 getconfig();
519
520 testno = config.testnum;
521
522 if(testno)
523 logmsg("Found test number %ld", testno);
524
525 buffer = malloc(buff_size);
526 if(!buffer) {
527 logmsg("Out of memory, unable to allocate buffer");
528 goto end;
529 }
530
531 do {
532 unsigned char usr_flag = 0x80;
533 unsigned char passwd_flag = 0x40;
534 unsigned char conn_flags;
535 const size_t client_id_offset = 12;
536 size_t start_usr;
537 size_t start_passwd;
538
539 /* get the fixed header */
540 rc = fixedheader(fd, &byte, &remaining_length, &bytes);
541 if(rc)
542 break;
543
544 if(remaining_length >= buff_size) {
545 unsigned char *newbuffer;
546 buff_size = remaining_length;
547 newbuffer = realloc(buffer, buff_size);
548 if(!newbuffer) {
549 logmsg("Failed realloc of size %zu", buff_size);
550 goto end;
551 }
552 buffer = newbuffer;
553 }
554
555 if(remaining_length) {
556 /* reading variable header and payload into buffer */
557 rc = sread(fd, (char *)buffer, remaining_length);
558 if(rc > 0) {
559 logmsg("READ %zd bytes", rc);
560 loghex(buffer, rc);
561 }
562 }
563
564 if(byte == MQTT_MSG_CONNECT) {
565 logprotocol(FROM_CLIENT, "CONNECT", remaining_length,
566 dump, buffer, rc);
567
568 if(memcmp(protocol, buffer, sizeof(protocol))) {
569 logmsg("Protocol preamble mismatch");
570 goto end;
571 }
572 /* ignore the connect flag byte and two keepalive bytes */
573 payload_len = (size_t)(buffer[10] << 8) | buffer[11];
574 /* first part of the payload is the client ID */
575 client_id_length = payload_len;
576
577 /* checking if user and password flags were set */
578 conn_flags = buffer[7];
579
580 start_usr = client_id_offset + payload_len;
581 if(usr_flag == (unsigned char)(conn_flags & usr_flag)) {
582 logmsg("User flag is present in CONN flag");
583 payload_len += (size_t)(buffer[start_usr] << 8) |
584 buffer[start_usr + 1];
585 payload_len += 2; /* MSB and LSB for user length */
586 }
587
588 start_passwd = client_id_offset + payload_len;
589 if(passwd_flag == (char)(conn_flags & passwd_flag)) {
590 logmsg("Password flag is present in CONN flags");
591 payload_len += (size_t)(buffer[start_passwd] << 8) |
592 buffer[start_passwd + 1];
593 payload_len += 2; /* MSB and LSB for password length */
594 }
595
596 /* check the length of the payload */
597 if((ssize_t)payload_len != (rc - 12)) {
598 logmsg("Payload length mismatch, expected %zx got %zx",
599 rc - 12, payload_len);
600 goto end;
601 }
602 /* check the length of the client ID */
603 else if((client_id_length + 1) > MAX_CLIENT_ID_LENGTH) {
604 logmsg("Too large client id");
605 goto end;
606 }
607 memcpy(client_id, &buffer[12], client_id_length);
608 client_id[client_id_length] = 0;
609
610 logmsg("MQTT client connect accepted: %s", client_id);
611
612 /* The first packet sent from the Server to the Client MUST be a
613 CONNACK Packet */
614
615 if(connack(dump, fd)) {
616 logmsg("failed sending CONNACK");
617 goto end;
618 }
619 }
620 else if(byte == MQTT_MSG_SUBSCRIBE) {
621 int error;
622 char *data;
623 size_t datalen;
624 logprotocol(FROM_CLIENT, "SUBSCRIBE", remaining_length,
625 dump, buffer, rc);
626 logmsg("Incoming SUBSCRIBE");
627
628 if(rc < 6) {
629 logmsg("Too small SUBSCRIBE");
630 goto end;
631 }
632
633 /* two bytes packet id */
634 packet_id = (unsigned short)((buffer[0] << 8) | buffer[1]);
635
636 /* two bytes topic length */
637 topic_len = (size_t)(buffer[2] << 8) | buffer[3];
638 if(topic_len != (remaining_length - 5)) {
639 logmsg("Wrong topic length, got %zu expected %zu",
640 topic_len, remaining_length - 5);
641 goto end;
642 }
643 memcpy(topic, &buffer[4], topic_len);
644 topic[topic_len] = 0;
645
646 /* there's a QoS byte (two bits) after the topic */
647
648 logmsg("SUBSCRIBE to '%s' [%d]", topic, packet_id);
649 stream = test2fopen(testno, logdir);
650 error = getpart(&data, &datalen, "reply", "data", stream);
651 if(!error) {
652 if(!config.publish_before_suback) {
653 if(suback(dump, fd, packet_id)) {
654 logmsg("failed sending SUBACK");
655 free(data);
656 goto end;
657 }
658 }
659 if(publish(dump, fd, packet_id, topic, data, datalen)) {
660 logmsg("PUBLISH failed");
661 free(data);
662 goto end;
663 }
664 free(data);
665 if(config.publish_before_suback) {
666 if(suback(dump, fd, packet_id)) {
667 logmsg("failed sending SUBACK");
668 goto end;
669 }
670 }
671 }
672 else {
673 char *def = (char *)"this is random payload yes yes it is";
674 publish(dump, fd, packet_id, topic, def, strlen(def));
675 }
676 disconnect(dump, fd);
677 }
678 else if((byte & 0xf0) == (MQTT_MSG_PUBLISH & 0xf0)) {
679 size_t topiclen;
680
681 logmsg("Incoming PUBLISH");
682 logprotocol(FROM_CLIENT, "PUBLISH", remaining_length,
683 dump, buffer, rc);
684
685 topiclen = (size_t)(buffer[1 + bytes] << 8) | buffer[2 + bytes];
686 logmsg("Got %zu bytes topic", topiclen);
687 /* TODO: verify topiclen */
688
689 #ifdef QOS
690 /* TODO: handle packetid if there is one. Send puback if QoS > 0 */
691 puback(dump, fd, 0);
692 #endif
693 /* expect a disconnect here */
694 /* get the request */
695 rc = sread(fd, (char *)&buffer[0], 2);
696
697 logmsg("READ %zd bytes [DISCONNECT]", rc);
698 loghex(buffer, rc);
699 logprotocol(FROM_CLIENT, "DISCONNECT", 0, dump, buffer, rc);
700 goto end;
701 }
702 else {
703 /* not supported (yet) */
704 goto end;
705 }
706 } while(1);
707
708 end:
709 if(buffer)
710 free(buffer);
711 if(dump)
712 fclose(dump);
713 if(stream)
714 fclose(stream);
715 return CURL_SOCKET_BAD;
716 }
717
718 /*
719 sockfdp is a pointer to an established stream or CURL_SOCKET_BAD
720
721 if sockfd is CURL_SOCKET_BAD, listendfd is a listening socket we must
722 accept()
723 */
incoming(curl_socket_t listenfd)724 static bool incoming(curl_socket_t listenfd)
725 {
726 fd_set fds_read;
727 fd_set fds_write;
728 fd_set fds_err;
729 int clients = 0; /* connected clients */
730
731 if(got_exit_signal) {
732 logmsg("signalled to die, exiting...");
733 return FALSE;
734 }
735
736 #ifdef HAVE_GETPPID
737 /* As a last resort, quit if socks5 process becomes orphan. */
738 if(getppid() <= 1) {
739 logmsg("process becomes orphan, exiting");
740 return FALSE;
741 }
742 #endif
743
744 do {
745 ssize_t rc;
746 int error = 0;
747 curl_socket_t sockfd = listenfd;
748 int maxfd = (int)sockfd;
749
750 FD_ZERO(&fds_read);
751 FD_ZERO(&fds_write);
752 FD_ZERO(&fds_err);
753
754 /* there's always a socket to wait for */
755 #if defined(__DJGPP__)
756 #pragma GCC diagnostic push
757 #pragma GCC diagnostic ignored "-Warith-conversion"
758 #endif
759 FD_SET(sockfd, &fds_read);
760 #if defined(__DJGPP__)
761 #pragma GCC diagnostic pop
762 #endif
763
764 do {
765 /* select() blocking behavior call on blocking descriptors please */
766 rc = select(maxfd + 1, &fds_read, &fds_write, &fds_err, NULL);
767 if(got_exit_signal) {
768 logmsg("signalled to die, exiting...");
769 return FALSE;
770 }
771 } while((rc == -1) && ((error = SOCKERRNO) == EINTR));
772
773 if(rc < 0) {
774 logmsg("select() failed with error: (%d) %s",
775 error, strerror(error));
776 return FALSE;
777 }
778
779 if(FD_ISSET(sockfd, &fds_read)) {
780 curl_socket_t newfd = accept(sockfd, NULL, NULL);
781 if(CURL_SOCKET_BAD == newfd) {
782 error = SOCKERRNO;
783 logmsg("accept(%" FMT_SOCKET_T ", NULL, NULL) "
784 "failed with error: (%d) %s", sockfd, error, sstrerror(error));
785 }
786 else {
787 logmsg("====> Client connect, fd %" FMT_SOCKET_T ". "
788 "Read config from %s", newfd, configfile);
789 set_advisor_read_lock(loglockfile);
790 (void)mqttit(newfd); /* until done */
791 clear_advisor_read_lock(loglockfile);
792
793 logmsg("====> Client disconnect");
794 sclose(newfd);
795 }
796 }
797 } while(clients);
798
799 return TRUE;
800 }
801
sockdaemon(curl_socket_t sock,unsigned short * listenport)802 static curl_socket_t sockdaemon(curl_socket_t sock,
803 unsigned short *listenport)
804 {
805 /* passive daemon style */
806 srvr_sockaddr_union_t listener;
807 int flag;
808 int rc;
809 int totdelay = 0;
810 int maxretr = 10;
811 int delay = 20;
812 int attempt = 0;
813 int error = 0;
814
815 do {
816 attempt++;
817 flag = 1;
818 rc = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
819 (void *)&flag, sizeof(flag));
820 if(rc) {
821 error = SOCKERRNO;
822 logmsg("setsockopt(SO_REUSEADDR) failed with error: (%d) %s",
823 error, sstrerror(error));
824 if(maxretr) {
825 rc = wait_ms(delay);
826 if(rc) {
827 /* should not happen */
828 logmsg("wait_ms() failed with error: %d", rc);
829 sclose(sock);
830 return CURL_SOCKET_BAD;
831 }
832 if(got_exit_signal) {
833 logmsg("signalled to die, exiting...");
834 sclose(sock);
835 return CURL_SOCKET_BAD;
836 }
837 totdelay += delay;
838 delay *= 2; /* double the sleep for next attempt */
839 }
840 }
841 } while(rc && maxretr--);
842
843 if(rc) {
844 logmsg("setsockopt(SO_REUSEADDR) failed %d times in %d ms. Error: (%d) %s",
845 attempt, totdelay, error, strerror(error));
846 logmsg("Continuing anyway...");
847 }
848
849 /* When the specified listener port is zero, it is actually a
850 request to let the system choose a non-zero available port. */
851
852 #ifdef USE_IPV6
853 if(!use_ipv6) {
854 #endif
855 memset(&listener.sa4, 0, sizeof(listener.sa4));
856 listener.sa4.sin_family = AF_INET;
857 listener.sa4.sin_addr.s_addr = INADDR_ANY;
858 listener.sa4.sin_port = htons(*listenport);
859 rc = bind(sock, &listener.sa, sizeof(listener.sa4));
860 #ifdef USE_IPV6
861 }
862 else {
863 memset(&listener.sa6, 0, sizeof(listener.sa6));
864 listener.sa6.sin6_family = AF_INET6;
865 listener.sa6.sin6_addr = in6addr_any;
866 listener.sa6.sin6_port = htons(*listenport);
867 rc = bind(sock, &listener.sa, sizeof(listener.sa6));
868 }
869 #endif /* USE_IPV6 */
870 if(rc) {
871 error = SOCKERRNO;
872 logmsg("Error binding socket on port %hu: (%d) %s",
873 *listenport, error, sstrerror(error));
874 sclose(sock);
875 return CURL_SOCKET_BAD;
876 }
877
878 if(!*listenport) {
879 /* The system was supposed to choose a port number, figure out which
880 port we actually got and update the listener port value with it. */
881 curl_socklen_t la_size;
882 srvr_sockaddr_union_t localaddr;
883 #ifdef USE_IPV6
884 if(!use_ipv6)
885 #endif
886 la_size = sizeof(localaddr.sa4);
887 #ifdef USE_IPV6
888 else
889 la_size = sizeof(localaddr.sa6);
890 #endif
891 memset(&localaddr.sa, 0, (size_t)la_size);
892 if(getsockname(sock, &localaddr.sa, &la_size) < 0) {
893 error = SOCKERRNO;
894 logmsg("getsockname() failed with error: (%d) %s",
895 error, sstrerror(error));
896 sclose(sock);
897 return CURL_SOCKET_BAD;
898 }
899 switch(localaddr.sa.sa_family) {
900 case AF_INET:
901 *listenport = ntohs(localaddr.sa4.sin_port);
902 break;
903 #ifdef USE_IPV6
904 case AF_INET6:
905 *listenport = ntohs(localaddr.sa6.sin6_port);
906 break;
907 #endif
908 default:
909 break;
910 }
911 if(!*listenport) {
912 /* Real failure, listener port shall not be zero beyond this point. */
913 logmsg("Apparently getsockname() succeeded, with listener port zero.");
914 logmsg("A valid reason for this failure is a binary built without");
915 logmsg("proper network library linkage. This might not be the only");
916 logmsg("reason, but double check it before anything else.");
917 sclose(sock);
918 return CURL_SOCKET_BAD;
919 }
920 }
921
922 /* start accepting connections */
923 rc = listen(sock, 5);
924 if(0 != rc) {
925 error = SOCKERRNO;
926 logmsg("listen(%" FMT_SOCKET_T ", 5) failed with error: (%d) %s",
927 sock, error, sstrerror(error));
928 sclose(sock);
929 return CURL_SOCKET_BAD;
930 }
931
932 return sock;
933 }
934
935
main(int argc,char * argv[])936 int main(int argc, char *argv[])
937 {
938 curl_socket_t sock = CURL_SOCKET_BAD;
939 curl_socket_t msgsock = CURL_SOCKET_BAD;
940 int wrotepidfile = 0;
941 int wroteportfile = 0;
942 const char *pidname = ".mqttd.pid";
943 const char *portname = ".mqttd.port";
944 bool juggle_again;
945 int error;
946 int arg = 1;
947
948 while(argc > arg) {
949 if(!strcmp("--version", argv[arg])) {
950 printf("mqttd IPv4%s\n",
951 #ifdef USE_IPV6
952 "/IPv6"
953 #else
954 ""
955 #endif
956 );
957 return 0;
958 }
959 else if(!strcmp("--pidfile", argv[arg])) {
960 arg++;
961 if(argc > arg)
962 pidname = argv[arg++];
963 }
964 else if(!strcmp("--portfile", argv[arg])) {
965 arg++;
966 if(argc > arg)
967 portname = argv[arg++];
968 }
969 else if(!strcmp("--config", argv[arg])) {
970 arg++;
971 if(argc > arg)
972 configfile = argv[arg++];
973 }
974 else if(!strcmp("--logfile", argv[arg])) {
975 arg++;
976 if(argc > arg)
977 serverlogfile = argv[arg++];
978 }
979 else if(!strcmp("--logdir", argv[arg])) {
980 arg++;
981 if(argc > arg)
982 logdir = argv[arg++];
983 }
984 else if(!strcmp("--ipv6", argv[arg])) {
985 #ifdef USE_IPV6
986 ipv_inuse = "IPv6";
987 use_ipv6 = TRUE;
988 #endif
989 arg++;
990 }
991 else if(!strcmp("--ipv4", argv[arg])) {
992 /* for completeness, we support this option as well */
993 #ifdef USE_IPV6
994 ipv_inuse = "IPv4";
995 use_ipv6 = FALSE;
996 #endif
997 arg++;
998 }
999 else if(!strcmp("--port", argv[arg])) {
1000 arg++;
1001 if(argc > arg) {
1002 char *endptr;
1003 unsigned long ulnum = strtoul(argv[arg], &endptr, 10);
1004 if((endptr != argv[arg] + strlen(argv[arg])) ||
1005 ((ulnum != 0UL) && ((ulnum < 1025UL) || (ulnum > 65535UL)))) {
1006 fprintf(stderr, "mqttd: invalid --port argument (%s)\n",
1007 argv[arg]);
1008 return 0;
1009 }
1010 port = curlx_ultous(ulnum);
1011 arg++;
1012 }
1013 }
1014 else {
1015 puts("Usage: mqttd [option]\n"
1016 " --config [file]\n"
1017 " --version\n"
1018 " --logfile [file]\n"
1019 " --logdir [directory]\n"
1020 " --pidfile [file]\n"
1021 " --portfile [file]\n"
1022 " --ipv4\n"
1023 " --ipv6\n"
1024 " --port [port]\n");
1025 return 0;
1026 }
1027 }
1028
1029 msnprintf(loglockfile, sizeof(loglockfile), "%s/%s/mqtt-%s.lock",
1030 logdir, SERVERLOGS_LOCKDIR, ipv_inuse);
1031
1032 #ifdef _WIN32
1033 win32_init();
1034 atexit(win32_cleanup);
1035 #endif
1036
1037 CURL_SET_BINMODE(stdin);
1038 CURL_SET_BINMODE(stdout);
1039 CURL_SET_BINMODE(stderr);
1040
1041 install_signal_handlers(FALSE);
1042
1043 #ifdef USE_IPV6
1044 if(!use_ipv6)
1045 #endif
1046 sock = socket(AF_INET, SOCK_STREAM, 0);
1047 #ifdef USE_IPV6
1048 else
1049 sock = socket(AF_INET6, SOCK_STREAM, 0);
1050 #endif
1051
1052 if(CURL_SOCKET_BAD == sock) {
1053 error = SOCKERRNO;
1054 logmsg("Error creating socket: (%d) %s", error, sstrerror(error));
1055 goto mqttd_cleanup;
1056 }
1057
1058 {
1059 /* passive daemon style */
1060 sock = sockdaemon(sock, &port);
1061 if(CURL_SOCKET_BAD == sock) {
1062 goto mqttd_cleanup;
1063 }
1064 msgsock = CURL_SOCKET_BAD; /* no stream socket yet */
1065 }
1066
1067 logmsg("Running %s version", ipv_inuse);
1068 logmsg("Listening on port %hu", port);
1069
1070 wrotepidfile = write_pidfile(pidname);
1071 if(!wrotepidfile) {
1072 goto mqttd_cleanup;
1073 }
1074
1075 wroteportfile = write_portfile(portname, port);
1076 if(!wroteportfile) {
1077 goto mqttd_cleanup;
1078 }
1079
1080 do {
1081 juggle_again = incoming(sock);
1082 } while(juggle_again);
1083
1084 mqttd_cleanup:
1085
1086 if((msgsock != sock) && (msgsock != CURL_SOCKET_BAD))
1087 sclose(msgsock);
1088
1089 if(sock != CURL_SOCKET_BAD)
1090 sclose(sock);
1091
1092 if(wrotepidfile)
1093 unlink(pidname);
1094 if(wroteportfile)
1095 unlink(portname);
1096
1097 restore_signal_handlers(FALSE);
1098
1099 if(got_exit_signal) {
1100 logmsg("============> mqttd exits with signal (%d)", exit_signal);
1101 /*
1102 * To properly set the return status of the process we
1103 * must raise the same signal SIGINT or SIGTERM that we
1104 * caught and let the old handler take care of it.
1105 */
1106 raise(exit_signal);
1107 }
1108
1109 logmsg("============> mqttd quits");
1110 return 0;
1111 }
1112