summaryrefslogtreecommitdiff
path: root/node
diff options
context:
space:
mode:
authorAdam Ierymenko <adam.ierymenko@gmail.com>2013-07-11 22:06:25 -0400
committerAdam Ierymenko <adam.ierymenko@gmail.com>2013-07-11 22:06:25 -0400
commit339b2314ea168d302bae4b5f7fbf828b210e5450 (patch)
tree5ba92262bfbeacd58935f7e332557086f6be53a0 /node
parentfd2b383c3e8e1a4b81454611b966c6c775c03913 (diff)
downloadinfinitytier-339b2314ea168d302bae4b5f7fbf828b210e5450.tar.gz
infinitytier-339b2314ea168d302bae4b5f7fbf828b210e5450.zip
More work in progress on Switch / PacketDecoder refactor.
Diffstat (limited to 'node')
-rw-r--r--node/Node.cpp3
-rw-r--r--node/PacketDecoder.cpp423
-rw-r--r--node/PacketDecoder.hpp38
-rw-r--r--node/Peer.cpp6
-rw-r--r--node/Peer.hpp38
-rw-r--r--node/RuntimeEnvironment.hpp3
-rw-r--r--node/Switch.cpp73
-rw-r--r--node/Switch.hpp53
8 files changed, 344 insertions, 293 deletions
diff --git a/node/Node.cpp b/node/Node.cpp
index ab7794b8..4b5a43ed 100644
--- a/node/Node.cpp
+++ b/node/Node.cpp
@@ -64,6 +64,7 @@
#include "Network.hpp"
#include "MulticastGroup.hpp"
#include "Mutex.hpp"
+#include "Multicaster.hpp"
#include "../version.h"
@@ -116,6 +117,7 @@ Node::~Node()
delete impl->renv.sysEnv;
delete impl->renv.topology;
delete impl->renv.sw;
+ delete impl->renv.multicaster;
delete impl->renv.demarc;
delete impl->renv.nc;
delete impl->renv.log;
@@ -211,6 +213,7 @@ Node::ReasonForTermination Node::run()
// watcher.
_r->nc = new NodeConfig(_r,_r->autoconfUrlPrefix + _r->identity.address().toString());
_r->demarc = new Demarc(_r);
+ _r->multicaster = new Multicaster();
_r->sw = new Switch(_r);
_r->topology = new Topology(_r,(_r->homePath + ZT_PATH_SEPARATOR_S + "peer.db").c_str());
_r->sysEnv = new SysEnv(_r);
diff --git a/node/PacketDecoder.cpp b/node/PacketDecoder.cpp
index 0fba9d86..93126e16 100644
--- a/node/PacketDecoder.cpp
+++ b/node/PacketDecoder.cpp
@@ -30,33 +30,33 @@
#include "PacketDecoder.hpp"
#include "Switch.hpp"
#include "Peer.hpp"
+#include "NodeConfig.hpp"
+#include "Filter.hpp"
namespace ZeroTier {
bool PacketDecoder::tryDecode(const RuntimeEnvironment *_r)
throw(std::out_of_range,std::runtime_error)
{
- Address source(source());
-
if ((!encrypted())&&(verb() == Packet::VERB_HELLO)) {
// Unencrypted HELLOs are handled here since they are used to
// populate our identity cache in the first place. Thus we might get
// a HELLO for someone for whom we don't have a Peer record.
- TRACE("HELLO from %s(%s)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("HELLO from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
return _doHELLO(_r);
}
- if (_step == DECODE_STEP_WAITING_FOR_ORIGINAL_SUBMITTER_LOOKUP) {
- // This means we've already decoded, decrypted, decompressed, and
- // validated, and we're processing a MULTICAST_FRAME. We're waiting
- // for a lookup on the frame's original submitter.
- return _doMULTICAST_FRAME(_r);
- }
-
- SharedPtr<Peer> peer = _r->topology->getPeer(source);
+ SharedPtr<Peer> peer = _r->topology->getPeer(source());
if (peer) {
+ if (_step == DECODE_STEP_WAITING_FOR_ORIGINAL_SUBMITTER_LOOKUP) {
+ // This means we've already decoded, decrypted, decompressed, and
+ // validated, and we're processing a MULTICAST_FRAME. We're waiting
+ // for a lookup on the frame's original submitter.
+ return _doMULTICAST_FRAME(_r,peer);
+ }
+
if (!hmacVerify(peer->macKey())) {
- TRACE("dropped packet from %s(%s), HMAC authentication failed (size: %u)",source.toString().c_str(),_remoteAddress.toString().c_str(),size());
+ TRACE("dropped packet from %s(%s), HMAC authentication failed (size: %u)",source().toString().c_str(),_remoteAddress.toString().c_str(),size());
return true;
}
@@ -66,58 +66,118 @@ bool PacketDecoder::tryDecode(const RuntimeEnvironment *_r)
// Unencrypted is tolerated in case we want to run this on
// devices where squeezing out cycles matters. HMAC is
// what's really important.
- TRACE("ODD: %s from %s(%s) wasn't encrypted",Packet::verbString(verb()),source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("ODD: %s from %s(%s) wasn't encrypted",Packet::verbString(verb()),source().toString().c_str(),_remoteAddress.toString().c_str());
}
if (!uncompress()) {
- TRACE("dropped packet from %s(%s), compressed data invalid",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped packet from %s(%s), compressed data invalid",source().toString().c_str(),_remoteAddress.toString().c_str());
return true;
}
+ Packet::Verb v = verb();
+
// Validated packets that have passed HMAC can result in us learning a new
// path to this peer.
- peer->onReceive(_r,localPort,_remoteAddress,hops(),verb(),Utils::now());
+ peer->onReceive(_r,_localPort,_remoteAddress,hops(),v,Utils::now());
- switch(verb()) {
+ switch(v) {
case Packet::VERB_NOP:
- TRACE("NOP from %s(%s)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("NOP from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
return true;
case Packet::VERB_HELLO:
return _doHELLO(_r);
case Packet::VERB_ERROR:
- return _doERROR(_r);
+ return _doERROR(_r,peer);
case Packet::VERB_OK:
- return _doOK(_r);
+ return _doOK(_r,peer);
case Packet::VERB_WHOIS:
- return _doWHOIS(_r);
+ return _doWHOIS(_r,peer);
case Packet::VERB_RENDEZVOUS:
- return _doRENDEZVOUS(_r);
+ return _doRENDEZVOUS(_r,peer);
case Packet::VERB_FRAME:
- return _doFRAME(_r);
+ return _doFRAME(_r,peer);
case Packet::VERB_MULTICAST_LIKE:
- return _doMULTICAST_LIKE(_r);
+ return _doMULTICAST_LIKE(_r,peer);
case Packet::VERB_MULTICAST_FRAME:
- return _doMULTICAST_FRAME(_r);
+ return _doMULTICAST_FRAME(_r,peer);
default:
// This might be something from a new or old version of the protocol.
// Technically it passed HMAC so the packet is still valid, but we
// ignore it.
- TRACE("ignored unrecognized verb %.2x from %s(%s)",(unsigned int)packet.verb(),source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("ignored unrecognized verb %.2x from %s(%s)",(unsigned int)v,source().toString().c_str(),_remoteAddress.toString().c_str());
return true;
}
} else {
- _r->sw->requestWhois(source);
+ _r->sw->requestWhois(source());
return false;
}
}
-bool PacketDecoder::_doERROR(const RuntimeEnvironment *_r)
+void PacketDecoder::_CBaddPeerFromHello(void *arg,const SharedPtr<Peer> &p,Topology::PeerVerifyResult result)
+{
+ _CBaddPeerFromHello_Data *req = (_CBaddPeerFromHello_Data *)arg;
+ const RuntimeEnvironment *_r = req->renv;
+
+ switch(result) {
+ case Topology::PEER_VERIFY_ACCEPTED_NEW:
+ case Topology::PEER_VERIFY_ACCEPTED_ALREADY_HAVE:
+ case Topology::PEER_VERIFY_ACCEPTED_DISPLACED_INVALID_ADDRESS: {
+ _r->sw->doAnythingWaitingForPeer(p);
+
+ Packet outp(req->source,_r->identity.address(),Packet::VERB_OK);
+ outp.append((unsigned char)Packet::VERB_HELLO);
+ outp.append(req->helloPacketId);
+ outp.append(req->helloTimestamp);
+ outp.encrypt(p->cryptKey());
+ outp.hmacSet(p->macKey());
+ _r->demarc->send(req->localPort,req->remoteAddress,outp.data(),outp.size(),-1);
+ } break;
+
+ case Topology::PEER_VERIFY_REJECTED_INVALID_IDENTITY: {
+ Packet outp(req->source,_r->identity.address(),Packet::VERB_ERROR);
+ outp.append((unsigned char)Packet::VERB_HELLO);
+ outp.append(req->helloPacketId);
+ outp.append((unsigned char)Packet::ERROR_IDENTITY_INVALID);
+ outp.encrypt(p->cryptKey());
+ outp.hmacSet(p->macKey());
+ _r->demarc->send(req->localPort,req->remoteAddress,outp.data(),outp.size(),-1);
+ } break;
+
+ case Topology::PEER_VERIFY_REJECTED_DUPLICATE:
+ case Topology::PEER_VERIFY_REJECTED_DUPLICATE_TRIAGED: {
+ Packet outp(req->source,_r->identity.address(),Packet::VERB_ERROR);
+ outp.append((unsigned char)Packet::VERB_HELLO);
+ outp.append(req->helloPacketId);
+ outp.append((unsigned char)Packet::ERROR_IDENTITY_COLLISION);
+ outp.encrypt(p->cryptKey());
+ outp.hmacSet(p->macKey());
+ _r->demarc->send(req->localPort,req->remoteAddress,outp.data(),outp.size(),-1);
+ } break;
+ }
+
+ delete req;
+}
+
+void PacketDecoder::_CBaddPeerFromWhois(void *arg,const SharedPtr<Peer> &p,Topology::PeerVerifyResult result)
+{
+ switch(result) {
+ case Topology::PEER_VERIFY_ACCEPTED_NEW:
+ case Topology::PEER_VERIFY_ACCEPTED_ALREADY_HAVE:
+ case Topology::PEER_VERIFY_ACCEPTED_DISPLACED_INVALID_ADDRESS:
+ ((const RuntimeEnvironment *)arg)->sw->doAnythingWaitingForPeer(p);
+ break;
+ default:
+ break;
+ }
+}
+
+bool PacketDecoder::_doERROR(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
try {
#ifdef ZT_TRACE
- Packet::Verb inReVerb = (Packet::Verb)packet[ZT_PROTO_VERB_ERROR_IDX_IN_RE_VERB];
- Packet::ErrorCode errorCode = (Packet::ErrorCode)packet[ZT_PROTO_VERB_ERROR_IDX_ERROR_CODE];
- TRACE("ERROR %s from %s(%s) in-re %s",Packet::errorString(errorCode),source.toString().c_str(),_remoteAddress.toString().c_str(),Packet::verbString(inReVerb));
+ Packet::Verb inReVerb = (Packet::Verb)(*this)[ZT_PROTO_VERB_ERROR_IDX_IN_RE_VERB];
+ Packet::ErrorCode errorCode = (Packet::ErrorCode)(*this)[ZT_PROTO_VERB_ERROR_IDX_ERROR_CODE];
+ TRACE("ERROR %s from %s(%s) in-re %s",Packet::errorString(errorCode),source().toString().c_str(),_remoteAddress.toString().c_str(),Packet::verbString(inReVerb));
#endif
// TODO (sorta):
// The fact is that the protocol works fine without error handling.
@@ -125,307 +185,328 @@ bool PacketDecoder::_doERROR(const RuntimeEnvironment *_r)
// identity collision, which if it comes from a supernode should cause
// us to restart and regenerate a new identity.
} catch (std::exception &ex) {
- TRACE("dropped ERROR from %s(%s): unexpected exception: %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped ERROR from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped ERROR from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped ERROR from %s(%s): unexpected exception: (unknown)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
bool PacketDecoder::_doHELLO(const RuntimeEnvironment *_r)
{
- Address source(source());
-
try {
- unsigned int protoVersion = packet[ZT_PROTO_VERB_HELLO_IDX_PROTOCOL_VERSION];
- unsigned int vMajor = packet[ZT_PROTO_VERB_HELLO_IDX_MAJOR_VERSION];
- unsigned int vMinor = packet[ZT_PROTO_VERB_HELLO_IDX_MINOR_VERSION];
- unsigned int vRevision = packet.at<uint16_t>(ZT_PROTO_VERB_HELLO_IDX_REVISION);
- uint64_t timestamp = packet.at<uint64_t>(ZT_PROTO_VERB_HELLO_IDX_TIMESTAMP);
- Identity id(packet,ZT_PROTO_VERB_HELLO_IDX_IDENTITY);
+ //unsigned int protoVersion = (*this)[ZT_PROTO_VERB_HELLO_IDX_PROTOCOL_VERSION];
+ unsigned int vMajor = (*this)[ZT_PROTO_VERB_HELLO_IDX_MAJOR_VERSION];
+ unsigned int vMinor = (*this)[ZT_PROTO_VERB_HELLO_IDX_MINOR_VERSION];
+ unsigned int vRevision = at<uint16_t>(ZT_PROTO_VERB_HELLO_IDX_REVISION);
+ uint64_t timestamp = at<uint64_t>(ZT_PROTO_VERB_HELLO_IDX_TIMESTAMP);
+ Identity id(*this,ZT_PROTO_VERB_HELLO_IDX_IDENTITY);
SharedPtr<Peer> candidate(new Peer(_r->identity,id));
candidate->setPathAddress(_remoteAddress,false);
// Initial sniff test
- if (protoVersion != ZT_PROTO_VERSION) {
- TRACE("rejected HELLO from %s(%s): invalid protocol version",source.toString().c_str(),_remoteAddress.toString().c_str());
- Packet outp(source,_r->identity.address(),Packet::VERB_ERROR);
- outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(packet.packetId());
- outp.append((unsigned char)Packet::ERROR_BAD_PROTOCOL_VERSION);
- outp.encrypt(candidate->cryptKey());
- outp.hmacSet(candidate->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
- return;
- }
if (id.address().isReserved()) {
- TRACE("rejected HELLO from %s(%s): identity has reserved address",source.toString().c_str(),_remoteAddress.toString().c_str());
- Packet outp(source,_r->identity.address(),Packet::VERB_ERROR);
+ TRACE("rejected HELLO from %s(%s): identity has reserved address",source().toString().c_str(),_remoteAddress.toString().c_str());
+ Packet outp(source(),_r->identity.address(),Packet::VERB_ERROR);
outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(packet.packetId());
+ outp.append(packetId());
outp.append((unsigned char)Packet::ERROR_IDENTITY_INVALID);
outp.encrypt(candidate->cryptKey());
outp.hmacSet(candidate->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
- return;
+ _r->demarc->send(_localPort,_remoteAddress,outp.data(),outp.size(),-1);
+ return true;
}
- if (id.address() != source) {
- TRACE("rejected HELLO from %s(%s): identity is not for sender of packet (HELLO is a self-announcement)",source.toString().c_str(),_remoteAddress.toString().c_str());
- Packet outp(source,_r->identity.address(),Packet::VERB_ERROR);
+ if (id.address() != source()) {
+ TRACE("rejected HELLO from %s(%s): identity is not for sender of packet (HELLO is a self-announcement)",source().toString().c_str(),_remoteAddress.toString().c_str());
+ Packet outp(source(),_r->identity.address(),Packet::VERB_ERROR);
outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(packet.packetId());
+ outp.append(packetId());
outp.append((unsigned char)Packet::ERROR_INVALID_REQUEST);
outp.encrypt(candidate->cryptKey());
outp.hmacSet(candidate->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
- return;
+ _r->demarc->send(_localPort,_remoteAddress,outp.data(),outp.size(),-1);
+ return true;
}
// Is this a HELLO for a peer we already know? If so just update its
// packet receive stats and send an OK.
SharedPtr<Peer> existingPeer(_r->topology->getPeer(id.address()));
if ((existingPeer)&&(existingPeer->identity() == id)) {
- existingPeer->onReceive(_r,localPort,_remoteAddress,0,packet.hops(),Packet::VERB_HELLO,Utils::now());
+ existingPeer->onReceive(_r,_localPort,_remoteAddress,hops(),Packet::VERB_HELLO,Utils::now());
+ existingPeer->setRemoteVersion(vMajor,vMinor,vRevision);
- Packet outp(source,_r->identity.address(),Packet::VERB_OK);
+ Packet outp(source(),_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(packet.packetId());
+ outp.append(packetId());
outp.append(timestamp);
outp.encrypt(existingPeer->cryptKey());
outp.hmacSet(existingPeer->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
- return;
+ _r->demarc->send(_localPort,_remoteAddress,outp.data(),outp.size(),-1);
+ return true;
}
// Otherwise we call addPeer() and set up a callback to handle the verdict
_CBaddPeerFromHello_Data *arg = new _CBaddPeerFromHello_Data;
- arg->parent = this;
- arg->source = source;
- arg->_remoteAddress = _remoteAddress;
- arg->localPort = localPort;
+ arg->renv = _r;
+ arg->source = source();
+ arg->remoteAddress = _remoteAddress;
+ arg->localPort = _localPort;
arg->vMajor = vMajor;
arg->vMinor = vMinor;
arg->vRevision = vRevision;
- arg->helloPacketId = packet.packetId();
+ arg->helloPacketId = packetId();
arg->helloTimestamp = timestamp;
- _r->topology->addPeer(candidate,&Switch::_CBaddPeerFromHello,arg);
+ _r->topology->addPeer(candidate,&PacketDecoder::_CBaddPeerFromHello,arg);
} catch (std::exception &ex) {
- TRACE("dropped HELLO from %s(%s): %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped HELLO from %s(%s): %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped HELLO from %s(%s): unexpected exception",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped HELLO from %s(%s): unexpected exception",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
-bool PacketDecoder::_doOK(const RuntimeEnvironment *_r)
+bool PacketDecoder::_doOK(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
try {
- Packet::Verb inReVerb = (Packet::Verb)packet[ZT_PROTO_VERB_OK_IDX_IN_RE_VERB];
+ Packet::Verb inReVerb = (Packet::Verb)(*this)[ZT_PROTO_VERB_OK_IDX_IN_RE_VERB];
switch(inReVerb) {
- case Packet::VERB_HELLO:
+ case Packet::VERB_HELLO: {
// OK from HELLO permits computation of latency.
- latency = std::min((unsigned int)(now - packet.at<uint64_t>(ZT_PROTO_VERB_HELLO__OK__IDX_TIMESTAMP)),(unsigned int)0xffff);
- TRACE("%s(%s): OK(HELLO), latency: %u",source.toString().c_str(),_remoteAddress.toString().c_str(),latency);
- break;
+ unsigned int latency = std::min((unsigned int)(Utils::now() - at<uint64_t>(ZT_PROTO_VERB_HELLO__OK__IDX_TIMESTAMP)),(unsigned int)0xffff);
+ TRACE("%s(%s): OK(HELLO), latency: %u",source().toString().c_str(),_remoteAddress.toString().c_str(),latency);
+ peer->setLatency(_remoteAddress,latency);
+ } break;
case Packet::VERB_WHOIS:
// Right now we only query supernodes for WHOIS and only accept
// OK back from them. If we query other nodes, we'll have to
// do something to prevent WHOIS cache poisoning such as
// using the packet ID field in the OK packet to match with the
// original query. Technically we should be doing this anyway.
- TRACE("%s(%s): OK(%s)",source.toString().c_str(),_remoteAddress.toString().c_str(),Packet::verbString(inReVerb));
- if (_r->topology->isSupernode(source))
- _r->topology->addPeer(SharedPtr<Peer>(new Peer(_r->identity,Identity(packet,ZT_PROTO_VERB_WHOIS__OK__IDX_IDENTITY))),&Switch::_CBaddPeerFromWhois,this);
+ TRACE("%s(%s): OK(%s)",source().toString().c_str(),_remoteAddress.toString().c_str(),Packet::verbString(inReVerb));
+ if (_r->topology->isSupernode(source()))
+ _r->topology->addPeer(SharedPtr<Peer>(new Peer(_r->identity,Identity(*this,ZT_PROTO_VERB_WHOIS__OK__IDX_IDENTITY))),&PacketDecoder::_CBaddPeerFromWhois,const_cast<void *>((const void *)_r));
break;
default:
- TRACE("%s(%s): OK(%s)",source.toString().c_str(),_remoteAddress.toString().c_str(),Packet::verbString(inReVerb));
+ TRACE("%s(%s): OK(%s)",source().toString().c_str(),_remoteAddress.toString().c_str(),Packet::verbString(inReVerb));
break;
}
} catch (std::exception &ex) {
- TRACE("dropped OK from %s(%s): unexpected exception: %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped OK from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped OK from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped OK from %s(%s): unexpected exception: (unknown)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
-bool PacketDecoder::_doWHOIS(const RuntimeEnvironment *_r)
+bool PacketDecoder::_doWHOIS(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
- if (packet.payloadLength() == ZT_ADDRESS_LENGTH) {
- SharedPtr<Peer> p(_r->topology->getPeer(Address(packet.payload())));
+ if (payloadLength() == ZT_ADDRESS_LENGTH) {
+ SharedPtr<Peer> p(_r->topology->getPeer(Address(payload())));
if (p) {
- Packet outp(source,_r->identity.address(),Packet::VERB_OK);
+ Packet outp(source(),_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_WHOIS);
- outp.append(packet.packetId());
+ outp.append(packetId());
p->identity().serialize(outp,false);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
- TRACE("sent WHOIS response to %s for %s",source.toString().c_str(),Address(packet.payload()).toString().c_str());
+ _r->demarc->send(_localPort,_remoteAddress,outp.data(),outp.size(),-1);
+ TRACE("sent WHOIS response to %s for %s",source().toString().c_str(),Address(payload()).toString().c_str());
} else {
- Packet outp(source,_r->identity.address(),Packet::VERB_ERROR);
+ Packet outp(source(),_r->identity.address(),Packet::VERB_ERROR);
outp.append((unsigned char)Packet::VERB_WHOIS);
- outp.append(packet.packetId());
+ outp.append(packetId());
outp.append((unsigned char)Packet::ERROR_NOT_FOUND);
- outp.append(packet.payload(),ZT_ADDRESS_LENGTH);
+ outp.append(payload(),ZT_ADDRESS_LENGTH);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
- TRACE("sent WHOIS ERROR to %s for %s (not found)",source.toString().c_str(),Address(packet.payload()).toString().c_str());
+ _r->demarc->send(_localPort,_remoteAddress,outp.data(),outp.size(),-1);
+ TRACE("sent WHOIS ERROR to %s for %s (not found)",source().toString().c_str(),Address(payload()).toString().c_str());
}
} else {
- TRACE("dropped WHOIS from %s(%s): missing or invalid address",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped WHOIS from %s(%s): missing or invalid address",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
-bool PacketDecoder::_doRENDEZVOUS(const RuntimeEnvironment *_r)
+bool PacketDecoder::_doRENDEZVOUS(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
try {
- Address with(packet.field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ZTADDRESS,ZT_ADDRESS_LENGTH));
- RendezvousQueueEntry qe;
+ Address with(field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ZTADDRESS,ZT_ADDRESS_LENGTH));
if (_r->topology->getPeer(with)) {
- unsigned int port = packet.at<uint16_t>(ZT_PROTO_VERB_RENDEZVOUS_IDX_PORT);
- unsigned int addrlen = packet[ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRLEN];
+ unsigned int port = at<uint16_t>(ZT_PROTO_VERB_RENDEZVOUS_IDX_PORT);
+ unsigned int addrlen = (*this)[ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRLEN];
if ((port > 0)&&((addrlen == 4)||(addrlen == 16))) {
- qe.inaddr.set(packet.field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRESS,addrlen),addrlen,port);
- qe.fireAtTime = now + ZT_RENDEZVOUS_NAT_T_DELAY; // then send real packet in a few ms
- qe.localPort = _r->demarc->pick(qe.inaddr);
- TRACE("RENDEZVOUS from %s says %s might be at %s, starting NAT-t",source.toString().c_str(),with.toString().c_str(),qe.inaddr.toString().c_str());
- _r->demarc->send(qe.localPort,qe.inaddr,"\0",1,ZT_FIREWALL_OPENER_HOPS); // start with firewall opener
- {
- Mutex::Lock _l(_rendezvousQueue_m);
- _rendezvousQueue[with] = qe;
- }
+ InetAddress atAddr(field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRESS,addrlen),addrlen,port);
+ TRACE("RENDEZVOUS from %s says %s might be at %s, starting NAT-t",source().toString().c_str(),with.toString().c_str(),atAddr.toString().c_str());
+ _r->sw->contact(peer,atAddr);
} else {
- TRACE("dropped corrupt RENDEZVOUS from %s(%s) (bad address or port)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped corrupt RENDEZVOUS from %s(%s) (bad address or port)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
} else {
- TRACE("ignored RENDEZVOUS from %s(%s) to meet unknown peer %s",source.toString().c_str(),_remoteAddress.toString().c_str(),with.toString().c_str());
+ TRACE("ignored RENDEZVOUS from %s(%s) to meet unknown peer %s",source().toString().c_str(),_remoteAddress.toString().c_str(),with.toString().c_str());
}
} catch (std::exception &ex) {
- TRACE("dropped RENDEZVOUS from %s(%s): %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped RENDEZVOUS from %s(%s): %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped RENDEZVOUS from %s(%s): unexpected exception",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped RENDEZVOUS from %s(%s): unexpected exception",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
-bool PacketDecoder::_doFRAME(const RuntimeEnvironment *_r)
+bool PacketDecoder::_doFRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
try {
- SharedPtr<Network> network(_r->nc->network(packet.at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID)));
+ SharedPtr<Network> network(_r->nc->network(at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID)));
if (network) {
- if (network->isAllowed(source)) {
- unsigned int etherType = packet.at<uint16_t>(ZT_PROTO_VERB_FRAME_IDX_ETHERTYPE);
+ if (network->isAllowed(source())) {
+ unsigned int etherType = at<uint16_t>(ZT_PROTO_VERB_FRAME_IDX_ETHERTYPE);
if ((etherType != ZT_ETHERTYPE_ARP)&&(etherType != ZT_ETHERTYPE_IPV4)&&(etherType != ZT_ETHERTYPE_IPV6)) {
- TRACE("dropped FRAME from %s: unsupported ethertype",source.toString().c_str());
- } else if (packet.size() > ZT_PROTO_VERB_FRAME_IDX_PAYLOAD) {
- network->tap().put(source.toMAC(),network->tap().mac(),etherType,packet.data() + ZT_PROTO_VERB_FRAME_IDX_PAYLOAD,packet.size() - ZT_PROTO_VERB_FRAME_IDX_PAYLOAD);
+ TRACE("dropped FRAME from %s: unsupported ethertype",source().toString().c_str());
+ } else if (size() > ZT_PROTO_VERB_FRAME_IDX_PAYLOAD) {
+ network->tap().put(source().toMAC(),network->tap().mac(),etherType,data() + ZT_PROTO_VERB_FRAME_IDX_PAYLOAD,size() - ZT_PROTO_VERB_FRAME_IDX_PAYLOAD);
}
} else {
- TRACE("dropped FRAME from %s(%s): not a member of closed network %llu",source.toString().c_str(),_remoteAddress.toString().c_str(),network->id());
+ TRACE("dropped FRAME from %s(%s): not a member of closed network %llu",source().toString().c_str(),_remoteAddress.toString().c_str(),network->id());
}
} else {
- TRACE("dropped FRAME from %s(%s): network %llu unknown",source.toString().c_str(),_remoteAddress.toString().c_str(),packet.at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID));
+ TRACE("dropped FRAME from %s(%s): network %llu unknown",source().toString().c_str(),_remoteAddress.toString().c_str(),at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID));
}
} catch (std::exception &ex) {
- TRACE("dropped FRAME from %s(%s): unexpected exception: %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped FRAME from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped FRAME from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped FRAME from %s(%s): unexpected exception: (unknown)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
-bool PacketDecoder::_doMULTICAST_LIKE(const RuntimeEnvironment *_r)
+bool PacketDecoder::_doMULTICAST_LIKE(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
try {
unsigned int ptr = ZT_PACKET_IDX_PAYLOAD;
unsigned int numAccepted = 0;
+ uint64_t now = Utils::now();
// Iterate through 18-byte network,MAC,ADI tuples:
- while ((ptr + 18) <= packet.size()) {
- uint64_t nwid = packet.at<uint64_t>(ptr); ptr += 8;
+ while ((ptr + 18) <= size()) {
+ uint64_t nwid = at<uint64_t>(ptr); ptr += 8;
SharedPtr<Network> network(_r->nc->network(nwid));
if (network) {
- if (network->isAllowed(source)) {
- MAC mac(packet.field(ptr,6)); ptr += 6;
- uint32_t adi = packet.at<uint32_t>(ptr); ptr += 4;
- TRACE("peer %s likes multicast group %s:%.8lx on network %llu",source.toString().c_str(),mac.toString().c_str(),(unsigned long)adi,nwid);
- _multicaster.likesMulticastGroup(nwid,MulticastGroup(mac,adi),source,now);
+ if (network->isAllowed(source())) {
+ MAC mac(field(ptr,6)); ptr += 6;
+ uint32_t adi = at<uint32_t>(ptr); ptr += 4;
+ TRACE("peer %s likes multicast group %s:%.8lx on network %llu",source().toString().c_str(),mac.toString().c_str(),(unsigned long)adi,nwid);
+ _r->multicaster->likesMulticastGroup(nwid,MulticastGroup(mac,adi),source(),now);
++numAccepted;
} else {
- TRACE("ignored MULTICAST_LIKE from %s(%s): not a member of closed network %llu",source.toString().c_str(),_remoteAddress.toString().c_str(),nwid);
+ TRACE("ignored MULTICAST_LIKE from %s(%s): not a member of closed network %llu",source().toString().c_str(),_remoteAddress.toString().c_str(),nwid);
}
} else {
- TRACE("ignored MULTICAST_LIKE from %s(%s): network %llu unknown or we are not a member",source.toString().c_str(),_remoteAddress.toString().c_str(),nwid);
+ TRACE("ignored MULTICAST_LIKE from %s(%s): network %llu unknown or we are not a member",source().toString().c_str(),_remoteAddress.toString().c_str(),nwid);
}
}
- Packet outp(source,_r->identity.address(),Packet::VERB_OK);
+ Packet outp(source(),_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_MULTICAST_LIKE);
- outp.append(packet.packetId());
+ outp.append(packetId());
outp.append((uint16_t)numAccepted);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
- _r->demarc->send(localPort,_remoteAddress,outp.data(),outp.size(),-1);
+ _r->demarc->send(_localPort,_remoteAddress,outp.data(),outp.size(),-1);
} catch (std::exception &ex) {
- TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: (unknown)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
-bool PacketDecoder::_doMULTICAST_FRAME(const RuntimeEnvironment *_r)
+bool PacketDecoder::_doMULTICAST_FRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer)
{
try {
- SharedPtr<Network> network(_r->nc->network(packet.at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID)));
+ SharedPtr<Network> network(_r->nc->network(at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID)));
if (network) {
- if (network->isAllowed(source)) {
- if (packet.size() > ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD) {
- Address originalSubmitterAddress(packet.field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SUBMITTER_ADDRESS,ZT_ADDRESS_LENGTH));
- MAC fromMac(packet.field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SOURCE_MAC,6));
- MulticastGroup mg(MAC(packet.field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_DESTINATION_MAC,6)),packet.at<uint32_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ADI));
- unsigned int hops = packet[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT];
- unsigned int etherType = packet.at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ETHERTYPE);
- unsigned int datalen = packet.at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD_LENGTH);
- unsigned int signaturelen = packet.at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SIGNATURE_LENGTH);
- unsigned char *dataAndSignature = packet.field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD,datalen + signaturelen);
-
- bool isDuplicate = _multicaster.checkAndUpdateMulticastHistory(fromMac,mg,payload,payloadLen,network->id(),now);
+ if (network->isAllowed(source())) {
+ if (size() > ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD) {
+
+ Address originalSubmitterAddress(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SUBMITTER_ADDRESS,ZT_ADDRESS_LENGTH));
+ MAC fromMac(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SOURCE_MAC,6));
+ MulticastGroup mg(MAC(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_DESTINATION_MAC,6)),at<uint32_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ADI));
+ unsigned int hops = (*this)[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT];
+ unsigned int etherType = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ETHERTYPE);
+ unsigned int datalen = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD_LENGTH);
+ unsigned int signaturelen = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SIGNATURE_LENGTH);
+ unsigned char *dataAndSignature = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD,datalen + signaturelen);
+
+ bool isDuplicate = _r->multicaster->checkAndUpdateMulticastHistory(fromMac,mg,dataAndSignature,datalen,network->id(),Utils::now());
if (originalSubmitterAddress == _r->identity.address()) {
// Technically should not happen, since the original submitter is
// excluded from consideration as a propagation recipient.
- TRACE("dropped boomerang MULTICAST_FRAME received from %s(%s)",source.toString().c_str(),_remoteAddress.toString().c_str());
- } else if ((!isDuplicate)||(_r->topology.isSupernode(_r->identity.address()))) {
+ TRACE("dropped boomerang MULTICAST_FRAME received from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
+ } else if ((!isDuplicate)||(_r->topology->isSupernode(_r->identity.address()))) {
// If I am a supernode, I will repeatedly propagate duplicates. That's
// because supernodes are used to bridge sparse multicast groups. Non-
// supernodes will ignore duplicates completely.
SharedPtr<Peer> originalSubmitter(_r->topology->getPeer(originalSubmitterAddress));
if (!originalSubmitter) {
TRACE("requesting WHOIS on original multicast frame submitter %s",originalSubmitterAddress.toString().c_str());
- _requestWhois(originalSubmitterAddress,packet.packetId());
+ _r->sw->requestWhois(originalSubmitterAddress);
+ _step = DECODE_STEP_WAITING_FOR_ORIGINAL_SUBMITTER_LOOKUP;
return false;
- } else if (Multicaster::verifyMulticastPacket(originalSubmitter->identity(),fromMac,mg,etherType,data,datalen,dataAndSignature + datalen,signaturelen)) {
+ } else if (Multicaster::verifyMulticastPacket(originalSubmitter->identity(),network->id(),fromMac,mg,etherType,dataAndSignature,datalen,dataAndSignature + datalen,signaturelen)) {
if (!isDuplicate)
- network->tap().put(fromMac,mg.mac(),etherType,payload,payloadLen);
- _propagateMulticast(network,originalSubmitterAddress,source,packet.field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_BLOOM,ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE),mg,hops+1,fromMac,etherType,payload,payloadLen);
+ network->tap().put(fromMac,mg.mac(),etherType,dataAndSignature,datalen);
+
+ if (++hops < ZT_MULTICAST_PROPAGATION_DEPTH) {
+ Multicaster::MulticastBloomFilter bloom(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_BLOOM_FILTER,ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES));
+ SharedPtr<Peer> propPeers[ZT_MULTICAST_PROPAGATION_BREADTH];
+ unsigned int np = _r->multicaster->pickNextPropagationPeers(
+ *(_r->topology),
+ network->id(),
+ mg,
+ originalSubmitterAddress,
+ source(),
+ bloom,
+ ZT_MULTICAST_PROPAGATION_BREADTH,
+ propPeers,
+ Utils::now());
+
+ (*this)[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT] = hops;
+ compress();
+
+ for(unsigned int i=0;i<np;++i) {
+ TRACE("propagating multicast from original node %s via %s toward %s",originalSubmitterAddress.toString().c_str(),source().toString().c_str(),propPeers[i]->address().toString().c_str());
+ // Re-use this packet to re-send multicast frame to everyone
+ // downstream from us.
+ newInitializationVector();
+ setDestination(propPeers[i]->address());
+ _r->sw->send(*this,true);
+ }
+ } else {
+ TRACE("terminating MULTICAST_FRAME propagation from %s(%s): max depth reached",source().toString().c_str(),_remoteAddress.toString().c_str());
+ }
} else {
- LOG("rejected MULTICAST_FRAME from %s(%s) due to failed signature check (claims original sender %s)",source.toString().c_str(),_remoteAddress.toString().c_str(),originalSubmitterAddress.toString().c_str());
+ LOG("rejected MULTICAST_FRAME from %s(%s) due to failed signature check (claims original sender %s)",source().toString().c_str(),_remoteAddress.toString().c_str(),originalSubmitterAddress.toString().c_str());
}
} else {
- TRACE("dropped redundant MULTICAST_FRAME from %s(%s)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped redundant MULTICAST_FRAME from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
} else {
- TRACE("dropped MULTICAST_FRAME from %s(%s): invalid short packet",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped MULTICAST_FRAME from %s(%s): invalid short packet",source().toString().c_str(),_remoteAddress.toString().c_str());
}
} else {
- TRACE("dropped MULTICAST_FRAME from %s(%s): not a member of closed network %llu",source.toString().c_str(),_remoteAddress.toString().c_str(),network->id());
+ TRACE("dropped MULTICAST_FRAME from %s(%s): not a member of closed network %llu",source().toString().c_str(),_remoteAddress.toString().c_str(),network->id());
}
} else {
- TRACE("dropped MULTICAST_FRAME from %s(%s): network %llu unknown or we are not a member",source.toString().c_str(),_remoteAddress.toString().c_str(),packet.at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID));
+ TRACE("dropped MULTICAST_FRAME from %s(%s): network %llu unknown or we are not a member",source().toString().c_str(),_remoteAddress.toString().c_str(),at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID));
}
} catch (std::exception &ex) {
- TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: %s",source.toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
+ TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
- TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),_remoteAddress.toString().c_str());
+ TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: (unknown)",source().toString().c_str(),_remoteAddress.toString().c_str());
}
+ return true;
}
} // namespace ZeroTier
diff --git a/node/PacketDecoder.hpp b/node/PacketDecoder.hpp
index 49d47a51..5da8f9c0 100644
--- a/node/PacketDecoder.hpp
+++ b/node/PacketDecoder.hpp
@@ -36,6 +36,7 @@
#include "Utils.hpp"
#include "SharedPtr.hpp"
#include "AtomicCounter.hpp"
+#include "Peer.hpp"
namespace ZeroTier {
@@ -56,7 +57,7 @@ public:
_receiveTime(Utils::now()),
_localPort(localPort),
_remoteAddress(remoteAddress),
- _step(DECODE_STEP_START),
+ _step(DECODE_STEP_WAITING_FOR_SENDER_LOOKUP),
__refCount()
{
}
@@ -76,21 +77,40 @@ public:
inline uint64_t receiveTime() const throw() { return _receiveTime; }
private:
- bool _doERROR(const RuntimeEnvironment *_r);
+ struct _CBaddPeerFromHello_Data
+ {
+ const RuntimeEnvironment *renv;
+ Address source;
+ InetAddress remoteAddress;
+ int localPort;
+ unsigned int vMajor,vMinor,vRevision;
+ uint64_t helloPacketId;
+ uint64_t helloTimestamp;
+ };
+ static void _CBaddPeerFromHello(
+ void *arg, // _CBaddPeerFromHello_Data
+ const SharedPtr<Peer> &p,
+ Topology::PeerVerifyResult result);
+
+ static void _CBaddPeerFromWhois(
+ void *arg, // RuntimeEnvironment
+ const SharedPtr<Peer> &p,
+ Topology::PeerVerifyResult result);
+
+ bool _doERROR(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
bool _doHELLO(const RuntimeEnvironment *_r);
- bool _doOK(const RuntimeEnvironment *_r);
- bool _doWHOIS(const RuntimeEnvironment *_r);
- bool _doRENDEZVOUS(const RuntimeEnvironment *_r);
- bool _doFRAME(const RuntimeEnvironment *_r);
- bool _doMULTICAST_LIKE(const RuntimeEnvironment *_r);
- bool _doMULTICAST_FRAME(const RuntimeEnvironment *_r);
+ bool _doOK(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
+ bool _doWHOIS(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
+ bool _doRENDEZVOUS(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
+ bool _doFRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
+ bool _doMULTICAST_LIKE(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
+ bool _doMULTICAST_FRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
uint64_t _receiveTime;
Demarc::Port _localPort;
InetAddress _remoteAddress;
enum {
- DECODE_STEP_START,
DECODE_STEP_WAITING_FOR_SENDER_LOOKUP, // on initial receipt, we need peer's identity
DECODE_STEP_WAITING_FOR_ORIGINAL_SUBMITTER_LOOKUP // this only applies to MULTICAST_FRAME
} _step;
diff --git a/node/Peer.cpp b/node/Peer.cpp
index 14eb49f0..d5037eb9 100644
--- a/node/Peer.cpp
+++ b/node/Peer.cpp
@@ -30,6 +30,9 @@
namespace ZeroTier {
Peer::Peer() :
+ _vMajor(0),
+ _vMinor(0),
+ _vRevision(0),
_dirty(false)
{
}
@@ -37,6 +40,9 @@ Peer::Peer() :
Peer::Peer(const Identity &myIdentity,const Identity &peerIdentity)
throw(std::runtime_error) :
_id(peerIdentity),
+ _vMajor(0),
+ _vMinor(0),
+ _vRevision(0),
_dirty(true)
{
if (!myIdentity.agree(peerIdentity,_keys,sizeof(_keys)))
diff --git a/node/Peer.hpp b/node/Peer.hpp
index 51527148..ef2d5851 100644
--- a/node/Peer.hpp
+++ b/node/Peer.hpp
@@ -213,21 +213,18 @@ public:
}
/**
- * @param latency measurment for IPv4 path
+ * @param addr Remote address
+ * @param latency Latency measurment
*/
- void setV4Latency(unsigned int latency)
+ void setLatency(const InetAddress &addr,unsigned int latency)
{
- _ipv4p.latency = latency;
- _dirty = true;
- }
-
- /**
- * @param latency Latency measurment for IPv6 path
- */
- void setV6Latency(unsigned int latency)
- {
- _ipv6p.latency = latency;
- _dirty = true;
+ if (addr == _ipv4p.addr) {
+ _ipv4p.latency = latency;
+ _dirty = true;
+ } else if (addr == _ipv6p.addr) {
+ _ipv6p.latency = latency;
+ _dirty = true;
+ }
}
/**
@@ -308,6 +305,20 @@ public:
}
/**
+ * Set the remote version of the peer (not persisted)
+ *
+ * @param vmaj Major version
+ * @param vmin Minor version
+ * @param vrev Revision
+ */
+ inline void setRemoteVersion(unsigned int vmaj,unsigned int vmin,unsigned int vrev)
+ {
+ _vMajor = vmaj;
+ _vMinor = vmin;
+ _vRevision = vrev;
+ }
+
+ /**
* Get and reset dirty flag
*
* @return Previous value of dirty flag before reset
@@ -482,6 +493,7 @@ private:
// Fields below this line are not persisted with serialize()
+ unsigned int _vMajor,_vMinor,_vRevision;
bool _dirty;
AtomicCounter __refCount;
diff --git a/node/RuntimeEnvironment.hpp b/node/RuntimeEnvironment.hpp
index a253af6e..ab19be49 100644
--- a/node/RuntimeEnvironment.hpp
+++ b/node/RuntimeEnvironment.hpp
@@ -39,6 +39,7 @@ class Demarc;
class Switch;
class Topology;
class SysEnv;
+class Multicaster;
/**
* Holds global state for an instance of ZeroTier::Node
@@ -60,6 +61,7 @@ public:
log((Logger *)0),
nc((NodeConfig *)0),
demarc((Demarc *)0),
+ multicaster((Multicaster *)0),
sw((Switch *)0),
topology((Topology *)0)
{
@@ -77,6 +79,7 @@ public:
Logger *log; // may be null
NodeConfig *nc;
Demarc *demarc;
+ Multicaster *multicaster;
Switch *sw;
Topology *topology;
SysEnv *sysEnv;
diff --git a/node/Switch.cpp b/node/Switch.cpp
index 7a1784ac..7f41eef0 100644
--- a/node/Switch.cpp
+++ b/node/Switch.cpp
@@ -105,7 +105,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
Multicaster::MulticastBloomFilter newbf;
SharedPtr<Peer> propPeers[ZT_MULTICAST_PROPAGATION_BREADTH];
- unsigned int np = _multicaster.pickNextPropagationPeers(
+ unsigned int np = _r->multicaster->pickNextPropagationPeers(
*(_r->topology),
network->id(),
mg,
@@ -324,7 +324,7 @@ unsigned long Switch::doTimerTasks()
{
Mutex::Lock _l(_rxQueue_m);
- for(std::multimap< Address,SharedPtr<PacketDecoder> >::iterator i(_rxQueue.begin());i!=_rxQueue.end();) {
+ for(std::list< SharedPtr<PacketDecoder> >::iterator i(_rxQueue.begin());i!=_rxQueue.end();) {
if ((now - i->second->receiveTime()) > ZT_RECEIVE_QUEUE_TIMEOUT) {
TRACE("RX %s -> %s timed out",i->second->source().toString().c_str(),i->second->destination().toString().c_str());
_rxQueue.erase(i++);
@@ -392,65 +392,7 @@ void Switch::requestWhois(const Address &addr)
_sendWhoisRequest(addr,(const Address *)0,0);
}
-void Switch::_CBaddPeerFromHello(void *arg,const SharedPtr<Peer> &p,Topology::PeerVerifyResult result)
-{
- _CBaddPeerFromHello_Data *req = (_CBaddPeerFromHello_Data *)arg;
- const RuntimeEnvironment *_r = req->parent->_r;
-
- switch(result) {
- case Topology::PEER_VERIFY_ACCEPTED_NEW:
- case Topology::PEER_VERIFY_ACCEPTED_ALREADY_HAVE:
- case Topology::PEER_VERIFY_ACCEPTED_DISPLACED_INVALID_ADDRESS: {
- req->parent->_finishWhoisRequest(p);
-
- Packet outp(req->source,_r->identity.address(),Packet::VERB_OK);
- outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(req->helloPacketId);
- outp.append(req->helloTimestamp);
- outp.encrypt(p->cryptKey());
- outp.hmacSet(p->macKey());
- req->parent->_r->demarc->send(req->localPort,req->fromAddr,outp.data(),outp.size(),-1);
- } break;
-
- case Topology::PEER_VERIFY_REJECTED_INVALID_IDENTITY: {
- Packet outp(req->source,_r->identity.address(),Packet::VERB_ERROR);
- outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(req->helloPacketId);
- outp.append((unsigned char)Packet::ERROR_IDENTITY_INVALID);
- outp.encrypt(p->cryptKey());
- outp.hmacSet(p->macKey());
- req->parent->_r->demarc->send(req->localPort,req->fromAddr,outp.data(),outp.size(),-1);
- } break;
-
- case Topology::PEER_VERIFY_REJECTED_DUPLICATE:
- case Topology::PEER_VERIFY_REJECTED_DUPLICATE_TRIAGED: {
- Packet outp(req->source,_r->identity.address(),Packet::VERB_ERROR);
- outp.append((unsigned char)Packet::VERB_HELLO);
- outp.append(req->helloPacketId);
- outp.append((unsigned char)Packet::ERROR_IDENTITY_COLLISION);
- outp.encrypt(p->cryptKey());
- outp.hmacSet(p->macKey());
- req->parent->_r->demarc->send(req->localPort,req->fromAddr,outp.data(),outp.size(),-1);
- } break;
- }
-
- delete req;
-}
-
-void Switch::_CBaddPeerFromWhois(void *arg,const SharedPtr<Peer> &p,Topology::PeerVerifyResult result)
-{
- switch(result) {
- case Topology::PEER_VERIFY_ACCEPTED_NEW:
- case Topology::PEER_VERIFY_ACCEPTED_ALREADY_HAVE:
- case Topology::PEER_VERIFY_ACCEPTED_DISPLACED_INVALID_ADDRESS:
- ((Switch *)arg)->_finishWhoisRequest(p);
- break;
- default:
- break;
- }
-}
-
-void Switch::_finishWhoisRequest(const SharedPtr<Peer> &peer)
+void Switch::doAnythingWaitingForPeer(const SharedPtr<Peer> &peer)
{
{
Mutex::Lock _l(_outstandingWhoisRequests_m);
@@ -459,8 +401,7 @@ void Switch::_finishWhoisRequest(const SharedPtr<Peer> &peer)
{
Mutex::Lock _l(_rxQueue_m);
- std::pair< std::multimap< Address,SharedPtr<PacketDecoder> >::iterator,std::multimap< Address,SharedPtr<PacketDecoder> >::iterator > waitingRxQueueItems(_rxQueue.equal_range(peer->address()));
- for(std::multimap< Address,SharedPtr<PacketDecoder> >::iterator rxi(waitingRxQueueItems.first);rxi!=waitingRxQueueItems.second;) {
+ for(std::list< SharedPtr<PacketDecoder> >::iterator rxi(_rxQueue.begin());rxi!=rxQueue.end();) {
if (rxi->second->tryDecode(_r))
_rxQueue.erase(rxi++);
else ++rxi;
@@ -539,7 +480,7 @@ void Switch::_handleRemotePacketFragment(Demarc::Port localPort,const InetAddres
if (!packet->tryDecode(_r)) {
Mutex::Lock _l(_rxQueue_m);
- _rxQueue.insert(std::pair< Address,SharedPtr<PacketDecoder> >(destination,packet));
+ _rxQueue.push_back(packet);
}
}
} // else this is a duplicate fragment, ignore
@@ -596,7 +537,7 @@ void Switch::_handleRemotePacketHead(Demarc::Port localPort,const InetAddress &f
if (!packet->tryDecode(_r)) {
Mutex::Lock _l(_rxQueue_m);
- _rxQueue.insert(std::pair< Address,SharedPtr<PacketDecoder> >(destination,packet));
+ _rxQueue.push_back(packet);
}
} else {
// Still waiting on more fragments, so queue the head
@@ -607,7 +548,7 @@ void Switch::_handleRemotePacketHead(Demarc::Port localPort,const InetAddress &f
// Packet is unfragmented, so just process it
if (!packet->tryDecode(_r)) {
Mutex::Lock _l(_rxQueue_m);
- _rxQueue.insert(std::pair< Address,SharedPtr<PacketDecoder> >(destination,packet));
+ _rxQueue.push_back(packet);
}
}
}
diff --git a/node/Switch.hpp b/node/Switch.hpp
index f6abee15..f69e26bf 100644
--- a/node/Switch.hpp
+++ b/node/Switch.hpp
@@ -31,6 +31,7 @@
#include <map>
#include <set>
#include <vector>
+#include <list>
#include "Mutex.hpp"
#include "MAC.hpp"
@@ -135,6 +136,14 @@ public:
bool unite(const Address &p1,const Address &p2,bool force);
/**
+ * Send NAT traversal messages to peer at the given candidate address
+ *
+ * @param peer Peer to contact
+ * @param atAddr Address of peer
+ */
+ void contact(const SharedPtr<Peer> &peer,const InetAddress &atAddr);
+
+ /**
* Perform retries and other periodic timer tasks
*
* @return Number of milliseconds until doTimerTasks() should be run again
@@ -158,30 +167,16 @@ public:
*/
void requestWhois(const Address &addr);
-private:
- struct _CBaddPeerFromHello_Data
- {
- Switch *parent;
- Address source;
- InetAddress fromAddr;
- int localPort;
- unsigned int vMajor,vMinor,vRevision;
- uint64_t helloPacketId;
- uint64_t helloTimestamp;
- };
- static void _CBaddPeerFromHello(
- void *arg, // _CBaddPeerFromHello_Data
- const SharedPtr<Peer> &p,
- Topology::PeerVerifyResult result);
-
- static void _CBaddPeerFromWhois(
- void *arg, // this (Switch)
- const SharedPtr<Peer> &p,
- Topology::PeerVerifyResult result);
-
- void _finishWhoisRequest(
- const SharedPtr<Peer> &peer);
+ /**
+ * Run any processes that are waiting for this peer
+ *
+ * Called when we learn of a peer's identity from HELLO, OK(WHOIS), etc.
+ *
+ * @param peer New peer
+ */
+ void doAnythingWaitingForPeer(const SharedPtr<Peer> &peer);
+private:
void _handleRemotePacketFragment(
Demarc::Port localPort,
const InetAddress &fromAddr,
@@ -202,7 +197,6 @@ private:
bool encrypt);
const RuntimeEnvironment *const _r;
- Multicaster _multicaster;
struct WhoisRequest
{
@@ -213,7 +207,7 @@ private:
std::map< Address,WhoisRequest > _outstandingWhoisRequests;
Mutex _outstandingWhoisRequests_m;
- std::multimap< Address,SharedPtr<PacketDecoder> > _rxQueue;
+ std::list< SharedPtr<PacketDecoder> > _rxQueue;
Mutex _rxQueue_m;
struct TXQueueEntry
@@ -244,15 +238,6 @@ private:
std::map< Array< Address,2 >,uint64_t > _lastUniteAttempt; // key is always sorted in ascending order, for set-like behavior
Mutex _lastUniteAttempt_m;
-
- struct RendezvousQueueEntry
- {
- InetAddress inaddr;
- uint64_t fireAtTime;
- Demarc::Port localPort;
- };
- std::map< Address,RendezvousQueueEntry > _rendezvousQueue;
- Mutex _rendezvousQueue_m;
};
} // namespace ZeroTier