mqttd.c 28 KB

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