sikkienl/dash
1
0
Fork 0
mirror of https://github.com/dashpay/dash.git synced 2024-12-25 12:02:48 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

merge bitcoin#16202: Refactor network message deserialization

Browse Source 
Co-authored-by: UdjinM6 <UdjinM6@users.noreply.github.com>
This commit is contained in:
Kittywhiskers Van Gogh 2019-06-13 10:39:44 +02:00
parent 907101098d
commit c0d4df0a06
4 changed files with 139 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
src/net.cpp
View File

@ -653,44 +653,29 @@ 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);
} else {
handled = msg.readData(pch, nBytes);
}
if (handled < 0)
return false;
if (msg.in_data && msg.hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_LENGTH) {
LogPrint(BCLog::NET, "Oversized message from peer=%i, disconnecting\n", GetId());
return false;
}
int handled = m_deserializer->Read(pch, nBytes);
if (handled < 0) 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(msg.m_command);
if (i == mapRecvBytesPerMsgCmd.end())
i = mapRecvBytesPerMsgCmd.find(NET_MESSAGE_COMMAND_OTHER);
assert(i != mapRecvBytesPerMsgCmd.end());
i->second += msg.hdr.nMessageSize + CMessageHeader::HEADER_SIZE;
statsClient.count("bandwidth.message." + std::string(msg.hdr.pchCommand) + ".bytesReceived", msg.hdr.nMessageSize + CMessageHeader::HEADER_SIZE, 1.0f);
i->second += msg.m_raw_message_size;
statsClient.count("bandwidth.message." + std::string(msg.m_command) + ".bytesReceived", msg.m_raw_message_size, 1.0f);
// push the message to the process queue,
vRecvMsg.push_back(std::move(msg));
msg.nTime = nTimeMicros;
complete = true;
}
}
@ -724,8 +709,7 @@ int CNode::GetSendVersion() const
return nSendVersion;
}
int CNetMessage::readHeader(const char *pch, unsigned int nBytes)
int V1TransportDeserializer::readHeader(const char *pch, unsigned int nBytes)
{
// copy data to temporary parsing buffer
unsigned int nRemaining = 24 - nHdrPos;
@ -746,9 +730,10 @@ int CNetMessage::readHeader(const char *pch, unsigned int nBytes)
return -1;
}
// reject messages larger than MAX_SIZE
if (hdr.nMessageSize > MAX_SIZE)
// reject messages larger than MAX_SIZE or MAX_PROTOCOL_MESSAGE_LENGTH
if (hdr.nMessageSize > MAX_SIZE || hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_LENGTH) {
return -1;
}
// switch state to reading message data
in_data = true;
@ -756,7 +741,7 @@ int CNetMessage::readHeader(const char *pch, unsigned int nBytes)
return nCopy;
}
int CNetMessage::readData(const char *pch, unsigned int nBytes)
int V1TransportDeserializer::readData(const char *pch, unsigned int nBytes)
{
unsigned int nRemaining = hdr.nMessageSize - nDataPos;
unsigned int nCopy = std::min(nRemaining, nBytes);
@ -773,14 +758,44 @@ int CNetMessage::readData(const char *pch, unsigned int nBytes)
return nCopy;
}
const uint256& CNetMessage::GetMessageHash() const
const uint256& V1TransportDeserializer::GetMessageHash() const
{
assert(complete());
assert(Complete());
if (data_hash.IsNull())
hasher.Finalize(data_hash);
return data_hash;
}
CNetMessage V1TransportDeserializer::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_raw_message_size = hdr.nMessageSize + CMessageHeader::HEADER_SIZE;
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(Span<uint8_t>(hash.begin(), hash.begin()+CMessageHeader::CHECKSUM_SIZE)),
HexStr(Span<uint8_t>(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) EXCLUSIVE_LOCKS_REQUIRED(pnode->cs_vSend)
{
auto it = pnode->vSendMsg.begin();
@ -1828,9 +1843,9 @@ size_t CConnman::SocketRecvData(CNode *pnode)
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_raw_message_size;
}
{
LOCK(pnode->cs_vProcessMsg);
@ -3732,6 +3747,8 @@ CNode::CNode(NodeId idIn, ServiceFlags nLocalServicesIn, int nMyStartingHeightIn
} else {
LogPrint(BCLog::NET, "Added connection peer=%d\n", id);
}
m_deserializer = MakeUnique<V1TransportDeserializer>(V1TransportDeserializer(Params().MessageStart(), SER_NETWORK, INIT_PROTO_VERSION));
}
CNode::~CNode()

83
src/net.h
View File

@ -782,56 +782,105 @@ 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
uint32_t m_raw_message_size = 0; // used wire size of the message (including header/checksum)
std::string m_command;
CNetMessage(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 {
public:
// returns true if the current deserialization is complete
virtual bool Complete() const = 0;
// set the serialization context version
virtual void SetVersion(int version) = 0;
// read and deserialize data
virtual int Read(const char *data, unsigned int bytes) = 0;
// decomposes a message from the context
virtual CNetMessage GetMessage(const CMessageHeader::MessageStartChars& message_start, int64_t time) = 0;
virtual ~TransportDeserializer() {}
};
class V1TransportDeserializer final : public TransportDeserializer
{
private:
mutable CHash256 hasher;
mutable uint256 data_hash;
public:
bool in_data; // parsing header (false) or data (true)
CDataStream hdrbuf; // partially received header
CMessageHeader hdr; // complete header
unsigned int nHdrPos;
CDataStream vRecv; // received message data
unsigned int nHdrPos;
unsigned int nDataPos;
int64_t nTime; // time (in microseconds) of message receipt.
const uint256& GetMessageHash() const;
int readHeader(const char *pch, unsigned int nBytes);
int readData(const char *pch, unsigned int nBytes);
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
public:
V1TransportDeserializer(const CMessageHeader::MessageStartChars& pchMessageStartIn, int nTypeIn, int nVersionIn) : hdrbuf(nTypeIn, nVersionIn), hdr(pchMessageStartIn), vRecv(nTypeIn, nVersionIn) {
Reset();
}
bool Complete() const override
{
if (!in_data)
return false;
return (hdr.nMessageSize == nDataPos);
}
const uint256& GetMessageHash() const;
void SetVersion(int nVersionIn)
void SetVersion(int nVersionIn) override
{
hdrbuf.SetVersion(nVersionIn);
vRecv.SetVersion(nVersionIn);
}
int readHeader(const char *pch, unsigned int nBytes);
int readData(const char *pch, unsigned int nBytes);
int Read(const char *pch, unsigned int nBytes) override {
int ret = in_data ? readData(pch, nBytes) : readHeader(pch, nBytes);
if (ret < 0) Reset();
return ret;
}
CNetMessage GetMessage(const CMessageHeader::MessageStartChars& message_start, int64_t time) override;
};
/** Information about a peer */
class CNode
{
friend class CConnman;
public:
std::unique_ptr<TransportDeserializer> m_deserializer;
// socket
std::atomic<ServiceFlags> nServices{NODE_NONE};
SOCKET hSocket GUARDED_BY(cs_hSocket);

32
src/net_processing.cpp
View File

@ -4026,41 +4026,37 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& 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_raw_message_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;
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(Span<uint8_t>(hash.begin(), hash.begin() + CMessageHeader::CHECKSUM_SIZE)),
HexStr(hdr.pchChecksum));
LogPrint(BCLog::NET, "%s(%s, %u bytes): CHECKSUM ERROR peer=%d\n", __func__,
SanitizeString(strCommand), nMessageSize, pfrom->GetId());
return fMoreWork;
}
@ -4068,7 +4064,7 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter
bool fRet = false;
try
{
fRet = ProcessMessage(pfrom, strCommand, vRecv, msg.nTime, chainparams, connman, interruptMsgProc, m_enable_bip61);
fRet = ProcessMessage(pfrom, strCommand, vRecv, msg.m_time, chainparams, connman, interruptMsgProc, m_enable_bip61);
if (interruptMsgProc)
return false;
if (!pfrom->vRecvGetData.empty())

11
test/functional/p2p_invalid_messages.py
View File

@ -97,11 +97,10 @@ class InvalidMessagesTest(BitcoinTestFramework):
msg_over_size = msg_unrecognized(str_data="b" * (valid_data_limit + 1))
assert len(msg_over_size.serialize()) == (msg_limit + 1)
with node.assert_debug_log(["Oversized message from peer=4, disconnecting"]):
# An unknown message type (or *any* message type) over
# MAX_PROTOCOL_MESSAGE_LENGTH should result in a disconnect.
node.p2p.send_message(msg_over_size)
node.p2p.wait_for_disconnect(timeout=4)
# An unknown message type (or *any* message type) over
# MAX_PROTOCOL_MESSAGE_LENGTH should result in a disconnect.
node.p2p.send_message(msg_over_size)
node.p2p.wait_for_disconnect(timeout=4)
node.disconnect_p2ps()
conn = node.add_p2p_connection(P2PDataStore())
@ -164,7 +163,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