DBMirrorSet.cpp 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294
  1. /*
  2. * Copyright (c)2019 ZeroTier, Inc.
  3. *
  4. * Use of this software is governed by the Business Source License included
  5. * in the LICENSE.TXT file in the project's root directory.
  6. *
  7. * Change Date: 2025-01-01
  8. *
  9. * On the date above, in accordance with the Business Source License, use
  10. * of this software will be governed by version 2.0 of the Apache License.
  11. */
  12. /****/
  13. #include "DBMirrorSet.hpp"
  14. namespace ZeroTier {
  15. DBMirrorSet::DBMirrorSet(DB::ChangeListener *listener) :
  16. _listener(listener),
  17. _running(true)
  18. {
  19. _syncCheckerThread = std::thread([this]() {
  20. for(;;) {
  21. for(int i=0;i<120;++i) { // 1 minute delay between checks
  22. if (!_running)
  23. return;
  24. std::this_thread::sleep_for(std::chrono::milliseconds(500));
  25. }
  26. std::vector< std::shared_ptr<DB> > dbs;
  27. {
  28. std::lock_guard<std::mutex> l(_dbs_l);
  29. if (_dbs.size() <= 1)
  30. continue; // no need to do this if there's only one DB, so skip the iteration
  31. dbs = _dbs;
  32. }
  33. for(auto db=dbs.begin();db!=dbs.end();++db) {
  34. (*db)->each([&dbs,&db](uint64_t networkId,const nlohmann::json &network,uint64_t memberId,const nlohmann::json &member) {
  35. try {
  36. if (network.is_object()) {
  37. if (memberId == 0) {
  38. for(auto db2=dbs.begin();db2!=dbs.end();++db2) {
  39. if (db->get() != db2->get()) {
  40. nlohmann::json nw2;
  41. if ((!(*db2)->get(networkId,nw2))||((nw2.is_object())&&(OSUtils::jsonInt(nw2["revision"],0) < OSUtils::jsonInt(network["revision"],0)))) {
  42. nw2 = network;
  43. (*db2)->save(nw2,false);
  44. }
  45. }
  46. }
  47. } else if (member.is_object()) {
  48. for(auto db2=dbs.begin();db2!=dbs.end();++db2) {
  49. if (db->get() != db2->get()) {
  50. nlohmann::json nw2,m2;
  51. if ((!(*db2)->get(networkId,nw2,memberId,m2))||((m2.is_object())&&(OSUtils::jsonInt(m2["revision"],0) < OSUtils::jsonInt(member["revision"],0)))) {
  52. m2 = member;
  53. (*db2)->save(m2,false);
  54. }
  55. }
  56. }
  57. }
  58. }
  59. } catch ( ... ) {} // skip entries that generate JSON errors
  60. });
  61. }
  62. }
  63. });
  64. }
  65. DBMirrorSet::~DBMirrorSet()
  66. {
  67. _running = false;
  68. _syncCheckerThread.join();
  69. }
  70. bool DBMirrorSet::hasNetwork(const uint64_t networkId) const
  71. {
  72. std::lock_guard<std::mutex> l(_dbs_l);
  73. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  74. if ((*d)->hasNetwork(networkId))
  75. return true;
  76. }
  77. return false;
  78. }
  79. bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network)
  80. {
  81. std::lock_guard<std::mutex> l(_dbs_l);
  82. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  83. if ((*d)->get(networkId,network)) {
  84. return true;
  85. }
  86. }
  87. return false;
  88. }
  89. bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member)
  90. {
  91. std::lock_guard<std::mutex> l(_dbs_l);
  92. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  93. if ((*d)->get(networkId,network,memberId,member))
  94. return true;
  95. }
  96. return false;
  97. }
  98. bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,DB::NetworkSummaryInfo &info)
  99. {
  100. std::lock_guard<std::mutex> l(_dbs_l);
  101. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  102. if ((*d)->get(networkId,network,memberId,member,info))
  103. return true;
  104. }
  105. return false;
  106. }
  107. bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members)
  108. {
  109. std::lock_guard<std::mutex> l(_dbs_l);
  110. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  111. if ((*d)->get(networkId,network,members))
  112. return true;
  113. }
  114. return false;
  115. }
  116. AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json &member, const std::string &redirectURL)
  117. {
  118. std::lock_guard<std::mutex> l(_dbs_l);
  119. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  120. AuthInfo info = (*d)->getSSOAuthInfo(member, redirectURL);
  121. if (info.enabled) {
  122. return info;
  123. }
  124. }
  125. return AuthInfo();
  126. }
  127. void DBMirrorSet::networkMemberSSOHasExpired(uint64_t nwid, int64_t ts)
  128. {
  129. std::lock_guard<std::mutex> l(_dbs_l);
  130. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  131. (*d)->networkMemberSSOHasExpired(nwid, ts);
  132. }
  133. }
  134. void DBMirrorSet::networks(std::set<uint64_t> &networks)
  135. {
  136. std::lock_guard<std::mutex> l(_dbs_l);
  137. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  138. (*d)->networks(networks);
  139. }
  140. }
  141. bool DBMirrorSet::waitForReady()
  142. {
  143. bool r = false;
  144. std::lock_guard<std::mutex> l(_dbs_l);
  145. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  146. r |= (*d)->waitForReady();
  147. }
  148. return r;
  149. }
  150. bool DBMirrorSet::isReady()
  151. {
  152. std::lock_guard<std::mutex> l(_dbs_l);
  153. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  154. if (!(*d)->isReady())
  155. return false;
  156. }
  157. return true;
  158. }
  159. bool DBMirrorSet::save(nlohmann::json &record,bool notifyListeners)
  160. {
  161. std::vector< std::shared_ptr<DB> > dbs;
  162. {
  163. std::lock_guard<std::mutex> l(_dbs_l);
  164. dbs = _dbs;
  165. }
  166. if (notifyListeners) {
  167. for(auto d=dbs.begin();d!=dbs.end();++d) {
  168. if ((*d)->save(record,true))
  169. return true;
  170. }
  171. return false;
  172. } else {
  173. bool modified = false;
  174. for(auto d=dbs.begin();d!=dbs.end();++d) {
  175. modified |= (*d)->save(record,false);
  176. }
  177. return modified;
  178. }
  179. }
  180. void DBMirrorSet::eraseNetwork(const uint64_t networkId)
  181. {
  182. std::lock_guard<std::mutex> l(_dbs_l);
  183. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  184. (*d)->eraseNetwork(networkId);
  185. }
  186. }
  187. void DBMirrorSet::eraseMember(const uint64_t networkId,const uint64_t memberId)
  188. {
  189. std::lock_guard<std::mutex> l(_dbs_l);
  190. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  191. (*d)->eraseMember(networkId,memberId);
  192. }
  193. }
  194. void DBMirrorSet::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress)
  195. {
  196. std::lock_guard<std::mutex> l(_dbs_l);
  197. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  198. (*d)->nodeIsOnline(networkId,memberId,physicalAddress);
  199. }
  200. }
  201. void DBMirrorSet::onNetworkUpdate(const void *db,uint64_t networkId,const nlohmann::json &network)
  202. {
  203. nlohmann::json record(network);
  204. std::lock_guard<std::mutex> l(_dbs_l);
  205. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  206. if (d->get() != db) {
  207. (*d)->save(record,false);
  208. }
  209. }
  210. _listener->onNetworkUpdate(this,networkId,network);
  211. }
  212. void DBMirrorSet::onNetworkMemberUpdate(const void *db,uint64_t networkId,uint64_t memberId,const nlohmann::json &member)
  213. {
  214. nlohmann::json record(member);
  215. std::lock_guard<std::mutex> l(_dbs_l);
  216. for(auto d=_dbs.begin();d!=_dbs.end();++d) {
  217. if (d->get() != db) {
  218. (*d)->save(record,false);
  219. }
  220. }
  221. _listener->onNetworkMemberUpdate(this,networkId,memberId,member);
  222. }
  223. void DBMirrorSet::onNetworkMemberDeauthorize(const void *db,uint64_t networkId,uint64_t memberId)
  224. {
  225. _listener->onNetworkMemberDeauthorize(this,networkId,memberId);
  226. }
  227. void DBMirrorSet::membersExpiring(std::set< std::pair<uint64_t, uint64_t> > &soon, std::set< std::pair<uint64_t, uint64_t> > &expired)
  228. {
  229. std::unique_lock<std::mutex> l(_membersExpiringSoon_l);
  230. int64_t now = OSUtils::now();
  231. for(auto next=_membersExpiringSoon.begin();next!=_membersExpiringSoon.end();) {
  232. if (next->first > now) {
  233. const uint64_t nwid = next->second.first;
  234. const uint64_t memberId = next->second.second;
  235. nlohmann::json network, member;
  236. if (this->get(nwid, network, memberId, member)) {
  237. try {
  238. const bool authorized = member["authorized"];
  239. const bool ssoExempt = member["ssoExempt"];
  240. const int64_t authenticationExpiryTime = member["authenticationExpiryTime"];
  241. if ((authenticationExpiryTime == next->first)&&(authorized)&&(!ssoExempt)) {
  242. if ((authenticationExpiryTime - now) > ZT_MEMBER_AUTH_TIMEOUT_NOTIFY_BEFORE) {
  243. // Stop when we get to entries too far in the future.
  244. break;
  245. } else {
  246. const bool ssoEnabled = network["ssoEnabled"];
  247. if (ssoEnabled)
  248. soon.insert(std::pair<uint64_t, uint64_t>(nwid, memberId));
  249. }
  250. } else {
  251. // Obsolete entry, no longer authorized, or SSO exempt.
  252. }
  253. } catch ( ... ) {
  254. // Invalid member object, erase.
  255. }
  256. } else {
  257. // Not found.
  258. }
  259. }
  260. _membersExpiringSoon.erase(next++);
  261. }
  262. }
  263. void DBMirrorSet::memberWillExpire(int64_t expTime, uint64_t nwid, uint64_t memberId)
  264. {
  265. std::unique_lock<std::mutex> l(_membersExpiringSoon_l);
  266. _membersExpiringSoon.insert(std::pair< int64_t, std::pair< uint64_t, uint64_t > >(expTime, std::pair< uint64_t, uint64_t >(nwid, memberId)));
  267. }
  268. } // namespace ZeroTier