From 6294ecdb8bb4eb7049a18c721ee8cb4a53d80a06 Mon Sep 17 00:00:00 2001 From: Jonas Schnelli Date: Thu, 13 Jun 2019 10:39:44 +0200 Subject: [PATCH] Refactor: split network transport deserializing from message container --- src/net.cpp | 78 +++++++++++++++++-------- src/net.h | 43 ++++++++++++-- src/net_processing.cpp | 32 +++++----- test/functional/p2p_invalid_messages.py | 2 +- 4 files changed, 107 insertions(+), 48 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 63b78338226..bd183ce52b2 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -570,42 +570,42 @@ bool CNode::ReceiveMsgBytes(const char *pch, unsigned int nBytes, bool& complete nLastRecv = nTimeMicros / 1000000; nRecvBytes += nBytes; while (nBytes > 0) { - - // get current incomplete message, or create a new one - if (vRecvMsg.empty() || - vRecvMsg.back().complete()) - vRecvMsg.push_back(CNetMessage(Params().MessageStart(), SER_NETWORK, INIT_PROTO_VERSION)); - - CNetMessage& msg = vRecvMsg.back(); - // absorb network data int handled; - if (!msg.in_data) - handled = msg.readHeader(pch, nBytes); + if (!m_deserializer->in_data) + handled = m_deserializer->readHeader(pch, nBytes); else - handled = msg.readData(pch, nBytes); + handled = m_deserializer->readData(pch, nBytes); - if (handled < 0) + if (handled < 0) { + m_deserializer->Reset(); return false; + } - if (msg.in_data && msg.hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_LENGTH) { + if (m_deserializer->in_data && m_deserializer->hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_LENGTH) { LogPrint(BCLog::NET, "Oversized message from peer=%i, disconnecting\n", GetId()); + m_deserializer->Reset(); return false; } pch += handled; nBytes -= handled; - if (msg.complete()) { + if (m_deserializer->complete()) { + // decompose a transport agnostic CNetMessage from the deserializer + CNetMessage msg = m_deserializer->GetMessage(Params().MessageStart(), nTimeMicros); + //store received bytes per message command //to prevent a memory DOS, only allow valid commands - mapMsgCmdSize::iterator i = mapRecvBytesPerMsgCmd.find(msg.hdr.pchCommand); + mapMsgCmdSize::iterator i = mapRecvBytesPerMsgCmd.find(m_deserializer->hdr.pchCommand); if (i == mapRecvBytesPerMsgCmd.end()) i = mapRecvBytesPerMsgCmd.find(NET_MESSAGE_COMMAND_OTHER); assert(i != mapRecvBytesPerMsgCmd.end()); - i->second += msg.hdr.nMessageSize + CMessageHeader::HEADER_SIZE; + i->second += m_deserializer->hdr.nMessageSize + CMessageHeader::HEADER_SIZE; + + // push the message to the process queue, + vRecvMsg.push_back(std::move(msg)); - msg.nTime = nTimeMicros; complete = true; } } @@ -639,8 +639,7 @@ int CNode::GetSendVersion() const return nSendVersion; } - -int CNetMessage::readHeader(const char *pch, unsigned int nBytes) +int TransportDeserializer::readHeader(const char *pch, unsigned int nBytes) { // copy data to temporary parsing buffer unsigned int nRemaining = 24 - nHdrPos; @@ -671,7 +670,7 @@ int CNetMessage::readHeader(const char *pch, unsigned int nBytes) return nCopy; } -int CNetMessage::readData(const char *pch, unsigned int nBytes) +int TransportDeserializer::readData(const char *pch, unsigned int nBytes) { unsigned int nRemaining = hdr.nMessageSize - nDataPos; unsigned int nCopy = std::min(nRemaining, nBytes); @@ -688,7 +687,7 @@ int CNetMessage::readData(const char *pch, unsigned int nBytes) return nCopy; } -const uint256& CNetMessage::GetMessageHash() const +const uint256& TransportDeserializer::GetMessageHash() const { assert(complete()); if (data_hash.IsNull()) @@ -696,6 +695,35 @@ const uint256& CNetMessage::GetMessageHash() const return data_hash; } +CNetMessage TransportDeserializer::GetMessage(const CMessageHeader::MessageStartChars& message_start, int64_t time) { + // decompose a single CNetMessage from the TransportDeserializer + CNetMessage msg(std::move(vRecv)); + + // store state about valid header, netmagic and checksum + msg.m_valid_header = hdr.IsValid(message_start); + msg.m_valid_netmagic = (memcmp(hdr.pchMessageStart, message_start, CMessageHeader::MESSAGE_START_SIZE) == 0); + uint256 hash = GetMessageHash(); + + // store command string, payload size + msg.m_command = hdr.GetCommand(); + msg.m_message_size = hdr.nMessageSize; + + msg.m_valid_checksum = (memcmp(hash.begin(), hdr.pchChecksum, CMessageHeader::CHECKSUM_SIZE) == 0); + if (!msg.m_valid_checksum) { + LogPrint(BCLog::NET, "CHECKSUM ERROR (%s, %u bytes), expected %s was %s\n", + SanitizeString(msg.m_command), msg.m_message_size, + HexStr(hash.begin(), hash.begin()+CMessageHeader::CHECKSUM_SIZE), + HexStr(hdr.pchChecksum, hdr.pchChecksum+CMessageHeader::CHECKSUM_SIZE)); + } + + // store receive time + msg.m_time = time; + + // reset the network deserializer (prepare for the next message) + Reset(); + return msg; +} + size_t CConnman::SocketSendData(CNode *pnode) const EXCLUSIVE_LOCKS_REQUIRED(pnode->cs_vSend) { auto it = pnode->vSendMsg.begin(); @@ -1347,9 +1375,9 @@ void CConnman::SocketHandler() size_t nSizeAdded = 0; auto it(pnode->vRecvMsg.begin()); for (; it != pnode->vRecvMsg.end(); ++it) { - if (!it->complete()) - break; - nSizeAdded += it->vRecv.size() + CMessageHeader::HEADER_SIZE; + // vRecvMsg contains only completed CNetMessage + // the single possible partially deserialized message are held by TransportDeserializer + nSizeAdded += it->m_recv.size() + CMessageHeader::HEADER_SIZE; } { LOCK(pnode->cs_vProcessMsg); @@ -2678,6 +2706,8 @@ CNode::CNode(NodeId idIn, ServiceFlags nLocalServicesIn, int nMyStartingHeightIn } else { LogPrint(BCLog::NET, "Added connection peer=%d\n", id); } + + m_deserializer = MakeUnique(TransportDeserializer(Params().MessageStart(), SER_NETWORK, INIT_PROTO_VERSION)); } CNode::~CNode() diff --git a/src/net.h b/src/net.h index 44655abf80d..a9ce33957d4 100644 --- a/src/net.h +++ b/src/net.h @@ -609,8 +609,33 @@ public: - +/** Transport protocol agnostic message container. + * Ideally it should only contain receive time, payload, + * command and size. + */ class CNetMessage { +public: + CDataStream m_recv; // received message data + int64_t m_time = 0; // time (in microseconds) of message receipt. + bool m_valid_netmagic = false; + bool m_valid_header = false; + bool m_valid_checksum = false; + uint32_t m_message_size = 0; // size of the payload + std::string m_command; + + CNetMessage(const CDataStream& recv_in) : m_recv(std::move(recv_in)) {} + + void SetVersion(int nVersionIn) + { + m_recv.SetVersion(nVersionIn); + } +}; + +/** The TransportDeserializer takes care of holding and deserializing the + * network receive buffer. It can deserialize the network buffer into a + * transport protocol agnostic CNetMessage (command & payload) + */ +class TransportDeserializer { private: mutable CHash256 hasher; mutable uint256 data_hash; @@ -624,14 +649,19 @@ public: CDataStream vRecv; // received message data unsigned int nDataPos; - int64_t nTime; // time (in microseconds) of message receipt. + TransportDeserializer(const CMessageHeader::MessageStartChars& pchMessageStartIn, int nTypeIn, int nVersionIn) : hdrbuf(nTypeIn, nVersionIn), hdr(pchMessageStartIn), vRecv(nTypeIn, nVersionIn) { + Reset(); + } - CNetMessage(const CMessageHeader::MessageStartChars& pchMessageStartIn, int nTypeIn, int nVersionIn) : hdrbuf(nTypeIn, nVersionIn), hdr(pchMessageStartIn), vRecv(nTypeIn, nVersionIn) { + void Reset() { + vRecv.clear(); + hdrbuf.clear(); hdrbuf.resize(24); in_data = false; nHdrPos = 0; nDataPos = 0; - nTime = 0; + data_hash.SetNull(); + hasher.Reset(); } bool complete() const @@ -651,14 +681,17 @@ public: int readHeader(const char *pch, unsigned int nBytes); int readData(const char *pch, unsigned int nBytes); -}; + CNetMessage GetMessage(const CMessageHeader::MessageStartChars& message_start, int64_t time); +}; /** Information about a peer */ class CNode { friend class CConnman; public: + std::unique_ptr m_deserializer; + // socket std::atomic nServices{NODE_NONE}; SOCKET hSocket GUARDED_BY(cs_hSocket); diff --git a/src/net_processing.cpp b/src/net_processing.cpp index b6839dcf215..c8ffbbbe9d7 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -3260,41 +3260,37 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic& inter return false; // Just take one message msgs.splice(msgs.begin(), pfrom->vProcessMsg, pfrom->vProcessMsg.begin()); - pfrom->nProcessQueueSize -= msgs.front().vRecv.size() + CMessageHeader::HEADER_SIZE; + pfrom->nProcessQueueSize -= msgs.front().m_recv.size() + CMessageHeader::HEADER_SIZE; pfrom->fPauseRecv = pfrom->nProcessQueueSize > connman->GetReceiveFloodSize(); fMoreWork = !pfrom->vProcessMsg.empty(); } CNetMessage& msg(msgs.front()); msg.SetVersion(pfrom->GetRecvVersion()); - // Scan for message start - if (memcmp(msg.hdr.pchMessageStart, chainparams.MessageStart(), CMessageHeader::MESSAGE_START_SIZE) != 0) { - LogPrint(BCLog::NET, "PROCESSMESSAGE: INVALID MESSAGESTART %s peer=%d\n", SanitizeString(msg.hdr.GetCommand()), pfrom->GetId()); + // Check network magic + if (!msg.m_valid_netmagic) { + LogPrint(BCLog::NET, "PROCESSMESSAGE: INVALID MESSAGESTART %s peer=%d\n", SanitizeString(msg.m_command), pfrom->GetId()); pfrom->fDisconnect = true; return false; } - // Read header - CMessageHeader& hdr = msg.hdr; - if (!hdr.IsValid(chainparams.MessageStart())) + // Check header + if (!msg.m_valid_header) { - LogPrint(BCLog::NET, "PROCESSMESSAGE: ERRORS IN HEADER %s peer=%d\n", SanitizeString(hdr.GetCommand()), pfrom->GetId()); + LogPrint(BCLog::NET, "PROCESSMESSAGE: ERRORS IN HEADER %s peer=%d\n", SanitizeString(msg.m_command), pfrom->GetId()); return fMoreWork; } - std::string strCommand = hdr.GetCommand(); + const std::string& strCommand = msg.m_command; // Message size - unsigned int nMessageSize = hdr.nMessageSize; + unsigned int nMessageSize = msg.m_message_size; // Checksum - CDataStream& vRecv = msg.vRecv; - const uint256& hash = msg.GetMessageHash(); - if (memcmp(hash.begin(), hdr.pchChecksum, CMessageHeader::CHECKSUM_SIZE) != 0) + CDataStream& vRecv = msg.m_recv; + if (!msg.m_valid_checksum) { - LogPrint(BCLog::NET, "%s(%s, %u bytes): CHECKSUM ERROR expected %s was %s\n", __func__, - SanitizeString(strCommand), nMessageSize, - HexStr(hash.begin(), hash.begin()+CMessageHeader::CHECKSUM_SIZE), - HexStr(hdr.pchChecksum, hdr.pchChecksum+CMessageHeader::CHECKSUM_SIZE)); + LogPrint(BCLog::NET, "%s(%s, %u bytes): CHECKSUM ERROR peer=%d\n", __func__, + SanitizeString(strCommand), nMessageSize, pfrom->GetId()); return fMoreWork; } @@ -3302,7 +3298,7 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic& inter bool fRet = false; try { - fRet = ProcessMessage(pfrom, strCommand, vRecv, msg.nTime, chainparams, connman, interruptMsgProc); + fRet = ProcessMessage(pfrom, strCommand, vRecv, msg.m_time, chainparams, connman, interruptMsgProc); if (interruptMsgProc) return false; if (!pfrom->vRecvGetData.empty()) diff --git a/test/functional/p2p_invalid_messages.py b/test/functional/p2p_invalid_messages.py index 58c72f89d81..8fcf5b878f2 100755 --- a/test/functional/p2p_invalid_messages.py +++ b/test/functional/p2p_invalid_messages.py @@ -168,7 +168,7 @@ class InvalidMessagesTest(BitcoinTestFramework): def test_checksum(self): conn = self.nodes[0].add_p2p_connection(P2PDataStore()) - with self.nodes[0].assert_debug_log(['ProcessMessages(badmsg, 2 bytes): CHECKSUM ERROR expected 78df0a04 was ffffffff']): + with self.nodes[0].assert_debug_log(['CHECKSUM ERROR (badmsg, 2 bytes), expected 78df0a04 was ffffffff']): msg = conn.build_message(msg_unrecognized(str_data="d")) cut_len = ( 4 + # magic