github/ZeroTierOne
1
0
Fork 0
mirror of https://github.com/ZeroTier/ZeroTierOne synced 2025-07-10 15:23:06 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

clang-format this branch

Browse source
This commit is contained in:
Adam Ierymenko 2025-07-03 12:02:18 -04:00
parent 8b77ef538a
commit 342fa9d33f
No known key found for this signature in database
GPG key ID: C8877CF2D7A5D7F3
135 changed files with 42729 additions and 42439 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.clang-format
View file

@ -57,7 +57,7 @@ SpacesInCStyleCastParentheses: 'false'
SpacesInContainerLiterals: 'true' SpacesInContainerLiterals: 'true'
SpacesInParentheses: 'false' SpacesInParentheses: 'false'
SpacesInSquareBrackets: 'false' SpacesInSquareBrackets: 'false'
UseTab: 'Never' UseTab: 'Always'
--- ---
Language: Cpp Language: Cpp

64
controller/ConnectionPool.hpp
View file

@ -14,38 +14,37 @@
#ifndef ZT_CONNECTION_POOL_H_ #ifndef ZT_CONNECTION_POOL_H_
#define ZT_CONNECTION_POOL_H_ #define ZT_CONNECTION_POOL_H_
#ifndef _DEBUG #ifndef _DEBUG
#define _DEBUG(x) #define _DEBUG(x)
#endif #endif
#include "../node/Metrics.hpp" #include "../node/Metrics.hpp"
#include <deque> #include <deque>
#include <set> #include <exception>
#include <memory> #include <memory>
#include <mutex> #include <mutex>
#include <exception> #include <set>
#include <string> #include <string>
namespace ZeroTier { namespace ZeroTier {
struct ConnectionUnavailable : std::exception { struct ConnectionUnavailable : std::exception {
char const* what() const throw() { char const* what() const throw()
{
return "Unable to allocate connection"; return "Unable to allocate connection";
}; };
}; };
class Connection { class Connection {
public: public:
virtual ~Connection() {}; virtual ~Connection() {};
}; };
class ConnectionFactory { class ConnectionFactory {
public: public:
virtual ~ConnectionFactory() {}; virtual ~ConnectionFactory() {};
virtual std::shared_ptr<Connection> create()=0; virtual std::shared_ptr<Connection> create() = 0;
}; };
struct ConnectionPoolStats { struct ConnectionPoolStats {
@ -53,23 +52,20 @@ struct ConnectionPoolStats {
size_t borrowed_size; size_t borrowed_size;
}; };
template<class T> template <class T> class ConnectionPool {
class ConnectionPool { public:
public: ConnectionPool(size_t max_pool_size, size_t min_pool_size, std::shared_ptr<ConnectionFactory> factory) : m_maxPoolSize(max_pool_size), m_minPoolSize(min_pool_size), m_factory(factory)
ConnectionPool(size_t max_pool_size, size_t min_pool_size, std::shared_ptr<ConnectionFactory> factory)
: m_maxPoolSize(max_pool_size)
, m_minPoolSize(min_pool_size)
, m_factory(factory)
{ {
Metrics::max_pool_size += max_pool_size; Metrics::max_pool_size += max_pool_size;
Metrics::min_pool_size += min_pool_size; Metrics::min_pool_size += min_pool_size;
while(m_pool.size() < m_minPoolSize){ while (m_pool.size() < m_minPoolSize) {
m_pool.push_back(m_factory->create()); m_pool.push_back(m_factory->create());
Metrics::pool_avail++; Metrics::pool_avail++;
} }
}; };
ConnectionPoolStats get_stats() { ConnectionPoolStats get_stats()
{
std::unique_lock<std::mutex> lock(m_poolMutex); std::unique_lock<std::mutex> lock(m_poolMutex);
ConnectionPoolStats stats; ConnectionPoolStats stats;
@ -79,8 +75,7 @@ public:
return stats; return stats;
}; };
~ConnectionPool() { ~ConnectionPool() {};
};
/** /**
* Borrow * Borrow
@ -90,30 +85,32 @@ public:
* When done, either (a) call unborrow() to return it, or (b) (if it's bad) just let it go out of scope. This will cause it to automatically be replaced. * When done, either (a) call unborrow() to return it, or (b) (if it's bad) just let it go out of scope. This will cause it to automatically be replaced.
* @retval a shared_ptr to the connection object * @retval a shared_ptr to the connection object
*/ */
std::shared_ptr<T> borrow() { std::shared_ptr<T> borrow()
{
std::unique_lock<std::mutex> l(m_poolMutex); std::unique_lock<std::mutex> l(m_poolMutex);
while((m_pool.size() + m_borrowed.size()) < m_minPoolSize) { while ((m_pool.size() + m_borrowed.size()) < m_minPoolSize) {
std::shared_ptr<Connection> conn = m_factory->create(); std::shared_ptr<Connection> conn = m_factory->create();
m_pool.push_back(conn); m_pool.push_back(conn);
Metrics::pool_avail++; Metrics::pool_avail++;
} }
if(m_pool.size()==0){ if (m_pool.size() == 0) {
if ((m_pool.size() + m_borrowed.size()) < m_maxPoolSize) { if ((m_pool.size() + m_borrowed.size()) < m_maxPoolSize) {
try { try {
std::shared_ptr<Connection> conn = m_factory->create(); std::shared_ptr<Connection> conn = m_factory->create();
m_borrowed.insert(conn); m_borrowed.insert(conn);
Metrics::pool_in_use++; Metrics::pool_in_use++;
return std::static_pointer_cast<T>(conn); return std::static_pointer_cast<T>(conn);
} catch (std::exception &e) { }
catch (std::exception& e) {
Metrics::pool_errors++; Metrics::pool_errors++;
throw ConnectionUnavailable(); throw ConnectionUnavailable();
} }
} else { }
for(auto it = m_borrowed.begin(); it != m_borrowed.end(); ++it){ else {
if((*it).unique()) { for (auto it = m_borrowed.begin(); it != m_borrowed.end(); ++it) {
if ((*it).unique()) {
// This connection has been abandoned! Destroy it and create a new connection // This connection has been abandoned! Destroy it and create a new connection
try { try {
// If we are able to create a new connection, return it // If we are able to create a new connection, return it
@ -122,7 +119,8 @@ public:
m_borrowed.erase(it); m_borrowed.erase(it);
m_borrowed.insert(conn); m_borrowed.insert(conn);
return std::static_pointer_cast<T>(conn); return std::static_pointer_cast<T>(conn);
} catch(std::exception& e) { }
catch (std::exception& e) {
// Error creating a replacement connection // Error creating a replacement connection
Metrics::pool_errors++; Metrics::pool_errors++;
throw ConnectionUnavailable(); throw ConnectionUnavailable();
@ -151,7 +149,8 @@ public:
* Only call this if you are returning a working connection. If the connection was bad, just let it go out of scope (so the connection manager can replace it). * Only call this if you are returning a working connection. If the connection was bad, just let it go out of scope (so the connection manager can replace it).
* @param the connection * @param the connection
*/ */
void unborrow(std::shared_ptr<T> conn) { void unborrow(std::shared_ptr<T> conn)
{
// Lock // Lock
std::unique_lock<std::mutex> lock(m_poolMutex); std::unique_lock<std::mutex> lock(m_poolMutex);
m_borrowed.erase(conn); m_borrowed.erase(conn);
@ -161,7 +160,8 @@ public:
m_pool.push_back(conn); m_pool.push_back(conn);
} }
}; };
protected:
protected:
size_t m_maxPoolSize; size_t m_maxPoolSize;
size_t m_minPoolSize; size_t m_minPoolSize;
std::shared_ptr<ConnectionFactory> m_factory; std::shared_ptr<ConnectionFactory> m_factory;
@ -170,6 +170,6 @@ protected:
std::mutex m_poolMutex; std::mutex m_poolMutex;
}; };
} } // namespace ZeroTier
#endif #endif

277
controller/DB.cpp
View file

@ -12,77 +12,114 @@
/****/ /****/
#include "DB.hpp" #include "DB.hpp"
#include "EmbeddedNetworkController.hpp"
#include "../node/Metrics.hpp"
#include <chrono> #include "../node/Metrics.hpp"
#include "EmbeddedNetworkController.hpp"
#include <algorithm> #include <algorithm>
#include <chrono>
#include <stdexcept> #include <stdexcept>
using json = nlohmann::json; using json = nlohmann::json;
namespace ZeroTier { namespace ZeroTier {
void DB::initNetwork(nlohmann::json &network) void DB::initNetwork(nlohmann::json& network)
{ {
if (!network.count("private")) network["private"] = true; if (! network.count("private"))
if (!network.count("creationTime")) network["creationTime"] = OSUtils::now(); network["private"] = true;
if (!network.count("name")) network["name"] = ""; if (! network.count("creationTime"))
if (!network.count("multicastLimit")) network["multicastLimit"] = (uint64_t)32; network["creationTime"] = OSUtils::now();
if (!network.count("enableBroadcast")) network["enableBroadcast"] = true; if (! network.count("name"))
if (!network.count("v4AssignMode")) network["v4AssignMode"] = {{"zt",false}}; network["name"] = "";
if (!network.count("v6AssignMode")) network["v6AssignMode"] = {{"rfc4193",false},{"zt",false},{"6plane",false}}; if (! network.count("multicastLimit"))
if (!network.count("authTokens")) network["authTokens"] = {{}}; network["multicastLimit"] = (uint64_t)32;
if (!network.count("capabilities")) network["capabilities"] = nlohmann::json::array(); if (! network.count("enableBroadcast"))
if (!network.count("tags")) network["tags"] = nlohmann::json::array(); network["enableBroadcast"] = true;
if (!network.count("routes")) network["routes"] = nlohmann::json::array(); if (! network.count("v4AssignMode"))
if (!network.count("ipAssignmentPools")) network["ipAssignmentPools"] = nlohmann::json::array(); network["v4AssignMode"] = { { "zt", false } };
if (!network.count("mtu")) network["mtu"] = ZT_DEFAULT_MTU; if (! network.count("v6AssignMode"))
if (!network.count("remoteTraceTarget")) network["remoteTraceTarget"] = nlohmann::json(); network["v6AssignMode"] = { { "rfc4193", false }, { "zt", false }, { "6plane", false } };
if (!network.count("removeTraceLevel")) network["remoteTraceLevel"] = 0; if (! network.count("authTokens"))
if (!network.count("rulesSource")) network["rulesSource"] = ""; network["authTokens"] = { {} };
if (!network.count("rules")) { if (! network.count("capabilities"))
network["capabilities"] = nlohmann::json::array();
if (! network.count("tags"))
network["tags"] = nlohmann::json::array();
if (! network.count("routes"))
network["routes"] = nlohmann::json::array();
if (! network.count("ipAssignmentPools"))
network["ipAssignmentPools"] = nlohmann::json::array();
if (! network.count("mtu"))
network["mtu"] = ZT_DEFAULT_MTU;
if (! network.count("remoteTraceTarget"))
network["remoteTraceTarget"] = nlohmann::json();
if (! network.count("removeTraceLevel"))
network["remoteTraceLevel"] = 0;
if (! network.count("rulesSource"))
network["rulesSource"] = "";
if (! network.count("rules")) {
// If unspecified, rules are set to allow anything and behave like a flat L2 segment // If unspecified, rules are set to allow anything and behave like a flat L2 segment
network["rules"] = {{ network["rules"] = { { { "not", false }, { "or", false }, { "type", "ACTION_ACCEPT" } } };
{ "not",false },
{ "or", false },
{ "type","ACTION_ACCEPT" }
}};
} }
if (!network.count("dns")) network["dns"] = nlohmann::json::array(); if (! network.count("dns"))
if (!network.count("ssoEnabled")) network["ssoEnabled"] = false; network["dns"] = nlohmann::json::array();
if (!network.count("clientId")) network["clientId"] = ""; if (! network.count("ssoEnabled"))
if (!network.count("authorizationEndpoint")) network["authorizationEndpoint"] = ""; network["ssoEnabled"] = false;
if (! network.count("clientId"))
network["clientId"] = "";
if (! network.count("authorizationEndpoint"))
network["authorizationEndpoint"] = "";
network["objtype"] = "network"; network["objtype"] = "network";
} }
void DB::initMember(nlohmann::json &member) void DB::initMember(nlohmann::json& member)
{ {
if (!member.count("authorized")) member["authorized"] = false; if (! member.count("authorized"))
if (!member.count("ssoExempt")) member["ssoExempt"] = false; member["authorized"] = false;
if (!member.count("ipAssignments")) member["ipAssignments"] = nlohmann::json::array(); if (! member.count("ssoExempt"))
if (!member.count("activeBridge")) member["activeBridge"] = false; member["ssoExempt"] = false;
if (!member.count("tags")) member["tags"] = nlohmann::json::array(); if (! member.count("ipAssignments"))
if (!member.count("capabilities")) member["capabilities"] = nlohmann::json::array(); member["ipAssignments"] = nlohmann::json::array();
if (!member.count("creationTime")) member["creationTime"] = OSUtils::now(); if (! member.count("activeBridge"))
if (!member.count("noAutoAssignIps")) member["noAutoAssignIps"] = false; member["activeBridge"] = false;
if (!member.count("revision")) member["revision"] = 0ULL; if (! member.count("tags"))
if (!member.count("lastDeauthorizedTime")) member["lastDeauthorizedTime"] = 0ULL; member["tags"] = nlohmann::json::array();
if (!member.count("lastAuthorizedTime")) member["lastAuthorizedTime"] = 0ULL; if (! member.count("capabilities"))
if (!member.count("lastAuthorizedCredentialType")) member["lastAuthorizedCredentialType"] = nlohmann::json(); member["capabilities"] = nlohmann::json::array();
if (!member.count("lastAuthorizedCredential")) member["lastAuthorizedCredential"] = nlohmann::json(); if (! member.count("creationTime"))
if (!member.count("authenticationExpiryTime")) member["authenticationExpiryTime"] = 0LL; member["creationTime"] = OSUtils::now();
if (!member.count("vMajor")) member["vMajor"] = -1; if (! member.count("noAutoAssignIps"))
if (!member.count("vMinor")) member["vMinor"] = -1; member["noAutoAssignIps"] = false;
if (!member.count("vRev")) member["vRev"] = -1; if (! member.count("revision"))
if (!member.count("vProto")) member["vProto"] = -1; member["revision"] = 0ULL;
if (!member.count("remoteTraceTarget")) member["remoteTraceTarget"] = nlohmann::json(); if (! member.count("lastDeauthorizedTime"))
if (!member.count("removeTraceLevel")) member["remoteTraceLevel"] = 0; member["lastDeauthorizedTime"] = 0ULL;
if (! member.count("lastAuthorizedTime"))
member["lastAuthorizedTime"] = 0ULL;
if (! member.count("lastAuthorizedCredentialType"))
member["lastAuthorizedCredentialType"] = nlohmann::json();
if (! member.count("lastAuthorizedCredential"))
member["lastAuthorizedCredential"] = nlohmann::json();
if (! member.count("authenticationExpiryTime"))
member["authenticationExpiryTime"] = 0LL;
if (! member.count("vMajor"))
member["vMajor"] = -1;
if (! member.count("vMinor"))
member["vMinor"] = -1;
if (! member.count("vRev"))
member["vRev"] = -1;
if (! member.count("vProto"))
member["vProto"] = -1;
if (! member.count("remoteTraceTarget"))
member["remoteTraceTarget"] = nlohmann::json();
if (! member.count("removeTraceLevel"))
member["remoteTraceLevel"] = 0;
member["objtype"] = "member"; member["objtype"] = "member";
} }
void DB::cleanNetwork(nlohmann::json &network) void DB::cleanNetwork(nlohmann::json& network)
{ {
network.erase("clock"); network.erase("clock");
network.erase("authorizedMemberCount"); network.erase("authorizedMemberCount");
@ -91,7 +128,7 @@ void DB::cleanNetwork(nlohmann::json &network)
network.erase("lastModified"); network.erase("lastModified");
} }
void DB::cleanMember(nlohmann::json &member) void DB::cleanMember(nlohmann::json& member)
{ {
member.erase("clock"); member.erase("clock");
member.erase("physicalAddr"); member.erase("physicalAddr");
@ -102,10 +139,14 @@ void DB::cleanMember(nlohmann::json &member)
member.erase("authenticationClientID"); // computed member.erase("authenticationClientID"); // computed
} }
DB::DB() {} DB::DB()
DB::~DB() {} {
}
DB::~DB()
{
}
bool DB::get(const uint64_t networkId,nlohmann::json &network) bool DB::get(const uint64_t networkId, nlohmann::json& network)
{ {
waitForReady(); waitForReady();
Metrics::db_get_network++; Metrics::db_get_network++;
@ -124,7 +165,7 @@ bool DB::get(const uint64_t networkId,nlohmann::json &network)
return true; return true;
} }
bool DB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member) bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member)
{ {
waitForReady(); waitForReady();
Metrics::db_get_network_and_member++; Metrics::db_get_network_and_member++;
@ -147,7 +188,7 @@ bool DB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t mem
return true; return true;
} }
bool DB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,NetworkSummaryInfo &info) bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, NetworkSummaryInfo& info)
{ {
waitForReady(); waitForReady();
Metrics::db_get_network_and_member_and_summary++; Metrics::db_get_network_and_member_and_summary++;
@ -162,7 +203,7 @@ bool DB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t mem
{ {
std::shared_lock<std::shared_mutex> l2(nw->lock); std::shared_lock<std::shared_mutex> l2(nw->lock);
network = nw->config; network = nw->config;
_fillSummaryInfo(nw,info); _fillSummaryInfo(nw, info);
auto m = nw->members.find(memberId); auto m = nw->members.find(memberId);
if (m == nw->members.end()) if (m == nw->members.end())
return false; return false;
@ -171,7 +212,7 @@ bool DB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t mem
return true; return true;
} }
bool DB::get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members) bool DB::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members)
{ {
waitForReady(); waitForReady();
Metrics::db_get_member_list++; Metrics::db_get_member_list++;
@ -186,23 +227,23 @@ bool DB::get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohma
{ {
std::shared_lock<std::shared_mutex> l2(nw->lock); std::shared_lock<std::shared_mutex> l2(nw->lock);
network = nw->config; network = nw->config;
for(auto m=nw->members.begin();m!=nw->members.end();++m) { for (auto m = nw->members.begin(); m != nw->members.end(); ++m) {
members.push_back(m->second); members.push_back(m->second);
} }
} }
return true; return true;
} }
void DB::networks(std::set<uint64_t> &networks) void DB::networks(std::set<uint64_t>& networks)
{ {
waitForReady(); waitForReady();
Metrics::db_get_network_list++; Metrics::db_get_network_list++;
std::shared_lock<std::shared_mutex> l(_networks_l); std::shared_lock<std::shared_mutex> l(_networks_l);
for(auto n=_networks.begin();n!=_networks.end();++n) for (auto n = _networks.begin(); n != _networks.end(); ++n)
networks.insert(n->first); networks.insert(n->first);
} }
void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool notifyListeners) void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners)
{ {
Metrics::db_member_change++; Metrics::db_member_change++;
uint64_t memberId = 0; uint64_t memberId = 0;
@ -212,9 +253,9 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
std::shared_ptr<_Network> nw; std::shared_ptr<_Network> nw;
if (old.is_object()) { if (old.is_object()) {
memberId = OSUtils::jsonIntHex(old["id"],0ULL); memberId = OSUtils::jsonIntHex(old["id"], 0ULL);
networkId = OSUtils::jsonIntHex(old["nwid"],0ULL); networkId = OSUtils::jsonIntHex(old["nwid"], 0ULL);
if ((memberId)&&(networkId)) { if ((memberId) && (networkId)) {
{ {
std::unique_lock<std::shared_mutex> l(_networks_l); std::unique_lock<std::shared_mutex> l(_networks_l);
auto nw2 = _networks.find(networkId); auto nw2 = _networks.find(networkId);
@ -224,17 +265,17 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
} }
if (nw) { if (nw) {
std::unique_lock<std::shared_mutex> l(nw->lock); std::unique_lock<std::shared_mutex> l(nw->lock);
if (OSUtils::jsonBool(old["activeBridge"],false)) { if (OSUtils::jsonBool(old["activeBridge"], false)) {
nw->activeBridgeMembers.erase(memberId); nw->activeBridgeMembers.erase(memberId);
} }
wasAuth = OSUtils::jsonBool(old["authorized"],false); wasAuth = OSUtils::jsonBool(old["authorized"], false);
if (wasAuth) { if (wasAuth) {
nw->authorizedMembers.erase(memberId); nw->authorizedMembers.erase(memberId);
} }
json &ips = old["ipAssignments"]; json& ips = old["ipAssignments"];
if (ips.is_array()) { if (ips.is_array()) {
for(unsigned long i=0;i<ips.size();++i) { for (unsigned long i = 0; i < ips.size(); ++i) {
json &ipj = ips[i]; json& ipj = ips[i];
if (ipj.is_string()) { if (ipj.is_string()) {
const std::string ips = ipj; const std::string ips = ipj;
InetAddress ipa(ips.c_str()); InetAddress ipa(ips.c_str());
@ -248,14 +289,14 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
} }
if (memberConfig.is_object()) { if (memberConfig.is_object()) {
if (!nw) { if (! nw) {
memberId = OSUtils::jsonIntHex(memberConfig["id"],0ULL); memberId = OSUtils::jsonIntHex(memberConfig["id"], 0ULL);
networkId = OSUtils::jsonIntHex(memberConfig["nwid"],0ULL); networkId = OSUtils::jsonIntHex(memberConfig["nwid"], 0ULL);
if ((!memberId)||(!networkId)) if ((! memberId) || (! networkId))
return; return;
std::unique_lock<std::shared_mutex> l(_networks_l); std::unique_lock<std::shared_mutex> l(_networks_l);
std::shared_ptr<_Network> &nw2 = _networks[networkId]; std::shared_ptr<_Network>& nw2 = _networks[networkId];
if (!nw2) if (! nw2)
nw2.reset(new _Network); nw2.reset(new _Network);
nw = nw2; nw = nw2;
} }
@ -265,18 +306,18 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
nw->members[memberId] = memberConfig; nw->members[memberId] = memberConfig;
if (OSUtils::jsonBool(memberConfig["activeBridge"],false)) { if (OSUtils::jsonBool(memberConfig["activeBridge"], false)) {
nw->activeBridgeMembers.insert(memberId); nw->activeBridgeMembers.insert(memberId);
} }
isAuth = OSUtils::jsonBool(memberConfig["authorized"],false); isAuth = OSUtils::jsonBool(memberConfig["authorized"], false);
if (isAuth) { if (isAuth) {
Metrics::member_auths++; Metrics::member_auths++;
nw->authorizedMembers.insert(memberId); nw->authorizedMembers.insert(memberId);
} }
json &ips = memberConfig["ipAssignments"]; json& ips = memberConfig["ipAssignments"];
if (ips.is_array()) { if (ips.is_array()) {
for(unsigned long i=0;i<ips.size();++i) { for (unsigned long i = 0; i < ips.size(); ++i) {
json &ipj = ips[i]; json& ipj = ips[i];
if (ipj.is_string()) { if (ipj.is_string()) {
const std::string ips = ipj; const std::string ips = ipj;
InetAddress ipa(ips.c_str()); InetAddress ipa(ips.c_str());
@ -286,8 +327,8 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
} }
} }
if (!isAuth) { if (! isAuth) {
const int64_t ldt = (int64_t)OSUtils::jsonInt(memberConfig["lastDeauthorizedTime"],0ULL); const int64_t ldt = (int64_t)OSUtils::jsonInt(memberConfig["lastDeauthorizedTime"], 0ULL);
if (ldt > nw->mostRecentDeauthTime) if (ldt > nw->mostRecentDeauthTime)
nw->mostRecentDeauthTime = ldt; nw->mostRecentDeauthTime = ldt;
} }
@ -295,11 +336,12 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
if (notifyListeners) { if (notifyListeners) {
std::unique_lock<std::shared_mutex> ll(_changeListeners_l); std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
for(auto i=_changeListeners.begin();i!=_changeListeners.end();++i) { for (auto i = _changeListeners.begin(); i != _changeListeners.end(); ++i) {
(*i)->onNetworkMemberUpdate(this,networkId,memberId,memberConfig); (*i)->onNetworkMemberUpdate(this, networkId, memberId, memberConfig);
} }
} }
} else if (memberId) { }
else if (memberId) {
if (nw) { if (nw) {
std::unique_lock<std::shared_mutex> l(nw->lock); std::unique_lock<std::shared_mutex> l(nw->lock);
nw->members.erase(memberId); nw->members.erase(memberId);
@ -307,7 +349,7 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
if (networkId) { if (networkId) {
std::unique_lock<std::shared_mutex> l(_networks_l); std::unique_lock<std::shared_mutex> l(_networks_l);
auto er = _networkByMember.equal_range(memberId); auto er = _networkByMember.equal_range(memberId);
for(auto i=er.first;i!=er.second;++i) { for (auto i = er.first; i != er.second; ++i) {
if (i->second == networkId) { if (i->second == networkId) {
_networkByMember.erase(i); _networkByMember.erase(i);
break; break;
@ -317,40 +359,45 @@ void DB::_memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool no
} }
if (notifyListeners) { if (notifyListeners) {
if(networkId != 0 && memberId != 0 && old.is_object() && !memberConfig.is_object()) { if (networkId != 0 && memberId != 0 && old.is_object() && ! memberConfig.is_object()) {
// member delete // member delete
Metrics::member_count--; Metrics::member_count--;
} else if (networkId != 0 && memberId != 0 && !old.is_object() && memberConfig.is_object()) { }
else if (networkId != 0 && memberId != 0 && ! old.is_object() && memberConfig.is_object()) {
// new member // new member
Metrics::member_count++; Metrics::member_count++;
} }
if (!wasAuth && isAuth) { if (! wasAuth && isAuth) {
Metrics::member_auths++; Metrics::member_auths++;
} else if (wasAuth && !isAuth) { }
else if (wasAuth && ! isAuth) {
Metrics::member_deauths++; Metrics::member_deauths++;
} else { }
else {
Metrics::member_changes++; Metrics::member_changes++;
} }
} }
if ((notifyListeners)&&((wasAuth)&&(!isAuth)&&(networkId)&&(memberId))) { if ((notifyListeners) && ((wasAuth) && (! isAuth) && (networkId) && (memberId))) {
std::unique_lock<std::shared_mutex> ll(_changeListeners_l); std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
for(auto i=_changeListeners.begin();i!=_changeListeners.end();++i) { for (auto i = _changeListeners.begin(); i != _changeListeners.end(); ++i) {
(*i)->onNetworkMemberDeauthorize(this,networkId,memberId); (*i)->onNetworkMemberDeauthorize(this, networkId, memberId);
} }
} }
} }
void DB::_networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool notifyListeners) void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners)
{ {
Metrics::db_network_change++; Metrics::db_network_change++;
if (notifyListeners) { if (notifyListeners) {
if (old.is_object() && old.contains("id") && networkConfig.is_object() && networkConfig.contains("id")) { if (old.is_object() && old.contains("id") && networkConfig.is_object() && networkConfig.contains("id")) {
Metrics::network_changes++; Metrics::network_changes++;
} else if (!old.is_object() && networkConfig.is_object() && networkConfig.contains("id")) { }
else if (! old.is_object() && networkConfig.is_object() && networkConfig.contains("id")) {
Metrics::network_count++; Metrics::network_count++;
} else if (old.is_object() && old.contains("id") && !networkConfig.is_object()) { }
else if (old.is_object() && old.contains("id") && ! networkConfig.is_object()) {
Metrics::network_count--; Metrics::network_count--;
} }
} }
@ -362,8 +409,8 @@ void DB::_networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool
std::shared_ptr<_Network> nw; std::shared_ptr<_Network> nw;
{ {
std::unique_lock<std::shared_mutex> l(_networks_l); std::unique_lock<std::shared_mutex> l(_networks_l);
std::shared_ptr<_Network> &nw2 = _networks[networkId]; std::shared_ptr<_Network>& nw2 = _networks[networkId];
if (!nw2) if (! nw2)
nw2.reset(new _Network); nw2.reset(new _Network);
nw = nw2; nw = nw2;
} }
@ -373,12 +420,13 @@ void DB::_networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool
} }
if (notifyListeners) { if (notifyListeners) {
std::unique_lock<std::shared_mutex> ll(_changeListeners_l); std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
for(auto i=_changeListeners.begin();i!=_changeListeners.end();++i) { for (auto i = _changeListeners.begin(); i != _changeListeners.end(); ++i) {
(*i)->onNetworkUpdate(this,networkId,networkConfig); (*i)->onNetworkUpdate(this, networkId, networkConfig);
} }
} }
} }
} else if (old.is_object()) { }
else if (old.is_object()) {
const std::string ids = old["id"]; const std::string ids = old["id"];
const uint64_t networkId = Utils::hexStrToU64(ids.c_str()); const uint64_t networkId = Utils::hexStrToU64(ids.c_str());
if (networkId) { if (networkId) {
@ -387,15 +435,16 @@ void DB::_networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool
nlohmann::json network; nlohmann::json network;
std::vector<nlohmann::json> members; std::vector<nlohmann::json> members;
this->get(networkId, network, members); this->get(networkId, network, members);
for(auto i=members.begin();i!=members.end();++i) { for (auto i = members.begin(); i != members.end(); ++i) {
const std::string nodeID = (*i)["id"]; const std::string nodeID = (*i)["id"];
const uint64_t memberId = Utils::hexStrToU64(nodeID.c_str()); const uint64_t memberId = Utils::hexStrToU64(nodeID.c_str());
std::unique_lock<std::shared_mutex> ll(_changeListeners_l); std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
for(auto j=_changeListeners.begin();j!=_changeListeners.end();++j) { for (auto j = _changeListeners.begin(); j != _changeListeners.end(); ++j) {
(*j)->onNetworkMemberDeauthorize(this,networkId,memberId); (*j)->onNetworkMemberDeauthorize(this, networkId, memberId);
} }
} }
} catch (std::exception &e) { }
catch (std::exception& e) {
std::cerr << "Error deauthorizing members on network delete: " << e.what() << std::endl; std::cerr << "Error deauthorizing members on network delete: " << e.what() << std::endl;
} }
@ -406,14 +455,14 @@ void DB::_networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool
} }
} }
void DB::_fillSummaryInfo(const std::shared_ptr<_Network> &nw,NetworkSummaryInfo &info) void DB::_fillSummaryInfo(const std::shared_ptr<_Network>& nw, NetworkSummaryInfo& info)
{ {
for(auto ab=nw->activeBridgeMembers.begin();ab!=nw->activeBridgeMembers.end();++ab) for (auto ab = nw->activeBridgeMembers.begin(); ab != nw->activeBridgeMembers.end(); ++ab)
info.activeBridges.push_back(Address(*ab)); info.activeBridges.push_back(Address(*ab));
std::sort(info.activeBridges.begin(),info.activeBridges.end()); std::sort(info.activeBridges.begin(), info.activeBridges.end());
for(auto ip=nw->allocatedIps.begin();ip!=nw->allocatedIps.end();++ip) for (auto ip = nw->allocatedIps.begin(); ip != nw->allocatedIps.end(); ++ip)
info.allocatedIps.push_back(*ip); info.allocatedIps.push_back(*ip);
std::sort(info.allocatedIps.begin(),info.allocatedIps.end()); std::sort(info.allocatedIps.begin(), info.allocatedIps.end());
info.authorizedMemberCount = (unsigned long)nw->authorizedMembers.size(); info.authorizedMemberCount = (unsigned long)nw->authorizedMembers.size();
info.totalMemberCount = (unsigned long)nw->members.size(); info.totalMemberCount = (unsigned long)nw->members.size();
info.mostRecentDeauthTime = nw->mostRecentDeauthTime; info.mostRecentDeauthTime = nw->mostRecentDeauthTime;

145
controller/DB.hpp
View file

@ -14,49 +14,36 @@
#ifndef ZT_CONTROLLER_DB_HPP #ifndef ZT_CONTROLLER_DB_HPP
#define ZT_CONTROLLER_DB_HPP #define ZT_CONTROLLER_DB_HPP
//#define ZT_CONTROLLER_USE_LIBPQ // #define ZT_CONTROLLER_USE_LIBPQ
#include "../node/Constants.hpp" #include "../node/Constants.hpp"
#include "../node/Identity.hpp" #include "../node/Identity.hpp"
#include "../node/InetAddress.hpp" #include "../node/InetAddress.hpp"
#include "../osdep/OSUtils.hpp"
#include "../osdep/BlockingQueue.hpp" #include "../osdep/BlockingQueue.hpp"
#include "../osdep/OSUtils.hpp"
#include <atomic>
#include <map>
#include <memory> #include <memory>
#include <nlohmann/json.hpp>
#include <prometheus/simpleapi.h>
#include <set>
#include <shared_mutex>
#include <string> #include <string>
#include <thread> #include <thread>
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
#include <atomic>
#include <shared_mutex>
#include <set>
#include <map>
#include <nlohmann/json.hpp>
#include <prometheus/simpleapi.h>
#define ZT_MEMBER_AUTH_TIMEOUT_NOTIFY_BEFORE 25000 #define ZT_MEMBER_AUTH_TIMEOUT_NOTIFY_BEFORE 25000
namespace ZeroTier namespace ZeroTier {
{
struct AuthInfo struct AuthInfo {
{ public:
public: AuthInfo() : enabled(false), version(0), authenticationURL(), authenticationExpiryTime(0), issuerURL(), centralAuthURL(), ssoNonce(), ssoState(), ssoClientID(), ssoProvider("default")
AuthInfo() {
: enabled(false) }
, version(0)
, authenticationURL()
, authenticationExpiryTime(0)
, issuerURL()
, centralAuthURL()
, ssoNonce()
, ssoState()
, ssoClientID()
, ssoProvider("default")
{}
bool enabled; bool enabled;
uint64_t version; uint64_t version;
@ -73,22 +60,31 @@ public:
/** /**
* Base class with common infrastructure for all controller DB implementations * Base class with common infrastructure for all controller DB implementations
*/ */
class DB class DB {
{
public:
class ChangeListener
{
public: public:
ChangeListener() {} class ChangeListener {
virtual ~ChangeListener() {} public:
virtual void onNetworkUpdate(const void *db,uint64_t networkId,const nlohmann::json &network) {} ChangeListener()
virtual void onNetworkMemberUpdate(const void *db,uint64_t networkId,uint64_t memberId,const nlohmann::json &member) {} {
virtual void onNetworkMemberDeauthorize(const void *db,uint64_t networkId,uint64_t memberId) {} }
virtual ~ChangeListener()
{
}
virtual void onNetworkUpdate(const void* db, uint64_t networkId, const nlohmann::json& network)
{
}
virtual void onNetworkMemberUpdate(const void* db, uint64_t networkId, uint64_t memberId, const nlohmann::json& member)
{
}
virtual void onNetworkMemberDeauthorize(const void* db, uint64_t networkId, uint64_t memberId)
{
}
}; };
struct NetworkSummaryInfo struct NetworkSummaryInfo {
NetworkSummaryInfo() : authorizedMemberCount(0), totalMemberCount(0), mostRecentDeauthTime(0)
{ {
NetworkSummaryInfo() : authorizedMemberCount(0),totalMemberCount(0),mostRecentDeauthTime(0) {} }
std::vector<Address> activeBridges; std::vector<Address> activeBridges;
std::vector<InetAddress> allocatedIps; std::vector<InetAddress> allocatedIps;
unsigned long authorizedMemberCount; unsigned long authorizedMemberCount;
@ -96,10 +92,10 @@ public:
int64_t mostRecentDeauthTime; int64_t mostRecentDeauthTime;
}; };
static void initNetwork(nlohmann::json &network); static void initNetwork(nlohmann::json& network);
static void initMember(nlohmann::json &member); static void initMember(nlohmann::json& member);
static void cleanNetwork(nlohmann::json &network); static void cleanNetwork(nlohmann::json& network);
static void cleanMember(nlohmann::json &member); static void cleanMember(nlohmann::json& member);
DB(); DB();
virtual ~DB(); virtual ~DB();
@ -113,41 +109,43 @@ public:
return (_networks.find(networkId) != _networks.end()); return (_networks.find(networkId) != _networks.end());
} }
bool get(const uint64_t networkId,nlohmann::json &network); bool get(const uint64_t networkId, nlohmann::json& network);
bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member); bool get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member);
bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,NetworkSummaryInfo &info); bool get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, NetworkSummaryInfo& info);
bool get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members); bool get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members);
void networks(std::set<uint64_t> &networks); void networks(std::set<uint64_t>& networks);
template<typename F> template <typename F> inline void each(F f)
inline void each(F f)
{ {
nlohmann::json nullJson; nlohmann::json nullJson;
std::unique_lock<std::shared_mutex> lck(_networks_l); std::unique_lock<std::shared_mutex> lck(_networks_l);
for(auto nw=_networks.begin();nw!=_networks.end();++nw) { for (auto nw = _networks.begin(); nw != _networks.end(); ++nw) {
f(nw->first,nw->second->config,0,nullJson); // first provide network with 0 for member ID f(nw->first, nw->second->config, 0, nullJson); // first provide network with 0 for member ID
for(auto m=nw->second->members.begin();m!=nw->second->members.end();++m) { for (auto m = nw->second->members.begin(); m != nw->second->members.end(); ++m) {
f(nw->first,nw->second->config,m->first,m->second); f(nw->first, nw->second->config, m->first, m->second);
} }
} }
} }
virtual bool save(nlohmann::json &record,bool notifyListeners) = 0; virtual bool save(nlohmann::json& record, bool notifyListeners) = 0;
virtual void eraseNetwork(const uint64_t networkId) = 0; virtual void eraseNetwork(const uint64_t networkId) = 0;
virtual void eraseMember(const uint64_t networkId,const uint64_t memberId) = 0; virtual void eraseMember(const uint64_t networkId, const uint64_t memberId) = 0;
virtual void nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress) = 0; virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress) = 0;
virtual AuthInfo getSSOAuthInfo(const nlohmann::json &member, const std::string &redirectURL) { return AuthInfo(); } virtual AuthInfo getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL)
{
return AuthInfo();
}
inline void addListener(DB::ChangeListener *const listener) inline void addListener(DB::ChangeListener* const listener)
{ {
std::unique_lock<std::shared_mutex> l(_changeListeners_l); std::unique_lock<std::shared_mutex> l(_changeListeners_l);
_changeListeners.push_back(listener); _changeListeners.push_back(listener);
} }
protected: protected:
static inline bool _compareRecords(const nlohmann::json &a,const nlohmann::json &b) static inline bool _compareRecords(const nlohmann::json& a, const nlohmann::json& b)
{ {
if (a.is_object() == b.is_object()) { if (a.is_object() == b.is_object()) {
if (a.is_object()) { if (a.is_object()) {
@ -155,10 +153,10 @@ protected:
return false; return false;
auto amap = a.get<nlohmann::json::object_t>(); auto amap = a.get<nlohmann::json::object_t>();
auto bmap = b.get<nlohmann::json::object_t>(); auto bmap = b.get<nlohmann::json::object_t>();
for(auto ai=amap.begin();ai!=amap.end();++ai) { for (auto ai = amap.begin(); ai != amap.end(); ++ai) {
if (ai->first != "revision") { // ignore revision, compare only non-revision-counter fields if (ai->first != "revision") { // ignore revision, compare only non-revision-counter fields
auto bi = bmap.find(ai->first); auto bi = bmap.find(ai->first);
if ((bi == bmap.end())||(bi->second != ai->second)) if ((bi == bmap.end()) || (bi->second != ai->second))
return false; return false;
} }
} }
@ -169,25 +167,26 @@ protected:
return false; return false;
} }
struct _Network struct _Network {
_Network() : mostRecentDeauthTime(0)
{ {
_Network() : mostRecentDeauthTime(0) {} }
nlohmann::json config; nlohmann::json config;
std::unordered_map<uint64_t,nlohmann::json> members; std::unordered_map<uint64_t, nlohmann::json> members;
std::unordered_set<uint64_t> activeBridgeMembers; std::unordered_set<uint64_t> activeBridgeMembers;
std::unordered_set<uint64_t> authorizedMembers; std::unordered_set<uint64_t> authorizedMembers;
std::unordered_set<InetAddress,InetAddress::Hasher> allocatedIps; std::unordered_set<InetAddress, InetAddress::Hasher> allocatedIps;
int64_t mostRecentDeauthTime; int64_t mostRecentDeauthTime;
std::shared_mutex lock; std::shared_mutex lock;
}; };
virtual void _memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool notifyListeners); virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners);
virtual void _networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool notifyListeners); virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners);
void _fillSummaryInfo(const std::shared_ptr<_Network> &nw,NetworkSummaryInfo &info); void _fillSummaryInfo(const std::shared_ptr<_Network>& nw, NetworkSummaryInfo& info);
std::vector<DB::ChangeListener *> _changeListeners; std::vector<DB::ChangeListener*> _changeListeners;
std::unordered_map< uint64_t,std::shared_ptr<_Network> > _networks; std::unordered_map<uint64_t, std::shared_ptr<_Network> > _networks;
std::unordered_multimap< uint64_t,uint64_t > _networkByMember; std::unordered_multimap<uint64_t, uint64_t> _networkByMember;
mutable std::shared_mutex _changeListeners_l; mutable std::shared_mutex _changeListeners_l;
mutable std::shared_mutex _networks_l; mutable std::shared_mutex _networks_l;
}; };

129
controller/DBMirrorSet.cpp
View file

@ -15,22 +15,17 @@
namespace ZeroTier { namespace ZeroTier {
DBMirrorSet::DBMirrorSet(DB::ChangeListener *listener) DBMirrorSet::DBMirrorSet(DB::ChangeListener* listener) : _listener(listener), _running(true), _syncCheckerThread(), _dbs(), _dbs_l()
: _listener(listener)
, _running(true)
, _syncCheckerThread()
, _dbs()
, _dbs_l()
{ {
_syncCheckerThread = std::thread([this]() { _syncCheckerThread = std::thread([this]() {
for(;;) { for (;;) {
for(int i=0;i<120;++i) { // 1 minute delay between checks for (int i = 0; i < 120; ++i) { // 1 minute delay between checks
if (!_running) if (! _running)
return; return;
std::this_thread::sleep_for(std::chrono::milliseconds(500)); std::this_thread::sleep_for(std::chrono::milliseconds(500));
} }
std::vector< std::shared_ptr<DB> > dbs; std::vector<std::shared_ptr<DB> > dbs;
{ {
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
if (_dbs.size() <= 1) if (_dbs.size() <= 1)
@ -38,33 +33,36 @@ DBMirrorSet::DBMirrorSet(DB::ChangeListener *listener)
dbs = _dbs; dbs = _dbs;
} }
for(auto db=dbs.begin();db!=dbs.end();++db) { for (auto db = dbs.begin(); db != dbs.end(); ++db) {
(*db)->each([&dbs,&db](uint64_t networkId,const nlohmann::json &network,uint64_t memberId,const nlohmann::json &member) { (*db)->each([&dbs, &db](uint64_t networkId, const nlohmann::json& network, uint64_t memberId, const nlohmann::json& member) {
try { try {
if (network.is_object()) { if (network.is_object()) {
if (memberId == 0) { if (memberId == 0) {
for(auto db2=dbs.begin();db2!=dbs.end();++db2) { for (auto db2 = dbs.begin(); db2 != dbs.end(); ++db2) {
if (db->get() != db2->get()) { if (db->get() != db2->get()) {
nlohmann::json nw2; nlohmann::json nw2;
if ((!(*db2)->get(networkId,nw2))||((nw2.is_object())&&(OSUtils::jsonInt(nw2["revision"],0) < OSUtils::jsonInt(network["revision"],0)))) { if ((! (*db2)->get(networkId, nw2)) || ((nw2.is_object()) && (OSUtils::jsonInt(nw2["revision"], 0) < OSUtils::jsonInt(network["revision"], 0)))) {
nw2 = network; nw2 = network;
(*db2)->save(nw2,false); (*db2)->save(nw2, false);
} }
} }
} }
} else if (member.is_object()) { }
for(auto db2=dbs.begin();db2!=dbs.end();++db2) { else if (member.is_object()) {
for (auto db2 = dbs.begin(); db2 != dbs.end(); ++db2) {
if (db->get() != db2->get()) { if (db->get() != db2->get()) {
nlohmann::json nw2,m2; nlohmann::json nw2, m2;
if ((!(*db2)->get(networkId,nw2,memberId,m2))||((m2.is_object())&&(OSUtils::jsonInt(m2["revision"],0) < OSUtils::jsonInt(member["revision"],0)))) { if ((! (*db2)->get(networkId, nw2, memberId, m2)) || ((m2.is_object()) && (OSUtils::jsonInt(m2["revision"], 0) < OSUtils::jsonInt(member["revision"], 0)))) {
m2 = member; m2 = member;
(*db2)->save(m2,false); (*db2)->save(m2, false);
} }
} }
} }
} }
} }
} catch ( ... ) {} // skip entries that generate JSON errors }
catch (...) {
} // skip entries that generate JSON errors
}); });
} }
} }
@ -80,58 +78,58 @@ DBMirrorSet::~DBMirrorSet()
bool DBMirrorSet::hasNetwork(const uint64_t networkId) const bool DBMirrorSet::hasNetwork(const uint64_t networkId) const
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->hasNetwork(networkId)) if ((*d)->hasNetwork(networkId))
return true; return true;
} }
return false; return false;
} }
bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId,network)) { if ((*d)->get(networkId, network)) {
return true; return true;
} }
} }
return false; return false;
} }
bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId,network,memberId,member)) if ((*d)->get(networkId, network, memberId, member))
return true; return true;
} }
return false; return false;
} }
bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,DB::NetworkSummaryInfo &info) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, DB::NetworkSummaryInfo& info)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId,network,memberId,member,info)) if ((*d)->get(networkId, network, memberId, member, info))
return true; return true;
} }
return false; return false;
} }
bool DBMirrorSet::get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members) bool DBMirrorSet::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if ((*d)->get(networkId,network,members)) if ((*d)->get(networkId, network, members))
return true; return true;
} }
return false; return false;
} }
AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json &member, const std::string &redirectURL) AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
AuthInfo info = (*d)->getSSOAuthInfo(member, redirectURL); AuthInfo info = (*d)->getSSOAuthInfo(member, redirectURL);
if (info.enabled) { if (info.enabled) {
return info; return info;
@ -140,10 +138,10 @@ AuthInfo DBMirrorSet::getSSOAuthInfo(const nlohmann::json &member, const std::st
return AuthInfo(); return AuthInfo();
} }
void DBMirrorSet::networks(std::set<uint64_t> &networks) void DBMirrorSet::networks(std::set<uint64_t>& networks)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->networks(networks); (*d)->networks(networks);
} }
} }
@ -152,7 +150,7 @@ bool DBMirrorSet::waitForReady()
{ {
bool r = false; bool r = false;
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
r |= (*d)->waitForReady(); r |= (*d)->waitForReady();
} }
return r; return r;
@ -161,30 +159,31 @@ bool DBMirrorSet::waitForReady()
bool DBMirrorSet::isReady() bool DBMirrorSet::isReady()
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if (!(*d)->isReady()) if (! (*d)->isReady())
return false; return false;
} }
return true; return true;
} }
bool DBMirrorSet::save(nlohmann::json &record,bool notifyListeners) bool DBMirrorSet::save(nlohmann::json& record, bool notifyListeners)
{ {
std::vector< std::shared_ptr<DB> > dbs; std::vector<std::shared_ptr<DB> > dbs;
{ {
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
dbs = _dbs; dbs = _dbs;
} }
if (notifyListeners) { if (notifyListeners) {
for(auto d=dbs.begin();d!=dbs.end();++d) { for (auto d = dbs.begin(); d != dbs.end(); ++d) {
if ((*d)->save(record,true)) if ((*d)->save(record, true))
return true; return true;
} }
return false; return false;
} else { }
else {
bool modified = false; bool modified = false;
for(auto d=dbs.begin();d!=dbs.end();++d) { for (auto d = dbs.begin(); d != dbs.end(); ++d) {
modified |= (*d)->save(record,false); modified |= (*d)->save(record, false);
} }
return modified; return modified;
} }
@ -193,54 +192,54 @@ bool DBMirrorSet::save(nlohmann::json &record,bool notifyListeners)
void DBMirrorSet::eraseNetwork(const uint64_t networkId) void DBMirrorSet::eraseNetwork(const uint64_t networkId)
{ {
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->eraseNetwork(networkId); (*d)->eraseNetwork(networkId);
} }
} }
void DBMirrorSet::eraseMember(const uint64_t networkId,const uint64_t memberId) void DBMirrorSet::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->eraseMember(networkId,memberId); (*d)->eraseMember(networkId, memberId);
} }
} }
void DBMirrorSet::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress) void DBMirrorSet::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress)
{ {
std::shared_lock<std::shared_mutex> l(_dbs_l); std::shared_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
(*d)->nodeIsOnline(networkId,memberId,physicalAddress); (*d)->nodeIsOnline(networkId, memberId, physicalAddress);
} }
} }
void DBMirrorSet::onNetworkUpdate(const void *db,uint64_t networkId,const nlohmann::json &network) void DBMirrorSet::onNetworkUpdate(const void* db, uint64_t networkId, const nlohmann::json& network)
{ {
nlohmann::json record(network); nlohmann::json record(network);
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if (d->get() != db) { if (d->get() != db) {
(*d)->save(record,false); (*d)->save(record, false);
} }
} }
_listener->onNetworkUpdate(this,networkId,network); _listener->onNetworkUpdate(this, networkId, network);
} }
void DBMirrorSet::onNetworkMemberUpdate(const void *db,uint64_t networkId,uint64_t memberId,const nlohmann::json &member) void DBMirrorSet::onNetworkMemberUpdate(const void* db, uint64_t networkId, uint64_t memberId, const nlohmann::json& member)
{ {
nlohmann::json record(member); nlohmann::json record(member);
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
for(auto d=_dbs.begin();d!=_dbs.end();++d) { for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
if (d->get() != db) { if (d->get() != db) {
(*d)->save(record,false); (*d)->save(record, false);
} }
} }
_listener->onNetworkMemberUpdate(this,networkId,memberId,member); _listener->onNetworkMemberUpdate(this, networkId, memberId, member);
} }
void DBMirrorSet::onNetworkMemberDeauthorize(const void *db,uint64_t networkId,uint64_t memberId) void DBMirrorSet::onNetworkMemberDeauthorize(const void* db, uint64_t networkId, uint64_t memberId)
{ {
_listener->onNetworkMemberDeauthorize(this,networkId,memberId); _listener->onNetworkMemberDeauthorize(this, networkId, memberId);
} }
} // namespace ZeroTier } // namespace ZeroTier

43
controller/DBMirrorSet.hpp
View file

@ -16,55 +16,54 @@
#include "DB.hpp" #include "DB.hpp"
#include <vector>
#include <memory> #include <memory>
#include <shared_mutex>
#include <set> #include <set>
#include <shared_mutex>
#include <thread> #include <thread>
#include <vector>
namespace ZeroTier { namespace ZeroTier {
class DBMirrorSet : public DB::ChangeListener class DBMirrorSet : public DB::ChangeListener {
{ public:
public: DBMirrorSet(DB::ChangeListener* listener);
DBMirrorSet(DB::ChangeListener *listener);
virtual ~DBMirrorSet(); virtual ~DBMirrorSet();
bool hasNetwork(const uint64_t networkId) const; bool hasNetwork(const uint64_t networkId) const;
bool get(const uint64_t networkId,nlohmann::json &network); bool get(const uint64_t networkId, nlohmann::json& network);
bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member); bool get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member);
bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,DB::NetworkSummaryInfo &info); bool get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member, DB::NetworkSummaryInfo& info);
bool get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members); bool get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members);
void networks(std::set<uint64_t> &networks); void networks(std::set<uint64_t>& networks);
bool waitForReady(); bool waitForReady();
bool isReady(); bool isReady();
bool save(nlohmann::json &record,bool notifyListeners); bool save(nlohmann::json& record, bool notifyListeners);
void eraseNetwork(const uint64_t networkId); void eraseNetwork(const uint64_t networkId);
void eraseMember(const uint64_t networkId,const uint64_t memberId); void eraseMember(const uint64_t networkId, const uint64_t memberId);
void nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress); void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress);
// These are called by various DB instances when changes occur. // These are called by various DB instances when changes occur.
virtual void onNetworkUpdate(const void *db,uint64_t networkId,const nlohmann::json &network); virtual void onNetworkUpdate(const void* db, uint64_t networkId, const nlohmann::json& network);
virtual void onNetworkMemberUpdate(const void *db,uint64_t networkId,uint64_t memberId,const nlohmann::json &member); virtual void onNetworkMemberUpdate(const void* db, uint64_t networkId, uint64_t memberId, const nlohmann::json& member);
virtual void onNetworkMemberDeauthorize(const void *db,uint64_t networkId,uint64_t memberId); virtual void onNetworkMemberDeauthorize(const void* db, uint64_t networkId, uint64_t memberId);
AuthInfo getSSOAuthInfo(const nlohmann::json &member, const std::string &redirectURL); AuthInfo getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL);
inline void addDB(const std::shared_ptr<DB> &db) inline void addDB(const std::shared_ptr<DB>& db)
{ {
db->addListener(this); db->addListener(this);
std::unique_lock<std::shared_mutex> l(_dbs_l); std::unique_lock<std::shared_mutex> l(_dbs_l);
_dbs.push_back(db); _dbs.push_back(db);
} }
private: private:
DB::ChangeListener *const _listener; DB::ChangeListener* const _listener;
std::atomic_bool _running; std::atomic_bool _running;
std::thread _syncCheckerThread; std::thread _syncCheckerThread;
std::vector< std::shared_ptr< DB > > _dbs; std::vector<std::shared_ptr<DB> > _dbs;
mutable std::shared_mutex _dbs_l; mutable std::shared_mutex _dbs_l;
}; };

1296
controller/EmbeddedNetworkController.cpp
View file

File diff suppressed because it is too large Load diff

131
controller/EmbeddedNetworkController.hpp
View file

@ -14,112 +14,109 @@
#ifndef ZT_SQLITENETWORKCONTROLLER_HPP #ifndef ZT_SQLITENETWORKCONTROLLER_HPP
#define ZT_SQLITENETWORKCONTROLLER_HPP #define ZT_SQLITENETWORKCONTROLLER_HPP
#include <stdint.h> #include "../node/Address.hpp"
#include <string>
#include <map>
#include <vector>
#include <set>
#include <list>
#include <thread>
#include <unordered_map>
#include <atomic>
#include "../node/Constants.hpp" #include "../node/Constants.hpp"
#include "../node/InetAddress.hpp"
#include "../node/NetworkController.hpp" #include "../node/NetworkController.hpp"
#include "../node/Utils.hpp" #include "../node/Utils.hpp"
#include "../node/Address.hpp" #include "../osdep/BlockingQueue.hpp"
#include "../node/InetAddress.hpp"
#include "../osdep/OSUtils.hpp" #include "../osdep/OSUtils.hpp"
#include "../osdep/Thread.hpp" #include "../osdep/Thread.hpp"
#include "../osdep/BlockingQueue.hpp"
#include <nlohmann/json.hpp>
#include <cpp-httplib/httplib.h>
#include "DB.hpp" #include "DB.hpp"
#include "DBMirrorSet.hpp" #include "DBMirrorSet.hpp"
#include <atomic>
#include <cpp-httplib/httplib.h>
#include <list>
#include <map>
#include <nlohmann/json.hpp>
#include <set>
#include <stdint.h>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
namespace ZeroTier { namespace ZeroTier {
class Node; class Node;
struct RedisConfig; struct RedisConfig;
class EmbeddedNetworkController : public NetworkController,public DB::ChangeListener class EmbeddedNetworkController
{ : public NetworkController
public: , public DB::ChangeListener {
public:
/** /**
* @param node Parent node * @param node Parent node
* @param dbPath Database path (file path or database credentials) * @param dbPath Database path (file path or database credentials)
*/ */
EmbeddedNetworkController(Node *node,const char *ztPath,const char *dbPath, int listenPort, RedisConfig *rc); EmbeddedNetworkController(Node* node, const char* ztPath, const char* dbPath, int listenPort, RedisConfig* rc);
virtual ~EmbeddedNetworkController(); virtual ~EmbeddedNetworkController();
virtual void init(const Identity &signingId,Sender *sender); virtual void init(const Identity& signingId, Sender* sender);
void setSSORedirectURL(const std::string &url); void setSSORedirectURL(const std::string& url);
virtual void request( virtual void request(uint64_t nwid, const InetAddress& fromAddr, uint64_t requestPacketId, const Identity& identity, const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY>& metaData);
uint64_t nwid,
const InetAddress &fromAddr,
uint64_t requestPacketId,
const Identity &identity,
const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData);
void configureHTTPControlPlane( void configureHTTPControlPlane(httplib::Server& s, httplib::Server& sV6, const std::function<void(const httplib::Request&, httplib::Response&, std::string)>);
httplib::Server &s,
httplib::Server &sV6,
const std::function<void(const httplib::Request&, httplib::Response&, std::string)>);
void handleRemoteTrace(const ZT_RemoteTrace &rt); void handleRemoteTrace(const ZT_RemoteTrace& rt);
virtual void onNetworkUpdate(const void *db,uint64_t networkId,const nlohmann::json &network); virtual void onNetworkUpdate(const void* db, uint64_t networkId, const nlohmann::json& network);
virtual void onNetworkMemberUpdate(const void *db,uint64_t networkId,uint64_t memberId,const nlohmann::json &member); virtual void onNetworkMemberUpdate(const void* db, uint64_t networkId, uint64_t memberId, const nlohmann::json& member);
virtual void onNetworkMemberDeauthorize(const void *db,uint64_t networkId,uint64_t memberId); virtual void onNetworkMemberDeauthorize(const void* db, uint64_t networkId, uint64_t memberId);
private: private:
void _request(uint64_t nwid,const InetAddress &fromAddr,uint64_t requestPacketId,const Identity &identity,const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData); void _request(uint64_t nwid, const InetAddress& fromAddr, uint64_t requestPacketId, const Identity& identity, const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY>& metaData);
void _startThreads(); void _startThreads();
void _ssoExpiryThread(); void _ssoExpiryThread();
std::string networkUpdateFromPostData(uint64_t networkID, const std::string &body); std::string networkUpdateFromPostData(uint64_t networkID, const std::string& body);
struct _RQEntry struct _RQEntry {
{
uint64_t nwid; uint64_t nwid;
uint64_t requestPacketId; uint64_t requestPacketId;
InetAddress fromAddr; InetAddress fromAddr;
Identity identity; Identity identity;
Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> metaData; Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> metaData;
enum { enum { RQENTRY_TYPE_REQUEST = 0 } type;
RQENTRY_TYPE_REQUEST = 0
} type;
}; };
struct _MemberStatusKey struct _MemberStatusKey {
_MemberStatusKey() : networkId(0), nodeId(0)
{ {
_MemberStatusKey() : networkId(0),nodeId(0) {} }
_MemberStatusKey(const uint64_t nwid,const uint64_t nid) : networkId(nwid),nodeId(nid) {} _MemberStatusKey(const uint64_t nwid, const uint64_t nid) : networkId(nwid), nodeId(nid)
{
}
uint64_t networkId; uint64_t networkId;
uint64_t nodeId; uint64_t nodeId;
inline bool operator==(const _MemberStatusKey &k) const { return ((k.networkId == networkId)&&(k.nodeId == nodeId)); } inline bool operator==(const _MemberStatusKey& k) const
inline bool operator<(const _MemberStatusKey &k) const { return (k.networkId < networkId) || ((k.networkId == networkId)&&(k.nodeId < nodeId)); }
};
struct _MemberStatus
{ {
_MemberStatus() : lastRequestTime(0),authenticationExpiryTime(-1),vMajor(-1),vMinor(-1),vRev(-1),vProto(-1) {} return ((k.networkId == networkId) && (k.nodeId == nodeId));
}
inline bool operator<(const _MemberStatusKey& k) const
{
return (k.networkId < networkId) || ((k.networkId == networkId) && (k.nodeId < nodeId));
}
};
struct _MemberStatus {
_MemberStatus() : lastRequestTime(0), authenticationExpiryTime(-1), vMajor(-1), vMinor(-1), vRev(-1), vProto(-1)
{
}
int64_t lastRequestTime; int64_t lastRequestTime;
int64_t authenticationExpiryTime; int64_t authenticationExpiryTime;
int vMajor,vMinor,vRev,vProto; int vMajor, vMinor, vRev, vProto;
Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> lastRequestMetaData; Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> lastRequestMetaData;
Identity identity; Identity identity;
inline bool online(const int64_t now) const { return ((now - lastRequestTime) < (ZT_NETWORK_AUTOCONF_DELAY * 2)); } inline bool online(const int64_t now) const
};
struct _MemberStatusHash
{ {
inline std::size_t operator()(const _MemberStatusKey &networkIdNodeId) const return ((now - lastRequestTime) < (ZT_NETWORK_AUTOCONF_DELAY * 2));
}
};
struct _MemberStatusHash {
inline std::size_t operator()(const _MemberStatusKey& networkIdNodeId) const
{ {
return (std::size_t)(networkIdNodeId.networkId + networkIdNodeId.nodeId); return (std::size_t)(networkIdNodeId.networkId + networkIdNodeId.nodeId);
} }
@ -127,26 +124,26 @@ private:
const int64_t _startTime; const int64_t _startTime;
int _listenPort; int _listenPort;
Node *const _node; Node* const _node;
std::string _ztPath; std::string _ztPath;
std::string _path; std::string _path;
Identity _signingId; Identity _signingId;
std::string _signingIdAddressString; std::string _signingIdAddressString;
NetworkController::Sender *_sender; NetworkController::Sender* _sender;
DBMirrorSet _db; DBMirrorSet _db;
BlockingQueue< _RQEntry * > _queue; BlockingQueue<_RQEntry*> _queue;
std::vector<std::thread> _threads; std::vector<std::thread> _threads;
std::mutex _threads_l; std::mutex _threads_l;
std::unordered_map< _MemberStatusKey,_MemberStatus,_MemberStatusHash > _memberStatus; std::unordered_map<_MemberStatusKey, _MemberStatus, _MemberStatusHash> _memberStatus;
std::mutex _memberStatus_l; std::mutex _memberStatus_l;
std::set< std::pair<int64_t, _MemberStatusKey> > _expiringSoon; std::set<std::pair<int64_t, _MemberStatusKey> > _expiringSoon;
std::mutex _expiringSoon_l; std::mutex _expiringSoon_l;
RedisConfig *_rc; RedisConfig* _rc;
std::string _ssoRedirectURL; std::string _ssoRedirectURL;
bool _ssoExpiryRunning; bool _ssoExpiryRunning;

137
controller/FileDB.cpp
View file

@ -15,51 +15,49 @@
#include "../node/Metrics.hpp" #include "../node/Metrics.hpp"
namespace ZeroTier namespace ZeroTier {
{
FileDB::FileDB(const char *path) : FileDB::FileDB(const char* path) : DB(), _path(path), _networksPath(_path + ZT_PATH_SEPARATOR_S + "network"), _tracePath(_path + ZT_PATH_SEPARATOR_S + "trace"), _running(true)
DB(),
_path(path),
_networksPath(_path + ZT_PATH_SEPARATOR_S + "network"),
_tracePath(_path + ZT_PATH_SEPARATOR_S + "trace"),
_running(true)
{ {
OSUtils::mkdir(_path.c_str()); OSUtils::mkdir(_path.c_str());
OSUtils::lockDownFile(_path.c_str(),true); OSUtils::lockDownFile(_path.c_str(), true);
OSUtils::mkdir(_networksPath.c_str()); OSUtils::mkdir(_networksPath.c_str());
OSUtils::mkdir(_tracePath.c_str()); OSUtils::mkdir(_tracePath.c_str());
std::vector<std::string> networks(OSUtils::listDirectory(_networksPath.c_str(),false)); std::vector<std::string> networks(OSUtils::listDirectory(_networksPath.c_str(), false));
std::string buf; std::string buf;
for(auto n=networks.begin();n!=networks.end();++n) { for (auto n = networks.begin(); n != networks.end(); ++n) {
buf.clear(); buf.clear();
if ((n->length() == 21)&&(OSUtils::readFile((_networksPath + ZT_PATH_SEPARATOR_S + *n).c_str(),buf))) { if ((n->length() == 21) && (OSUtils::readFile((_networksPath + ZT_PATH_SEPARATOR_S + *n).c_str(), buf))) {
try { try {
nlohmann::json network(OSUtils::jsonParse(buf)); nlohmann::json network(OSUtils::jsonParse(buf));
const std::string nwids = network["id"]; const std::string nwids = network["id"];
if (nwids.length() == 16) { if (nwids.length() == 16) {
nlohmann::json nullJson; nlohmann::json nullJson;
_networkChanged(nullJson,network,false); _networkChanged(nullJson, network, false);
Metrics::network_count++; Metrics::network_count++;
std::string membersPath(_networksPath + ZT_PATH_SEPARATOR_S + nwids + ZT_PATH_SEPARATOR_S "member"); std::string membersPath(_networksPath + ZT_PATH_SEPARATOR_S + nwids + ZT_PATH_SEPARATOR_S "member");
std::vector<std::string> members(OSUtils::listDirectory(membersPath.c_str(),false)); std::vector<std::string> members(OSUtils::listDirectory(membersPath.c_str(), false));
for(auto m=members.begin();m!=members.end();++m) { for (auto m = members.begin(); m != members.end(); ++m) {
buf.clear(); buf.clear();
if ((m->length() == 15)&&(OSUtils::readFile((membersPath + ZT_PATH_SEPARATOR_S + *m).c_str(),buf))) { if ((m->length() == 15) && (OSUtils::readFile((membersPath + ZT_PATH_SEPARATOR_S + *m).c_str(), buf))) {
try { try {
nlohmann::json member(OSUtils::jsonParse(buf)); nlohmann::json member(OSUtils::jsonParse(buf));
const std::string addrs = member["id"]; const std::string addrs = member["id"];
if (addrs.length() == 10) { if (addrs.length() == 10) {
nlohmann::json nullJson2; nlohmann::json nullJson2;
_memberChanged(nullJson2,member,false); _memberChanged(nullJson2, member, false);
Metrics::member_count++; Metrics::member_count++;
} }
} catch ( ... ) {} }
catch (...) {
} }
} }
} }
} catch ( ... ) {} }
}
catch (...) {
}
} }
} }
} }
@ -71,94 +69,101 @@ FileDB::~FileDB()
_running = false; _running = false;
_online_l.unlock(); _online_l.unlock();
_onlineUpdateThread.join(); _onlineUpdateThread.join();
} catch ( ... ) {} }
catch (...) {
}
} }
bool FileDB::waitForReady() { return true; } bool FileDB::waitForReady()
bool FileDB::isReady() { return true; }
bool FileDB::save(nlohmann::json &record,bool notifyListeners)
{ {
char p1[4096],p2[4096],pb[4096]; return true;
}
bool FileDB::isReady()
{
return true;
}
bool FileDB::save(nlohmann::json& record, bool notifyListeners)
{
char p1[4096], p2[4096], pb[4096];
bool modified = false; bool modified = false;
try { try {
const std::string objtype = record["objtype"]; const std::string objtype = record["objtype"];
if (objtype == "network") { if (objtype == "network") {
const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL);
const uint64_t nwid = OSUtils::jsonIntHex(record["id"],0ULL);
if (nwid) { if (nwid) {
nlohmann::json old; nlohmann::json old;
get(nwid,old); get(nwid, old);
if ((!old.is_object())||(!_compareRecords(old,record))) { if ((! old.is_object()) || (! _compareRecords(old, record))) {
record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1ULL; record["revision"] = OSUtils::jsonInt(record["revision"], 0ULL) + 1ULL;
OSUtils::ztsnprintf(p1,sizeof(p1),"%s" ZT_PATH_SEPARATOR_S "%.16llx.json",_networksPath.c_str(),nwid); OSUtils::ztsnprintf(p1, sizeof(p1), "%s" ZT_PATH_SEPARATOR_S "%.16llx.json", _networksPath.c_str(), nwid);
if (!OSUtils::writeFile(p1,OSUtils::jsonDump(record,-1))) { if (! OSUtils::writeFile(p1, OSUtils::jsonDump(record, -1))) {
fprintf(stderr,"WARNING: controller unable to write to path: %s" ZT_EOL_S,p1); fprintf(stderr, "WARNING: controller unable to write to path: %s" ZT_EOL_S, p1);
} }
_networkChanged(old,record,notifyListeners); _networkChanged(old, record, notifyListeners);
modified = true; modified = true;
} }
} }
}
} else if (objtype == "member") { else if (objtype == "member") {
const uint64_t id = OSUtils::jsonIntHex(record["id"], 0ULL);
const uint64_t id = OSUtils::jsonIntHex(record["id"],0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL);
const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"],0ULL); if ((id) && (nwid)) {
if ((id)&&(nwid)) { nlohmann::json network, old;
nlohmann::json network,old; get(nwid, network, id, old);
get(nwid,network,id,old); if ((! old.is_object()) || (! _compareRecords(old, record))) {
if ((!old.is_object())||(!_compareRecords(old,record))) { record["revision"] = OSUtils::jsonInt(record["revision"], 0ULL) + 1ULL;
record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1ULL; OSUtils::ztsnprintf(pb, sizeof(pb), "%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member", _networksPath.c_str(), (unsigned long long)nwid);
OSUtils::ztsnprintf(pb,sizeof(pb),"%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member",_networksPath.c_str(),(unsigned long long)nwid); OSUtils::ztsnprintf(p1, sizeof(p1), "%s" ZT_PATH_SEPARATOR_S "%.10llx.json", pb, (unsigned long long)id);
OSUtils::ztsnprintf(p1,sizeof(p1),"%s" ZT_PATH_SEPARATOR_S "%.10llx.json",pb,(unsigned long long)id); if (! OSUtils::writeFile(p1, OSUtils::jsonDump(record, -1))) {
if (!OSUtils::writeFile(p1,OSUtils::jsonDump(record,-1))) { OSUtils::ztsnprintf(p2, sizeof(p2), "%s" ZT_PATH_SEPARATOR_S "%.16llx", _networksPath.c_str(), (unsigned long long)nwid);
OSUtils::ztsnprintf(p2,sizeof(p2),"%s" ZT_PATH_SEPARATOR_S "%.16llx",_networksPath.c_str(),(unsigned long long)nwid);
OSUtils::mkdir(p2); OSUtils::mkdir(p2);
OSUtils::mkdir(pb); OSUtils::mkdir(pb);
if (!OSUtils::writeFile(p1,OSUtils::jsonDump(record,-1))) { if (! OSUtils::writeFile(p1, OSUtils::jsonDump(record, -1))) {
fprintf(stderr,"WARNING: controller unable to write to path: %s" ZT_EOL_S,p1); fprintf(stderr, "WARNING: controller unable to write to path: %s" ZT_EOL_S, p1);
} }
} }
_memberChanged(old,record,notifyListeners); _memberChanged(old, record, notifyListeners);
modified = true; modified = true;
} }
} }
} }
} catch ( ... ) {} // drop invalid records missing fields }
catch (...) {
} // drop invalid records missing fields
return modified; return modified;
} }
void FileDB::eraseNetwork(const uint64_t networkId) void FileDB::eraseNetwork(const uint64_t networkId)
{ {
nlohmann::json network,nullJson; nlohmann::json network, nullJson;
get(networkId,network); get(networkId, network);
char p[16384]; char p[16384];
OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx.json",_networksPath.c_str(),networkId); OSUtils::ztsnprintf(p, sizeof(p), "%s" ZT_PATH_SEPARATOR_S "%.16llx.json", _networksPath.c_str(), networkId);
OSUtils::rm(p); OSUtils::rm(p);
OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx",_networksPath.c_str(),(unsigned long long)networkId); OSUtils::ztsnprintf(p, sizeof(p), "%s" ZT_PATH_SEPARATOR_S "%.16llx", _networksPath.c_str(), (unsigned long long)networkId);
OSUtils::rmDashRf(p); OSUtils::rmDashRf(p);
_networkChanged(network,nullJson,true); _networkChanged(network, nullJson, true);
std::lock_guard<std::mutex> l(this->_online_l); std::lock_guard<std::mutex> l(this->_online_l);
this->_online.erase(networkId); this->_online.erase(networkId);
} }
void FileDB::eraseMember(const uint64_t networkId,const uint64_t memberId) void FileDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
nlohmann::json network,member,nullJson; nlohmann::json network, member, nullJson;
get(networkId,network,memberId,member); get(networkId, network, memberId, member);
char p[4096]; char p[4096];
OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member" ZT_PATH_SEPARATOR_S "%.10llx.json",_networksPath.c_str(),networkId,memberId); OSUtils::ztsnprintf(p, sizeof(p), "%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member" ZT_PATH_SEPARATOR_S "%.10llx.json", _networksPath.c_str(), networkId, memberId);
OSUtils::rm(p); OSUtils::rm(p);
_memberChanged(member,nullJson,true); _memberChanged(member, nullJson, true);
std::lock_guard<std::mutex> l(this->_online_l); std::lock_guard<std::mutex> l(this->_online_l);
this->_online[networkId].erase(memberId); this->_online[networkId].erase(memberId);
} }
void FileDB::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress) void FileDB::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress)
{ {
char mid[32],atmp[64]; char mid[32], atmp[64];
OSUtils::ztsnprintf(mid,sizeof(mid),"%.10llx",(unsigned long long)memberId); OSUtils::ztsnprintf(mid, sizeof(mid), "%.10llx", (unsigned long long)memberId);
physicalAddress.toString(atmp); physicalAddress.toString(atmp);
std::lock_guard<std::mutex> l(this->_online_l); std::lock_guard<std::mutex> l(this->_online_l);
this->_online[networkId][memberId][OSUtils::now()] = physicalAddress; this->_online[networkId][memberId][OSUtils::now()] = physicalAddress;

20
controller/FileDB.hpp
View file

@ -16,28 +16,26 @@
#include "DB.hpp" #include "DB.hpp"
namespace ZeroTier namespace ZeroTier {
{
class FileDB : public DB class FileDB : public DB {
{ public:
public: FileDB(const char* path);
FileDB(const char *path);
virtual ~FileDB(); virtual ~FileDB();
virtual bool waitForReady(); virtual bool waitForReady();
virtual bool isReady(); virtual bool isReady();
virtual bool save(nlohmann::json &record,bool notifyListeners); virtual bool save(nlohmann::json& record, bool notifyListeners);
virtual void eraseNetwork(const uint64_t networkId); virtual void eraseNetwork(const uint64_t networkId);
virtual void eraseMember(const uint64_t networkId,const uint64_t memberId); virtual void eraseMember(const uint64_t networkId, const uint64_t memberId);
virtual void nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress); virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress);
protected: protected:
std::string _path; std::string _path;
std::string _networksPath; std::string _networksPath;
std::string _tracePath; std::string _tracePath;
std::thread _onlineUpdateThread; std::thread _onlineUpdateThread;
std::map< uint64_t,std::map<uint64_t,std::map<int64_t,InetAddress> > > _online; std::map<uint64_t, std::map<uint64_t, std::map<int64_t, InetAddress> > > _online;
std::mutex _online_l; std::mutex _online_l;
bool _running; bool _running;
}; };

326
controller/LFDB.cpp
View file

@ -13,51 +13,52 @@
#include "LFDB.hpp" #include "LFDB.hpp"
#include <thread> #include "../ext/cpp-httplib/httplib.h"
#include "../osdep/OSUtils.hpp"
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
#include <sstream> #include <sstream>
#include <thread>
#include "../osdep/OSUtils.hpp" namespace ZeroTier {
#include "../ext/cpp-httplib/httplib.h"
namespace ZeroTier LFDB::LFDB(const Identity& myId, const char* path, const char* lfOwnerPrivate, const char* lfOwnerPublic, const char* lfNodeHost, int lfNodePort, bool storeOnlineState)
{ : DB()
, _myId(myId)
LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,const char *lfOwnerPublic,const char *lfNodeHost,int lfNodePort,bool storeOnlineState) : , _lfOwnerPrivate((lfOwnerPrivate) ? lfOwnerPrivate : "")
DB(), , _lfOwnerPublic((lfOwnerPublic) ? lfOwnerPublic : "")
_myId(myId), , _lfNodeHost((lfNodeHost) ? lfNodeHost : "127.0.0.1")
_lfOwnerPrivate((lfOwnerPrivate) ? lfOwnerPrivate : ""), , _lfNodePort(((lfNodePort > 0) && (lfNodePort < 65536)) ? lfNodePort : 9980)
_lfOwnerPublic((lfOwnerPublic) ? lfOwnerPublic : ""), , _running(true)
_lfNodeHost((lfNodeHost) ? lfNodeHost : "127.0.0.1"), , _ready(false)
_lfNodePort(((lfNodePort > 0)&&(lfNodePort < 65536)) ? lfNodePort : 9980), , _storeOnlineState(storeOnlineState)
_running(true),
_ready(false),
_storeOnlineState(storeOnlineState)
{ {
_syncThread = std::thread([this]() { _syncThread = std::thread([this]() {
char controllerAddress[24]; char controllerAddress[24];
const uint64_t controllerAddressInt = _myId.address().toInt(); const uint64_t controllerAddressInt = _myId.address().toInt();
_myId.address().toString(controllerAddress); _myId.address().toString(controllerAddress);
std::string networksSelectorName("com.zerotier.controller.lfdb:"); networksSelectorName.append(controllerAddress); networksSelectorName.append("/network"); std::string networksSelectorName("com.zerotier.controller.lfdb:");
networksSelectorName.append(controllerAddress);
networksSelectorName.append("/network");
// LF record masking key is the first 32 bytes of SHA512(controller private key) in hex, // LF record masking key is the first 32 bytes of SHA512(controller private key) in hex,
// hiding record values from anything but the controller or someone who has its key. // hiding record values from anything but the controller or someone who has its key.
uint8_t sha512pk[64]; uint8_t sha512pk[64];
_myId.sha512PrivateKey(sha512pk); _myId.sha512PrivateKey(sha512pk);
char maskingKey [128]; char maskingKey[128];
Utils::hex(sha512pk,32,maskingKey); Utils::hex(sha512pk, 32, maskingKey);
httplib::Client htcli(_lfNodeHost.c_str(),_lfNodePort); httplib::Client htcli(_lfNodeHost.c_str(), _lfNodePort);
int64_t timeRangeStart = 0; int64_t timeRangeStart = 0;
while (_running.load()) { while (_running.load()) {
{ {
std::lock_guard<std::mutex> sl(_state_l); std::lock_guard<std::mutex> sl(_state_l);
for(auto ns=_state.begin();ns!=_state.end();++ns) { for (auto ns = _state.begin(); ns != _state.end(); ++ns) {
if (ns->second.dirty) { if (ns->second.dirty) {
nlohmann::json network; nlohmann::json network;
if (get(ns->first,network)) { if (get(ns->first, network)) {
nlohmann::json newrec,selector0; nlohmann::json newrec, selector0;
selector0["Name"] = networksSelectorName; selector0["Name"] = networksSelectorName;
selector0["Ordinal"] = ns->first; selector0["Ordinal"] = ns->first;
newrec["Selectors"].push_back(selector0); newrec["Selectors"].push_back(selector0);
@ -66,30 +67,34 @@ LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,cons
newrec["MaskingKey"] = maskingKey; newrec["MaskingKey"] = maskingKey;
newrec["PulseIfUnchanged"] = true; newrec["PulseIfUnchanged"] = true;
try { try {
auto resp = htcli.Post("/makerecord",newrec.dump(),"application/json"); auto resp = htcli.Post("/makerecord", newrec.dump(), "application/json");
if (resp) { if (resp) {
if (resp->status == 200) { if (resp->status == 200) {
ns->second.dirty = false; ns->second.dirty = false;
//printf("SET network %.16llx %s\n",ns->first,resp->body.c_str()); // printf("SET network %.16llx %s\n",ns->first,resp->body.c_str());
} else {
fprintf(stderr,"ERROR: LFDB: %d from node (create/update network): %s" ZT_EOL_S,resp->status,resp->body.c_str());
} }
} else { else {
fprintf(stderr,"ERROR: LFDB: node is offline" ZT_EOL_S); fprintf(stderr, "ERROR: LFDB: %d from node (create/update network): %s" ZT_EOL_S, resp->status, resp->body.c_str());
} }
} catch (std::exception &e) { }
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (create/update network): %s" ZT_EOL_S,e.what()); else {
} catch ( ... ) { fprintf(stderr, "ERROR: LFDB: node is offline" ZT_EOL_S);
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (create/update network): unknown exception" ZT_EOL_S); }
}
catch (std::exception& e) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (create/update network): %s" ZT_EOL_S, e.what());
}
catch (...) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (create/update network): unknown exception" ZT_EOL_S);
} }
} }
} }
for(auto ms=ns->second.members.begin();ms!=ns->second.members.end();++ms) { for (auto ms = ns->second.members.begin(); ms != ns->second.members.end(); ++ms) {
if ((_storeOnlineState)&&(ms->second.lastOnlineDirty)&&(ms->second.lastOnlineAddress)) { if ((_storeOnlineState) && (ms->second.lastOnlineDirty) && (ms->second.lastOnlineAddress)) {
nlohmann::json newrec,selector0,selector1,selectors,ip; nlohmann::json newrec, selector0, selector1, selectors, ip;
char tmp[1024],tmp2[128]; char tmp[1024], tmp2[128];
OSUtils::ztsnprintf(tmp,sizeof(tmp),"com.zerotier.controller.lfdb:%s/network/%.16llx/online",controllerAddress,(unsigned long long)ns->first); OSUtils::ztsnprintf(tmp, sizeof(tmp), "com.zerotier.controller.lfdb:%s/network/%.16llx/online", controllerAddress, (unsigned long long)ns->first);
ms->second.lastOnlineAddress.toIpString(tmp2); ms->second.lastOnlineAddress.toIpString(tmp2);
selector0["Name"] = tmp; selector0["Name"] = tmp;
selector0["Ordinal"] = ms->first; selector0["Ordinal"] = ms->first;
@ -98,14 +103,14 @@ LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,cons
selectors.push_back(selector0); selectors.push_back(selector0);
selectors.push_back(selector1); selectors.push_back(selector1);
newrec["Selectors"] = selectors; newrec["Selectors"] = selectors;
const uint8_t *const rawip = (const uint8_t *)ms->second.lastOnlineAddress.rawIpData(); const uint8_t* const rawip = (const uint8_t*)ms->second.lastOnlineAddress.rawIpData();
switch(ms->second.lastOnlineAddress.ss_family) { switch (ms->second.lastOnlineAddress.ss_family) {
case AF_INET: case AF_INET:
for(int j=0;j<4;++j) for (int j = 0; j < 4; ++j)
ip.push_back((unsigned int)rawip[j]); ip.push_back((unsigned int)rawip[j]);
break; break;
case AF_INET6: case AF_INET6:
for(int j=0;j<16;++j) for (int j = 0; j < 16; ++j)
ip.push_back((unsigned int)rawip[j]); ip.push_back((unsigned int)rawip[j]);
break; break;
default: default:
@ -118,28 +123,32 @@ LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,cons
newrec["Timestamp"] = ms->second.lastOnlineTime; newrec["Timestamp"] = ms->second.lastOnlineTime;
newrec["PulseIfUnchanged"] = true; newrec["PulseIfUnchanged"] = true;
try { try {
auto resp = htcli.Post("/makerecord",newrec.dump(),"application/json"); auto resp = htcli.Post("/makerecord", newrec.dump(), "application/json");
if (resp) { if (resp) {
if (resp->status == 200) { if (resp->status == 200) {
ms->second.lastOnlineDirty = false; ms->second.lastOnlineDirty = false;
//printf("SET member online %.16llx %.10llx %s\n",ns->first,ms->first,resp->body.c_str()); // printf("SET member online %.16llx %.10llx %s\n",ns->first,ms->first,resp->body.c_str());
} else {
fprintf(stderr,"ERROR: LFDB: %d from node (create/update member online status): %s" ZT_EOL_S,resp->status,resp->body.c_str());
} }
} else { else {
fprintf(stderr,"ERROR: LFDB: node is offline" ZT_EOL_S); fprintf(stderr, "ERROR: LFDB: %d from node (create/update member online status): %s" ZT_EOL_S, resp->status, resp->body.c_str());
} }
} catch (std::exception &e) { }
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (create/update member online status): %s" ZT_EOL_S,e.what()); else {
} catch ( ... ) { fprintf(stderr, "ERROR: LFDB: node is offline" ZT_EOL_S);
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (create/update member online status): unknown exception" ZT_EOL_S); }
}
catch (std::exception& e) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (create/update member online status): %s" ZT_EOL_S, e.what());
}
catch (...) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (create/update member online status): unknown exception" ZT_EOL_S);
} }
} }
if (ms->second.dirty) { if (ms->second.dirty) {
nlohmann::json network,member; nlohmann::json network, member;
if (get(ns->first,network,ms->first,member)) { if (get(ns->first, network, ms->first, member)) {
nlohmann::json newrec,selector0,selector1,selectors; nlohmann::json newrec, selector0, selector1, selectors;
selector0["Name"] = networksSelectorName; selector0["Name"] = networksSelectorName;
selector0["Ordinal"] = ns->first; selector0["Ordinal"] = ns->first;
selector1["Name"] = "member"; selector1["Name"] = "member";
@ -152,21 +161,25 @@ LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,cons
newrec["MaskingKey"] = maskingKey; newrec["MaskingKey"] = maskingKey;
newrec["PulseIfUnchanged"] = true; newrec["PulseIfUnchanged"] = true;
try { try {
auto resp = htcli.Post("/makerecord",newrec.dump(),"application/json"); auto resp = htcli.Post("/makerecord", newrec.dump(), "application/json");
if (resp) { if (resp) {
if (resp->status == 200) { if (resp->status == 200) {
ms->second.dirty = false; ms->second.dirty = false;
//printf("SET member %.16llx %.10llx %s\n",ns->first,ms->first,resp->body.c_str()); // printf("SET member %.16llx %.10llx %s\n",ns->first,ms->first,resp->body.c_str());
} else {
fprintf(stderr,"ERROR: LFDB: %d from node (create/update member): %s" ZT_EOL_S,resp->status,resp->body.c_str());
} }
} else { else {
fprintf(stderr,"ERROR: LFDB: node is offline" ZT_EOL_S); fprintf(stderr, "ERROR: LFDB: %d from node (create/update member): %s" ZT_EOL_S, resp->status, resp->body.c_str());
} }
} catch (std::exception &e) { }
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (create/update member): %s" ZT_EOL_S,e.what()); else {
} catch ( ... ) { fprintf(stderr, "ERROR: LFDB: node is offline" ZT_EOL_S);
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (create/update member): unknown exception" ZT_EOL_S); }
}
catch (std::exception& e) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (create/update member): %s" ZT_EOL_S, e.what());
}
catch (...) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (create/update member): unknown exception" ZT_EOL_S);
} }
} }
} }
@ -176,31 +189,37 @@ LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,cons
try { try {
std::ostringstream query; std::ostringstream query;
query << query << "{"
"{"
"\"Ranges\":[{" "\"Ranges\":[{"
"\"Name\":\"" << networksSelectorName << "\"," "\"Name\":\""
<< networksSelectorName
<< "\","
"\"Range\":[0,18446744073709551615]" "\"Range\":[0,18446744073709551615]"
"}]," "}],"
"\"TimeRange\":[" << timeRangeStart << ",9223372036854775807]," "\"TimeRange\":["
"\"MaskingKey\":\"" << maskingKey << "\"," << timeRangeStart
"\"Owners\":[\"" << _lfOwnerPublic << "\"]" << ",9223372036854775807],"
"\"MaskingKey\":\""
<< maskingKey
<< "\","
"\"Owners\":[\""
<< _lfOwnerPublic
<< "\"]"
"}"; "}";
auto resp = htcli.Post("/query",query.str(),"application/json"); auto resp = htcli.Post("/query", query.str(), "application/json");
if (resp) { if (resp) {
if (resp->status == 200) { if (resp->status == 200) {
nlohmann::json results(OSUtils::jsonParse(resp->body)); nlohmann::json results(OSUtils::jsonParse(resp->body));
if ((results.is_array())&&(!results.empty())) { if ((results.is_array()) && (! results.empty())) {
for(std::size_t ri=0;ri<results.size();++ri) { for (std::size_t ri = 0; ri < results.size(); ++ri) {
nlohmann::json &rset = results[ri]; nlohmann::json& rset = results[ri];
if ((rset.is_array())&&(!rset.empty())) { if ((rset.is_array()) && (! rset.empty())) {
nlohmann::json& result = rset[0];
nlohmann::json &result = rset[0];
if (result.is_object()) { if (result.is_object()) {
nlohmann::json &record = result["Record"]; nlohmann::json& record = result["Record"];
if (record.is_object()) { if (record.is_object()) {
const std::string recordValue = result["Value"]; const std::string recordValue = result["Value"];
//printf("GET network %s\n",recordValue.c_str()); // printf("GET network %s\n",recordValue.c_str());
nlohmann::json network(OSUtils::jsonParse(recordValue)); nlohmann::json network(OSUtils::jsonParse(recordValue));
if (network.is_object()) { if (network.is_object()) {
const std::string idstr = network["id"]; const std::string idstr = network["id"];
@ -208,111 +227,123 @@ LFDB::LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,cons
if ((id >> 24) == controllerAddressInt) { // sanity check if ((id >> 24) == controllerAddressInt) { // sanity check
nlohmann::json oldNetwork; nlohmann::json oldNetwork;
if ((timeRangeStart > 0)&&(get(id,oldNetwork))) { if ((timeRangeStart > 0) && (get(id, oldNetwork))) {
const uint64_t revision = network["revision"]; const uint64_t revision = network["revision"];
const uint64_t prevRevision = oldNetwork["revision"]; const uint64_t prevRevision = oldNetwork["revision"];
if (prevRevision < revision) { if (prevRevision < revision) {
_networkChanged(oldNetwork,network,timeRangeStart > 0); _networkChanged(oldNetwork, network, timeRangeStart > 0);
} }
} else { }
else {
nlohmann::json nullJson; nlohmann::json nullJson;
_networkChanged(nullJson,network,timeRangeStart > 0); _networkChanged(nullJson, network, timeRangeStart > 0);
}
} }
} }
} }
} }
} }
} }
} }
} else {
fprintf(stderr,"ERROR: LFDB: %d from node (check for network updates): %s" ZT_EOL_S,resp->status,resp->body.c_str());
} }
} else {
fprintf(stderr,"ERROR: LFDB: node is offline" ZT_EOL_S);
} }
} catch (std::exception &e) { else {
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (check for network updates): %s" ZT_EOL_S,e.what()); fprintf(stderr, "ERROR: LFDB: %d from node (check for network updates): %s" ZT_EOL_S, resp->status, resp->body.c_str());
} catch ( ... ) { }
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (check for network updates): unknown exception" ZT_EOL_S); }
else {
fprintf(stderr, "ERROR: LFDB: node is offline" ZT_EOL_S);
}
}
catch (std::exception& e) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (check for network updates): %s" ZT_EOL_S, e.what());
}
catch (...) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (check for network updates): unknown exception" ZT_EOL_S);
} }
try { try {
std::ostringstream query; std::ostringstream query;
query << query << "{"
"{"
"\"Ranges\":[{" "\"Ranges\":[{"
"\"Name\":\"" << networksSelectorName << "\"," "\"Name\":\""
<< networksSelectorName
<< "\","
"\"Range\":[0,18446744073709551615]" "\"Range\":[0,18446744073709551615]"
"},{" "},{"
"\"Name\":\"member\"," "\"Name\":\"member\","
"\"Range\":[0,18446744073709551615]" "\"Range\":[0,18446744073709551615]"
"}]," "}],"
"\"TimeRange\":[" << timeRangeStart << ",9223372036854775807]," "\"TimeRange\":["
"\"MaskingKey\":\"" << maskingKey << "\"," << timeRangeStart
"\"Owners\":[\"" << _lfOwnerPublic << "\"]" << ",9223372036854775807],"
"\"MaskingKey\":\""
<< maskingKey
<< "\","
"\"Owners\":[\""
<< _lfOwnerPublic
<< "\"]"
"}"; "}";
auto resp = htcli.Post("/query",query.str(),"application/json"); auto resp = htcli.Post("/query", query.str(), "application/json");
if (resp) { if (resp) {
if (resp->status == 200) { if (resp->status == 200) {
nlohmann::json results(OSUtils::jsonParse(resp->body)); nlohmann::json results(OSUtils::jsonParse(resp->body));
if ((results.is_array())&&(!results.empty())) { if ((results.is_array()) && (! results.empty())) {
for(std::size_t ri=0;ri<results.size();++ri) { for (std::size_t ri = 0; ri < results.size(); ++ri) {
nlohmann::json &rset = results[ri]; nlohmann::json& rset = results[ri];
if ((rset.is_array())&&(!rset.empty())) { if ((rset.is_array()) && (! rset.empty())) {
nlohmann::json& result = rset[0];
nlohmann::json &result = rset[0];
if (result.is_object()) { if (result.is_object()) {
nlohmann::json &record = result["Record"]; nlohmann::json& record = result["Record"];
if (record.is_object()) { if (record.is_object()) {
const std::string recordValue = result["Value"]; const std::string recordValue = result["Value"];
//printf("GET member %s\n",recordValue.c_str()); // printf("GET member %s\n",recordValue.c_str());
nlohmann::json member(OSUtils::jsonParse(recordValue)); nlohmann::json member(OSUtils::jsonParse(recordValue));
if (member.is_object()) { if (member.is_object()) {
const std::string nwidstr = member["nwid"]; const std::string nwidstr = member["nwid"];
const std::string idstr = member["id"]; const std::string idstr = member["id"];
const uint64_t nwid = Utils::hexStrToU64(nwidstr.c_str()); const uint64_t nwid = Utils::hexStrToU64(nwidstr.c_str());
const uint64_t id = Utils::hexStrToU64(idstr.c_str()); const uint64_t id = Utils::hexStrToU64(idstr.c_str());
if ((id)&&((nwid >> 24) == controllerAddressInt)) { // sanity check if ((id) && ((nwid >> 24) == controllerAddressInt)) { // sanity check
nlohmann::json network,oldMember; nlohmann::json network, oldMember;
if ((timeRangeStart > 0)&&(get(nwid,network,id,oldMember))) { if ((timeRangeStart > 0) && (get(nwid, network, id, oldMember))) {
const uint64_t revision = member["revision"]; const uint64_t revision = member["revision"];
const uint64_t prevRevision = oldMember["revision"]; const uint64_t prevRevision = oldMember["revision"];
if (prevRevision < revision) if (prevRevision < revision)
_memberChanged(oldMember,member,timeRangeStart > 0); _memberChanged(oldMember, member, timeRangeStart > 0);
} else if (hasNetwork(nwid)) { }
else if (hasNetwork(nwid)) {
nlohmann::json nullJson; nlohmann::json nullJson;
_memberChanged(nullJson,member,timeRangeStart > 0); _memberChanged(nullJson, member, timeRangeStart > 0);
}
} }
} }
} }
} }
} }
} }
} }
} else {
fprintf(stderr,"ERROR: LFDB: %d from node (check for member updates): %s" ZT_EOL_S,resp->status,resp->body.c_str());
} }
} else {
fprintf(stderr,"ERROR: LFDB: node is offline" ZT_EOL_S);
} }
} catch (std::exception &e) { else {
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (check for member updates): %s" ZT_EOL_S,e.what()); fprintf(stderr, "ERROR: LFDB: %d from node (check for member updates): %s" ZT_EOL_S, resp->status, resp->body.c_str());
} catch ( ... ) { }
fprintf(stderr,"ERROR: LFDB: unexpected exception querying node (check for member updates): unknown exception" ZT_EOL_S); }
else {
fprintf(stderr, "ERROR: LFDB: node is offline" ZT_EOL_S);
}
}
catch (std::exception& e) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (check for member updates): %s" ZT_EOL_S, e.what());
}
catch (...) {
fprintf(stderr, "ERROR: LFDB: unexpected exception querying node (check for member updates): unknown exception" ZT_EOL_S);
} }
timeRangeStart = time(nullptr) - 120; // start next query 2m before now to avoid losing updates timeRangeStart = time(nullptr) - 120; // start next query 2m before now to avoid losing updates
_ready.store(true); _ready.store(true);
for(int k=0;k<4;++k) { // 2s delay between queries for remotely modified networks or members for (int k = 0; k < 4; ++k) { // 2s delay between queries for remotely modified networks or members
if (!_running.load()) if (! _running.load())
return; return;
std::this_thread::sleep_for(std::chrono::milliseconds(500)); std::this_thread::sleep_for(std::chrono::milliseconds(500));
} }
@ -328,7 +359,7 @@ LFDB::~LFDB()
bool LFDB::waitForReady() bool LFDB::waitForReady()
{ {
while (!_ready.load()) { while (! _ready.load()) {
std::this_thread::sleep_for(std::chrono::milliseconds(500)); std::this_thread::sleep_for(std::chrono::milliseconds(500));
} }
return true; return true;
@ -339,18 +370,18 @@ bool LFDB::isReady()
return (_ready.load()); return (_ready.load());
} }
bool LFDB::save(nlohmann::json &record,bool notifyListeners) bool LFDB::save(nlohmann::json& record, bool notifyListeners)
{ {
bool modified = false; bool modified = false;
const std::string objtype = record["objtype"]; const std::string objtype = record["objtype"];
if (objtype == "network") { if (objtype == "network") {
const uint64_t nwid = OSUtils::jsonIntHex(record["id"],0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL);
if (nwid) { if (nwid) {
nlohmann::json old; nlohmann::json old;
get(nwid,old); get(nwid, old);
if ((!old.is_object())||(!_compareRecords(old,record))) { if ((! old.is_object()) || (! _compareRecords(old, record))) {
record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1ULL; record["revision"] = OSUtils::jsonInt(record["revision"], 0ULL) + 1ULL;
_networkChanged(old,record,notifyListeners); _networkChanged(old, record, notifyListeners);
{ {
std::lock_guard<std::mutex> l(_state_l); std::lock_guard<std::mutex> l(_state_l);
_state[nwid].dirty = true; _state[nwid].dirty = true;
@ -358,15 +389,16 @@ bool LFDB::save(nlohmann::json &record,bool notifyListeners)
modified = true; modified = true;
} }
} }
} else if (objtype == "member") { }
const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"],0ULL); else if (objtype == "member") {
const uint64_t id = OSUtils::jsonIntHex(record["id"],0ULL); const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL);
if ((id)&&(nwid)) { const uint64_t id = OSUtils::jsonIntHex(record["id"], 0ULL);
nlohmann::json network,old; if ((id) && (nwid)) {
get(nwid,network,id,old); nlohmann::json network, old;
if ((!old.is_object())||(!_compareRecords(old,record))) { get(nwid, network, id, old);
record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1ULL; if ((! old.is_object()) || (! _compareRecords(old, record))) {
_memberChanged(old,record,notifyListeners); record["revision"] = OSUtils::jsonInt(record["revision"], 0ULL) + 1ULL;
_memberChanged(old, record, notifyListeners);
{ {
std::lock_guard<std::mutex> l(_state_l); std::lock_guard<std::mutex> l(_state_l);
_state[nwid].members[id].dirty = true; _state[nwid].members[id].dirty = true;
@ -383,12 +415,12 @@ void LFDB::eraseNetwork(const uint64_t networkId)
// TODO // TODO
} }
void LFDB::eraseMember(const uint64_t networkId,const uint64_t memberId) void LFDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
{ {
// TODO // TODO
} }
void LFDB::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress) void LFDB::nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress)
{ {
std::lock_guard<std::mutex> l(_state_l); std::lock_guard<std::mutex> l(_state_l);
auto nw = _state.find(networkId); auto nw = _state.find(networkId);

39
controller/LFDB.hpp
View file

@ -16,19 +16,18 @@
#include "DB.hpp" #include "DB.hpp"
#include <atomic>
#include <mutex> #include <mutex>
#include <string> #include <string>
#include <unordered_map> #include <unordered_map>
#include <atomic>
namespace ZeroTier { namespace ZeroTier {
/** /**
* DB implementation for controller that stores data in LF * DB implementation for controller that stores data in LF
*/ */
class LFDB : public DB class LFDB : public DB {
{ public:
public:
/** /**
* @param myId This controller's identity * @param myId This controller's identity
* @param path Base path for ZeroTier node itself * @param path Base path for ZeroTier node itself
@ -38,44 +37,40 @@ public:
* @param lfNodePort LF node http (not https) port * @param lfNodePort LF node http (not https) port
* @param storeOnlineState If true, store online/offline state and IP info in LF (a lot of data, only for private networks!) * @param storeOnlineState If true, store online/offline state and IP info in LF (a lot of data, only for private networks!)
*/ */
LFDB(const Identity &myId,const char *path,const char *lfOwnerPrivate,const char *lfOwnerPublic,const char *lfNodeHost,int lfNodePort,bool storeOnlineState); LFDB(const Identity& myId, const char* path, const char* lfOwnerPrivate, const char* lfOwnerPublic, const char* lfNodeHost, int lfNodePort, bool storeOnlineState);
virtual ~LFDB(); virtual ~LFDB();
virtual bool waitForReady(); virtual bool waitForReady();
virtual bool isReady(); virtual bool isReady();
virtual bool save(nlohmann::json &record,bool notifyListeners); virtual bool save(nlohmann::json& record, bool notifyListeners);
virtual void eraseNetwork(const uint64_t networkId); virtual void eraseNetwork(const uint64_t networkId);
virtual void eraseMember(const uint64_t networkId,const uint64_t memberId); virtual void eraseMember(const uint64_t networkId, const uint64_t memberId);
virtual void nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress); virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress);
protected: protected:
const Identity _myId; const Identity _myId;
std::string _lfOwnerPrivate,_lfOwnerPublic; std::string _lfOwnerPrivate, _lfOwnerPublic;
std::string _lfNodeHost; std::string _lfNodeHost;
int _lfNodePort; int _lfNodePort;
struct _MemberState struct _MemberState {
_MemberState() : lastOnlineAddress(), lastOnlineTime(0), dirty(false), lastOnlineDirty(false)
{ {
_MemberState() : }
lastOnlineAddress(),
lastOnlineTime(0),
dirty(false),
lastOnlineDirty(false) {}
InetAddress lastOnlineAddress; InetAddress lastOnlineAddress;
int64_t lastOnlineTime; int64_t lastOnlineTime;
bool dirty; bool dirty;
bool lastOnlineDirty; bool lastOnlineDirty;
}; };
struct _NetworkState struct _NetworkState {
_NetworkState() : members(), dirty(false)
{ {
_NetworkState() : }
members(), std::unordered_map<uint64_t, _MemberState> members;
dirty(false) {}
std::unordered_map<uint64_t,_MemberState> members;
bool dirty; bool dirty;
}; };
std::unordered_map<uint64_t,_NetworkState> _state; std::unordered_map<uint64_t, _NetworkState> _state;
std::mutex _state_l; std::mutex _state_l;
std::atomic_bool _running; std::atomic_bool _running;

856
controller/PostgreSQL.cpp
View file

File diff suppressed because it is too large Load diff

101
controller/PostgreSQL.hpp
View file

@ -20,78 +20,81 @@
#define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 4 #define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 4
#include "../node/Metrics.hpp"
#include "ConnectionPool.hpp" #include "ConnectionPool.hpp"
#include <pqxx/pqxx>
#include <memory> #include <memory>
#include <pqxx/pqxx>
#include <redis++/redis++.h> #include <redis++/redis++.h>
#include "../node/Metrics.hpp"
extern "C" { extern "C" {
typedef struct pg_conn PGconn; typedef struct pg_conn PGconn;
} }
namespace smeeclient { namespace smeeclient {
struct SmeeClient; struct SmeeClient;
} }
namespace ZeroTier { namespace ZeroTier {
struct RedisConfig; struct RedisConfig;
class PostgresConnection : public Connection { class PostgresConnection : public Connection {
public: public:
virtual ~PostgresConnection() { virtual ~PostgresConnection()
{
} }
std::shared_ptr<pqxx::connection> c; std::shared_ptr<pqxx::connection> c;
int a; int a;
}; };
class PostgresConnFactory : public ConnectionFactory { class PostgresConnFactory : public ConnectionFactory {
public: public:
PostgresConnFactory(std::string &connString) PostgresConnFactory(std::string& connString) : m_connString(connString)
: m_connString(connString)
{ {
} }
virtual std::shared_ptr<Connection> create() { virtual std::shared_ptr<Connection> create()
{
Metrics::conn_counter++; Metrics::conn_counter++;
auto c = std::shared_ptr<PostgresConnection>(new PostgresConnection()); auto c = std::shared_ptr<PostgresConnection>(new PostgresConnection());
c->c = std::make_shared<pqxx::connection>(m_connString); c->c = std::make_shared<pqxx::connection>(m_connString);
return std::static_pointer_cast<Connection>(c); return std::static_pointer_cast<Connection>(c);
} }
private:
private:
std::string m_connString; std::string m_connString;
}; };
class PostgreSQL; class PostgreSQL;
class MemberNotificationReceiver : public pqxx::notification_receiver { class MemberNotificationReceiver : public pqxx::notification_receiver {
public: public:
MemberNotificationReceiver(PostgreSQL *p, pqxx::connection &c, const std::string &channel); MemberNotificationReceiver(PostgreSQL* p, pqxx::connection& c, const std::string& channel);
virtual ~MemberNotificationReceiver() { virtual ~MemberNotificationReceiver()
{
fprintf(stderr, "MemberNotificationReceiver destroyed\n"); fprintf(stderr, "MemberNotificationReceiver destroyed\n");
} }
virtual void operator() (const std::string &payload, int backendPid); virtual void operator()(const std::string& payload, int backendPid);
private:
PostgreSQL *_psql; private:
PostgreSQL* _psql;
}; };
class NetworkNotificationReceiver : public pqxx::notification_receiver { class NetworkNotificationReceiver : public pqxx::notification_receiver {
public: public:
NetworkNotificationReceiver(PostgreSQL *p, pqxx::connection &c, const std::string &channel); NetworkNotificationReceiver(PostgreSQL* p, pqxx::connection& c, const std::string& channel);
virtual ~NetworkNotificationReceiver() { virtual ~NetworkNotificationReceiver()
{
fprintf(stderr, "NetworkNotificationReceiver destroyed\n"); fprintf(stderr, "NetworkNotificationReceiver destroyed\n");
}; };
virtual void operator() (const std::string &payload, int packend_pid); virtual void operator()(const std::string& payload, int packend_pid);
private:
PostgreSQL *_psql; private:
PostgreSQL* _psql;
}; };
/** /**
@ -100,36 +103,40 @@ private:
* This is for use with ZeroTier Central. Others are free to build and use it * This is for use with ZeroTier Central. Others are free to build and use it
* but be aware that we might change it at any time. * but be aware that we might change it at any time.
*/ */
class PostgreSQL : public DB class PostgreSQL : public DB {
{
friend class MemberNotificationReceiver; friend class MemberNotificationReceiver;
friend class NetworkNotificationReceiver; friend class NetworkNotificationReceiver;
public:
PostgreSQL(const Identity &myId, const char *path, int listenPort, RedisConfig *rc); public:
PostgreSQL(const Identity& myId, const char* path, int listenPort, RedisConfig* rc);
virtual ~PostgreSQL(); virtual ~PostgreSQL();
virtual bool waitForReady(); virtual bool waitForReady();
virtual bool isReady(); virtual bool isReady();
virtual bool save(nlohmann::json &record,bool notifyListeners); virtual bool save(nlohmann::json& record, bool notifyListeners);
virtual void eraseNetwork(const uint64_t networkId); virtual void eraseNetwork(const uint64_t networkId);
virtual void eraseMember(const uint64_t networkId, const uint64_t memberId); virtual void eraseMember(const uint64_t networkId, const uint64_t memberId);
virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress &physicalAddress); virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress);
virtual AuthInfo getSSOAuthInfo(const nlohmann::json &member, const std::string &redirectURL); virtual AuthInfo getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL);
protected: protected:
struct _PairHasher struct _PairHasher {
inline std::size_t operator()(const std::pair<uint64_t, uint64_t>& p) const
{ {
inline std::size_t operator()(const std::pair<uint64_t,uint64_t> &p) const { return (std::size_t)(p.first ^ p.second); } return (std::size_t)(p.first ^ p.second);
}
}; };
virtual void _memberChanged(nlohmann::json &old,nlohmann::json &memberConfig,bool notifyListeners) { virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners)
{
DB::_memberChanged(old, memberConfig, notifyListeners); DB::_memberChanged(old, memberConfig, notifyListeners);
} }
virtual void _networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool notifyListeners) { virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners)
{
DB::_networkChanged(old, networkConfig, notifyListeners); DB::_networkChanged(old, networkConfig, notifyListeners);
} }
private: private:
void initializeNetworks(); void initializeNetworks();
void initializeMembers(); void initializeMembers();
void heartbeat(); void heartbeat();
@ -145,16 +152,12 @@ private:
void onlineNotificationThread(); void onlineNotificationThread();
void onlineNotification_Postgres(); void onlineNotification_Postgres();
void onlineNotification_Redis(); void onlineNotification_Redis();
uint64_t _doRedisUpdate(sw::redis::Transaction &tx, std::string &controllerId, uint64_t _doRedisUpdate(sw::redis::Transaction& tx, std::string& controllerId, std::unordered_map<std::pair<uint64_t, uint64_t>, std::pair<int64_t, InetAddress>, _PairHasher>& lastOnline);
std::unordered_map< std::pair<uint64_t,uint64_t>,std::pair<int64_t,InetAddress>,_PairHasher > &lastOnline);
void configureSmee(); void configureSmee();
void notifyNewMember(const std::string &networkID, const std::string &memberID); void notifyNewMember(const std::string& networkID, const std::string& memberID);
enum OverrideMode { enum OverrideMode { ALLOW_PGBOUNCER_OVERRIDE = 0, NO_OVERRIDE = 1 };
ALLOW_PGBOUNCER_OVERRIDE = 0,
NO_OVERRIDE = 1
};
std::shared_ptr<ConnectionPool<PostgresConnection> > _pool; std::shared_ptr<ConnectionPool<PostgresConnection> > _pool;
@ -163,7 +166,7 @@ private:
std::string _myAddressStr; std::string _myAddressStr;
std::string _connString; std::string _connString;
BlockingQueue< std::pair<nlohmann::json,bool> > _commitQueue; BlockingQueue<std::pair<nlohmann::json, bool> > _commitQueue;
std::thread _heartbeatThread; std::thread _heartbeatThread;
std::thread _membersDbWatcher; std::thread _membersDbWatcher;
@ -171,7 +174,7 @@ private:
std::thread _commitThread[ZT_CENTRAL_CONTROLLER_COMMIT_THREADS]; std::thread _commitThread[ZT_CENTRAL_CONTROLLER_COMMIT_THREADS];
std::thread _onlineNotificationThread; std::thread _onlineNotificationThread;
std::unordered_map< std::pair<uint64_t,uint64_t>,std::pair<int64_t,InetAddress>,_PairHasher > _lastOnline; std::unordered_map<std::pair<uint64_t, uint64_t>, std::pair<int64_t, InetAddress>, _PairHasher> _lastOnline;
mutable std::mutex _lastOnline_l; mutable std::mutex _lastOnline_l;
mutable std::mutex _readyLock; mutable std::mutex _readyLock;
@ -181,12 +184,12 @@ private:
int _listenPort; int _listenPort;
uint8_t _ssoPsk[48]; uint8_t _ssoPsk[48];
RedisConfig *_rc; RedisConfig* _rc;
std::shared_ptr<sw::redis::Redis> _redis; std::shared_ptr<sw::redis::Redis> _redis;
std::shared_ptr<sw::redis::RedisCluster> _cluster; std::shared_ptr<sw::redis::RedisCluster> _cluster;
bool _redisMemberStatus; bool _redisMemberStatus;
smeeclient::SmeeClient *_smee; smeeclient::SmeeClient* _smee;
}; };
} // namespace ZeroTier } // namespace ZeroTier

2
controller/Redis.hpp
View file

@ -10,6 +10,6 @@ struct RedisConfig {
std::string password; std::string password;
bool clusterMode; bool clusterMode;
}; };
} } // namespace ZeroTier
#endif #endif

1
node/Peer.cpp
View file

@ -25,7 +25,6 @@
#include "Switch.hpp" #include "Switch.hpp"
#include "Trace.hpp" #include "Trace.hpp"
#include "Utils.hpp" #include "Utils.hpp"
#include "Switch.hpp"
namespace ZeroTier { namespace ZeroTier {

8
node/Switch.cpp
View file

@ -495,9 +495,7 @@ void Switch::onLocalEthernet(void* tPtr, const SharedPtr<Network>& network, cons
// this prevents problems related to trying to do rx while inside of doing tx, such as acquiring same lock recursively // this prevents problems related to trying to do rx while inside of doing tx, such as acquiring same lock recursively
// //
std::thread([=]() { std::thread([=]() { RR->node->putFrame(tPtr, network->id(), network->userPtr(), peerMac, from, ZT_ETHERTYPE_IPV6, 0, adv, 72); }).detach();
RR->node->putFrame(tPtr, network->id(), network->userPtr(), peerMac, from, ZT_ETHERTYPE_IPV6, 0, adv, 72);
}).detach();
return; // NDP emulation done. We have forged a "fake" reply, so no need to send actual NDP query. return; // NDP emulation done. We have forged a "fake" reply, so no need to send actual NDP query.
} // else no NDP emulation } // else no NDP emulation
@ -532,9 +530,7 @@ void Switch::onLocalEthernet(void* tPtr, const SharedPtr<Network>& network, cons
// //
// same pattern as putFrame call above // same pattern as putFrame call above
// //
std::thread([=]() { std::thread([=]() { RR->node->putFrame(tPtr, network->id(), network->userPtr(), from, to, etherType, vlanId, data, len); }).detach();
RR->node->putFrame(tPtr, network->id(), network->userPtr(), from, to, etherType, vlanId, data, len);
}).detach();
} }
else if (to[0] == MAC::firstOctetForNetwork(network->id())) { else if (to[0] == MAC::firstOctetForNetwork(network->id())) {
// Destination is another ZeroTier peer on the same network // Destination is another ZeroTier peer on the same network

35
osdep/BSDEthernetTap.cpp
View file

@ -445,7 +445,6 @@ void BSDEthernetTap::threadMain() throw()
for (unsigned int i = 0; i < _concurrency; ++i) { for (unsigned int i = 0; i < _concurrency; ++i) {
_rxThreads.push_back(std::thread([this, i, pinning] { _rxThreads.push_back(std::thread([this, i, pinning] {
if (pinning) { if (pinning) {
int pinCore = i % _concurrency; int pinCore = i % _concurrency;
fprintf(stderr, "Pinning thread %d to core %d\n", i, pinCore); fprintf(stderr, "Pinning thread %d to core %d\n", i, pinCore);
@ -453,10 +452,9 @@ void BSDEthernetTap::threadMain() throw()
cpu_set_t cpuset; cpu_set_t cpuset;
CPU_ZERO(&cpuset); CPU_ZERO(&cpuset);
CPU_SET(pinCore, &cpuset); CPU_SET(pinCore, &cpuset);
//int rc = sched_setaffinity(self, sizeof(cpu_set_t), &cpuset); // int rc = sched_setaffinity(self, sizeof(cpu_set_t), &cpuset);
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset); int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
if (rc != 0) if (rc != 0) {
{
fprintf(stderr, "Failed to pin thread %d to core %d: %s\n", i, pinCore, strerror(errno)); fprintf(stderr, "Failed to pin thread %d to core %d: %s\n", i, pinCore, strerror(errno));
exit(1); exit(1);
} }
@ -470,24 +468,25 @@ void BSDEthernetTap::threadMain() throw()
FD_ZERO(&readfds); FD_ZERO(&readfds);
FD_ZERO(&nullfds); FD_ZERO(&nullfds);
nfds = (int)std::max(_shutdownSignalPipe[0],_fd) + 1; nfds = (int)std::max(_shutdownSignalPipe[0], _fd) + 1;
r = 0; r = 0;
for(;;) { for (;;) {
FD_SET(_shutdownSignalPipe[0],&readfds); FD_SET(_shutdownSignalPipe[0], &readfds);
FD_SET(_fd,&readfds); FD_SET(_fd, &readfds);
select(nfds,&readfds,&nullfds,&nullfds,(struct timeval *)0); select(nfds, &readfds, &nullfds, &nullfds, (struct timeval*)0);
if (FD_ISSET(_shutdownSignalPipe[0],&readfds)) // writes to shutdown pipe terminate thread if (FD_ISSET(_shutdownSignalPipe[0], &readfds)) // writes to shutdown pipe terminate thread
break; break;
if (FD_ISSET(_fd,&readfds)) { if (FD_ISSET(_fd, &readfds)) {
n = (int)::read(_fd,b + r,sizeof(b) - r); n = (int)::read(_fd, b + r, sizeof(b) - r);
if (n < 0) { if (n < 0) {
if ((errno != EINTR)&&(errno != ETIMEDOUT)) if ((errno != EINTR) && (errno != ETIMEDOUT))
break; break;
} else { }
else {
// Some tap drivers like to send the ethernet frame and the // Some tap drivers like to send the ethernet frame and the
// payload in two chunks, so handle that by accumulating // payload in two chunks, so handle that by accumulating
// data until we have at least a frame. // data until we have at least a frame.
@ -497,10 +496,10 @@ void BSDEthernetTap::threadMain() throw()
r = _mtu + 14; r = _mtu + 14;
if (_enabled) { if (_enabled) {
to.setTo(b,6); to.setTo(b, 6);
from.setTo(b + 6,6); from.setTo(b + 6, 6);
unsigned int etherType = ntohs(((const uint16_t *)b)[6]); unsigned int etherType = ntohs(((const uint16_t*)b)[6]);
_handler(_arg,(void *)0,_nwid,from,to,etherType,0,(const void *)(b + 14),r - 14); _handler(_arg, (void*)0, _nwid, from, to, etherType, 0, (const void*)(b + 14), r - 14);
} }
r = 0; r = 0;

2
osdep/EthernetTap.cpp
View file

@ -132,7 +132,7 @@ std::shared_ptr<EthernetTap> EthernetTap::newInstance(
#endif // __NetBSD__ #endif // __NetBSD__
#ifdef __OpenBSD__ #ifdef __OpenBSD__
return std::shared_ptr<EthernetTap>(new BSDEthernetTap(homePath,concurrency,pinning,mac,mtu,metric,nwid,friendlyName,handler,arg)); return std::shared_ptr<EthernetTap>(new BSDEthernetTap(homePath, concurrency, pinning, mac, mtu, metric, nwid, friendlyName, handler, arg));
#endif // __OpenBSD__ #endif // __OpenBSD__
#endif // ZT_SDK? #endif // ZT_SDK?

178
tcp-proxy/tcp-proxy.cpp
View file

@ -19,32 +19,30 @@
// HACK! Will eventually use epoll() or something in Phy<> instead of select(). // HACK! Will eventually use epoll() or something in Phy<> instead of select().
// Also be sure to change ulimit -n and fs.file-max in /etc/sysctl.conf on relays. // Also be sure to change ulimit -n and fs.file-max in /etc/sysctl.conf on relays.
#if defined(__linux__) || defined(__LINUX__) || defined(__LINUX) || defined(LINUX) #if defined(__linux__) || defined(__LINUX__) || defined(__LINUX) || defined(LINUX)
#include <linux/posix_types.h>
#include <bits/types.h> #include <bits/types.h>
#include <linux/posix_types.h>
#undef __FD_SETSIZE #undef __FD_SETSIZE
#define __FD_SETSIZE 1048576 #define __FD_SETSIZE 1048576
#undef FD_SETSIZE #undef FD_SETSIZE
#define FD_SETSIZE 1048576 #define FD_SETSIZE 1048576
#endif #endif
#include "../node/Metrics.hpp"
#include "../osdep/Phy.hpp"
#include <algorithm>
#include <map>
#include <set>
#include <signal.h>
#include <stdint.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <time.h>
#include <stdint.h>
#include <unistd.h>
#include <signal.h>
#include <map>
#include <set>
#include <string> #include <string>
#include <algorithm> #include <time.h>
#include <unistd.h>
#include <vector> #include <vector>
#include "../osdep/Phy.hpp"
#include "../node/Metrics.hpp"
#define ZT_TCP_PROXY_CONNECTION_TIMEOUT_SECONDS 300 #define ZT_TCP_PROXY_CONNECTION_TIMEOUT_SECONDS 300
#define ZT_TCP_PROXY_TCP_PORT 443 #define ZT_TCP_PROXY_TCP_PORT 443
@ -88,55 +86,53 @@ using namespace ZeroTier;
*/ */
struct TcpProxyService; struct TcpProxyService;
struct TcpProxyService struct TcpProxyService {
{ Phy<TcpProxyService*>* phy;
Phy<TcpProxyService *> *phy;
int udpPortCounter; int udpPortCounter;
struct Client struct Client {
{
char tcpReadBuf[131072]; char tcpReadBuf[131072];
char tcpWriteBuf[131072]; char tcpWriteBuf[131072];
unsigned long tcpWritePtr; unsigned long tcpWritePtr;
unsigned long tcpReadPtr; unsigned long tcpReadPtr;
PhySocket *tcp; PhySocket* tcp;
PhySocket *udp; PhySocket* udp;
time_t lastActivity; time_t lastActivity;
bool newVersion; bool newVersion;
}; };
std::map< PhySocket *,Client > clients; std::map<PhySocket*, Client> clients;
PhySocket *getUnusedUdp(void *uptr) PhySocket* getUnusedUdp(void* uptr)
{ {
for(int i=0;i<65535;++i) { for (int i = 0; i < 65535; ++i) {
++udpPortCounter; ++udpPortCounter;
if (udpPortCounter > 0xfffe) if (udpPortCounter > 0xfffe)
udpPortCounter = 1024; udpPortCounter = 1024;
struct sockaddr_in laddr; struct sockaddr_in laddr;
memset(&laddr,0,sizeof(struct sockaddr_in)); memset(&laddr, 0, sizeof(struct sockaddr_in));
laddr.sin_family = AF_INET; laddr.sin_family = AF_INET;
laddr.sin_port = htons((uint16_t)udpPortCounter); laddr.sin_port = htons((uint16_t)udpPortCounter);
PhySocket *udp = phy->udpBind(reinterpret_cast<struct sockaddr *>(&laddr),uptr); PhySocket* udp = phy->udpBind(reinterpret_cast<struct sockaddr*>(&laddr), uptr);
if (udp) if (udp)
return udp; return udp;
} }
return (PhySocket *)0; return (PhySocket*)0;
} }
void phyOnDatagram(PhySocket *sock,void **uptr,const struct sockaddr *localAddr,const struct sockaddr *from,void *data,unsigned long len) void phyOnDatagram(PhySocket* sock, void** uptr, const struct sockaddr* localAddr, const struct sockaddr* from, void* data, unsigned long len)
{ {
if (!*uptr) if (! *uptr)
return; return;
if ((from->sa_family == AF_INET)&&(len >= 16)&&(len < 2048)) { if ((from->sa_family == AF_INET) && (len >= 16) && (len < 2048)) {
Client &c = *((Client *)*uptr); Client& c = *((Client*)*uptr);
c.lastActivity = time((time_t *)0); c.lastActivity = time((time_t*)0);
unsigned long mlen = len; unsigned long mlen = len;
if (c.newVersion) if (c.newVersion)
mlen += 7; // new clients get IP info mlen += 7; // new clients get IP info
if ((c.tcpWritePtr + 5 + mlen) <= sizeof(c.tcpWriteBuf)) { if ((c.tcpWritePtr + 5 + mlen) <= sizeof(c.tcpWriteBuf)) {
if (!c.tcpWritePtr) if (! c.tcpWritePtr)
phy->setNotifyWritable(c.tcp,true); phy->setNotifyWritable(c.tcp, true);
c.tcpWriteBuf[c.tcpWritePtr++] = 0x17; // look like TLS data c.tcpWriteBuf[c.tcpWritePtr++] = 0x17; // look like TLS data
c.tcpWriteBuf[c.tcpWritePtr++] = 0x03; // look like TLS 1.2 c.tcpWriteBuf[c.tcpWritePtr++] = 0x03; // look like TLS 1.2
@ -147,30 +143,30 @@ struct TcpProxyService
if (c.newVersion) { if (c.newVersion) {
c.tcpWriteBuf[c.tcpWritePtr++] = (char)4; // IPv4 c.tcpWriteBuf[c.tcpWritePtr++] = (char)4; // IPv4
*((uint32_t *)(c.tcpWriteBuf + c.tcpWritePtr)) = ((const struct sockaddr_in *)from)->sin_addr.s_addr; *((uint32_t*)(c.tcpWriteBuf + c.tcpWritePtr)) = ((const struct sockaddr_in*)from)->sin_addr.s_addr;
c.tcpWritePtr += 4; c.tcpWritePtr += 4;
*((uint16_t *)(c.tcpWriteBuf + c.tcpWritePtr)) = ((const struct sockaddr_in *)from)->sin_port; *((uint16_t*)(c.tcpWriteBuf + c.tcpWritePtr)) = ((const struct sockaddr_in*)from)->sin_port;
c.tcpWritePtr += 2; c.tcpWritePtr += 2;
} }
for(unsigned long i=0;i<len;++i) for (unsigned long i = 0; i < len; ++i)
c.tcpWriteBuf[c.tcpWritePtr++] = ((const char *)data)[i]; c.tcpWriteBuf[c.tcpWritePtr++] = ((const char*)data)[i];
} }
printf("<< UDP %s:%d -> %.16llx\n",inet_ntoa(reinterpret_cast<const struct sockaddr_in *>(from)->sin_addr),(int)ntohs(reinterpret_cast<const struct sockaddr_in *>(from)->sin_port),(unsigned long long)&c); printf("<< UDP %s:%d -> %.16llx\n", inet_ntoa(reinterpret_cast<const struct sockaddr_in*>(from)->sin_addr), (int)ntohs(reinterpret_cast<const struct sockaddr_in*>(from)->sin_port), (unsigned long long)&c);
} }
} }
void phyOnTcpConnect(PhySocket *sock,void **uptr,bool success) void phyOnTcpConnect(PhySocket* sock, void** uptr, bool success)
{ {
// unused, we don't initiate outbound connections // unused, we don't initiate outbound connections
} }
void phyOnTcpAccept(PhySocket *sockL,PhySocket *sockN,void **uptrL,void **uptrN,const struct sockaddr *from) void phyOnTcpAccept(PhySocket* sockL, PhySocket* sockN, void** uptrL, void** uptrN, const struct sockaddr* from)
{ {
Client &c = clients[sockN]; Client& c = clients[sockN];
PhySocket *udp = getUnusedUdp((void *)&c); PhySocket* udp = getUnusedUdp((void*)&c);
if (!udp) { if (! udp) {
phy->close(sockN); phy->close(sockN);
clients.erase(sockN); clients.erase(sockN);
printf("** TCP rejected, no more UDP ports to assign\n"); printf("** TCP rejected, no more UDP ports to assign\n");
@ -180,59 +176,61 @@ struct TcpProxyService
c.tcpReadPtr = 0; c.tcpReadPtr = 0;
c.tcp = sockN; c.tcp = sockN;
c.udp = udp; c.udp = udp;
c.lastActivity = time((time_t *)0); c.lastActivity = time((time_t*)0);
c.newVersion = false; c.newVersion = false;
*uptrN = (void *)&c; *uptrN = (void*)&c;
printf("<< TCP from %s -> %.16llx\n",inet_ntoa(reinterpret_cast<const struct sockaddr_in *>(from)->sin_addr),(unsigned long long)&c); printf("<< TCP from %s -> %.16llx\n", inet_ntoa(reinterpret_cast<const struct sockaddr_in*>(from)->sin_addr), (unsigned long long)&c);
} }
void phyOnTcpClose(PhySocket *sock,void **uptr) void phyOnTcpClose(PhySocket* sock, void** uptr)
{ {
if (!*uptr) if (! *uptr)
return; return;
Client &c = *((Client *)*uptr); Client& c = *((Client*)*uptr);
phy->close(c.udp); phy->close(c.udp);
clients.erase(sock); clients.erase(sock);
printf("** TCP %.16llx closed\n",(unsigned long long)*uptr); printf("** TCP %.16llx closed\n", (unsigned long long)*uptr);
} }
void phyOnTcpData(PhySocket *sock,void **uptr,void *data,unsigned long len) void phyOnTcpData(PhySocket* sock, void** uptr, void* data, unsigned long len)
{ {
Client &c = *((Client *)*uptr); Client& c = *((Client*)*uptr);
c.lastActivity = time((time_t *)0); c.lastActivity = time((time_t*)0);
for(unsigned long i=0;i<len;++i) { for (unsigned long i = 0; i < len; ++i) {
if (c.tcpReadPtr >= sizeof(c.tcpReadBuf)) { if (c.tcpReadPtr >= sizeof(c.tcpReadBuf)) {
phy->close(sock); phy->close(sock);
return; return;
} }
c.tcpReadBuf[c.tcpReadPtr++] = ((const char *)data)[i]; c.tcpReadBuf[c.tcpReadPtr++] = ((const char*)data)[i];
if (c.tcpReadPtr >= 5) { if (c.tcpReadPtr >= 5) {
unsigned long mlen = ( ((((unsigned long)c.tcpReadBuf[3]) & 0xff) << 8) | (((unsigned long)c.tcpReadBuf[4]) & 0xff) ); unsigned long mlen = (((((unsigned long)c.tcpReadBuf[3]) & 0xff) << 8) | (((unsigned long)c.tcpReadBuf[4]) & 0xff));
if (c.tcpReadPtr >= (mlen + 5)) { if (c.tcpReadPtr >= (mlen + 5)) {
if (mlen == 4) { if (mlen == 4) {
// Right now just sending this means the client is 'new enough' for the IP header // Right now just sending this means the client is 'new enough' for the IP header
c.newVersion = true; c.newVersion = true;
printf("<< TCP %.16llx HELLO\n",(unsigned long long)*uptr); printf("<< TCP %.16llx HELLO\n", (unsigned long long)*uptr);
} else if (mlen >= 7) { }
char *payload = c.tcpReadBuf + 5; else if (mlen >= 7) {
char* payload = c.tcpReadBuf + 5;
unsigned long payloadLen = mlen; unsigned long payloadLen = mlen;
struct sockaddr_in dest; struct sockaddr_in dest;
memset(&dest,0,sizeof(dest)); memset(&dest, 0, sizeof(dest));
if (c.newVersion) { if (c.newVersion) {
if (*payload == (char)4) { if (*payload == (char)4) {
// New clients tell us where their packets go. // New clients tell us where their packets go.
++payload; ++payload;
dest.sin_family = AF_INET; dest.sin_family = AF_INET;
dest.sin_addr.s_addr = *((uint32_t *)payload); dest.sin_addr.s_addr = *((uint32_t*)payload);
payload += 4; payload += 4;
dest.sin_port = *((uint16_t *)payload); // will be in network byte order already dest.sin_port = *((uint16_t*)payload); // will be in network byte order already
payload += 2; payload += 2;
payloadLen -= 7; payloadLen -= 7;
} }
} else { }
else {
// For old clients we will just proxy everything to a local ZT instance. The // For old clients we will just proxy everything to a local ZT instance. The
// fact that this will come from 127.0.0.1 will in turn prevent that instance // fact that this will come from 127.0.0.1 will in turn prevent that instance
// from doing unite() with us. It'll just forward. There will not be many of // from doing unite() with us. It'll just forward. There will not be many of
@ -243,72 +241,74 @@ struct TcpProxyService
} }
// Note: we do not relay to privileged ports... just an abuse prevention rule. // Note: we do not relay to privileged ports... just an abuse prevention rule.
if ((ntohs(dest.sin_port) > 1024)&&(payloadLen >= 16)) { if ((ntohs(dest.sin_port) > 1024) && (payloadLen >= 16)) {
phy->udpSend(c.udp,(const struct sockaddr *)&dest,payload,payloadLen); phy->udpSend(c.udp, (const struct sockaddr*)&dest, payload, payloadLen);
printf(">> TCP %.16llx to %s:%d\n",(unsigned long long)*uptr,inet_ntoa(dest.sin_addr),(int)ntohs(dest.sin_port)); printf(">> TCP %.16llx to %s:%d\n", (unsigned long long)*uptr, inet_ntoa(dest.sin_addr), (int)ntohs(dest.sin_port));
} }
} }
memmove(c.tcpReadBuf,c.tcpReadBuf + (mlen + 5),c.tcpReadPtr -= (mlen + 5)); memmove(c.tcpReadBuf, c.tcpReadBuf + (mlen + 5), c.tcpReadPtr -= (mlen + 5));
} }
} }
} }
} }
void phyOnTcpWritable(PhySocket *sock,void **uptr) void phyOnTcpWritable(PhySocket* sock, void** uptr)
{ {
Client &c = *((Client *)*uptr); Client& c = *((Client*)*uptr);
if (c.tcpWritePtr) { if (c.tcpWritePtr) {
long n = phy->streamSend(sock,c.tcpWriteBuf,c.tcpWritePtr); long n = phy->streamSend(sock, c.tcpWriteBuf, c.tcpWritePtr);
if (n > 0) { if (n > 0) {
memmove(c.tcpWriteBuf,c.tcpWriteBuf + n,c.tcpWritePtr -= (unsigned long)n); memmove(c.tcpWriteBuf, c.tcpWriteBuf + n, c.tcpWritePtr -= (unsigned long)n);
if (!c.tcpWritePtr) if (! c.tcpWritePtr)
phy->setNotifyWritable(sock,false); phy->setNotifyWritable(sock, false);
} }
} else phy->setNotifyWritable(sock,false); }
else
phy->setNotifyWritable(sock, false);
} }
void doHousekeeping() void doHousekeeping()
{ {
std::vector<PhySocket *> toClose; std::vector<PhySocket*> toClose;
time_t now = time((time_t *)0); time_t now = time((time_t*)0);
for(std::map< PhySocket *,Client >::iterator c(clients.begin());c!=clients.end();++c) { for (std::map<PhySocket*, Client>::iterator c(clients.begin()); c != clients.end(); ++c) {
if ((now - c->second.lastActivity) >= ZT_TCP_PROXY_CONNECTION_TIMEOUT_SECONDS) { if ((now - c->second.lastActivity) >= ZT_TCP_PROXY_CONNECTION_TIMEOUT_SECONDS) {
toClose.push_back(c->first); toClose.push_back(c->first);
toClose.push_back(c->second.udp); toClose.push_back(c->second.udp);
} }
} }
for(std::vector<PhySocket *>::iterator s(toClose.begin());s!=toClose.end();++s) for (std::vector<PhySocket*>::iterator s(toClose.begin()); s != toClose.end(); ++s)
phy->close(*s); phy->close(*s);
} }
}; };
int main(int argc,char **argv) int main(int argc, char** argv)
{ {
signal(SIGPIPE,SIG_IGN); signal(SIGPIPE, SIG_IGN);
signal(SIGHUP,SIG_IGN); signal(SIGHUP, SIG_IGN);
srand(time((time_t *)0)); srand(time((time_t*)0));
TcpProxyService svc; TcpProxyService svc;
Phy<TcpProxyService *> phy(&svc,false,true); Phy<TcpProxyService*> phy(&svc, false, true);
svc.phy = &phy; svc.phy = &phy;
svc.udpPortCounter = 1023; svc.udpPortCounter = 1023;
{ {
struct sockaddr_in laddr; struct sockaddr_in laddr;
memset(&laddr,0,sizeof(laddr)); memset(&laddr, 0, sizeof(laddr));
laddr.sin_family = AF_INET; laddr.sin_family = AF_INET;
laddr.sin_port = htons(ZT_TCP_PROXY_TCP_PORT); laddr.sin_port = htons(ZT_TCP_PROXY_TCP_PORT);
if (!phy.tcpListen((const struct sockaddr *)&laddr)) { if (! phy.tcpListen((const struct sockaddr*)&laddr)) {
fprintf(stderr,"%s: fatal error: unable to bind TCP port %d\n",argv[0],ZT_TCP_PROXY_TCP_PORT); fprintf(stderr, "%s: fatal error: unable to bind TCP port %d\n", argv[0], ZT_TCP_PROXY_TCP_PORT);
return 1; return 1;
} }
} }
time_t lastDidHousekeeping = time((time_t *)0); time_t lastDidHousekeeping = time((time_t*)0);
for(;;) { for (;;) {
phy.poll(120000); phy.poll(120000);
time_t now = time((time_t *)0); time_t now = time((time_t*)0);
if ((now - lastDidHousekeeping) > 120) { if ((now - lastDidHousekeeping) > 120) {
lastDidHousekeeping = now; lastDidHousekeeping = now;
svc.doHousekeeping(); svc.doHousekeeping();