Cluster.cpp 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717
  1. /*
  2. * ZeroTier One - Network Virtualization Everywhere
  3. * Copyright (C) 2011-2015 ZeroTier, Inc.
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU General Public License as published by
  7. * the Free Software Foundation, either version 3 of the License, or
  8. * (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU General Public License
  16. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  17. *
  18. * --
  19. *
  20. * ZeroTier may be used and distributed under the terms of the GPLv3, which
  21. * are available at: http://www.gnu.org/licenses/gpl-3.0.html
  22. *
  23. * If you would like to embed ZeroTier into a commercial application or
  24. * redistribute it in a modified binary form, please contact ZeroTier Networks
  25. * LLC. Start here: http://www.zerotier.com/
  26. */
  27. #ifdef ZT_ENABLE_CLUSTER
  28. #include <stdint.h>
  29. #include <stdio.h>
  30. #include <stdlib.h>
  31. #include <string.h>
  32. #include <math.h>
  33. #include <algorithm>
  34. #include <utility>
  35. #include "../version.h"
  36. #include "Cluster.hpp"
  37. #include "RuntimeEnvironment.hpp"
  38. #include "MulticastGroup.hpp"
  39. #include "CertificateOfMembership.hpp"
  40. #include "Salsa20.hpp"
  41. #include "Poly1305.hpp"
  42. #include "Identity.hpp"
  43. #include "Topology.hpp"
  44. #include "Packet.hpp"
  45. #include "Switch.hpp"
  46. #include "Node.hpp"
  47. namespace ZeroTier {
  48. static inline double _dist3d(int x1,int y1,int z1,int x2,int y2,int z2)
  49. throw()
  50. {
  51. double dx = ((double)x2 - (double)x1);
  52. double dy = ((double)y2 - (double)y1);
  53. double dz = ((double)z2 - (double)z1);
  54. return sqrt((dx * dx) + (dy * dy) + (dz * dz));
  55. }
  56. Cluster::Cluster(
  57. const RuntimeEnvironment *renv,
  58. uint16_t id,
  59. const std::vector<InetAddress> &zeroTierPhysicalEndpoints,
  60. int32_t x,
  61. int32_t y,
  62. int32_t z,
  63. void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
  64. void *sendFunctionArg,
  65. int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
  66. void *addressToLocationFunctionArg) :
  67. RR(renv),
  68. _sendFunction(sendFunction),
  69. _sendFunctionArg(sendFunctionArg),
  70. _addressToLocationFunction(addressToLocationFunction),
  71. _addressToLocationFunctionArg(addressToLocationFunctionArg),
  72. _x(x),
  73. _y(y),
  74. _z(z),
  75. _id(id),
  76. _zeroTierPhysicalEndpoints(zeroTierPhysicalEndpoints),
  77. _members(new _Member[ZT_CLUSTER_MAX_MEMBERS]),
  78. _peerAffinities(65536)
  79. {
  80. uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
  81. // Generate master secret by hashing the secret from our Identity key pair
  82. RR->identity.sha512PrivateKey(_masterSecret);
  83. // Generate our inbound message key, which is the master secret XORed with our ID and hashed twice
  84. memcpy(stmp,_masterSecret,sizeof(stmp));
  85. stmp[0] ^= Utils::hton(id);
  86. SHA512::hash(stmp,stmp,sizeof(stmp));
  87. SHA512::hash(stmp,stmp,sizeof(stmp));
  88. memcpy(_key,stmp,sizeof(_key));
  89. Utils::burn(stmp,sizeof(stmp));
  90. }
  91. Cluster::~Cluster()
  92. {
  93. Utils::burn(_masterSecret,sizeof(_masterSecret));
  94. Utils::burn(_key,sizeof(_key));
  95. delete [] _members;
  96. }
  97. void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
  98. {
  99. Buffer<ZT_CLUSTER_MAX_MESSAGE_LENGTH> dmsg;
  100. {
  101. // FORMAT: <[16] iv><[8] MAC><... data>
  102. if ((len < 24)||(len > ZT_CLUSTER_MAX_MESSAGE_LENGTH))
  103. return;
  104. // 16-byte IV: first 8 bytes XORed with key, last 8 bytes used as Salsa20 64-bit IV
  105. char keytmp[32];
  106. memcpy(keytmp,_key,32);
  107. for(int i=0;i<8;++i)
  108. keytmp[i] ^= reinterpret_cast<const char *>(msg)[i];
  109. Salsa20 s20(keytmp,256,reinterpret_cast<const char *>(msg) + 8);
  110. Utils::burn(keytmp,sizeof(keytmp));
  111. // One-time-use Poly1305 key from first 32 bytes of Salsa20 keystream (as per DJB/NaCl "standard")
  112. char polykey[ZT_POLY1305_KEY_LEN];
  113. memset(polykey,0,sizeof(polykey));
  114. s20.encrypt12(polykey,polykey,sizeof(polykey));
  115. // Compute 16-byte MAC
  116. char mac[ZT_POLY1305_MAC_LEN];
  117. Poly1305::compute(mac,reinterpret_cast<const char *>(msg) + 24,len - 24,polykey);
  118. // Check first 8 bytes of MAC against 64-bit MAC in stream
  119. if (!Utils::secureEq(mac,reinterpret_cast<const char *>(msg) + 16,8))
  120. return;
  121. // Decrypt!
  122. dmsg.setSize(len - 24);
  123. s20.decrypt12(reinterpret_cast<const char *>(msg) + 24,const_cast<void *>(dmsg.data()),dmsg.size());
  124. }
  125. if (dmsg.size() < 4)
  126. return;
  127. const uint16_t fromMemberId = dmsg.at<uint16_t>(0);
  128. unsigned int ptr = 2;
  129. if (fromMemberId == _id) // sanity check: we don't talk to ourselves
  130. return;
  131. const uint16_t toMemberId = dmsg.at<uint16_t>(ptr);
  132. ptr += 2;
  133. if (toMemberId != _id) // sanity check: message not for us?
  134. return;
  135. { // make sure sender is actually considered a member
  136. Mutex::Lock _l3(_memberIds_m);
  137. if (std::find(_memberIds.begin(),_memberIds.end(),fromMemberId) == _memberIds.end())
  138. return;
  139. }
  140. {
  141. _Member &m = _members[fromMemberId];
  142. Mutex::Lock mlck(m.lock);
  143. try {
  144. while (ptr < dmsg.size()) {
  145. const unsigned int mlen = dmsg.at<uint16_t>(ptr); ptr += 2;
  146. const unsigned int nextPtr = ptr + mlen;
  147. if (nextPtr > dmsg.size())
  148. break;
  149. int mtype = -1;
  150. try {
  151. switch((StateMessageType)(mtype = (int)dmsg[ptr++])) {
  152. default:
  153. break;
  154. case STATE_MESSAGE_ALIVE: {
  155. ptr += 7; // skip version stuff, not used yet
  156. m.x = dmsg.at<int32_t>(ptr); ptr += 4;
  157. m.y = dmsg.at<int32_t>(ptr); ptr += 4;
  158. m.z = dmsg.at<int32_t>(ptr); ptr += 4;
  159. ptr += 8; // skip local clock, not used
  160. m.load = dmsg.at<uint64_t>(ptr); ptr += 8;
  161. ptr += 8; // skip flags, unused
  162. #ifdef ZT_TRACE
  163. std::string addrs;
  164. #endif
  165. unsigned int physicalAddressCount = dmsg[ptr++];
  166. m.zeroTierPhysicalEndpoints.clear();
  167. for(unsigned int i=0;i<physicalAddressCount;++i) {
  168. m.zeroTierPhysicalEndpoints.push_back(InetAddress());
  169. ptr += m.zeroTierPhysicalEndpoints.back().deserialize(dmsg,ptr);
  170. if (!(m.zeroTierPhysicalEndpoints.back())) {
  171. m.zeroTierPhysicalEndpoints.pop_back();
  172. }
  173. #ifdef ZT_TRACE
  174. else {
  175. if (addrs.length() > 0)
  176. addrs.push_back(',');
  177. addrs.append(m.zeroTierPhysicalEndpoints.back().toString());
  178. }
  179. #endif
  180. }
  181. #ifdef ZT_TRACE
  182. if ((RR->node->now() - m.lastReceivedAliveAnnouncement) >= ZT_CLUSTER_TIMEOUT) {
  183. TRACE("[%u] I'm alive! peers close to %d,%d,%d can be redirected to: %s",(unsigned int)fromMemberId,m.x,m.y,m.z,addrs.c_str());
  184. }
  185. #endif
  186. m.lastReceivedAliveAnnouncement = RR->node->now();
  187. } break;
  188. case STATE_MESSAGE_HAVE_PEER: {
  189. const uint64_t now = RR->node->now();
  190. Identity id;
  191. InetAddress physicalAddress;
  192. ptr += id.deserialize(dmsg,ptr);
  193. ptr += physicalAddress.deserialize(dmsg,ptr);
  194. if (id) {
  195. // Forget any paths that we have to this peer at its address
  196. if (physicalAddress) {
  197. SharedPtr<Peer> myPeerRecord(RR->topology->getPeerNoCache(id.address(),now));
  198. if (myPeerRecord)
  199. myPeerRecord->removePathByAddress(physicalAddress);
  200. }
  201. // Always save identity to update file time
  202. RR->topology->saveIdentity(id);
  203. // Set peer affinity to its new home
  204. {
  205. Mutex::Lock _l2(_peerAffinities_m);
  206. _PA &pa = _peerAffinities[id.address()];
  207. pa.ts = now;
  208. pa.mid = fromMemberId;
  209. }
  210. TRACE("[%u] has %s @ %s",(unsigned int)fromMemberId,id.address().toString().c_str(),physicalAddress.toString().c_str());
  211. }
  212. } break;
  213. case STATE_MESSAGE_MULTICAST_LIKE: {
  214. const uint64_t nwid = dmsg.at<uint64_t>(ptr); ptr += 8;
  215. const Address address(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  216. const MAC mac(dmsg.field(ptr,6),6); ptr += 6;
  217. const uint32_t adi = dmsg.at<uint32_t>(ptr); ptr += 4;
  218. RR->mc->add(RR->node->now(),nwid,MulticastGroup(mac,adi),address);
  219. TRACE("[%u] %s likes %s/%.8x on %.16llx",(unsigned int)fromMemberId,address.toString().c_str(),mac.toString().c_str(),(unsigned int)adi,nwid);
  220. } break;
  221. case STATE_MESSAGE_COM: {
  222. CertificateOfMembership com;
  223. ptr += com.deserialize(dmsg,ptr);
  224. if (com) {
  225. TRACE("[%u] COM for %s on %.16llu rev %llu",(unsigned int)fromMemberId,com.issuedTo().toString().c_str(),com.networkId(),com.revision());
  226. }
  227. } break;
  228. case STATE_MESSAGE_PROXY_UNITE: {
  229. const Address localPeerAddress(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  230. const Address remotePeerAddress(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  231. const unsigned int numRemotePeerPaths = dmsg[ptr++];
  232. InetAddress remotePeerPaths[256]; // size is 8-bit, so 256 is max
  233. for(unsigned int i=0;i<numRemotePeerPaths;++i)
  234. ptr += remotePeerPaths[i].deserialize(dmsg,ptr);
  235. TRACE("[%u] requested proxy unite between local peer %s and remote peer %s",(unsigned int)fromMemberId,localPeerAddress.toString().c_str(),remotePeerAddress.toString().c_str());
  236. SharedPtr<Peer> localPeer(RR->topology->getPeer(localPeerAddress));
  237. if ((localPeer)&&(numRemotePeerPaths > 0)) {
  238. InetAddress bestLocalV4,bestLocalV6;
  239. localPeer->getBestActiveAddresses(RR->node->now(),bestLocalV4,bestLocalV6);
  240. InetAddress bestRemoteV4,bestRemoteV6;
  241. for(unsigned int i=0;i<numRemotePeerPaths;++i) {
  242. if ((bestRemoteV4)&&(bestRemoteV6))
  243. break;
  244. switch(remotePeerPaths[i].ss_family) {
  245. case AF_INET:
  246. if (!bestRemoteV4)
  247. bestRemoteV4 = remotePeerPaths[i];
  248. break;
  249. case AF_INET6:
  250. if (!bestRemoteV6)
  251. bestRemoteV6 = remotePeerPaths[i];
  252. break;
  253. }
  254. }
  255. Packet rendezvousForLocal(localPeerAddress,RR->identity.address(),Packet::VERB_RENDEZVOUS);
  256. rendezvousForLocal.append((uint8_t)0);
  257. remotePeerAddress.appendTo(rendezvousForLocal);
  258. Buffer<2048> rendezvousForRemote;
  259. remotePeerAddress.appendTo(rendezvousForRemote);
  260. rendezvousForRemote.append((uint8_t)Packet::VERB_RENDEZVOUS);
  261. const unsigned int rendezvousForOtherEndPayloadSizePtr = rendezvousForRemote.size();
  262. rendezvousForRemote.addSize(2); // space for actual packet payload length
  263. rendezvousForRemote.append((uint8_t)0); // flags == 0
  264. localPeerAddress.appendTo(rendezvousForRemote);
  265. bool haveMatch = false;
  266. if ((bestLocalV6)&&(bestRemoteV6)) {
  267. haveMatch = true;
  268. rendezvousForLocal.append((uint16_t)bestRemoteV6.port());
  269. rendezvousForLocal.append((uint8_t)16);
  270. rendezvousForLocal.append(bestRemoteV6.rawIpData(),16);
  271. rendezvousForRemote.append((uint16_t)bestLocalV6.port());
  272. rendezvousForRemote.append((uint8_t)16);
  273. rendezvousForRemote.append(bestLocalV6.rawIpData(),16);
  274. rendezvousForRemote.setAt<uint16_t>(rendezvousForOtherEndPayloadSizePtr,(uint16_t)(9 + 16));
  275. } else if ((bestLocalV4)&&(bestRemoteV4)) {
  276. haveMatch = true;
  277. rendezvousForLocal.append((uint16_t)bestRemoteV4.port());
  278. rendezvousForLocal.append((uint8_t)4);
  279. rendezvousForLocal.append(bestRemoteV4.rawIpData(),4);
  280. rendezvousForRemote.append((uint16_t)bestLocalV4.port());
  281. rendezvousForRemote.append((uint8_t)4);
  282. rendezvousForRemote.append(bestLocalV4.rawIpData(),4);
  283. rendezvousForRemote.setAt<uint16_t>(rendezvousForOtherEndPayloadSizePtr,(uint16_t)(9 + 4));
  284. }
  285. if (haveMatch) {
  286. _send(fromMemberId,STATE_MESSAGE_PROXY_SEND,rendezvousForRemote.data(),rendezvousForRemote.size());
  287. RR->sw->send(rendezvousForLocal,true,0);
  288. }
  289. }
  290. } break;
  291. case STATE_MESSAGE_PROXY_SEND: {
  292. const Address rcpt(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  293. const Packet::Verb verb = (Packet::Verb)dmsg[ptr++];
  294. const unsigned int len = dmsg.at<uint16_t>(ptr); ptr += 2;
  295. Packet outp(rcpt,RR->identity.address(),verb);
  296. outp.append(dmsg.field(ptr,len),len); ptr += len;
  297. RR->sw->send(outp,true,0);
  298. TRACE("[%u] proxy send %s to %s length %u",(unsigned int)fromMemberId,Packet::verbString(verb),rcpt.toString().c_str(),len);
  299. } break;
  300. }
  301. } catch ( ... ) {
  302. TRACE("invalid message of size %u type %d (inner decode), discarding",mlen,mtype);
  303. // drop invalids
  304. }
  305. ptr = nextPtr;
  306. }
  307. } catch ( ... ) {
  308. TRACE("invalid message (outer loop), discarding");
  309. // drop invalids
  310. }
  311. }
  312. }
  313. bool Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len,bool unite)
  314. {
  315. if (len > 16384) // sanity check
  316. return false;
  317. const uint64_t now = RR->node->now();
  318. unsigned int canHasPeer = 0;
  319. { // Anyone got this peer?
  320. Mutex::Lock _l2(_peerAffinities_m);
  321. _PA *pa = _peerAffinities.get(toPeerAddress);
  322. if ((pa)&&(pa->mid != _id)&&((now - pa->ts) < ZT_PEER_ACTIVITY_TIMEOUT))
  323. canHasPeer = pa->mid;
  324. else return false;
  325. }
  326. Buffer<2048> buf;
  327. if (unite) {
  328. InetAddress v4,v6;
  329. if (fromPeerAddress) {
  330. SharedPtr<Peer> fromPeer(RR->topology->getPeer(fromPeerAddress));
  331. if (fromPeer)
  332. fromPeer->getBestActiveAddresses(now,v4,v6);
  333. }
  334. uint8_t addrCount = 0;
  335. if (v4)
  336. ++addrCount;
  337. if (v6)
  338. ++addrCount;
  339. if (addrCount) {
  340. toPeerAddress.appendTo(buf);
  341. fromPeerAddress.appendTo(buf);
  342. buf.append(addrCount);
  343. if (v4)
  344. v4.serialize(buf);
  345. if (v6)
  346. v6.serialize(buf);
  347. }
  348. }
  349. {
  350. Mutex::Lock _l2(_members[canHasPeer].lock);
  351. if (buf.size() > 0)
  352. _send(canHasPeer,STATE_MESSAGE_PROXY_UNITE,buf.data(),buf.size());
  353. if (_members[canHasPeer].zeroTierPhysicalEndpoints.size() > 0)
  354. RR->node->putPacket(InetAddress(),_members[canHasPeer].zeroTierPhysicalEndpoints.front(),data,len);
  355. }
  356. TRACE("sendViaCluster(): relaying %u bytes from %s to %s by way of %u",len,fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str(),(unsigned int)canHasPeer);
  357. return true;
  358. }
  359. void Cluster::replicateHavePeer(const Identity &peerId,const InetAddress &physicalAddress)
  360. {
  361. const uint64_t now = RR->node->now();
  362. { // Use peer affinity table to track our own last announce time for peers
  363. Mutex::Lock _l2(_peerAffinities_m);
  364. _PA &pa = _peerAffinities[peerId.address()];
  365. if (pa.mid != _id) {
  366. pa.ts = now;
  367. pa.mid = _id;
  368. } else if ((now - pa.ts) < ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD) {
  369. return;
  370. } else {
  371. pa.ts = now;
  372. }
  373. }
  374. // announcement
  375. Buffer<4096> buf;
  376. peerId.serialize(buf,false);
  377. physicalAddress.serialize(buf);
  378. {
  379. Mutex::Lock _l(_memberIds_m);
  380. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  381. Mutex::Lock _l2(_members[*mid].lock);
  382. _send(*mid,STATE_MESSAGE_HAVE_PEER,buf.data(),buf.size());
  383. }
  384. }
  385. }
  386. void Cluster::replicateMulticastLike(uint64_t nwid,const Address &peerAddress,const MulticastGroup &group)
  387. {
  388. Buffer<2048> buf;
  389. buf.append((uint64_t)nwid);
  390. peerAddress.appendTo(buf);
  391. group.mac().appendTo(buf);
  392. buf.append((uint32_t)group.adi());
  393. TRACE("replicating %s MULTICAST_LIKE %.16llx/%s/%u to all members",peerAddress.toString().c_str(),nwid,group.mac().toString().c_str(),(unsigned int)group.adi());
  394. {
  395. Mutex::Lock _l(_memberIds_m);
  396. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  397. Mutex::Lock _l2(_members[*mid].lock);
  398. _send(*mid,STATE_MESSAGE_MULTICAST_LIKE,buf.data(),buf.size());
  399. }
  400. }
  401. }
  402. void Cluster::replicateCertificateOfNetworkMembership(const CertificateOfMembership &com)
  403. {
  404. Buffer<2048> buf;
  405. com.serialize(buf);
  406. TRACE("replicating %s COM for %.16llx to all members",com.issuedTo().toString().c_str(),com.networkId());
  407. {
  408. Mutex::Lock _l(_memberIds_m);
  409. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  410. Mutex::Lock _l2(_members[*mid].lock);
  411. _send(*mid,STATE_MESSAGE_COM,buf.data(),buf.size());
  412. }
  413. }
  414. }
  415. void Cluster::doPeriodicTasks()
  416. {
  417. const uint64_t now = RR->node->now();
  418. {
  419. Mutex::Lock _l(_memberIds_m);
  420. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  421. Mutex::Lock _l2(_members[*mid].lock);
  422. if ((now - _members[*mid].lastAnnouncedAliveTo) >= ((ZT_CLUSTER_TIMEOUT / 2) - 1000)) {
  423. Buffer<2048> alive;
  424. alive.append((uint16_t)ZEROTIER_ONE_VERSION_MAJOR);
  425. alive.append((uint16_t)ZEROTIER_ONE_VERSION_MINOR);
  426. alive.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
  427. alive.append((uint8_t)ZT_PROTO_VERSION);
  428. if (_addressToLocationFunction) {
  429. alive.append((int32_t)_x);
  430. alive.append((int32_t)_y);
  431. alive.append((int32_t)_z);
  432. } else {
  433. alive.append((int32_t)0);
  434. alive.append((int32_t)0);
  435. alive.append((int32_t)0);
  436. }
  437. alive.append((uint64_t)now);
  438. alive.append((uint64_t)0); // TODO: compute and send load average
  439. alive.append((uint64_t)0); // unused/reserved flags
  440. alive.append((uint8_t)_zeroTierPhysicalEndpoints.size());
  441. for(std::vector<InetAddress>::const_iterator pe(_zeroTierPhysicalEndpoints.begin());pe!=_zeroTierPhysicalEndpoints.end();++pe)
  442. pe->serialize(alive);
  443. _send(*mid,STATE_MESSAGE_ALIVE,alive.data(),alive.size());
  444. _members[*mid].lastAnnouncedAliveTo = now;
  445. }
  446. _flush(*mid); // does nothing if nothing to flush
  447. }
  448. }
  449. }
  450. void Cluster::addMember(uint16_t memberId)
  451. {
  452. if ((memberId >= ZT_CLUSTER_MAX_MEMBERS)||(memberId == _id))
  453. return;
  454. Mutex::Lock _l2(_members[memberId].lock);
  455. {
  456. Mutex::Lock _l(_memberIds_m);
  457. if (std::find(_memberIds.begin(),_memberIds.end(),memberId) != _memberIds.end())
  458. return;
  459. _memberIds.push_back(memberId);
  460. std::sort(_memberIds.begin(),_memberIds.end());
  461. }
  462. _members[memberId].clear();
  463. // Generate this member's message key from the master and its ID
  464. uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
  465. memcpy(stmp,_masterSecret,sizeof(stmp));
  466. stmp[0] ^= Utils::hton(memberId);
  467. SHA512::hash(stmp,stmp,sizeof(stmp));
  468. SHA512::hash(stmp,stmp,sizeof(stmp));
  469. memcpy(_members[memberId].key,stmp,sizeof(_members[memberId].key));
  470. Utils::burn(stmp,sizeof(stmp));
  471. // Prepare q
  472. _members[memberId].q.clear();
  473. char iv[16];
  474. Utils::getSecureRandom(iv,16);
  475. _members[memberId].q.append(iv,16);
  476. _members[memberId].q.addSize(8); // room for MAC
  477. _members[memberId].q.append((uint16_t)_id);
  478. _members[memberId].q.append((uint16_t)memberId);
  479. }
  480. void Cluster::removeMember(uint16_t memberId)
  481. {
  482. Mutex::Lock _l(_memberIds_m);
  483. std::vector<uint16_t> newMemberIds;
  484. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  485. if (*mid != memberId)
  486. newMemberIds.push_back(*mid);
  487. }
  488. _memberIds = newMemberIds;
  489. }
  490. bool Cluster::findBetterEndpoint(InetAddress &redirectTo,const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload)
  491. {
  492. if (_addressToLocationFunction) {
  493. // Pick based on location if it can be determined
  494. int px = 0,py = 0,pz = 0;
  495. if (_addressToLocationFunction(_addressToLocationFunctionArg,reinterpret_cast<const struct sockaddr_storage *>(&peerPhysicalAddress),&px,&py,&pz) == 0) {
  496. TRACE("no geolocation data for %s (geo-lookup is lazy/async so it may work next time)",peerPhysicalAddress.toIpString().c_str());
  497. return false;
  498. }
  499. // Find member closest to this peer
  500. const uint64_t now = RR->node->now();
  501. std::vector<InetAddress> best; // initial "best" is for peer to stay put
  502. const double currentDistance = _dist3d(_x,_y,_z,px,py,pz);
  503. double bestDistance = (offload ? 2147483648.0 : currentDistance);
  504. unsigned int bestMember = _id;
  505. {
  506. Mutex::Lock _l(_memberIds_m);
  507. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  508. _Member &m = _members[*mid];
  509. Mutex::Lock _ml(m.lock);
  510. // Consider member if it's alive and has sent us a location and one or more physical endpoints to send peers to
  511. if ( ((now - m.lastReceivedAliveAnnouncement) < ZT_CLUSTER_TIMEOUT) && ((m.x != 0)||(m.y != 0)||(m.z != 0)) && (m.zeroTierPhysicalEndpoints.size() > 0) ) {
  512. double mdist = _dist3d(m.x,m.y,m.z,px,py,pz);
  513. if (mdist < bestDistance) {
  514. bestDistance = mdist;
  515. bestMember = *mid;
  516. best = m.zeroTierPhysicalEndpoints;
  517. }
  518. }
  519. }
  520. }
  521. // Suggestion redirection if a closer member was found
  522. for(std::vector<InetAddress>::const_iterator a(best.begin());a!=best.end();++a) {
  523. if (a->ss_family == peerPhysicalAddress.ss_family) {
  524. TRACE("%s at [%d,%d,%d] is %f from us but %f from %u, can redirect to %s",peerAddress.toString().c_str(),px,py,pz,currentDistance,bestDistance,bestMember,a->toString().c_str());
  525. redirectTo = *a;
  526. return true;
  527. }
  528. }
  529. TRACE("%s at [%d,%d,%d] is %f from us, no better endpoints found",peerAddress.toString().c_str(),px,py,pz,currentDistance);
  530. return false;
  531. } else {
  532. // TODO: pick based on load if no location info?
  533. return false;
  534. }
  535. }
  536. void Cluster::status(ZT_ClusterStatus &status) const
  537. {
  538. const uint64_t now = RR->node->now();
  539. memset(&status,0,sizeof(ZT_ClusterStatus));
  540. ZT_ClusterMemberStatus *ms[ZT_CLUSTER_MAX_MEMBERS];
  541. memset(ms,0,sizeof(ms));
  542. status.myId = _id;
  543. ms[_id] = &(status.members[status.clusterSize++]);
  544. ms[_id]->id = _id;
  545. ms[_id]->alive = 1;
  546. ms[_id]->x = _x;
  547. ms[_id]->y = _y;
  548. ms[_id]->z = _z;
  549. ms[_id]->peers = RR->topology->countAlive();
  550. for(std::vector<InetAddress>::const_iterator ep(_zeroTierPhysicalEndpoints.begin());ep!=_zeroTierPhysicalEndpoints.end();++ep) {
  551. if (ms[_id]->numZeroTierPhysicalEndpoints >= ZT_CLUSTER_MAX_ZT_PHYSICAL_ADDRESSES) // sanity check
  552. break;
  553. memcpy(&(ms[_id]->zeroTierPhysicalEndpoints[ms[_id]->numZeroTierPhysicalEndpoints++]),&(*ep),sizeof(struct sockaddr_storage));
  554. }
  555. {
  556. Mutex::Lock _l1(_memberIds_m);
  557. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  558. if (status.clusterSize >= ZT_CLUSTER_MAX_MEMBERS) // sanity check
  559. break;
  560. ZT_ClusterMemberStatus *s = ms[*mid] = &(status.members[status.clusterSize++]);
  561. _Member &m = _members[*mid];
  562. Mutex::Lock ml(m.lock);
  563. s->id = *mid;
  564. s->msSinceLastHeartbeat = (unsigned int)std::min((uint64_t)(~((unsigned int)0)),(now - m.lastReceivedAliveAnnouncement));
  565. s->alive = (s->msSinceLastHeartbeat < ZT_CLUSTER_TIMEOUT) ? 1 : 0;
  566. s->x = m.x;
  567. s->y = m.y;
  568. s->z = m.z;
  569. s->load = m.load;
  570. for(std::vector<InetAddress>::const_iterator ep(m.zeroTierPhysicalEndpoints.begin());ep!=m.zeroTierPhysicalEndpoints.end();++ep) {
  571. if (s->numZeroTierPhysicalEndpoints >= ZT_CLUSTER_MAX_ZT_PHYSICAL_ADDRESSES) // sanity check
  572. break;
  573. memcpy(&(s->zeroTierPhysicalEndpoints[s->numZeroTierPhysicalEndpoints++]),&(*ep),sizeof(struct sockaddr_storage));
  574. }
  575. }
  576. }
  577. {
  578. Mutex::Lock _l2(_peerAffinities_m);
  579. Address *k = (Address *)0;
  580. _PA *v = (_PA *)0;
  581. Hashtable< Address,_PA >::Iterator i(const_cast<Cluster *>(this)->_peerAffinities);
  582. while (i.next(k,v)) {
  583. if ( (ms[v->mid]) && (v->mid != _id) && ((now - v->ts) < ZT_PEER_ACTIVITY_TIMEOUT) )
  584. ++ms[v->mid]->peers;
  585. }
  586. }
  587. }
  588. void Cluster::_send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len)
  589. {
  590. if ((len + 3) > (ZT_CLUSTER_MAX_MESSAGE_LENGTH - (24 + 2 + 2))) // sanity check
  591. return;
  592. _Member &m = _members[memberId];
  593. // assumes m.lock is locked!
  594. if ((m.q.size() + len + 3) > ZT_CLUSTER_MAX_MESSAGE_LENGTH)
  595. _flush(memberId);
  596. m.q.append((uint16_t)(len + 1));
  597. m.q.append((uint8_t)type);
  598. m.q.append(msg,len);
  599. }
  600. void Cluster::_flush(uint16_t memberId)
  601. {
  602. _Member &m = _members[memberId];
  603. // assumes m.lock is locked!
  604. if (m.q.size() > (24 + 2 + 2)) { // 16-byte IV + 8-byte MAC + 2 byte from-member-ID + 2 byte to-member-ID
  605. // Create key from member's key and IV
  606. char keytmp[32];
  607. memcpy(keytmp,m.key,32);
  608. for(int i=0;i<8;++i)
  609. keytmp[i] ^= m.q[i];
  610. Salsa20 s20(keytmp,256,m.q.field(8,8));
  611. Utils::burn(keytmp,sizeof(keytmp));
  612. // One-time-use Poly1305 key from first 32 bytes of Salsa20 keystream (as per DJB/NaCl "standard")
  613. char polykey[ZT_POLY1305_KEY_LEN];
  614. memset(polykey,0,sizeof(polykey));
  615. s20.encrypt12(polykey,polykey,sizeof(polykey));
  616. // Encrypt m.q in place
  617. s20.encrypt12(reinterpret_cast<const char *>(m.q.data()) + 24,const_cast<char *>(reinterpret_cast<const char *>(m.q.data())) + 24,m.q.size() - 24);
  618. // Add MAC for authentication (encrypt-then-MAC)
  619. char mac[ZT_POLY1305_MAC_LEN];
  620. Poly1305::compute(mac,reinterpret_cast<const char *>(m.q.data()) + 24,m.q.size() - 24,polykey);
  621. memcpy(m.q.field(16,8),mac,8);
  622. // Send!
  623. _sendFunction(_sendFunctionArg,memberId,m.q.data(),m.q.size());
  624. // Prepare for more
  625. m.q.clear();
  626. char iv[16];
  627. Utils::getSecureRandom(iv,16);
  628. m.q.append(iv,16);
  629. m.q.addSize(8); // room for MAC
  630. m.q.append((uint16_t)_id); // from member ID
  631. m.q.append((uint16_t)memberId); // to member ID
  632. }
  633. }
  634. } // namespace ZeroTier
  635. #endif // ZT_ENABLE_CLUSTER