CentralDB.cpp 64 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833
  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: 2026-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 "CentralDB.hpp"
  14. #ifdef ZT_CONTROLLER_USE_LIBPQ
  15. #include "../../node/Constants.hpp"
  16. #include "../../node/SHA512.hpp"
  17. #include "../../version.h"
  18. #include "BigTableStatusWriter.hpp"
  19. #include "ControllerChangeNotifier.hpp"
  20. #include "ControllerConfig.hpp"
  21. #include "CtlUtil.hpp"
  22. #include "EmbeddedNetworkController.hpp"
  23. #include "PostgresStatusWriter.hpp"
  24. #include "PubSubListener.hpp"
  25. #include "PubSubWriter.hpp"
  26. #include "Redis.hpp"
  27. #include "RedisListener.hpp"
  28. #include "RedisStatusWriter.hpp"
  29. #include "opentelemetry/trace/provider.h"
  30. #include <chrono>
  31. #include <climits>
  32. #include <iomanip>
  33. #include <libpq-fe.h>
  34. #include <optional>
  35. #include <pqxx/pqxx>
  36. #include <rustybits.h>
  37. #include <sstream>
  38. // #define REDIS_TRACE 1
  39. using json = nlohmann::json;
  40. using namespace ZeroTier;
  41. using Attrs = std::vector<std::pair<std::string, std::string> >;
  42. using Item = std::pair<std::string, Attrs>;
  43. using ItemStream = std::vector<Item>;
  44. CentralDB::CentralDB(
  45. const Identity& myId,
  46. const char* connString,
  47. int listenPort,
  48. CentralDB::ListenerMode listenMode,
  49. CentralDB::StatusWriterMode statusMode,
  50. const ControllerConfig* cc)
  51. : DB()
  52. , _listenerMode(listenMode)
  53. , _statusWriterMode(statusMode)
  54. , _cc(cc)
  55. , _pool()
  56. , _myId(myId)
  57. , _myAddress(myId.address())
  58. , _ready(0)
  59. , _connected(1)
  60. , _run(1)
  61. , _waitNoticePrinted(false)
  62. , _listenPort(listenPort)
  63. , _redis(NULL)
  64. , _cluster(NULL)
  65. , _redisMemberStatus(false)
  66. , _smee(NULL)
  67. {
  68. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  69. auto tracer = provider->GetTracer("CentralDB");
  70. auto span = tracer->StartSpan("CentralDB::CentralDB");
  71. auto scope = tracer->WithActiveSpan(span);
  72. rustybits::init_async_runtime();
  73. char myAddress[64];
  74. _myAddressStr = myId.address().toString(myAddress);
  75. _connString = std::string(connString);
  76. auto f = std::make_shared<PostgresConnFactory>(_connString);
  77. _pool =
  78. std::make_shared<ConnectionPool<PostgresConnection> >(15, 5, std::static_pointer_cast<ConnectionFactory>(f));
  79. memset(_ssoPsk, 0, sizeof(_ssoPsk));
  80. char* const ssoPskHex = getenv("ZT_SSO_PSK");
  81. #ifdef ZT_TRACE
  82. fprintf(stderr, "ZT_SSO_PSK: %s\n", ssoPskHex);
  83. #endif
  84. if (ssoPskHex) {
  85. // SECURITY: note that ssoPskHex will always be null-terminated if libc actually
  86. // returns something non-NULL. If the hex encodes something shorter than 48 bytes,
  87. // it will be padded at the end with zeroes. If longer, it'll be truncated.
  88. Utils::unhex(ssoPskHex, _ssoPsk, sizeof(_ssoPsk));
  89. }
  90. const char* redisMemberStatus = getenv("ZT_REDIS_MEMBER_STATUS");
  91. if (redisMemberStatus && (strcmp(redisMemberStatus, "true") == 0)) {
  92. _redisMemberStatus = true;
  93. fprintf(stderr, "Using redis for member status\n");
  94. }
  95. if ((listenMode == LISTENER_MODE_REDIS || statusMode == STATUS_WRITER_MODE_REDIS) && _cc->redisConfig != NULL) {
  96. auto innerspan = tracer->StartSpan("CentralDB::CentralDB::configureRedis");
  97. auto innerscope = tracer->WithActiveSpan(innerspan);
  98. sw::redis::ConnectionOptions opts;
  99. sw::redis::ConnectionPoolOptions poolOpts;
  100. opts.host = _cc->redisConfig->hostname;
  101. opts.port = _cc->redisConfig->port;
  102. opts.password = _cc->redisConfig->password;
  103. opts.db = 0;
  104. opts.keep_alive = true;
  105. opts.connect_timeout = std::chrono::seconds(3);
  106. poolOpts.size = 25;
  107. poolOpts.wait_timeout = std::chrono::seconds(5);
  108. poolOpts.connection_lifetime = std::chrono::minutes(3);
  109. poolOpts.connection_idle_time = std::chrono::minutes(1);
  110. if (_cc->redisConfig->clusterMode) {
  111. innerspan->SetAttribute("cluster_mode", "true");
  112. fprintf(stderr, "Using Redis in Cluster Mode\n");
  113. _cluster = std::make_shared<sw::redis::RedisCluster>(opts, poolOpts);
  114. }
  115. else {
  116. innerspan->SetAttribute("cluster_mode", "false");
  117. fprintf(stderr, "Using Redis in Standalone Mode\n");
  118. _redis = std::make_shared<sw::redis::Redis>(opts, poolOpts);
  119. }
  120. }
  121. _readyLock.lock();
  122. fprintf(
  123. stderr, "[%s] NOTICE: %.10llx controller PostgreSQL waiting for initial data download..." ZT_EOL_S,
  124. ::_timestr(), (unsigned long long)_myAddress.toInt());
  125. _waitNoticePrinted = true;
  126. initializeNetworks();
  127. initializeMembers();
  128. _heartbeatThread = std::thread(&CentralDB::heartbeat, this);
  129. switch (listenMode) {
  130. case LISTENER_MODE_REDIS:
  131. fprintf(stderr, "Using Redis for change listeners\n");
  132. if (_cc->redisConfig != NULL) {
  133. if (_cc->redisConfig->clusterMode) {
  134. _membersDbWatcher = std::make_shared<RedisMemberListener>(_myAddressStr, _cluster, this);
  135. _networksDbWatcher = std::make_shared<RedisNetworkListener>(_myAddressStr, _cluster, this);
  136. }
  137. else {
  138. _membersDbWatcher = std::make_shared<RedisMemberListener>(_myAddressStr, _redis, this);
  139. _networksDbWatcher = std::make_shared<RedisNetworkListener>(_myAddressStr, _redis, this);
  140. }
  141. }
  142. else {
  143. throw std::runtime_error("CentralDB: Redis listener mode selected but no Redis configuration provided");
  144. }
  145. case LISTENER_MODE_PUBSUB:
  146. fprintf(stderr, "Using PubSub for change listeners\n");
  147. if (cc->pubSubConfig != NULL) {
  148. _membersDbWatcher = std::make_shared<PubSubMemberListener>(
  149. _myAddressStr, cc->pubSubConfig->project_id, cc->pubSubConfig->member_change_recv_topic, this);
  150. _networksDbWatcher = std::make_shared<PubSubNetworkListener>(
  151. _myAddressStr, cc->pubSubConfig->project_id, cc->pubSubConfig->network_change_recv_topic, this);
  152. _changeNotifier = std::make_shared<PubSubChangeNotifier>(
  153. _myAddressStr, cc->pubSubConfig->project_id, cc->pubSubConfig->member_change_send_topic,
  154. cc->pubSubConfig->network_change_send_topic);
  155. }
  156. else {
  157. throw std::runtime_error(
  158. "CentralDB: PubSub listener mode selected but no PubSub configuration provided");
  159. }
  160. break;
  161. case LISTENER_MODE_PGSQL:
  162. default:
  163. fprintf(stderr, "Using PostgreSQL for change listeners\n");
  164. _membersDbWatcher = std::make_shared<PostgresMemberListener>(this, _pool, "member_" + _myAddressStr, 5);
  165. _networksDbWatcher = std::make_shared<PostgresNetworkListener>(this, _pool, "network_" + _myAddressStr, 5);
  166. break;
  167. }
  168. std::shared_ptr<PubSubWriter> pubsubWriter;
  169. switch (statusMode) {
  170. case STATUS_WRITER_MODE_REDIS:
  171. fprintf(stderr, "Using Redis for status writer\n");
  172. if (_cc->redisConfig != NULL) {
  173. if (_cc->redisConfig->clusterMode) {
  174. _statusWriter = std::make_shared<RedisStatusWriter>(_cluster, _myAddressStr);
  175. }
  176. else {
  177. _statusWriter = std::make_shared<RedisStatusWriter>(_redis, _myAddressStr);
  178. }
  179. }
  180. else {
  181. throw std::runtime_error("CentralDB: Redis status mode selected but no Redis configuration provided");
  182. }
  183. break;
  184. case STATUS_WRITER_MODE_BIGTABLE:
  185. fprintf(stderr, "Using BigTable for status writer\n");
  186. if (cc->bigTableConfig == NULL) {
  187. throw std::runtime_error(
  188. "CentralDB: BigTable status mode selected but no BigTable configuration provided");
  189. }
  190. if (cc->pubSubConfig == NULL) {
  191. throw std::runtime_error(
  192. "CentralDB: BigTable status mode selected but no PubSub configuration provided");
  193. }
  194. _statusWriter = std::make_shared<BigTableStatusWriter>(
  195. cc->bigTableConfig->project_id, cc->bigTableConfig->instance_id, cc->bigTableConfig->table_id);
  196. break;
  197. case STATUS_WRITER_MODE_PGSQL:
  198. default:
  199. fprintf(stderr, "Using PostgreSQL for status writer\n");
  200. _statusWriter = std::make_shared<PostgresStatusWriter>(_pool);
  201. break;
  202. }
  203. // start background threads
  204. for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {
  205. _commitThread[i] = std::thread(&CentralDB::commitThread, this);
  206. }
  207. _onlineNotificationThread = std::thread(&CentralDB::onlineNotificationThread, this);
  208. configureSmee();
  209. }
  210. CentralDB::~CentralDB()
  211. {
  212. if (_smee != NULL) {
  213. rustybits::smee_client_delete(_smee);
  214. _smee = NULL;
  215. }
  216. rustybits::shutdown_async_runtime();
  217. _run = 0;
  218. std::this_thread::sleep_for(std::chrono::milliseconds(100));
  219. _heartbeatThread.join();
  220. _commitQueue.stop();
  221. for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {
  222. _commitThread[i].join();
  223. }
  224. _onlineNotificationThread.join();
  225. }
  226. void CentralDB::configureSmee()
  227. {
  228. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  229. auto tracer = provider->GetTracer("CentralDB");
  230. auto span = tracer->StartSpan("CentralDB::configureSmee");
  231. auto scope = tracer->WithActiveSpan(span);
  232. const char* TEMPORAL_SCHEME = "ZT_TEMPORAL_SCHEME";
  233. const char* TEMPORAL_HOST = "ZT_TEMPORAL_HOST";
  234. const char* TEMPORAL_PORT = "ZT_TEMPORAL_PORT";
  235. const char* TEMPORAL_NAMESPACE = "ZT_TEMPORAL_NAMESPACE";
  236. const char* SMEE_TASK_QUEUE = "ZT_SMEE_TASK_QUEUE";
  237. const char* scheme = getenv(TEMPORAL_SCHEME);
  238. if (scheme == NULL) {
  239. scheme = "http";
  240. }
  241. const char* host = getenv(TEMPORAL_HOST);
  242. const char* port = getenv(TEMPORAL_PORT);
  243. const char* ns = getenv(TEMPORAL_NAMESPACE);
  244. const char* task_queue = getenv(SMEE_TASK_QUEUE);
  245. if (scheme != NULL && host != NULL && port != NULL && ns != NULL && task_queue != NULL) {
  246. fprintf(stderr, "creating smee client\n");
  247. std::string hostPort =
  248. std::string(scheme) + std::string("://") + std::string(host) + std::string(":") + std::string(port);
  249. this->_smee = rustybits::smee_client_new(hostPort.c_str(), ns, task_queue);
  250. }
  251. else {
  252. fprintf(stderr, "Smee client not configured\n");
  253. }
  254. }
  255. bool CentralDB::waitForReady()
  256. {
  257. while (_ready < 2) {
  258. _readyLock.lock();
  259. _readyLock.unlock();
  260. }
  261. return true;
  262. }
  263. bool CentralDB::isReady()
  264. {
  265. return ((_ready == 2) && (_connected));
  266. }
  267. bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
  268. {
  269. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  270. auto tracer = provider->GetTracer("CentralDB");
  271. auto span = tracer->StartSpan("CentralDB::save");
  272. auto scope = tracer->WithActiveSpan(span);
  273. bool modified = false;
  274. try {
  275. if (! record.is_object()) {
  276. fprintf(stderr, "record is not an object?!?\n");
  277. return false;
  278. }
  279. const std::string objtype = record["objtype"];
  280. if (objtype == "network") {
  281. fprintf(stderr, "CentralDB network save %s\n", record["id"].get<std::string>().c_str());
  282. const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL);
  283. if (nwid) {
  284. nlohmann::json old;
  285. get(nwid, old);
  286. if ((! old.is_object()) || (! _compareRecords(old, record))) {
  287. fprintf(stderr, "posting network change to commit queue\n");
  288. record["revision"] = OSUtils::jsonInt(record["revision"], 0ULL) + 1ULL;
  289. _commitQueue.post(std::pair<nlohmann::json, bool>(record, notifyListeners));
  290. modified = true;
  291. }
  292. }
  293. }
  294. else if (objtype == "member") {
  295. std::string networkId = record["nwid"];
  296. std::string memberId = record["id"];
  297. const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL);
  298. const uint64_t id = OSUtils::jsonIntHex(record["id"], 0ULL);
  299. fprintf(stderr, "member save %s-%s\n", networkId.c_str(), memberId.c_str());
  300. if ((id) && (nwid)) {
  301. nlohmann::json network, old;
  302. get(nwid, network, id, old);
  303. if ((! old.is_object()) || (! _compareRecords(old, record))) {
  304. fprintf(stderr, "posting member change to commit queue\n");
  305. record["revision"] = OSUtils::jsonInt(record["revision"], 0ULL) + 1ULL;
  306. _commitQueue.post(std::pair<nlohmann::json, bool>(record, notifyListeners));
  307. modified = true;
  308. }
  309. else {
  310. // fprintf(stderr, "no change\n");
  311. }
  312. }
  313. }
  314. else {
  315. fprintf(stderr, "uhh waaat\n");
  316. }
  317. }
  318. catch (std::exception& e) {
  319. fprintf(stderr, "Error on CentralDB::save: %s\n", e.what());
  320. }
  321. catch (...) {
  322. fprintf(stderr, "Unknown error on CentralDB::save\n");
  323. }
  324. return modified;
  325. }
  326. void CentralDB::eraseNetwork(const uint64_t networkId)
  327. {
  328. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  329. auto tracer = provider->GetTracer("CentralDB");
  330. auto span = tracer->StartSpan("CentralDB::eraseNetwork");
  331. auto scope = tracer->WithActiveSpan(span);
  332. char networkIdStr[17];
  333. span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
  334. fprintf(stderr, "CentralDB::eraseNetwork\n");
  335. char tmp2[24];
  336. waitForReady();
  337. Utils::hex(networkId, tmp2);
  338. std::pair<nlohmann::json, bool> tmp;
  339. tmp.first["id"] = tmp2;
  340. tmp.first["objtype"] = "_delete_network";
  341. tmp.second = true;
  342. _commitQueue.post(tmp);
  343. nlohmann::json nullJson;
  344. _networkChanged(tmp.first, nullJson, true);
  345. }
  346. void CentralDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
  347. {
  348. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  349. auto tracer = provider->GetTracer("CentralDB");
  350. auto span = tracer->StartSpan("CentralDB::eraseMember");
  351. auto scope = tracer->WithActiveSpan(span);
  352. char networkIdStr[17];
  353. char memberIdStr[11];
  354. span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
  355. span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
  356. fprintf(stderr, "CentralDB::eraseMember\n");
  357. char tmp2[24];
  358. waitForReady();
  359. std::pair<nlohmann::json, bool> tmp, nw;
  360. Utils::hex(networkId, tmp2);
  361. tmp.first["nwid"] = tmp2;
  362. Utils::hex(memberId, tmp2);
  363. tmp.first["id"] = tmp2;
  364. tmp.first["objtype"] = "_delete_member";
  365. tmp.second = true;
  366. _commitQueue.post(tmp);
  367. nlohmann::json nullJson;
  368. _memberChanged(tmp.first, nullJson, true);
  369. }
  370. void CentralDB::nodeIsOnline(
  371. const uint64_t networkId,
  372. const uint64_t memberId,
  373. const InetAddress& physicalAddress,
  374. const char* osArch)
  375. {
  376. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  377. auto tracer = provider->GetTracer("CentralDB");
  378. auto span = tracer->StartSpan("CentralDB::nodeIsOnline");
  379. auto scope = tracer->WithActiveSpan(span);
  380. char networkIdStr[17];
  381. char memberIdStr[11];
  382. char ipStr[INET6_ADDRSTRLEN];
  383. span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
  384. span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
  385. span->SetAttribute("physical_address", physicalAddress.toString(ipStr));
  386. span->SetAttribute("os_arch", osArch);
  387. std::lock_guard<std::mutex> l(_lastOnline_l);
  388. NodeOnlineRecord& i = _lastOnline[std::pair<uint64_t, uint64_t>(networkId, memberId)];
  389. i.lastSeen = OSUtils::now();
  390. if (physicalAddress) {
  391. i.physicalAddress = physicalAddress;
  392. }
  393. i.osArch = std::string(osArch);
  394. }
  395. void CentralDB::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress)
  396. {
  397. this->nodeIsOnline(networkId, memberId, physicalAddress, "unknown/unknown");
  398. }
  399. AuthInfo CentralDB::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL)
  400. {
  401. if (_cc->ssoEnabled) {
  402. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  403. auto tracer = provider->GetTracer("CentralDB");
  404. auto span = tracer->StartSpan("CentralDB::getSSOAuthInfo");
  405. auto scope = tracer->WithActiveSpan(span);
  406. Metrics::db_get_sso_info++;
  407. // NONCE is just a random character string. no semantic meaning
  408. // state = HMAC SHA384 of Nonce based on shared sso key
  409. //
  410. // need nonce timeout in database? make sure it's used within X time
  411. // X is 5 minutes for now. Make configurable later?
  412. //
  413. // how do we tell when a nonce is used? if auth_expiration_time is set
  414. std::string networkId = member["nwid"];
  415. std::string memberId = member["id"];
  416. char authenticationURL[4096] = { 0 };
  417. AuthInfo info;
  418. info.enabled = true;
  419. // if (memberId == "a10dccea52" && networkId == "8056c2e21c24673d") {
  420. // fprintf(stderr, "invalid authinfo for grant's machine\n");
  421. // info.version=1;
  422. // return info;
  423. // }
  424. // fprintf(stderr, "CentralDB::updateMemberOnLoad: %s-%s\n", networkId.c_str(), memberId.c_str());
  425. std::shared_ptr<PostgresConnection> c;
  426. try {
  427. c = _pool->borrow();
  428. pqxx::work w(*c->c);
  429. char nonceBytes[16] = { 0 };
  430. std::string nonce = "";
  431. // check if the member exists first.
  432. pqxx::row count =
  433. w.exec(
  434. "SELECT count(id) FROM ztc_member WHERE id = $1 AND network_id = $2 AND deleted = false",
  435. pqxx::params { memberId, networkId })
  436. .one_row();
  437. if (count[0].as<int>() == 1) {
  438. // get active nonce, if exists.
  439. pqxx::result r = w.exec(
  440. "SELECT nonce FROM ztc_sso_expiry "
  441. "WHERE network_id = $1 AND member_id = $2 "
  442. "AND ((NOW() AT TIME ZONE 'UTC') <= authentication_expiry_time) AND ((NOW() AT TIME ZONE 'UTC') <= "
  443. "nonce_expiration)",
  444. pqxx::params { networkId, memberId });
  445. if (r.size() == 0) {
  446. // no active nonce.
  447. // find an unused nonce, if one exists.
  448. pqxx::result r = w.exec(
  449. "SELECT nonce FROM ztc_sso_expiry "
  450. "WHERE network_id = $1 AND member_id = $2 "
  451. "AND authentication_expiry_time IS NULL AND ((NOW() AT TIME ZONE 'UTC') <= nonce_expiration)",
  452. pqxx::params { networkId, memberId });
  453. if (r.size() == 1) {
  454. // we have an existing nonce. Use it
  455. nonce = r.at(0)[0].as<std::string>();
  456. Utils::unhex(nonce.c_str(), nonceBytes, sizeof(nonceBytes));
  457. }
  458. else if (r.empty()) {
  459. // create a nonce
  460. Utils::getSecureRandom(nonceBytes, 16);
  461. char nonceBuf[64] = { 0 };
  462. Utils::hex(nonceBytes, sizeof(nonceBytes), nonceBuf);
  463. nonce = std::string(nonceBuf);
  464. pqxx::result ir = w.exec(
  465. "INSERT INTO ztc_sso_expiry "
  466. "(nonce, nonce_expiration, network_id, member_id) VALUES "
  467. "($1, TO_TIMESTAMP($2::double precision/1000), $3, $4)",
  468. pqxx::params { nonce, OSUtils::now() + 300000, networkId, memberId });
  469. w.commit();
  470. }
  471. else {
  472. // > 1 ?!? Thats an error!
  473. fprintf(stderr, "> 1 unused nonce!\n");
  474. exit(6);
  475. }
  476. }
  477. else if (r.size() == 1) {
  478. nonce = r.at(0)[0].as<std::string>();
  479. Utils::unhex(nonce.c_str(), nonceBytes, sizeof(nonceBytes));
  480. }
  481. else {
  482. // more than 1 nonce in use? Uhhh...
  483. fprintf(stderr, "> 1 nonce in use for network member?!?\n");
  484. exit(7);
  485. }
  486. r = w.exec(
  487. "SELECT oc.client_id, oc.authorization_endpoint, oc.issuer, oc.provider, oc.sso_impl_version "
  488. "FROM ztc_network AS n "
  489. "INNER JOIN ztc_org o "
  490. " ON o.owner_id = n.owner_id "
  491. "LEFT OUTER JOIN ztc_network_oidc_config noc "
  492. " ON noc.network_id = n.id "
  493. "LEFT OUTER JOIN ztc_oidc_config oc "
  494. " ON noc.client_id = oc.client_id AND oc.org_id = o.org_id "
  495. "WHERE n.id = $1 AND n.sso_enabled = true",
  496. pqxx::params { networkId });
  497. std::string client_id = "";
  498. std::string authorization_endpoint = "";
  499. std::string issuer = "";
  500. std::string provider = "";
  501. uint64_t sso_version = 0;
  502. if (r.size() == 1) {
  503. client_id = r.at(0)[0].as<std::optional<std::string> >().value_or("");
  504. authorization_endpoint = r.at(0)[1].as<std::optional<std::string> >().value_or("");
  505. issuer = r.at(0)[2].as<std::optional<std::string> >().value_or("");
  506. provider = r.at(0)[3].as<std::optional<std::string> >().value_or("");
  507. sso_version = r.at(0)[4].as<std::optional<uint64_t> >().value_or(1);
  508. }
  509. else if (r.size() > 1) {
  510. fprintf(
  511. stderr, "ERROR: More than one auth endpoint for an organization?!?!? NetworkID: %s\n",
  512. networkId.c_str());
  513. }
  514. else {
  515. fprintf(stderr, "No client or auth endpoint?!?\n");
  516. }
  517. info.version = sso_version;
  518. // no catch all else because we don't actually care if no records exist here. just continue as normal.
  519. if ((! client_id.empty()) && (! authorization_endpoint.empty())) {
  520. uint8_t state[48];
  521. HMACSHA384(_ssoPsk, nonceBytes, sizeof(nonceBytes), state);
  522. char state_hex[256];
  523. Utils::hex(state, 48, state_hex);
  524. if (info.version == 0) {
  525. char url[2048] = { 0 };
  526. OSUtils::ztsnprintf(
  527. url, sizeof(authenticationURL),
  528. "%s?response_type=id_token&response_mode=form_post&scope=openid+email+profile&redirect_uri="
  529. "%s&nonce=%s&state=%s&client_id=%s",
  530. authorization_endpoint.c_str(), url_encode(redirectURL).c_str(), nonce.c_str(), state_hex,
  531. client_id.c_str());
  532. info.authenticationURL = std::string(url);
  533. }
  534. else if (info.version == 1) {
  535. info.ssoClientID = client_id;
  536. info.issuerURL = issuer;
  537. info.ssoProvider = provider;
  538. info.ssoNonce = nonce;
  539. info.ssoState = std::string(state_hex) + "_" + networkId;
  540. info.centralAuthURL = redirectURL;
  541. #ifdef ZT_DEBUG
  542. fprintf(
  543. stderr,
  544. "ssoClientID: %s\nissuerURL: %s\nssoNonce: %s\nssoState: %s\ncentralAuthURL: %s\nprovider: "
  545. "%s\n",
  546. info.ssoClientID.c_str(), info.issuerURL.c_str(), info.ssoNonce.c_str(),
  547. info.ssoState.c_str(), info.centralAuthURL.c_str(), provider.c_str());
  548. #endif
  549. }
  550. }
  551. else {
  552. fprintf(
  553. stderr, "client_id: %s\nauthorization_endpoint: %s\n", client_id.c_str(),
  554. authorization_endpoint.c_str());
  555. }
  556. }
  557. _pool->unborrow(c);
  558. }
  559. catch (std::exception& e) {
  560. span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  561. fprintf(stderr, "ERROR: Error updating member on load for network %s: %s\n", networkId.c_str(), e.what());
  562. }
  563. return info; // std::string(authenticationURL);
  564. }
  565. return AuthInfo();
  566. }
  567. void CentralDB::initializeNetworks()
  568. {
  569. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  570. auto tracer = provider->GetTracer("CentralDB");
  571. auto span = tracer->StartSpan("CentralDB::initializeNetworks");
  572. auto scope = tracer->WithActiveSpan(span);
  573. fprintf(stderr, "Initializing networks...\n");
  574. try {
  575. char qbuf[2048];
  576. sprintf(
  577. qbuf,
  578. "SELECT id, name, configuration , (EXTRACT(EPOCH FROM creation_time AT TIME ZONE 'UTC')*1000)::bigint, "
  579. "(EXTRACT(EPOCH FROM last_modified AT TIME ZONE 'UTC')*1000)::bigint, revision, frontend "
  580. "FROM networks_ctl WHERE controller_id = '%s'",
  581. _myAddressStr.c_str());
  582. auto c = _pool->borrow();
  583. pqxx::work w(*c->c);
  584. fprintf(stderr, "Load networks from psql...\n");
  585. auto stream = pqxx::stream_from::query(w, qbuf);
  586. std::tuple<
  587. std::string // network ID
  588. ,
  589. std::optional<std::string> // name
  590. ,
  591. std::string // configuration
  592. ,
  593. std::optional<uint64_t> // creation_time
  594. ,
  595. std::optional<uint64_t> // last_modified
  596. ,
  597. std::optional<uint64_t> // revision
  598. ,
  599. std::string // frontend
  600. >
  601. row;
  602. uint64_t count = 0;
  603. uint64_t total = 0;
  604. while (stream >> row) {
  605. auto start = std::chrono::high_resolution_clock::now();
  606. json empty;
  607. json config;
  608. initNetwork(config);
  609. std::string nwid = std::get<0>(row);
  610. std::string name = std::get<1>(row).value_or("");
  611. json cfgtmp = json::parse(std::get<2>(row));
  612. std::optional<uint64_t> created_at = std::get<3>(row);
  613. std::optional<uint64_t> last_modified = std::get<4>(row);
  614. std::optional<uint64_t> revision = std::get<5>(row);
  615. std::string frontend = std::get<6>(row);
  616. config["id"] = nwid;
  617. config["name"] = name;
  618. config["creationTime"] = created_at.value_or(0);
  619. config["lastModified"] = last_modified.value_or(0);
  620. config["revision"] = revision.value_or(0);
  621. config["capabilities"] = cfgtmp["capabilities"].is_array() ? cfgtmp["capabilities"] : json::array();
  622. config["enableBroadcast"] =
  623. cfgtmp["enableBroadcast"].is_boolean() ? cfgtmp["enableBroadcast"].get<bool>() : false;
  624. config["mtu"] = cfgtmp["mtu"].is_number() ? cfgtmp["mtu"].get<int32_t>() : 2800;
  625. config["multicastLimit"] =
  626. cfgtmp["multicastLimit"].is_number() ? cfgtmp["multicastLimit"].get<int32_t>() : 64;
  627. config["private"] = cfgtmp["private"].is_boolean() ? cfgtmp["private"].get<bool>() : true;
  628. config["remoteTraceLevel"] =
  629. cfgtmp["remoteTraceLevel"].is_number() ? cfgtmp["remoteTraceLevel"].get<int32_t>() : 0;
  630. config["remoteTraceTarget"] =
  631. cfgtmp["remoteTraceTarget"].is_string() ? cfgtmp["remoteTraceTarget"].get<std::string>() : "";
  632. config["revision"] = revision.value_or(0);
  633. config["rules"] = cfgtmp["rules"].is_array() ? cfgtmp["rules"] : json::array();
  634. config["tags"] = cfgtmp["tags"].is_array() ? cfgtmp["tags"] : json::array();
  635. if (cfgtmp["v4AssignMode"].is_object()) {
  636. config["v4AssignMode"] = cfgtmp["v4AssignMode"];
  637. }
  638. else {
  639. config["v4AssignMode"] = json::object();
  640. config["v4AssignMode"]["zt"] = true;
  641. }
  642. if (cfgtmp["v6AssignMode"].is_object()) {
  643. config["v6AssignMode"] = cfgtmp["v6AssignMode"];
  644. }
  645. else {
  646. config["v6AssignMode"] = json::object();
  647. config["v6AssignMode"]["zt"] = true;
  648. config["v6AssignMode"]["6plane"] = true;
  649. config["v6AssignMode"]["rfc4193"] = false;
  650. }
  651. config["ssoEnabled"] = cfgtmp["ssoEnabled"].is_boolean() ? cfgtmp["ssoEnabled"].get<bool>() : false;
  652. if (config["ssoConfig"].is_object()) {
  653. config["ssoConfig"] = cfgtmp["ssoConfig"];
  654. }
  655. else {
  656. config["ssoConfig"] = empty;
  657. }
  658. config["objtype"] = "network";
  659. config["routes"] = cfgtmp["routes"].is_array() ? cfgtmp["routes"] : json::array();
  660. config["clientId"] = cfgtmp["clientId"].is_string() ? cfgtmp["clientId"].get<std::string>() : "";
  661. config["authorizationEndpoint"] =
  662. cfgtmp["authorizationEndpoint"].is_string() ? cfgtmp["authorizationEndpoint"].get<std::string>() : "";
  663. config["provider"] = cfgtmp["ssoProvider"].is_string() ? cfgtmp["ssoProvider"].get<std::string>() : "";
  664. if (! cfgtmp["dns"].is_object()) {
  665. cfgtmp["dns"] = json::object();
  666. cfgtmp["dns"]["domain"] = "";
  667. cfgtmp["dns"]["servers"] = json::array();
  668. }
  669. else {
  670. config["dns"] = cfgtmp["dns"];
  671. }
  672. config["ipAssignmentPools"] =
  673. cfgtmp["ipAssignmentPools"].is_array() ? cfgtmp["ipAssignmentPools"] : json::array();
  674. config["frontend"] = frontend;
  675. Metrics::network_count++;
  676. _networkChanged(empty, config, false);
  677. auto end = std::chrono::high_resolution_clock::now();
  678. auto dur = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
  679. total += dur.count();
  680. ++count;
  681. if (count > 0 && count % 10000 == 0) {
  682. fprintf(stderr, "Averaging %lu us per network\n", (total / count));
  683. }
  684. }
  685. w.commit();
  686. _pool->unborrow(c);
  687. fprintf(stderr, "done.\n");
  688. if (++this->_ready == 2) {
  689. if (_waitNoticePrinted) {
  690. fprintf(
  691. stderr, "[%s] NOTICE: %.10llx controller PostgreSQL data download complete." ZT_EOL_S, _timestr(),
  692. (unsigned long long)_myAddress.toInt());
  693. }
  694. _readyLock.unlock();
  695. }
  696. fprintf(stderr, "network init done\n");
  697. }
  698. catch (std::exception& e) {
  699. fprintf(stderr, "ERROR: Error initializing networks: %s\n", e.what());
  700. span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  701. std::this_thread::sleep_for(std::chrono::milliseconds(5000));
  702. exit(-1);
  703. }
  704. }
  705. void CentralDB::initializeMembers()
  706. {
  707. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  708. auto tracer = provider->GetTracer("CentralDB");
  709. auto span = tracer->StartSpan("CentralDB::initializeMembers");
  710. auto scope = tracer->WithActiveSpan(span);
  711. std::string memberId;
  712. std::string networkId;
  713. try {
  714. std::unordered_map<std::string, std::string> networkMembers;
  715. fprintf(stderr, "Initializing Members...\n");
  716. std::string setKeyBase = "network-nodes-all:{" + _myAddressStr + "}:";
  717. if (_redisMemberStatus) {
  718. fprintf(stderr, "Initialize Redis for members...\n");
  719. std::unique_lock<std::shared_mutex> l(_networks_l);
  720. std::unordered_set<std::string> deletes;
  721. for (auto it : _networks) {
  722. uint64_t nwid_i = it.first;
  723. char nwidTmp[64] = { 0 };
  724. OSUtils::ztsnprintf(nwidTmp, sizeof(nwidTmp), "%.16llx", nwid_i);
  725. std::string nwid(nwidTmp);
  726. std::string key = setKeyBase + nwid;
  727. deletes.insert(key);
  728. }
  729. if (! deletes.empty()) {
  730. try {
  731. if (_cc->redisConfig->clusterMode) {
  732. auto tx = _cluster->transaction(_myAddressStr, true, false);
  733. for (std::string k : deletes) {
  734. tx.del(k);
  735. }
  736. tx.exec();
  737. }
  738. else {
  739. auto tx = _redis->transaction(true, false);
  740. for (std::string k : deletes) {
  741. tx.del(k);
  742. }
  743. tx.exec();
  744. }
  745. }
  746. catch (sw::redis::Error& e) {
  747. // ignore
  748. }
  749. }
  750. }
  751. char qbuf[2048];
  752. sprintf(
  753. qbuf,
  754. "SELECT nm.device_id, nm.network_id, nm.authorized, nm.active_bridge, nm.ip_assignments, "
  755. "nm.no_auto_assign_ips, "
  756. "nm.sso_exempt, (EXTRACT(EPOCH FROM nm.authentication_expiry_time AT TIME ZONE 'UTC')*1000)::bigint, "
  757. "(EXTRACT(EPOCH FROM nm.creation_time AT TIME ZONE 'UTC')*1000)::bigint, nm.identity, "
  758. "(EXTRACT(EPOCH FROM nm.last_authorized_time AT TIME ZONE 'UTC')*1000)::bigint, "
  759. "(EXTRACT(EPOCH FROM nm.last_deauthorized_time AT TIME ZONE 'UTC')*1000)::bigint, "
  760. "nm.remote_trace_level, nm.remote_trace_target, nm.revision, nm.capabilities, nm.tags, "
  761. "nm.frontend "
  762. "FROM network_memberships_ctl nm "
  763. "INNER JOIN networks_ctl n "
  764. " ON nm.network_id = n.id "
  765. "WHERE n.controller_id = '%s'",
  766. _myAddressStr.c_str());
  767. auto c = _pool->borrow();
  768. pqxx::work w(*c->c);
  769. fprintf(stderr, "Load members from psql...\n");
  770. auto stream = pqxx::stream_from::query(w, qbuf);
  771. std::tuple<
  772. std::string // device ID
  773. ,
  774. std::string // network ID
  775. ,
  776. bool // authorized
  777. ,
  778. std::optional<bool> // active_bridge
  779. ,
  780. std::optional<std::string> // ip_assignments
  781. ,
  782. std::optional<bool> // no_auto_assign_ips
  783. ,
  784. std::optional<bool> // sso_exempt
  785. ,
  786. std::optional<uint64_t> // authentication_expiry_time
  787. ,
  788. std::optional<uint64_t> // creation_time
  789. ,
  790. std::optional<std::string> // identity
  791. ,
  792. std::optional<uint64_t> // last_authorized_time
  793. ,
  794. std::optional<uint64_t> // last_deauthorized_time
  795. ,
  796. std::optional<int32_t> // remote_trace_level
  797. ,
  798. std::optional<std::string> // remote_trace_target
  799. ,
  800. std::optional<uint64_t> // revision
  801. ,
  802. std::optional<std::string> // capabilities
  803. ,
  804. std::optional<std::string> // tags
  805. ,
  806. std::string // frontend
  807. >
  808. row;
  809. auto tmp = std::chrono::high_resolution_clock::now();
  810. uint64_t count = 0;
  811. uint64_t total = 0;
  812. while (stream >> row) {
  813. auto start = std::chrono::high_resolution_clock::now();
  814. json empty;
  815. json config;
  816. initMember(config);
  817. memberId = std::get<0>(row);
  818. networkId = std::get<1>(row);
  819. bool authorized = std::get<2>(row);
  820. std::optional<bool> active_bridge = std::get<3>(row);
  821. std::string ip_assignments = std::get<4>(row).value_or("");
  822. std::optional<bool> no_auto_assign_ips = std::get<5>(row);
  823. std::optional<bool> sso_exempt = std::get<6>(row);
  824. std::optional<uint64_t> authentication_expiry_time = std::get<7>(row);
  825. std::optional<uint64_t> creation_time = std::get<8>(row);
  826. std::optional<std::string> identity = std::get<9>(row);
  827. std::optional<uint64_t> last_authorized_time = std::get<10>(row);
  828. std::optional<uint64_t> last_deauthorized_time = std::get<11>(row);
  829. std::optional<int32_t> remote_trace_level = std::get<12>(row);
  830. std::optional<std::string> remote_trace_target = std::get<13>(row);
  831. std::optional<uint64_t> revision = std::get<14>(row);
  832. std::optional<std::string> capabilities = std::get<15>(row);
  833. std::optional<std::string> tags = std::get<16>(row);
  834. networkMembers.insert(std::pair<std::string, std::string>(setKeyBase + networkId, memberId));
  835. config["objtype"] = "member";
  836. config["id"] = memberId;
  837. config["address"] = identity.value_or("");
  838. config["nwid"] = networkId;
  839. config["authorized"] = authorized;
  840. config["activeBridge"] = active_bridge.value_or(false);
  841. config["ipAssignments"] = json::array();
  842. if (ip_assignments != "{}") {
  843. std::string tmp = ip_assignments.substr(1, ip_assignments.length() - 2);
  844. std::vector<std::string> addrs = split(tmp, ',');
  845. for (auto it = addrs.begin(); it != addrs.end(); ++it) {
  846. config["ipAssignments"].push_back(*it);
  847. }
  848. }
  849. config["capabilities"] = json::parse(capabilities.value_or("[]"));
  850. config["creationTime"] = creation_time.value_or(0);
  851. config["lastAuthorizedTime"] = last_authorized_time.value_or(0);
  852. config["lastDeauthorizedTime"] = last_deauthorized_time.value_or(0);
  853. config["noAutoAssignIPs"] = no_auto_assign_ips.value_or(false);
  854. config["remoteTraceLevel"] = remote_trace_level.value_or(0);
  855. config["remoteTraceTarget"] = remote_trace_target.value_or(nullptr);
  856. config["revision"] = revision.value_or(0);
  857. config["ssoExempt"] = sso_exempt.value_or(false);
  858. config["authenticationExpiryTime"] = authentication_expiry_time.value_or(0);
  859. config["tags"] = json::parse(tags.value_or("[]"));
  860. config["frontend"] = std::get<17>(row);
  861. Metrics::member_count++;
  862. _memberChanged(empty, config, false);
  863. memberId = "";
  864. networkId = "";
  865. auto end = std::chrono::high_resolution_clock::now();
  866. auto dur = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
  867. total += dur.count();
  868. ++count;
  869. if (count > 0 && count % 10000 == 0) {
  870. fprintf(stderr, "Averaging %llu us per member\n", (total / count));
  871. }
  872. }
  873. if (count > 0) {
  874. fprintf(stderr, "Took %llu us per member to load\n", (total / count));
  875. }
  876. stream.complete();
  877. w.commit();
  878. _pool->unborrow(c);
  879. fprintf(stderr, "done.\n");
  880. if (_listenerMode == LISTENER_MODE_REDIS)
  881. if (! networkMembers.empty()) {
  882. if (_redisMemberStatus) {
  883. fprintf(stderr, "Load member data into redis...\n");
  884. if (_cc->redisConfig->clusterMode) {
  885. auto tx = _cluster->transaction(_myAddressStr, true, false);
  886. uint64_t count = 0;
  887. for (auto it : networkMembers) {
  888. tx.sadd(it.first, it.second);
  889. if (++count % 30000 == 0) {
  890. tx.exec();
  891. tx = _cluster->transaction(_myAddressStr, true, false);
  892. }
  893. }
  894. tx.exec();
  895. }
  896. else {
  897. auto tx = _redis->transaction(true, false);
  898. uint64_t count = 0;
  899. for (auto it : networkMembers) {
  900. tx.sadd(it.first, it.second);
  901. if (++count % 30000 == 0) {
  902. tx.exec();
  903. tx = _redis->transaction(true, false);
  904. }
  905. }
  906. tx.exec();
  907. }
  908. fprintf(stderr, "done.\n");
  909. }
  910. }
  911. fprintf(stderr, "Done loading members...\n");
  912. if (++this->_ready == 2) {
  913. if (_waitNoticePrinted) {
  914. fprintf(
  915. stderr, "[%s] NOTICE: %.10llx controller PostgreSQL data download complete." ZT_EOL_S, _timestr(),
  916. (unsigned long long)_myAddress.toInt());
  917. }
  918. _readyLock.unlock();
  919. }
  920. }
  921. catch (sw::redis::Error& e) {
  922. span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  923. fprintf(stderr, "ERROR: Error initializing members (redis): %s\n", e.what());
  924. exit(-1);
  925. }
  926. catch (std::exception& e) {
  927. span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  928. fprintf(stderr, "ERROR: Error initializing member: %s-%s %s\n", networkId.c_str(), memberId.c_str(), e.what());
  929. exit(-1);
  930. }
  931. }
  932. void CentralDB::heartbeat()
  933. {
  934. char publicId[1024];
  935. char hostnameTmp[1024];
  936. _myId.toString(false, publicId);
  937. if (gethostname(hostnameTmp, sizeof(hostnameTmp)) != 0) {
  938. hostnameTmp[0] = (char)0;
  939. }
  940. else {
  941. for (int i = 0; i < (int)sizeof(hostnameTmp); ++i) {
  942. if ((hostnameTmp[i] == '.') || (hostnameTmp[i] == 0)) {
  943. hostnameTmp[i] = (char)0;
  944. break;
  945. }
  946. }
  947. }
  948. const char* controllerId = _myAddressStr.c_str();
  949. const char* publicIdentity = publicId;
  950. const char* hostname = hostnameTmp;
  951. while (_run == 1) {
  952. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  953. auto tracer = provider->GetTracer("CentralDB");
  954. auto span = tracer->StartSpan("CentralDB::heartbeat");
  955. auto scope = tracer->WithActiveSpan(span);
  956. // fprintf(stderr, "%s: heartbeat\n", controllerId);
  957. auto c = _pool->borrow();
  958. int64_t ts = OSUtils::now();
  959. if (c->c) {
  960. std::string major = std::to_string(ZEROTIER_ONE_VERSION_MAJOR);
  961. std::string minor = std::to_string(ZEROTIER_ONE_VERSION_MINOR);
  962. std::string rev = std::to_string(ZEROTIER_ONE_VERSION_REVISION);
  963. std::string version = major + "." + minor + "." + rev;
  964. std::string versionStr = "v" + version;
  965. try {
  966. pqxx::work w { *c->c };
  967. w.exec(
  968. "INSERT INTO controllers_ctl (id, hostname, last_heartbeat, public_identity, version) VALUES "
  969. "($1, $2, TO_TIMESTAMP($3::double precision/1000), $4, $5) "
  970. "ON CONFLICT (id) DO UPDATE SET hostname = EXCLUDED.hostname, last_heartbeat = "
  971. "EXCLUDED.last_heartbeat, "
  972. "public_identity = EXCLUDED.public_identity, version = EXCLUDED.version",
  973. pqxx::params { controllerId, hostname, ts, publicIdentity, versionStr })
  974. .no_rows();
  975. w.commit();
  976. }
  977. catch (std::exception& e) {
  978. fprintf(stderr, "%s: Heartbeat update failed: %s\n", controllerId, e.what());
  979. span->End();
  980. std::this_thread::sleep_for(std::chrono::milliseconds(1000));
  981. continue;
  982. }
  983. }
  984. _pool->unborrow(c);
  985. try {
  986. if (_listenerMode == LISTENER_MODE_REDIS && _redisMemberStatus) {
  987. if (_cc->redisConfig->clusterMode) {
  988. _cluster->zadd("controllers", "controllerId", ts);
  989. }
  990. else {
  991. _redis->zadd("controllers", "controllerId", ts);
  992. }
  993. }
  994. }
  995. catch (sw::redis::Error& e) {
  996. fprintf(stderr, "ERROR: Redis error in heartbeat thread: %s\n", e.what());
  997. }
  998. span->End();
  999. std::this_thread::sleep_for(std::chrono::milliseconds(1000));
  1000. }
  1001. fprintf(stderr, "Exited heartbeat thread\n");
  1002. }
  1003. void CentralDB::commitThread()
  1004. {
  1005. fprintf(stderr, "%s: commitThread start\n", _myAddressStr.c_str());
  1006. std::pair<nlohmann::json, bool> qitem;
  1007. while (_commitQueue.get(qitem) & (_run == 1)) {
  1008. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  1009. auto tracer = provider->GetTracer("CentralDB");
  1010. auto span = tracer->StartSpan("CentralDB::commitThread");
  1011. auto scope = tracer->WithActiveSpan(span);
  1012. fprintf(stderr, "commitThread tick\n");
  1013. if (! qitem.first.is_object()) {
  1014. fprintf(stderr, "not an object\n");
  1015. continue;
  1016. }
  1017. std::shared_ptr<PostgresConnection> c;
  1018. try {
  1019. c = _pool->borrow();
  1020. }
  1021. catch (std::exception& e) {
  1022. fprintf(stderr, "ERROR: %s\n", e.what());
  1023. continue;
  1024. }
  1025. if (! c) {
  1026. fprintf(stderr, "Error getting database connection\n");
  1027. continue;
  1028. }
  1029. Metrics::pgsql_commit_ticks++;
  1030. try {
  1031. nlohmann::json& config = (qitem.first);
  1032. const std::string objtype = config["objtype"];
  1033. if (objtype == "member") {
  1034. auto mspan = tracer->StartSpan("CentralDB::commitThread::member");
  1035. auto mscope = tracer->WithActiveSpan(mspan);
  1036. // fprintf(stderr, "%s: commitThread: member\n", _myAddressStr.c_str());
  1037. std::string memberId;
  1038. std::string networkId;
  1039. try {
  1040. pqxx::work w(*c->c);
  1041. memberId = config["id"];
  1042. networkId = config["nwid"];
  1043. fprintf(stderr, "commit member %s-%s\n", networkId.c_str(), memberId.c_str());
  1044. std::string target = "NULL";
  1045. if (! config["remoteTraceTarget"].is_null()) {
  1046. target = config["remoteTraceTarget"];
  1047. }
  1048. // get network and the frontend it is assigned to
  1049. // if network does not exist, skip member update
  1050. pqxx::row nwrow =
  1051. w.exec(
  1052. "SELECT COUNT(id), frontend FROM networks_ctl WHERE id = $1 GROUP BY frontend",
  1053. pqxx::params { networkId })
  1054. .one_row();
  1055. int nwcount = nwrow[0].as<int>();
  1056. std::string frontend = nwrow[1].as<std::string>();
  1057. if (nwcount != 1) {
  1058. fprintf(stderr, "network %s does not exist. skipping member upsert\n", networkId.c_str());
  1059. w.abort();
  1060. _pool->unborrow(c);
  1061. continue;
  1062. }
  1063. pqxx::row mrow = w.exec(
  1064. "SELECT COUNT(device_id) FROM network_memberships_ctl WHERE device_id = $1 "
  1065. "AND network_id = $2",
  1066. pqxx::params { memberId, networkId })
  1067. .one_row();
  1068. int membercount = mrow[0].as<int>();
  1069. bool isNewMember = (membercount == 0);
  1070. std::string change_source;
  1071. if (! config["change_source"].is_null()) {
  1072. change_source = config["change_source"];
  1073. }
  1074. else {
  1075. change_source = "controller";
  1076. }
  1077. if (! isNewMember && change_source != "controller" && frontend != change_source) {
  1078. fprintf(
  1079. stderr, "skipping member %s-%s update. change source: %s, frontend: %s\n",
  1080. networkId.c_str(), memberId.c_str(), change_source.c_str(), frontend.c_str());
  1081. // if it is not a new member and the change source is not the controller and doesn't match the
  1082. // frontend, don't apply the change.
  1083. continue;
  1084. }
  1085. std::vector<std::string> ipAssignments;
  1086. fprintf(
  1087. stderr, "Saving IP Assignments: \n\tipAssignments: %s\n",
  1088. OSUtils::jsonDump(config["ipAssignments"], -1).c_str());
  1089. if (config["ipAssignments"].is_array()) {
  1090. for (auto& ip : config["ipAssignments"]) {
  1091. if (ip.is_string()) {
  1092. ipAssignments.push_back(ip.get<std::string>());
  1093. }
  1094. }
  1095. }
  1096. fprintf(stderr, "member json: %s\n", config.dump().c_str());
  1097. int64_t vMajor = OSUtils::jsonUInt(config["vMajor"], 0);
  1098. int64_t vMinor = OSUtils::jsonUInt(config["vMinor"], 0);
  1099. int64_t vRev = OSUtils::jsonUInt(config["vRev"], 0);
  1100. int64_t vProto = OSUtils::jsonUInt(config["vProto"], 0);
  1101. if (vMajor < 0)
  1102. vMajor = 0;
  1103. if (vMinor < 0)
  1104. vMinor = 0;
  1105. if (vRev < 0)
  1106. vRev = 0;
  1107. if (vProto < 0)
  1108. vProto = 0;
  1109. pqxx::result res =
  1110. w.exec(
  1111. "INSERT INTO network_memberships_ctl (device_id, network_id, authorized, active_bridge, "
  1112. "ip_assignments, "
  1113. "no_auto_assign_ips, sso_exempt, authentication_expiry_time, capabilities, creation_time, "
  1114. "identity, last_authorized_time, last_deauthorized_time, "
  1115. "remote_trace_level, remote_trace_target, revision, tags, version_major, version_minor, "
  1116. "version_revision, version_protocol) "
  1117. "VALUES ($1, $2, $3, $4, $5, $6, $7, TO_TIMESTAMP($8::double precision/1000), $9, "
  1118. "TO_TIMESTAMP($10::double precision/1000), $11, TO_TIMESTAMP($12::double precision/1000), "
  1119. "TO_TIMESTAMP($13::double precision/1000), $14, $15, $16, $17, $18, $19, $20, $21) "
  1120. "ON CONFLICT (device_id, network_id) DO UPDATE SET "
  1121. "authorized = EXCLUDED.authorized, active_bridge = EXCLUDED.active_bridge, "
  1122. "ip_assignments = EXCLUDED.ip_assignments, no_auto_assign_ips = "
  1123. "EXCLUDED.no_auto_assign_ips, "
  1124. "sso_exempt = EXCLUDED.sso_exempt, authentication_expiry_time = "
  1125. "EXCLUDED.authentication_expiry_time, "
  1126. "capabilities = EXCLUDED.capabilities, creation_time = EXCLUDED.creation_time, "
  1127. "identity = EXCLUDED.identity, last_authorized_time = EXCLUDED.last_authorized_time, "
  1128. "last_deauthorized_time = EXCLUDED.last_deauthorized_time, "
  1129. "remote_trace_level = EXCLUDED.remote_trace_level, remote_trace_target = "
  1130. "EXCLUDED.remote_trace_target, "
  1131. "revision = EXCLUDED.revision, tags = EXCLUDED.tags, version_major = "
  1132. "EXCLUDED.version_major, "
  1133. "version_minor = EXCLUDED.version_minor, version_revision = EXCLUDED.version_revision, "
  1134. "version_protocol = EXCLUDED.version_protocol",
  1135. pqxx::params { memberId,
  1136. networkId,
  1137. OSUtils::jsonBool(config["authorized"], false),
  1138. OSUtils::jsonBool(config["activeBridge"], false),
  1139. ipAssignments,
  1140. OSUtils::jsonBool(config["noAutoAssignIps"], false),
  1141. OSUtils::jsonBool(config["ssoExempt"], false),
  1142. OSUtils::jsonInt(config["authenticationExpiryTime"], 0),
  1143. OSUtils::jsonDump(config["capabilities"], -1),
  1144. OSUtils::jsonInt(config["creationTime"], OSUtils::now()),
  1145. OSUtils::jsonString(config["identity"], ""),
  1146. OSUtils::jsonInt(config["lastAuthorizedTime"], 0),
  1147. OSUtils::jsonInt(config["lastDeauthorizedTime"], 0),
  1148. OSUtils::jsonInt(config["remoteTraceLevel"], 0),
  1149. target,
  1150. OSUtils::jsonInt(config["revision"], 0),
  1151. OSUtils::jsonDump(config["tags"], -1),
  1152. vMajor,
  1153. vMinor,
  1154. vRev,
  1155. vProto })
  1156. .no_rows();
  1157. w.commit();
  1158. if (_listenerMode == LISTENER_MODE_PUBSUB) {
  1159. // Publish change to pubsub stream
  1160. if (config["change_source"].is_null() || config["change_source"] == "controller") {
  1161. nlohmann::json oldMember;
  1162. nlohmann::json newMember = config;
  1163. if (! isNewMember) {
  1164. oldMember = _getNetworkMember(w, networkId, memberId);
  1165. }
  1166. _changeNotifier->notifyMemberChange(oldMember, newMember, frontend);
  1167. }
  1168. }
  1169. if (_smee != NULL && isNewMember) {
  1170. // TODO: Smee Notifications for New Members
  1171. // pqxx::row row = w.exec_params1(
  1172. // "SELECT "
  1173. // " count(h.hook_id) "
  1174. // "FROM "
  1175. // " ztc_hook h "
  1176. // " INNER JOIN ztc_org o ON o.org_id = h.org_id "
  1177. // " INNER JOIN ztc_network n ON n.owner_id = o.owner_id "
  1178. // " WHERE "
  1179. // "n.id = $1 ",
  1180. // networkId);
  1181. // int64_t hookCount = row[0].as<int64_t>();
  1182. // if (hookCount > 0) {
  1183. // notifyNewMember(networkId, memberId);
  1184. // }
  1185. }
  1186. const uint64_t nwidInt = OSUtils::jsonIntHex(config["nwid"], 0ULL);
  1187. const uint64_t memberidInt = OSUtils::jsonIntHex(config["id"], 0ULL);
  1188. if (nwidInt && memberidInt) {
  1189. nlohmann::json nwOrig;
  1190. nlohmann::json memOrig;
  1191. nlohmann::json memNew(config);
  1192. get(nwidInt, nwOrig, memberidInt, memOrig);
  1193. _memberChanged(memOrig, memNew, qitem.second);
  1194. }
  1195. else {
  1196. fprintf(
  1197. stderr, "%s: Can't notify of change. Error parsing nwid or memberid: %llu-%llu\n",
  1198. _myAddressStr.c_str(), (unsigned long long)nwidInt, (unsigned long long)memberidInt);
  1199. }
  1200. }
  1201. catch (std::exception& e) {
  1202. fprintf(
  1203. stderr, "%s ERROR: Error updating member %s-%s: %s\n", _myAddressStr.c_str(), networkId.c_str(),
  1204. memberId.c_str(), e.what());
  1205. mspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1206. }
  1207. }
  1208. else if (objtype == "network") {
  1209. auto nspan = tracer->StartSpan("CentralDB::commitThread::network");
  1210. auto nscope = tracer->WithActiveSpan(nspan);
  1211. try {
  1212. // fprintf(stderr, "%s: commitThread: network\n", _myAddressStr.c_str());
  1213. pqxx::work w(*c->c);
  1214. std::string id = config["id"];
  1215. fprintf(stderr, "commit network %s\n", id.c_str());
  1216. pqxx::row nwrow =
  1217. w.exec("SELECT COUNT(id) frontend FROM networks_ctl WHERE id = $1", pqxx::params { id })
  1218. .one_row();
  1219. int nwcount = nwrow[0].as<int>();
  1220. bool isNewNetwork = (nwcount == 0);
  1221. std::string frontend = "";
  1222. if (! isNewNetwork) {
  1223. pqxx::row nwrow =
  1224. w.exec("SELECT frontend FROM networks_ctl WHERE id = $1", pqxx::params { id }).one_row();
  1225. frontend = nwrow[0].as<std::string>();
  1226. }
  1227. std::string change_source;
  1228. if (! config["change_source"].is_null()) {
  1229. change_source = config["change_source"];
  1230. }
  1231. if (! isNewNetwork && change_source != "controller" && frontend != change_source) {
  1232. // if it is not a new network and the change source is not the controller and doesn't match the
  1233. // frontend, don't apply the change.
  1234. fprintf(
  1235. stderr, "Skipping network update %s. isNewNetwork: %s, change_source: %s, frontend: %s\n",
  1236. id.c_str(), isNewNetwork ? "true" : "false", change_source.c_str(), frontend.c_str());
  1237. continue;
  1238. }
  1239. pqxx::result res = w.exec(
  1240. "INSERT INTO networks_ctl (id, name, configuration, controller_id, revision, frontend) "
  1241. "VALUES ($1, $2, $3, $4, $5, $6) "
  1242. "ON CONFLICT (id) DO UPDATE SET "
  1243. "name = EXCLUDED.name, configuration = EXCLUDED.configuration, revision = EXCLUDED.revision+1, "
  1244. "frontend = EXCLUDED.frontend",
  1245. pqxx::params { id, OSUtils::jsonString(config["name"], ""), OSUtils::jsonDump(config, -1),
  1246. _myAddressStr, ((uint64_t)config["revision"]), change_source });
  1247. w.commit();
  1248. if (_listenerMode == LISTENER_MODE_PUBSUB) {
  1249. // Publish change to pubsub stream
  1250. if (config["change_source"].is_null() || config["change_source"] == "controller") {
  1251. nlohmann::json oldNetwork;
  1252. nlohmann::json newNetwork = config;
  1253. if (! isNewNetwork) {
  1254. oldNetwork = _getNetwork(w, id);
  1255. }
  1256. _changeNotifier->notifyNetworkChange(oldNetwork, newNetwork, frontend);
  1257. }
  1258. }
  1259. const uint64_t nwidInt = OSUtils::jsonIntHex(config["id"], 0ULL);
  1260. if (nwidInt) {
  1261. nlohmann::json nwOrig;
  1262. nlohmann::json nwNew(config);
  1263. get(nwidInt, nwOrig);
  1264. _networkChanged(nwOrig, nwNew, qitem.second);
  1265. }
  1266. else {
  1267. fprintf(
  1268. stderr, "%s: Can't notify network changed: %llu\n", _myAddressStr.c_str(),
  1269. (unsigned long long)nwidInt);
  1270. }
  1271. }
  1272. catch (std::exception& e) {
  1273. nspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1274. fprintf(stderr, "%s ERROR: Error updating network: %s\n", _myAddressStr.c_str(), e.what());
  1275. }
  1276. if (_listenerMode == LISTENER_MODE_REDIS && _redisMemberStatus) {
  1277. try {
  1278. std::string id = config["id"];
  1279. std::string controllerId = _myAddressStr.c_str();
  1280. std::string key = "networks:{" + controllerId + "}";
  1281. if (_cc->redisConfig->clusterMode) {
  1282. _cluster->sadd(key, id);
  1283. }
  1284. else {
  1285. _redis->sadd(key, id);
  1286. }
  1287. }
  1288. catch (sw::redis::Error& e) {
  1289. nspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1290. fprintf(stderr, "ERROR: Error adding network to Redis: %s\n", e.what());
  1291. }
  1292. }
  1293. }
  1294. else if (objtype == "_delete_network") {
  1295. auto dspan = tracer->StartSpan("CentralDB::commitThread::_delete_network");
  1296. auto dscope = tracer->WithActiveSpan(dspan);
  1297. // fprintf(stderr, "%s: commitThread: delete network\n", _myAddressStr.c_str());
  1298. try {
  1299. pqxx::work w(*c->c);
  1300. std::string networkId = config["id"];
  1301. fprintf(stderr, "Deleting network %s\n", networkId.c_str());
  1302. w.exec("DELETE FROM network_memberships_ctl WHERE network_id = $1", pqxx::params { networkId });
  1303. w.exec("DELETE FROM networks_ctl WHERE id = $1", pqxx::params { networkId });
  1304. w.commit();
  1305. uint64_t nwidInt = OSUtils::jsonIntHex(config["nwid"], 0ULL);
  1306. json oldConfig;
  1307. get(nwidInt, oldConfig);
  1308. json empty;
  1309. _networkChanged(oldConfig, empty, qitem.second);
  1310. }
  1311. catch (std::exception& e) {
  1312. dspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1313. fprintf(stderr, "%s ERROR: Error deleting network: %s\n", _myAddressStr.c_str(), e.what());
  1314. }
  1315. if (_listenerMode == LISTENER_MODE_REDIS && _redisMemberStatus) {
  1316. try {
  1317. std::string id = config["id"];
  1318. std::string controllerId = _myAddressStr.c_str();
  1319. std::string key = "networks:{" + controllerId + "}";
  1320. if (_cc->redisConfig->clusterMode) {
  1321. _cluster->srem(key, id);
  1322. _cluster->del("network-nodes-online:{" + controllerId + "}:" + id);
  1323. }
  1324. else {
  1325. _redis->srem(key, id);
  1326. _redis->del("network-nodes-online:{" + controllerId + "}:" + id);
  1327. }
  1328. }
  1329. catch (sw::redis::Error& e) {
  1330. dspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1331. fprintf(stderr, "ERROR: Error adding network to Redis: %s\n", e.what());
  1332. }
  1333. }
  1334. }
  1335. else if (objtype == "_delete_member") {
  1336. auto mspan = tracer->StartSpan("CentralDB::commitThread::_delete_member");
  1337. auto mscope = tracer->WithActiveSpan(mspan);
  1338. // fprintf(stderr, "%s commitThread: delete member\n", _myAddressStr.c_str());
  1339. try {
  1340. pqxx::work w(*c->c);
  1341. std::string memberId = config["id"];
  1342. std::string networkId = config["nwid"];
  1343. fprintf(stderr, "Deleting member %s-%s\n", networkId.c_str(), memberId.c_str());
  1344. pqxx::result res =
  1345. w.exec(
  1346. "DELETE FROM network_memberships_ctl WHERE device_id = $1 AND network_id = $2",
  1347. pqxx::params { memberId, networkId })
  1348. .no_rows();
  1349. w.commit();
  1350. uint64_t nwidInt = OSUtils::jsonIntHex(config["nwid"], 0ULL);
  1351. uint64_t memberidInt = OSUtils::jsonIntHex(config["id"], 0ULL);
  1352. nlohmann::json networkConfig;
  1353. nlohmann::json oldConfig;
  1354. get(nwidInt, networkConfig, memberidInt, oldConfig);
  1355. json empty;
  1356. _memberChanged(oldConfig, empty, qitem.second);
  1357. }
  1358. catch (std::exception& e) {
  1359. mspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1360. fprintf(stderr, "%s ERROR: Error deleting member: %s\n", _myAddressStr.c_str(), e.what());
  1361. }
  1362. if (_listenerMode == LISTENER_MODE_REDIS && _redisMemberStatus) {
  1363. try {
  1364. std::string memberId = config["id"];
  1365. std::string networkId = config["nwid"];
  1366. std::string controllerId = _myAddressStr.c_str();
  1367. std::string key = "network-nodes-all:{" + controllerId + "}:" + networkId;
  1368. if (_cc->redisConfig->clusterMode) {
  1369. _cluster->srem(key, memberId);
  1370. _cluster->del("member:{" + controllerId + "}:" + networkId + ":" + memberId);
  1371. }
  1372. else {
  1373. _redis->srem(key, memberId);
  1374. _redis->del("member:{" + controllerId + "}:" + networkId + ":" + memberId);
  1375. }
  1376. }
  1377. catch (sw::redis::Error& e) {
  1378. mspan->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1379. fprintf(stderr, "ERROR: Error deleting member from Redis: %s\n", e.what());
  1380. }
  1381. }
  1382. }
  1383. else {
  1384. fprintf(stderr, "%s ERROR: unknown objtype\n", _myAddressStr.c_str());
  1385. }
  1386. }
  1387. catch (std::exception& e) {
  1388. span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
  1389. fprintf(stderr, "%s ERROR: Error getting objtype: %s\n", _myAddressStr.c_str(), e.what());
  1390. }
  1391. _pool->unborrow(c);
  1392. c.reset();
  1393. }
  1394. fprintf(stderr, "%s commitThread finished\n", _myAddressStr.c_str());
  1395. }
  1396. void CentralDB::notifyNewMember(const std::string& networkID, const std::string& memberID)
  1397. {
  1398. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  1399. auto tracer = provider->GetTracer("CentralDB");
  1400. auto span = tracer->StartSpan("CentralDB::notifyNewMember");
  1401. auto scope = tracer->WithActiveSpan(span);
  1402. rustybits::smee_client_notify_network_joined(_smee, networkID.c_str(), memberID.c_str());
  1403. }
  1404. void CentralDB::onlineNotificationThread()
  1405. {
  1406. waitForReady();
  1407. while (_run == 1) {
  1408. auto provider = opentelemetry::trace::Provider::GetTracerProvider();
  1409. auto tracer = provider->GetTracer("CentralDB");
  1410. auto span = tracer->StartSpan("CentralDB::onlineNotificationThread");
  1411. auto scope = tracer->WithActiveSpan(span);
  1412. try {
  1413. std::unordered_map<std::pair<uint64_t, uint64_t>, NodeOnlineRecord, _PairHasher> lastOnline;
  1414. {
  1415. std::lock_guard<std::mutex> l(_lastOnline_l);
  1416. lastOnline.swap(_lastOnline);
  1417. }
  1418. uint64_t updateCount = 0;
  1419. auto c = _pool->borrow();
  1420. pqxx::work w(*c->c);
  1421. for (auto i = lastOnline.begin(); i != lastOnline.end(); ++i) {
  1422. updateCount += 1;
  1423. uint64_t nwid_i = i->first.first;
  1424. char nwidTmp[64];
  1425. char memTmp[64];
  1426. char ipTmp[64];
  1427. OSUtils::ztsnprintf(nwidTmp, sizeof(nwidTmp), "%.16llx", nwid_i);
  1428. OSUtils::ztsnprintf(memTmp, sizeof(memTmp), "%.10llx", i->first.second);
  1429. nlohmann::json network, member;
  1430. if (! get(nwid_i, network, i->first.second, member)) {
  1431. continue; // skip non existent networks/members
  1432. }
  1433. std::string networkId(nwidTmp);
  1434. std::string memberId(memTmp);
  1435. try {
  1436. // check if the member exists first.
  1437. //
  1438. // exec_params1 will throw pqxx::unexpected_rows if not exactly one row is returned. If that's the
  1439. // case, skip this record and move on.
  1440. pqxx::row r = w.exec(
  1441. "SELECT device_id, network_id FROM network_memberships_ctl WHERE network_id = "
  1442. "$1 AND device_id "
  1443. "= $2",
  1444. pqxx::params { networkId, memberId })
  1445. .one_row();
  1446. }
  1447. catch (pqxx::unexpected_rows& e) {
  1448. continue;
  1449. }
  1450. int64_t ts = i->second.lastSeen;
  1451. std::string ipAddr = i->second.physicalAddress.toIpString(ipTmp);
  1452. std::string timestamp = std::to_string(ts);
  1453. std::string osArch = i->second.osArch;
  1454. std::vector<std::string> osArchSplit = split(osArch, '/');
  1455. std::string os = "unknown";
  1456. std::string arch = "unknown";
  1457. std::string frontend = OSUtils::jsonString(network["frontend"], "");
  1458. int vMajor = OSUtils::jsonInt(member["vMajor"], 0);
  1459. int vMinor = OSUtils::jsonInt(member["vMinor"], 0);
  1460. int vRev = OSUtils::jsonInt(member["vRev"], 0);
  1461. std::string version;
  1462. if (vMajor <= 0 && vMinor <= 0 && vRev <= 0) {
  1463. vMajor = 0;
  1464. vMinor = 0;
  1465. vRev = 0;
  1466. version = "unknown";
  1467. }
  1468. else {
  1469. version = "v" + std::to_string(vMajor) + "." + std::to_string(vMinor) + "." + std::to_string(vRev);
  1470. }
  1471. if (osArchSplit.size() == 2) {
  1472. os = osArchSplit[0];
  1473. arch = osArchSplit[1];
  1474. }
  1475. _statusWriter->updateNodeStatus(
  1476. networkId, memberId, os, arch, version, i->second.physicalAddress, ts, frontend);
  1477. }
  1478. _statusWriter->writePending();
  1479. w.commit();
  1480. _pool->unborrow(c);
  1481. }
  1482. catch (std::exception& e) {
  1483. fprintf(stderr, "%s: error in onlinenotification thread: %s\n", _myAddressStr.c_str(), e.what());
  1484. }
  1485. std::this_thread::sleep_for(std::chrono::seconds(10));
  1486. }
  1487. }
  1488. nlohmann::json CentralDB::_getNetworkMember(pqxx::work& tx, const std::string networkID, const std::string memberID)
  1489. {
  1490. nlohmann::json out;
  1491. try {
  1492. pqxx::row row =
  1493. tx.exec(
  1494. "SELECT nm.device_id, nm.network_id, nm.authorized, nm.active_bridge, nm.ip_assignments, "
  1495. "nm.no_auto_assign_ips, "
  1496. "nm.sso_exempt, (EXTRACT(EPOCH FROM nm.authentication_expiry_time AT TIME ZONE 'UTC')*1000)::bigint, "
  1497. "(EXTRACT(EPOCH FROM nm.creation_time AT TIME ZONE 'UTC')*1000)::bigint, nm.identity, "
  1498. "(EXTRACT(EPOCH FROM nm.last_authorized_time AT TIME ZONE 'UTC')*1000)::bigint, "
  1499. "(EXTRACT(EPOCH FROM nm.last_deauthorized_time AT TIME ZONE 'UTC')*1000)::bigint, "
  1500. "nm.remote_trace_level, nm.remote_trace_target, nm.revision, nm.capabilities, nm.tags, "
  1501. "nm.frontend "
  1502. "FROM network_memberships_ctl nm "
  1503. "INNER JOIN networks_ctl n "
  1504. " ON nm.network_id = n.id "
  1505. "WHERE nm.network_id = $1 AND nm.device_id = $2",
  1506. pqxx::params { networkID, memberID })
  1507. .one_row();
  1508. bool authorized = row[2].as<bool>();
  1509. std::optional<bool> active_bridge =
  1510. row[3].is_null() ? std::optional<bool>() : std::optional<bool>(row[3].as<bool>());
  1511. std::string ip_assignments = row[4].is_null() ? "{}" : row[4].as<std::string>();
  1512. std::optional<bool> no_auto_assign_ips =
  1513. row[5].is_null() ? std::optional<bool>() : std::optional<bool>(row[5].as<bool>());
  1514. std::optional<bool> sso_exempt =
  1515. row[6].is_null() ? std::optional<bool>() : std::optional<bool>(row[6].as<bool>());
  1516. std::optional<uint64_t> authentication_expiry_time =
  1517. row[7].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[7].as<uint64_t>());
  1518. std::optional<uint64_t> creation_time =
  1519. row[8].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[8].as<uint64_t>());
  1520. std::optional<std::string> identity =
  1521. row[9].is_null() ? std::optional<std::string>() : std::optional<std::string>(row[9].as<std::string>());
  1522. std::optional<uint64_t> last_authorized_time =
  1523. row[10].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[10].as<uint64_t>());
  1524. std::optional<uint64_t> last_deauthorized_time =
  1525. row[11].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[11].as<uint64_t>());
  1526. std::optional<int32_t> remote_trace_level =
  1527. row[12].is_null() ? std::optional<int32_t>() : std::optional<int32_t>(row[12].as<int32_t>());
  1528. std::optional<std::string> remote_trace_target =
  1529. row[13].is_null() ? std::optional<std::string>() : std::optional<std::string>(row[13].as<std::string>());
  1530. std::optional<uint64_t> revision =
  1531. row[14].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[14].as<uint64_t>());
  1532. std::optional<std::string> capabilities =
  1533. row[15].is_null() ? std::optional<std::string>() : std::optional<std::string>(row[15].as<std::string>());
  1534. std::optional<std::string> tags =
  1535. row[16].is_null() ? std::optional<std::string>() : std::optional<std::string>(row[16].as<std::string>());
  1536. std::string frontend = row[17].is_null() ? "" : row[17].as<std::string>();
  1537. out["objtype"] = "member";
  1538. out["id"] = memberID;
  1539. out["nwid"] = networkID;
  1540. out["address"] = identity.value_or("");
  1541. out["authorized"] = authorized;
  1542. out["activeBridge"] = active_bridge.value_or(false);
  1543. out["ipAssignments"] = json::array();
  1544. if (ip_assignments != "{}" && ip_assignments != "[]") {
  1545. std::string tmp = ip_assignments.substr(1, ip_assignments.length() - 2);
  1546. std::vector<std::string> addrs = split(tmp, ',');
  1547. for (auto it = addrs.begin(); it != addrs.end(); ++it) {
  1548. out["ipAssignments"].push_back(*it);
  1549. }
  1550. }
  1551. out["capabilities"] = json::parse(capabilities.value_or("[]"));
  1552. out["creationTime"] = creation_time.value_or(0);
  1553. out["lastAuthorizedTime"] = last_authorized_time.value_or(0);
  1554. out["lastDeauthorizedTime"] = last_deauthorized_time.value_or(0);
  1555. out["noAutoAssignIps"] = no_auto_assign_ips.value_or(false);
  1556. out["remoteTraceLevel"] = remote_trace_level.value_or(0);
  1557. out["remoteTraceTarget"] = remote_trace_target.value_or(nullptr);
  1558. out["revision"] = revision.value_or(0);
  1559. out["ssoExempt"] = sso_exempt.value_or(false);
  1560. out["authenticationExpiryTime"] = authentication_expiry_time.value_or(0);
  1561. out["tags"] = json::parse(tags.value_or("[]"));
  1562. out["frontend"] = frontend;
  1563. }
  1564. catch (std::exception& e) {
  1565. fprintf(
  1566. stderr, "ERROR: Error getting network member %s-%s: %s\n", networkID.c_str(), memberID.c_str(), e.what());
  1567. return nlohmann::json();
  1568. }
  1569. return out;
  1570. }
  1571. nlohmann::json CentralDB::_getNetwork(pqxx::work& tx, const std::string networkID)
  1572. {
  1573. nlohmann::json out;
  1574. try {
  1575. std::optional<std::string> name;
  1576. std::string cfg;
  1577. std::optional<uint64_t> creation_time;
  1578. std::optional<uint64_t> last_modified;
  1579. std::optional<uint64_t> revision;
  1580. std::string frontend;
  1581. pqxx::row row = tx.exec(
  1582. "SELECT id, name, configuration , (EXTRACT(EPOCH FROM creation_time AT TIME ZONE "
  1583. "'UTC')*1000)::bigint, "
  1584. "(EXTRACT(EPOCH FROM last_modified AT TIME ZONE 'UTC')*1000)::bigint, revision, frontend "
  1585. "FROM networks_ctl WHERE id = $1",
  1586. pqxx::params { networkID })
  1587. .one_row();
  1588. cfg = row[2].as<std::string>();
  1589. creation_time = row[3].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[3].as<uint64_t>());
  1590. last_modified = row[4].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[4].as<uint64_t>());
  1591. revision = row[5].is_null() ? std::optional<uint64_t>() : std::optional<uint64_t>(row[5].as<uint64_t>());
  1592. frontend = row[6].is_null() ? "" : row[6].as<std::string>();
  1593. nlohmann::json cfgtmp = nlohmann::json::parse(cfg);
  1594. if (! cfgtmp.is_object()) {
  1595. fprintf(stderr, "ERROR: Network %s configuration is not a JSON object\n", networkID.c_str());
  1596. return nlohmann::json();
  1597. }
  1598. out["objtype"] = "network";
  1599. out["id"] = row[0].as<std::string>();
  1600. out["name"] = row[1].is_null() ? "" : row[1].as<std::string>();
  1601. out["creationTime"] = creation_time.value_or(0);
  1602. out["lastModified"] = last_modified.value_or(0);
  1603. out["revision"] = revision.value_or(0);
  1604. out["capabilities"] = cfgtmp["capabilities"].is_array() ? cfgtmp["capabilities"] : json::array();
  1605. out["enableBroadcast"] = cfgtmp["enableBroadcast"].is_boolean() ? cfgtmp["enableBroadcast"].get<bool>() : false;
  1606. out["mtu"] = cfgtmp["mtu"].is_number() ? cfgtmp["mtu"].get<int32_t>() : 2800;
  1607. out["multicastLimit"] = cfgtmp["multicastLimit"].is_number() ? cfgtmp["multicastLimit"].get<int32_t>() : 64;
  1608. out["private"] = cfgtmp["private"].is_boolean() ? cfgtmp["private"].get<bool>() : true;
  1609. out["remoteTraceLevel"] =
  1610. cfgtmp["remoteTraceLevel"].is_number() ? cfgtmp["remoteTraceLevel"].get<int32_t>() : 0;
  1611. out["remoteTraceTarget"] =
  1612. cfgtmp["remoteTraceTarget"].is_string() ? cfgtmp["remoteTraceTarget"].get<std::string>() : "";
  1613. out["revision"] = revision.value_or(0);
  1614. out["rules"] = cfgtmp["rules"].is_array() ? cfgtmp["rules"] : json::array();
  1615. out["tags"] = cfgtmp["tags"].is_array() ? cfgtmp["tags"] : json::array();
  1616. if (cfgtmp["v4AssignMode"].is_object()) {
  1617. out["v4AssignMode"] = cfgtmp["v4AssignMode"];
  1618. }
  1619. else {
  1620. out["v4AssignMode"] = json::object();
  1621. out["v4AssignMode"]["zt"] = true;
  1622. }
  1623. if (cfgtmp["v6AssignMode"].is_object()) {
  1624. out["v6AssignMode"] = cfgtmp["v6AssignMode"];
  1625. }
  1626. else {
  1627. out["v6AssignMode"] = json::object();
  1628. out["v6AssignMode"]["zt"] = true;
  1629. out["v6AssignMode"]["6plane"] = true;
  1630. out["v6AssignMode"]["rfc4193"] = false;
  1631. }
  1632. out["ssoEnabled"] = cfgtmp["ssoEnabled"].is_boolean() ? cfgtmp["ssoEnabled"].get<bool>() : false;
  1633. out["objtype"] = "network";
  1634. out["routes"] = cfgtmp["routes"].is_array() ? cfgtmp["routes"] : json::array();
  1635. out["clientId"] = cfgtmp["clientId"].is_string() ? cfgtmp["clientId"].get<std::string>() : "";
  1636. out["authorizationEndpoint"] =
  1637. cfgtmp["authorizationEndpoint"].is_string() ? cfgtmp["authorizationEndpoint"].get<std::string>() : nullptr;
  1638. out["provider"] = cfgtmp["ssoProvider"].is_string() ? cfgtmp["ssoProvider"].get<std::string>() : "";
  1639. if (! cfgtmp["dns"].is_object()) {
  1640. cfgtmp["dns"] = json::object();
  1641. cfgtmp["dns"]["domain"] = "";
  1642. cfgtmp["dns"]["servers"] = json::array();
  1643. }
  1644. else {
  1645. out["dns"] = cfgtmp["dns"];
  1646. }
  1647. out["ipAssignmentPools"] = cfgtmp["ipAssignmentPools"].is_array() ? cfgtmp["ipAssignmentPools"] : json::array();
  1648. out["frontend"] = row[6].as<std::string>();
  1649. }
  1650. catch (std::exception& e) {
  1651. fprintf(stderr, "ERROR: Error getting network %s: %s\n", networkID.c_str(), e.what());
  1652. return nlohmann::json();
  1653. }
  1654. return out;
  1655. }
  1656. #endif // ZT_CONTROLLER_USE_LIBPQ