mqttd.c 28 KB

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