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 buff_size = remaining_length;
546 buffer = realloc(buffer, buff_size);
547 if(!buffer) {
548 logmsg("Failed realloc of size %zu", buff_size);
549 goto end;
550 }
551 }
552
553 if(remaining_length) {
554 /* reading variable header and payload into buffer */
555 rc = sread(fd, (char *)buffer, remaining_length);
556 if(rc > 0) {
557 logmsg("READ %zd bytes", rc);
558 loghex(buffer, rc);
559 }
560 }
561
562 if(byte == MQTT_MSG_CONNECT) {
563 logprotocol(FROM_CLIENT, "CONNECT", remaining_length,
564 dump, buffer, rc);
565
566 if(memcmp(protocol, buffer, sizeof(protocol))) {
567 logmsg("Protocol preamble mismatch");
568 goto end;
569 }
570 /* ignore the connect flag byte and two keepalive bytes */
571 payload_len = (size_t)(buffer[10] << 8) | buffer[11];
572 /* first part of the payload is the client ID */
573 client_id_length = payload_len;
574
575 /* checking if user and password flags were set */
576 conn_flags = buffer[7];
577
578 start_usr = client_id_offset + payload_len;
579 if(usr_flag == (unsigned char)(conn_flags & usr_flag)) {
580 logmsg("User flag is present in CONN flag");
581 payload_len += (size_t)(buffer[start_usr] << 8) |
582 buffer[start_usr + 1];
583 payload_len += 2; /* MSB and LSB for user length */
584 }
585
586 start_passwd = client_id_offset + payload_len;
587 if(passwd_flag == (char)(conn_flags & passwd_flag)) {
588 logmsg("Password flag is present in CONN flags");
589 payload_len += (size_t)(buffer[start_passwd] << 8) |
590 buffer[start_passwd + 1];
591 payload_len += 2; /* MSB and LSB for password length */
592 }
593
594 /* check the length of the payload */
595 if((ssize_t)payload_len != (rc - 12)) {
596 logmsg("Payload length mismatch, expected %zx got %zx",
597 rc - 12, payload_len);
598 goto end;
599 }
600 /* check the length of the client ID */
601 else if((client_id_length + 1) > MAX_CLIENT_ID_LENGTH) {
602 logmsg("Too large client id");
603 goto end;
604 }
605 memcpy(client_id, &buffer[12], client_id_length);
606 client_id[client_id_length] = 0;
607
608 logmsg("MQTT client connect accepted: %s", client_id);
609
610 /* The first packet sent from the Server to the Client MUST be a
611 CONNACK Packet */
612
613 if(connack(dump, fd)) {
614 logmsg("failed sending CONNACK");
615 goto end;
616 }
617 }
618 else if(byte == MQTT_MSG_SUBSCRIBE) {
619 int error;
620 char *data;
621 size_t datalen;
622 logprotocol(FROM_CLIENT, "SUBSCRIBE", remaining_length,
623 dump, buffer, rc);
624 logmsg("Incoming SUBSCRIBE");
625
626 if(rc < 6) {
627 logmsg("Too small SUBSCRIBE");
628 goto end;
629 }
630
631 /* two bytes packet id */
632 packet_id = (unsigned short)((buffer[0] << 8) | buffer[1]);
633
634 /* two bytes topic length */
635 topic_len = (size_t)(buffer[2] << 8) | buffer[3];
636 if(topic_len != (remaining_length - 5)) {
637 logmsg("Wrong topic length, got %zu expected %zu",
638 topic_len, remaining_length - 5);
639 goto end;
640 }
641 memcpy(topic, &buffer[4], topic_len);
642 topic[topic_len] = 0;
643
644 /* there's a QoS byte (two bits) after the topic */
645
646 logmsg("SUBSCRIBE to '%s' [%d]", topic, packet_id);
647 stream = test2fopen(testno, logdir);
648 error = getpart(&data, &datalen, "reply", "data", stream);
649 if(!error) {
650 if(!config.publish_before_suback) {
651 if(suback(dump, fd, packet_id)) {
652 logmsg("failed sending SUBACK");
653 free(data);
654 goto end;
655 }
656 }
657 if(publish(dump, fd, packet_id, topic, data, datalen)) {
658 logmsg("PUBLISH failed");
659 free(data);
660 goto end;
661 }
662 free(data);
663 if(config.publish_before_suback) {
664 if(suback(dump, fd, packet_id)) {
665 logmsg("failed sending SUBACK");
666 goto end;
667 }
668 }
669 }
670 else {
671 char *def = (char *)"this is random payload yes yes it is";
672 publish(dump, fd, packet_id, topic, def, strlen(def));
673 }
674 disconnect(dump, fd);
675 }
676 else if((byte & 0xf0) == (MQTT_MSG_PUBLISH & 0xf0)) {
677 size_t topiclen;
678
679 logmsg("Incoming PUBLISH");
680 logprotocol(FROM_CLIENT, "PUBLISH", remaining_length,
681 dump, buffer, rc);
682
683 topiclen = (size_t)(buffer[1 + bytes] << 8) | buffer[2 + bytes];
684 logmsg("Got %zu bytes topic", topiclen);
685 /* TODO: verify topiclen */
686
687 #ifdef QOS
688 /* TODO: handle packetid if there is one. Send puback if QoS > 0 */
689 puback(dump, fd, 0);
690 #endif
691 /* expect a disconnect here */
692 /* get the request */
693 rc = sread(fd, (char *)&buffer[0], 2);
694
695 logmsg("READ %zd bytes [DISCONNECT]", rc);
696 loghex(buffer, rc);
697 logprotocol(FROM_CLIENT, "DISCONNECT", 0, dump, buffer, rc);
698 goto end;
699 }
700 else {
701 /* not supported (yet) */
702 goto end;
703 }
704 } while(1);
705
706 end:
707 if(buffer)
708 free(buffer);
709 if(dump)
710 fclose(dump);
711 if(stream)
712 fclose(stream);
713 return CURL_SOCKET_BAD;
714 }
715
716 /*
717 sockfdp is a pointer to an established stream or CURL_SOCKET_BAD
718
719 if sockfd is CURL_SOCKET_BAD, listendfd is a listening socket we must
720 accept()
721 */
incoming(curl_socket_t listenfd)722 static bool incoming(curl_socket_t listenfd)
723 {
724 fd_set fds_read;
725 fd_set fds_write;
726 fd_set fds_err;
727 int clients = 0; /* connected clients */
728
729 if(got_exit_signal) {
730 logmsg("signalled to die, exiting...");
731 return FALSE;
732 }
733
734 #ifdef HAVE_GETPPID
735 /* As a last resort, quit if socks5 process becomes orphan. */
736 if(getppid() <= 1) {
737 logmsg("process becomes orphan, exiting");
738 return FALSE;
739 }
740 #endif
741
742 do {
743 ssize_t rc;
744 int error = 0;
745 curl_socket_t sockfd = listenfd;
746 int maxfd = (int)sockfd;
747
748 FD_ZERO(&fds_read);
749 FD_ZERO(&fds_write);
750 FD_ZERO(&fds_err);
751
752 /* there's always a socket to wait for */
753 #if defined(__DJGPP__)
754 #pragma GCC diagnostic push
755 #pragma GCC diagnostic ignored "-Warith-conversion"
756 #endif
757 FD_SET(sockfd, &fds_read);
758 #if defined(__DJGPP__)
759 #pragma GCC diagnostic pop
760 #endif
761
762 do {
763 /* select() blocking behavior call on blocking descriptors please */
764 rc = select(maxfd + 1, &fds_read, &fds_write, &fds_err, NULL);
765 if(got_exit_signal) {
766 logmsg("signalled to die, exiting...");
767 return FALSE;
768 }
769 } while((rc == -1) && ((error = SOCKERRNO) == EINTR));
770
771 if(rc < 0) {
772 logmsg("select() failed with error: (%d) %s",
773 error, strerror(error));
774 return FALSE;
775 }
776
777 if(FD_ISSET(sockfd, &fds_read)) {
778 curl_socket_t newfd = accept(sockfd, NULL, NULL);
779 if(CURL_SOCKET_BAD == newfd) {
780 error = SOCKERRNO;
781 logmsg("accept(%" FMT_SOCKET_T ", NULL, NULL) "
782 "failed with error: (%d) %s", sockfd, error, sstrerror(error));
783 }
784 else {
785 logmsg("====> Client connect, fd %" FMT_SOCKET_T ". "
786 "Read config from %s", newfd, configfile);
787 set_advisor_read_lock(loglockfile);
788 (void)mqttit(newfd); /* until done */
789 clear_advisor_read_lock(loglockfile);
790
791 logmsg("====> Client disconnect");
792 sclose(newfd);
793 }
794 }
795 } while(clients);
796
797 return TRUE;
798 }
799
sockdaemon(curl_socket_t sock,unsigned short * listenport)800 static curl_socket_t sockdaemon(curl_socket_t sock,
801 unsigned short *listenport)
802 {
803 /* passive daemon style */
804 srvr_sockaddr_union_t listener;
805 int flag;
806 int rc;
807 int totdelay = 0;
808 int maxretr = 10;
809 int delay = 20;
810 int attempt = 0;
811 int error = 0;
812
813 do {
814 attempt++;
815 flag = 1;
816 rc = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
817 (void *)&flag, sizeof(flag));
818 if(rc) {
819 error = SOCKERRNO;
820 logmsg("setsockopt(SO_REUSEADDR) failed with error: (%d) %s",
821 error, sstrerror(error));
822 if(maxretr) {
823 rc = wait_ms(delay);
824 if(rc) {
825 /* should not happen */
826 logmsg("wait_ms() failed with error: %d", rc);
827 sclose(sock);
828 return CURL_SOCKET_BAD;
829 }
830 if(got_exit_signal) {
831 logmsg("signalled to die, exiting...");
832 sclose(sock);
833 return CURL_SOCKET_BAD;
834 }
835 totdelay += delay;
836 delay *= 2; /* double the sleep for next attempt */
837 }
838 }
839 } while(rc && maxretr--);
840
841 if(rc) {
842 logmsg("setsockopt(SO_REUSEADDR) failed %d times in %d ms. Error: (%d) %s",
843 attempt, totdelay, error, strerror(error));
844 logmsg("Continuing anyway...");
845 }
846
847 /* When the specified listener port is zero, it is actually a
848 request to let the system choose a non-zero available port. */
849
850 #ifdef USE_IPV6
851 if(!use_ipv6) {
852 #endif
853 memset(&listener.sa4, 0, sizeof(listener.sa4));
854 listener.sa4.sin_family = AF_INET;
855 listener.sa4.sin_addr.s_addr = INADDR_ANY;
856 listener.sa4.sin_port = htons(*listenport);
857 rc = bind(sock, &listener.sa, sizeof(listener.sa4));
858 #ifdef USE_IPV6
859 }
860 else {
861 memset(&listener.sa6, 0, sizeof(listener.sa6));
862 listener.sa6.sin6_family = AF_INET6;
863 listener.sa6.sin6_addr = in6addr_any;
864 listener.sa6.sin6_port = htons(*listenport);
865 rc = bind(sock, &listener.sa, sizeof(listener.sa6));
866 }
867 #endif /* USE_IPV6 */
868 if(rc) {
869 error = SOCKERRNO;
870 logmsg("Error binding socket on port %hu: (%d) %s",
871 *listenport, error, sstrerror(error));
872 sclose(sock);
873 return CURL_SOCKET_BAD;
874 }
875
876 if(!*listenport) {
877 /* The system was supposed to choose a port number, figure out which
878 port we actually got and update the listener port value with it. */
879 curl_socklen_t la_size;
880 srvr_sockaddr_union_t localaddr;
881 #ifdef USE_IPV6
882 if(!use_ipv6)
883 #endif
884 la_size = sizeof(localaddr.sa4);
885 #ifdef USE_IPV6
886 else
887 la_size = sizeof(localaddr.sa6);
888 #endif
889 memset(&localaddr.sa, 0, (size_t)la_size);
890 if(getsockname(sock, &localaddr.sa, &la_size) < 0) {
891 error = SOCKERRNO;
892 logmsg("getsockname() failed with error: (%d) %s",
893 error, sstrerror(error));
894 sclose(sock);
895 return CURL_SOCKET_BAD;
896 }
897 switch(localaddr.sa.sa_family) {
898 case AF_INET:
899 *listenport = ntohs(localaddr.sa4.sin_port);
900 break;
901 #ifdef USE_IPV6
902 case AF_INET6:
903 *listenport = ntohs(localaddr.sa6.sin6_port);
904 break;
905 #endif
906 default:
907 break;
908 }
909 if(!*listenport) {
910 /* Real failure, listener port shall not be zero beyond this point. */
911 logmsg("Apparently getsockname() succeeded, with listener port zero.");
912 logmsg("A valid reason for this failure is a binary built without");
913 logmsg("proper network library linkage. This might not be the only");
914 logmsg("reason, but double check it before anything else.");
915 sclose(sock);
916 return CURL_SOCKET_BAD;
917 }
918 }
919
920 /* start accepting connections */
921 rc = listen(sock, 5);
922 if(0 != rc) {
923 error = SOCKERRNO;
924 logmsg("listen(%" FMT_SOCKET_T ", 5) failed with error: (%d) %s",
925 sock, error, sstrerror(error));
926 sclose(sock);
927 return CURL_SOCKET_BAD;
928 }
929
930 return sock;
931 }
932
933
main(int argc,char * argv[])934 int main(int argc, char *argv[])
935 {
936 curl_socket_t sock = CURL_SOCKET_BAD;
937 curl_socket_t msgsock = CURL_SOCKET_BAD;
938 int wrotepidfile = 0;
939 int wroteportfile = 0;
940 const char *pidname = ".mqttd.pid";
941 const char *portname = ".mqttd.port";
942 bool juggle_again;
943 int error;
944 int arg = 1;
945
946 while(argc > arg) {
947 if(!strcmp("--version", argv[arg])) {
948 printf("mqttd IPv4%s\n",
949 #ifdef USE_IPV6
950 "/IPv6"
951 #else
952 ""
953 #endif
954 );
955 return 0;
956 }
957 else if(!strcmp("--pidfile", argv[arg])) {
958 arg++;
959 if(argc > arg)
960 pidname = argv[arg++];
961 }
962 else if(!strcmp("--portfile", argv[arg])) {
963 arg++;
964 if(argc > arg)
965 portname = argv[arg++];
966 }
967 else if(!strcmp("--config", argv[arg])) {
968 arg++;
969 if(argc > arg)
970 configfile = argv[arg++];
971 }
972 else if(!strcmp("--logfile", argv[arg])) {
973 arg++;
974 if(argc > arg)
975 serverlogfile = argv[arg++];
976 }
977 else if(!strcmp("--logdir", argv[arg])) {
978 arg++;
979 if(argc > arg)
980 logdir = argv[arg++];
981 }
982 else if(!strcmp("--ipv6", argv[arg])) {
983 #ifdef USE_IPV6
984 ipv_inuse = "IPv6";
985 use_ipv6 = TRUE;
986 #endif
987 arg++;
988 }
989 else if(!strcmp("--ipv4", argv[arg])) {
990 /* for completeness, we support this option as well */
991 #ifdef USE_IPV6
992 ipv_inuse = "IPv4";
993 use_ipv6 = FALSE;
994 #endif
995 arg++;
996 }
997 else if(!strcmp("--port", argv[arg])) {
998 arg++;
999 if(argc > arg) {
1000 char *endptr;
1001 unsigned long ulnum = strtoul(argv[arg], &endptr, 10);
1002 if((endptr != argv[arg] + strlen(argv[arg])) ||
1003 ((ulnum != 0UL) && ((ulnum < 1025UL) || (ulnum > 65535UL)))) {
1004 fprintf(stderr, "mqttd: invalid --port argument (%s)\n",
1005 argv[arg]);
1006 return 0;
1007 }
1008 port = curlx_ultous(ulnum);
1009 arg++;
1010 }
1011 }
1012 else {
1013 puts("Usage: mqttd [option]\n"
1014 " --config [file]\n"
1015 " --version\n"
1016 " --logfile [file]\n"
1017 " --logdir [directory]\n"
1018 " --pidfile [file]\n"
1019 " --portfile [file]\n"
1020 " --ipv4\n"
1021 " --ipv6\n"
1022 " --port [port]\n");
1023 return 0;
1024 }
1025 }
1026
1027 msnprintf(loglockfile, sizeof(loglockfile), "%s/%s/mqtt-%s.lock",
1028 logdir, SERVERLOGS_LOCKDIR, ipv_inuse);
1029
1030 #ifdef _WIN32
1031 win32_init();
1032 atexit(win32_cleanup);
1033 #endif
1034
1035 CURL_SET_BINMODE(stdin);
1036 CURL_SET_BINMODE(stdout);
1037 CURL_SET_BINMODE(stderr);
1038
1039 install_signal_handlers(FALSE);
1040
1041 #ifdef USE_IPV6
1042 if(!use_ipv6)
1043 #endif
1044 sock = socket(AF_INET, SOCK_STREAM, 0);
1045 #ifdef USE_IPV6
1046 else
1047 sock = socket(AF_INET6, SOCK_STREAM, 0);
1048 #endif
1049
1050 if(CURL_SOCKET_BAD == sock) {
1051 error = SOCKERRNO;
1052 logmsg("Error creating socket: (%d) %s", error, sstrerror(error));
1053 goto mqttd_cleanup;
1054 }
1055
1056 {
1057 /* passive daemon style */
1058 sock = sockdaemon(sock, &port);
1059 if(CURL_SOCKET_BAD == sock) {
1060 goto mqttd_cleanup;
1061 }
1062 msgsock = CURL_SOCKET_BAD; /* no stream socket yet */
1063 }
1064
1065 logmsg("Running %s version", ipv_inuse);
1066 logmsg("Listening on port %hu", port);
1067
1068 wrotepidfile = write_pidfile(pidname);
1069 if(!wrotepidfile) {
1070 goto mqttd_cleanup;
1071 }
1072
1073 wroteportfile = write_portfile(portname, port);
1074 if(!wroteportfile) {
1075 goto mqttd_cleanup;
1076 }
1077
1078 do {
1079 juggle_again = incoming(sock);
1080 } while(juggle_again);
1081
1082 mqttd_cleanup:
1083
1084 if((msgsock != sock) && (msgsock != CURL_SOCKET_BAD))
1085 sclose(msgsock);
1086
1087 if(sock != CURL_SOCKET_BAD)
1088 sclose(sock);
1089
1090 if(wrotepidfile)
1091 unlink(pidname);
1092 if(wroteportfile)
1093 unlink(portname);
1094
1095 restore_signal_handlers(FALSE);
1096
1097 if(got_exit_signal) {
1098 logmsg("============> mqttd exits with signal (%d)", exit_signal);
1099 /*
1100 * To properly set the return status of the process we
1101 * must raise the same signal SIGINT or SIGTERM that we
1102 * caught and let the old handler take care of it.
1103 */
1104 raise(exit_signal);
1105 }
1106
1107 logmsg("============> mqttd quits");
1108 return 0;
1109 }
1110
