root.cpp 39 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146
  1. /*
  2. * Copyright (c)2019 ZeroTier, Inc.
  3. *
  4. * Use of this software is governed by the Business Source License included
  5. * in the LICENSE.TXT file in the project's root directory.
  6. *
  7. * Change Date: 2023-01-01
  8. *
  9. * On the date above, in accordance with the Business Source License, use
  10. * of this software will be governed by version 2.0 of the Apache License.
  11. */
  12. /****/
  13. /*
  14. * This is a high-throughput minimal root server. It implements only
  15. * those functions of a ZT node that a root must perform and does so
  16. * using highly efficient multithreaded I/O code. It's only been
  17. * thoroughly tested on Linux but should also run on BSDs.
  18. *
  19. * Root configuration file format (JSON):
  20. *
  21. * {
  22. * "name": Name of this root for documentation/UI purposes (string)
  23. * "port": UDP port (int)
  24. * "httpPort": Local HTTP port for basic stats (int)
  25. * "statsRoot": If present, path to periodically save stats files (string)
  26. * "siblings": [
  27. * {
  28. * "name": Sibling name for UI/documentation purposes (string)
  29. * "id": Full public identity of subling (string)
  30. * "ip": IP address of sibling (string)
  31. * "port": port of subling (for ZeroTier UDP) (int)
  32. * }, ...
  33. * ]
  34. * }
  35. *
  36. * The only required field is port. If statsRoot is present then files
  37. * are periodically written there containing the root's current state.
  38. * It should be a memory filesystem like /dev/shm on Linux as these
  39. * files are large and rewritten frequently and do not need to be
  40. * persisted.
  41. *
  42. * Siblings are other root servers that should receive packets to peers
  43. * that we can't find. This can occur due to e.g. network topology
  44. * hiccups, IP blockages, etc. Siblings are used in the order in which
  45. * they appear with the first alive sibling being used.
  46. */
  47. #include <Constants.hpp>
  48. #include <stdio.h>
  49. #include <stdlib.h>
  50. #include <unistd.h>
  51. #include <string.h>
  52. #include <fcntl.h>
  53. #include <signal.h>
  54. #include <errno.h>
  55. #include <sys/stat.h>
  56. #include <sys/types.h>
  57. #include <sys/socket.h>
  58. #include <sys/select.h>
  59. #include <sys/time.h>
  60. #include <sys/un.h>
  61. #include <sys/ioctl.h>
  62. #include <arpa/inet.h>
  63. #include <netinet/in.h>
  64. #include <netinet/ip.h>
  65. #include <netinet/ip6.h>
  66. #include <netinet/tcp.h>
  67. #include <netinet/udp.h>
  68. #include <json.hpp>
  69. #include <httplib.h>
  70. #include <Packet.hpp>
  71. #include <Utils.hpp>
  72. #include <Address.hpp>
  73. #include <Identity.hpp>
  74. #include <InetAddress.hpp>
  75. #include <Mutex.hpp>
  76. #include <SharedPtr.hpp>
  77. #include <MulticastGroup.hpp>
  78. #include <CertificateOfMembership.hpp>
  79. #include <OSUtils.hpp>
  80. #include <Meter.hpp>
  81. #include <string>
  82. #include <thread>
  83. #include <map>
  84. #include <set>
  85. #include <vector>
  86. #include <iostream>
  87. #include <unordered_map>
  88. #include <unordered_set>
  89. #include <vector>
  90. #include <atomic>
  91. #include <mutex>
  92. #include <sstream>
  93. using namespace ZeroTier;
  94. using json = nlohmann::json;
  95. #ifdef MSG_DONTWAIT
  96. #define SENDTO_FLAGS MSG_DONTWAIT
  97. #else
  98. #define SENDTO_FLAGS 0
  99. #endif
  100. //////////////////////////////////////////////////////////////////////////////
  101. //////////////////////////////////////////////////////////////////////////////
  102. // Hashers for std::unordered_map
  103. struct IdentityHasher { ZT_ALWAYS_INLINE std::size_t operator()(const Identity &id) const { return (std::size_t)id.hashCode(); } };
  104. struct AddressHasher { ZT_ALWAYS_INLINE std::size_t operator()(const Address &a) const { return (std::size_t)a.toInt(); } };
  105. struct InetAddressHasher { ZT_ALWAYS_INLINE std::size_t operator()(const InetAddress &ip) const { return (std::size_t)ip.hashCode(); } };
  106. struct MulticastGroupHasher { ZT_ALWAYS_INLINE std::size_t operator()(const MulticastGroup &mg) const { return (std::size_t)mg.hashCode(); } };
  107. // An ordered tuple key representing an introduction of one peer to another
  108. struct RendezvousKey
  109. {
  110. RendezvousKey(const Address &aa,const Address &bb)
  111. {
  112. if (aa > bb) {
  113. a = aa;
  114. b = bb;
  115. } else {
  116. a = bb;
  117. b = aa;
  118. }
  119. }
  120. Address a,b;
  121. ZT_ALWAYS_INLINE bool operator==(const RendezvousKey &k) const { return ((a == k.a)&&(b == k.b)); }
  122. ZT_ALWAYS_INLINE bool operator!=(const RendezvousKey &k) const { return ((a != k.a)||(b != k.b)); }
  123. struct Hasher { ZT_ALWAYS_INLINE std::size_t operator()(const RendezvousKey &k) const { return (std::size_t)(k.a.toInt() ^ k.b.toInt()); } };
  124. };
  125. /**
  126. * RootPeer is a normal peer known to this root
  127. *
  128. * This can also be a sibling root, which is itself a peer. Sibling roots
  129. * are sent HELLO while for other peers we only listen for HELLO.
  130. */
  131. struct RootPeer
  132. {
  133. ZT_ALWAYS_INLINE RootPeer() : lastSend(0),lastReceive(0),lastSync(0),lastEcho(0),lastHello(0),vMajor(-1),vMinor(-1),vRev(-1) {}
  134. ZT_ALWAYS_INLINE ~RootPeer() { Utils::burn(key,sizeof(key)); }
  135. Identity id; // Identity
  136. uint8_t key[32]; // Shared secret key
  137. InetAddress ip4,ip6; // IPv4 and IPv6 addresses
  138. int64_t lastSend; // Time of last send (any packet)
  139. int64_t lastReceive; // Time of last receive (any packet)
  140. int64_t lastSync; // Time of last data synchronization with LF or other root state backend (currently unused)
  141. int64_t lastEcho; // Time of last received ECHO
  142. int64_t lastHello; // Time of last received HELLO
  143. int vMajor,vMinor,vRev; // Peer version or -1,-1,-1 if unknown
  144. bool sibling; // If true, this is a sibling root that will get forwards we don't know where to send
  145. std::mutex lock;
  146. AtomicCounter __refCount;
  147. };
  148. static int64_t startTime;
  149. static std::vector<int> ports;
  150. static Identity self;
  151. static std::atomic_bool run;
  152. static json config;
  153. static std::string statsRoot;
  154. static Meter inputRate;
  155. static Meter outputRate;
  156. static Meter forwardRate;
  157. static Meter siblingForwardRate;
  158. static std::vector< SharedPtr<RootPeer> > siblings;
  159. static std::unordered_map< uint64_t,std::unordered_map< MulticastGroup,std::unordered_map< Address,int64_t,AddressHasher >,MulticastGroupHasher > > multicastSubscriptions;
  160. static std::unordered_map< Identity,SharedPtr<RootPeer>,IdentityHasher > peersByIdentity;
  161. static std::unordered_map< Address,std::set< SharedPtr<RootPeer> >,AddressHasher > peersByVirtAddr;
  162. static std::unordered_map< InetAddress,std::set< SharedPtr<RootPeer> >,InetAddressHasher > peersByPhysAddr;
  163. static std::unordered_map< RendezvousKey,int64_t,RendezvousKey::Hasher > lastRendezvous;
  164. static std::mutex siblings_l;
  165. static std::mutex multicastSubscriptions_l;
  166. static std::mutex peersByIdentity_l;
  167. static std::mutex peersByVirtAddr_l;
  168. static std::mutex peersByPhysAddr_l;
  169. static std::mutex lastRendezvous_l;
  170. //////////////////////////////////////////////////////////////////////////////
  171. //////////////////////////////////////////////////////////////////////////////
  172. static void handlePacket(const int v4s,const int v6s,const InetAddress *const ip,Packet &pkt)
  173. {
  174. char ipstr[128],ipstr2[128],astr[32],astr2[32],tmpstr[256];
  175. const bool fragment = pkt[ZT_PACKET_FRAGMENT_IDX_FRAGMENT_INDICATOR] == ZT_PACKET_FRAGMENT_INDICATOR;
  176. const Address source(pkt.source());
  177. const Address dest(pkt.destination());
  178. const int64_t now = OSUtils::now();
  179. inputRate.log(now,pkt.size());
  180. if ((!fragment)&&(!pkt.fragmented())&&(dest == self.address())) {
  181. SharedPtr<RootPeer> peer;
  182. // If this is an un-encrypted HELLO, either learn a new peer or verify
  183. // that this is a peer we already know.
  184. if ((pkt.cipher() == ZT_PROTO_CIPHER_SUITE__POLY1305_NONE)&&(pkt.verb() == Packet::VERB_HELLO)) {
  185. std::lock_guard<std::mutex> pbi_l(peersByIdentity_l);
  186. std::lock_guard<std::mutex> pbv_l(peersByVirtAddr_l);
  187. Identity id;
  188. if (id.deserialize(pkt,ZT_PROTO_VERB_HELLO_IDX_IDENTITY)) {
  189. {
  190. auto pById = peersByIdentity.find(id);
  191. if (pById != peersByIdentity.end()) {
  192. peer = pById->second;
  193. //printf("%s has %s (known (1))" ZT_EOL_S,ip->toString(ipstr),source().toString(astr));
  194. }
  195. }
  196. if (peer) {
  197. if (!pkt.dearmor(peer->key)) {
  198. printf("%s HELLO rejected: packet authentication failed" ZT_EOL_S,ip->toString(ipstr));
  199. return;
  200. }
  201. } else {
  202. peer.set(new RootPeer);
  203. if (self.agree(id,peer->key)) {
  204. if (pkt.dearmor(peer->key)) {
  205. if (!pkt.uncompress()) {
  206. printf("%s HELLO rejected: decompression failed" ZT_EOL_S,ip->toString(ipstr));
  207. return;
  208. }
  209. peer->id = id;
  210. peer->lastReceive = now;
  211. peersByIdentity.emplace(id,peer);
  212. peersByVirtAddr[id.address()].emplace(peer);
  213. } else {
  214. printf("%s HELLO rejected: packet authentication failed" ZT_EOL_S,ip->toString(ipstr));
  215. return;
  216. }
  217. } else {
  218. printf("%s HELLO rejected: key agreement failed" ZT_EOL_S,ip->toString(ipstr));
  219. return;
  220. }
  221. }
  222. }
  223. }
  224. // If it wasn't a HELLO, check to see if any known identities for the sender's
  225. // short ZT address successfully decrypt the packet.
  226. if (!peer) {
  227. std::lock_guard<std::mutex> pbv_l(peersByVirtAddr_l);
  228. auto peers = peersByVirtAddr.find(source);
  229. if (peers != peersByVirtAddr.end()) {
  230. for(auto p=peers->second.begin();p!=peers->second.end();++p) {
  231. if (pkt.dearmor((*p)->key)) {
  232. if (!pkt.uncompress()) {
  233. printf("%s packet rejected: decompression failed" ZT_EOL_S,ip->toString(ipstr));
  234. return;
  235. }
  236. peer = (*p);
  237. //printf("%s has %s (known (2))" ZT_EOL_S,ip->toString(ipstr),source().toString(astr));
  238. break;
  239. }
  240. }
  241. }
  242. }
  243. // If we found the peer, update IP and/or time and handle certain key packet types that the
  244. // root must concern itself with.
  245. if (peer) {
  246. std::lock_guard<std::mutex> pl(peer->lock);
  247. InetAddress *const peerIp = ip->isV4() ? &(peer->ip4) : &(peer->ip6);
  248. if (*peerIp != ip) {
  249. std::lock_guard<std::mutex> pbp_l(peersByPhysAddr_l);
  250. if (*peerIp) {
  251. auto prev = peersByPhysAddr.find(*peerIp);
  252. if (prev != peersByPhysAddr.end()) {
  253. prev->second.erase(peer);
  254. if (prev->second.empty())
  255. peersByPhysAddr.erase(prev);
  256. }
  257. }
  258. *peerIp = ip;
  259. peersByPhysAddr[ip].emplace(peer);
  260. }
  261. const int64_t now = OSUtils::now();
  262. peer->lastReceive = now;
  263. switch(pkt.verb()) {
  264. case Packet::VERB_HELLO:
  265. try {
  266. if ((now - peer->lastHello) > 1000) {
  267. peer->lastHello = now;
  268. peer->vMajor = (int)pkt[ZT_PROTO_VERB_HELLO_IDX_MAJOR_VERSION];
  269. peer->vMinor = (int)pkt[ZT_PROTO_VERB_HELLO_IDX_MINOR_VERSION];
  270. peer->vRev = (int)pkt.template at<uint16_t>(ZT_PROTO_VERB_HELLO_IDX_REVISION);
  271. const uint64_t origId = pkt.packetId();
  272. const uint64_t ts = pkt.template at<uint64_t>(ZT_PROTO_VERB_HELLO_IDX_TIMESTAMP);
  273. pkt.reset(source,self.address(),Packet::VERB_OK);
  274. pkt.append((uint8_t)Packet::VERB_HELLO);
  275. pkt.append(origId);
  276. pkt.append(ts);
  277. pkt.append((uint8_t)ZT_PROTO_VERSION);
  278. pkt.append((uint8_t)0);
  279. pkt.append((uint8_t)0);
  280. pkt.append((uint16_t)0);
  281. ip->serialize(pkt);
  282. pkt.armor(peer->key,true);
  283. sendto(ip->isV4() ? v4s : v6s,pkt.data(),pkt.size(),SENDTO_FLAGS,(const struct sockaddr *)ip,(socklen_t)((ip->ss_family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)));
  284. outputRate.log(now,pkt.size());
  285. peer->lastSend = now;
  286. }
  287. } catch ( ... ) {
  288. printf("* unexpected exception handling HELLO from %s" ZT_EOL_S,ip->toString(ipstr));
  289. }
  290. break;
  291. case Packet::VERB_ECHO:
  292. try {
  293. if ((now - peer->lastEcho) > 1000) {
  294. peer->lastEcho = now;
  295. Packet outp(source,self.address(),Packet::VERB_OK);
  296. outp.append((uint8_t)Packet::VERB_ECHO);
  297. outp.append(pkt.packetId());
  298. outp.append(((const uint8_t *)pkt.data()) + ZT_PACKET_IDX_PAYLOAD,pkt.size() - ZT_PACKET_IDX_PAYLOAD);
  299. outp.compress();
  300. outp.armor(peer->key,true);
  301. sendto(ip->isV4() ? v4s : v6s,outp.data(),outp.size(),SENDTO_FLAGS,(const struct sockaddr *)ip,(socklen_t)((ip->ss_family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)));
  302. outputRate.log(now,outp.size());
  303. peer->lastSend = now;
  304. }
  305. } catch ( ... ) {
  306. printf("* unexpected exception handling ECHO from %s" ZT_EOL_S,ip->toString(ipstr));
  307. }
  308. case Packet::VERB_WHOIS:
  309. try {
  310. std::vector< SharedPtr<RootPeer> > results;
  311. {
  312. std::lock_guard<std::mutex> l(peersByVirtAddr_l);
  313. for(unsigned int ptr=ZT_PACKET_IDX_PAYLOAD;(ptr+ZT_ADDRESS_LENGTH)<=pkt.size();ptr+=ZT_ADDRESS_LENGTH) {
  314. auto peers = peersByVirtAddr.find(Address(pkt.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH));
  315. if (peers != peersByVirtAddr.end()) {
  316. for(auto p=peers->second.begin();p!=peers->second.end();++p)
  317. results.push_back(*p);
  318. }
  319. }
  320. }
  321. if (!results.empty()) {
  322. const uint64_t origId = pkt.packetId();
  323. pkt.reset(source,self.address(),Packet::VERB_OK);
  324. pkt.append((uint8_t)Packet::VERB_WHOIS);
  325. pkt.append(origId);
  326. for(auto p=results.begin();p!=results.end();++p)
  327. (*p)->id.serialize(pkt,false);
  328. pkt.armor(peer->key,true);
  329. sendto(ip->isV4() ? v4s : v6s,pkt.data(),pkt.size(),SENDTO_FLAGS,(const struct sockaddr *)ip,(socklen_t)((ip->ss_family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)));
  330. outputRate.log(now,pkt.size());
  331. peer->lastSend = now;
  332. }
  333. } catch ( ... ) {
  334. printf("* unexpected exception handling ECHO from %s" ZT_EOL_S,ip->toString(ipstr));
  335. }
  336. case Packet::VERB_MULTICAST_LIKE:
  337. try {
  338. std::lock_guard<std::mutex> l(multicastSubscriptions_l);
  339. for(unsigned int ptr=ZT_PACKET_IDX_PAYLOAD;(ptr+18)<=pkt.size();ptr+=18) {
  340. const uint64_t nwid = pkt.template at<uint64_t>(ptr);
  341. const MulticastGroup mg(MAC(pkt.field(ptr + 8,6),6),pkt.template at<uint32_t>(ptr + 14));
  342. multicastSubscriptions[nwid][mg][source] = now;
  343. //printf("%s %s subscribes to %s/%.8lx on network %.16llx" ZT_EOL_S,ip->toString(ipstr),source.toString(astr),mg.mac().toString(tmpstr),(unsigned long)mg.adi(),(unsigned long long)nwid);
  344. }
  345. } catch ( ... ) {
  346. printf("* unexpected exception handling MULTICAST_LIKE from %s" ZT_EOL_S,ip->toString(ipstr));
  347. }
  348. break;
  349. case Packet::VERB_MULTICAST_GATHER:
  350. try {
  351. const uint64_t nwid = pkt.template at<uint64_t>(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_NETWORK_ID);
  352. //const unsigned int flags = pkt[ZT_PROTO_VERB_MULTICAST_GATHER_IDX_FLAGS];
  353. const MulticastGroup mg(MAC(pkt.field(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_MAC,6),6),pkt.template at<uint32_t>(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_ADI));
  354. unsigned int gatherLimit = pkt.template at<uint32_t>(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_GATHER_LIMIT);
  355. if (gatherLimit > 255)
  356. gatherLimit = 255;
  357. const uint64_t origId = pkt.packetId();
  358. pkt.reset(source,self.address(),Packet::VERB_OK);
  359. pkt.append((uint8_t)Packet::VERB_MULTICAST_GATHER);
  360. pkt.append(origId);
  361. pkt.append(nwid);
  362. mg.mac().appendTo(pkt);
  363. pkt.append((uint32_t)mg.adi());
  364. {
  365. std::lock_guard<std::mutex> l(multicastSubscriptions_l);
  366. auto forNet = multicastSubscriptions.find(nwid);
  367. if (forNet != multicastSubscriptions.end()) {
  368. auto forGroup = forNet->second.find(mg);
  369. if (forGroup != forNet->second.end()) {
  370. pkt.append((uint32_t)forGroup->second.size());
  371. const unsigned int countAt = pkt.size();
  372. pkt.addSize(2);
  373. unsigned int l = 0;
  374. for(auto g=forGroup->second.begin();((l<gatherLimit)&&(g!=forGroup->second.end()));++g) {
  375. if (g->first != source) {
  376. ++l;
  377. g->first.appendTo(pkt);
  378. }
  379. }
  380. if (l > 0) {
  381. pkt.setAt<uint16_t>(countAt,(uint16_t)l);
  382. pkt.armor(peer->key,true);
  383. sendto(ip->isV4() ? v4s : v6s,pkt.data(),pkt.size(),SENDTO_FLAGS,(const struct sockaddr *)ip,(socklen_t)(ip->isV4() ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)));
  384. outputRate.log(now,pkt.size());
  385. peer->lastSend = now;
  386. //printf("%s %s gathered %u subscribers to %s/%.8lx on network %.16llx" ZT_EOL_S,ip->toString(ipstr),source.toString(astr),l,mg.mac().toString(tmpstr),(unsigned long)mg.adi(),(unsigned long long)nwid);
  387. }
  388. }
  389. }
  390. }
  391. } catch ( ... ) {
  392. printf("* unexpected exception handling MULTICAST_GATHER from %s" ZT_EOL_S,ip->toString(ipstr));
  393. }
  394. break;
  395. default:
  396. break;
  397. }
  398. return;
  399. }
  400. }
  401. // If we made it here, we are forwarding this packet to someone else and also possibly
  402. // sending a RENDEZVOUS message.
  403. bool introduce = false;
  404. if (!fragment) {
  405. RendezvousKey rk(source,dest);
  406. std::lock_guard<std::mutex> l(lastRendezvous_l);
  407. int64_t &lr = lastRendezvous[rk];
  408. if ((now - lr) >= 45000) {
  409. lr = now;
  410. introduce = true;
  411. }
  412. }
  413. std::vector< std::pair< InetAddress *,SharedPtr<RootPeer> > > toAddrs;
  414. {
  415. std::lock_guard<std::mutex> pbv_l(peersByVirtAddr_l);
  416. auto peers = peersByVirtAddr.find(dest);
  417. if (peers != peersByVirtAddr.end()) {
  418. for(auto p=peers->second.begin();p!=peers->second.end();++p) {
  419. if ((*p)->ip4) {
  420. toAddrs.push_back(std::pair< InetAddress *,SharedPtr<RootPeer> >(&((*p)->ip4),*p));
  421. } else if ((*p)->ip6) {
  422. toAddrs.push_back(std::pair< InetAddress *,SharedPtr<RootPeer> >(&((*p)->ip6),*p));
  423. }
  424. }
  425. }
  426. }
  427. if (toAddrs.empty()) {
  428. std::lock_guard<std::mutex> sib_l(siblings_l);
  429. for(auto s=siblings.begin();s!=siblings.end();++s) {
  430. if (((now - (*s)->lastReceive) < (ZT_PEER_PING_PERIOD * 2))&&((*s)->sibling)) {
  431. if ((*s)->ip4) {
  432. toAddrs.push_back(std::pair< InetAddress *,SharedPtr<RootPeer> >(&((*s)->ip4),*s));
  433. } else if ((*s)->ip6) {
  434. toAddrs.push_back(std::pair< InetAddress *,SharedPtr<RootPeer> >(&((*s)->ip6),*s));
  435. }
  436. }
  437. }
  438. }
  439. if (toAddrs.empty())
  440. return;
  441. if (introduce) {
  442. std::lock_guard<std::mutex> l(peersByVirtAddr_l);
  443. auto sources = peersByVirtAddr.find(source);
  444. if (sources != peersByVirtAddr.end()) {
  445. for(auto a=sources->second.begin();a!=sources->second.end();++a) {
  446. for(auto b=toAddrs.begin();b!=toAddrs.end();++b) {
  447. if (((*a)->ip6)&&(b->second->ip6)) {
  448. //printf("* introducing %s(%s) to %s(%s)" ZT_EOL_S,ip->toString(ipstr),source.toString(astr),b->second->ip6.toString(ipstr2),dest.toString(astr2));
  449. // Introduce source to destination (V6)
  450. Packet outp(source,self.address(),Packet::VERB_RENDEZVOUS);
  451. outp.append((uint8_t)0);
  452. dest.appendTo(outp);
  453. outp.append((uint16_t)b->second->ip6.port());
  454. outp.append((uint8_t)16);
  455. outp.append((const uint8_t *)b->second->ip6.rawIpData(),16);
  456. outp.armor((*a)->key,true);
  457. sendto(v6s,outp.data(),outp.size(),SENDTO_FLAGS,(const struct sockaddr *)&((*a)->ip6),(socklen_t)sizeof(struct sockaddr_in6));
  458. outputRate.log(now,outp.size());
  459. (*a)->lastSend = now;
  460. // Introduce destination to source (V6)
  461. outp.reset(dest,self.address(),Packet::VERB_RENDEZVOUS);
  462. outp.append((uint8_t)0);
  463. source.appendTo(outp);
  464. outp.append((uint16_t)ip->port());
  465. outp.append((uint8_t)16);
  466. outp.append((const uint8_t *)ip->rawIpData(),16);
  467. outp.armor(b->second->key,true);
  468. sendto(v6s,outp.data(),outp.size(),SENDTO_FLAGS,(const struct sockaddr *)&(b->second->ip6),(socklen_t)sizeof(struct sockaddr_in6));
  469. outputRate.log(now,outp.size());
  470. b->second->lastSend = now;
  471. }
  472. if (((*a)->ip4)&&(b->second->ip4)) {
  473. //printf("* introducing %s(%s) to %s(%s)" ZT_EOL_S,ip->toString(ipstr),source.toString(astr),b->second->ip4.toString(ipstr2),dest.toString(astr2));
  474. // Introduce source to destination (V4)
  475. Packet outp(source,self.address(),Packet::VERB_RENDEZVOUS);
  476. outp.append((uint8_t)0);
  477. dest.appendTo(outp);
  478. outp.append((uint16_t)b->second->ip4.port());
  479. outp.append((uint8_t)4);
  480. outp.append((const uint8_t *)b->second->ip4.rawIpData(),4);
  481. outp.armor((*a)->key,true);
  482. sendto(v4s,outp.data(),outp.size(),SENDTO_FLAGS,(const struct sockaddr *)&((*a)->ip4),(socklen_t)sizeof(struct sockaddr_in));
  483. outputRate.log(now,outp.size());
  484. (*a)->lastSend = now;
  485. // Introduce destination to source (V4)
  486. outp.reset(dest,self.address(),Packet::VERB_RENDEZVOUS);
  487. outp.append((uint8_t)0);
  488. source.appendTo(outp);
  489. outp.append((uint16_t)ip->port());
  490. outp.append((uint8_t)4);
  491. outp.append((const uint8_t *)ip->rawIpData(),4);
  492. outp.armor(b->second->key,true);
  493. sendto(v4s,outp.data(),outp.size(),SENDTO_FLAGS,(const struct sockaddr *)&(b->second->ip4),(socklen_t)sizeof(struct sockaddr_in));
  494. outputRate.log(now,outp.size());
  495. b->second->lastSend = now;
  496. }
  497. }
  498. }
  499. }
  500. }
  501. if (fragment) {
  502. if (reinterpret_cast<Packet::Fragment *>(&pkt)->incrementHops() >= ZT_PROTO_MAX_HOPS) {
  503. printf("%s refused to forward to %s: max hop count exceeded" ZT_EOL_S,ip->toString(ipstr),dest.toString(astr));
  504. return;
  505. }
  506. } else {
  507. if (pkt.incrementHops() >= ZT_PROTO_MAX_HOPS) {
  508. printf("%s refused to forward to %s: max hop count exceeded" ZT_EOL_S,ip->toString(ipstr),dest.toString(astr));
  509. return;
  510. }
  511. }
  512. for(auto i=toAddrs.begin();i!=toAddrs.end();++i) {
  513. //printf("%s -> %s for %s -> %s (%u bytes)" ZT_EOL_S,ip->toString(ipstr),i->first->toString(ipstr2),source.toString(astr),dest.toString(astr2),pkt.size());
  514. if (sendto(i->first->isV4() ? v4s : v6s,pkt.data(),pkt.size(),SENDTO_FLAGS,(const struct sockaddr *)i->first,(socklen_t)(i->first->isV4() ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6))) <= 0) {
  515. printf("* write error forwarding packet to %s: %s" ZT_EOL_S,i->first->toString(ipstr),strerror(errno));
  516. } else {
  517. outputRate.log(now,pkt.size());
  518. forwardRate.log(now,pkt.size());
  519. if (i->second->sibling)
  520. siblingForwardRate.log(now,pkt.size());
  521. i->second->lastSend = now;
  522. }
  523. }
  524. }
  525. //////////////////////////////////////////////////////////////////////////////
  526. //////////////////////////////////////////////////////////////////////////////
  527. static int bindSocket(struct sockaddr *const bindAddr)
  528. {
  529. const int s = socket(bindAddr->sa_family,SOCK_DGRAM,0);
  530. if (s < 0) {
  531. close(s);
  532. return -1;
  533. }
  534. int f = 1048576;
  535. while (f > 16384) {
  536. if (setsockopt(s,SOL_SOCKET,SO_RCVBUF,(const char *)&f,sizeof(f)) == 0)
  537. break;
  538. f -= 16384;
  539. }
  540. f = 1048576;
  541. while (f > 16384) {
  542. if (setsockopt(s,SOL_SOCKET,SO_SNDBUF,(const char *)&f,sizeof(f)) == 0)
  543. break;
  544. f -= 16384;
  545. }
  546. if (bindAddr->sa_family == AF_INET6) {
  547. f = 1; setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&f,sizeof(f));
  548. #ifdef IPV6_MTU_DISCOVER
  549. f = 0; setsockopt(s,IPPROTO_IPV6,IPV6_MTU_DISCOVER,&f,sizeof(f));
  550. #endif
  551. #ifdef IPV6_DONTFRAG
  552. f = 0; setsockopt(s,IPPROTO_IPV6,IPV6_DONTFRAG,&f,sizeof(f));
  553. #endif
  554. }
  555. #ifdef IP_DONTFRAG
  556. f = 0; setsockopt(s,IPPROTO_IP,IP_DONTFRAG,&f,sizeof(f));
  557. #endif
  558. #ifdef IP_MTU_DISCOVER
  559. f = IP_PMTUDISC_DONT; setsockopt(s,IPPROTO_IP,IP_MTU_DISCOVER,&f,sizeof(f));
  560. #endif
  561. #ifdef SO_NO_CHECK
  562. if (bindAddr->sa_family == AF_INET) {
  563. f = 1; setsockopt(s,SOL_SOCKET,SO_NO_CHECK,(void *)&f,sizeof(f));
  564. }
  565. #endif
  566. #if defined(SO_REUSEPORT)
  567. f = 1; setsockopt(s,SOL_SOCKET,SO_REUSEPORT,(void *)&f,sizeof(f));
  568. #endif
  569. #ifndef __LINUX__ // linux wants just SO_REUSEPORT
  570. f = 1; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
  571. #endif
  572. if (bind(s,bindAddr,(bindAddr->sa_family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6))) {
  573. close(s);
  574. //printf("%s\n",strerror(errno));
  575. return -1;
  576. }
  577. return s;
  578. }
  579. static void shutdownSigHandler(int sig) { run = false; }
  580. int main(int argc,char **argv)
  581. {
  582. signal(SIGTERM,shutdownSigHandler);
  583. signal(SIGINT,shutdownSigHandler);
  584. signal(SIGQUIT,shutdownSigHandler);
  585. signal(SIGPIPE,SIG_IGN);
  586. signal(SIGUSR1,SIG_IGN);
  587. signal(SIGUSR2,SIG_IGN);
  588. signal(SIGCHLD,SIG_IGN);
  589. startTime = OSUtils::now();
  590. if (argc < 3) {
  591. printf("Usage: zerotier-root <identity.secret> <config path>" ZT_EOL_S);
  592. return 1;
  593. }
  594. {
  595. std::string myIdStr;
  596. if (!OSUtils::readFile(argv[1],myIdStr)) {
  597. printf("FATAL: cannot read identity.secret at %s" ZT_EOL_S,argv[1]);
  598. return 1;
  599. }
  600. if (!self.fromString(myIdStr.c_str())) {
  601. printf("FATAL: cannot read identity.secret at %s (invalid identity)" ZT_EOL_S,argv[1]);
  602. return 1;
  603. }
  604. if (!self.hasPrivate()) {
  605. printf("FATAL: cannot read identity.secret at %s (missing secret key)" ZT_EOL_S,argv[1]);
  606. return 1;
  607. }
  608. }
  609. {
  610. std::string configStr;
  611. if (!OSUtils::readFile(argv[2],configStr)) {
  612. printf("FATAL: cannot read config file at %s" ZT_EOL_S,argv[2]);
  613. return 1;
  614. }
  615. try {
  616. config = json::parse(configStr);
  617. } catch (std::exception &exc) {
  618. printf("FATAL: config file at %s invalid: %s" ZT_EOL_S,argv[2],exc.what());
  619. return 1;
  620. } catch ( ... ) {
  621. printf("FATAL: config file at %s invalid: unknown exception" ZT_EOL_S,argv[2]);
  622. return 1;
  623. }
  624. if (!config.is_object()) {
  625. printf("FATAL: config file at %s invalid: does not contain a JSON object" ZT_EOL_S,argv[2]);
  626. return 1;
  627. }
  628. }
  629. try {
  630. auto jport = config["port"];
  631. if (jport.is_array()) {
  632. for(long i=0;i<(long)jport.size();++i) {
  633. int port = jport[i];
  634. if ((port <= 0)||(port > 65535)) {
  635. printf("FATAL: invalid port in config file %d" ZT_EOL_S,port);
  636. return 1;
  637. }
  638. ports.push_back(port);
  639. }
  640. } else {
  641. int port = jport;
  642. if ((port <= 0)||(port > 65535)) {
  643. printf("FATAL: invalid port in config file %d" ZT_EOL_S,port);
  644. return 1;
  645. }
  646. ports.push_back(port);
  647. }
  648. } catch ( ... ) {}
  649. if (ports.empty())
  650. ports.push_back(ZT_DEFAULT_PORT);
  651. std::sort(ports.begin(),ports.end());
  652. int httpPort = ZT_DEFAULT_PORT;
  653. try {
  654. httpPort = config["httpPort"];
  655. if ((httpPort <= 0)||(httpPort > 65535)) {
  656. printf("FATAL: invalid HTTP port in config file %d" ZT_EOL_S,httpPort);
  657. return 1;
  658. }
  659. } catch ( ... ) {
  660. httpPort = ZT_DEFAULT_PORT;
  661. }
  662. try {
  663. statsRoot = config["statsRoot"];
  664. while ((statsRoot.length() > 0)&&(statsRoot[statsRoot.length()-1] == ZT_PATH_SEPARATOR))
  665. statsRoot = statsRoot.substr(0,statsRoot.length()-1);
  666. if (statsRoot.length() > 0)
  667. OSUtils::mkdir(statsRoot);
  668. } catch ( ... ) {
  669. statsRoot = "";
  670. }
  671. try {
  672. auto sibs = config["siblings"];
  673. if (sibs.is_array()) {
  674. for(long i=0;i<(long)siblings.size();++i) {
  675. auto sib = sibs[i];
  676. if (sib.is_object()) {
  677. std::string idStr = sib["id"];
  678. std::string ipStr = sib["ip"];
  679. Identity id;
  680. if (!id.fromString(idStr.c_str())) {
  681. printf("FATAL: invalid JSON while parsing siblings section in config file: invalid identity in sibling entry" ZT_EOL_S);
  682. return 1;
  683. }
  684. InetAddress ip;
  685. if (!ip.fromString(ipStr.c_str())) {
  686. printf("FATAL: invalid JSON while parsing siblings section in config file: invalid IP address in sibling entry" ZT_EOL_S);
  687. return 1;
  688. }
  689. ip.setPort((unsigned int)sib["port"]);
  690. SharedPtr<RootPeer> rp(new RootPeer);
  691. rp->id = id;
  692. if (!self.agree(id,rp->key)) {
  693. printf("FATAL: invalid JSON while parsing siblings section in config file: invalid identity in sibling entry (unable to execute key agreement)" ZT_EOL_S);
  694. return 1;
  695. }
  696. if (ip.isV4()) {
  697. rp->ip4 = ip;
  698. } else if (ip.isV6()) {
  699. rp->ip6 = ip;
  700. } else {
  701. printf("FATAL: invalid JSON while parsing siblings section in config file: invalid IP address in sibling entry" ZT_EOL_S);
  702. return 1;
  703. }
  704. rp->sibling = true;
  705. siblings.push_back(rp);
  706. peersByIdentity[id] = rp;
  707. peersByVirtAddr[id.address()].insert(rp);
  708. peersByPhysAddr[ip].insert(rp);
  709. } else {
  710. printf("FATAL: invalid JSON while parsing siblings section in config file: sibling entry is not a JSON object" ZT_EOL_S);
  711. return 1;
  712. }
  713. }
  714. } else {
  715. printf("FATAL: invalid JSON while parsing siblings section in config file: siblings is not a JSON array" ZT_EOL_S);
  716. return 1;
  717. }
  718. } catch ( ... ) {
  719. printf("FATAL: invalid JSON while parsing siblings section in config file: parse error" ZT_EOL_S);
  720. return 1;
  721. }
  722. unsigned int ncores = std::thread::hardware_concurrency();
  723. if (ncores == 0) ncores = 1;
  724. run = true;
  725. std::vector<std::thread> threads;
  726. std::vector<int> sockets;
  727. int v4Sock = -1,v6Sock = -1;
  728. for(auto port=ports.begin();port!=ports.end();++port) {
  729. for(unsigned int tn=0;tn<ncores;++tn) {
  730. struct sockaddr_in6 in6;
  731. memset(&in6,0,sizeof(in6));
  732. in6.sin6_family = AF_INET6;
  733. in6.sin6_port = htons((uint16_t)*port);
  734. const int s6 = bindSocket((struct sockaddr *)&in6);
  735. if (s6 < 0) {
  736. std::cout << "ERROR: unable to bind to port " << *port << ZT_EOL_S;
  737. exit(1);
  738. }
  739. struct sockaddr_in in4;
  740. memset(&in4,0,sizeof(in4));
  741. in4.sin_family = AF_INET;
  742. in4.sin_port = htons((uint16_t)*port);
  743. const int s4 = bindSocket((struct sockaddr *)&in4);
  744. if (s4 < 0) {
  745. std::cout << "ERROR: unable to bind to port " << *port << ZT_EOL_S;
  746. exit(1);
  747. }
  748. sockets.push_back(s6);
  749. sockets.push_back(s4);
  750. if (v4Sock < 0) v4Sock = s4;
  751. if (v6Sock < 0) v6Sock = s6;
  752. threads.push_back(std::thread([s6,s4]() {
  753. struct sockaddr_in6 in6;
  754. Packet pkt;
  755. memset(&in6,0,sizeof(in6));
  756. for(;;) {
  757. socklen_t sl = sizeof(in6);
  758. const int pl = (int)recvfrom(s6,pkt.unsafeData(),pkt.capacity(),0,(struct sockaddr *)&in6,&sl);
  759. if (pl > 0) {
  760. if (pl >= ZT_PROTO_MIN_FRAGMENT_LENGTH) {
  761. try {
  762. pkt.setSize((unsigned int)pl);
  763. handlePacket(s4,s6,reinterpret_cast<const InetAddress *>(&in6),pkt);
  764. } catch ( ... ) {
  765. char ipstr[128];
  766. printf("* unexpected exception handling packet from %s" ZT_EOL_S,reinterpret_cast<const InetAddress *>(&in6)->toString(ipstr));
  767. }
  768. }
  769. } else {
  770. break;
  771. }
  772. }
  773. }));
  774. threads.push_back(std::thread([s6,s4]() {
  775. struct sockaddr_in in4;
  776. Packet pkt;
  777. memset(&in4,0,sizeof(in4));
  778. for(;;) {
  779. socklen_t sl = sizeof(in4);
  780. const int pl = (int)recvfrom(s4,pkt.unsafeData(),pkt.capacity(),0,(struct sockaddr *)&in4,&sl);
  781. if (pl > 0) {
  782. if (pl >= ZT_PROTO_MIN_FRAGMENT_LENGTH) {
  783. try {
  784. pkt.setSize((unsigned int)pl);
  785. handlePacket(s4,s6,reinterpret_cast<const InetAddress *>(&in4),pkt);
  786. } catch ( ... ) {
  787. char ipstr[128];
  788. printf("* unexpected exception handling packet from %s" ZT_EOL_S,reinterpret_cast<const InetAddress *>(&in4)->toString(ipstr));
  789. }
  790. }
  791. } else {
  792. break;
  793. }
  794. }
  795. }));
  796. }
  797. }
  798. // Minimal local API for use with monitoring clients, etc.
  799. httplib::Server apiServ;
  800. threads.push_back(std::thread([&apiServ,httpPort]() {
  801. apiServ.Get("/",[](const httplib::Request &req,httplib::Response &res) {
  802. std::ostringstream o;
  803. std::lock_guard<std::mutex> l0(peersByIdentity_l);
  804. std::lock_guard<std::mutex> l1(peersByPhysAddr_l);
  805. o << "ZeroTier Root Server " << ZEROTIER_ONE_VERSION_MAJOR << '.' << ZEROTIER_ONE_VERSION_MINOR << '.' << ZEROTIER_ONE_VERSION_REVISION << ZT_EOL_S;
  806. o << "(c)2019 ZeroTier, Inc." ZT_EOL_S "Licensed under the ZeroTier BSL 1.1" ZT_EOL_S ZT_EOL_S;
  807. o << "Peers Online: " << peersByIdentity.size() << ZT_EOL_S;
  808. o << "Physical Addresses: " << peersByPhysAddr.size() << ZT_EOL_S;
  809. res.set_content(o.str(),"text/plain");
  810. });
  811. apiServ.Get("/peer",[](const httplib::Request &req,httplib::Response &res) {
  812. char tmp[256];
  813. std::ostringstream o;
  814. o << '[';
  815. {
  816. bool first = true;
  817. std::lock_guard<std::mutex> l(peersByIdentity_l);
  818. for(auto p=peersByIdentity.begin();p!=peersByIdentity.end();++p) {
  819. if (first)
  820. first = false;
  821. else o << ',';
  822. o <<
  823. "{\"address\":\"" << p->first.address().toString(tmp) << "\""
  824. ",\"latency\":-1"
  825. ",\"paths\":[";
  826. if (p->second->ip4) {
  827. o <<
  828. "{\"active\":true"
  829. ",\"address\":\"" << p->second->ip4.toIpString(tmp) << "\\/" << p->second->ip4.port() << "\""
  830. ",\"expired\":false"
  831. ",\"lastReceive\":" << p->second->lastReceive <<
  832. ",\"lastSend\":" << p->second->lastSend <<
  833. ",\"preferred\":true"
  834. ",\"trustedPathId\":0}";
  835. }
  836. if (p->second->ip6) {
  837. if (p->second->ip4)
  838. o << ',';
  839. o <<
  840. "{\"active\":true"
  841. ",\"address\":\"" << p->second->ip6.toIpString(tmp) << "\\/" << p->second->ip6.port() << "\""
  842. ",\"expired\":false"
  843. ",\"lastReceive\":" << p->second->lastReceive <<
  844. ",\"lastSend\":" << p->second->lastSend <<
  845. ",\"preferred\":" << ((p->second->ip4) ? "false" : "true") <<
  846. ",\"trustedPathId\":0}";
  847. }
  848. o << "]"
  849. ",\"role\":\"LEAF\""
  850. ",\"version\":\"" << p->second->vMajor << '.' << p->second->vMinor << '.' << p->second->vRev << "\""
  851. ",\"versionMajor\":" << p->second->vMajor <<
  852. ",\"versionMinor\":" << p->second->vMinor <<
  853. ",\"versionRev\":" << p->second->vRev << "}";
  854. }
  855. }
  856. o << ']';
  857. res.set_content(o.str(),"application/json");
  858. });
  859. apiServ.listen("127.0.0.1",httpPort,0);
  860. }));
  861. // In the main thread periodically clean stuff up
  862. int64_t lastCleaned = 0;
  863. int64_t lastWroteStats = 0;
  864. int64_t lastPingedSiblings = 0;
  865. while (run) {
  866. //peersByIdentity_l.lock();
  867. //peersByPhysAddr_l.lock();
  868. //printf("*** have %lu peers at %lu physical endpoints" ZT_EOL_S,(unsigned long)peersByIdentity.size(),(unsigned long)peersByPhysAddr.size());
  869. //peersByPhysAddr_l.unlock();
  870. //peersByIdentity_l.unlock();
  871. sleep(1);
  872. const int64_t now = OSUtils::now();
  873. // Send HELLO to sibling roots
  874. if ((now - lastPingedSiblings) >= ZT_PEER_PING_PERIOD) {
  875. lastPingedSiblings = now;
  876. std::lock_guard<std::mutex> l(siblings_l);
  877. for(auto s=siblings.begin();s!=siblings.end();++s) {
  878. const InetAddress *ip = nullptr;
  879. socklen_t sl = 0;
  880. Packet outp((*s)->id.address(),self.address(),Packet::VERB_HELLO);
  881. outp.append((uint8_t)ZT_PROTO_VERSION);
  882. outp.append((uint8_t)ZEROTIER_ONE_VERSION_MAJOR);
  883. outp.append((uint8_t)ZEROTIER_ONE_VERSION_MINOR);
  884. outp.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
  885. outp.append((uint64_t)now);
  886. self.serialize(outp,false);
  887. if ((*s)->ip4) {
  888. (*s)->ip4.serialize(outp);
  889. ip = &((*s)->ip4);
  890. sl = sizeof(struct sockaddr_in);
  891. } else if ((*s)->ip6) {
  892. (*s)->ip6.serialize(outp);
  893. ip = &((*s)->ip6);
  894. sl = sizeof(struct sockaddr_in6);
  895. }
  896. if (ip) {
  897. outp.armor((*s)->key,false);
  898. sendto(ip->isV4() ? v4Sock : v6Sock,outp.data(),outp.size(),SENDTO_FLAGS,(const struct sockaddr *)ip,sl);
  899. }
  900. }
  901. }
  902. if ((now - lastCleaned) > 120000) {
  903. lastCleaned = now;
  904. // Old multicast subscription cleanup
  905. {
  906. std::lock_guard<std::mutex> l(multicastSubscriptions_l);
  907. for(auto a=multicastSubscriptions.begin();a!=multicastSubscriptions.end();) {
  908. for(auto b=a->second.begin();b!=a->second.end();) {
  909. for(auto c=b->second.begin();c!=b->second.end();) {
  910. if ((now - c->second) > ZT_MULTICAST_LIKE_EXPIRE)
  911. b->second.erase(c++);
  912. else ++c;
  913. }
  914. if (b->second.empty())
  915. a->second.erase(b++);
  916. else ++b;
  917. }
  918. if (a->second.empty())
  919. multicastSubscriptions.erase(a++);
  920. else ++a;
  921. }
  922. }
  923. // Remove expired peers
  924. {
  925. std::lock_guard<std::mutex> pbi_l(peersByIdentity_l);
  926. for(auto p=peersByIdentity.begin();p!=peersByIdentity.end();) {
  927. if (((now - p->second->lastReceive) > ZT_PEER_ACTIVITY_TIMEOUT)&&(!p->second->sibling)) {
  928. std::lock_guard<std::mutex> pbv_l(peersByVirtAddr_l);
  929. std::lock_guard<std::mutex> pbp_l(peersByPhysAddr_l);
  930. auto pbv = peersByVirtAddr.find(p->second->id.address());
  931. if (pbv != peersByVirtAddr.end()) {
  932. pbv->second.erase(p->second);
  933. if (pbv->second.empty())
  934. peersByVirtAddr.erase(pbv);
  935. }
  936. if (p->second->ip4) {
  937. auto pbp = peersByPhysAddr.find(p->second->ip4);
  938. if (pbp != peersByPhysAddr.end()) {
  939. pbp->second.erase(p->second);
  940. if (pbp->second.empty())
  941. peersByPhysAddr.erase(pbp);
  942. }
  943. }
  944. if (p->second->ip6) {
  945. auto pbp = peersByPhysAddr.find(p->second->ip6);
  946. if (pbp != peersByPhysAddr.end()) {
  947. pbp->second.erase(p->second);
  948. if (pbp->second.empty())
  949. peersByPhysAddr.erase(pbp);
  950. }
  951. }
  952. peersByIdentity.erase(p++);
  953. } else ++p;
  954. }
  955. }
  956. // Remove old rendezvous tracking entries
  957. {
  958. std::lock_guard<std::mutex> l(lastRendezvous_l);
  959. for(auto lr=lastRendezvous.begin();lr!=lastRendezvous.end();) {
  960. if ((now - lr->second) > ZT_PEER_ACTIVITY_TIMEOUT)
  961. lastRendezvous.erase(lr++);
  962. else ++lr;
  963. }
  964. }
  965. }
  966. // Write stats if configured to do so
  967. if (((now - lastWroteStats) > 15000)&&(statsRoot.length() > 0)) {
  968. lastWroteStats = now;
  969. std::string peersFilePath(statsRoot);
  970. peersFilePath.append("/.peers.tmp");
  971. FILE *pf = fopen(peersFilePath.c_str(),"wb");
  972. if (pf) {
  973. std::vector< SharedPtr<RootPeer> > sp;
  974. {
  975. std::lock_guard<std::mutex> pbi_l(peersByIdentity_l);
  976. sp.reserve(peersByIdentity.size());
  977. for(auto p=peersByIdentity.begin();p!=peersByIdentity.end();++p) {
  978. sp.push_back(p->second);
  979. }
  980. }
  981. std::sort(sp.begin(),sp.end(),[](const SharedPtr<RootPeer> &a,const SharedPtr<RootPeer> &b) { return (a->id < b->id); });
  982. char ip4[128],ip6[128];
  983. for(auto p=sp.begin();p!=sp.end();++p) {
  984. if ((*p)->ip4) {
  985. (*p)->ip4.toString(ip4);
  986. } else {
  987. ip4[0] = '-';
  988. ip4[1] = 0;
  989. }
  990. if ((*p)->ip6) {
  991. (*p)->ip6.toString(ip6);
  992. } else {
  993. ip6[0] = '-';
  994. ip6[1] = 0;
  995. }
  996. fprintf(pf,"%.10llx %21s %45s %5.4f %d.%d.%d" ZT_EOL_S,(unsigned long long)(*p)->id.address().toInt(),ip4,ip6,fabs((double)(now - (*p)->lastReceive) / 1000.0),(*p)->vMajor,(*p)->vMinor,(*p)->vRev);
  997. }
  998. fclose(pf);
  999. std::string peersFilePath2(statsRoot);
  1000. peersFilePath2.append("/peers");
  1001. OSUtils::rm(peersFilePath2);
  1002. OSUtils::rename(peersFilePath.c_str(),peersFilePath2.c_str());
  1003. }
  1004. std::string statsFilePath(statsRoot);
  1005. statsFilePath.append("/.stats.tmp");
  1006. FILE *sf = fopen(statsFilePath.c_str(),"wb");
  1007. if (sf) {
  1008. fprintf(sf,"Uptime (seconds) : %ld" ZT_EOL_S,(long)((now - startTime) / 1000));
  1009. peersByIdentity_l.lock();
  1010. fprintf(sf,"Peers : %llu" ZT_EOL_S,(unsigned long long)peersByIdentity.size());
  1011. peersByVirtAddr_l.lock();
  1012. fprintf(sf,"Virtual Address Collisions : %llu" ZT_EOL_S,(unsigned long long)(peersByIdentity.size() - peersByVirtAddr.size()));
  1013. peersByVirtAddr_l.unlock();
  1014. peersByIdentity_l.unlock();
  1015. peersByPhysAddr_l.lock();
  1016. fprintf(sf,"Physical Endpoints : %llu" ZT_EOL_S,(unsigned long long)peersByPhysAddr.size());
  1017. peersByPhysAddr_l.unlock();
  1018. lastRendezvous_l.lock();
  1019. fprintf(sf,"Recent P2P Graph Edges : %llu" ZT_EOL_S,(unsigned long long)lastRendezvous.size());
  1020. lastRendezvous_l.unlock();
  1021. fprintf(sf,"Input BPS : %.4f" ZT_EOL_S,inputRate.perSecond(now));
  1022. fprintf(sf,"Output BPS : %.4f" ZT_EOL_S,outputRate.perSecond(now));
  1023. fprintf(sf,"Forwarded BPS : %.4f" ZT_EOL_S,forwardRate.perSecond(now));
  1024. fprintf(sf,"Sibling Forwarded BPS : %.4f" ZT_EOL_S,siblingForwardRate.perSecond(now));
  1025. fclose(sf);
  1026. std::string statsFilePath2(statsRoot);
  1027. statsFilePath2.append("/stats");
  1028. OSUtils::rm(statsFilePath2);
  1029. OSUtils::rename(statsFilePath.c_str(),statsFilePath2.c_str());
  1030. }
  1031. }
  1032. }
  1033. // If we received a kill signal, close everything and wait
  1034. // for threads to die before exiting.
  1035. apiServ.stop();
  1036. for(auto s=sockets.begin();s!=sockets.end();++s) {
  1037. shutdown(*s,SHUT_RDWR);
  1038. close(*s);
  1039. }
  1040. for(auto t=threads.begin();t!=threads.end();++t)
  1041. t->join();
  1042. return 0;
  1043. }