Cluster.cpp 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694
  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 "Packet.hpp"
  43. #include "Identity.hpp"
  44. #include "Peer.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. {
  79. uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
  80. // Generate master secret by hashing the secret from our Identity key pair
  81. RR->identity.sha512PrivateKey(_masterSecret);
  82. // Generate our inbound message key, which is the master secret XORed with our ID and hashed twice
  83. memcpy(stmp,_masterSecret,sizeof(stmp));
  84. stmp[0] ^= Utils::hton(id);
  85. SHA512::hash(stmp,stmp,sizeof(stmp));
  86. SHA512::hash(stmp,stmp,sizeof(stmp));
  87. memcpy(_key,stmp,sizeof(_key));
  88. Utils::burn(stmp,sizeof(stmp));
  89. }
  90. Cluster::~Cluster()
  91. {
  92. Utils::burn(_masterSecret,sizeof(_masterSecret));
  93. Utils::burn(_key,sizeof(_key));
  94. delete [] _members;
  95. }
  96. void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
  97. {
  98. Buffer<ZT_CLUSTER_MAX_MESSAGE_LENGTH> dmsg;
  99. {
  100. // FORMAT: <[16] iv><[8] MAC><... data>
  101. if ((len < 24)||(len > ZT_CLUSTER_MAX_MESSAGE_LENGTH))
  102. return;
  103. // 16-byte IV: first 8 bytes XORed with key, last 8 bytes used as Salsa20 64-bit IV
  104. char keytmp[32];
  105. memcpy(keytmp,_key,32);
  106. for(int i=0;i<8;++i)
  107. keytmp[i] ^= reinterpret_cast<const char *>(msg)[i];
  108. Salsa20 s20(keytmp,256,reinterpret_cast<const char *>(msg) + 8);
  109. Utils::burn(keytmp,sizeof(keytmp));
  110. // One-time-use Poly1305 key from first 32 bytes of Salsa20 keystream (as per DJB/NaCl "standard")
  111. char polykey[ZT_POLY1305_KEY_LEN];
  112. memset(polykey,0,sizeof(polykey));
  113. s20.encrypt12(polykey,polykey,sizeof(polykey));
  114. // Compute 16-byte MAC
  115. char mac[ZT_POLY1305_MAC_LEN];
  116. Poly1305::compute(mac,reinterpret_cast<const char *>(msg) + 24,len - 24,polykey);
  117. // Check first 8 bytes of MAC against 64-bit MAC in stream
  118. if (!Utils::secureEq(mac,reinterpret_cast<const char *>(msg) + 16,8))
  119. return;
  120. // Decrypt!
  121. dmsg.setSize(len - 24);
  122. s20.decrypt12(reinterpret_cast<const char *>(msg) + 24,const_cast<void *>(dmsg.data()),dmsg.size());
  123. }
  124. if (dmsg.size() < 4)
  125. return;
  126. const uint16_t fromMemberId = dmsg.at<uint16_t>(0);
  127. unsigned int ptr = 2;
  128. if (fromMemberId == _id) // sanity check: we don't talk to ourselves
  129. return;
  130. const uint16_t toMemberId = dmsg.at<uint16_t>(ptr);
  131. ptr += 2;
  132. if (toMemberId != _id) // sanity check: message not for us?
  133. return;
  134. { // make sure sender is actually considered a member
  135. Mutex::Lock _l3(_memberIds_m);
  136. if (std::find(_memberIds.begin(),_memberIds.end(),fromMemberId) == _memberIds.end())
  137. return;
  138. }
  139. {
  140. _Member &m = _members[fromMemberId];
  141. Mutex::Lock mlck(m.lock);
  142. try {
  143. while (ptr < dmsg.size()) {
  144. const unsigned int mlen = dmsg.at<uint16_t>(ptr); ptr += 2;
  145. const unsigned int nextPtr = ptr + mlen;
  146. if (nextPtr > dmsg.size())
  147. break;
  148. int mtype = -1;
  149. try {
  150. switch((StateMessageType)(mtype = (int)dmsg[ptr++])) {
  151. default:
  152. break;
  153. case STATE_MESSAGE_ALIVE: {
  154. ptr += 7; // skip version stuff, not used yet
  155. m.x = dmsg.at<int32_t>(ptr); ptr += 4;
  156. m.y = dmsg.at<int32_t>(ptr); ptr += 4;
  157. m.z = dmsg.at<int32_t>(ptr); ptr += 4;
  158. ptr += 8; // skip local clock, not used
  159. m.load = dmsg.at<uint64_t>(ptr); ptr += 8;
  160. ptr += 8; // skip flags, unused
  161. #ifdef ZT_TRACE
  162. std::string addrs;
  163. #endif
  164. unsigned int physicalAddressCount = dmsg[ptr++];
  165. m.zeroTierPhysicalEndpoints.clear();
  166. for(unsigned int i=0;i<physicalAddressCount;++i) {
  167. m.zeroTierPhysicalEndpoints.push_back(InetAddress());
  168. ptr += m.zeroTierPhysicalEndpoints.back().deserialize(dmsg,ptr);
  169. if (!(m.zeroTierPhysicalEndpoints.back())) {
  170. m.zeroTierPhysicalEndpoints.pop_back();
  171. }
  172. #ifdef ZT_TRACE
  173. else {
  174. if (addrs.length() > 0)
  175. addrs.push_back(',');
  176. addrs.append(m.zeroTierPhysicalEndpoints.back().toString());
  177. }
  178. #endif
  179. }
  180. m.lastReceivedAliveAnnouncement = RR->node->now();
  181. #ifdef ZT_TRACE
  182. 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());
  183. #endif
  184. } break;
  185. case STATE_MESSAGE_HAVE_PEER: {
  186. try {
  187. Identity id;
  188. ptr += id.deserialize(dmsg,ptr);
  189. if (id) {
  190. RR->topology->saveIdentity(id);
  191. { // Add or update peer affinity entry
  192. _PeerAffinity pa(id.address(),fromMemberId,RR->node->now());
  193. Mutex::Lock _l2(_peerAffinities_m);
  194. std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),pa)); // O(log(n))
  195. if ((i != _peerAffinities.end())&&(i->key == pa.key)) {
  196. i->timestamp = pa.timestamp;
  197. } else {
  198. _peerAffinities.push_back(pa);
  199. std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now
  200. }
  201. }
  202. TRACE("[%u] has %s",(unsigned int)fromMemberId,id.address().toString().c_str());
  203. }
  204. } catch ( ... ) {
  205. // ignore invalid identities
  206. }
  207. } break;
  208. case STATE_MESSAGE_MULTICAST_LIKE: {
  209. const uint64_t nwid = dmsg.at<uint64_t>(ptr); ptr += 8;
  210. const Address address(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  211. const MAC mac(dmsg.field(ptr,6),6); ptr += 6;
  212. const uint32_t adi = dmsg.at<uint32_t>(ptr); ptr += 4;
  213. RR->mc->add(RR->node->now(),nwid,MulticastGroup(mac,adi),address);
  214. TRACE("[%u] %s likes %s/%u on %.16llu",(unsigned int)fromMemberId,address.toString().c_str(),mac.toString().c_str(),(unsigned int)adi,nwid);
  215. } break;
  216. case STATE_MESSAGE_COM: {
  217. CertificateOfMembership com;
  218. ptr += com.deserialize(dmsg,ptr);
  219. if (com) {
  220. TRACE("[%u] COM for %s on %.16llu rev %llu",(unsigned int)fromMemberId,com.issuedTo().toString().c_str(),com.networkId(),com.revision());
  221. }
  222. } break;
  223. case STATE_MESSAGE_RELAY: {
  224. const unsigned int numRemotePeerPaths = dmsg[ptr++];
  225. InetAddress remotePeerPaths[256]; // size is 8-bit, so 256 is max
  226. for(unsigned int i=0;i<numRemotePeerPaths;++i)
  227. ptr += remotePeerPaths[i].deserialize(dmsg,ptr);
  228. const unsigned int packetLen = dmsg.at<uint16_t>(ptr); ptr += 2;
  229. const void *packet = (const void *)dmsg.field(ptr,packetLen); ptr += packetLen;
  230. if (packetLen >= ZT_PROTO_MIN_FRAGMENT_LENGTH) { // ignore anything too short to contain a dest address
  231. const Address destinationAddress(reinterpret_cast<const char *>(packet) + 8,ZT_ADDRESS_LENGTH);
  232. TRACE("[%u] relay %u bytes to %s (%u remote paths included)",(unsigned int)fromMemberId,packetLen,destinationAddress.toString().c_str(),numRemotePeerPaths);
  233. SharedPtr<Peer> destinationPeer(RR->topology->getPeer(destinationAddress));
  234. if (destinationPeer) {
  235. if (
  236. (destinationPeer->send(RR,packet,packetLen,RR->node->now()))&&
  237. (numRemotePeerPaths > 0)&&
  238. (packetLen >= 18)&&
  239. (reinterpret_cast<const unsigned char *>(packet)[ZT_PACKET_FRAGMENT_IDX_FRAGMENT_INDICATOR] == ZT_PACKET_FRAGMENT_INDICATOR)
  240. ) {
  241. // If remote peer paths were sent with this relayed packet, we do
  242. // RENDEZVOUS. It's handled here for cluster-relayed packets since
  243. // we don't have both Peer records so this is a different path.
  244. const Address remotePeerAddress(reinterpret_cast<const char *>(packet) + 13,ZT_ADDRESS_LENGTH);
  245. InetAddress bestDestV4,bestDestV6;
  246. destinationPeer->getBestActiveAddresses(RR->node->now(),bestDestV4,bestDestV6);
  247. InetAddress bestRemoteV4,bestRemoteV6;
  248. for(unsigned int i=0;i<numRemotePeerPaths;++i) {
  249. if ((bestRemoteV4)&&(bestRemoteV6))
  250. break;
  251. switch(remotePeerPaths[i].ss_family) {
  252. case AF_INET:
  253. if (!bestRemoteV4)
  254. bestRemoteV4 = remotePeerPaths[i];
  255. break;
  256. case AF_INET6:
  257. if (!bestRemoteV6)
  258. bestRemoteV6 = remotePeerPaths[i];
  259. break;
  260. }
  261. }
  262. Packet rendezvousForDest(destinationAddress,RR->identity.address(),Packet::VERB_RENDEZVOUS);
  263. rendezvousForDest.append((uint8_t)0);
  264. remotePeerAddress.appendTo(rendezvousForDest);
  265. Buffer<2048> rendezvousForOtherEnd;
  266. remotePeerAddress.appendTo(rendezvousForOtherEnd);
  267. rendezvousForOtherEnd.append((uint8_t)Packet::VERB_RENDEZVOUS);
  268. const unsigned int rendezvousForOtherEndPayloadSizePtr = rendezvousForOtherEnd.size();
  269. rendezvousForOtherEnd.addSize(2); // space for actual packet payload length
  270. rendezvousForOtherEnd.append((uint8_t)0); // flags == 0
  271. destinationAddress.appendTo(rendezvousForOtherEnd);
  272. bool haveMatch = false;
  273. if ((bestDestV6)&&(bestRemoteV6)) {
  274. haveMatch = true;
  275. rendezvousForDest.append((uint16_t)bestRemoteV6.port());
  276. rendezvousForDest.append((uint8_t)16);
  277. rendezvousForDest.append(bestRemoteV6.rawIpData(),16);
  278. rendezvousForOtherEnd.append((uint16_t)bestDestV6.port());
  279. rendezvousForOtherEnd.append((uint8_t)16);
  280. rendezvousForOtherEnd.append(bestDestV6.rawIpData(),16);
  281. rendezvousForOtherEnd.setAt<uint16_t>(rendezvousForOtherEndPayloadSizePtr,(uint16_t)(9 + 16));
  282. } else if ((bestDestV4)&&(bestRemoteV4)) {
  283. haveMatch = true;
  284. rendezvousForDest.append((uint16_t)bestRemoteV4.port());
  285. rendezvousForDest.append((uint8_t)4);
  286. rendezvousForDest.append(bestRemoteV4.rawIpData(),4);
  287. rendezvousForOtherEnd.append((uint16_t)bestDestV4.port());
  288. rendezvousForOtherEnd.append((uint8_t)4);
  289. rendezvousForOtherEnd.append(bestDestV4.rawIpData(),4);
  290. rendezvousForOtherEnd.setAt<uint16_t>(rendezvousForOtherEndPayloadSizePtr,(uint16_t)(9 + 4));
  291. }
  292. if (haveMatch) {
  293. _send(fromMemberId,STATE_MESSAGE_PROXY_SEND,rendezvousForOtherEnd.data(),rendezvousForOtherEnd.size());
  294. RR->sw->send(rendezvousForDest,true,0);
  295. }
  296. }
  297. }
  298. }
  299. } break;
  300. case STATE_MESSAGE_PROXY_SEND: {
  301. const Address rcpt(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH);
  302. const Packet::Verb verb = (Packet::Verb)dmsg[ptr++];
  303. const unsigned int len = dmsg.at<uint16_t>(ptr); ptr += 2;
  304. Packet outp(rcpt,RR->identity.address(),verb);
  305. outp.append(dmsg.field(ptr,len),len);
  306. RR->sw->send(outp,true,0);
  307. TRACE("[%u] proxy send %s to %s length %u",(unsigned int)fromMemberId,Packet::verbString(verb),rcpt.toString().c_str(),len);
  308. } break;
  309. }
  310. } catch ( ... ) {
  311. TRACE("invalid message of size %u type %d (inner decode), discarding",mlen,mtype);
  312. // drop invalids
  313. }
  314. ptr = nextPtr;
  315. }
  316. } catch ( ... ) {
  317. TRACE("invalid message (outer loop), discarding");
  318. // drop invalids
  319. }
  320. }
  321. }
  322. bool Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len)
  323. {
  324. if (len > 16384) // sanity check
  325. return false;
  326. uint64_t mostRecentTimestamp = 0;
  327. uint16_t canHasPeer = 0;
  328. { // Anyone got this peer?
  329. Mutex::Lock _l2(_peerAffinities_m);
  330. std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),_PeerAffinity(toPeerAddress,0,0))); // O(log(n))
  331. while ((i != _peerAffinities.end())&&(i->address() == toPeerAddress)) {
  332. uint16_t mid = i->clusterMemberId();
  333. if ((mid != _id)&&(i->timestamp > mostRecentTimestamp)) {
  334. mostRecentTimestamp = i->timestamp;
  335. canHasPeer = mid;
  336. }
  337. }
  338. }
  339. const uint64_t now = RR->node->now();
  340. if ((now - mostRecentTimestamp) < ZT_PEER_ACTIVITY_TIMEOUT) {
  341. Buffer<16384> buf;
  342. InetAddress v4,v6;
  343. if (fromPeerAddress) {
  344. SharedPtr<Peer> fromPeer(RR->topology->getPeer(fromPeerAddress));
  345. if (fromPeer)
  346. fromPeer->getBestActiveAddresses(now,v4,v6);
  347. }
  348. buf.append((uint8_t)( (v4) ? ((v6) ? 2 : 1) : ((v6) ? 1 : 0) ));
  349. if (v4)
  350. v4.serialize(buf);
  351. if (v6)
  352. v6.serialize(buf);
  353. buf.append((uint16_t)len);
  354. buf.append(data,len);
  355. {
  356. Mutex::Lock _l2(_members[canHasPeer].lock);
  357. _send(canHasPeer,STATE_MESSAGE_RELAY,buf.data(),buf.size());
  358. }
  359. 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);
  360. return true;
  361. } else {
  362. TRACE("sendViaCluster(): unable to relay %u bytes from %s to %s since no cluster members seem to have it!",len,fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str());
  363. return false;
  364. }
  365. }
  366. void Cluster::replicateHavePeer(const Identity &peerId)
  367. {
  368. { // Use peer affinity table to track our own last announce time for peers
  369. _PeerAffinity pa(peerId.address(),_id,RR->node->now());
  370. Mutex::Lock _l2(_peerAffinities_m);
  371. std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),pa)); // O(log(n))
  372. if ((i != _peerAffinities.end())&&(i->key == pa.key)) {
  373. if ((pa.timestamp - i->timestamp) >= ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD) {
  374. i->timestamp = pa.timestamp;
  375. // continue to announcement
  376. } else {
  377. // we've already announced this peer recently, so skip
  378. return;
  379. }
  380. } else {
  381. _peerAffinities.push_back(pa);
  382. std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now
  383. // continue to announcement
  384. }
  385. }
  386. // announcement
  387. Buffer<4096> buf;
  388. peerId.serialize(buf,false);
  389. {
  390. Mutex::Lock _l(_memberIds_m);
  391. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  392. Mutex::Lock _l2(_members[*mid].lock);
  393. _send(*mid,STATE_MESSAGE_HAVE_PEER,buf.data(),buf.size());
  394. }
  395. }
  396. }
  397. void Cluster::replicateMulticastLike(uint64_t nwid,const Address &peerAddress,const MulticastGroup &group)
  398. {
  399. Buffer<4096> buf;
  400. buf.append((uint64_t)nwid);
  401. peerAddress.appendTo(buf);
  402. group.mac().appendTo(buf);
  403. buf.append((uint32_t)group.adi());
  404. {
  405. Mutex::Lock _l(_memberIds_m);
  406. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  407. Mutex::Lock _l2(_members[*mid].lock);
  408. _send(*mid,STATE_MESSAGE_MULTICAST_LIKE,buf.data(),buf.size());
  409. }
  410. }
  411. }
  412. void Cluster::replicateCertificateOfNetworkMembership(const CertificateOfMembership &com)
  413. {
  414. Buffer<4096> buf;
  415. com.serialize(buf);
  416. {
  417. Mutex::Lock _l(_memberIds_m);
  418. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  419. Mutex::Lock _l2(_members[*mid].lock);
  420. _send(*mid,STATE_MESSAGE_COM,buf.data(),buf.size());
  421. }
  422. }
  423. }
  424. void Cluster::doPeriodicTasks()
  425. {
  426. const uint64_t now = RR->node->now();
  427. {
  428. Mutex::Lock _l(_memberIds_m);
  429. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  430. Mutex::Lock _l2(_members[*mid].lock);
  431. if ((now - _members[*mid].lastAnnouncedAliveTo) >= ((ZT_CLUSTER_TIMEOUT / 2) - 1000)) {
  432. Buffer<2048> alive;
  433. alive.append((uint16_t)ZEROTIER_ONE_VERSION_MAJOR);
  434. alive.append((uint16_t)ZEROTIER_ONE_VERSION_MINOR);
  435. alive.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
  436. alive.append((uint8_t)ZT_PROTO_VERSION);
  437. if (_addressToLocationFunction) {
  438. alive.append((int32_t)_x);
  439. alive.append((int32_t)_y);
  440. alive.append((int32_t)_z);
  441. } else {
  442. alive.append((int32_t)0);
  443. alive.append((int32_t)0);
  444. alive.append((int32_t)0);
  445. }
  446. alive.append((uint64_t)now);
  447. alive.append((uint64_t)0); // TODO: compute and send load average
  448. alive.append((uint64_t)0); // unused/reserved flags
  449. alive.append((uint8_t)_zeroTierPhysicalEndpoints.size());
  450. for(std::vector<InetAddress>::const_iterator pe(_zeroTierPhysicalEndpoints.begin());pe!=_zeroTierPhysicalEndpoints.end();++pe)
  451. pe->serialize(alive);
  452. _send(*mid,STATE_MESSAGE_ALIVE,alive.data(),alive.size());
  453. _members[*mid].lastAnnouncedAliveTo = now;
  454. }
  455. _flush(*mid); // does nothing if nothing to flush
  456. }
  457. }
  458. }
  459. void Cluster::addMember(uint16_t memberId)
  460. {
  461. if ((memberId >= ZT_CLUSTER_MAX_MEMBERS)||(memberId == _id))
  462. return;
  463. Mutex::Lock _l2(_members[memberId].lock);
  464. {
  465. Mutex::Lock _l(_memberIds_m);
  466. if (std::find(_memberIds.begin(),_memberIds.end(),memberId) != _memberIds.end())
  467. return;
  468. _memberIds.push_back(memberId);
  469. std::sort(_memberIds.begin(),_memberIds.end());
  470. }
  471. _members[memberId].clear();
  472. // Generate this member's message key from the master and its ID
  473. uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
  474. memcpy(stmp,_masterSecret,sizeof(stmp));
  475. stmp[0] ^= Utils::hton(memberId);
  476. SHA512::hash(stmp,stmp,sizeof(stmp));
  477. SHA512::hash(stmp,stmp,sizeof(stmp));
  478. memcpy(_members[memberId].key,stmp,sizeof(_members[memberId].key));
  479. Utils::burn(stmp,sizeof(stmp));
  480. // Prepare q
  481. _members[memberId].q.clear();
  482. char iv[16];
  483. Utils::getSecureRandom(iv,16);
  484. _members[memberId].q.append(iv,16);
  485. _members[memberId].q.addSize(8); // room for MAC
  486. _members[memberId].q.append((uint16_t)_id);
  487. _members[memberId].q.append((uint16_t)memberId);
  488. }
  489. void Cluster::removeMember(uint16_t memberId)
  490. {
  491. Mutex::Lock _l(_memberIds_m);
  492. std::vector<uint16_t> newMemberIds;
  493. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  494. if (*mid != memberId)
  495. newMemberIds.push_back(*mid);
  496. }
  497. _memberIds = newMemberIds;
  498. }
  499. bool Cluster::redirectPeer(const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload)
  500. {
  501. if (!peerPhysicalAddress) // sanity check
  502. return false;
  503. if (_addressToLocationFunction) {
  504. // Pick based on location if it can be determined
  505. int px = 0,py = 0,pz = 0;
  506. if (_addressToLocationFunction(_addressToLocationFunctionArg,reinterpret_cast<const struct sockaddr_storage *>(&peerPhysicalAddress),&px,&py,&pz) == 0) {
  507. TRACE("no geolocation available for %s",peerPhysicalAddress.toIpString().c_str());
  508. return false;
  509. }
  510. // Find member closest to this peer
  511. const uint64_t now = RR->node->now();
  512. std::vector<InetAddress> best; // initial "best" is for peer to stay put
  513. const double currentDistance = _dist3d(_x,_y,_z,px,py,pz);
  514. double bestDistance = (offload ? 2147483648.0 : currentDistance);
  515. unsigned int bestMember = _id;
  516. TRACE("%s is at %d,%d,%d -- looking for anyone closer than %d,%d,%d (%fkm)",peerPhysicalAddress.toString().c_str(),px,py,pz,_x,_y,_z,bestDistance);
  517. {
  518. Mutex::Lock _l(_memberIds_m);
  519. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  520. _Member &m = _members[*mid];
  521. Mutex::Lock _ml(m.lock);
  522. // Consider member if it's alive and has sent us a location and one or more physical endpoints to send peers to
  523. if ( ((now - m.lastReceivedAliveAnnouncement) < ZT_CLUSTER_TIMEOUT) && ((m.x != 0)||(m.y != 0)||(m.z != 0)) && (m.zeroTierPhysicalEndpoints.size() > 0) ) {
  524. double mdist = _dist3d(m.x,m.y,m.z,px,py,pz);
  525. if (mdist < bestDistance) {
  526. bestDistance = mdist;
  527. bestMember = *mid;
  528. best = m.zeroTierPhysicalEndpoints;
  529. }
  530. }
  531. }
  532. }
  533. if (best.size() > 0) {
  534. TRACE("%s seems closer to %u at %fkm, suggesting redirect...",peerAddress.toString().c_str(),bestMember,bestDistance);
  535. /* if (peer->remoteVersionProtocol() >= 5) {
  536. // If it's a newer peer send VERB_PUSH_DIRECT_PATHS which is more idiomatic
  537. } else { */
  538. // Otherwise send VERB_RENDEZVOUS for ourselves, which will trick peers into trying other endpoints for us even if they're too old for PUSH_DIRECT_PATHS
  539. for(std::vector<InetAddress>::const_iterator a(best.begin());a!=best.end();++a) {
  540. if ((a->ss_family == AF_INET)||(a->ss_family == AF_INET6)) {
  541. Packet outp(peerAddress,RR->identity.address(),Packet::VERB_RENDEZVOUS);
  542. outp.append((uint8_t)0); // no flags
  543. RR->identity.address().appendTo(outp); // HACK: rendezvous with ourselves! with really old peers this will only work if I'm a root server!
  544. outp.append((uint16_t)a->port());
  545. if (a->ss_family == AF_INET) {
  546. outp.append((uint8_t)4);
  547. outp.append(a->rawIpData(),4);
  548. } else {
  549. outp.append((uint8_t)16);
  550. outp.append(a->rawIpData(),16);
  551. }
  552. RR->sw->send(outp,true,0);
  553. }
  554. }
  555. //}
  556. return true;
  557. } else {
  558. TRACE("peer %s is at [%d,%d,%d], distance to us is %f and this seems to be the best",peerAddress.toString().c_str(),px,py,pz,currentDistance);
  559. return false;
  560. }
  561. } else {
  562. // TODO: pick based on load if no location info?
  563. return false;
  564. }
  565. }
  566. void Cluster::_send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len)
  567. {
  568. if ((len + 3) > (ZT_CLUSTER_MAX_MESSAGE_LENGTH - (24 + 2 + 2))) // sanity check
  569. return;
  570. _Member &m = _members[memberId];
  571. // assumes m.lock is locked!
  572. if ((m.q.size() + len + 3) > ZT_CLUSTER_MAX_MESSAGE_LENGTH)
  573. _flush(memberId);
  574. m.q.append((uint16_t)(len + 1));
  575. m.q.append((uint8_t)type);
  576. m.q.append(msg,len);
  577. }
  578. void Cluster::_flush(uint16_t memberId)
  579. {
  580. _Member &m = _members[memberId];
  581. // assumes m.lock is locked!
  582. if (m.q.size() > (24 + 2 + 2)) { // 16-byte IV + 8-byte MAC + 2 byte from-member-ID + 2 byte to-member-ID
  583. // Create key from member's key and IV
  584. char keytmp[32];
  585. memcpy(keytmp,m.key,32);
  586. for(int i=0;i<8;++i)
  587. keytmp[i] ^= m.q[i];
  588. Salsa20 s20(keytmp,256,m.q.field(8,8));
  589. Utils::burn(keytmp,sizeof(keytmp));
  590. // One-time-use Poly1305 key from first 32 bytes of Salsa20 keystream (as per DJB/NaCl "standard")
  591. char polykey[ZT_POLY1305_KEY_LEN];
  592. memset(polykey,0,sizeof(polykey));
  593. s20.encrypt12(polykey,polykey,sizeof(polykey));
  594. // Encrypt m.q in place
  595. 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);
  596. // Add MAC for authentication (encrypt-then-MAC)
  597. char mac[ZT_POLY1305_MAC_LEN];
  598. Poly1305::compute(mac,reinterpret_cast<const char *>(m.q.data()) + 24,m.q.size() - 24,polykey);
  599. memcpy(m.q.field(16,8),mac,8);
  600. // Send!
  601. _sendFunction(_sendFunctionArg,memberId,m.q.data(),m.q.size());
  602. // Prepare for more
  603. m.q.clear();
  604. char iv[16];
  605. Utils::getSecureRandom(iv,16);
  606. m.q.append(iv,16);
  607. m.q.addSize(8); // room for MAC
  608. m.q.append((uint16_t)_id); // from member ID
  609. m.q.append((uint16_t)memberId); // to member ID
  610. }
  611. }
  612. } // namespace ZeroTier
  613. #endif // ZT_ENABLE_CLUSTER