Cluster.cpp 36 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023
  1. /*
  2. * ZeroTier One - Network Virtualization Everywhere
  3. * Copyright (C) 2011-2016 ZeroTier, Inc. https://www.zerotier.com/
  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. #ifdef ZT_ENABLE_CLUSTER
  19. #include <stdint.h>
  20. #include <stdio.h>
  21. #include <stdlib.h>
  22. #include <string.h>
  23. #include <math.h>
  24. #include <map>
  25. #include <algorithm>
  26. #include <set>
  27. #include <utility>
  28. #include <list>
  29. #include <stdexcept>
  30. #include "../version.h"
  31. #include "Cluster.hpp"
  32. #include "RuntimeEnvironment.hpp"
  33. #include "MulticastGroup.hpp"
  34. #include "CertificateOfMembership.hpp"
  35. #include "Salsa20.hpp"
  36. #include "Poly1305.hpp"
  37. #include "Identity.hpp"
  38. #include "Topology.hpp"
  39. #include "Packet.hpp"
  40. #include "Switch.hpp"
  41. #include "Node.hpp"
  42. #include "Network.hpp"
  43. #include "Array.hpp"
  44. namespace ZeroTier {
  45. static inline double _dist3d(int x1,int y1,int z1,int x2,int y2,int z2)
  46. throw()
  47. {
  48. double dx = ((double)x2 - (double)x1);
  49. double dy = ((double)y2 - (double)y1);
  50. double dz = ((double)z2 - (double)z1);
  51. return sqrt((dx * dx) + (dy * dy) + (dz * dz));
  52. }
  53. // An entry in _ClusterSendQueue
  54. struct _ClusterSendQueueEntry
  55. {
  56. uint64_t timestamp;
  57. Address fromPeerAddress;
  58. Address toPeerAddress;
  59. // if we ever support larger transport MTUs this must be increased
  60. unsigned char data[ZT_CLUSTER_SEND_QUEUE_DATA_MAX];
  61. unsigned int len;
  62. bool unite;
  63. };
  64. // A multi-index map with entry memory pooling -- this allows our queue to
  65. // be O(log(N)) and is complex enough that it makes the code a lot cleaner
  66. // to break it out from Cluster.
  67. class _ClusterSendQueue
  68. {
  69. public:
  70. _ClusterSendQueue() :
  71. _poolCount(0) {}
  72. ~_ClusterSendQueue() {} // memory is automatically freed when _chunks is destroyed
  73. inline void enqueue(uint64_t now,const Address &from,const Address &to,const void *data,unsigned int len,bool unite)
  74. {
  75. if (len > ZT_CLUSTER_SEND_QUEUE_DATA_MAX)
  76. return;
  77. Mutex::Lock _l(_lock);
  78. // Delete oldest queue entry for this sender if this enqueue() would take them over the per-sender limit
  79. {
  80. std::set< std::pair<Address,_ClusterSendQueueEntry *> >::iterator qi(_bySrc.lower_bound(std::pair<Address,_ClusterSendQueueEntry *>(from,(_ClusterSendQueueEntry *)0)));
  81. std::set< std::pair<Address,_ClusterSendQueueEntry *> >::iterator oldest(qi);
  82. unsigned long countForSender = 0;
  83. while ((qi != _bySrc.end())&&(qi->first == from)) {
  84. if (qi->second->timestamp < oldest->second->timestamp)
  85. oldest = qi;
  86. ++countForSender;
  87. ++qi;
  88. }
  89. if (countForSender >= ZT_CLUSTER_MAX_QUEUE_PER_SENDER) {
  90. _byDest.erase(std::pair<Address,_ClusterSendQueueEntry *>(oldest->second->toPeerAddress,oldest->second));
  91. _pool[_poolCount++] = oldest->second;
  92. _bySrc.erase(oldest);
  93. }
  94. }
  95. _ClusterSendQueueEntry *e;
  96. if (_poolCount > 0) {
  97. e = _pool[--_poolCount];
  98. } else {
  99. if (_chunks.size() >= ZT_CLUSTER_MAX_QUEUE_CHUNKS)
  100. return; // queue is totally full!
  101. _chunks.push_back(Array<_ClusterSendQueueEntry,ZT_CLUSTER_QUEUE_CHUNK_SIZE>());
  102. e = &(_chunks.back().data[0]);
  103. for(unsigned int i=1;i<ZT_CLUSTER_QUEUE_CHUNK_SIZE;++i)
  104. _pool[_poolCount++] = &(_chunks.back().data[i]);
  105. }
  106. e->timestamp = now;
  107. e->fromPeerAddress = from;
  108. e->toPeerAddress = to;
  109. memcpy(e->data,data,len);
  110. e->len = len;
  111. e->unite = unite;
  112. _bySrc.insert(std::pair<Address,_ClusterSendQueueEntry *>(from,e));
  113. _byDest.insert(std::pair<Address,_ClusterSendQueueEntry *>(to,e));
  114. }
  115. inline void expire(uint64_t now)
  116. {
  117. Mutex::Lock _l(_lock);
  118. for(std::set< std::pair<Address,_ClusterSendQueueEntry *> >::iterator qi(_bySrc.begin());qi!=_bySrc.end();) {
  119. if ((now - qi->second->timestamp) > ZT_CLUSTER_QUEUE_EXPIRATION) {
  120. _byDest.erase(std::pair<Address,_ClusterSendQueueEntry *>(qi->second->toPeerAddress,qi->second));
  121. _pool[_poolCount++] = qi->second;
  122. _bySrc.erase(qi++);
  123. } else ++qi;
  124. }
  125. }
  126. /**
  127. * Get and dequeue entries for a given destination address
  128. *
  129. * After use these entries must be returned with returnToPool()!
  130. *
  131. * @param dest Destination address
  132. * @param results Array to fill with results
  133. * @param maxResults Size of results[] in pointers
  134. * @return Number of actual results returned
  135. */
  136. inline unsigned int getByDest(const Address &dest,_ClusterSendQueueEntry **results,unsigned int maxResults)
  137. {
  138. unsigned int count = 0;
  139. Mutex::Lock _l(_lock);
  140. std::set< std::pair<Address,_ClusterSendQueueEntry *> >::iterator qi(_byDest.lower_bound(std::pair<Address,_ClusterSendQueueEntry *>(dest,(_ClusterSendQueueEntry *)0)));
  141. while ((qi != _byDest.end())&&(qi->first == dest)) {
  142. _bySrc.erase(std::pair<Address,_ClusterSendQueueEntry *>(qi->second->fromPeerAddress,qi->second));
  143. results[count++] = qi->second;
  144. if (count == maxResults)
  145. break;
  146. _byDest.erase(qi++);
  147. }
  148. return count;
  149. }
  150. /**
  151. * Return entries to pool after use
  152. *
  153. * @param entries Array of entries
  154. * @param count Number of entries
  155. */
  156. inline void returnToPool(_ClusterSendQueueEntry **entries,unsigned int count)
  157. {
  158. Mutex::Lock _l(_lock);
  159. for(unsigned int i=0;i<count;++i)
  160. _pool[_poolCount++] = entries[i];
  161. }
  162. private:
  163. std::list< Array<_ClusterSendQueueEntry,ZT_CLUSTER_QUEUE_CHUNK_SIZE> > _chunks;
  164. _ClusterSendQueueEntry *_pool[ZT_CLUSTER_QUEUE_CHUNK_SIZE * ZT_CLUSTER_MAX_QUEUE_CHUNKS];
  165. unsigned long _poolCount;
  166. std::set< std::pair<Address,_ClusterSendQueueEntry *> > _bySrc;
  167. std::set< std::pair<Address,_ClusterSendQueueEntry *> > _byDest;
  168. Mutex _lock;
  169. };
  170. Cluster::Cluster(
  171. const RuntimeEnvironment *renv,
  172. uint16_t id,
  173. const std::vector<InetAddress> &zeroTierPhysicalEndpoints,
  174. int32_t x,
  175. int32_t y,
  176. int32_t z,
  177. void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
  178. void *sendFunctionArg,
  179. int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
  180. void *addressToLocationFunctionArg) :
  181. RR(renv),
  182. _sendQueue(new _ClusterSendQueue()),
  183. _sendFunction(sendFunction),
  184. _sendFunctionArg(sendFunctionArg),
  185. _addressToLocationFunction(addressToLocationFunction),
  186. _addressToLocationFunctionArg(addressToLocationFunctionArg),
  187. _x(x),
  188. _y(y),
  189. _z(z),
  190. _id(id),
  191. _zeroTierPhysicalEndpoints(zeroTierPhysicalEndpoints),
  192. _members(new _Member[ZT_CLUSTER_MAX_MEMBERS]),
  193. _lastFlushed(0),
  194. _lastCleanedRemotePeers(0),
  195. _lastCleanedQueue(0)
  196. {
  197. uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
  198. // Generate master secret by hashing the secret from our Identity key pair
  199. RR->identity.sha512PrivateKey(_masterSecret);
  200. // Generate our inbound message key, which is the master secret XORed with our ID and hashed twice
  201. memcpy(stmp,_masterSecret,sizeof(stmp));
  202. stmp[0] ^= Utils::hton(id);
  203. SHA512::hash(stmp,stmp,sizeof(stmp));
  204. SHA512::hash(stmp,stmp,sizeof(stmp));
  205. memcpy(_key,stmp,sizeof(_key));
  206. Utils::burn(stmp,sizeof(stmp));
  207. }
  208. Cluster::~Cluster()
  209. {
  210. Utils::burn(_masterSecret,sizeof(_masterSecret));
  211. Utils::burn(_key,sizeof(_key));
  212. delete [] _members;
  213. delete _sendQueue;
  214. }
  215. void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
  216. {
  217. Buffer<ZT_CLUSTER_MAX_MESSAGE_LENGTH> dmsg;
  218. {
  219. // FORMAT: <[16] iv><[8] MAC><... data>
  220. if ((len < 24)||(len > ZT_CLUSTER_MAX_MESSAGE_LENGTH))
  221. return;
  222. // 16-byte IV: first 8 bytes XORed with key, last 8 bytes used as Salsa20 64-bit IV
  223. char keytmp[32];
  224. memcpy(keytmp,_key,32);
  225. for(int i=0;i<8;++i)
  226. keytmp[i] ^= reinterpret_cast<const char *>(msg)[i];
  227. Salsa20 s20(keytmp,256,reinterpret_cast<const char *>(msg) + 8);
  228. Utils::burn(keytmp,sizeof(keytmp));
  229. // One-time-use Poly1305 key from first 32 bytes of Salsa20 keystream (as per DJB/NaCl "standard")
  230. char polykey[ZT_POLY1305_KEY_LEN];
  231. memset(polykey,0,sizeof(polykey));
  232. s20.encrypt12(polykey,polykey,sizeof(polykey));
  233. // Compute 16-byte MAC
  234. char mac[ZT_POLY1305_MAC_LEN];
  235. Poly1305::compute(mac,reinterpret_cast<const char *>(msg) + 24,len - 24,polykey);
  236. // Check first 8 bytes of MAC against 64-bit MAC in stream
  237. if (!Utils::secureEq(mac,reinterpret_cast<const char *>(msg) + 16,8))
  238. return;
  239. // Decrypt!
  240. dmsg.setSize(len - 24);
  241. s20.decrypt12(reinterpret_cast<const char *>(msg) + 24,const_cast<void *>(dmsg.data()),dmsg.size());
  242. }
  243. if (dmsg.size() < 4)
  244. return;
  245. const uint16_t fromMemberId = dmsg.at<uint16_t>(0);
  246. unsigned int ptr = 2;
  247. if (fromMemberId == _id) // sanity check: we don't talk to ourselves
  248. return;
  249. const uint16_t toMemberId = dmsg.at<uint16_t>(ptr);
  250. ptr += 2;
  251. if (toMemberId != _id) // sanity check: message not for us?
  252. return;
  253. { // make sure sender is actually considered a member
  254. Mutex::Lock _l3(_memberIds_m);
  255. if (std::find(_memberIds.begin(),_memberIds.end(),fromMemberId) == _memberIds.end())
  256. return;
  257. }
  258. try {
  259. while (ptr < dmsg.size()) {
  260. const unsigned int mlen = dmsg.at<uint16_t>(ptr); ptr += 2;
  261. const unsigned int nextPtr = ptr + mlen;
  262. if (nextPtr > dmsg.size())
  263. break;
  264. int mtype = -1;
  265. try {
  266. switch((StateMessageType)(mtype = (int)dmsg[ptr++])) {
  267. default:
  268. break;
  269. case CLUSTER_MESSAGE_ALIVE: {
  270. _Member &m = _members[fromMemberId];
  271. Mutex::Lock mlck(m.lock);
  272. ptr += 7; // skip version stuff, not used yet
  273. m.x = dmsg.at<int32_t>(ptr); ptr += 4;
  274. m.y = dmsg.at<int32_t>(ptr); ptr += 4;
  275. m.z = dmsg.at<int32_t>(ptr); ptr += 4;
  276. ptr += 8; // skip local clock, not used
  277. m.load = dmsg.at<uint64_t>(ptr); ptr += 8;
  278. m.peers = dmsg.at<uint64_t>(ptr); ptr += 8;
  279. ptr += 8; // skip flags, unused
  280. #ifdef ZT_TRACE
  281. std::string addrs;
  282. #endif
  283. unsigned int physicalAddressCount = dmsg[ptr++];
  284. m.zeroTierPhysicalEndpoints.clear();
  285. for(unsigned int i=0;i<physicalAddressCount;++i) {
  286. m.zeroTierPhysicalEndpoints.push_back(InetAddress());
  287. ptr += m.zeroTierPhysicalEndpoints.back().deserialize(dmsg,ptr);
  288. if (!(m.zeroTierPhysicalEndpoints.back())) {
  289. m.zeroTierPhysicalEndpoints.pop_back();
  290. }
  291. #ifdef ZT_TRACE
  292. else {
  293. if (addrs.length() > 0)
  294. addrs.push_back(',');
  295. addrs.append(m.zeroTierPhysicalEndpoints.back().toString());
  296. }
  297. #endif
  298. }
  299. #ifdef ZT_TRACE
  300. if ((RR->node->now() - m.lastReceivedAliveAnnouncement) >= ZT_CLUSTER_TIMEOUT) {
  301. 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());
  302. }
  303. #endif
  304. m.lastReceivedAliveAnnouncement = RR->node->now();
  305. } break;
  306. case CLUSTER_MESSAGE_HAVE_PEER: {
  307. Identity id;
  308. ptr += id.deserialize(dmsg,ptr);
  309. if (id) {
  310. {
  311. Mutex::Lock _l(_remotePeers_m);
  312. _RemotePeer &rp = _remotePeers[std::pair<Address,unsigned int>(id.address(),(unsigned int)fromMemberId)];
  313. if (!rp.lastHavePeerReceived) {
  314. RR->topology->saveIdentity(id);
  315. id.agree(RR->identity,rp.key,ZT_PEER_SECRET_KEY_LENGTH);
  316. }
  317. rp.lastHavePeerReceived = RR->node->now();
  318. }
  319. _ClusterSendQueueEntry *q[16384]; // 16384 is "tons"
  320. unsigned int qc = _sendQueue->getByDest(id.address(),q,16384);
  321. for(unsigned int i=0;i<qc;++i)
  322. this->relayViaCluster(q[i]->fromPeerAddress,q[i]->toPeerAddress,q[i]->data,q[i]->len,q[i]->unite);
  323. _sendQueue->returnToPool(q,qc);
  324. TRACE("[%u] has %s (retried %u queued sends)",(unsigned int)fromMemberId,id.address().toString().c_str(),qc);
  325. }
  326. } break;
  327. case CLUSTER_MESSAGE_WANT_PEER: {
  328. const Address zeroTierAddress(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  329. SharedPtr<Peer> peer(RR->topology->getPeerNoCache(zeroTierAddress));
  330. if ( (peer) && (peer->hasLocalClusterOptimalPath(RR->node->now())) ) {
  331. Buffer<1024> buf;
  332. peer->identity().serialize(buf);
  333. Mutex::Lock _l2(_members[fromMemberId].lock);
  334. _send(fromMemberId,CLUSTER_MESSAGE_HAVE_PEER,buf.data(),buf.size());
  335. }
  336. } break;
  337. case CLUSTER_MESSAGE_REMOTE_PACKET: {
  338. const unsigned int plen = dmsg.at<uint16_t>(ptr); ptr += 2;
  339. if (plen) {
  340. Packet remotep(dmsg.field(ptr,plen),plen); ptr += plen;
  341. //TRACE("remote %s from %s via %u (%u bytes)",Packet::verbString(remotep.verb()),remotep.source().toString().c_str(),fromMemberId,plen);
  342. switch(remotep.verb()) {
  343. case Packet::VERB_WHOIS: _doREMOTE_WHOIS(fromMemberId,remotep); break;
  344. case Packet::VERB_MULTICAST_GATHER: _doREMOTE_MULTICAST_GATHER(fromMemberId,remotep); break;
  345. default: break; // ignore things we don't care about across cluster
  346. }
  347. }
  348. } break;
  349. case CLUSTER_MESSAGE_PROXY_UNITE: {
  350. const Address localPeerAddress(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  351. const Address remotePeerAddress(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  352. const unsigned int numRemotePeerPaths = dmsg[ptr++];
  353. InetAddress remotePeerPaths[256]; // size is 8-bit, so 256 is max
  354. for(unsigned int i=0;i<numRemotePeerPaths;++i)
  355. ptr += remotePeerPaths[i].deserialize(dmsg,ptr);
  356. TRACE("[%u] requested that we unite local %s with remote %s",(unsigned int)fromMemberId,localPeerAddress.toString().c_str(),remotePeerAddress.toString().c_str());
  357. const uint64_t now = RR->node->now();
  358. SharedPtr<Peer> localPeer(RR->topology->getPeerNoCache(localPeerAddress));
  359. if ((localPeer)&&(numRemotePeerPaths > 0)) {
  360. InetAddress bestLocalV4,bestLocalV6;
  361. localPeer->getBestActiveAddresses(now,bestLocalV4,bestLocalV6);
  362. InetAddress bestRemoteV4,bestRemoteV6;
  363. for(unsigned int i=0;i<numRemotePeerPaths;++i) {
  364. if ((bestRemoteV4)&&(bestRemoteV6))
  365. break;
  366. switch(remotePeerPaths[i].ss_family) {
  367. case AF_INET:
  368. if (!bestRemoteV4)
  369. bestRemoteV4 = remotePeerPaths[i];
  370. break;
  371. case AF_INET6:
  372. if (!bestRemoteV6)
  373. bestRemoteV6 = remotePeerPaths[i];
  374. break;
  375. }
  376. }
  377. Packet rendezvousForLocal(localPeerAddress,RR->identity.address(),Packet::VERB_RENDEZVOUS);
  378. rendezvousForLocal.append((uint8_t)0);
  379. remotePeerAddress.appendTo(rendezvousForLocal);
  380. Buffer<2048> rendezvousForRemote;
  381. remotePeerAddress.appendTo(rendezvousForRemote);
  382. rendezvousForRemote.append((uint8_t)Packet::VERB_RENDEZVOUS);
  383. rendezvousForRemote.addSize(2); // space for actual packet payload length
  384. rendezvousForRemote.append((uint8_t)0); // flags == 0
  385. localPeerAddress.appendTo(rendezvousForRemote);
  386. bool haveMatch = false;
  387. if ((bestLocalV6)&&(bestRemoteV6)) {
  388. haveMatch = true;
  389. rendezvousForLocal.append((uint16_t)bestRemoteV6.port());
  390. rendezvousForLocal.append((uint8_t)16);
  391. rendezvousForLocal.append(bestRemoteV6.rawIpData(),16);
  392. rendezvousForRemote.append((uint16_t)bestLocalV6.port());
  393. rendezvousForRemote.append((uint8_t)16);
  394. rendezvousForRemote.append(bestLocalV6.rawIpData(),16);
  395. rendezvousForRemote.setAt<uint16_t>(ZT_ADDRESS_LENGTH + 1,(uint16_t)(9 + 16));
  396. } else if ((bestLocalV4)&&(bestRemoteV4)) {
  397. haveMatch = true;
  398. rendezvousForLocal.append((uint16_t)bestRemoteV4.port());
  399. rendezvousForLocal.append((uint8_t)4);
  400. rendezvousForLocal.append(bestRemoteV4.rawIpData(),4);
  401. rendezvousForRemote.append((uint16_t)bestLocalV4.port());
  402. rendezvousForRemote.append((uint8_t)4);
  403. rendezvousForRemote.append(bestLocalV4.rawIpData(),4);
  404. rendezvousForRemote.setAt<uint16_t>(ZT_ADDRESS_LENGTH + 1,(uint16_t)(9 + 4));
  405. }
  406. if (haveMatch) {
  407. {
  408. Mutex::Lock _l2(_members[fromMemberId].lock);
  409. _send(fromMemberId,CLUSTER_MESSAGE_PROXY_SEND,rendezvousForRemote.data(),rendezvousForRemote.size());
  410. }
  411. RR->sw->send(rendezvousForLocal,true);
  412. }
  413. }
  414. } break;
  415. case CLUSTER_MESSAGE_PROXY_SEND: {
  416. const Address rcpt(dmsg.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH); ptr += ZT_ADDRESS_LENGTH;
  417. const Packet::Verb verb = (Packet::Verb)dmsg[ptr++];
  418. const unsigned int len = dmsg.at<uint16_t>(ptr); ptr += 2;
  419. Packet outp(rcpt,RR->identity.address(),verb);
  420. outp.append(dmsg.field(ptr,len),len); ptr += len;
  421. RR->sw->send(outp,true);
  422. //TRACE("[%u] proxy send %s to %s length %u",(unsigned int)fromMemberId,Packet::verbString(verb),rcpt.toString().c_str(),len);
  423. } break;
  424. case CLUSTER_MESSAGE_NETWORK_CONFIG: {
  425. const SharedPtr<Network> network(RR->node->network(dmsg.at<uint64_t>(ptr)));
  426. if (network) {
  427. // Copy into a Packet just to conform to Network API. Eventually
  428. // will want to refactor.
  429. network->handleConfigChunk(0,Address(),Buffer<ZT_PROTO_MAX_PACKET_LENGTH>(dmsg),ptr);
  430. }
  431. } break;
  432. }
  433. } catch ( ... ) {
  434. TRACE("invalid message of size %u type %d (inner decode), discarding",mlen,mtype);
  435. // drop invalids
  436. }
  437. ptr = nextPtr;
  438. }
  439. } catch ( ... ) {
  440. TRACE("invalid message (outer loop), discarding");
  441. // drop invalids
  442. }
  443. }
  444. void Cluster::broadcastHavePeer(const Identity &id)
  445. {
  446. Buffer<1024> buf;
  447. id.serialize(buf);
  448. Mutex::Lock _l(_memberIds_m);
  449. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  450. Mutex::Lock _l2(_members[*mid].lock);
  451. _send(*mid,CLUSTER_MESSAGE_HAVE_PEER,buf.data(),buf.size());
  452. }
  453. }
  454. void Cluster::broadcastNetworkConfigChunk(const void *chunk,unsigned int len)
  455. {
  456. Mutex::Lock _l(_memberIds_m);
  457. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  458. Mutex::Lock _l2(_members[*mid].lock);
  459. _send(*mid,CLUSTER_MESSAGE_NETWORK_CONFIG,chunk,len);
  460. }
  461. }
  462. int Cluster::prepSendViaCluster(const Address &toPeerAddress,void *peerSecret)
  463. {
  464. const uint64_t now = RR->node->now();
  465. uint64_t mostRecentTs = 0;
  466. int mostRecentMemberId = -1;
  467. {
  468. Mutex::Lock _l2(_remotePeers_m);
  469. std::map< std::pair<Address,unsigned int>,_RemotePeer >::const_iterator rpe(_remotePeers.lower_bound(std::pair<Address,unsigned int>(toPeerAddress,0)));
  470. for(;;) {
  471. if ((rpe == _remotePeers.end())||(rpe->first.first != toPeerAddress))
  472. break;
  473. else if (rpe->second.lastHavePeerReceived > mostRecentTs) {
  474. mostRecentTs = rpe->second.lastHavePeerReceived;
  475. memcpy(peerSecret,rpe->second.key,ZT_PEER_SECRET_KEY_LENGTH);
  476. mostRecentMemberId = (int)rpe->first.second;
  477. }
  478. ++rpe;
  479. }
  480. }
  481. if (mostRecentMemberId >= 0) {
  482. const uint64_t ageOfMostRecentHavePeerAnnouncement = now - mostRecentTs;
  483. if (ageOfMostRecentHavePeerAnnouncement >= (ZT_PEER_ACTIVITY_TIMEOUT / 3)) {
  484. if (ageOfMostRecentHavePeerAnnouncement >= ZT_PEER_ACTIVITY_TIMEOUT)
  485. return -1;
  486. bool sendWantPeer = true;
  487. {
  488. Mutex::Lock _l(_remotePeers_m);
  489. _RemotePeer &rp = _remotePeers[std::pair<Address,unsigned int>(toPeerAddress,(unsigned int)_id)];
  490. if ((now - rp.lastSentWantPeer) >= ZT_CLUSTER_WANT_PEER_EVERY) {
  491. rp.lastSentWantPeer = now;
  492. } else {
  493. sendWantPeer = false; // don't flood WANT_PEER
  494. }
  495. }
  496. if (sendWantPeer) {
  497. char tmp[ZT_ADDRESS_LENGTH];
  498. toPeerAddress.copyTo(tmp,ZT_ADDRESS_LENGTH);
  499. {
  500. Mutex::Lock _l(_memberIds_m);
  501. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  502. Mutex::Lock _l2(_members[*mid].lock);
  503. _send(*mid,CLUSTER_MESSAGE_WANT_PEER,tmp,ZT_ADDRESS_LENGTH);
  504. }
  505. }
  506. }
  507. }
  508. return mostRecentMemberId;
  509. } else return -1;
  510. }
  511. bool Cluster::sendViaCluster(int mostRecentMemberId,const Address &toPeerAddress,const void *data,unsigned int len)
  512. {
  513. if ((mostRecentMemberId < 0)||(mostRecentMemberId >= ZT_CLUSTER_MAX_MEMBERS)) // sanity check
  514. return false;
  515. Mutex::Lock _l2(_members[mostRecentMemberId].lock);
  516. for(std::vector<InetAddress>::const_iterator i1(_zeroTierPhysicalEndpoints.begin());i1!=_zeroTierPhysicalEndpoints.end();++i1) {
  517. for(std::vector<InetAddress>::const_iterator i2(_members[mostRecentMemberId].zeroTierPhysicalEndpoints.begin());i2!=_members[mostRecentMemberId].zeroTierPhysicalEndpoints.end();++i2) {
  518. if (i1->ss_family == i2->ss_family) {
  519. TRACE("sendViaCluster sending %u bytes to %s by way of %u (%s->%s)",len,toPeerAddress.toString().c_str(),(unsigned int)mostRecentMemberId,i1->toString().c_str(),i2->toString().c_str());
  520. RR->node->putPacket(*i1,*i2,data,len);
  521. return true;
  522. }
  523. }
  524. }
  525. return false;
  526. }
  527. void Cluster::relayViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len,bool unite)
  528. {
  529. if (len > ZT_PROTO_MAX_PACKET_LENGTH) // sanity check
  530. return;
  531. const uint64_t now = RR->node->now();
  532. uint64_t mostRecentTs = 0;
  533. int mostRecentMemberId = -1;
  534. {
  535. Mutex::Lock _l2(_remotePeers_m);
  536. std::map< std::pair<Address,unsigned int>,_RemotePeer >::const_iterator rpe(_remotePeers.lower_bound(std::pair<Address,unsigned int>(toPeerAddress,0)));
  537. for(;;) {
  538. if ((rpe == _remotePeers.end())||(rpe->first.first != toPeerAddress))
  539. break;
  540. else if (rpe->second.lastHavePeerReceived > mostRecentTs) {
  541. mostRecentTs = rpe->second.lastHavePeerReceived;
  542. mostRecentMemberId = (int)rpe->first.second;
  543. }
  544. ++rpe;
  545. }
  546. }
  547. const uint64_t ageOfMostRecentHavePeerAnnouncement = now - mostRecentTs;
  548. if (ageOfMostRecentHavePeerAnnouncement >= (ZT_PEER_ACTIVITY_TIMEOUT / 3)) {
  549. // Enqueue and wait if peer seems alive, but do WANT_PEER to refresh homing
  550. const bool enqueueAndWait = ((ageOfMostRecentHavePeerAnnouncement >= ZT_PEER_ACTIVITY_TIMEOUT)||(mostRecentMemberId < 0));
  551. // Poll everyone with WANT_PEER if the age of our most recent entry is
  552. // approaching expiration (or has expired, or does not exist).
  553. bool sendWantPeer = true;
  554. {
  555. Mutex::Lock _l(_remotePeers_m);
  556. _RemotePeer &rp = _remotePeers[std::pair<Address,unsigned int>(toPeerAddress,(unsigned int)_id)];
  557. if ((now - rp.lastSentWantPeer) >= ZT_CLUSTER_WANT_PEER_EVERY) {
  558. rp.lastSentWantPeer = now;
  559. } else {
  560. sendWantPeer = false; // don't flood WANT_PEER
  561. }
  562. }
  563. if (sendWantPeer) {
  564. char tmp[ZT_ADDRESS_LENGTH];
  565. toPeerAddress.copyTo(tmp,ZT_ADDRESS_LENGTH);
  566. {
  567. Mutex::Lock _l(_memberIds_m);
  568. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  569. Mutex::Lock _l2(_members[*mid].lock);
  570. _send(*mid,CLUSTER_MESSAGE_WANT_PEER,tmp,ZT_ADDRESS_LENGTH);
  571. }
  572. }
  573. }
  574. // If there isn't a good place to send via, then enqueue this for retrying
  575. // later and return after having broadcasted a WANT_PEER.
  576. if (enqueueAndWait) {
  577. TRACE("relayViaCluster %s -> %s enqueueing to wait for HAVE_PEER",fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str());
  578. _sendQueue->enqueue(now,fromPeerAddress,toPeerAddress,data,len,unite);
  579. return;
  580. }
  581. }
  582. if (mostRecentMemberId >= 0) {
  583. Buffer<1024> buf;
  584. if (unite) {
  585. InetAddress v4,v6;
  586. if (fromPeerAddress) {
  587. SharedPtr<Peer> fromPeer(RR->topology->getPeerNoCache(fromPeerAddress));
  588. if (fromPeer)
  589. fromPeer->getBestActiveAddresses(now,v4,v6);
  590. }
  591. uint8_t addrCount = 0;
  592. if (v4)
  593. ++addrCount;
  594. if (v6)
  595. ++addrCount;
  596. if (addrCount) {
  597. toPeerAddress.appendTo(buf);
  598. fromPeerAddress.appendTo(buf);
  599. buf.append(addrCount);
  600. if (v4)
  601. v4.serialize(buf);
  602. if (v6)
  603. v6.serialize(buf);
  604. }
  605. }
  606. {
  607. Mutex::Lock _l2(_members[mostRecentMemberId].lock);
  608. if (buf.size() > 0)
  609. _send(mostRecentMemberId,CLUSTER_MESSAGE_PROXY_UNITE,buf.data(),buf.size());
  610. for(std::vector<InetAddress>::const_iterator i1(_zeroTierPhysicalEndpoints.begin());i1!=_zeroTierPhysicalEndpoints.end();++i1) {
  611. for(std::vector<InetAddress>::const_iterator i2(_members[mostRecentMemberId].zeroTierPhysicalEndpoints.begin());i2!=_members[mostRecentMemberId].zeroTierPhysicalEndpoints.end();++i2) {
  612. if (i1->ss_family == i2->ss_family) {
  613. TRACE("relayViaCluster relaying %u bytes from %s to %s by way of %u (%s->%s)",len,fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str(),(unsigned int)mostRecentMemberId,i1->toString().c_str(),i2->toString().c_str());
  614. RR->node->putPacket(*i1,*i2,data,len);
  615. return;
  616. }
  617. }
  618. }
  619. TRACE("relayViaCluster relaying %u bytes from %s to %s by way of %u failed: no common endpoints with the same address family!",len,fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str(),(unsigned int)mostRecentMemberId);
  620. }
  621. }
  622. }
  623. void Cluster::sendDistributedQuery(const Packet &pkt)
  624. {
  625. Buffer<4096> buf;
  626. buf.append((uint16_t)pkt.size());
  627. buf.append(pkt.data(),pkt.size());
  628. Mutex::Lock _l(_memberIds_m);
  629. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  630. Mutex::Lock _l2(_members[*mid].lock);
  631. _send(*mid,CLUSTER_MESSAGE_REMOTE_PACKET,buf.data(),buf.size());
  632. }
  633. }
  634. void Cluster::doPeriodicTasks()
  635. {
  636. const uint64_t now = RR->node->now();
  637. if ((now - _lastFlushed) >= ZT_CLUSTER_FLUSH_PERIOD) {
  638. _lastFlushed = now;
  639. Mutex::Lock _l(_memberIds_m);
  640. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  641. Mutex::Lock _l2(_members[*mid].lock);
  642. if ((now - _members[*mid].lastAnnouncedAliveTo) >= ((ZT_CLUSTER_TIMEOUT / 2) - 1000)) {
  643. _members[*mid].lastAnnouncedAliveTo = now;
  644. Buffer<2048> alive;
  645. alive.append((uint16_t)ZEROTIER_ONE_VERSION_MAJOR);
  646. alive.append((uint16_t)ZEROTIER_ONE_VERSION_MINOR);
  647. alive.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
  648. alive.append((uint8_t)ZT_PROTO_VERSION);
  649. if (_addressToLocationFunction) {
  650. alive.append((int32_t)_x);
  651. alive.append((int32_t)_y);
  652. alive.append((int32_t)_z);
  653. } else {
  654. alive.append((int32_t)0);
  655. alive.append((int32_t)0);
  656. alive.append((int32_t)0);
  657. }
  658. alive.append((uint64_t)now);
  659. alive.append((uint64_t)0); // TODO: compute and send load average
  660. alive.append((uint64_t)RR->topology->countActive(now));
  661. alive.append((uint64_t)0); // unused/reserved flags
  662. alive.append((uint8_t)_zeroTierPhysicalEndpoints.size());
  663. for(std::vector<InetAddress>::const_iterator pe(_zeroTierPhysicalEndpoints.begin());pe!=_zeroTierPhysicalEndpoints.end();++pe)
  664. pe->serialize(alive);
  665. _send(*mid,CLUSTER_MESSAGE_ALIVE,alive.data(),alive.size());
  666. }
  667. _flush(*mid);
  668. }
  669. }
  670. if ((now - _lastCleanedRemotePeers) >= (ZT_PEER_ACTIVITY_TIMEOUT * 2)) {
  671. _lastCleanedRemotePeers = now;
  672. Mutex::Lock _l(_remotePeers_m);
  673. for(std::map< std::pair<Address,unsigned int>,_RemotePeer >::iterator rp(_remotePeers.begin());rp!=_remotePeers.end();) {
  674. if ((now - rp->second.lastHavePeerReceived) >= ZT_PEER_ACTIVITY_TIMEOUT)
  675. _remotePeers.erase(rp++);
  676. else ++rp;
  677. }
  678. }
  679. if ((now - _lastCleanedQueue) >= ZT_CLUSTER_QUEUE_EXPIRATION) {
  680. _lastCleanedQueue = now;
  681. _sendQueue->expire(now);
  682. }
  683. }
  684. void Cluster::addMember(uint16_t memberId)
  685. {
  686. if ((memberId >= ZT_CLUSTER_MAX_MEMBERS)||(memberId == _id))
  687. return;
  688. Mutex::Lock _l2(_members[memberId].lock);
  689. {
  690. Mutex::Lock _l(_memberIds_m);
  691. if (std::find(_memberIds.begin(),_memberIds.end(),memberId) != _memberIds.end())
  692. return;
  693. _memberIds.push_back(memberId);
  694. std::sort(_memberIds.begin(),_memberIds.end());
  695. }
  696. _members[memberId].clear();
  697. // Generate this member's message key from the master and its ID
  698. uint16_t stmp[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)];
  699. memcpy(stmp,_masterSecret,sizeof(stmp));
  700. stmp[0] ^= Utils::hton(memberId);
  701. SHA512::hash(stmp,stmp,sizeof(stmp));
  702. SHA512::hash(stmp,stmp,sizeof(stmp));
  703. memcpy(_members[memberId].key,stmp,sizeof(_members[memberId].key));
  704. Utils::burn(stmp,sizeof(stmp));
  705. // Prepare q
  706. _members[memberId].q.clear();
  707. char iv[16];
  708. Utils::getSecureRandom(iv,16);
  709. _members[memberId].q.append(iv,16);
  710. _members[memberId].q.addSize(8); // room for MAC
  711. _members[memberId].q.append((uint16_t)_id);
  712. _members[memberId].q.append((uint16_t)memberId);
  713. }
  714. void Cluster::removeMember(uint16_t memberId)
  715. {
  716. Mutex::Lock _l(_memberIds_m);
  717. std::vector<uint16_t> newMemberIds;
  718. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  719. if (*mid != memberId)
  720. newMemberIds.push_back(*mid);
  721. }
  722. _memberIds = newMemberIds;
  723. }
  724. bool Cluster::findBetterEndpoint(InetAddress &redirectTo,const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload)
  725. {
  726. if (_addressToLocationFunction) {
  727. // Pick based on location if it can be determined
  728. int px = 0,py = 0,pz = 0;
  729. if (_addressToLocationFunction(_addressToLocationFunctionArg,reinterpret_cast<const struct sockaddr_storage *>(&peerPhysicalAddress),&px,&py,&pz) == 0) {
  730. TRACE("no geolocation data for %s",peerPhysicalAddress.toIpString().c_str());
  731. return false;
  732. }
  733. // Find member closest to this peer
  734. const uint64_t now = RR->node->now();
  735. std::vector<InetAddress> best;
  736. const double currentDistance = _dist3d(_x,_y,_z,px,py,pz);
  737. double bestDistance = (offload ? 2147483648.0 : currentDistance);
  738. #ifdef ZT_TRACE
  739. unsigned int bestMember = _id;
  740. #endif
  741. {
  742. Mutex::Lock _l(_memberIds_m);
  743. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  744. _Member &m = _members[*mid];
  745. Mutex::Lock _ml(m.lock);
  746. // Consider member if it's alive and has sent us a location and one or more physical endpoints to send peers to
  747. if ( ((now - m.lastReceivedAliveAnnouncement) < ZT_CLUSTER_TIMEOUT) && ((m.x != 0)||(m.y != 0)||(m.z != 0)) && (m.zeroTierPhysicalEndpoints.size() > 0) ) {
  748. const double mdist = _dist3d(m.x,m.y,m.z,px,py,pz);
  749. if (mdist < bestDistance) {
  750. bestDistance = mdist;
  751. #ifdef ZT_TRACE
  752. bestMember = *mid;
  753. #endif
  754. best = m.zeroTierPhysicalEndpoints;
  755. }
  756. }
  757. }
  758. }
  759. // Redirect to a closer member if it has a ZeroTier endpoint address in the same ss_family
  760. for(std::vector<InetAddress>::const_iterator a(best.begin());a!=best.end();++a) {
  761. if (a->ss_family == peerPhysicalAddress.ss_family) {
  762. 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());
  763. redirectTo = *a;
  764. return true;
  765. }
  766. }
  767. TRACE("%s at [%d,%d,%d] is %f from us, no better endpoints found",peerAddress.toString().c_str(),px,py,pz,currentDistance);
  768. return false;
  769. } else {
  770. // TODO: pick based on load if no location info?
  771. return false;
  772. }
  773. }
  774. void Cluster::status(ZT_ClusterStatus &status) const
  775. {
  776. const uint64_t now = RR->node->now();
  777. memset(&status,0,sizeof(ZT_ClusterStatus));
  778. status.myId = _id;
  779. {
  780. ZT_ClusterMemberStatus *const s = &(status.members[status.clusterSize++]);
  781. s->id = _id;
  782. s->alive = 1;
  783. s->x = _x;
  784. s->y = _y;
  785. s->z = _z;
  786. s->load = 0; // TODO
  787. s->peers = RR->topology->countActive(now);
  788. for(std::vector<InetAddress>::const_iterator ep(_zeroTierPhysicalEndpoints.begin());ep!=_zeroTierPhysicalEndpoints.end();++ep) {
  789. if (s->numZeroTierPhysicalEndpoints >= ZT_CLUSTER_MAX_ZT_PHYSICAL_ADDRESSES) // sanity check
  790. break;
  791. memcpy(&(s->zeroTierPhysicalEndpoints[s->numZeroTierPhysicalEndpoints++]),&(*ep),sizeof(struct sockaddr_storage));
  792. }
  793. }
  794. {
  795. Mutex::Lock _l1(_memberIds_m);
  796. for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
  797. if (status.clusterSize >= ZT_CLUSTER_MAX_MEMBERS) // sanity check
  798. break;
  799. _Member &m = _members[*mid];
  800. Mutex::Lock ml(m.lock);
  801. ZT_ClusterMemberStatus *const s = &(status.members[status.clusterSize++]);
  802. s->id = *mid;
  803. s->msSinceLastHeartbeat = (unsigned int)std::min((uint64_t)(~((unsigned int)0)),(now - m.lastReceivedAliveAnnouncement));
  804. s->alive = (s->msSinceLastHeartbeat < ZT_CLUSTER_TIMEOUT) ? 1 : 0;
  805. s->x = m.x;
  806. s->y = m.y;
  807. s->z = m.z;
  808. s->load = m.load;
  809. s->peers = m.peers;
  810. for(std::vector<InetAddress>::const_iterator ep(m.zeroTierPhysicalEndpoints.begin());ep!=m.zeroTierPhysicalEndpoints.end();++ep) {
  811. if (s->numZeroTierPhysicalEndpoints >= ZT_CLUSTER_MAX_ZT_PHYSICAL_ADDRESSES) // sanity check
  812. break;
  813. memcpy(&(s->zeroTierPhysicalEndpoints[s->numZeroTierPhysicalEndpoints++]),&(*ep),sizeof(struct sockaddr_storage));
  814. }
  815. }
  816. }
  817. }
  818. void Cluster::_send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len)
  819. {
  820. if ((len + 3) > (ZT_CLUSTER_MAX_MESSAGE_LENGTH - (24 + 2 + 2))) // sanity check
  821. return;
  822. _Member &m = _members[memberId];
  823. // assumes m.lock is locked!
  824. if ((m.q.size() + len + 3) > ZT_CLUSTER_MAX_MESSAGE_LENGTH)
  825. _flush(memberId);
  826. m.q.append((uint16_t)(len + 1));
  827. m.q.append((uint8_t)type);
  828. m.q.append(msg,len);
  829. }
  830. void Cluster::_flush(uint16_t memberId)
  831. {
  832. _Member &m = _members[memberId];
  833. // assumes m.lock is locked!
  834. if (m.q.size() > (24 + 2 + 2)) { // 16-byte IV + 8-byte MAC + 2 byte from-member-ID + 2 byte to-member-ID
  835. // Create key from member's key and IV
  836. char keytmp[32];
  837. memcpy(keytmp,m.key,32);
  838. for(int i=0;i<8;++i)
  839. keytmp[i] ^= m.q[i];
  840. Salsa20 s20(keytmp,256,m.q.field(8,8));
  841. Utils::burn(keytmp,sizeof(keytmp));
  842. // One-time-use Poly1305 key from first 32 bytes of Salsa20 keystream (as per DJB/NaCl "standard")
  843. char polykey[ZT_POLY1305_KEY_LEN];
  844. memset(polykey,0,sizeof(polykey));
  845. s20.encrypt12(polykey,polykey,sizeof(polykey));
  846. // Encrypt m.q in place
  847. 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);
  848. // Add MAC for authentication (encrypt-then-MAC)
  849. char mac[ZT_POLY1305_MAC_LEN];
  850. Poly1305::compute(mac,reinterpret_cast<const char *>(m.q.data()) + 24,m.q.size() - 24,polykey);
  851. memcpy(m.q.field(16,8),mac,8);
  852. // Send!
  853. _sendFunction(_sendFunctionArg,memberId,m.q.data(),m.q.size());
  854. // Prepare for more
  855. m.q.clear();
  856. char iv[16];
  857. Utils::getSecureRandom(iv,16);
  858. m.q.append(iv,16);
  859. m.q.addSize(8); // room for MAC
  860. m.q.append((uint16_t)_id); // from member ID
  861. m.q.append((uint16_t)memberId); // to member ID
  862. }
  863. }
  864. void Cluster::_doREMOTE_WHOIS(uint64_t fromMemberId,const Packet &remotep)
  865. {
  866. if (remotep.payloadLength() >= ZT_ADDRESS_LENGTH) {
  867. Identity queried(RR->topology->getIdentity(Address(remotep.payload(),ZT_ADDRESS_LENGTH)));
  868. if (queried) {
  869. Buffer<1024> routp;
  870. remotep.source().appendTo(routp);
  871. routp.append((uint8_t)Packet::VERB_OK);
  872. routp.addSize(2); // space for length
  873. routp.append((uint8_t)Packet::VERB_WHOIS);
  874. routp.append(remotep.packetId());
  875. queried.serialize(routp);
  876. routp.setAt<uint16_t>(ZT_ADDRESS_LENGTH + 1,(uint16_t)(routp.size() - ZT_ADDRESS_LENGTH - 3));
  877. TRACE("responding to remote WHOIS from %s @ %u with identity of %s",remotep.source().toString().c_str(),(unsigned int)fromMemberId,queried.address().toString().c_str());
  878. Mutex::Lock _l2(_members[fromMemberId].lock);
  879. _send(fromMemberId,CLUSTER_MESSAGE_PROXY_SEND,routp.data(),routp.size());
  880. }
  881. }
  882. }
  883. void Cluster::_doREMOTE_MULTICAST_GATHER(uint64_t fromMemberId,const Packet &remotep)
  884. {
  885. const uint64_t nwid = remotep.at<uint64_t>(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_NETWORK_ID);
  886. const MulticastGroup mg(MAC(remotep.field(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_MAC,6),6),remotep.at<uint32_t>(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_ADI));
  887. unsigned int gatherLimit = remotep.at<uint32_t>(ZT_PROTO_VERB_MULTICAST_GATHER_IDX_GATHER_LIMIT);
  888. const Address remotePeerAddress(remotep.source());
  889. if (gatherLimit) {
  890. Buffer<ZT_PROTO_MAX_PACKET_LENGTH> routp;
  891. remotePeerAddress.appendTo(routp);
  892. routp.append((uint8_t)Packet::VERB_OK);
  893. routp.addSize(2); // space for length
  894. routp.append((uint8_t)Packet::VERB_MULTICAST_GATHER);
  895. routp.append(remotep.packetId());
  896. routp.append(nwid);
  897. mg.mac().appendTo(routp);
  898. routp.append((uint32_t)mg.adi());
  899. if (gatherLimit > ((ZT_CLUSTER_MAX_MESSAGE_LENGTH - 80) / 5))
  900. gatherLimit = ((ZT_CLUSTER_MAX_MESSAGE_LENGTH - 80) / 5);
  901. if (RR->mc->gather(remotePeerAddress,nwid,mg,routp,gatherLimit)) {
  902. routp.setAt<uint16_t>(ZT_ADDRESS_LENGTH + 1,(uint16_t)(routp.size() - ZT_ADDRESS_LENGTH - 3));
  903. TRACE("responding to remote MULTICAST_GATHER from %s @ %u with %u bytes",remotePeerAddress.toString().c_str(),(unsigned int)fromMemberId,routp.size());
  904. Mutex::Lock _l2(_members[fromMemberId].lock);
  905. _send(fromMemberId,CLUSTER_MESSAGE_PROXY_SEND,routp.data(),routp.size());
  906. }
  907. }
  908. }
  909. } // namespace ZeroTier
  910. #endif // ZT_ENABLE_CLUSTER