More work in progress on Switch / PacketDecoder refactor.

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);

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

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;

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)))

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;
+		}
 	}
 
 	/**
@@ -307,6 +304,20 @@ public:
 		return (_keys + 32); // mac key is second 32-byte key
 	}
 
+	/**
+	 * 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
 	 * 
@@ -482,6 +493,7 @@ private:
 
 	// Fields below this line are not persisted with serialize()
 
+	unsigned int _vMajor,_vMinor,_vRevision;
 	bool _dirty;
 
 	AtomicCounter __refCount;

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;

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);
 		}
 	}
 }

node/Switch.hpp

@@ -31,6 +31,7 @@
 #include <map>
 #include <set>
 #include <vector>
+#include <list>
 
 #include "Mutex.hpp"
 #include "MAC.hpp"
@@ -134,6 +135,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
 	 * 
@@ -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

objects.mk

@@ -21,6 +21,7 @@ OBJS=\
 	node/Node.o \
 	node/NodeConfig.o \
 	node/Packet.o \
+	node/PacketDecoder.o \
 	node/Pack.o \
 	node/Peer.o \
 	node/Salsa20.o \