From 5028aca3722fbdc989a904a3ffe7d07392ab0add Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Fri, 18 Jun 2021 16:14:59 -0400 Subject: [PATCH] Fix some identity verification stuff, performance improvements, build of root now requires libjemalloc. --- root/CMakeLists.txt | 2 +- root/root.cpp | 218 ++++++++++++++++++++++++++++---------------- 2 files changed, 142 insertions(+), 78 deletions(-) diff --git a/root/CMakeLists.txt b/root/CMakeLists.txt index a89e4807..593bc131 100644 --- a/root/CMakeLists.txt +++ b/root/CMakeLists.txt @@ -7,7 +7,7 @@ endif(WIN32) add_executable(${PROJECT_NAME} root.cpp) -target_link_libraries(${PROJECT_NAME} zt_core zt_osdep pthread resolv) +target_link_libraries(${PROJECT_NAME} zt_core zt_osdep pthread resolv jemalloc) target_compile_features(${PROJECT_NAME} PUBLIC cxx_std_11) diff --git a/root/root.cpp b/root/root.cpp index 7c77bccd..c54f5869 100644 --- a/root/root.cpp +++ b/root/root.cpp @@ -48,7 +48,8 @@ * they appear with the first alive sibling being used. */ -#include + +#include "../node/Constants.hpp" #include #include @@ -71,20 +72,22 @@ #include #include -#include -#include +#include "../ext/json/json.hpp" +#include "../ext/cpp-httplib/httplib.h" -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include +#include "../node/Packet.hpp" +#include "../node/Utils.hpp" +#include "../node/Address.hpp" +#include "../node/Identity.hpp" +#include "../node/InetAddress.hpp" +#include "../node/Mutex.hpp" +#include "../node/SharedPtr.hpp" +#include "../node/MulticastGroup.hpp" +#include "../node/CertificateOfMembership.hpp" +#include "../node/Meter.hpp" + +#include "../osdep/OSUtils.hpp" +#include "../osdep/BlockingQueue.hpp" #include #include @@ -126,7 +129,7 @@ using json = nlohmann::json; */ struct RootPeer { - ZT_ALWAYS_INLINE RootPeer() : v4s(-1),v6s(-1),lastSend(0),lastReceive(0),lastReceiveV4(0),lastReceiveV6(0),lastEcho(0),lastHello(0),vProto(-1),vMajor(-1),vMinor(-1),vRev(-1) {} + ZT_ALWAYS_INLINE RootPeer() : v4s(-1),v6s(-1),lastSend(0),lastReceive(0),lastReceiveV4(0),lastReceiveV6(0),lastEcho(0),lastHello(0),vProto(-1),vMajor(-1),vMinor(-1),vRev(-1),identityValidated(false),identityInvalid(false) {} ZT_ALWAYS_INLINE ~RootPeer() { Utils::burn(key,sizeof(key)); } Identity id; // Identity @@ -141,6 +144,8 @@ struct RootPeer int64_t lastHello; // Time of last received HELLO int vProto; // Protocol version or -1 if unknown int vMajor,vMinor,vRev; // Peer version or -1,-1,-1 if unknown + bool identityValidated; // Identity has been fully verified + bool identityInvalid; // Identity validation failed, to be deleted AtomicCounter __refCount; }; @@ -200,7 +205,8 @@ static Meter s_discardedForwardRate; // These fields are locked using mutexes below as they're modified during runtime static std::string s_planet; -static std::list< SharedPtr > s_peers; +static std::vector< SharedPtr > s_peers; +static std::vector< SharedPtr > s_peersToValidate; static std::unordered_map< uint64_t,std::unordered_map< MulticastGroup,std::unordered_map< Address,int64_t,AddressHasher >,MulticastGroupHasher > > s_multicastSubscriptions; static std::unordered_map< Identity,SharedPtr,IdentityHasher > s_peersByIdentity; static std::unordered_map< Address,std::set< SharedPtr >,AddressHasher > s_peersByVirtAddr; @@ -208,6 +214,7 @@ static std::unordered_map< RendezvousKey,RendezvousStats,RendezvousKey::Hasher > static std::mutex s_planet_l; static std::mutex s_peers_l; +static std::mutex s_peersToValidate_l; static std::mutex s_multicastSubscriptions_l; static std::mutex s_peersByIdentity_l; static std::mutex s_peersByVirtAddr_l; @@ -262,12 +269,30 @@ static void handlePacket(const int sock,const InetAddress *const ip,Packet &pkt) } } if (peer) { + // Peer found with this identity. if (!pkt.dearmor(peer->key)) { printf("%s HELLO rejected: packet authentication failed" ZT_EOL_S,ip->toString(ipstr)); return; } } else { + // Check to ensure that there is no peer with the same address as this identity. If there is, + // verify both identities to pick the one with this address. + bool needsValidation = false; + bool identityValidated = false; + { + std::lock_guard pbv_l(s_peersByVirtAddr_l); + needsValidation = s_peersByVirtAddr.find(id.address()) != s_peersByVirtAddr.end(); + } + if (unlikely(needsValidation)) { + if (!id.locallyValidate()) { + printf("%s HELLO rejected: identity validate failed" ZT_EOL_S,ip->toString(ipstr)); + return; + } + identityValidated = true; + } + peer.set(new RootPeer); + peer->identityValidated = identityValidated; if (!s_self.agree(id,peer->key)) { printf("%s HELLO rejected: key agreement failed" ZT_EOL_S,ip->toString(ipstr)); @@ -289,20 +314,29 @@ static void handlePacket(const int sock,const InetAddress *const ip,Packet &pkt) { std::lock_guard pbi_l(s_peersByIdentity_l); auto existing = s_peersByIdentity.find(id); // make sure another thread didn't do this while we were - if (existing == s_peersByIdentity.end()) { + if (likely(existing == s_peersByIdentity.end())) { s_peersByIdentity.emplace(id,peer); added = true; } else { peer = existing->second; } } - if (added) { + if (likely(added)) { { std::lock_guard pl(s_peers_l); s_peers.emplace_back(peer); } + if (!peer->identityValidated) { + std::lock_guard pv(s_peersToValidate_l); + s_peersToValidate.emplace_back(peer); + } { std::lock_guard pbv_l(s_peersByVirtAddr_l); + std::set< SharedPtr > &byVirt = s_peersByVirtAddr[id.address()]; + for(auto i=byVirt.begin();i!=byVirt.end();++i) { + if (!(*i)->identityValidated) + (*i)->identityInvalid = !(*i)->id.locallyValidate(); + } s_peersByVirtAddr[id.address()].emplace(peer); } } @@ -317,13 +351,15 @@ static void handlePacket(const int sock,const InetAddress *const ip,Packet &pkt) auto peers = s_peersByVirtAddr.find(source); if (peers != s_peersByVirtAddr.end()) { for(auto p=peers->second.begin();p!=peers->second.end();++p) { - if (pkt.dearmor((*p)->key)) { - if (!pkt.uncompress()) { - printf("%s packet rejected: decompression failed" ZT_EOL_S,ip->toString(ipstr)); - return; + if (!(*p)->identityInvalid) { + if (pkt.dearmor((*p)->key)) { + if (!pkt.uncompress()) { + printf("%s packet rejected: decompression failed" ZT_EOL_S,ip->toString(ipstr)); + return; + } + peer = (*p); + break; } - peer = (*p); - break; } } } @@ -418,8 +454,10 @@ static void handlePacket(const int sock,const InetAddress *const ip,Packet &pkt) for(unsigned int ptr=ZT_PACKET_IDX_PAYLOAD;(ptr+ZT_ADDRESS_LENGTH)<=pkt.size();ptr+=ZT_ADDRESS_LENGTH) { auto peers = s_peersByVirtAddr.find(Address(pkt.field(ptr,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH)); if (peers != s_peersByVirtAddr.end()) { - for(auto p=peers->second.begin();p!=peers->second.end();++p) - results.push_back(*p); + for(auto p=peers->second.begin();p!=peers->second.end();++p) { + if (!(*p)->identityInvalid) + results.push_back(*p); + } } } } @@ -550,16 +588,18 @@ static void handlePacket(const int sock,const InetAddress *const ip,Packet &pkt) auto peers = s_peersByVirtAddr.find(dest); if (peers != s_peersByVirtAddr.end()) { for(auto p=peers->second.begin();p!=peers->second.end();++p) { - if (((*p)->v4s >= 0)&&((*p)->v6s >= 0)) { - if ((*p)->lastReceiveV4 > (*p)->lastReceiveV6) { + if (!(*p)->identityInvalid) { + if (((*p)->v4s >= 0)&&((*p)->v6s >= 0)) { + if ((*p)->lastReceiveV4 > (*p)->lastReceiveV6) { + toAddrs.emplace_back(std::pair< InetAddress *,SharedPtr >(&((*p)->ip4),*p)); + } else { + toAddrs.emplace_back(std::pair< InetAddress *,SharedPtr >(&((*p)->ip6),*p)); + } + } else if ((*p)->v4s >= 0) { toAddrs.emplace_back(std::pair< InetAddress *,SharedPtr >(&((*p)->ip4),*p)); - } else { + } else if ((*p)->v6s >= 0) { toAddrs.emplace_back(std::pair< InetAddress *,SharedPtr >(&((*p)->ip6),*p)); } - } else if ((*p)->v4s >= 0) { - toAddrs.emplace_back(std::pair< InetAddress *,SharedPtr >(&((*p)->ip4),*p)); - } else if ((*p)->v6s >= 0) { - toAddrs.emplace_back(std::pair< InetAddress *,SharedPtr >(&((*p)->ip6),*p)); } } } @@ -895,6 +935,20 @@ int main(int argc,char **argv) s_run = true; + threads.push_back(std::thread([]() { + while (s_run) { + std::vector< SharedPtr > toValidate; + { + std::lock_guard l(s_peersToValidate_l); + toValidate.swap(s_peersToValidate); + } + for(auto p=toValidate.begin();p!=toValidate.end();++p) { + if (!(*p)->identityValidated) + (*p)->identityInvalid = !(*p)->id.locallyValidate(); + } + } + })); + for(auto port=s_ports.begin();port!=s_ports.end();++port) { for(unsigned int tn=0;tn l(s_peers_l); for(auto p=s_peers.begin();p!=s_peers.end();++p) { - if (first) - first = false; - else o << ','; - o << - "{\"address\":\"" << (*p)->id.address().toString(tmp) << "\"" - ",\"latency\":-1" - ",\"paths\":["; - if ((*p)->v4s >= 0) { + if (likely(!(*p)->identityInvalid)) { + if (first) + first = false; + else o << ','; o << - "{\"active\":true" - ",\"address\":\"" << (*p)->ip4.toIpString(tmp) << "\\/" << (*p)->ip4.port() << "\"" - ",\"expired\":false" - ",\"lastReceive\":" << (*p)->lastReceive << - ",\"lastSend\":" << (*p)->lastSend << - ",\"preferred\":true" - ",\"trustedPathId\":0}"; + "{\"address\":\"" << (*p)->id.address().toString(tmp) << "\"" + ",\"latency\":-1" + ",\"paths\":["; + if ((*p)->v4s >= 0) { + o << + "{\"active\":true" + ",\"address\":\"" << (*p)->ip4.toIpString(tmp) << "\\/" << (*p)->ip4.port() << "\"" + ",\"expired\":false" + ",\"lastReceive\":" << (*p)->lastReceive << + ",\"lastSend\":" << (*p)->lastSend << + ",\"preferred\":true" + ",\"trustedPathId\":0}"; + } + if ((*p)->v6s >= 0) { + if ((*p)->v4s >= 0) + o << ','; + o << + "{\"active\":true" + ",\"address\":\"" << (*p)->ip6.toIpString(tmp) << "\\/" << (*p)->ip6.port() << "\"" + ",\"expired\":false" + ",\"lastReceive\":" << (*p)->lastReceive << + ",\"lastSend\":" << (*p)->lastSend << + ",\"preferred\":" << (((*p)->ip4) ? "false" : "true") << + ",\"trustedPathId\":0}"; + } + o << "]" + ",\"role\":\"LEAF\"" + ",\"version\":\"" << (*p)->vMajor << '.' << (*p)->vMinor << '.' << (*p)->vRev << "\"" + ",\"versionMajor\":" << (*p)->vMajor << + ",\"versionMinor\":" << (*p)->vMinor << + ",\"versionRev\":" << (*p)->vRev << "}"; } - if ((*p)->v6s >= 0) { - if ((*p)->v4s >= 0) - o << ','; - o << - "{\"active\":true" - ",\"address\":\"" << (*p)->ip6.toIpString(tmp) << "\\/" << (*p)->ip6.port() << "\"" - ",\"expired\":false" - ",\"lastReceive\":" << (*p)->lastReceive << - ",\"lastSend\":" << (*p)->lastSend << - ",\"preferred\":" << (((*p)->ip4) ? "false" : "true") << - ",\"trustedPathId\":0}"; - } - o << "]" - ",\"role\":\"LEAF\"" - ",\"version\":\"" << (*p)->vMajor << '.' << (*p)->vMinor << '.' << (*p)->vRev << "\"" - ",\"versionMajor\":" << (*p)->vMajor << - ",\"versionMinor\":" << (*p)->vMinor << - ",\"versionRev\":" << (*p)->vRev << "}"; } } catch ( ... ) {} o << ']'; @@ -1094,10 +1150,12 @@ int main(int argc,char **argv) { std::lock_guard l(s_peers_l); for(auto p=s_peers.begin();p!=s_peers.end();++p) { - if ((*p)->v4s >= 0) - ips[(*p)->ip4].insert((*p)->id.address()); - if ((*p)->v6s >= 0) - ips[(*p)->ip6].insert((*p)->id.address()); + if (likely(!(*p)->identityInvalid)) { + if ((*p)->v4s >= 0) + ips[(*p)->ip4].insert((*p)->id.address()); + if ((*p)->v6s >= 0) + ips[(*p)->ip6].insert((*p)->id.address()); + } } } @@ -1201,18 +1259,24 @@ int main(int argc,char **argv) } } - // Remove expired peers + // Remove expired or otherwise invalid peers try { std::vector< SharedPtr > toRemove; toRemove.reserve(1024); { std::lock_guard pbi_l(s_peers_l); - for(auto p=s_peers.begin();p!=s_peers.end();) { - if ((now - (*p)->lastReceive) > ZT_PEER_ACTIVITY_TIMEOUT) { - toRemove.emplace_back(*p); - s_peers.erase(p++); - } else ++p; + std::vector< SharedPtr > newPeers; + newPeers.reserve(s_peers.size()); + for(auto p=s_peers.begin();p!=s_peers.end();++p) { + if (((now - (*p)->lastReceive) > ZT_PEER_ACTIVITY_TIMEOUT)||((*p)->identityInvalid)) { + toRemove.emplace_back(); + p->swap(toRemove.back()); + } else { + newPeers.emplace_back(); + p->swap(newPeers.back()); + } } + newPeers.swap(s_peers); } for(auto p=toRemove.begin();p!=toRemove.end();++p) { { @@ -1316,7 +1380,7 @@ int main(int argc,char **argv) s_peersByIdentity_l.unlock(); fprintf(sf,"Peers : %llu" ZT_EOL_S,(unsigned long long)peersByIdentitySize); s_peersByVirtAddr_l.lock(); - fprintf(sf,"Virtual Address Collisions : %llu" ZT_EOL_S,(unsigned long long)(peersByIdentitySize - s_peersByVirtAddr.size())); + fprintf(sf,"Virtual Address Collisions : %lld" ZT_EOL_S,(long long)peersByIdentitySize - (long long)s_peersByVirtAddr.size()); s_peersByVirtAddr_l.unlock(); s_rendezvousTracking_l.lock(); uint64_t unsuccessfulp2p = 0;