From 84619a77884b202a7d57dc188cfb8424307d9f61 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Thu, 6 Feb 2020 18:06:50 -0800 Subject: [PATCH] A bunch more refactoring including splitting Switch into VL1 and VL2 --- {node => attic}/IncomingPacket.cpp | 0 {node => attic}/IncomingPacket.hpp | 0 node/Buf.cpp | 10 +- node/Buf.hpp | 233 +++++++----- node/CMakeLists.txt | 5 +- node/Constants.hpp | 5 + node/Defragmenter.hpp | 204 +++++------ node/FCV.hpp | 278 +++++++++++++++ node/InetAddress.hpp | 11 +- node/Node.cpp | 5 + node/OS.hpp | 1 + node/Path.hpp | 4 +- node/Protocol.cpp | 114 ++---- node/Protocol.hpp | 112 +++--- node/Salsa20.hpp | 1 + node/VL1.cpp | 546 +++++++++++++++++++++++++++++ node/VL1.hpp | 88 +++++ node/VL2.hpp | 54 +++ 18 files changed, 1287 insertions(+), 384 deletions(-) rename {node => attic}/IncomingPacket.cpp (100%) rename {node => attic}/IncomingPacket.hpp (100%) create mode 100644 node/FCV.hpp create mode 100644 node/VL1.cpp create mode 100644 node/VL1.hpp create mode 100644 node/VL2.hpp diff --git a/node/IncomingPacket.cpp b/attic/IncomingPacket.cpp similarity index 100% rename from node/IncomingPacket.cpp rename to attic/IncomingPacket.cpp diff --git a/node/IncomingPacket.hpp b/attic/IncomingPacket.hpp similarity index 100% rename from node/IncomingPacket.hpp rename to attic/IncomingPacket.hpp diff --git a/node/Buf.cpp b/node/Buf.cpp index 3816471d8..e36f1d5de 100644 --- a/node/Buf.cpp +++ b/node/Buf.cpp @@ -36,7 +36,7 @@ void _Buf_release(void *ptr,std::size_t sz) break; } - ((Buf<> *)ptr)->__nextInPool = bb; + ((Buf *)ptr)->__nextInPool = bb; #ifdef __GNUC__ __sync_fetch_and_and(&_Buf_pool,(uintptr_t)ptr); #else @@ -59,18 +59,18 @@ void *_Buf_get() break; } - Buf<> *b; + Buf *b; if (bb == 0) { #ifdef __GNUC__ __sync_fetch_and_and(&_Buf_pool,bb); #else s_pool.store(bb); #endif - b = (Buf<> *)malloc(sizeof(Buf<>)); + b = (Buf *)malloc(sizeof(Buf<>)); if (!b) throw std::bad_alloc(); } else { - b = (Buf<> *)bb; + b = (Buf *)bb; #ifdef __GNUC__ __sync_fetch_and_and(&_Buf_pool,b->__nextInPool); #else @@ -103,7 +103,7 @@ void freeBufPool() #endif while (bb != 0) { - uintptr_t next = ((Buf<> *)bb)->__nextInPool; + uintptr_t next = ((Buf *)bb)->__nextInPool; free((void *)bb); bb = next; } diff --git a/node/Buf.hpp b/node/Buf.hpp index 6c5c3225c..677fbcf77 100644 --- a/node/Buf.hpp +++ b/node/Buf.hpp @@ -20,11 +20,14 @@ #include "SharedPtr.hpp" #include "Mutex.hpp" #include "TriviallyCopyable.hpp" +#include "FCV.hpp" #include #include #include #include +#include +#include #ifndef __GNUC__ #include @@ -47,7 +50,7 @@ void _Buf_release(void *ptr,std::size_t sz); void *_Buf_get(); /** - * Free buffers in the pool + * Free all instances of Buf in shared pool. * * New buffers will be created and the pool repopulated if get() is called * and outstanding buffers will still be returned to the pool. This just @@ -55,11 +58,6 @@ void *_Buf_get(); */ void freeBufPool(); -/** - * Macro to declare and get a new buffer templated with the given type - */ -#define ZT_GET_NEW_BUF(vvv,xxx) SharedPtr< Buf > vvv(reinterpret_cast< Buf * >(_Buf_get())) - /** * Buffer and methods for branch-free bounds-checked data assembly and parsing * @@ -100,10 +98,9 @@ void freeBufPool(); * * @tparam U Type to overlap with data bytes in data union (can't be larger than ZT_BUF_MEM_SIZE) */ -template class Buf { - friend class SharedPtr< Buf >; + friend class SharedPtr< Buf >; friend void _Buf_release(void *,std::size_t); friend void *_Buf_get(); friend void freeBufPool(); @@ -139,17 +136,65 @@ public: unsigned int e; }; - ZT_ALWAYS_INLINE Buf() {} + /** + * Assemble all slices in a vector into a single slice starting at position 0 + * + * The returned slice will start at 0 and contain the entire vector unless the + * vector is too large to fit in a single buffer. If that or any other error + * occurs the returned slice will be empty and contain a NULL Buf. + * + * The vector may be modified by this function and should be considered + * undefined after it is called. + * + * @tparam FCVC Capacity of FCV (generally inferred automatically) + * @param fcv FCV containing one or more slices + * @return Single slice containing fully assembled buffer (empty on error) + */ + template + static ZT_ALWAYS_INLINE Buf::Slice assembleSliceVector(FCV &fcv) + { + Buf::Slice r; - template - ZT_ALWAYS_INLINE Buf(const Buf &b) { memcpy(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE); } + typename FCV::const_iterator s(fcv.begin()); + unsigned int l = s->e - s->s; + if (l <= ZT_BUF_MEM_SIZE) { + r.b.move(s->b); + if (s->s > 0) + memmove(r.b->b,r.b->b + s->s,l); + r.e = l; + + while (++s != fcv.end()) { + l = s->e - s->s; + if (l > (ZT_BUF_MEM_SIZE - r.e)) { + r.b.zero(); + r.e = 0; + break; + } + memcpy(r.b->b + r.e,s->b->b + s->s,l); + s->b.zero(); // let go of buffer in vector as soon as possible + r.e += l; + } + } + + return r; + } + + ZT_ALWAYS_INLINE Buf() {} + ZT_ALWAYS_INLINE Buf(const Buf &b2) { memcpy(b,b2.b,ZT_BUF_MEM_SIZE); } + + ZT_ALWAYS_INLINE Buf &operator=(const Buf &b2) + { + if (this != &b2) + memcpy(b,b2.b,ZT_BUF_MEM_SIZE); + return *this; + } /** * Get obtains a buffer from the pool or allocates a new buffer if the pool is empty * * @return Buffer instance */ - static ZT_ALWAYS_INLINE SharedPtr< Buf > get() { return SharedPtr((Buf *)_Buf_get()); } + static ZT_ALWAYS_INLINE SharedPtr< Buf > get() { return SharedPtr((Buf *)_Buf_get()); } /** * Check for overflow beyond the size of the buffer @@ -175,29 +220,14 @@ public: static ZT_ALWAYS_INLINE bool readOverflow(const int &ii,const unsigned int size) { return ((ii - (int)size) > 0); } /** - * Shortcut to cast between buffers whose data can be viewed through a different struct type - * - * @tparam X A packed struct or other primitive type that should be placed in the data union - * @return Reference to this Buf templated with the supplied parameter + * Set all memory to zero */ - template - ZT_ALWAYS_INLINE Buf &view() { return *reinterpret_cast< Buf * >(this); } + ZT_ALWAYS_INLINE void clear() { memset(b,0,ZT_BUF_MEM_SIZE); } /** - * Shortcut to cast between buffers whose data can be viewed through a different struct type - * - * @tparam X A packed struct or other primitive type that should be placed in the data union - * @return Reference to this Buf templated with the supplied parameter + * Zero security critical data using Utils::burn() to ensure it's never optimized out. */ - template - ZT_ALWAYS_INLINE const Buf &view() const { return *reinterpret_cast< Buf * >(this); } - - /** - * Zero memory - * - * For performance reasons Buf does not do this on every get(). - */ - ZT_ALWAYS_INLINE void clear() { memset(data.bytes,0,ZT_BUF_MEM_SIZE); } + ZT_ALWAYS_INLINE void burn() { Utils::burn(b,ZT_BUF_MEM_SIZE); } /** * Read a byte @@ -208,7 +238,7 @@ public: ZT_ALWAYS_INLINE uint8_t rI8(int &ii) const { const int s = ii++; - return data.bytes[(unsigned int)s & ZT_BUF_MEM_MASK]; + return b[(unsigned int)s & ZT_BUF_MEM_MASK]; } /** @@ -226,7 +256,7 @@ public: ((uint16_t)data.bytes[s] << 8U) | (uint16_t)data.bytes[s + 1]); #else - return Utils::ntoh(*reinterpret_cast(data.bytes + s)); + return Utils::ntoh(*reinterpret_cast(b + s)); #endif } @@ -247,7 +277,7 @@ public: ((uint32_t)data.bytes[s + 2] << 8U) | (uint32_t)data.bytes[s + 3]); #else - return Utils::ntoh(*reinterpret_cast(data.bytes + s)); + return Utils::ntoh(*reinterpret_cast(b + s)); #endif } @@ -272,7 +302,7 @@ public: ((uint64_t)data.bytes[s + 6] << 8U) | (uint64_t)data.bytes[s + 7]); #else - return Utils::ntoh(*reinterpret_cast(data.bytes + s)); + return Utils::ntoh(*reinterpret_cast(b + s)); #endif } @@ -295,7 +325,7 @@ public: ZT_ALWAYS_INLINE int rO(int &ii,T &obj) const { if (ii < ZT_BUF_MEM_SIZE) { - int ms = obj.unmarshal(data.bytes + ii,ZT_BUF_MEM_SIZE - ii); + int ms = obj.unmarshal(b + ii,ZT_BUF_MEM_SIZE - ii); if (ms > 0) ii += ms; return ms; @@ -316,10 +346,10 @@ public: */ ZT_ALWAYS_INLINE char *rS(int &ii,char *const buf,const unsigned int bufSize) const { - const char *const s = (const char *)(data.bytes + ii); + const char *const s = (const char *)(b + ii); const int sii = ii; while (ii < ZT_BUF_MEM_SIZE) { - if (data.bytes[ii++] == 0) { + if (b[ii++] == 0) { memcpy(buf,s,ii - sii); return buf; } @@ -342,9 +372,9 @@ public: */ ZT_ALWAYS_INLINE const char *rSnc(int &ii) const { - const char *const s = (const char *)(data.bytes + ii); + const char *const s = (const char *)(b + ii); while (ii < ZT_BUF_MEM_SIZE) { - if (data.bytes[ii++] == 0) + if (b[ii++] == 0) return s; } return nullptr; @@ -364,7 +394,7 @@ public: ZT_ALWAYS_INLINE uint8_t *rB(int &ii,void *bytes,unsigned int len) const { if ((ii += (int)len) <= ZT_BUF_MEM_SIZE) { - memcpy(bytes,data.bytes + ii,len); + memcpy(bytes,b + ii,len); return reinterpret_cast(bytes); } return nullptr; @@ -385,7 +415,7 @@ public: */ ZT_ALWAYS_INLINE const uint8_t *rBnc(int &ii,unsigned int len) const { - const uint8_t *const b = data.bytes + ii; + const uint8_t *const b = b + ii; return ((ii += (int)len) <= ZT_BUF_MEM_SIZE) ? b : nullptr; } @@ -398,7 +428,7 @@ public: ZT_ALWAYS_INLINE void wI(int &ii,uint8_t n) { const int s = ii++; - data[(unsigned int)s & ZT_BUF_MEM_MASK] = n; + b[(unsigned int)s & ZT_BUF_MEM_MASK] = n; } /** @@ -412,10 +442,10 @@ public: const unsigned int s = ((unsigned int)ii) & ZT_BUF_MEM_MASK; ii += 2; #ifdef ZT_NO_UNALIGNED_ACCESS - data[s] = (uint8_t)(n >> 8U); - data[s + 1] = (uint8_t)n; + b[s] = (uint8_t)(n >> 8U); + b[s + 1] = (uint8_t)n; #else - *reinterpret_cast(data.bytes + s) = Utils::hton(n); + *reinterpret_cast(b + s) = Utils::hton(n); #endif } @@ -430,12 +460,12 @@ public: const unsigned int s = ((unsigned int)ii) & ZT_BUF_MEM_MASK; ii += 4; #ifdef ZT_NO_UNALIGNED_ACCESS - data[s] = (uint8_t)(n >> 24U); - data[s + 1] = (uint8_t)(n >> 16U); - data[s + 2] = (uint8_t)(n >> 8U); - data[s + 3] = (uint8_t)n; + b[s] = (uint8_t)(n >> 24U); + b[s + 1] = (uint8_t)(n >> 16U); + b[s + 2] = (uint8_t)(n >> 8U); + b[s + 3] = (uint8_t)n; #else - *reinterpret_cast(data.bytes + s) = Utils::hton(n); + *reinterpret_cast(b + s) = Utils::hton(n); #endif } @@ -450,16 +480,16 @@ public: const unsigned int s = ((unsigned int)ii) & ZT_BUF_MEM_MASK; ii += 8; #ifdef ZT_NO_UNALIGNED_ACCESS - data[s] = (uint8_t)(n >> 56U); - data[s + 1] = (uint8_t)(n >> 48U); - data[s + 2] = (uint8_t)(n >> 40U); - data[s + 3] = (uint8_t)(n >> 32U); - data[s + 4] = (uint8_t)(n >> 24U); - data[s + 5] = (uint8_t)(n >> 16U); - data[s + 6] = (uint8_t)(n >> 8U); - data[s + 7] = (uint8_t)n; + b[s] = (uint8_t)(n >> 56U); + b[s + 1] = (uint8_t)(n >> 48U); + b[s + 2] = (uint8_t)(n >> 40U); + b[s + 3] = (uint8_t)(n >> 32U); + b[s + 4] = (uint8_t)(n >> 24U); + b[s + 5] = (uint8_t)(n >> 16U); + b[s + 6] = (uint8_t)(n >> 8U); + b[s + 7] = (uint8_t)n; #else - *reinterpret_cast(data.bytes + s) = Utils::hton(n); + *reinterpret_cast(b + s) = Utils::hton(n); #endif } @@ -475,7 +505,7 @@ public: { const int s = ii; if ((s + T::marshalSizeMax()) <= ZT_BUF_MEM_SIZE) { - int ms = t.marshal(data.bytes + s); + int ms = t.marshal(b + s); if (ms > 0) ii += ms; } else { @@ -513,42 +543,63 @@ public: { const int s = ii; if ((ii += (int)len) <= ZT_BUF_MEM_SIZE) - memcpy(data.bytes + s,bytes,len); + memcpy(b + s,bytes,len); } - template - ZT_ALWAYS_INLINE Buf &operator=(const Buf &b) const - { - memcpy(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE); - return *this; - } - - template - ZT_ALWAYS_INLINE bool operator==(const Buf &b) const { return (memcmp(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE) == 0); } - template - ZT_ALWAYS_INLINE bool operator!=(const Buf &b) const { return (memcmp(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE) != 0); } - template - ZT_ALWAYS_INLINE bool operator<(const Buf &b) const { return (memcmp(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE) < 0); } - template - ZT_ALWAYS_INLINE bool operator<=(const Buf &b) const { return (memcmp(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE) <= 0); } - template - ZT_ALWAYS_INLINE bool operator>(const Buf &b) const { return (memcmp(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE) > 0); } - template - ZT_ALWAYS_INLINE bool operator>=(const Buf &b) const { return (memcmp(data.bytes,b.data.bytes,ZT_BUF_MEM_SIZE) >= 0); } - /** - * Raw data and fields (if U template parameter is set) - * - * The extra eight bytes permit silent overflow of integer types without reading or writing - * beyond Buf's memory and without branching or extra masks. They can be ignored otherwise. + * @return Capacity of this buffer (usable size of data.bytes) */ - ZT_PACKED_STRUCT(union { - uint8_t bytes[ZT_BUF_MEM_SIZE + 8]; - U fields; - }) data; + static constexpr unsigned int capacity() { return ZT_BUF_MEM_SIZE; } + + /** + * Cast data in 'b' to a (usually packed) structure type + * + * Warning: this does no bounds checking. It should only be used with packed + * struct types designed for use in packet decoding such as those in + * Protocol.hpp, and if 'i' is non-zero the caller must check bounds. + * + * @tparam T Structure type to cast 'b' to + * @param i Index of start of structure (default: 0) + * @return Reference to 'b' cast to type T + */ + template + ZT_ALWAYS_INLINE T &as(const unsigned int i = 0) { return *reinterpret_cast(b + i); } + + /** + * Cast data in 'b' to a (usually packed) structure type (const) + * + * Warning: this does no bounds checking. It should only be used with packed + * struct types designed for use in packet decoding such as those in + * Protocol.hpp, and if 'i' is non-zero the caller must check bounds. + * + * @tparam T Structure type to cast 'b' to + * @param i Index of start of structure (default: 0) + * @return Reference to 'b' cast to type T + */ + template + ZT_ALWAYS_INLINE const T &as(const unsigned int i = 0) const { return *reinterpret_cast(b + i); } + + ZT_ALWAYS_INLINE bool operator==(const Buf &b2) const { return (memcmp(b,b2.b,ZT_BUF_MEM_SIZE) == 0); } + ZT_ALWAYS_INLINE bool operator!=(const Buf &b2) const { return (memcmp(b,b2.b,ZT_BUF_MEM_SIZE) != 0); } + ZT_ALWAYS_INLINE bool operator<(const Buf &b2) const { return (memcmp(b,b2.b,ZT_BUF_MEM_SIZE) < 0); } + ZT_ALWAYS_INLINE bool operator<=(const Buf &b2) const { return (memcmp(b,b2.b,ZT_BUF_MEM_SIZE) <= 0); } + ZT_ALWAYS_INLINE bool operator>(const Buf &b2) const { return (memcmp(b,b2.b,ZT_BUF_MEM_SIZE) > 0); } + ZT_ALWAYS_INLINE bool operator>=(const Buf &b2) const { return (memcmp(b,b2.b,ZT_BUF_MEM_SIZE) >= 0); } + + /** + * Raw data held in buffer + * + * The additional eight bytes should not be used and should be considered undefined. + * They exist to allow reads and writes of integer types to silently overflow if a + * read or write is performed at the end of the buffer. + */ + uint8_t b[ZT_BUF_MEM_SIZE + 8]; private: - volatile uintptr_t __nextInPool; // next item in free pool if this Buf is in Buf_pool + // Next item in free buffer pool linked list if Buf is placed in pool, undefined and unused otherwise + volatile uintptr_t __nextInPool; + + // Reference counter for SharedPtr<> AtomicCounter __refCount; }; diff --git a/node/CMakeLists.txt b/node/CMakeLists.txt index 920efc2d6..358bbaa73 100644 --- a/node/CMakeLists.txt +++ b/node/CMakeLists.txt @@ -15,6 +15,7 @@ set(core_headers Defragmenter.hpp Dictionary.hpp ECC384.hpp + FCV.hpp Hashtable.hpp Identity.hpp InetAddress.hpp @@ -45,6 +46,8 @@ set(core_headers Trace.hpp TriviallyCopyable.hpp Utils.hpp + VL1.hpp + VL2.hpp ) set(core_src @@ -59,7 +62,6 @@ set(core_src ECC384.cpp Endpoint.cpp Identity.cpp - IncomingPacket.cpp InetAddress.cpp Locator.cpp LZ4.cpp @@ -80,6 +82,7 @@ set(core_src Topology.cpp Trace.cpp Utils.cpp + VL1.cpp ) add_library(${PROJECT_NAME} STATIC ${core_src} ${core_headers}) diff --git a/node/Constants.hpp b/node/Constants.hpp index f348523b0..538c1f6be 100644 --- a/node/Constants.hpp +++ b/node/Constants.hpp @@ -56,6 +56,11 @@ */ #define ZT_MAX_PACKET_FRAGMENTS 11 +/** + * Anti-DOS limit on the maximum incoming fragments per path + */ +#define ZT_MAX_INCOMING_FRAGMENTS_PER_PATH 32 + /** * Sanity limit on the maximum size of a network config object */ diff --git a/node/Defragmenter.hpp b/node/Defragmenter.hpp index 7dea7187f..faadf2e29 100644 --- a/node/Defragmenter.hpp +++ b/node/Defragmenter.hpp @@ -21,6 +21,7 @@ #include "Hashtable.hpp" #include "Mutex.hpp" #include "Path.hpp" +#include "FCV.hpp" #include #include @@ -40,20 +41,27 @@ namespace ZeroTier { * Here be dragons! * * @tparam MF Maximum number of fragments that each message can possess + * @tparam GCS Garbage collection target size for the incoming message queue + * @tparam GCT Garbage collection trigger threshold, usually 2X GCS */ -template +template class Defragmenter { public: /** - * Error codes for assemble() + * Return values from assemble() */ - enum ErrorCode + enum ResultCode { /** - * No error occurred + * No error occurred, fragment accepted */ - ERR_NONE, + OK, + + /** + * Message fully assembled and placed in message vector + */ + COMPLETE, /** * This fragment duplicates another with the same fragment number for this message @@ -82,32 +90,6 @@ public: ERR_OUT_OF_MEMORY }; - /** - * Return tuple for assemble() - */ - struct Result - { - ZT_ALWAYS_INLINE Result() : message(),messageFragmentCount(0),error(Defragmenter::ERR_NONE) {} - explicit ZT_ALWAYS_INLINE Result(const Defragmenter::ErrorCode e) : message(),messageFragmentCount(0),error(e) {} - - /** - * Fully assembled message as a series of slices of fragments - */ - Buf<>::Slice message[MF]; - - /** - * Fully assembled message fragment count (number of slices) - * - * This will be nonzero if the message is fully assembled. - */ - unsigned int messageFragmentCount; - - /** - * Error code or ERR_NONE if none - */ - Defragmenter::ErrorCode error; - }; - /** * Process a fragment of a multi-part message * @@ -132,119 +114,76 @@ public: * off, so the SharedPtr<> passed in as 'fragment' will be NULL after this * function is called. * - * The result returned by this function is a structure containing a series - * of assembled and complete fragments, a fragment count, and an error. - * If the message fragment count is non-zero then the message has been - * successfully assembled. If the fragment count is zero then an error may - * have occurred or the message may simply not yet be complete. - * - * The calling code must decide what to do with the assembled and ordered - * fragments, such as memcpy'ing them into a contiguous buffer or handling - * them as a vector of fragments. - * * The 'via' parameter causes this fragment to be registered with a path and * unregistered when done or abandoned. It's only used the first time it's * supplied (the first non-NULL) for a given message ID. This is a mitigation * against memory exhausting DOS attacks. * - * Lastly the message queue size target and GC trigger parameters control - * garbage collection of defragmenter message queue entries. If the size - * target parameter is non-zero then the message queue is cleaned when its - * size reaches the GC trigger parameter, which MUST be larger than the size - * target. Cleaning is done by sorting all entries by their last modified - * timestamp and removing the oldest N entries so as to bring the size down - * to under the size target. The use of a trigger size that is larger than - * the size target reduces CPU-wasting thrashing. A good value for the trigger - * is 2X the size target, causing cleanups to happen only occasionally. - * - * If the GC parameters are set to zero then clear() must be called from time - * to time or memory use will grow without bound. - * * @tparam X Template parameter type for Buf<> containing fragment (inferred) * @param messageId Message ID (a unique ID identifying this message) + * @param message Fixed capacity vector that will be filled with the result if result code is DONE * @param fragment Buffer containing fragment that will be filed under this message's ID * @param fragmentDataIndex Index of data in fragment's data.bytes (fragment's data.fields type is ignored) * @param fragmentDataSize Length of data in fragment's data.bytes (fragment's data.fields type is ignored) * @param fragmentNo Number of fragment (0..totalFragmentsExpected, non-inclusive) - * @param totalFragmentsExpected Total number of expected fragments in this message + * @param totalFragmentsExpected Total number of expected fragments in this message or 0 to use cached value * @param now Current time * @param via If non-NULL this is the path on which this message fragment was received * @param maxIncomingFragmentsPerPath If via is non-NULL this is a cutoff for maximum fragments in flight via this path - * @param messageQueueSizeTarget If non-zero periodically clean the message queue to bring it under this size - * @param messageQueueSizeGCTrigger A value larger than messageQueueSizeTarget that is when cleaning is performed - * @return Result buffer (pointer to 'result' or newly allocated buffer) or NULL if message not complete + * @return Result code */ - ZT_ALWAYS_INLINE Result assemble( + ZT_ALWAYS_INLINE ResultCode assemble( const uint64_t messageId, - SharedPtr< Buf<> > &fragment, + FCV< Buf::Slice,MF > &message, + SharedPtr &fragment, const unsigned int fragmentDataIndex, const unsigned int fragmentDataSize, const unsigned int fragmentNo, const unsigned int totalFragmentsExpected, const int64_t now, const SharedPtr< Path > &via, - const unsigned int maxIncomingFragmentsPerPath, - const unsigned long messageQueueSizeTarget, - const unsigned long messageQueueSizeGCTrigger) + const unsigned int maxIncomingFragmentsPerPath) { // Sanity checks for malformed fragments or invalid input parameters. if ((fragmentNo >= totalFragmentsExpected)||(totalFragmentsExpected > MF)||(totalFragmentsExpected == 0)) - return Result(ERR_INVALID_FRAGMENT); - - // If there is only one fragment just return that fragment and we are done. - if (totalFragmentsExpected < 2) { - if (fragmentNo == 0) { - Result r; - r.message[0].b.move(fragment); - r.message[0].s = fragmentDataIndex; - r.message[0].e = fragmentDataSize; - r.messageFragmentCount = 1; - return r; - } else { - return Result(ERR_INVALID_FRAGMENT); - } - } + return ERR_INVALID_FRAGMENT; // Lock messages for read and look up current entry. Also check the // GC trigger and if we've exceeded that threshold then older message // entries are garbage collected. _messages_l.rlock(); - if (messageQueueSizeTarget > 0) { - if (_messages.size() >= messageQueueSizeGCTrigger) { - try { - // Scan messages with read lock still locked first and make a sorted list of - // message entries by last modified time. Then lock for writing and delete - // the oldest entries to bring the size of the messages hash table down to - // under the target size. This tries to minimize the amount of time the write - // lock is held since many threads can hold the read lock but all threads must - // wait if someone holds the write lock. - std::vector< std::pair > messagesByLastUsedTime; - messagesByLastUsedTime.reserve(_messages.size()); + if (_messages.size() >= GCT) { + try { + // Scan messages with read lock still locked first and make a sorted list of + // message entries by last modified time. Then lock for writing and delete + // the oldest entries to bring the size of the messages hash table down to + // under the target size. This tries to minimize the amount of time the write + // lock is held since many threads can hold the read lock but all threads must + // wait if someone holds the write lock. + std::vector< std::pair > messagesByLastUsedTime; + messagesByLastUsedTime.reserve(_messages.size()); - typename Hashtable::Iterator i(_messages); - uint64_t *mk = nullptr; - _E *mv = nullptr; - while (i.next(mk,mv)) - messagesByLastUsedTime.push_back(std::pair(mv->lastUsed,*mk)); + typename Hashtable::Iterator i(_messages); + uint64_t *mk = nullptr; + _E *mv = nullptr; + while (i.next(mk,mv)) + messagesByLastUsedTime.push_back(std::pair(mv->lastUsed,*mk)); - std::sort(messagesByLastUsedTime.begin(),messagesByLastUsedTime.end()); + std::sort(messagesByLastUsedTime.begin(),messagesByLastUsedTime.end()); - _messages_l.runlock(); - _messages_l.lock(); - for (unsigned long x = 0,y = (messagesByLastUsedTime.size() - messageQueueSizeTarget); x <= y; ++x) - _messages.erase(messagesByLastUsedTime[x].second); - _messages_l.unlock(); - _messages_l.rlock(); - } catch (...) { - // The only way something in that code can throw is if a bad_alloc occurs when - // reserve() is called in the vector. In this case we flush the entire queue - // and error out. This is very rare and on some platforms impossible. - _messages_l.runlock(); - _messages_l.lock(); - _messages.clear(); - _messages_l.unlock(); - return Result(ERR_OUT_OF_MEMORY); - } + _messages_l.runlock(); _messages_l.lock(); + for (unsigned long x = 0,y = (messagesByLastUsedTime.size() - GCS); x <= y; ++x) + _messages.erase(messagesByLastUsedTime[x].second); + _messages_l.unlock(); _messages_l.rlock(); + } catch (...) { + // The only way something in that code can throw is if a bad_alloc occurs when + // reserve() is called in the vector. In this case we flush the entire queue + // and error out. This is very rare and on some platforms impossible. + _messages_l.runlock(); + _messages_l.lock(); + _messages.clear(); + _messages_l.unlock(); + return ERR_OUT_OF_MEMORY; } } _E *e = _messages.get(messageId); @@ -256,7 +195,7 @@ public: RWMutex::Lock ml(_messages_l); e = &(_messages[messageId]); } catch ( ... ) { - return Result(ERR_OUT_OF_MEMORY); + return ERR_OUT_OF_MEMORY; } e->id = messageId; } @@ -268,6 +207,19 @@ public: // is locked or a deadlock could occur due to GC or clear() being called // in another thread. + // This magic value means this message has already been assembled and is done. + if (e->lastUsed < 0) + return ERR_DUPLICATE_FRAGMENT; + + // Update last-activity timestamp for this entry. + e->lastUsed = now; + + // Learn total fragments expected if a value is given. Otherwise the cached + // value gets used. This is to support the implementation of fragmentation + // in the ZT protocol where only fragments carry the total. + if (totalFragmentsExpected > 0) + e->totalFragmentsExpected = totalFragmentsExpected; + // If there is a path associated with this fragment make sure we've registered // ourselves as in flight, check the limit, and abort if exceeded. if ((via)&&(!e->via)) { @@ -286,22 +238,18 @@ public: tooManyPerPath = true; } via->_inboundFragmentedMessages_l.unlock(); - if (tooManyPerPath) { - return Result(ERR_TOO_MANY_FRAGMENTS_FOR_PATH); - } + if (tooManyPerPath) + return ERR_TOO_MANY_FRAGMENTS_FOR_PATH; } - // Update last-activity timestamp for this entry. - e->lastUsed = now; - // If we already have fragment number X, abort. Note that we do not // actually compare data here. Two same-numbered fragments with different // data would just mean the transfer is corrupt and would be detected // later e.g. by packet MAC check. Other use cases of this code like // network configs check each fragment so this basically can't happen. - Buf<>::Slice &s = e->result.message[fragmentNo]; + Buf::Slice &s = e->message.at(fragmentNo); if (s.b) - return Result(ERR_DUPLICATE_FRAGMENT); + return ERR_DUPLICATE_FRAGMENT; // Take ownership of fragment, setting 'fragment' pointer to NULL. The simple // transfer of the pointer avoids a synchronized increment/decrement of the object's @@ -311,7 +259,7 @@ public: s.e = fragmentDataIndex + fragmentDataSize; // If we now have all fragments then assemble them. - if (++e->result.messageFragmentCount >= totalFragmentsExpected) { + if ((e->message.size() >= e->totalFragmentsExpected)&&(e->totalFragmentsExpected > 0)) { // This message is done so de-register it with its path if one is associated. if (e->via) { e->via->_inboundFragmentedMessages_l.lock(); @@ -320,10 +268,15 @@ public: e->via.zero(); } - return e->result; + // Slices are TriviallyCopyable and so may be memcpy'd from e->message to + // the result parameter. This is fast. + e->message.unsafeMoveTo(message); + e->message.lastUsed = -1; // mark as "done" and force GC to collect + + return COMPLETE; } - return Result(ERR_NONE); + return OK; } /** @@ -338,7 +291,7 @@ public: private: struct _E { - ZT_ALWAYS_INLINE _E() : id(0),lastUsed(0),via() {} + ZT_ALWAYS_INLINE _E() : id(0),lastUsed(0),totalFragmentsExpected(0),via(),message(),lock() {} ZT_ALWAYS_INLINE ~_E() { // Ensure that this entry is not in use while it is being deleted! @@ -352,8 +305,9 @@ private: } uint64_t id; volatile int64_t lastUsed; + unsigned int totalFragmentsExpected; SharedPtr via; - Result result; + FCV< Buf::Slice,MF > message; Mutex lock; }; diff --git a/node/FCV.hpp b/node/FCV.hpp new file mode 100644 index 000000000..dadbdbaa7 --- /dev/null +++ b/node/FCV.hpp @@ -0,0 +1,278 @@ +/* + * Copyright (c)2013-2020 ZeroTier, Inc. + * + * Use of this software is governed by the Business Source License included + * in the LICENSE.TXT file in the project's root directory. + * + * Change Date: 2024-01-01 + * + * On the date above, in accordance with the Business Source License, use + * of this software will be governed by version 2.0 of the Apache License. + */ +/****/ + +#ifndef ZT_FCV_HPP +#define ZT_FCV_HPP + +#include "Constants.hpp" + +#include +#include +#include +#include +#include + +namespace ZeroTier { + +/** + * FCV is a Fixed Capacity Vector + * + * Attempts to resize, push, or access this vector beyond its capacity will + * silently fail. The [] operator is NOT bounds checked! + * + * This doesn't implement everything in std::vector, just what we need. It + * also adds a few special things for use in ZT core code. + * + * @tparam T Type to contain + * @tparam C Maximum capacity of vector + */ +template +class FCV +{ +public: + typedef T * iterator; + typedef const T * const_iterator; + + ZT_ALWAYS_INLINE FCV() : _s(0) {} + + template + ZT_ALWAYS_INLINE FCV(const FCV &v) : _s(0) { *this = v; } + + ZT_ALWAYS_INLINE ~FCV() { this->clear(); } + + template + ZT_ALWAYS_INLINE FCV &operator=(const FCV &v) + { + if ((C != C2)||(&v != this)) { + this->clear(); + const unsigned int vs = ((C2 > C) && (v._s > C)) ? C : v._s; + _s = vs; + for (unsigned int i = 0; i < vs; ++i) + new(reinterpret_cast(_m) + i) T(*(reinterpret_cast(v._m) + i)); + } + return *this; + } + + /** + * Clear this vector, destroying all content objects + */ + ZT_ALWAYS_INLINE void clear() + { + const unsigned int s = _s; + _s = 0; + for(unsigned int i=0;i(_m) + i)->~T(); + } + + /** + * Clear without calling destructors (same as unsafeResize(0)) + */ + ZT_ALWAYS_INLINE void unsafeClear() { _s = 0; } + + /** + * This does a straight copy of one vector's data to another + * + * If the other vector is larger than this one's capacity the data is + * silently truncated. This is unsafe in that it does not call any + * constructors or destructors and copies data with memcpy, so it can + * only be used with primitive types or TriviallyCopyable objects. + * + * @tparam C2 Inferred capacity of other vector + * @param v Other vector to copy to this one + */ + template + ZT_ALWAYS_INLINE void unsafeAssign(const FCV &v) + { + _s = ((C2 > C)&&(v._s > C)) ? C : v._s; + memcpy(_m,v._m,_s * sizeof(T)); + } + + /** + * Move contents from this vector to another and clear this vector + * + * This uses a straight memcpy and so is only safe for primitive types or + * types that are TriviallyCopyable. + * + * @param v Target vector + */ + ZT_ALWAYS_INLINE void unsafeMoveTo(FCV &v) + { + memcpy(v._m,_m,(v._s = _s) * sizeof(T)); + _s = 0; + } + + ZT_ALWAYS_INLINE iterator begin() { return reinterpret_cast(_m); } + ZT_ALWAYS_INLINE const_iterator begin() const { return reinterpret_cast(_m); } + ZT_ALWAYS_INLINE iterator end() { return reinterpret_cast(_m) + _s; } + ZT_ALWAYS_INLINE const_iterator end() const { return reinterpret_cast(_m) + _s; } + + ZT_ALWAYS_INLINE T &operator[](const unsigned int i) { return reinterpret_cast(_m)[i]; } + ZT_ALWAYS_INLINE const T &operator[](const unsigned int i) const { return reinterpret_cast(_m)[i]; } + + ZT_ALWAYS_INLINE unsigned int size() const { return _s; } + ZT_ALWAYS_INLINE bool empty() const { return (_s == 0); } + static constexpr unsigned int capacity() { return C; } + + /** + * Push a value onto the back of this vector + * + * If the vector is at capacity this silently fails. + * + * @param v Value to push + */ + ZT_ALWAYS_INLINE void push_back(const T &v) + { + if (_s < C) + new (reinterpret_cast(_m) + _s++) T(v); + } + + /** + * Push a new value onto the vector and return it, or return last item if capacity is reached + * + * @return Reference to new item + */ + ZT_ALWAYS_INLINE T &push() + { + if (_s < C) { + return *(new(reinterpret_cast(_m) + _s++) T()); + } else { + return *(reinterpret_cast(_m) + (C - 1)); + } + } + + /** + * Push a new value onto the vector and return it, or return last item if capacity is reached + * + * @return Reference to new item + */ + ZT_ALWAYS_INLINE T &push(const T &v) + { + if (_s < C) { + return *(new(reinterpret_cast(_m) + _s++) T(v)); + } else { + T &tmp = *(reinterpret_cast(_m) + (C - 1)); + tmp = v; + return tmp; + } + } + + /** + * Remove the last element if this vector is not empty + */ + ZT_ALWAYS_INLINE void pop_back() + { + if (_s != 0) + (reinterpret_cast(_m) + --_s)->~T(); + } + + /** + * Resize vector + * + * @param ns New size (clipped to C if larger than capacity) + */ + ZT_ALWAYS_INLINE void resize(unsigned int ns) + { + if (ns > C) + ns = C; + unsigned int s = _s; + while (s < ns) + new(reinterpret_cast(_m) + s++) T(); + while (s > ns) + (reinterpret_cast(_m) + --s)->~T(); + _s = s; + } + + /** + * Resize without calling any constructors or destructors on T + * + * This must only be called if T is a primitive type or is TriviallyCopyable and + * safe to initialize from undefined contents. + * + * @param ns New size (clipped to C if larger than capacity) + */ + ZT_ALWAYS_INLINE void unsafeResize(const unsigned int ns) { _s = (ns > C) ? C : ns; } + + /** + * This is a bounds checked auto-resizing variant of the [] operator + * + * If 'i' is out of bounds vs the current size of the vector, the vector is + * resized. If that size would exceed C (capacity), 'i' is clipped to C-1. + * + * @param i Index to obtain as a reference, resizing if needed + * @return Reference to value at this index + */ + ZT_ALWAYS_INLINE T &at(unsigned int i) + { + if (i >= _s) { + if (unlikely(i >= C)) + i = C - 1; + do { + new(reinterpret_cast(_m) + _s++) T(); + } while (i >= _s); + } + return *(reinterpret_cast(_m) + i); + } + + /** + * Assign this vector's contents from a range of pointers or iterators + * + * If the range is larger than C it is truncated at C. + * + * @tparam X Inferred type of interators or pointers + * @param start Starting iterator + * @param end Ending iterator (must be greater than start) + */ + template + ZT_ALWAYS_INLINE void assign(X start,const X &end) + { + const int l = std::min((int)std::distance(start,end),(int)C); + if (l > 0) { + this->resize((unsigned int)l); + for(int i=0;i(_m)[i] = *(start++); + } else { + this->clear(); + } + } + + template + ZT_ALWAYS_INLINE bool operator==(const FCV &v) const + { + if (_s == v._s) { + for(unsigned int i=0;i<_s;++i) { + if (!(*(reinterpret_cast(_m) + i) == *(reinterpret_cast(v._m) + i))) + return false; + } + return true; + } + return false; + } + template + ZT_ALWAYS_INLINE bool operator!=(const FCV &v) const { return (!(*this == v)); } + template + ZT_ALWAYS_INLINE bool operator<(const FCV &v) const { return std::lexicographical_compare(begin(),end(),v.begin(),v.end()); } + template + ZT_ALWAYS_INLINE bool operator>(const FCV &v) const { return (v < *this); } + template + ZT_ALWAYS_INLINE bool operator<=(const FCV &v) const { return !(v < *this); } + template + ZT_ALWAYS_INLINE bool operator>=(const FCV &v) const { return !(*this < v); } + +private: + unsigned int _s; + uint8_t _m[sizeof(T) * C]; +}; + +} // namespace ZeroTier + +#endif diff --git a/node/InetAddress.hpp b/node/InetAddress.hpp index 0b2de4d45..d048d6bbf 100644 --- a/node/InetAddress.hpp +++ b/node/InetAddress.hpp @@ -106,7 +106,6 @@ public: memoryCopyUnsafe(this,&ss); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr_storage *ss) { if (ss) @@ -114,13 +113,11 @@ public: else memoryZero(this); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr_in &sa) { copySockaddrToThis(&sa); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr_in *sa) { if (sa) @@ -128,13 +125,11 @@ public: else memset(reinterpret_cast(this),0,sizeof(InetAddress)); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr_in6 &sa) { copySockaddrToThis(&sa); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr_in6 *sa) { if (sa) @@ -142,7 +137,6 @@ public: else memset(reinterpret_cast(this),0,sizeof(InetAddress)); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr &sa) { if (sa.sa_family == AF_INET) @@ -152,7 +146,6 @@ public: else memset(reinterpret_cast(this),0,sizeof(InetAddress)); return *this; } - ZT_ALWAYS_INLINE InetAddress &operator=(const struct sockaddr *sa) { if (sa) { @@ -239,9 +232,9 @@ public: ZT_ALWAYS_INLINE unsigned int port() const { switch(ss_family) { - case AF_INET: return Utils::ntoh((uint16_t)(reinterpret_cast(this)->sin_port)); + case AF_INET: return Utils::ntoh((uint16_t)(reinterpret_cast(this)->sin_port)); case AF_INET6: return Utils::ntoh((uint16_t)(reinterpret_cast(this)->sin6_port)); - default: return 0; + default: return 0; } } diff --git a/node/Node.cpp b/node/Node.cpp index 0d10b40e3..dd8ed9772 100644 --- a/node/Node.cpp +++ b/node/Node.cpp @@ -111,11 +111,16 @@ Node::~Node() for(std::vector< SharedPtr >::iterator i(_networks.begin());i!=_networks.end();++i) i->zero(); } + if (RR->sa) RR->sa->~SelfAwareness(); if (RR->topology) RR->topology->~Topology(); if (RR->sw) RR->sw->~Switch(); if (RR->t) RR->t->~Trace(); free(RR->rtmem); + + // Let go of cached Buf objects. This does no harm if other nodes are running + // but usually that won't be the case. + freeBufPool(); } void Node::shutdown(void *tPtr) diff --git a/node/OS.hpp b/node/OS.hpp index de356b156..49c87aa6f 100644 --- a/node/OS.hpp +++ b/node/OS.hpp @@ -144,6 +144,7 @@ #endif #ifndef __CPP11__ #define nullptr (0) +#define constexpr ZT_ALWAYS_INLINE #endif #ifdef SOCKET diff --git a/node/Path.hpp b/node/Path.hpp index ffc0f6769..18b97b709 100644 --- a/node/Path.hpp +++ b/node/Path.hpp @@ -32,7 +32,7 @@ namespace ZeroTier { class RuntimeEnvironment; -template +template class Defragmenter; /** @@ -43,7 +43,7 @@ class Path friend class SharedPtr; // Allow defragmenter to access fragment in flight info stored in Path for performance reasons. - template + template friend class Defragmenter; public: diff --git a/node/Protocol.cpp b/node/Protocol.cpp index 462af7d19..7236eff10 100644 --- a/node/Protocol.cpp +++ b/node/Protocol.cpp @@ -15,6 +15,9 @@ #include "Buf.hpp" #include "Utils.hpp" +#include +#include + #if defined(__GCC__) && (defined(__amd64) || defined(__amd64__) || defined(__x86_64) || defined(__x86_64__) || defined(__AMD64) || defined(__AMD64__) || defined(_M_X64)) #define ZT_PACKET_USE_ATOMIC_INTRINSICS #endif @@ -27,41 +30,6 @@ namespace Protocol { namespace { -const uint64_t ZEROES32[4] = { 0,0,0,0 }; - -/** - * Deterministically mangle a 256-bit crypto key based on packet - * - * This uses extra data from the packet to mangle the secret, giving us an - * effective IV that is somewhat more than 64 bits. This is "free" for - * Salsa20 since it has negligible key setup time so using a different - * key each time is fine. - * - * @param in Input key (32 bytes) - * @param out Output buffer (32 bytes) - */ -ZT_ALWAYS_INLINE void _salsa20MangleKey(const uint8_t *const in,uint8_t *const out,const Buf< Header > &packet,const unsigned int packetSize) -{ - // IV and source/destination addresses. Using the addresses divides the - // key space into two halves-- A->B and B->A (since order will change). - for(int i=0;i<18;++i) // 8 + (ZT_ADDRESS_LENGTH * 2) == 18 - out[i] = in[i] ^ packet.data.bytes[i]; - - // Flags, but with hop count masked off. Hop count is altered by forwarding - // nodes. It's one of the only parts of a packet modifiable by people - // without the key. - out[18] = in[18] ^ (packet.data.fields.flags & 0xf8U); - - // Raw packet size in bytes -- thus each packet size defines a new - // key space. - out[19] = in[19] ^ (uint8_t)packetSize; - out[20] = in[20] ^ (uint8_t)(packetSize >> 8U); // little endian - - // Rest of raw key is used unchanged - for(int i=21;i<32;++i) - out[i] = in[i]; -} - unsigned long long _initPacketID() { unsigned long long tmp = 0; @@ -76,8 +44,29 @@ unsigned long long _packetIdCtr = _initPacketID(); static std::atomic _packetIdCtr(_initPacketID()); #endif +uintptr_t _checkStructureSizing() +{ + if (sizeof(Header) != ZT_PROTO_MIN_PACKET_LENGTH) + throw std::runtime_error("sizeof(Header) != ZT_PROTO_MIN_PACKET_LENGTH"); + if (sizeof(FragmentHeader) != ZT_PROTO_MIN_FRAGMENT_LENGTH) + throw std::runtime_error("sizeof(FragmentHeader) != ZT_PROTO_MIN_FRAGMENT_LENGTH"); + return (uintptr_t)Utils::getSecureRandomU64(); // also prevents compiler from optimizing out +} + } // anonymous namespace +volatile uintptr_t _compileTimeStructCheckHappened = _checkStructureSizing(); + +uint64_t getPacketId() +{ +#ifdef ZT_PACKET_USE_ATOMIC_INTRINSICS + return __sync_add_and_fetch(&_packetIdCtr,1ULL); +#else + return ++_packetIdCtr; +#endif +} + +#if 0 void _armor(Buf< Header > &packet,const unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH],const uint8_t cipherSuite) { packet.data.fields.flags = (packet.data.fields.flags & 0xc7U) | ((cipherSuite << 3U) & 0x38U); // FFCCCHHH @@ -102,32 +91,6 @@ void _armor(Buf< Header > &packet,const unsigned int packetSize,const uint8_t ke } } -int _dearmor(Buf< Header > &packet,const unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH]) -{ - const int cipherSuite = (int)(packet.data.fields.flags & 0x38U); - if (cipherSuite == ZT_PROTO_CIPHER_SUITE__AES_GCM) { - // TODO - } else if (cipherSuite != ZT_PROTO_CIPHER_SUITE__NONE) { - uint8_t mangledKey[ZT_PEER_SECRET_KEY_LENGTH],macKey[ZT_POLY1305_KEY_LEN]; - uint64_t mac[2]; - - _salsa20MangleKey(key,mangledKey,packet,packetSize); - Salsa20 s20(mangledKey,&(packet.data.fields.packetId)); - s20.crypt12(ZEROES32,macKey,sizeof(macKey)); - - uint8_t *payload = packet.data.bytes + ZT_PROTO_PACKET_ENCRYPTED_SECTION_START; - const unsigned int payloadLen = packetSize - ZT_PROTO_PACKET_ENCRYPTED_SECTION_START; - - if (cipherSuite == ZT_PROTO_CIPHER_SUITE__POLY1305_SALSA2012) - s20.crypt12(payload,payload,payloadLen); - - poly1305(mac,payload,payloadLen,macKey); - if (packet.data.fields.mac != mac[0]) - return -1; - } - return cipherSuite; -} - unsigned int _compress(Buf< Header > &packet,const unsigned int packetSize) { uint8_t tmp[ZT_BUF_MEM_SIZE + 32]; @@ -149,36 +112,7 @@ unsigned int _compress(Buf< Header > &packet,const unsigned int packetSize) return packetSize; } - -int _uncompress(Buf< Header > &packet,const unsigned int packetSize) -{ - uint8_t tmp[ZT_BUF_MEM_SIZE]; - - if ((packet.data.fields.verb & ZT_PROTO_VERB_FLAG_COMPRESSED) == 0) - return (int)packetSize; - - const int uncompressedLen = LZ4_decompress_safe( - reinterpret_cast(packet.data.bytes + ZT_PROTO_PACKET_PAYLOAD_START), - reinterpret_cast(tmp), - (int)(packetSize - ZT_PROTO_PACKET_PAYLOAD_START), - sizeof(tmp) - ZT_PROTO_PACKET_PAYLOAD_START); - - if ((uncompressedLen > 0)&&(uncompressedLen <= (sizeof(tmp) - ZT_PROTO_PACKET_PAYLOAD_START))) { - packet.data.fields.verb &= (uint8_t)(~((uint8_t)ZT_PROTO_VERB_FLAG_COMPRESSED)); - memcpy(packet.data.bytes + ZT_PROTO_PACKET_PAYLOAD_START,tmp,uncompressedLen); - return uncompressedLen + ZT_PROTO_PACKET_PAYLOAD_START; - } - return -1; -} - -uint64_t getPacketId() -{ -#ifdef ZT_PACKET_USE_ATOMIC_INTRINSICS - return __sync_add_and_fetch(&_packetIdCtr,1ULL); -#else - return ++_packetIdCtr; #endif -} } // namespace Protocol } // namespace ZeroTier diff --git a/node/Protocol.hpp b/node/Protocol.hpp index 32c3f5859..5f4c56acf 100644 --- a/node/Protocol.hpp +++ b/node/Protocol.hpp @@ -119,7 +119,12 @@ /** * Magic number indicating a fragment */ -#define ZT_PACKET_FRAGMENT_INDICATOR 0xff +#define ZT_PROTO_PACKET_FRAGMENT_INDICATOR 0xff + +/** + * Index at which fragment indicator is found in fragments + */ +#define ZT_PROTO_PACKET_FRAGMENT_INDICATOR_INDEX 13 /** * Minimum viable length for a fragment @@ -129,17 +134,17 @@ /** * Index at which packet fragment payload starts */ -#define ZT_PROTO_PACKET_FRAGMENT_PAYLOAD_START_AT 16 +#define ZT_PROTO_PACKET_FRAGMENT_PAYLOAD_START_AT ZT_PROTO_MIN_FRAGMENT_LENGTH /** * Header flag indicating that a packet is fragmented and more fragments should be expected */ -#define ZT_PROTO_FLAG_FRAGMENTED 0x40 +#define ZT_PROTO_FLAG_FRAGMENTED 0x40U /** * Verb flag indicating payload is compressed with LZ4 */ -#define ZT_PROTO_VERB_FLAG_COMPRESSED 0x80 +#define ZT_PROTO_VERB_FLAG_COMPRESSED 0x80U /** * HELLO exchange meta-data: signed locator for this node @@ -203,10 +208,6 @@ * For unencrypted packets, MAC is computed on plaintext. Only HELLO is ever * sent in the clear, as it's the "here is my public key" message. * - * Fragments are sent if a packet is larger than UDP MTU. The first fragment - * is sent with its normal header with the fragmented flag set. Remaining - * fragments are sent this way. - * * The fragmented bit indicates that there is at least one fragment. Fragments * themselves contain the total, so the receiver must "learn" this from the * first fragment it receives. @@ -755,7 +756,6 @@ ZT_PACKED_STRUCT(struct FragmentHeader uint8_t fragmentIndicator; // always 0xff for fragments uint8_t counts; // total: most significant four bits, number: least significant four bits uint8_t hops; // top 5 bits unused and must be zero - uint8_t p[]; }); ZT_PACKED_STRUCT(struct HELLO @@ -766,7 +766,6 @@ ZT_PACKED_STRUCT(struct HELLO uint8_t versionMinor; uint16_t versionRev; uint64_t timestamp; - uint8_t p[]; }); ZT_PACKED_STRUCT(struct RENDEZVOUS @@ -776,7 +775,6 @@ ZT_PACKED_STRUCT(struct RENDEZVOUS uint8_t peerAddress[5]; uint16_t port; uint8_t addressLength; - uint8_t address[]; }); ZT_PACKED_STRUCT(struct FRAME @@ -784,7 +782,6 @@ ZT_PACKED_STRUCT(struct FRAME Header h; uint64_t networkId; uint16_t etherType; - uint8_t data[]; }); ZT_PACKED_STRUCT(struct EXT_FRAME @@ -792,7 +789,6 @@ ZT_PACKED_STRUCT(struct EXT_FRAME Header h; uint64_t networkId; uint8_t flags; - uint8_t p[]; }); ZT_PACKED_STRUCT(struct MULTICAST_LIKE @@ -805,7 +801,6 @@ ZT_PACKED_STRUCT(struct MULTICAST_LIKE }); Header h; - Entry groups[]; }); namespace OK { @@ -922,59 +917,48 @@ ZT_ALWAYS_INLINE uint8_t packetHops(const Header &h) { return (h.flags & 0x07U); */ ZT_ALWAYS_INLINE uint8_t packetCipher(const Header &h) { return ((h.flags >> 3U) & 0x07U); } -void _armor(Buf< Header > &packet,unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH],uint8_t cipherSuite); -int _dearmor(Buf< Header > &packet,unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH]); -unsigned int _compress(Buf< Header > &packet,unsigned int packetSize); -int _uncompress(Buf< Header > &packet,unsigned int packetSize); - /** - * Armor a packet for transport + * Deterministically mangle a 256-bit crypto key based on packet characteristics * - * @param packet Packet to armor - * @param packetSize Size of data in packet (must be at least the minimum packet size) - * @param key 256-bit symmetric key - * @param cipherSuite Cipher suite to apply + * This uses extra data from the packet to mangle the secret, yielding when + * combined with Salsa20's conventional 64-bit nonce an effective nonce that's + * more like 68 bits. + * + * @param in Input key (32 bytes) + * @param out Output buffer (32 bytes) */ -template -static ZT_ALWAYS_INLINE void armor(Buf< X > &packet,unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH],uint8_t cipherSuite) -{ _armor(reinterpret_cast< Buf< Header > & >(packet),packetSize,key,cipherSuite); } +ZT_ALWAYS_INLINE void salsa2012DeriveKey(const uint8_t *const in,uint8_t *const out,const Buf &packet,const unsigned int packetSize) +{ + // IV and source/destination addresses. Using the addresses divides the + // key space into two halves-- A->B and B->A (since order will change). +#ifdef ZT_NO_UNALIGNED_ACCESS + for(int i=0;i<18;++i) + out[i] = in[i] ^ packet.b[i]; +#else + *reinterpret_cast(out) = *reinterpret_cast(in) ^ *reinterpret_cast(packet.b); + *reinterpret_cast(out + 8) = *reinterpret_cast(in + 8) ^ *reinterpret_cast(packet.b + 8); + *reinterpret_cast(out + 16) = *reinterpret_cast(in + 16) ^ *reinterpret_cast(packet.b + 16); +#endif -/** - * Dearmor a packet and check message authentication code - * - * If the packet is valid and MAC (if indicated) passes, the cipher suite - * is returned. Otherwise -1 is returned to indicate a MAC failure. - * - * @param packet Packet to dearmor - * @param packetSize Size of data in packet (must be at least the minimum packet size) - * @param key 256-bit symmetric key - * @return Cipher suite or -1 if MAC validation failed - */ -template -static ZT_ALWAYS_INLINE int dearmor(Buf< X > &packet,unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH]) -{ return _dearmor(reinterpret_cast< Buf < Header > & >(packet),packetSize,key); } + // Flags, but with hop count masked off. Hop count is altered by forwarding + // nodes and is the only field that is mutable by unauthenticated third parties. + out[18] = in[18] ^ (packet.b[18] & 0xf8U); -/** - * Compress packet payload - * - * @param packet Packet to compress - * @param packetSize Original packet size - * @return New packet size (returns original size of compression didn't help, in which case packet is unmodified) - */ -template -static ZT_ALWAYS_INLINE unsigned int compress(Buf< X > &packet,unsigned int packetSize) -{ return _compress(reinterpret_cast< Buf< Header > & >(packet),packetSize); } + // Raw packet size in bytes -- thus each packet size defines a new key space. + out[19] = in[19] ^ (uint8_t)packetSize; + out[20] = in[20] ^ (uint8_t)(packetSize >> 8U); // little endian -/** - * Uncompress packet payload (if compressed) - * - * @param packet Packet to uncompress - * @param packetSize Original packet size - * @return New packet size or -1 on decompression error (returns original packet size if packet wasn't compressed) - */ -template -static ZT_ALWAYS_INLINE int uncompress(Buf< X > &packet,unsigned int packetSize) -{ return _uncompress(reinterpret_cast< Buf< Header > & >(packet),packetSize); } + // Rest of raw key is used unchanged +#ifdef ZT_NO_UNALIGNED_ACCESS + for(int i=21;i<32;++i) + out[i] = in[i]; +#else + out[21] = in[21]; + out[22] = in[22]; + out[23] = in[23]; + *reinterpret_cast(out + 24) = *reinterpret_cast(in + 24); +#endif +} /** * Get a sequential non-repeating packet ID for the next packet (thread-safe) @@ -983,6 +967,12 @@ static ZT_ALWAYS_INLINE int uncompress(Buf< X > &packet,unsigned int packetSize) */ uint64_t getPacketId(); +void armor(Buf &packet,unsigned int packetSize,const uint8_t key[ZT_PEER_SECRET_KEY_LENGTH],uint8_t cipherSuite); + +unsigned int compress(Buf &packet,unsigned int packetSize); + +int uncompress(Buf &packet,unsigned int packetSize); + } // namespace Protocol } // namespace ZeroTier diff --git a/node/Salsa20.hpp b/node/Salsa20.hpp index 4294a8d6c..6a20d0d84 100644 --- a/node/Salsa20.hpp +++ b/node/Salsa20.hpp @@ -25,6 +25,7 @@ #if (defined(__amd64) || defined(__amd64__) || defined(__x86_64) || defined(__x86_64__) || defined(__AMD64) || defined(__AMD64__) || defined(_M_X64)) #include #include +#include #define ZT_SALSA20_SSE 1 #endif diff --git a/node/VL1.cpp b/node/VL1.cpp new file mode 100644 index 000000000..3db88f173 --- /dev/null +++ b/node/VL1.cpp @@ -0,0 +1,546 @@ +/* + * Copyright (c)2013-2020 ZeroTier, Inc. + * + * Use of this software is governed by the Business Source License included + * in the LICENSE.TXT file in the project's root directory. + * + * Change Date: 2024-01-01 + * + * On the date above, in accordance with the Business Source License, use + * of this software will be governed by version 2.0 of the Apache License. + */ +/****/ + +#include "VL1.hpp" +#include "RuntimeEnvironment.hpp" +#include "Topology.hpp" +#include "VL2.hpp" +#include "Salsa20.hpp" +#include "LZ4.hpp" +#include "Poly1305.hpp" + +namespace ZeroTier { + +namespace { + +#if 0 +ZT_ALWAYS_INLINE bool _doHELLO(SharedPtr &pkt,int len,const RuntimeEnvironment *const RR,void *const tPtr,const bool alreadyAuthenticated) +{ + if (p.size < sizeof(Protocol::HELLO)) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,Identity(),p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + return true; + } + Buf< Protocol::HELLO > &pkt = reinterpret_cast &>(*p.pkt); + + Identity id; + int ptr = sizeof(Protocol::HELLO); + if (pkt.rO(ptr,id) < 0) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,Identity(),p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_INVALID_OBJECT); + return true; + } + + if (pkt.data.fields.versionProtocol < ZT_PROTO_VERSION_MIN) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_PEER_TOO_OLD); + return true; + } + if (Address(pkt.data.fields.h.source) != id.address()) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + return true; + } + + const int64_t now = RR->node->now(); + + SharedPtr peer(RR->topology->get(tPtr,id.address())); + if (peer) { + // We already have an identity with this address -- check for collisions + if (!alreadyAuthenticated) { + if (peer->identity() != id) { + // Identity is different from the one we already have -- address collision + + // Check rate limits + if (!RR->node->rateGateIdentityVerification(now,p.path->address())) + return true; + + uint8_t key[ZT_PEER_SECRET_KEY_LENGTH]; + if (RR->identity.agree(id,key)) { + if (Protocol::dearmor(pkt,p.size,key) < 0) { // ensure packet is authentic, otherwise drop + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return true; + } else { + // TODO: we handle identity collisions differently now + } + } else { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return true; + } + + return true; + } else { + // Identity is the same as the one we already have -- check packet integrity + + if (Protocol::dearmor(pkt,p.size,peer->key()) < 0) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return true; + } + + // Continue at // VALID + } + } // else if alreadyAuthenticated then continue at // VALID + } else { + // We don't already have an identity with this address -- validate and learn it + + // Sanity check: this basically can't happen + if (alreadyAuthenticated) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,Identity(),p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_UNSPECIFIED); + return true; + } + + // Check rate limits + if (!RR->node->rateGateIdentityVerification(now,p.path->address())) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_RATE_LIMIT_EXCEEDED); + return true; + } + + // Check packet integrity and MAC (this is faster than locallyValidate() so do it first to filter out total crap) + SharedPtr newPeer(new Peer(RR)); + if (!newPeer->init(RR->identity,id)) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return true; + } + if (Protocol::dearmor(pkt,p.size,newPeer->key())) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return true; + } + + // Check that identity's address is valid as per the derivation function + if (!id.locallyValidate()) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_INVALID_OBJECT); + return true; + } + + peer = RR->topology->add(tPtr,newPeer); + + // Continue at // VALID + } + + // VALID -- if we made it here, packet passed identity and authenticity checks! + + // Get address to which this packet was sent to learn our external surface address if packet was direct. + InetAddress externalSurfaceAddress; + if (ptr < p.size) { + if (pkt.rO(ptr,externalSurfaceAddress) < 0) { + RR->t->incomingPacketDropped(tPtr,p.idBE,0,id,p.path->address(),p.hops,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_INVALID_OBJECT); + return true; + } + if ((p.hops == 0)&&(externalSurfaceAddress)) + RR->sa->iam(tPtr,id,p.path->localSocket(),p.path->address(),externalSurfaceAddress,RR->topology->isRoot(id),now); + } + + // Send OK(HELLO) with an echo of the packet's timestamp and some of the same + // information about us: version, sent-to address, etc. + + ZT_GET_NEW_BUF(outp,Protocol::OK::HELLO); + + outp->data.fields.h.packetId = Protocol::getPacketId(); + peer->address().copyTo(outp->data.fields.h.destination); + RR->identity.address().copyTo(outp->data.fields.h.source); + outp->data.fields.h.flags = 0; + outp->data.fields.h.verb = Protocol::VERB_OK; + + outp->data.fields.oh.inReVerb = Protocol::VERB_HELLO; + outp->data.fields.oh.inRePacketId = p.idBE; + + outp->data.fields.timestampEcho = pkt.data.fields.timestamp; + outp->data.fields.versionProtocol = ZT_PROTO_VERSION; + outp->data.fields.versionMajor = ZEROTIER_ONE_VERSION_MAJOR; + outp->data.fields.versionMinor = ZEROTIER_ONE_VERSION_MINOR; + outp->data.fields.versionRev = CONST_TO_BE_UINT16(ZEROTIER_ONE_VERSION_REVISION); + + int outl = sizeof(Protocol::OK::HELLO); + outp->wO(outl,p.path->address()); + if (!Buf<>::writeOverflow(outl)) { + Protocol::armor(*outp,outl,peer->key(),ZT_PROTO_CIPHER_SUITE__POLY1305_SALSA2012); + p.path->send(RR,tPtr,outp->data.bytes,outl,RR->node->now()); + } + + peer->setRemoteVersion(pkt.data.fields.versionProtocol,pkt.data.fields.versionMajor,pkt.data.fields.versionMinor,Utils::ntoh(pkt.data.fields.versionRev)); + peer->received(tPtr,p.path,p.hops,p.idBE,p.size,Protocol::VERB_HELLO,0,Protocol::VERB_NOP,0); + + return true; +} +#endif + +} // anonymous namespace + +VL1::VL1(const RuntimeEnvironment *renv) : + RR(renv), + _vl2(nullptr) +{ +} + +VL1::~VL1() +{ +} + +void VL1::onRemotePacket(void *const tPtr,const int64_t localSocket,const InetAddress &fromAddr,SharedPtr &data,const unsigned int len) +{ + static const uint64_t ZEROES32[4] = { 0,0,0,0 }; + const int64_t now = RR->node->now(); + const SharedPtr path(RR->topology->getPath(localSocket,fromAddr)); + path->received(now); + + if (len < ZT_PROTO_MIN_FRAGMENT_LENGTH) + return; + + try { + FCV pktv; + Address destination; + + if (data->b[ZT_PROTO_PACKET_FRAGMENT_INDICATOR_INDEX] == ZT_PROTO_PACKET_FRAGMENT_INDICATOR) { + // Fragment ----------------------------------------------------------------------------------------------------- + + const Protocol::FragmentHeader &fh = data->as(); + destination.setTo(fh.destination); + + if (destination != RR->identity.address()) { + // Fragment is not address to this node ----------------------------------------------------------------------- + _relay(tPtr,path,destination,data,len); + return; + } + + switch (_inputPacketAssembler.assemble( + fh.packetId, + pktv, + data, + ZT_PROTO_PACKET_FRAGMENT_PAYLOAD_START_AT, + (unsigned int)(len - ZT_PROTO_PACKET_FRAGMENT_PAYLOAD_START_AT), + fh.counts & 0xfU, // fragment number + fh.counts >> 4U, // total number of fragments in message is specified in each fragment + now, + path, + ZT_MAX_INCOMING_FRAGMENTS_PER_PATH)) { + case Defragmenter::COMPLETE: + break; + default: + //case Defragmenter::OK: + //case Defragmenter::ERR_DUPLICATE_FRAGMENT: + //case Defragmenter::ERR_INVALID_FRAGMENT: + //case Defragmenter::ERR_TOO_MANY_FRAGMENTS_FOR_PATH: + //case Defragmenter::ERR_OUT_OF_MEMORY: + return; + } + } else { + // Not fragment, meaning whole packet or head of series with fragments ------------------------------------------ + + if (len < ZT_PROTO_MIN_PACKET_LENGTH) + return; + const Protocol::Header &ph = data->as(); + destination.setTo(ph.destination); + + if (destination != RR->identity.address()) { + // Packet or packet head is not address to this node ---------------------------------------------------------- + _relay(tPtr,path,destination,data,len); + return; + } + + if ((ph.flags & ZT_PROTO_FLAG_FRAGMENTED) != 0) { + // Head of fragmented packet ---------------------------------------------------------------------------------- + switch (_inputPacketAssembler.assemble( + ph.packetId, + pktv, + data, + 0, + len, + 0, // always the zero'eth fragment + 0, // this is specified in fragments, not in the head + now, + path, + ZT_MAX_INCOMING_FRAGMENTS_PER_PATH)) { + case Defragmenter::COMPLETE: + break; + default: + //case Defragmenter::OK: + //case Defragmenter::ERR_DUPLICATE_FRAGMENT: + //case Defragmenter::ERR_INVALID_FRAGMENT: + //case Defragmenter::ERR_TOO_MANY_FRAGMENTS_FOR_PATH: + //case Defragmenter::ERR_OUT_OF_MEMORY: + return; + } + } else { + // Unfragmented packet, skip defrag engine and just handle it ------------------------------------------------- + Buf::Slice &s = pktv.push(); + s.b = data; + s.s = 0; + s.e = len; + } + } + + // Packet defragmented and apparently addressed to this node ------------------------------------------------------ + + // Subject pktv to a few sanity checks just to make sure Defragmenter worked correctly and + // there is enough room in each slice to shift their contents to sizes that are multiples + // of 64 if needed for crypto. + if ((pktv.empty()) || (((int)pktv[0].e - (int)pktv[0].s) < sizeof(Protocol::Header))) + return; + for(FCV::const_iterator s(pktv.begin());s!=pktv.end();++s) { + if ((s->e > (ZT_BUF_MEM_SIZE - 64))||(s->s > s->e)) + return; + } + + Protocol::Header *ph = &(pktv[0].b->as(pktv[0].s)); + const Address source(ph->source); + + if (source == RR->identity.address()) + return; + SharedPtr peer(RR->topology->get(tPtr,source)); + + Buf::Slice pkt; + bool authenticated = false; + + const uint8_t hops = Protocol::packetHops(*ph); + const uint8_t cipher = Protocol::packetCipher(*ph); + + unsigned int packetSize = pktv[0].e - pktv[0].s; + for(FCV::const_iterator s(pktv.begin()+1);s!=pktv.end();++s) + packetSize += s->e - s->s; + if (packetSize > ZT_PROTO_MAX_PACKET_LENGTH) { + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,Identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + return; + } + + // If we don't know this peer and this is not a HELLO, issue a WHOIS and enqueue this packet to try again. + if ((!peer)&&(!(((cipher == ZT_PROTO_CIPHER_SUITE__POLY1305_NONE)||(cipher == ZT_PROTO_CIPHER_SUITE__NONE))&&((ph->verb & 0x1fU) == Protocol::VERB_HELLO)))) { + pkt = Buf::assembleSliceVector(pktv); + if (pkt.e < ZT_PROTO_MIN_PACKET_LENGTH) { + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,Identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + return; + } + { + Mutex::Lock wl(_whoisQueue_l); + _WhoisQueueItem &wq = _whoisQueue[source]; + wq.inboundPackets.push_back(pkt); + if (wq.retries == 0) { + wq.retries = 1; + _sendPendingWhois(); + } + } + return; + } + + switch(cipher) { + case ZT_PROTO_CIPHER_SUITE__POLY1305_NONE: + if (peer) { + pkt = Buf::assembleSliceVector(pktv); + if (pkt.e < ZT_PROTO_MIN_PACKET_LENGTH) + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,Identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + ph = &(pkt.b->as()); + + // Generate one-time-use MAC key using Salsa20. + uint8_t perPacketKey[ZT_PEER_SECRET_KEY_LENGTH]; + uint8_t macKey[ZT_POLY1305_KEY_LEN]; + Protocol::salsa2012DeriveKey(peer->key(),perPacketKey,*pktv[0].b,packetSize); + Salsa20(perPacketKey,&ph->packetId).crypt12(ZEROES32,macKey,ZT_POLY1305_KEY_LEN); + + // Verify packet MAC. + uint64_t mac[2]; + poly1305(mac,pkt.b->b,pkt.e,macKey); + if (ph->mac != mac[0]) { + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,peer->identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return; + } + authenticated = true; + } + break; + + case ZT_PROTO_CIPHER_SUITE__POLY1305_SALSA2012: + if (peer) { + // Derive per-packet key using symmetric key plus some data from the packet header. + uint8_t perPacketKey[ZT_PEER_SECRET_KEY_LENGTH]; + Protocol::salsa2012DeriveKey(peer->key(),perPacketKey,*pktv[0].b,packetSize); + Salsa20 s20(perPacketKey,&ph->packetId); + + // Do one Salsa20 block to generate the one-time-use Poly1305 key. + uint8_t macKey[ZT_POLY1305_KEY_LEN]; + s20.crypt12(ZEROES32,macKey,ZT_POLY1305_KEY_LEN); + + // Get a buffer to store the decrypted and fully contiguous packet. + pkt.b = Buf::get(); + if (!pkt.b) // only possible on out of memory condition + return; + + // Salsa20 is a stream cipher but it's only seekable to multiples of 64 bytes. + // This moves data in slices around so that all slices have sizes that are + // multiples of 64 except the last slice. Note that this does not corrupt + // the assembled slice vector, just moves data around. + if (pktv.size() > 1) { + unsigned int prevOverflow,i; + for (typename FCV::iterator ps(pktv.begin()),s(ps + 1);s!=pktv.end();) { + prevOverflow = (ps->e - ps->s) & 63U; // amount by which previous exceeds a multiple of 64 + for(i=0;is >= s->e) + goto next_slice; + ps->b->b[ps->e++] = s->b->b[s->s++]; // move from head of current to end of previous + } + next_slice: ps = s++; + } + } + + // Simultaneously decrypt and assemble packet into a contiguous buffer. + memcpy(pkt.b->b,ph,sizeof(Protocol::Header)); + pkt.e = sizeof(Protocol::Header); + for(FCV::iterator s(pktv.begin());s!=pktv.end();++s) { + const unsigned int sliceSize = s->e - s->s; + s20.crypt12(s->b->b + s->s,pkt.b->b + pkt.e,sliceSize); + pkt.e += sliceSize; + } + ph = &(pkt.b->as()); + + // Verify packet MAC. + uint64_t mac[2]; + poly1305(mac,pkt.b->b,pkt.e,macKey); + if (ph->mac != mac[0]) { + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,peer->identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return; + } + authenticated = true; + } else { + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,Identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return; + } + break; + + case ZT_PROTO_CIPHER_SUITE__NONE: { + // CIPHER_SUITE__NONE is only used with trusted paths. Verification is performed by + // checking the address and the presence of its corresponding trusted path ID in the + // packet header's MAC field. + + pkt = Buf::assembleSliceVector(pktv); + if (pkt.e < ZT_PROTO_MIN_PACKET_LENGTH) + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,Identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + ph = &(pkt.b->as()); + + if (RR->topology->shouldInboundPathBeTrusted(path->address(),Utils::ntoh(ph->mac))) { + authenticated = true; + } else { + if (peer) + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,peer->identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_MAC_FAILED); + return; + } + } break; + + //case ZT_PROTO_CIPHER_SUITE__AES_GCM_NRH: + // if (peer) { + // } + // break; + + default: + if (peer) + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,peer->identity(),path->address(),hops,Protocol::VERB_NOP,ZT_TRACE_PACKET_DROP_REASON_INVALID_OBJECT); + return; + } + + // Packet fully assembled and may be authenticated ---------------------------------------------------------------- + + // Return any still held buffers in pktv to the buffer pool. + pktv.clear(); + + // Decompress packet payload if compressed. + if (((ph->verb & ZT_PROTO_VERB_FLAG_COMPRESSED) != 0)&&(authenticated)) { + SharedPtr nb(Buf::get()); + if (!nb) // can only happen if we're out of memory + return; + + const int uncompressedLen = LZ4_decompress_safe( + reinterpret_cast(pkt.b->b + ZT_PROTO_PACKET_PAYLOAD_START), + reinterpret_cast(nb->b), + (int)(packetSize - ZT_PROTO_PACKET_PAYLOAD_START), + ZT_BUF_MEM_SIZE - ZT_PROTO_PACKET_PAYLOAD_START); + + if ((uncompressedLen > 0)&&(uncompressedLen <= (ZT_BUF_MEM_SIZE - ZT_PROTO_PACKET_PAYLOAD_START))) { + pkt.b.swap(nb); + pkt.e = packetSize = (unsigned int)uncompressedLen; + } else { + if (peer) + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,peer->identity(),path->address(),hops,(Protocol::Verb)(ph->verb & 0x1fU),ZT_TRACE_PACKET_DROP_REASON_INVALID_COMPRESSED_DATA); + return; + } + } + + const Protocol::Verb verb = (Protocol::Verb)(ph->verb & 0x1fU); + switch(verb) { + case Protocol::VERB_NOP: + peer->received(tPtr,path,hops,ph->packetId,packetSize - ZT_PROTO_PACKET_PAYLOAD_START,Protocol::VERB_NOP,0,Protocol::VERB_NOP,0); + break; + + case Protocol::VERB_HELLO: _HELLO(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_ERROR: _ERROR(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_OK: _OK(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_WHOIS: _WHOIS(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_RENDEZVOUS: _RENDEZVOUS(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_FRAME: _vl2->_FRAME(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_EXT_FRAME: _vl2->_EXT_FRAME(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_ECHO: _ECHO(tPtr,path,peer,*pkt.b,packetSize,authenticated); + case Protocol::VERB_MULTICAST_LIKE: _vl2->_MULTICAST_LIKE(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_NETWORK_CREDENTIALS: _vl2->_NETWORK_CREDENTIALS(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_NETWORK_CONFIG_REQUEST: _vl2->_NETWORK_CONFIG_REQUEST(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_NETWORK_CONFIG: _vl2->_NETWORK_CONFIG(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_MULTICAST_GATHER: _vl2->_MULTICAST_LIKE(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_MULTICAST_FRAME_deprecated: _vl2->_MULTICAST_FRAME_deprecated(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_PUSH_DIRECT_PATHS: _PUSH_DIRECT_PATHS(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_USER_MESSAGE: _USER_MESSAGE(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_MULTICAST: _vl2->_MULTICAST(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + case Protocol::VERB_ENCAP: _ENCAP(tPtr,path,peer,*pkt.b,packetSize,authenticated); break; + + default: + if (peer) + RR->t->incomingPacketDropped(tPtr,ph->packetId,0,peer->identity(),path->address(),hops,verb,ZT_TRACE_PACKET_DROP_REASON_UNRECOGNIZED_VERB); + break; + } + } catch ( ... ) { + uint64_t packetId = 0; + if (len >= 8) { + for(int i=0;i<8;++i) + reinterpret_cast(&packetId)[i] = data->b[i]; + } + RR->t->incomingPacketDropped(tPtr,packetId,0,Identity(),path->address(),0,Protocol::VERB_HELLO,ZT_TRACE_PACKET_DROP_REASON_MALFORMED_PACKET); + } +} + +void VL1::_relay(void *tPtr,const SharedPtr &path,const Address &destination,SharedPtr &data,unsigned int len) +{ +} + +void VL1::_sendPendingWhois() +{ + // assume _whoisQueue_l locked +} + +void VL1::_HELLO(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +void VL1::_ERROR(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +void VL1::_OK(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +void VL1::_WHOIS(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +void VL1::_PUSH_DIRECT_PATHS(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +void VL1::_USER_MESSAGE(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +void VL1::_ENCAP(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated) +{ +} + +} // namespace ZeroTier diff --git a/node/VL1.hpp b/node/VL1.hpp new file mode 100644 index 000000000..b0e0e0fb6 --- /dev/null +++ b/node/VL1.hpp @@ -0,0 +1,88 @@ +/* + * Copyright (c)2013-2020 ZeroTier, Inc. + * + * Use of this software is governed by the Business Source License included + * in the LICENSE.TXT file in the project's root directory. + * + * Change Date: 2024-01-01 + * + * On the date above, in accordance with the Business Source License, use + * of this software will be governed by version 2.0 of the Apache License. + */ +/****/ + +#ifndef ZT_VL1_HPP +#define ZT_VL1_HPP + +#include "Constants.hpp" +#include "Defragmenter.hpp" +#include "Buf.hpp" +#include "Address.hpp" +#include "Protocol.hpp" +#include "Hashtable.hpp" +#include "Mutex.hpp" +#include "FCV.hpp" + +#include + +namespace ZeroTier { + +class RuntimeEnvironment; +class Peer; +class VL2; + +/** + * VL1 (virtual layer 1) packet I/O and messaging + */ +class VL1 +{ +public: + explicit VL1(const RuntimeEnvironment *renv); + ~VL1(); + + /** + * Called when a packet is received from the real network + * + * @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call + * @param localSocket Local I/O socket as supplied by external code + * @param fromAddr Internet IP address of origin + * @param data Packet data + * @param len Packet length + */ + void onRemotePacket(void *tPtr,int64_t localSocket,const InetAddress &fromAddr,SharedPtr &data,unsigned int len); + +private: + void _relay(void *tPtr,const SharedPtr &path,const Address &destination,SharedPtr &data,unsigned int len); + void _sendPendingWhois(); + + // Handlers for VL1 verbs + void _HELLO(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _ERROR(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _OK(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _WHOIS(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _RENDEZVOUS(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _ECHO(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _PUSH_DIRECT_PATHS(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _USER_MESSAGE(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _ENCAP(void *tPtr,const SharedPtr &path,const SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + + const RuntimeEnvironment *RR; + VL2 *const _vl2; + + struct _WhoisQueueItem + { + ZT_ALWAYS_INLINE _WhoisQueueItem() : lastRetry(0),inboundPackets(),retries(0) {} + int64_t lastRetry; + FCV inboundPackets; // capacity can be changed but this should be plenty + unsigned int retries; + }; + + Defragmenter _inputPacketAssembler; + + Hashtable _whoisQueue; + Mutex _whoisQueue_l; +}; + +} // namespace ZeroTier + +#endif diff --git a/node/VL2.hpp b/node/VL2.hpp new file mode 100644 index 000000000..027a9e44e --- /dev/null +++ b/node/VL2.hpp @@ -0,0 +1,54 @@ +/* + * Copyright (c)2013-2020 ZeroTier, Inc. + * + * Use of this software is governed by the Business Source License included + * in the LICENSE.TXT file in the project's root directory. + * + * Change Date: 2024-01-01 + * + * On the date above, in accordance with the Business Source License, use + * of this software will be governed by version 2.0 of the Apache License. + */ +/****/ + +#ifndef ZT_VL2_HPP +#define ZT_VL2_HPP + +#include "Constants.hpp" +#include "Buf.hpp" +#include "Address.hpp" +#include "Protocol.hpp" +#include "Hashtable.hpp" +#include "Mutex.hpp" +#include "FCV.hpp" + +namespace ZeroTier { + +class RuntimeEnvironment; +class VL1; + +class VL2 +{ + friend class VL1; + +public: + VL1(const RuntimeEnvironment *renv); + ~VL1(); + +protected: + void _FRAME(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _EXT_FRAME(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _MULTICAST_LIKE(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _NETWORK_CREDENTIALS(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _NETWORK_CONFIG_REQUEST(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _NETWORK_CONFIG(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _MULTICAST_GATHER(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _MULTICAST_FRAME_deprecated(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + void _MULTICAST(void *tPtr,const SharedPtr &path,SharedPtr &peer,Buf &pkt,unsigned int len,bool authenticated); + +private: +}; + +} // namespace ZeroTier + +#endif