mirror
/
devilutionX-ZeroTierOne
mirror of https://github.com/diasurgical/ZeroTierOne.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

More major Switch refactor work... still in progress.

pull/1/head
Adam Ierymenko 13 years ago
parent
ffad0b2780
commit
ae93c95151
4 changed files with 452 additions and 453 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 368
      node/PacketDecoder.cpp
  2. 4
      node/PacketDecoder.hpp
  3. 516
      node/Switch.cpp
  4. 17
      node/Switch.hpp

368
node/PacketDecoder.cpp
Unescape View File

@ -29,12 +29,380 @@
#include "Topology.hpp"
#include "PacketDecoder.hpp"
#include "Switch.hpp"
#include "Peer.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(),fromAddr.toString().c_str());
_doHELLO(localPort,fromAddr,packet);
return true;
}
SharedPtr<Peer> peer = _r->topology->getPeer(source);
if (peer) {
uint64_t now = Utils::now();
unsigned int latency = 0;
if (!packet.hmacVerify(peer->macKey())) {
TRACE("dropped packet from %s(%s), HMAC authentication failed (size: %u)",source.toString().c_str(),fromAddr.toString().c_str(),packet.size());
return true;
}
if (packet.encrypted()) {
packet.decrypt(peer->cryptKey());
} else {
// 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(packet.verb()),source.toString().c_str(),fromAddr.toString().c_str());
}
if (!packet.uncompress()) {
TRACE("dropped packet from %s(%s), compressed data invalid",source.toString().c_str(),fromAddr.toString().c_str());
return true;
}
switch(packet.verb()) {
case Packet::VERB_NOP:
TRACE("NOP from %s(%s)",source.toString().c_str(),fromAddr.toString().c_str());
break;
case Packet::VERB_HELLO:
// HELLO is normally handled up top, but this is legal. Pointless, but legal.
_doHELLO(localPort,fromAddr,packet);
break;
case Packet::VERB_ERROR:
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(),fromAddr.toString().c_str(),Packet::verbString(inReVerb));
#endif
// TODO:
// The fact is that the protocol works fine without error handling.
// The only error that really needs to be handled here is duplicate
// 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(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped ERROR from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_OK:
try {
Packet::Verb inReVerb = (Packet::Verb)packet[ZT_PROTO_VERB_OK_IDX_IN_RE_VERB];
switch(inReVerb) {
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(),fromAddr.toString().c_str(),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(),fromAddr.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);
break;
default:
TRACE("%s(%s): OK(%s)",source.toString().c_str(),fromAddr.toString().c_str(),Packet::verbString(inReVerb));
break;
}
} catch (std::exception &ex) {
TRACE("dropped OK from %s(%s): unexpected exception: %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped OK from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_WHOIS:
if (packet.payloadLength() == ZT_ADDRESS_LENGTH) {
SharedPtr<Peer> p(_r->topology->getPeer(Address(packet.payload())));
if (p) {
Packet outp(source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_WHOIS);
outp.append(packet.packetId());
p->identity().serialize(outp,false);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
TRACE("sent WHOIS response to %s for %s",source.toString().c_str(),Address(packet.payload()).toString().c_str());
} else {
Packet outp(source,_r->identity.address(),Packet::VERB_ERROR);
outp.append((unsigned char)Packet::VERB_WHOIS);
outp.append(packet.packetId());
outp.append((unsigned char)Packet::ERROR_NOT_FOUND);
outp.append(packet.payload(),ZT_ADDRESS_LENGTH);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
_r->demarc->send(localPort,fromAddr,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());
}
} else {
TRACE("dropped WHOIS from %s(%s): missing or invalid address",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_RENDEZVOUS:
try {
Address with(packet.field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ZTADDRESS,ZT_ADDRESS_LENGTH));
RendezvousQueueEntry qe;
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];
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;
}
} else {
TRACE("dropped corrupt RENDEZVOUS from %s(%s) (bad address or port)",source.toString().c_str(),fromAddr.toString().c_str());
}
} else {
TRACE("ignored RENDEZVOUS from %s(%s) to meet unknown peer %s",source.toString().c_str(),fromAddr.toString().c_str(),with.toString().c_str());
}
} catch (std::exception &ex) {
TRACE("dropped RENDEZVOUS from %s(%s): %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped RENDEZVOUS from %s(%s): unexpected exception",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_FRAME:
try {
SharedPtr<Network> network(_r->nc->network(packet.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 ((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);
}
} else {
TRACE("dropped FRAME from %s(%s): not a member of closed network %llu",source.toString().c_str(),fromAddr.toString().c_str(),network->id());
}
} else {
TRACE("dropped FRAME from %s(%s): network %llu unknown",source.toString().c_str(),fromAddr.toString().c_str(),packet.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(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped FRAME from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_MULTICAST_LIKE:
try {
unsigned int ptr = ZT_PACKET_IDX_PAYLOAD;
unsigned int numAccepted = 0;
// Iterate through 18-byte network,MAC,ADI tuples:
while ((ptr + 18) <= packet.size()) {
uint64_t nwid = packet.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);
++numAccepted;
} else {
TRACE("ignored MULTICAST_LIKE from %s(%s): not a member of closed network %llu",source.toString().c_str(),fromAddr.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(),fromAddr.toString().c_str(),nwid);
}
}
Packet outp(source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_MULTICAST_LIKE);
outp.append(packet.packetId());
outp.append((uint16_t)numAccepted);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
} catch (std::exception &ex) {
TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_MULTICAST_FRAME:
try {
SharedPtr<Network> network(_r->nc->network(packet.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 (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(),fromAddr.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());
return false;
} else if (Multicaster::verifyMulticastPacket(originalSubmitter->identity(),fromMac,mg,etherType,data,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);
} else {
LOG("rejected MULTICAST_FRAME from %s(%s) due to failed signature check (claims original sender %s)",source.toString().c_str(),fromAddr.toString().c_str(),originalSubmitterAddress.toString().c_str());
}
} else {
TRACE("dropped redundant MULTICAST_FRAME from %s(%s)",source.toString().c_str(),fromAddr.toString().c_str());
}
} else {
TRACE("dropped MULTICAST_FRAME from %s(%s): invalid short packet",source.toString().c_str(),fromAddr.toString().c_str());
}
} else {
TRACE("dropped MULTICAST_FRAME from %s(%s): not a member of closed network %llu",source.toString().c_str(),fromAddr.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(),fromAddr.toString().c_str(),packet.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(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
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(),fromAddr.toString().c_str());
break;
}
// Update peer timestamps and learn new links. This must only ever
// be called on an authenticated and technically valid packet, since
// we only learn paths to peers over the WAN by hearing directly
// from them over those paths. (Or by having them authoritatively
// and statically defined, like with supernodes, but that's done
// elsewhere.)
peer->onReceive(_r,localPort,fromAddr,latency,packet.hops(),packet.verb(),now);
} else {
_requestWhois(source,packet.packetId());
return false;
}
return true;
}
void PacketDecoder::_doHELLO(Demarc::Port localPort,const InetAddress &fromAddr)
{
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);
SharedPtr<Peer> candidate(new Peer(_r->identity,id));
candidate->setPathAddress(fromAddr,false);
// Initial sniff test
if (protoVersion != ZT_PROTO_VERSION) {
TRACE("rejected HELLO from %s(%s): invalid protocol version",source.toString().c_str(),fromAddr.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,fromAddr,outp.data(),outp.size(),-1);
return;
}
if (id.address().isReserved()) {
TRACE("rejected HELLO from %s(%s): identity has reserved address",source.toString().c_str(),fromAddr.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_IDENTITY_INVALID);
outp.encrypt(candidate->cryptKey());
outp.hmacSet(candidate->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
return;
}
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(),fromAddr.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_INVALID_REQUEST);
outp.encrypt(candidate->cryptKey());
outp.hmacSet(candidate->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
return;
}
// 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,fromAddr,0,packet.hops(),Packet::VERB_HELLO,Utils::now());
Packet outp(source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_HELLO);
outp.append(packet.packetId());
outp.append(timestamp);
outp.encrypt(existingPeer->cryptKey());
outp.hmacSet(existingPeer->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
return;
}
// 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->fromAddr = fromAddr;
arg->localPort = localPort;
arg->vMajor = vMajor;
arg->vMinor = vMinor;
arg->vRevision = vRevision;
arg->helloPacketId = packet.packetId();
arg->helloTimestamp = timestamp;
_r->topology->addPeer(candidate,&Switch::_CBaddPeerFromHello,arg);
} catch (std::exception &ex) {
TRACE("dropped HELLO from %s(%s): %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped HELLO from %s(%s): unexpected exception",source.toString().c_str(),fromAddr.toString().c_str());
}
}
} // namespace ZeroTier

4
node/PacketDecoder.hpp
Unescape View File

@ -52,7 +52,7 @@ public:
template<unsigned int C2>
PacketDecoder(const Buffer<C2> &b,Demarc::Port localPort,const InetAddress &remoteAddress)
throw(std::out_of_range) :
Packet(b) :
Packet(b),
_receiveTime(Utils::now()),
_localPort(localPort),
_remoteAddress(remoteAddress),
@ -76,6 +76,8 @@ public:
inline uint64_t receiveTime() const throw() { return _receiveTime; }
private:
void _doHELLO(Demarc::Port localPort,const InetAddress &fromAddr);
uint64_t _receiveTime;
Demarc::Port _localPort;
InetAddress _remoteAddress;

516
node/Switch.cpp
Unescape View File

@ -58,13 +58,14 @@ Switch::~Switch()
void Switch::onRemotePacket(Demarc::Port localPort,const InetAddress &fromAddr,const Buffer<4096> &data)
{
Packet packet;
try {
if (data.size() > ZT_PROTO_MIN_FRAGMENT_LENGTH) {
if (data[ZT_PACKET_FRAGMENT_IDX_FRAGMENT_INDICATOR] == ZT_PACKET_FRAGMENT_INDICATOR) {
if (data[ZT_PACKET_FRAGMENT_IDX_FRAGMENT_INDICATOR] == ZT_PACKET_FRAGMENT_INDICATOR)
_handleRemotePacketFragment(localPort,fromAddr,data);
} else if (data.size() > ZT_PROTO_MIN_PACKET_LENGTH) {
else if (data.size() > ZT_PROTO_MIN_PACKET_LENGTH)
_handleRemotePacketHead(localPort,fromAddr,data);
else {
TRACE("dropped runt packet from %s",fromAddr.toString().c_str());
}
}
} catch (std::exception &ex) {
@ -118,7 +119,7 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
if (!np)
return;
std::string signature(Multicaster::signMulticastPacket(_r->identity,network->id(),from,mg,etherType,data,len));
std::string signature(Multicaster::signMulticastPacket(_r->identity,network->id(),from,mg,etherType,data.data(),data.size()));
if (!signature.length()) {
TRACE("failure signing multicast message!");
return;
@ -131,12 +132,12 @@ void Switch::onLocalEthernet(const SharedPtr<Network> &network,const MAC &from,c
outpTmpl.append(from.data,6);
outpTmpl.append(mg.mac().data,6);
outpTmpl.append((uint32_t)mg.adi());
outpTmpl.append(newBloom.data(),ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE);
outpTmpl.append(newbf.data(),ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES);
outpTmpl.append((uint8_t)0); // 0 hops
outpTmpl.append((uint16_t)etherType);
outpTmpl.append((uint16_t)len);
outpTmpl.append((uint16_t)data.size());
outpTmpl.append((uint16_t)signature.length());
outpTmpl.append(data,len);
outpTmpl.append(data.data(),data.size());
outpTmpl.append(signature.data(),signature.length());
outpTmpl.compress();
send(outpTmpl,true);
@ -168,7 +169,7 @@ void Switch::send(const Packet &packet,bool encrypt)
//TRACE("%.16llx %s -> %s (size: %u) (enc: %s)",packet.packetId(),Packet::verbString(packet.verb()),packet.destination().toString().c_str(),packet.size(),(encrypt ? "yes" : "no"));
if (!_trySend(packet,encrypt)) {
Mutex::Lock _l(_txQueue_m);
_txQueue.insert(std::pair< uint64_t,TXQueueEntry >(packet.packetId(),TXQueueEntry(Utils::now(),packet,encrypt)));
_txQueue.insert(std::pair< Address,TXQueueEntry >(packet.destination(),TXQueueEntry(Utils::now(),packet,encrypt)));
}
}
@ -311,7 +312,7 @@ unsigned long Switch::doTimerTasks()
{
Mutex::Lock _l(_txQueue_m);
for(std::map< uint64_t,TXQueueEntry >::iterator i(_txQueue.begin());i!=_txQueue.end();) {
for(std::multimap< Address,TXQueueEntry >::iterator i(_txQueue.begin());i!=_txQueue.end();) {
if (_trySend(i->second.packet,i->second.encrypt))
_txQueue.erase(i++);
else if ((now - i->second.creationTime) > ZT_TRANSMIT_QUEUE_TIMEOUT) {
@ -320,13 +321,12 @@ unsigned long Switch::doTimerTasks()
} else ++i;
}
}
{
Mutex::Lock _l(_rxQueue_m);
for(std::map< uint64_t,RXQueueEntry >::iterator i(_rxQueue.begin());i!=_rxQueue.end();) {
if (_tryHandleRemotePacket(i->second.localPort,i->second.fromAddr,i->second.packet))
_rxQueue.erase(i++);
else if ((now - i->second.creationTime) > ZT_RECEIVE_QUEUE_TIMEOUT) {
TRACE("RX from %s timed out waiting for WHOIS",i->second.packet.source().toString().c_str());
for(std::multimap< Address,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++);
} else ++i;
}
@ -342,7 +342,7 @@ unsigned long Switch::doTimerTasks()
}
}
return std::max(nextDelay,(unsigned long)50); // minimum delay
return std::max(nextDelay,(unsigned long)10); // minimum delay
}
void Switch::announceMulticastGroups(const std::map< SharedPtr<Network>,std::set<MulticastGroup> > &allMemberships)
@ -380,15 +380,16 @@ void Switch::announceMulticastGroups(const std::map< SharedPtr<Network>,std::set
}
}
void Switch::requestWhois(const Address &addr,const SharedPtr<PacketDecoder> &pd)
void Switch::requestWhois(const Address &addr)
{
TRACE("requesting WHOIS for %s",addr.toString().c_str());
{
Mutex::Lock _l(_outstandingWhoisRequests_m);
std::pair< std::map< Address,WhoisRequest >::iterator,bool > entry(_outstandingWhoisRequests.insert(std::pair<Address,WhoisRequest>(addr,WhoisRequest())));
entry.first->second.lastSent = Utils::now();
entry.first->second.retries = 0; // reset retry count if entry already existed
}
_sendWhoisRequest(addr,(const Address *)0,0);
Mutex::Lock _l(_outstandingWhoisRequests_m);
std::pair< std::map< Address,WhoisRequest >::iterator,bool > entry(_outstandingWhoisRequests.insert(std::pair<Address,WhoisRequest>(addr,WhoisRequest())));
entry.first->second.lastSent = Utils::now();
entry.first->second.retries = 0; // reset retry count if entry already existed
entry.first->second.waitingPackets.insert(pd);
}
void Switch::_CBaddPeerFromHello(void *arg,const SharedPtr<Peer> &p,Topology::PeerVerifyResult result)
@ -400,7 +401,7 @@ void Switch::_CBaddPeerFromHello(void *arg,const SharedPtr<Peer> &p,Topology::Pe
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); // terminate any outstanding WHOIS too
req->parent->_finishWhoisRequest(p);
Packet outp(req->source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_HELLO);
@ -449,31 +450,31 @@ void Switch::_CBaddPeerFromWhois(void *arg,const SharedPtr<Peer> &p,Topology::Pe
}
}
void Switch::_finishWhoisRequest(
const SharedPtr<Peer> &peer)
void Switch::_finishWhoisRequest(const SharedPtr<Peer> &peer)
{
Mutex::Lock _l(_outstandingWhoisRequests_m);
std::map< Address,WhoisRequest >::iterator wr(_outstandingWhoisRequests.find(peer->address()));
if (wr != _outstandingWhoisRequests.end()) {
for(std::set<uint64_t>::iterator pid(wr->second.waitingPackets.begin());pid!=wr->second.waitingPackets.end();++pid) {
{
Mutex::Lock _l(_txQueue_m);
std::map< uint64_t,TXQueueEntry >::iterator txitem(_txQueue.find(*pid));
if (txitem != _txQueue.end()) {
if (_trySend(txitem->second.packet,txitem->second.encrypt))
_txQueue.erase(txitem);
}
}
{
Mutex::Lock _l(_rxQueue_m);
std::map< uint64_t,RXQueueEntry >::iterator rxitem(_rxQueue.find(*pid));
if (rxitem != _rxQueue.end()) {
if (_tryHandleRemotePacket(rxitem->second.localPort,rxitem->second.fromAddr,rxitem->second.packet))
_rxQueue.erase(rxitem);
}
}
{
Mutex::Lock _l(_outstandingWhoisRequests_m);
_outstandingWhoisRequests.erase(peer->address());
}
{
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;) {
if (rxi->second->tryDecode(_r))
_rxQueue.erase(rxi++);
else ++rxi;
}
}
{
Mutex::Lock _l(_txQueue_m);
std::pair< std::multimap< Address,TXQueueEntry >::iterator,std::multimap< Address,TXQueueEntry >::iterator > waitingTxQueueItems(_txQueue.equal_range(peer->address()));
for(std::multimap< Address,TXQueueEntry >::iterator txi(waitingTxQueueItems.first);txi!=waitingTxQueueItems.second;) {
if (_trySend(txi->second.packet,txi->second.encrypt))
_txQueue.erase(txi++);
else ++txi;
}
_outstandingWhoisRequests.erase(wr);
}
}
@ -484,7 +485,6 @@ void Switch::_handleRemotePacketFragment(Demarc::Port localPort,const InetAddres
Address destination(fragment.destination());
if (destination != _r->identity.address()) {
// Fragment is not for us, so try to relay it
if (fragment.hops() < ZT_RELAY_MAX_HOPS) {
fragment.incrementHops();
@ -499,7 +499,6 @@ void Switch::_handleRemotePacketFragment(Demarc::Port localPort,const InetAddres
}
} else {
// Fragment looks like ours
uint64_t pid = fragment.packetId();
unsigned int fno = fragment.fragmentNumber();
unsigned int tf = fragment.totalFragments();
@ -531,53 +530,53 @@ void Switch::_handleRemotePacketFragment(Demarc::Port localPort,const InetAddres
if (Utils::countBits(dqe->second.haveFragments |= (1 << fno)) == tf) {
// We have all fragments -- assemble and process full Packet
//TRACE("packet %.16llx is complete, assembling and processing...",pid);
Packet packet(dqe->second.frag0);
SharedPtr<PacketDecoder> packet(dqe->second.frag0);
for(unsigned int f=1;f<tf;++f)
packet.append(dqe->second.frags[f - 1].payload(),dqe->second.frags[f - 1].payloadLength());
packet->append(dqe->second.frags[f - 1].payload(),dqe->second.frags[f - 1].payloadLength());
_defragQueue.erase(dqe);
_handleRemotePacket(localPort,fromAddr,packet);
if (!packet->tryDecode(_r)) {
Mutex::Lock _l(_rxQueue_m);
_rxQueue.insert(std::pair< Address,SharedPtr<PacketDecoder> >(destination,packet));
}
}
} // else this is a duplicate fragment, ignore
}
}
}
bool Switch::_handleRemotePacketHead(Demarc::Port localPort,const InetAddress &fromAddr,const Buffer<4096> &data)
void Switch::_handleRemotePacketHead(Demarc::Port localPort,const InetAddress &fromAddr,const Buffer<4096> &data)
{
Packet packet(data);
SharedPtr<PacketDecoder> packet(new PacketDecoder(data,localPort,fromAddr));
Address destination(packet->destination());
Address destination(packet.destination());
if (destination != _r->identity.address()) {
// Packet is not for us, so try to relay it
if (packet.hops() < ZT_RELAY_MAX_HOPS) {
packet.incrementHops();
if (packet->hops() < ZT_RELAY_MAX_HOPS) {
packet->incrementHops();
SharedPtr<Peer> relayTo = _r->topology->getPeer(destination);
if ((relayTo)&&(relayTo->send(_r,packet.data(),packet.size(),true,Packet::VERB_NOP,Utils::now()))) {
// TODO: don't unite immediately, wait until the peers have exchanged a packet or two
unite(packet.source(),destination,false); // periodically try to get them to talk directly
if ((relayTo)&&(relayTo->send(_r,packet->data(),packet->size(),true,Packet::VERB_NOP,Utils::now()))) {
unite(packet->source(),destination,false); // periodically try to get them to talk directly
} else {
relayTo = _r->topology->getBestSupernode();
if (relayTo)
relayTo->send(_r,packet.data(),packet.size(),true,Packet::VERB_NOP,Utils::now());
relayTo->send(_r,packet->data(),packet->size(),true,Packet::VERB_NOP,Utils::now());
}
} else {
TRACE("dropped relay %s(%s) -> %s, max hops exceeded",packet.source().toString().c_str(),fromAddr.toString().c_str(),destination.toString().c_str());
TRACE("dropped relay %s(%s) -> %s, max hops exceeded",packet->source().toString().c_str(),fromAddr.toString().c_str(),destination.toString().c_str());
}
} else if (packet.fragmented()) {
} else if (packet->fragmented()) {
// Packet is the head of a fragmented packet series
uint64_t pid = packet.packetId();
uint64_t pid = packet->packetId();
Mutex::Lock _l(_defragQueue_m);
std::map< uint64_t,DefragQueueEntry >::iterator dqe(_defragQueue.find(pid));
if (dqe == _defragQueue.end()) {
// If we have no other fragments yet, create an entry and save the head
DefragQueueEntry &dq = _defragQueue[pid];
dq.creationTime = Utils::now();
dq.frag0 = packet;
@ -586,398 +585,30 @@ bool Switch::_handleRemotePacketHead(Demarc::Port localPort,const InetAddress &f
//TRACE("fragment (0/?) of %.16llx from %s",pid,fromAddr.toString().c_str());
} else if (!(dqe->second.haveFragments & 1)) {
// If we have other fragments but no head, see if we are complete with the head
if ((dqe->second.totalFragments)&&(Utils::countBits(dqe->second.haveFragments |= 1) == dqe->second.totalFragments)) {
// We have all fragments -- assemble and process full Packet
//TRACE("packet %.16llx is complete, assembling and processing...",pid);
// packet already contains head, so append fragments
for(unsigned int f=1;f<dqe->second.totalFragments;++f)
packet.append(dqe->second.frags[f - 1].payload(),dqe->second.frags[f - 1].payloadLength());
packet->append(dqe->second.frags[f - 1].payload(),dqe->second.frags[f - 1].payloadLength());
_defragQueue.erase(dqe);
_handleRemotePacket(localPort,fromAddr,packet);
if (!packet->tryDecode(_r)) {
Mutex::Lock _l(_rxQueue_m);
_rxQueue.insert(std::pair< Address,SharedPtr<PacketDecoder> >(destination,packet));
}
} else {
// Still waiting on more fragments, so queue the head
dqe->second.frag0 = packet;
}
} // else this is a duplicate head, ignore
} else {
// Packet is unfragmented, so just process it
_handleRemotePacket(localPort,fromAddr,packet);
}
}
//////////////////// OBSOLETE
bool Switch::_tryHandleRemotePacket(Demarc::Port localPort,const InetAddress &fromAddr,Packet &packet)
{
Address source(packet.source());
if ((!packet.encrypted())&&(packet.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(),fromAddr.toString().c_str());
_doHELLO(localPort,fromAddr,packet);
return true;
}
SharedPtr<Peer> peer = _r->topology->getPeer(source);
if (peer) {
uint64_t now = Utils::now();
unsigned int latency = 0;
if (!packet.hmacVerify(peer->macKey())) {
TRACE("dropped packet from %s(%s), HMAC authentication failed (size: %u)",source.toString().c_str(),fromAddr.toString().c_str(),packet.size());
return true;
}
if (packet.encrypted()) {
packet.decrypt(peer->cryptKey());
} else {
// 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(packet.verb()),source.toString().c_str(),fromAddr.toString().c_str());
}
if (!packet.uncompress()) {
TRACE("dropped packet from %s(%s), compressed data invalid",source.toString().c_str(),fromAddr.toString().c_str());
return true;
if (!packet->tryDecode(_r)) {
Mutex::Lock _l(_rxQueue_m);
_rxQueue.insert(std::pair< Address,SharedPtr<PacketDecoder> >(destination,packet));
}
switch(packet.verb()) {
case Packet::VERB_NOP:
TRACE("NOP from %s(%s)",source.toString().c_str(),fromAddr.toString().c_str());
break;
case Packet::VERB_HELLO:
// HELLO is normally handled up top, but this is legal. Pointless, but legal.
_doHELLO(localPort,fromAddr,packet);
break;
case Packet::VERB_ERROR:
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(),fromAddr.toString().c_str(),Packet::verbString(inReVerb));
#endif
// TODO:
// The fact is that the protocol works fine without error handling.
// The only error that really needs to be handled here is duplicate
// 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(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped ERROR from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_OK:
try {
Packet::Verb inReVerb = (Packet::Verb)packet[ZT_PROTO_VERB_OK_IDX_IN_RE_VERB];
switch(inReVerb) {
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(),fromAddr.toString().c_str(),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(),fromAddr.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);
break;
default:
TRACE("%s(%s): OK(%s)",source.toString().c_str(),fromAddr.toString().c_str(),Packet::verbString(inReVerb));
break;
}
} catch (std::exception &ex) {
TRACE("dropped OK from %s(%s): unexpected exception: %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped OK from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_WHOIS:
if (packet.payloadLength() == ZT_ADDRESS_LENGTH) {
SharedPtr<Peer> p(_r->topology->getPeer(Address(packet.payload())));
if (p) {
Packet outp(source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_WHOIS);
outp.append(packet.packetId());
p->identity().serialize(outp,false);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
TRACE("sent WHOIS response to %s for %s",source.toString().c_str(),Address(packet.payload()).toString().c_str());
} else {
Packet outp(source,_r->identity.address(),Packet::VERB_ERROR);
outp.append((unsigned char)Packet::VERB_WHOIS);
outp.append(packet.packetId());
outp.append((unsigned char)Packet::ERROR_NOT_FOUND);
outp.append(packet.payload(),ZT_ADDRESS_LENGTH);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
_r->demarc->send(localPort,fromAddr,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());
}
} else {
TRACE("dropped WHOIS from %s(%s): missing or invalid address",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_RENDEZVOUS:
try {
Address with(packet.field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ZTADDRESS,ZT_ADDRESS_LENGTH));
RendezvousQueueEntry qe;
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];
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;
}
} else {
TRACE("dropped corrupt RENDEZVOUS from %s(%s) (bad address or port)",source.toString().c_str(),fromAddr.toString().c_str());
}
} else {
TRACE("ignored RENDEZVOUS from %s(%s) to meet unknown peer %s",source.toString().c_str(),fromAddr.toString().c_str(),with.toString().c_str());
}
} catch (std::exception &ex) {
TRACE("dropped RENDEZVOUS from %s(%s): %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped RENDEZVOUS from %s(%s): unexpected exception",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_FRAME:
try {
SharedPtr<Network> network(_r->nc->network(packet.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 ((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);
}
} else {
TRACE("dropped FRAME from %s(%s): not a member of closed network %llu",source.toString().c_str(),fromAddr.toString().c_str(),network->id());
}
} else {
TRACE("dropped FRAME from %s(%s): network %llu unknown",source.toString().c_str(),fromAddr.toString().c_str(),packet.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(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped FRAME from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_MULTICAST_LIKE:
try {
unsigned int ptr = ZT_PACKET_IDX_PAYLOAD;
unsigned int numAccepted = 0;
// Iterate through 18-byte network,MAC,ADI tuples:
while ((ptr + 18) <= packet.size()) {
uint64_t nwid = packet.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);
++numAccepted;
} else {
TRACE("ignored MULTICAST_LIKE from %s(%s): not a member of closed network %llu",source.toString().c_str(),fromAddr.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(),fromAddr.toString().c_str(),nwid);
}
}
Packet outp(source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_MULTICAST_LIKE);
outp.append(packet.packetId());
outp.append((uint16_t)numAccepted);
outp.encrypt(peer->cryptKey());
outp.hmacSet(peer->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
} catch (std::exception &ex) {
TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
case Packet::VERB_MULTICAST_FRAME:
try {
SharedPtr<Network> network(_r->nc->network(packet.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 (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(),fromAddr.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());
return false;
} else if (Multicaster::verifyMulticastPacket(originalSubmitter->identity(),fromMac,mg,etherType,data,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);
} else {
LOG("rejected MULTICAST_FRAME from %s(%s) due to failed signature check (claims original sender %s)",source.toString().c_str(),fromAddr.toString().c_str(),originalSubmitterAddress.toString().c_str());
}
} else {
TRACE("dropped redundant MULTICAST_FRAME from %s(%s)",source.toString().c_str(),fromAddr.toString().c_str());
}
} else {
TRACE("dropped MULTICAST_FRAME from %s(%s): invalid short packet",source.toString().c_str(),fromAddr.toString().c_str());
}
} else {
TRACE("dropped MULTICAST_FRAME from %s(%s): not a member of closed network %llu",source.toString().c_str(),fromAddr.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(),fromAddr.toString().c_str(),packet.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(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: (unknown)",source.toString().c_str(),fromAddr.toString().c_str());
}
break;
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(),fromAddr.toString().c_str());
break;
}
// Update peer timestamps and learn new links. This must only ever
// be called on an authenticated and technically valid packet, since
// we only learn paths to peers over the WAN by hearing directly
// from them over those paths. (Or by having them authoritatively
// and statically defined, like with supernodes, but that's done
// elsewhere.)
peer->onReceive(_r,localPort,fromAddr,latency,packet.hops(),packet.verb(),now);
} else {
_requestWhois(source,packet.packetId());
return false;
}
return true;
}
void Switch::_doHELLO(Demarc::Port localPort,const InetAddress &fromAddr,Packet &packet)
{
Address source(packet.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);
SharedPtr<Peer> candidate(new Peer(_r->identity,id));
candidate->setPathAddress(fromAddr,false);
// Initial sniff test
if (protoVersion != ZT_PROTO_VERSION) {
TRACE("rejected HELLO from %s(%s): invalid protocol version",source.toString().c_str(),fromAddr.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,fromAddr,outp.data(),outp.size(),-1);
return;
}
if (id.address().isReserved()) {
TRACE("rejected HELLO from %s(%s): identity has reserved address",source.toString().c_str(),fromAddr.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_IDENTITY_INVALID);
outp.encrypt(candidate->cryptKey());
outp.hmacSet(candidate->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
return;
}
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(),fromAddr.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_INVALID_REQUEST);
outp.encrypt(candidate->cryptKey());
outp.hmacSet(candidate->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
return;
}
// 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,fromAddr,0,packet.hops(),Packet::VERB_HELLO,Utils::now());
Packet outp(source,_r->identity.address(),Packet::VERB_OK);
outp.append((unsigned char)Packet::VERB_HELLO);
outp.append(packet.packetId());
outp.append(timestamp);
outp.encrypt(existingPeer->cryptKey());
outp.hmacSet(existingPeer->macKey());
_r->demarc->send(localPort,fromAddr,outp.data(),outp.size(),-1);
return;
}
// 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->fromAddr = fromAddr;
arg->localPort = localPort;
arg->vMajor = vMajor;
arg->vMinor = vMinor;
arg->vRevision = vRevision;
arg->helloPacketId = packet.packetId();
arg->helloTimestamp = timestamp;
_r->topology->addPeer(candidate,&Switch::_CBaddPeerFromHello,arg);
} catch (std::exception &ex) {
TRACE("dropped HELLO from %s(%s): %s",source.toString().c_str(),fromAddr.toString().c_str(),ex.what());
} catch ( ... ) {
TRACE("dropped HELLO from %s(%s): unexpected exception",source.toString().c_str(),fromAddr.toString().c_str());
}
}
@ -998,6 +629,7 @@ Address Switch::_sendWhoisRequest(const Address &addr,const Address *peersAlread
bool Switch::_trySend(const Packet &packet,bool encrypt)
{
SharedPtr<Peer> peer(_r->topology->getPeer(packet.destination()));
if (peer) {
uint64_t now = Utils::now();
@ -1050,7 +682,7 @@ bool Switch::_trySend(const Packet &packet,bool encrypt)
return false;
}
_requestWhois(packet.destination(),packet.packetId());
requestWhois(packet.destination());
return false;
}

17
node/Switch.hpp
Unescape View File

@ -45,6 +45,7 @@
#include "SharedPtr.hpp"
#include "Demarc.hpp"
#include "Multicaster.hpp"
#include "PacketDecoder.hpp"
namespace ZeroTier {
@ -154,9 +155,8 @@ public:
* Request WHOIS on a given address
*
* @param addr Address to look up
* @param pd Packet decoder to link to request or NULL for none
*/
void requestWhois(const Address &addr,const SharedPtr<PacketDecoder> &pd);
void requestWhois(const Address &addr);
private:
struct _CBaddPeerFromHello_Data
@ -192,11 +192,6 @@ private:
const InetAddress &fromAddr,
const Buffer<4096> &data);
void _doHELLO(
Demarc::Port localPort,
const InetAddress &fromAddr,
Packet &packet);
Address _sendWhoisRequest(
const Address &addr,
const Address *peersAlreadyConsulted,
@ -214,11 +209,13 @@ private:
uint64_t lastSent;
Address peersConsulted[ZT_MAX_WHOIS_RETRIES]; // by retry
unsigned int retries; // 0..ZT_MAX_WHOIS_RETRIES
std::set< SharedPtr<PacketDecoder> > waitingPackets;
};
std::map< Address,WhoisRequest > _outstandingWhoisRequests;
Mutex _outstandingWhoisRequests_m;
std::multimap< Address,SharedPtr<PacketDecoder> > _rxQueue;
Mutex _rxQueue_m;
struct TXQueueEntry
{
TXQueueEntry() {}
@ -231,13 +228,13 @@ private:
Packet packet; // unencrypted/untagged for TX queue
bool encrypt;
};
std::map< uint64_t,TXQueueEntry > _txQueue;
std::multimap< Address,TXQueueEntry > _txQueue;
Mutex _txQueue_m;
struct DefragQueueEntry
{
uint64_t creationTime;
Packet frag0;
SharedPtr<PacketDecoder> frag0;
Packet::Fragment frags[ZT_MAX_PACKET_FRAGMENTS - 1];
unsigned int totalFragments; // 0 if only frag0 received, waiting for frags
uint32_t haveFragments; // bit mask, LSB to MSB

Write Preview
Loading…
Cancel
Save